Make changes to approach

Author: DylanRussell

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

@@ -17,7 +17,7 @@
 import enum
 import logging
 import sys
-import traceback
+from contextvars import ContextVar
 from os import environ, linesep
 from typing import IO, Callable, Optional, Sequence
 
@@ -50,6 +50,10 @@
 _logger = logging.getLogger(__name__)
 _logger.addFilter(DuplicateFilter())
 
+_propagate_false_logger = logging.getLogger(__name__ + ".propagate.false")
+_propagate_false_logger.addFilter(DuplicateFilter())
+_propagate_false_logger.propagate = False
+
 
 class LogExportResult(enum.Enum):
     SUCCESS = 0
@@ -115,26 +119,31 @@ class SimpleLogRecordProcessor(LogRecordProcessor):
     """
 
     def __init__(self, exporter: LogExporter):
+        self._emit_executing = ContextVar("var", default=False)
         self._exporter = exporter
         self._shutdown = False
 
     def on_emit(self, log_data: LogData):
-        # Prevent entering recursive loop.
-        if sum(
-            item.name == "on_emit"
-            and item.filename.endswith("export/__init__.py")
-            for item in traceback.extract_stack()
-        ) > 1:
-            return
-        if self._shutdown:
-            _logger.warning("Processor is already shutdown, ignoring call")
+        # Prevent entering a recursive loop.
+        if self._emit_executing.get():
+            _propagate_false_logger.warning(
+                "SimpleLogRecordProcessor.on_emit has entered a recursive loop. Dropping log and exiting the loop."
+            )
             return
-        token = attach(set_value(_SUPPRESS_INSTRUMENTATION_KEY, True))
+        emit_token = self._emit_executing.set(True)
+        suppress_token = attach(set_value(_SUPPRESS_INSTRUMENTATION_KEY, True))
         try:
-            self._exporter.export((log_data,))
-        except Exception:  # pylint: disable=broad-exception-caught
-            _logger.exception("Exception while exporting logs.")
-        detach(token)
+            if self._shutdown:
+                _logger.warning("Processor is already shutdown, ignoring call")
+                return
+
+            try:
+                self._exporter.export((log_data,))
+            except Exception:  # pylint: disable=broad-exception-caught
+                _logger.exception("Exception while exporting logs.")
+        finally:
+            self._emit_executing.reset(emit_token)
+            detach(suppress_token)
 
     def shutdown(self):
         self._shutdown = True

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

@@ -42,8 +42,8 @@
 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."""
+    Currently applied to `logger`'s on the export logs path to prevent endlessly logging the same log
+    in cases where logging itself is failing."""
 
     def filter(self, record):
         current_log = (
@@ -83,6 +83,9 @@ def shutdown(self):
 
 _logger = logging.getLogger(__name__)
 _logger.addFilter(DuplicateFilter())
+_propagate_false_logger = logging.getLogger(__name__ + ".propagate.false")
+_propagate_false_logger.addFilter(DuplicateFilter())
+_propagate_false_logger.propagate = False
 
 
 class BatchProcessor(Generic[Telemetry]):
@@ -196,15 +199,19 @@ def _export(self, batch_strategy: BatchExportStrategy) -> None:
                     )
                 detach(token)
 
-    # Do not add any logging.log statements to this function, they can be being routed back to this `emit` function,
-    # resulting in endless recursive calls that crash the program.
-    # See https://github.com/open-telemetry/opentelemetry-python/issues/4261
     def emit(self, data: Telemetry) -> None:
         if self._shutdown:
+            _logger.info(
+                "Shutdown called, ignoring %s.", self._exporting
+            )
             return
         if self._pid != os.getpid():
             self._bsp_reset_once.do_once(self._at_fork_reinit)
-        # This will drop a log from the right side if the queue is at _max_queue_length.
+        if len(self._queue) == self._max_queue_size:
+            _logger.warning(
+                "Queue full, dropping %s.", self._exporting
+            )
+        # This will drop a log from the right side if the queue is at _max_queue_size.
         self._queue.appendleft(data)
         if len(self._queue) >= self._max_export_batch_size:
             self._worker_awaken.set()

opentelemetry-sdk/tests/logs/test_export.py

@@ -15,10 +15,10 @@
 # pylint: disable=protected-access
 import logging
 import os
+import random
 import time
 import unittest
 from concurrent.futures import ThreadPoolExecutor
-from sys import version_info
 from typing import Sequence
 from unittest.mock import Mock, patch
 
@@ -63,23 +63,45 @@
 
 
 class TestSimpleLogRecordProcessor(unittest.TestCase):
+    @mark.skipif(
+        version_info=(3, 13),
+        reason="This will fail on 3.13 due to https://github.com/python/cpython/pull/131812 which prevents recursive log messages but was later rolled back.",
+    )
     def test_simple_log_record_processor_doesnt_enter_recursive_loop(self):
         class Exporter(LogExporter):
             def shutdown(self):
                 pass
 
             def export(self, batch: Sequence[LogData]):
-                raise ValueError("Exception raised !")
+                raise ValueError(
+                    "Exception raised ! {}".format(random.randint(1, 10000))
+                )
 
         exporter = Exporter()
         logger_provider = LoggerProvider()
         logger_provider.add_log_record_processor(
             SimpleLogRecordProcessor(exporter)
         )
         root_logger = logging.getLogger()
-        root_logger.addHandler(LoggingHandler(level=logging.NOTSET,logger_provider=logger_provider))
+        # Add the OTLP handler to the root logger like is done in auto instrumentation.
+        # This means logs generated from within SimpleLogRecordProcessor.on_emit are sent back to SimpleLogRecordProcessor.on_emit
+        handler = LoggingHandler(
+            level=logging.DEBUG, logger_provider=logger_provider
+        )
+        root_logger.addHandler(handler)
+        propagate_false_logger = logging.getLogger(
+            "opentelemetry.sdk._logs._internal.export.propagate.false"
+        )
         # This would cause a max recursion depth exceeded error..
-        root_logger.warning("Something is wrong")
+        try:
+            with self.assertLogs(propagate_false_logger) as cm:
+                root_logger.warning("hello!")
+            assert (
+                "SimpleLogRecordProcessor.on_emit has entered a recursive loop"
+                in cm.output[0]
+            )
+        finally:
+            root_logger.removeHandler(handler)
 
     def test_simple_log_record_processor_default_level(self):
         exporter = InMemoryLogExporter()
@@ -403,39 +425,6 @@ def bulk_emit(num_emit):
         time.sleep(2)
         assert len(exporter.get_finished_logs()) == total_expected_logs
 
-    @mark.skipif(
-        version_info < (3, 10),
-        reason="assertNoLogs only exists in python 3.10+.",
-    )
-    def test_logging_lib_not_invoked_in_batch_log_record_emit(self):  # pylint: disable=no-self-use
-        # See https://github.com/open-telemetry/opentelemetry-python/issues/4261
-        exporter = Mock()
-        processor = BatchLogRecordProcessor(exporter)
-        logger_provider = LoggerProvider(
-            resource=SDKResource.create(
-                {
-                    "service.name": "shoppingcart",
-                    "service.instance.id": "instance-12",
-                }
-            ),
-        )
-        logger_provider.add_log_record_processor(processor)
-        handler = LoggingHandler(
-            level=logging.INFO, logger_provider=logger_provider
-        )
-        sdk_logger = logging.getLogger("opentelemetry.sdk")
-        # Attach OTLP handler to SDK logger
-        sdk_logger.addHandler(handler)
-        # If `emit` calls logging.log then this test will throw a maximum recursion depth exceeded exception and fail.
-        try:
-            with self.assertNoLogs(sdk_logger, logging.NOTSET):
-                processor.on_emit(EMPTY_LOG)
-            processor.shutdown()
-            with self.assertNoLogs(sdk_logger, logging.NOTSET):
-                processor.on_emit(EMPTY_LOG)
-        finally:
-            sdk_logger.removeHandler(handler)
-
     def test_args(self):
         exporter = InMemoryLogExporter()
         log_record_processor = BatchLogRecordProcessor(