added a safeguard for processors that raise exceptions (#1138)

* added a safeguard for processors that raise exceptions

Without this safeguard, exceptions in processors would bubble all the way
up, killing the transport thread. This would then result in the agent
stopping to report any data.

* fix style issue

Author: beniwohli

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

@@ -178,13 +178,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
@@ -215,7 +225,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",
     )