fix: Add recursion guard in LoggingHandler.emit to prevent deadlock (#4302)

* fix: Add recursion guard in LoggingHandler.emit to prevent deadlock

LoggingHandler.emit() can deadlock when _translate() triggers internal
logging (e.g., _clean_extended_attribute calling _logger.warning() for
invalid attribute types). The warning re-enters the handler chain on the
same thread, creating infinite recursion that prevents the handler lock
from ever being released, blocking all other threads.

This adds a ContextVar-based guard that detects re-entrant calls to
emit() on the same thread and skips the recursive call, preventing the
deadlock while preserving all application logs.

Fixes open-telemetry/opentelemetry-python#3858
Relates to open-telemetry/opentelemetry-python#3560
Relates to open-telemetry/opentelemetry-python#4585

* Update CHANGELOG.md

---------

Co-authored-by: Deepanshu Kartikey <kartikey406@gmail.com>
Co-authored-by: Riccardo Magliocchetti <riccardo.magliocchetti@gmail.com>

Author: 3 people

CHANGELOG.md

@@ -26,6 +26,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#4305](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4305))
 - Don't import module in unwrap if not already imported
   ([#4321](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4321))
+- `opentelemetry-instrumentation-logging`: Add recursion guard in LoggingHandler.emit to prevent deadlock
+  ([#4302](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4302))
 
 ### Breaking changes
 

instrumentation/opentelemetry-instrumentation-logging/src/opentelemetry/instrumentation/logging/handler.py

@@ -18,6 +18,7 @@
 import logging.config
 import threading
 import traceback
+from contextvars import ContextVar
 from time import time_ns
 from typing import Callable
 
@@ -35,6 +36,10 @@
 from opentelemetry.semconv.attributes import exception_attributes
 from opentelemetry.util.types import _ExtendedAttributes
 
+_internal_logger = logging.getLogger(__name__ + ".internal")
+_internal_logger.propagate = False
+_internal_logger.addHandler(logging.StreamHandler())
+
 
 def _setup_logging_handler(
     logger_provider: LoggerProvider, log_code_attributes: bool = False
@@ -112,6 +117,8 @@ class LoggingHandler(logging.Handler):
     https://docs.python.org/3/library/logging.html
     """
 
+    _is_emitting: ContextVar[bool] = ContextVar("_is_emitting", default=False)
+
     def __init__(
         self,
         level: int = logging.NOTSET,
@@ -203,9 +210,28 @@ def emit(self, record: logging.LogRecord) -> None:
 
         The record is translated to OTel format, and then sent across the pipeline.
         """
-        logger = get_logger(record.name, logger_provider=self._logger_provider)
-        if not isinstance(logger, NoOpLogger):
-            logger.emit(self._translate(record))
+        # Prevent recursive logging that can cause infinite recursion or deadlock.
+        # During _translate(), internal OTel code (e.g., _clean_extended_attribute)
+        # may call _logger.warning() for invalid attributes. If the OTel
+        # LoggingHandler is in the logger chain, this warning re-enters emit(),
+        # creating an infinite loop that prevents the handler lock from ever
+        # being released, blocking all other threads.
+        # See: https://github.com/open-telemetry/opentelemetry-python/issues/3858
+
+        if self._is_emitting.get():
+            _internal_logger.warning(
+                "LoggingHandler.emit detected recursive logging, skipping to prevent deadlock."
+            )
+            return
+        token = self._is_emitting.set(True)
+        try:
+            logger = get_logger(
+                record.name, logger_provider=self._logger_provider
+            )
+            if not isinstance(logger, NoOpLogger):
+                logger.emit(self._translate(record))
+        finally:
+            self._is_emitting.reset(token)
 
     def flush(self) -> None:
         """

instrumentation/opentelemetry-instrumentation-logging/tests/test_logging_handler_recursion.py

@@ -0,0 +1,91 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+import threading
+import unittest
+
+from opentelemetry.instrumentation.logging.handler import LoggingHandler
+from opentelemetry.sdk._logs import LoggerProvider
+from opentelemetry.sdk._logs.export import (
+    InMemoryLogExporter,
+    SimpleLogRecordProcessor,
+)
+
+
+class TestLoggingHandlerRecursionGuard(unittest.TestCase):
+    @staticmethod
+    def _create_handler():
+        exporter = InMemoryLogExporter()
+        provider = LoggerProvider()
+        provider.add_log_record_processor(SimpleLogRecordProcessor(exporter))
+        handler = LoggingHandler(logger_provider=provider)
+        return handler, exporter
+
+    @staticmethod
+    def _make_record(msg="test"):
+        return logging.LogRecord(
+            name="test",
+            level=logging.INFO,
+            pathname="",
+            lineno=0,
+            msg=msg,
+            args=(),
+            exc_info=None,
+        )
+
+    def test_recursive_emit_is_skipped(self):
+        """Recursive emit() should be skipped to prevent deadlock.
+        Regression test for https://github.com/open-telemetry/opentelemetry-python/issues/3858
+        """
+        handler, exporter = self._create_handler()
+        token = handler._is_emitting.set(True)
+        handler.emit(self._make_record("should be skipped"))
+        self.assertEqual(len(exporter.get_finished_logs()), 0)
+        handler._is_emitting.reset(token)
+
+    def test_normal_emit_works(self):
+        """Non-recursive emit() should process logs normally."""
+        handler, exporter = self._create_handler()
+        handler.emit(self._make_record("should be captured"))
+        logs = exporter.get_finished_logs()
+        self.assertEqual(len(logs), 1)
+        self.assertEqual(logs[0].log_record.body, "should be captured")
+
+    def test_is_emitting_resets_after_emit(self):
+        """Context must reset after emit(), allowing subsequent logs."""
+        handler, exporter = self._create_handler()
+        handler.emit(self._make_record("first"))
+        self.assertFalse(handler._is_emitting.get())
+        handler.emit(self._make_record("second"))
+        self.assertEqual(len(exporter.get_finished_logs()), 2)
+
+    def test_is_emitting_is_isolated_across_threads(self):
+        """Emit context on one thread must not block other threads."""
+        handler, exporter = self._create_handler()
+        token = handler._is_emitting.set(True)
+
+        result = [False]
+
+        def log_from_other_thread():
+            handler.emit(self._make_record("from other thread"))
+            result[0] = True
+
+        thread = threading.Thread(target=log_from_other_thread)
+        thread.start()
+        thread.join(timeout=5)
+
+        self.assertTrue(result[0], "Other thread should not be blocked")
+        self.assertEqual(len(exporter.get_finished_logs()), 1)
+        handler._is_emitting.reset(token)