Filter duplicate logs out of some logger's logs that might otherwise endlessly log (#4695)

* Add DuplicateFilter to http/grpc exporter

* Precommit and Changelog

* Fix lint issue. Add comment

* Update CHANGELOG.md

Co-authored-by: Riccardo Magliocchetti <[email protected]>

* Update exporter/opentelemetry-exporter-otlp-proto-common/src/opentelemetry/exporter/otlp/proto/common/_internal/__init__.py

Co-authored-by: Riccardo Magliocchetti <[email protected]>

* Add filter to BatchProcessor class

* Run precommit and add DuplicateFilter to another place..

* Move DuplicateFilter to SDK

* improve changelog entry

* Precommit and comment changes

* Fix broken test

* Precommit

* Fix lint issue

* precommit

* test repro of issue

* add print statements

* undo debug stuff

* Update time to 20s

---------

Co-authored-by: Riccardo Magliocchetti <[email protected]>

Author: DylanRussell

CHANGELOG.md

@@ -12,6 +12,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+- Filter duplicate logs out of some internal `logger`'s logs on the export logs path that might otherwise endlessly log or cause a recursion depth exceeded issue in cases where logging itself results in an exception.
+  ([#4695](https://github.com/open-telemetry/opentelemetry-python/pull/4695)).
 - docs: linked the examples with their github source code location and added Prometheus example
   ([#4728](https://github.com/open-telemetry/opentelemetry-python/pull/4728))
 

exporter/opentelemetry-exporter-otlp-proto-grpc/src/opentelemetry/exporter/otlp/proto/grpc/exporter.py

@@ -59,6 +59,7 @@
     KeyValue,
 )
 from opentelemetry.proto.resource.v1.resource_pb2 import Resource  # noqa: F401
+from opentelemetry.sdk._shared_internal import DuplicateFilter
 from opentelemetry.sdk.environment_variables import (
     OTEL_EXPORTER_OTLP_CERTIFICATE,
     OTEL_EXPORTER_OTLP_CLIENT_CERTIFICATE,
@@ -87,6 +88,8 @@
 )
 _MAX_RETRYS = 6
 logger = getLogger(__name__)
+# This prevents logs generated when a log fails to be written to generate another log which fails to be written etc. etc.
+logger.addFilter(DuplicateFilter())
 SDKDataT = TypeVar("SDKDataT")
 ResourceDataT = TypeVar("ResourceDataT")
 TypingResourceT = TypeVar("TypingResourceT")

exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/_log_exporter/__init__.py

@@ -38,6 +38,7 @@
     LogExporter,
     LogExportResult,
 )
+from opentelemetry.sdk._shared_internal import DuplicateFilter
 from opentelemetry.sdk.environment_variables import (
     OTEL_EXPORTER_OTLP_CERTIFICATE,
     OTEL_EXPORTER_OTLP_CLIENT_CERTIFICATE,
@@ -57,6 +58,8 @@
 from opentelemetry.util.re import parse_env_headers
 
 _logger = logging.getLogger(__name__)
+# This prevents logs generated when a log fails to be written to generate another log which fails to be written etc. etc.
+_logger.addFilter(DuplicateFilter())
 
 
 DEFAULT_COMPRESSION = Compression.NoCompression

opentelemetry-sdk/src/opentelemetry/sdk/_logs/_internal/export/__init__.py

@@ -27,7 +27,7 @@
     set_value,
 )
 from opentelemetry.sdk._logs import LogData, LogRecord, LogRecordProcessor
-from opentelemetry.sdk._shared_internal import BatchProcessor
+from opentelemetry.sdk._shared_internal import BatchProcessor, DuplicateFilter
 from opentelemetry.sdk.environment_variables import (
     OTEL_BLRP_EXPORT_TIMEOUT,
     OTEL_BLRP_MAX_EXPORT_BATCH_SIZE,
@@ -43,6 +43,7 @@
     "Unable to parse value for %s as integer. Defaulting to %s."
 )
 _logger = logging.getLogger(__name__)
+_logger.addFilter(DuplicateFilter())
 
 
 class LogExportResult(enum.Enum):

opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py

@@ -39,6 +39,29 @@
 from opentelemetry.util._once import Once
 
 
+class DuplicateFilter(logging.Filter):
+    """Filter that can be applied to internal `logger`'s.
+
+    Currently applied to `logger`s on the export logs path that could otherwise cause endless logging of errors or a
+    recursion depth exceeded issue in cases where logging itself results in an exception."""
+
+    def filter(self, record):
+        current_log = (
+            record.module,
+            record.levelno,
+            record.msg,
+            # We need to pick a time longer than the OTLP LogExporter timeout
+            # which defaults to 10 seconds, but not pick something so long that
+            # it filters out useful logs.
+            time.time() // 20,
+        )
+        if current_log != getattr(self, "last_log", None):
+            self.last_log = current_log  # pylint: disable=attribute-defined-outside-init
+            return True
+        # False means python's `logging` module will no longer process this log.
+        return False
+
+
 class BatchExportStrategy(enum.Enum):
     EXPORT_ALL = 0
     EXPORT_WHILE_BATCH_EXCEEDS_THRESHOLD = 1
@@ -89,6 +112,7 @@ def __init__(
             daemon=True,
         )
         self._logger = logging.getLogger(__name__)
+        self._logger.addFilter(DuplicateFilter())
         self._exporting = exporting
 
         self._shutdown = False

opentelemetry-sdk/tests/shared_internal/test_batch_processor.py

@@ -14,6 +14,7 @@
 
 # pylint: disable=protected-access
 import gc
+import logging
 import multiprocessing
 import os
 import threading
@@ -33,6 +34,9 @@
 from opentelemetry.sdk._logs.export import (
     BatchLogRecordProcessor,
 )
+from opentelemetry.sdk._shared_internal import (
+    DuplicateFilter,
+)
 from opentelemetry.sdk.trace import ReadableSpan
 from opentelemetry.sdk.trace.export import BatchSpanProcessor
 from opentelemetry.sdk.util.instrumentation import InstrumentationScope
@@ -56,6 +60,7 @@ def __init__(self, export_sleep: int):
         self.num_export_calls = 0
         self.export_sleep = export_sleep
         self._shutdown = False
+        self.sleep_interrupted = False
         self.export_sleep_event = threading.Event()
 
     def export(self, _: list[Any]):
@@ -65,6 +70,7 @@ def export(self, _: list[Any]):
 
         sleep_interrupted = self.export_sleep_event.wait(self.export_sleep)
         if sleep_interrupted:
+            self.sleep_interrupted = True
             raise ValueError("Did not get to finish !")
 
     def shutdown(self):
@@ -219,7 +225,7 @@ def test_record_processor_is_garbage_collected(
         assert weak_ref() is None
 
     def test_shutdown_allows_1_export_to_finish(
-        self, batch_processor_class, telemetry, caplog
+        self, batch_processor_class, telemetry
     ):
         # This exporter throws an exception if it's export sleep cannot finish.
         exporter = MockExporterForTesting(export_sleep=2)
@@ -244,5 +250,15 @@ def test_shutdown_allows_1_export_to_finish(
         time.sleep(0.1)
         assert processor._batch_processor._worker_thread.is_alive() is False
         # Expect the second call to be interrupted by shutdown, and the third call to never be made.
-        assert "Exception while exporting" in caplog.text
+        assert exporter.sleep_interrupted is True
         assert 2 == exporter.num_export_calls
+
+
+class TestCommonFuncs(unittest.TestCase):
+    def test_duplicate_logs_filter_works(self):
+        test_logger = logging.getLogger("testLogger")
+        test_logger.addFilter(DuplicateFilter())
+        with self.assertLogs("testLogger") as cm:
+            test_logger.info("message")
+            test_logger.info("message")
+        self.assertEqual(len(cm.output), 1)