Initial commit

Author: DylanRussell

opentelemetry-api/test-requirements.txt

@@ -12,5 +12,5 @@ wrapt==1.16.0
 zipp==3.20.2
 -e opentelemetry-sdk
 -e opentelemetry-semantic-conventions
--e tests/opentelemetry-test-utils
 -e opentelemetry-api
+-e tests/opentelemetry-test-utils

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

@@ -17,6 +17,7 @@
 import enum
 import logging
 import sys
+import traceback
 from os import environ, linesep
 from typing import IO, Callable, Optional, Sequence
 
@@ -118,6 +119,9 @@ def __init__(self, exporter: LogExporter):
         self._shutdown = False
 
     def on_emit(self, log_data: LogData):
+        # Prevent entering recursive loop.
+        if any(item[2] == "on_emit" for item in traceback.extract_stack()):
+            return
         if self._shutdown:
             _logger.warning("Processor is already shutdown, ignoring call")
             return

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

@@ -81,6 +81,10 @@ def shutdown(self):
         raise NotImplementedError
 
 
+_logger = logging.getLogger(__name__)
+_logger.addFilter(DuplicateFilter())
+
+
 class BatchProcessor(Generic[Telemetry]):
     """This class can be used with exporter's that implement the above
     Exporter interface to buffer and send telemetry in batch through
@@ -111,8 +115,6 @@ def __init__(
             target=self.worker,
             daemon=True,
         )
-        self._logger = logging.getLogger(__name__)
-        self._logger.addFilter(DuplicateFilter())
         self._exporting = exporting
 
         self._shutdown = False
@@ -189,7 +191,7 @@ def _export(self, batch_strategy: BatchExportStrategy) -> None:
                         ]
                     )
                 except Exception:  # pylint: disable=broad-exception-caught
-                    self._logger.exception(
+                    _logger.exception(
                         "Exception while exporting %s.", self._exporting
                     )
                 detach(token)

opentelemetry-sdk/test-requirements.txt

@@ -11,7 +11,7 @@ tomli==2.0.1
 typing_extensions==4.10.0
 wrapt==1.16.0
 zipp==3.19.2
--e tests/opentelemetry-test-utils
 -e opentelemetry-api
+-e tests/opentelemetry-test-utils
 -e opentelemetry-semantic-conventions
 -e opentelemetry-sdk

opentelemetry-sdk/tests/logs/test_export.py

@@ -19,6 +19,7 @@
 import unittest
 from concurrent.futures import ThreadPoolExecutor
 from sys import version_info
+from typing import Sequence
 from unittest.mock import Mock, patch
 
 from pytest import mark
@@ -36,6 +37,7 @@
     BatchLogRecordProcessor,
     ConsoleLogExporter,
     InMemoryLogExporter,
+    LogExporter,
     SimpleLogRecordProcessor,
 )
 from opentelemetry.sdk.environment_variables import (
@@ -61,6 +63,24 @@
 
 
 class TestSimpleLogRecordProcessor(unittest.TestCase):
+    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 !")
+
+        exporter = Exporter()
+        logger_provider = LoggerProvider()
+        logger_provider.add_log_record_processor(
+            SimpleLogRecordProcessor(exporter)
+        )
+        logger = logging.getLogger("default_level")
+        logger.addHandler(LoggingHandler(logger_provider=logger_provider))
+        # This would cause a max recursion depth exceeded error..
+        logger.warning("Something is wrong")
+
     def test_simple_log_record_processor_default_level(self):
         exporter = InMemoryLogExporter()
         logger_provider = LoggerProvider()