Actually make use of Error Monitor and add configuration option for it

connectors/config.py

@@ -87,6 +87,13 @@ def _default_config():
  "preflight_idle": 30,
  "max_errors": 20,
  "max_errors_span": 600,
+ "error_monitor": {
+ "max_total_errors": 1000,
+ "max_consecutive_errors": 10,
+ "max_error_rate": 0.15,
+ "error_window_size": 100,
+ "error_queue_size": 10,
+ },
  "max_concurrent_content_syncs": 1,
  "max_concurrent_access_control_syncs": 1,
  "max_file_download_size": DEFAULT_MAX_FILE_SIZE,

connectors/es/sink.py

@@ -110,6 +110,10 @@ def __init__(self, cause=None):
  self.__cause__ = cause
 
 
+class DocumentIngestionError(Exception):
+ pass
+
+
 class Sink:
  """Send bulk operations in batches by consuming a queue.
 
@@ -136,6 +140,7 @@ def __init__(
  max_concurrency,
  max_retries,
  retry_interval,
+ error_monitor,
  logger_=None,
  enable_bulk_operations_logging=False,
  ):
@@ -145,6 +150,7 @@ def __init__(
  self.pipeline = pipeline
  self.chunk_mem_size = chunk_mem_size * 1024 * 1024
  self.bulk_tasks = ConcurrentTasks(max_concurrency=max_concurrency)
+ self.error_monitor = error_monitor
  self.max_retires = max_retries
  self.retry_interval = retry_interval
  self.error = None
@@ -272,18 +278,19 @@ async def _process_bulk_response(self, res, ids_to_ops, do_log=False):
  successful_result = result in SUCCESSFUL_RESULTS
  if not successful_result:
  if "error" in item[action_item]:
+ message = f"Failed to execute '{action_item}' on document with id '{doc_id}'. Error: {item[action_item].get('error')}"
+ self.error_monitor.track_error(DocumentIngestionError(message))
  if do_log:
- self._logger.debug(
- f"Failed to execute '{action_item}' on document with id '{doc_id}'. Error: {item[action_item].get('error')}"
- )
+ self._logger.debug(message)
  self.counters.increment(RESULT_ERROR, namespace=BULK_RESPONSES)
  else:
+ message = f"Executed '{action_item}' on document with id '{doc_id}', but got non-successful result: {result}"
+ self.error_monitor.track_error(DocumentIngestionError(message))
  if do_log:
- self._logger.debug(
- f"Executed '{action_item}' on document with id '{doc_id}', but got non-successful result: {result}"
- )
+ self._logger.debug(message)
  self.counters.increment(RESULT_UNDEFINED, namespace=BULK_RESPONSES)
  else:
+ self.error_monitor.track_success()
  if do_log:
  self._logger.debug(
  f"Successfully executed '{action_item}' on document with id '{doc_id}'. Result: {result}"
@@ -424,6 +431,7 @@ def __init__(
  client,
  queue,
  index,
+ error_monitor,
  filter_=None,
  sync_rules_enabled=False,
  content_extraction_enabled=True,
@@ -449,27 +457,37 @@ def __init__(
  self.concurrent_downloads = concurrent_downloads
  self._logger = logger_ or logger
  self._canceled = False
+ self.error_monitor = error_monitor
  self.skip_unchanged_documents = skip_unchanged_documents
 
  async def _deferred_index(self, lazy_download, doc_id, doc, operation):
- data = await lazy_download(doit=True, timestamp=doc[TIMESTAMP_FIELD])
-
- if data is not None:
- self.counters.increment(BIN_DOCS_DOWNLOADED)
- data.pop("_id", None)
- data.pop(TIMESTAMP_FIELD, None)
- doc.update(data)
-
- doc.pop("_original_filename", None)
-
- await self.put_doc(
- {
- "_op_type": operation,
- "_index": self.index,
- "_id": doc_id,
- "doc": doc,
- }
- )
+ try:
+ data = await lazy_download(doit=True, timestamp=doc[TIMESTAMP_FIELD])
+
+ if data is not None:
+ self.counters.increment(BIN_DOCS_DOWNLOADED)
+ data.pop("_id", None)
+ data.pop(TIMESTAMP_FIELD, None)
+ doc.update(data)
+
+ doc.pop("_original_filename", None)
+
+ await self.put_doc(
+ {
+ "_op_type": operation,
+ "_index": self.index,
+ "_id": doc_id,
+ "doc": doc,
+ }
+ )
+ self.error_monitor.track_success()
+ except ForceCanceledError:
+ raise
+ except Exception as ex:
+ self._logger.error(
+ f"Failed to do deferred index operation for doc {doc_id}: {ex}"
+ )
+ self.error_monitor.track_error(ex)
 
  def force_cancel(self):
  self._canceled = True
@@ -599,6 +617,8 @@ async def get_docs(self, generator, skip_unchanged_documents=False):
  }
  )
 
+ lazy_downloads.raise_any_exception()
+
  await asyncio.sleep(0)
  finally:
  # wait for all downloads to be finished
@@ -803,7 +823,8 @@ class SyncOrchestrator:
  - once they are both over, returns totals
  """
 
- def __init__(self, elastic_config, logger_=None):
+ def __init__(self, elastic_config, error_monitor, logger_=None):
+ self.error_monitor = error_monitor
  self._logger = logger_ or logger
  self._logger.debug(f"SyncOrchestrator connecting to {elastic_config['host']}")
  self.es_management_client = ESManagementClient(elastic_config)
@@ -884,7 +905,7 @@ def _extractor_task_running(self):
  async def cancel(self):
  if self._sink_task_running():
  self._logger.info(
- f"Cancling the Sink task: {self._sink_task.name}" # pyright: ignore
+ f"Canceling the Sink task: {self._sink_task.get_name()}" # pyright: ignore
  )
  self._sink_task.cancel()
  else:
@@ -894,7 +915,7 @@ async def cancel(self):
 
  if self._extractor_task_running():
  self._logger.info(
- f"Canceling the Extractor task: {self._extractor_task.name}" # pyright: ignore
+ f"Canceling the Extractor task: {self._extractor_task.get_name()}" # pyright: ignore
  )
  self._extractor_task.cancel()
  else:
@@ -1005,6 +1026,7 @@ async def async_bulk(
  self.es_management_client,
  stream,
  index,
+ error_monitor=self.error_monitor,
  filter_=filter_,
  sync_rules_enabled=sync_rules_enabled,
  content_extraction_enabled=content_extraction_enabled,
@@ -1031,6 +1053,7 @@ async def async_bulk(
  max_concurrency=max_concurrency,
  max_retries=max_bulk_retries,
  retry_interval=retry_interval,
+ error_monitor=self.error_monitor,
  logger_=self._logger,
  enable_bulk_operations_logging=enable_bulk_operations_logging,
  )

connectors/source.py

@@ -32,7 +32,6 @@
 from connectors.logger import logger
 from connectors.utils import (
  TIKA_SUPPORTED_FILETYPES,
- ErrorMonitor,
  convert_to_b64,
  epoch_timestamp_zulu,
  get_file_extension,
@@ -410,7 +409,7 @@ def __init__(self, configuration):
 
  # this will be overwritten by set_framework_config()
  self.framework_config = DataSourceFrameworkConfig.Builder().build()
- self.error_monitor = ErrorMonitor()
+ self.error_monitor = None
 
  def __str__(self):
  return f"Datasource `{self.__class__.name}`"
@@ -419,6 +418,9 @@ def set_logger(self, logger_):
  self._logger = logger_
  self._set_internal_logger()
 
+ def set_error_monitor(self, error_monitor):
+ self.error_monitor = error_monitor
+
  def _set_internal_logger(self):
  # no op for BaseDataSource
  # if there are internal class (e.g. Client class) to which the logger need to be set,
@@ -778,12 +780,14 @@ async def download_and_extract_file(
  doc = await self.handle_file_content_extraction(
  doc, source_filename, temp_filename
  )
+ self.error_monitor.track_success()
  return doc
  except Exception as e:
  self._logger.warning(
  f"File download and extraction or conversion for file {source_filename} failed: {e}",
  exc_info=True,
  )
+ self.error_monitor.track_error(e)
  if return_doc_if_failed:
  return doc
  else:

connectors/sources/mongo.py

@@ -240,8 +240,9 @@ def _serialize(value):
  value = _serialize(value.as_doc().to_dict())
  return value
 
- for key, value in doc.items():
- doc[key] = _serialize(value)
+ with self.with_error_monitoring():
+ for key, value in doc.items():
+ doc[key] = _serialize(value)
 
  return doc
 

connectors/sync_job_runner.py

@@ -35,7 +35,7 @@
  INDEXED_DOCUMENT_VOLUME,
 )
 from connectors.source import BaseDataSource
-from connectors.utils import truncate_id
+from connectors.utils import ErrorMonitor, truncate_id
 
 UTF_8 = "utf-8"
 
@@ -113,6 +113,8 @@ def __init__(
  self.es_config = es_config
  self.service_config = service_config
  self.sync_orchestrator = None
+ error_monitor_config = service_config.get("error_monitor", {})
+ self.error_monitor = ErrorMonitor(**error_monitor_config)
  self.job_reporting_task = None
  self.bulk_options = self.es_config.get("bulk", {})
  self._start_time = None
@@ -149,6 +151,7 @@ async def execute(self):
  configuration=self.sync_job.configuration
  )
  self.data_provider.set_logger(self.sync_job.logger)
+ self.data_provider.set_error_monitor(self.error_monitor)
  self.data_provider.set_framework_config(
  self._data_source_framework_config()
  )
@@ -183,7 +186,7 @@ async def execute(self):
  await self._update_native_connector_authentication()
 
  self.sync_orchestrator = SyncOrchestrator(
- self.es_config, self.sync_job.logger
+ self.es_config, self.error_monitor, self.sync_job.logger
  )
 
  if job_type in [JobType.INCREMENTAL, JobType.FULL]:

connectors/utils.py

@@ -1009,7 +1009,7 @@ def __init__(
  max_consecutive_errors=10,
  max_error_rate=0.15,
  error_window_size=100,
- error_queue_size=20,
+ error_queue_size=10,
  ):
  self.max_error_rate = max_error_rate
  self.error_window_size = error_window_size
@@ -1082,13 +1082,13 @@ def _error_window_error_rate(self):
 
  def _raise_if_necessary(self):
  if self.consecutive_error_count > self.max_consecutive_errors:
- msg = f"Exceeded maximum consecutive errors - saw {self.consecutive_error_count} errors in a row"
+ msg = f"Exceeded maximum consecutive errors - saw {self.consecutive_error_count} errors in a row. Last error: {self.last_error}"
  raise TooManyErrors(msg) from self.last_error
  elif self.total_error_count > self.max_total_errors:
- msg = f"Exceeded maximum total error count - saw {self.total_error_count} errors"
+ msg = f"Exceeded maximum total error count - saw {self.total_error_count} errors. Last error: {self.last_error}"
  raise TooManyErrors(msg) from self.last_error
  elif self.error_window_size > 0:
  error_rate = self._error_window_error_rate()
  if error_rate > self.max_error_rate:
- msg = f"Exceeded maximum error ratio of {self.max_error_rate} for last {self.error_window_size} operations."
+ msg = f"Exceeded maximum error ratio of {self.max_error_rate} for last {self.error_window_size} operations. Last error: {self.last_error}"
  raise TooManyErrors(msg) from self.last_error

tests/sources/support.py

@@ -4,6 +4,7 @@
 # you may not use this file except in compliance with the Elastic License 2.0.
 #
 from contextlib import asynccontextmanager
+from unittest.mock import Mock
 
 from connectors.source import DEFAULT_CONFIGURATION, DataSourceConfiguration
 
@@ -18,6 +19,7 @@ async def create_source(klass, **extras):
  config[k] = DEFAULT_CONFIGURATION.copy() | {"value": v}
 
  source = klass(configuration=DataSourceConfiguration(config))
+ source.set_error_monitor(Mock())
  try:
  yield source
  finally: