diff --git a/ddtrace/internal/logger.py b/ddtrace/internal/logger.py
index 0592afa0a74..ed190314503 100644
--- a/ddtrace/internal/logger.py
+++ b/ddtrace/internal/logger.py
@@ -1,6 +1,7 @@
 import collections
 import logging
 import os
+import traceback
 import typing
 from typing import Optional  # noqa:F401
 from typing import cast  # noqa:F401
@@ -46,7 +47,7 @@ def get_logger(name):
         logger = manager.loggerDict[name]
         if isinstance(manager.loggerDict[name], logging.PlaceHolder):
             placeholder = logger
-            logger = DDLogger(name=name)
+            logger = _new_logger(name=name)
             manager.loggerDict[name] = logger
             # DEV: `_fixupChildren` and `_fixupParents` have been around for awhile,
             # DEV: but add the `hasattr` guard... just in case.
@@ -55,7 +56,7 @@ def get_logger(name):
             if hasattr(manager, "_fixupParents"):
                 manager._fixupParents(logger)
     else:
-        logger = DDLogger(name=name)
+        logger = _new_logger(name=name)
         manager.loggerDict[name] = logger
         if hasattr(manager, "_fixupParents"):
             manager._fixupParents(logger)
@@ -64,6 +65,13 @@ def get_logger(name):
     return cast(DDLogger, logger)
 
 
+def _new_logger(name):
+    if _TelemetryConfig.LOG_COLLECTION_ENABLED:
+        if name.startswith("ddtrace.contrib."):
+            return DDTelemetryLogger(name=name)
+    return DDLogger(name=name)
+
+
 def hasHandlers(self):
     # type: (DDLogger) -> bool
     """
@@ -178,3 +186,66 @@ def handle(self, record):
             # Increment the count of records we have skipped
             # DEV: `self.buckets[key]` is a tuple which is immutable so recreate instead
             self.buckets[key] = DDLogger.LoggingBucket(logging_bucket.bucket, logging_bucket.skipped + 1)
+
+
+class DDTelemetryLogger(DDLogger):
+    """
+    Logger that intercepts and reports exceptions to the telemetry.
+    """
+
+    def __init__(self, *args, **kwargs):
+        # type: (*Any, **Any) -> None
+        """Constructor for ``DDTelemetryLogger``"""
+        super(DDTelemetryLogger, self).__init__(*args, **kwargs)
+
+        self.telemetry_log_buckets = collections.defaultdict(
+            lambda: DDLogger.LoggingBucket(0, 0)
+        )  # type: DefaultDict[Tuple[str, int, str, int], DDLogger.LoggingBucket]
+
+
+    def handle(self, record):
+        # type: (logging.LogRecord) -> None
+
+        from ddtrace.internal.telemetry.constants import TELEMETRY_LOG_LEVEL
+
+        key = (record.name, record.levelno, record.pathname, record.lineno)
+        current_bucket = int(record.created / _TelemetryConfig.TELEMETRY_HEARTBEAT_INTERVAL)
+        key_bucket = self.telemetry_log_buckets[key]
+        if key_bucket.bucket == current_bucket:
+            self.telemetry_log_buckets[key] = DDLogger.LoggingBucket(key_bucket.bucket, key_bucket.skipped + 1)
+        else:
+            self.telemetry_log_buckets[key] = DDLogger.LoggingBucket(current_bucket, 0)
+            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
+            )
+            from ddtrace.internal import telemetry
+
+            tags = {
+                "lib_language": "python",
+            }
+            stack_trace = None
+            if record.exc_info:
+                _, _, traceback_object = record.exc_info
+                if traceback_object:
+                    stack_trace = "".join(traceback.format_tb(traceback_object))
+                    # TODO redact absolute file paths and unknown packages
+            if record.levelno >= logging.ERROR or stack_trace is not None:
+                # Report only an error or an exception with a stack trace
+                telemetry.telemetry_writer.add_log(
+                    level, record.msg, tags=tags, stack_trace=stack_trace, count=key_bucket.skipped + 1
+                )
+
+        super().handle(record)
+
+
+class _TelemetryConfig:
+    TELEMETRY_ENABLED = os.getenv("DD_INSTRUMENTATION_TELEMETRY_ENABLED", "true").lower() in ("true", "1")
+    LOG_COLLECTION_ENABLED = TELEMETRY_ENABLED and os.getenv("DD_TELEMETRY_LOG_COLLECTION_ENABLED", "true").lower() in (
+        "true",
+        "1",
+    )
+    TELEMETRY_HEARTBEAT_INTERVAL = float(os.getenv("DD_TELEMETRY_HEARTBEAT_INTERVAL", "60"))
diff --git a/ddtrace/internal/telemetry/writer.py b/ddtrace/internal/telemetry/writer.py
index 71de6b03907..b0ee860f5af 100644
--- a/ddtrace/internal/telemetry/writer.py
+++ b/ddtrace/internal/telemetry/writer.py
@@ -478,8 +478,8 @@ def add_configurations(self, configuration_list):
                     "value": value,
                 }
 
-    def add_log(self, level, message, stack_trace="", tags=None):
-        # type: (TELEMETRY_LOG_LEVEL, str, str, Optional[Dict]) -> None
+    def add_log(self, level, message, stack_trace="", tags=None, count=1):
+        # type: (TELEMETRY_LOG_LEVEL, str, str, Optional[Dict], int) -> None
         """
         Queues log. This event is meant to send library logs to Datadog’s backend through the Telemetry intake.
         This will make support cycles easier and ensure we know about potentially silent issues in libraries.
@@ -499,6 +499,8 @@ 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
+            if count > 1:
+                data["count"] = count
             self._logs.add(data)
 
     def add_gauge_metric(self, namespace, name, value, tags=None):