added a safeguard for processors that raise exceptions

docs/sanitizing-data.asciidoc

@@ -10,6 +10,9 @@ and returns the modified event.
 
 To completely drop an event, your processor should return `False` (or any other "falsy" value) instead of the event.
 
+An event will also be dropped if any processor raises an exception while processing it.
+A log message with level `WARNING` will be issued in this case.
+
 This is an example of a processor that removes the exception stacktrace from an error:
 
 [source,python]
@@ -46,7 +49,7 @@ ELASTIC_APM = {
 }
 ----
 
-NOTE: We recommend to use the above list of processors that sanitize passwords and secrets in different places of the event object.
+NOTE: We recommend using the above list of processors that sanitize passwords and secrets in different places of the event object.
 
 The default set of processors sanitize fields based on a set of defaults defined in `elasticapm.conf.constants`. This set can be configured with the `SANITIZE_FIELD_NAMES` configuration option. For example, if your application produces a sensitive field called `My-Sensitive-Field`, the default processors can be used to automatically sanitize this field. You can specify what fields to santize within default processors like this:
 

elasticapm/transport/base.py

@@ -176,13 +176,23 @@ def _process_event(self, event_type, data):
  # Run the data through processors
  for processor in self._processors:
  if not hasattr(processor, "event_types") or event_type in processor.event_types:
- data = processor(self.client, data)
- if not data:
- logger.debug(
- "Dropped event of type %s due to processor %s.%s",
+ try:
+ data = processor(self.client, data)
+ if not data:
+ logger.debug(
+ "Dropped event of type %s due to processor %s.%s",
+ event_type,
+ processor.__module__,
+ processor.__name__,
+ )
+ return None
+ except Exception:
+ logger.warning(
+ "Dropped event of type %s due to exception in processor %s.%s",
  event_type,
- getattr(processor, "__module__"),
- getattr(processor, "__name__"),
+ processor.__module__,
+ processor.__name__,
+ exc_info=True,
  )
  return None
  return data
@@ -213,7 +223,6 @@ def _init_event_queue(self, chill_until, max_chill_time):
  def _flush(self, buffer):
  """
  Flush the queue. This method should only be called from the event processing queue
- :param sync: if true, flushes the queue synchronously in the current thread
  :return: None
  """
  if not self.state.should_try():

tests/processors/tests.py

@@ -356,6 +356,10 @@ def dummy_processor_no_events(client, data):
  return data
 
 
+def dummy_processor_failing(client, data):
+ raise ValueError("oops")
+
+
 @pytest.mark.parametrize(
  "elasticapm_client",
  [{"processors": "tests.processors.tests.dummy_processor,tests.processors.tests.dummy_processor_no_events"}],
@@ -428,17 +432,36 @@ def foo(client, event):
 
 
 def test_drop_events_in_processor(elasticapm_client, caplog):
- dropping_processor = mock.MagicMock(return_value=None, event_types=[SPAN], __name__="dropper")
+ dropping_processor = mock.MagicMock(return_value=None, event_types=[TRANSACTION], __name__="dropper")
  shouldnt_be_called_processor = mock.Mock(event_types=[])
 
  elasticapm_client._transport._processors = [dropping_processor, shouldnt_be_called_processor]
  with caplog.at_level(logging.DEBUG, logger="elasticapm.transport"):
- elasticapm_client.queue(SPAN, {"some": "data"})
+ elasticapm_client.begin_transaction("test")
+ elasticapm_client.end_transaction("test", "FAIL")
  assert dropping_processor.call_count == 1
  assert shouldnt_be_called_processor.call_count == 0
- assert elasticapm_client._transport.events[SPAN][0] is None
+ assert elasticapm_client._transport.events[TRANSACTION][0] is None
+ assert_any_record_contains(
+ caplog.records, "Dropped event of type transaction due to processor mock.mock.dropper", "elasticapm.transport"
+ )
+
+
+@pytest.mark.parametrize(
+ "elasticapm_client",
+ [{"processors": "tests.processors.tests.dummy_processor,tests.processors.tests.dummy_processor_failing"}],
+ indirect=True,
+)
+def test_drop_events_in_failing_processor(elasticapm_client, caplog):
+
+ with caplog.at_level(logging.WARNING, logger="elasticapm.transport"):
+ elasticapm_client.begin_transaction("test")
+ elasticapm_client.end_transaction("test", "FAIL")
+ assert elasticapm_client._transport.events[TRANSACTION][0] is None
  assert_any_record_contains(
- caplog.records, "Dropped event of type span due to processor mock.mock.dropper", "elasticapm.transport"
+ caplog.records,
+ "Dropped event of type transaction due to exception in processor tests.processors.tests.dummy_processor_failing",
+ "elasticapm.transport",
  )