chore(telemetry): replace add_error by add_error_log

Author: dubloom

ddtrace/internal/telemetry/logging.py

@@ -14,6 +14,4 @@ def emit(self, record: logging.LogRecord) -> None:
         - Log all records with a level of ERROR or higher with telemetry
         """
         if record.levelno >= logging.ERROR:
-            # Capture start up errors
-            full_file_name = os.path.join(record.pathname, record.filename)
-            self.telemetry_writer.add_error(1, record.msg, full_file_name, record.lineno)
+            self.telemetry_writer.add_error_log(record.msg, record.exc_info)

ddtrace/internal/telemetry/writer.py

@@ -163,10 +163,6 @@ def __init__(self, is_periodic=True, agentless=None):
         self._periodic_count = 0
         self._is_periodic = is_periodic
         self._integrations_queue = dict()  # type: Dict[str, Dict]
-        # Currently telemetry only supports reporting a single error.
-        # If we'd like to report multiple errors in the future
-        # we could hack it in by xor-ing error codes and concatenating strings
-        self._error = (0, "")  # type: Tuple[int, str]
         self._namespace = MetricNamespace()
         self._logs = set()  # type: Set[Dict[str, Any]]
         self._forked = False  # type: bool
@@ -301,15 +297,6 @@ def add_integration(self, integration_name, patched, auto_patched=None, error_ms
                 self._integrations_queue[integration_name]["compatible"] = error_msg == ""
                 self._integrations_queue[integration_name]["error"] = error_msg
 
-    def add_error(self, code, msg, filename, line_number):
-        # type: (int, str, Optional[str], Optional[int]) -> None
-        """Add an error to be submitted with an event.
-        Note that this overwrites any previously set errors.
-        """
-        if filename and line_number is not None:
-            msg = "%s:%s: %s" % (filename, line_number, msg)
-        self._error = (code, msg)
-
     def _app_started(self, register_app_shutdown=True):
         # type: (bool) -> None
         """Sent when TelemetryWriter is enabled or forks"""
@@ -330,10 +317,6 @@ def _app_started(self, register_app_shutdown=True):
 
         payload = {
             "configuration": self._flush_configuration_queue(),
-            "error": {
-                "code": self._error[0],
-                "message": self._error[1],
-            },
             "products": products,
         }  # type: Dict[str, Union[Dict[str, Any], List[Any]]]
         # Add time to value telemetry metrics for single step instrumentation
@@ -343,9 +326,6 @@ def _app_started(self, register_app_shutdown=True):
                 "install_type": config.INSTALL_TYPE,
                 "install_time": config.INSTALL_TIME,
             }
-
-        # Reset the error after it has been reported.
-        self._error = (0, "")
         self.add_event(payload, "app-started")
 
     def _app_heartbeat_event(self):
@@ -524,18 +504,21 @@ def add_log(self, level, message, stack_trace="", tags=None):
             # Logs are hashed using the message, level, tags, and stack_trace. This should prevent duplicatation.
             self._logs.add(data)
 
-    def add_error_log(self, msg: str, exc: BaseException) -> None:
+    def add_error_log(self, msg: str, exc: Union[BaseException, tuple, None]) -> None:
         if config.LOG_COLLECTION_ENABLED:
-            stack_trace = self._format_stack_trace(exc)
+            stack_trace = None if exc is None else self._format_stack_trace(exc)
 
             self.add_log(
                 TELEMETRY_LOG_LEVEL.ERROR,
                 msg,
                 stack_trace=stack_trace if stack_trace is not None else "",
             )
 
-    def _format_stack_trace(self, exc: BaseException) -> Optional[str]:
-        exc_type, _, exc_traceback = type(exc), exc, getattr(exc, "__traceback__", None)
+    def _format_stack_trace(self, exc: Union[BaseException, tuple]) -> Optional[str]:
+        if isinstance(exc, tuple) and len(exc) == 3:
+            exc_type, _, exc_traceback = exc
+        else:
+            exc_type, _, exc_traceback = type(exc), exc, getattr(exc, "__traceback__", None)
 
         if not exc_traceback:
             return None
@@ -759,7 +742,8 @@ def _telemetry_excepthook(self, tp, value, root_traceback):
 
             lineno = traceback.tb_frame.f_code.co_firstlineno
             filename = traceback.tb_frame.f_code.co_filename
-            self.add_error(1, str(value), filename, lineno)
+
+            self.add_error_log("Unhandled exception from ddtrace code", (tp, None, root_traceback))
 
             dir_parts = filename.split(os.path.sep)
             # Check if exception was raised in the  `ddtrace.contrib` package

tests/telemetry/test_telemetry.py

@@ -1,6 +1,4 @@
 import os
-import re
-
 import pytest
 
 
@@ -173,16 +171,13 @@ def process_trace(self, trace):
 
     app_started_events = [event for event in events if event["request_type"] == "app-started"]
     assert len(app_started_events) == 1
-    assert app_started_events[0]["payload"]["error"]["code"] == 1
-    assert (
-        "error applying processor <__main__.FailingFilture object at"
-        not in app_started_events[0]["payload"]["error"]["message"]
-    )
-    assert "error applying processor %r" in app_started_events[0]["payload"]["error"]["message"]
-    pattern = re.compile(".*ddtrace/_trace/processor/__init__.py/__init__.py:[0-9]+: " "error applying processor %r")
-    assert pattern.match(app_started_events[0]["payload"]["error"]["message"]), app_started_events[0]["payload"][
-        "error"
-    ]["message"]
+
+    logs_event = test_agent_session.get_events("logs", subprocess=True)
+    error_log = logs_event[0]["payload"]["logs"][0]
+    assert error_log["message"] == "error applying processor %r to trace %d"
+    assert error_log["level"] == "ERROR"
+    assert "in on_span_finish" in error_log["stack_trace"]
+    assert "spans = tp.process_trace(spans) or []" in error_log["stack_trace"]
 
 
 def test_register_telemetry_excepthook_after_another_hook(test_agent_session, run_python_code_in_subprocess):
@@ -211,11 +206,11 @@ def pre_ddtrace_exc_hook(exctype, value, traceback):
 
     app_starteds = test_agent_session.get_events("app-started", subprocess=True)
     assert len(app_starteds) == 1
-    # app-started captures unhandled exceptions raised in application code
-    assert app_starteds[0]["payload"]["error"]["code"] == 1
-    assert re.search(r"test\.py:\d+:\sbad_code$", app_starteds[0]["payload"]["error"]["message"]), app_starteds[0][
-        "payload"
-    ]["error"]["message"]
+
+    # the tracer should not capture logs from non ddtrace code
+    # i will not test further as it will be removed in next PR
+    logs_event = test_agent_session.get_events("logs", subprocess=True)
+    assert len(logs_event) == 1
 
 
 def test_handled_integration_error(test_agent_session, run_python_code_in_subprocess):
@@ -271,9 +266,12 @@ def test_unhandled_integration_error(test_agent_session, ddtrace_run_python_code
 
     app_started_event = test_agent_session.get_events("app-started", subprocess=True)
     assert len(app_started_event) == 1
-    assert app_started_event[0]["payload"]["error"]["code"] == 1
-    assert "ddtrace/contrib/internal/flask/patch.py" in app_started_event[0]["payload"]["error"]["message"]
-    assert "not enough values to unpack (expected 2, got 0)" in app_started_event[0]["payload"]["error"]["message"]
+
+    logs_event = test_agent_session.get_events("logs", subprocess=True)
+    error_log = logs_event[0]["payload"]["logs"][0]
+    assert error_log["message"] == "Unhandled exception from ddtrace code"
+    assert error_log["level"] == "ERROR"
+    assert "patched_wsgi_app" in error_log["stack_trace"]
 
     integration_events = test_agent_session.get_events("app-integrations-change", subprocess=True)
     integrations = integration_events[0]["payload"]["integrations"]