chore(telemetry): integration exception tracking

ddtrace/internal/logger.py

@@ -134,14 +134,6 @@ def handle(self, record):
         :param record: The log record being logged
         :type record: ``logging.LogRecord``
         """
-        if record.levelno >= logging.ERROR:
-            # avoid circular import
-            from ddtrace.internal import telemetry
-
-            # currently we only have one error code
-            full_file_name = os.path.join(record.pathname, record.filename)
-            telemetry.telemetry_writer.add_error(1, record.msg % record.args, full_file_name, record.lineno)
-
         # If rate limiting has been disabled (`DD_TRACE_LOGGING_RATE=0`) then apply no rate limit
         # If the logging is in debug, then do not apply any limits to any log
         if not self.rate_limit or self.getEffectiveLevel() == logging.DEBUG:

ddtrace/internal/telemetry/logging.py

@@ -0,0 +1,67 @@
+import logging
+import os
+import traceback
+
+from ddtrace.internal.telemetry.constants import TELEMETRY_LOG_LEVEL
+
+
+class DDTelemetryLogHandler(logging.StreamHandler):
+    CWD = os.getcwd()
+
+    def __init__(self, telemetry_writer):
+        self.telemetry_writer = telemetry_writer
+        super().__init__()
+
+    def emit(self, record):
+        # type: (logging.LogRecord) -> None
+        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 % record.args, full_file_name, record.lineno)
+
+        # Capture errors logged in the ddtrace integrations
+        if record.name.startswith("ddtrace.contrib"):
+            telemetry_level = (
+                TELEMETRY_LOG_LEVEL.ERROR
+                if record.levelno >= logging.ERROR
+                else TELEMETRY_LOG_LEVEL.WARNING
+                if record.levelno == logging.WARNING
+                else TELEMETRY_LOG_LEVEL.DEBUG
+            )
+            # Only collect telemetry for logs with a traceback
+            stack_trace = self._format_stack_trace(record.exc_info)
+            if stack_trace is not None:
+                # Report only exceptions with a stack trace
+                self.telemetry_writer.add_log(
+                    telemetry_level,
+                    record.msg,
+                    # Do we need to set this tag? Should we allow telemetry intake to infer this value?
+                    tags={"lib_language": "python"},
+                    stack_trace=stack_trace,
+                )
+
+    def _format_stack_trace(self, exc_info):
+        if exc_info is None:
+            return None
+        exc_type, exc_value, exc_traceback = exc_info
+        if exc_traceback:
+            tb = traceback.extract_tb(exc_traceback)
+            formatted_tb = ["Traceback (most recent call last):"]
+            for filename, lineno, funcname, srcline in tb:
+                if self._should_redact(filename):
+                    formatted_tb.append("  <REDACTED>")
+                else:
+                    relative_filename = self._format_file_path(filename)
+                    formatted_line = f'  File "{relative_filename}", line {lineno}, in {funcname}\n    {srcline}'
+                    formatted_tb.append(formatted_line)
+            formatted_tb.append(f"{exc_type.__module__}.{exc_type.__name__}: {exc_value}")
+            return "\n".join(formatted_tb)
+
+    def _should_redact(self, filename):
+        return "ddtrace" not in filename
+
+    def _format_file_path(self, filename):
+        try:
+            return os.path.relpath(filename, start=self.CWD)
+        except ValueError:
+            return filename

ddtrace/internal/telemetry/writer.py

@@ -39,6 +39,7 @@
 from .data import get_host_info
 from .data import get_python_config_vars
 from .data import update_imported_dependencies
+from .logging import DDTelemetryLogHandler
 from .metrics import CountMetric
 from .metrics import DistributionMetric
 from .metrics import GaugeMetric
@@ -68,6 +69,10 @@ class _TelemetryConfig:
     INSTALL_TYPE = os.environ.get("DD_INSTRUMENTATION_INSTALL_TYPE", None)
     INSTALL_TIME = os.environ.get("DD_INSTRUMENTATION_INSTALL_TIME", None)
     FORCE_START = asbool(os.environ.get("_DD_INSTRUMENTATION_TELEMETRY_TESTS_FORCE_APP_STARTED", "false"))
+    LOG_COLLECTION_ENABLED = TELEMETRY_ENABLED and os.getenv("DD_TELEMETRY_LOG_COLLECTION_ENABLED", "true").lower() in (
+        "true",
+        "1",
+    )
 
 
 class LogData(dict):
@@ -222,6 +227,8 @@ def __init__(self, is_periodic=True, agentless=None):
             # Force app started for unit tests
             if _TelemetryConfig.FORCE_START:
                 self._app_started()
+            if _TelemetryConfig.LOG_COLLECTION_ENABLED:
+                getLogger("ddtrace").addHandler(DDTelemetryLogHandler(self))
 
     def enable(self):
         # type: () -> bool
@@ -504,6 +511,7 @@ def add_log(self, level, message, stack_trace="", tags=None):
                 data["tags"] = ",".join(["%s:%s" % (k, str(v).lower()) for k, v in tags.items()])
             if stack_trace:
                 data["stack_trace"] = stack_trace
+            # Logs are hashed using the message, level, tags, and stack_trace. This should prevent duplicatation.
             self._logs.add(data)
 
     def add_gauge_metric(self, namespace: TELEMETRY_NAMESPACE, name: str, value: float, tags: MetricTagType = None):