Merge branch 'main' into openai-v2-hook-and-fixes

lmolkova · web-flow · commit 1d3f5e9d93ef · 2026-03-16T16:55:06.000-07:00
diff --git a/.github/workflows/generate_workflows_lib/src/generate_workflows_lib/lint.yml.j2 b/.github/workflows/generate_workflows_lib/src/generate_workflows_lib/lint.yml.j2
@@ -5,9 +5,8 @@ name: Lint {{ file_number }}
 
 on:
   push:
-    branches-ignore:
-    - 'release/*'
-    - 'otelbot/*'
+    branches:
+    - 'main'
   pull_request:
 
 permissions:
diff --git a/.github/workflows/generate_workflows_lib/src/generate_workflows_lib/misc.yml.j2 b/.github/workflows/generate_workflows_lib/src/generate_workflows_lib/misc.yml.j2
@@ -5,9 +5,8 @@ name: Misc {{ file_number }}
 
 on:
   push:
-    branches-ignore:
-    - 'release/*'
-    - 'otelbot/*'
+    branches:
+    - 'main'
   pull_request:
 
 permissions:
diff --git a/.github/workflows/generate_workflows_lib/src/generate_workflows_lib/test.yml.j2 b/.github/workflows/generate_workflows_lib/src/generate_workflows_lib/test.yml.j2
@@ -5,9 +5,8 @@ name: Test {{ file_number }}
 
 on:
   push:
-    branches-ignore:
-    - 'release/*'
-    - 'otelbot/*'
+    branches:
+    - 'main'
   pull_request:
 
 permissions:
diff --git a/.github/workflows/lint_0.yml b/.github/workflows/lint_0.yml
@@ -5,9 +5,8 @@ name: Lint 0
 
 on:
   push:
-    branches-ignore:
-    - 'release/*'
-    - 'otelbot/*'
+    branches:
+    - 'main'
   pull_request:
 
 permissions:
diff --git a/.github/workflows/misc_0.yml b/.github/workflows/misc_0.yml
@@ -5,9 +5,8 @@ name: Misc 0
 
 on:
   push:
-    branches-ignore:
-    - 'release/*'
-    - 'otelbot/*'
+    branches:
+    - 'main'
   pull_request:
 
 permissions:
diff --git a/.github/workflows/test_0.yml b/.github/workflows/test_0.yml
@@ -5,9 +5,8 @@ name: Test 0
 
 on:
   push:
-    branches-ignore:
-    - 'release/*'
-    - 'otelbot/*'
+    branches:
+    - 'main'
   pull_request:
 
 permissions:
diff --git a/.github/workflows/test_1.yml b/.github/workflows/test_1.yml
@@ -5,9 +5,8 @@ name: Test 1
 
 on:
   push:
-    branches-ignore:
-    - 'release/*'
-    - 'otelbot/*'
+    branches:
+    - 'main'
   pull_request:
 
 permissions:
diff --git a/.github/workflows/test_2.yml b/.github/workflows/test_2.yml
@@ -5,9 +5,8 @@ name: Test 2
 
 on:
   push:
-    branches-ignore:
-    - 'release/*'
-    - 'otelbot/*'
+    branches:
+    - 'main'
   pull_request:
 
 permissions:
diff --git a/.github/workflows/test_3.yml b/.github/workflows/test_3.yml
@@ -5,9 +5,8 @@ name: Test 3
 
 on:
   push:
-    branches-ignore:
-    - 'release/*'
-    - 'otelbot/*'
+    branches:
+    - 'main'
   pull_request:
 
 permissions:
diff --git a/CHANGELOG.md b/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
 
diff --git a/instrumentation/opentelemetry-instrumentation-logging/src/opentelemetry/instrumentation/logging/handler.py b/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:
         """
diff --git a/instrumentation/opentelemetry-instrumentation-logging/tests/test_logging_handler_recursion.py b/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)
diff --git a/util/opentelemetry-util-genai/CHANGELOG.md b/util/opentelemetry-util-genai/CHANGELOG.md
@@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Populate schema_url on metrics
   ([#4320](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4320))
+- Add workflow invocation type to genAI utils
+  ([https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4310](#4310))
 - `TelemetryHandler` now accepts a `completion_hook` parameter and calls it after each LLM invocation, passing inputs, outputs, the active span, and the log record. Content capture is enabled automatically when a real hook is configured. 
   ([#4315](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4315))
 
diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/types.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/types.py
@@ -203,6 +203,23 @@ class GenAIInvocation:
     attributes: dict[str, Any] = field(default_factory=_new_str_any_dict)
 
 
+@dataclass
+class WorkflowInvocation(GenAIInvocation):
+    """
+    Represents predetermined static sequence of operations eg: Agent, LLM, tool, and retrieval invocations.
+    A workflow groups multiple operations together, accepting input(s) and producing final output(s).
+    """
+
+    name: str = ""
+    operation_name: str = "invoke_workflow"
+    input_messages: list[InputMessage] = field(
+        default_factory=_new_input_messages
+    )
+    output_messages: list[OutputMessage] = field(
+        default_factory=_new_output_messages
+    )
+
+
 @dataclass
 class LLMInvocation(GenAIInvocation):
     """
diff --git a/util/opentelemetry-util-genai/tests/test_workflow_invocation.py b/util/opentelemetry-util-genai/tests/test_workflow_invocation.py
@@ -0,0 +1,75 @@
+from opentelemetry.util.genai.types import (
+    InputMessage,
+    OutputMessage,
+    Text,
+    WorkflowInvocation,
+)
+
+
+class TestWorkflowInvocation:  # pylint: disable=no-self-use
+    def test_default_values(self):
+        invocation = WorkflowInvocation()
+        assert invocation.name == ""
+        assert invocation.operation_name == "invoke_workflow"
+        assert not invocation.input_messages
+        assert not invocation.output_messages
+        assert invocation.span is None
+        assert invocation.context_token is None
+        assert not invocation.attributes
+
+    def test_custom_name(self):
+        invocation = WorkflowInvocation(name="customer_support_pipeline")
+        assert invocation.name == "customer_support_pipeline"
+
+    def test_custom_operation_name(self):
+        invocation = WorkflowInvocation(operation_name="run_pipeline")
+        assert invocation.operation_name == "run_pipeline"
+
+    def test_with_input_messages(self):
+        msg = InputMessage(role="user", parts=[Text(content="hello")])
+        invocation = WorkflowInvocation(input_messages=[msg])
+        assert len(invocation.input_messages) == 1
+        assert invocation.input_messages[0].role == "user"
+
+    def test_with_output_messages(self):
+        msg = OutputMessage(
+            role="assistant", parts=[Text(content="hi")], finish_reason="stop"
+        )
+        invocation = WorkflowInvocation(output_messages=[msg])
+        assert len(invocation.output_messages) == 1
+        assert invocation.output_messages[0].finish_reason == "stop"
+
+    def test_inherits_genai_invocation(self):
+        invocation = WorkflowInvocation(attributes={"key": "value"})
+        assert invocation.attributes == {"key": "value"}
+
+    def test_default_lists_are_independent(self):
+        """Ensure default factory creates separate list instances."""
+        inv1 = WorkflowInvocation()
+        inv2 = WorkflowInvocation()
+        inv1.input_messages.append(InputMessage(role="user", parts=[]))
+        assert len(inv2.input_messages) == 0
+
+    def test_default_attributes_are_independent(self):
+        inv1 = WorkflowInvocation()
+        inv2 = WorkflowInvocation()
+        inv1.attributes["foo"] = "bar"
+        assert "foo" not in inv2.attributes
+
+    def test_full_construction(self):
+        inp = InputMessage(role="user", parts=[Text(content="query")])
+        out = OutputMessage(
+            role="assistant",
+            parts=[Text(content="answer")],
+            finish_reason="stop",
+        )
+        invocation = WorkflowInvocation(
+            name="my_workflow",
+            operation_name="invoke_workflow",
+            input_messages=[inp],
+            output_messages=[out],
+        )
+        assert invocation.name == "my_workflow"
+        assert len(invocation.input_messages) == 1
+        assert len(invocation.output_messages) == 1
+        assert invocation.output_messages[0].parts[0].content == "answer"
