Merge branch 'main' into ci-concurrency

Author: xrmx

CHANGELOG.md

@@ -11,6 +11,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+- `opentelemetry-instrumentation-asyncio` Fix duplicate instrumentation.
+  ([[#3383](https://github.com/open-telemetry/opentelemetry-python-contrib/issues/3383)])
+
+### Added
+
+### Fixed
+
+- `opentelemetry-instrumentation` Catch `ModuleNotFoundError` when the library is not installed
+  and log as debug instead of exception
+  ([#3423](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3423))
+
 ## Version 1.32.0/0.53b0 (2025-04-10)
 
 ### Added
@@ -31,9 +42,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#3113](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3113))
 - `opentelemetry-instrumentation-grpc` Fix error when using gprc versions <= 1.50.0 with unix sockets.
   ([[#3393](https://github.com/open-telemetry/opentelemetry-python-contrib/issues/3393)])
+- `opentelemetry-instrumentation-asyncio` Fix duplicate instrumentation.
+  ([[#3383](https://github.com/open-telemetry/opentelemetry-python-contrib/issues/3383)])
 - `opentelemetry-instrumentation-aiokafka` Fix send_and_wait method no headers kwargs error.
   ([[#3332](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3332)])
 
+
 ## Version 1.31.0/0.52b0 (2025-03-12)
 
 ### Added

instrumentation-genai/opentelemetry-instrumentation-google-genai/CHANGELOG.md

@@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Restructure tests to keep in line with repository conventions ([#3344](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3344))
 
+- Fix [bug](https://github.com/open-telemetry/opentelemetry-python-contrib/issues/3416) where
+span attribute `gen_ai.response.finish_reasons` is empty ([#3417](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3417))
+
 ## Version 0.1b0 (2025-03-05)
 
 - Add support for async and streaming.

instrumentation-genai/opentelemetry-instrumentation-google-genai/src/opentelemetry/instrumentation/google_genai/generate_content.py

@@ -252,6 +252,7 @@ def process_response(self, response: GenerateContentResponse):
         # need to be reflected back into the span attributes.
         #
         # See also: TODOS.md.
+        self._update_finish_reasons(response)
         self._maybe_update_token_counts(response)
         self._maybe_update_error_type(response)
         self._maybe_log_response(response)
@@ -275,6 +276,18 @@ def finalize_processing(self):
         self._record_token_usage_metric()
         self._record_duration_metric()
 
+    def _update_finish_reasons(self, response):
+        if not response.candidates:
+            return
+        for candidate in response.candidates:
+            finish_reason = candidate.finish_reason
+            if finish_reason is None:
+                continue
+            finish_reason_str = finish_reason.name.lower().removeprefix(
+                "finish_reason_"
+            )
+            self._finish_reasons_set.add(finish_reason_str)
+
     def _maybe_update_token_counts(self, response: GenerateContentResponse):
         input_tokens = _get_response_property(
             response, "usage_metadata.prompt_token_count"

instrumentation-genai/opentelemetry-instrumentation-google-genai/tests/generate_content/test_finish_reasons.py

@@ -0,0 +1,143 @@
+# 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.
+
+
+from google.genai import types as genai_types
+
+from .base import TestCase
+
+
+class FinishReasonsTestCase(TestCase):
+    def generate_and_get_span_finish_reasons(self):
+        self.client.models.generate_content(
+            model="gemini-2.5-flash-001", contents="Some prompt"
+        )
+        span = self.otel.get_span_named(
+            "generate_content gemini-2.5-flash-001"
+        )
+        assert span is not None
+        assert "gen_ai.response.finish_reasons" in span.attributes
+        return list(span.attributes["gen_ai.response.finish_reasons"])
+
+    def test_single_candidate_with_valid_reason(self):
+        self.configure_valid_response(
+            candidate=genai_types.Candidate(
+                finish_reason=genai_types.FinishReason.STOP
+            )
+        )
+        self.assertEqual(self.generate_and_get_span_finish_reasons(), ["stop"])
+
+    def test_single_candidate_with_safety_reason(self):
+        self.configure_valid_response(
+            candidate=genai_types.Candidate(
+                finish_reason=genai_types.FinishReason.SAFETY
+            )
+        )
+        self.assertEqual(
+            self.generate_and_get_span_finish_reasons(), ["safety"]
+        )
+
+    def test_single_candidate_with_max_tokens_reason(self):
+        self.configure_valid_response(
+            candidate=genai_types.Candidate(
+                finish_reason=genai_types.FinishReason.MAX_TOKENS
+            )
+        )
+        self.assertEqual(
+            self.generate_and_get_span_finish_reasons(), ["max_tokens"]
+        )
+
+    def test_single_candidate_with_no_reason(self):
+        self.configure_valid_response(
+            candidate=genai_types.Candidate(finish_reason=None)
+        )
+        self.assertEqual(self.generate_and_get_span_finish_reasons(), [])
+
+    def test_single_candidate_with_unspecified_reason(self):
+        self.configure_valid_response(
+            candidate=genai_types.Candidate(
+                finish_reason=genai_types.FinishReason.FINISH_REASON_UNSPECIFIED
+            )
+        )
+        self.assertEqual(
+            self.generate_and_get_span_finish_reasons(), ["unspecified"]
+        )
+
+    def test_multiple_candidates_with_valid_reasons(self):
+        self.configure_valid_response(
+            candidates=[
+                genai_types.Candidate(
+                    finish_reason=genai_types.FinishReason.MAX_TOKENS
+                ),
+                genai_types.Candidate(
+                    finish_reason=genai_types.FinishReason.STOP
+                ),
+            ]
+        )
+        self.assertEqual(
+            self.generate_and_get_span_finish_reasons(), ["max_tokens", "stop"]
+        )
+
+    def test_sorts_finish_reasons(self):
+        self.configure_valid_response(
+            candidates=[
+                genai_types.Candidate(
+                    finish_reason=genai_types.FinishReason.STOP
+                ),
+                genai_types.Candidate(
+                    finish_reason=genai_types.FinishReason.MAX_TOKENS
+                ),
+                genai_types.Candidate(
+                    finish_reason=genai_types.FinishReason.SAFETY
+                ),
+            ]
+        )
+        self.assertEqual(
+            self.generate_and_get_span_finish_reasons(),
+            ["max_tokens", "safety", "stop"],
+        )
+
+    def test_deduplicates_finish_reasons(self):
+        self.configure_valid_response(
+            candidates=[
+                genai_types.Candidate(
+                    finish_reason=genai_types.FinishReason.STOP
+                ),
+                genai_types.Candidate(
+                    finish_reason=genai_types.FinishReason.MAX_TOKENS
+                ),
+                genai_types.Candidate(
+                    finish_reason=genai_types.FinishReason.STOP
+                ),
+                genai_types.Candidate(
+                    finish_reason=genai_types.FinishReason.STOP
+                ),
+                genai_types.Candidate(
+                    finish_reason=genai_types.FinishReason.SAFETY
+                ),
+                genai_types.Candidate(
+                    finish_reason=genai_types.FinishReason.STOP
+                ),
+                genai_types.Candidate(
+                    finish_reason=genai_types.FinishReason.STOP
+                ),
+                genai_types.Candidate(
+                    finish_reason=genai_types.FinishReason.STOP
+                ),
+            ]
+        )
+        self.assertEqual(
+            self.generate_and_get_span_finish_reasons(),
+            ["max_tokens", "safety", "stop"],
+        )

instrumentation/opentelemetry-instrumentation-asyncio/src/opentelemetry/instrumentation/asyncio/__init__.py

@@ -93,6 +93,9 @@ def func():
 
 from wrapt import wrap_function_wrapper as _wrap
 
+from opentelemetry.instrumentation.asyncio.instrumentation_state import (
+    _is_instrumented,
+)
 from opentelemetry.instrumentation.asyncio.package import _instruments
 from opentelemetry.instrumentation.asyncio.utils import (
     get_coros_to_trace,
@@ -237,7 +240,12 @@ def wrap_taskgroup_create_task(method, instance, args, kwargs) -> None:
         )
 
     def trace_to_thread(self, func: callable):
-        """Trace a function."""
+        """
+        Trace a function, but if already instrumented, skip double-wrapping.
+        """
+        if _is_instrumented(func):
+            return func
+
         start = default_timer()
         func_name = getattr(func, "__name__", None)
         if func_name is None and isinstance(func, functools.partial):
@@ -270,6 +278,13 @@ def trace_item(self, coro_or_future):
         return coro_or_future
 
     async def trace_coroutine(self, coro):
+        """
+        Wrap a coroutine so that we measure its duration, metrics, etc.
+        If already instrumented, simply 'await coro' to preserve call behavior.
+        """
+        if _is_instrumented(coro):
+            return await coro
+
         if not hasattr(coro, "__name__"):
             return await coro
         start = default_timer()
@@ -303,6 +318,12 @@ async def trace_coroutine(self, coro):
             self.record_process(start, attr, span, exception)
 
     def trace_future(self, future):
+        """
+        Wrap a Future's done callback. If already instrumented, skip re-wrapping.
+        """
+        if _is_instrumented(future):
+            return future
+
         start = default_timer()
         span = (
             self._tracer.start_span(f"{ASYNCIO_PREFIX} future")

instrumentation/opentelemetry-instrumentation-asyncio/src/opentelemetry/instrumentation/asyncio/instrumentation_state.py

@@ -0,0 +1,66 @@
+# 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.
+
+"""
+Instrumentation State Tracker
+
+This module provides helper functions to safely track whether a coroutine,
+Future, or function has already been instrumented by the OpenTelemetry
+asyncio instrumentation layer.
+
+Some Python objects (like coroutines or functions) may not support setting
+custom attributes or weak references. To avoid memory leaks and runtime
+errors, this module uses a WeakKeyDictionary to safely track instrumented
+objects.
+
+If an object cannot be weak-referenced, it is silently skipped.
+
+Usage:
+    if not _is_instrumented(obj):
+        _mark_instrumented(obj)
+        # instrument the object...
+"""
+
+import weakref
+from typing import Any
+
+# A global WeakSet to track instrumented objects.
+# Entries are automatically removed when the objects are garbage collected.
+_instrumented_tasks = weakref.WeakSet()
+
+
+def _is_instrumented(obj: Any) -> bool:
+    """
+    Check whether the object has already been instrumented.
+    If not, mark it as instrumented (only if weakref is supported).
+
+    Args:
+        obj: A coroutine, function, or Future.
+
+    Returns:
+        True if the object was already instrumented.
+        False if the object is not trackable (no weakref support), or just marked now.
+
+    Note:
+        In Python 3.12+, some internal types like `async_generator_asend`
+        raise TypeError when weakref is attempted.
+    """
+    try:
+        if obj in _instrumented_tasks:
+            return True
+        _instrumented_tasks.add(obj)
+        return False
+    except TypeError:
+        # Object doesn't support weak references → can't track instrumentation
+        return False