Merge remote-tracking branch 'origin/main' into zhaez/gen-ai-support

Author: ezhang6811

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))
+
 - `opentelemetry-instrumentation-botocore` Add GenAI instrumentation for additional Bedrock models for InvokeModel API
   ([#3419](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3419))
 
@@ -34,9 +45,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/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

instrumentation/opentelemetry-instrumentation-asyncio/tests/test_asyncio_duplicate_instrument.py

@@ -0,0 +1,70 @@
+# 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.
+
+"""
+A general test verifying that when the same Future objects (or coroutines) are
+repeatedly instrumented (for example, via `trace_future`), callback references
+do not leak. In this example, we mimic a typical scenario where a small set of
+Futures might be reused throughout an application's lifecycle.
+"""
+
+import asyncio
+
+from opentelemetry.instrumentation.asyncio import AsyncioInstrumentor
+from opentelemetry.test.test_base import TestBase
+
+
+class TestAsyncioDuplicateInstrument(TestBase):
+    """
+    Tests whether repeated instrumentation of the same Futures leads to
+    exponential callback growth (potential memory leak).
+    """
+
+    def setUp(self):
+        super().setUp()
+        self.loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(self.loop)
+
+        self.instrumentor = AsyncioInstrumentor()
+        self.instrumentor.instrument()
+
+    def tearDown(self):
+        self.instrumentor.uninstrument()
+        self.loop.close()
+        asyncio.set_event_loop(None)
+        super().tearDown()
+
+    def test_duplicate_instrumentation_of_futures(self):
+        """
+        If instrumentor.trace_future is called multiple times on the same Future,
+        we should NOT see an unbounded accumulation of callbacks.
+        """
+        fut1 = asyncio.Future()
+        fut2 = asyncio.Future()
+
+        num_iterations = 10
+        for _ in range(num_iterations):
+            self.instrumentor.trace_future(fut1)
+            self.instrumentor.trace_future(fut2)
+
+        self.assertLessEqual(
+            len(fut1._callbacks),
+            1,
+            f"fut1 has {len(fut1._callbacks)} callbacks. Potential leak!",
+        )
+        self.assertLessEqual(
+            len(fut2._callbacks),
+            1,
+            f"fut2 has {len(fut2._callbacks)} callbacks. Potential leak!",
+        )

instrumentation/opentelemetry-instrumentation-tornado/test-requirements.txt

@@ -1,16 +1,11 @@
 asgiref==3.8.1
-blinker==1.7.0
 certifi==2024.7.4
 charset-normalizer==3.3.2
 click==8.1.7
 Deprecated==1.2.14
-Flask==3.0.2
 http_server_mock==1.7
 idna==3.7
 iniconfig==2.0.0
-itsdangerous==2.1.2
-Jinja2==3.1.6
-MarkupSafe==2.1.5
 packaging==24.0
 pluggy==1.5.0
 py-cpuinfo==9.0.0
@@ -20,7 +15,6 @@ tomli==2.0.1
 tornado==6.4.2
 typing_extensions==4.12.2
 urllib3==2.2.2
-Werkzeug==3.0.6
 wrapt==1.16.0
 zipp==3.19.2
 -e opentelemetry-instrumentation

opentelemetry-instrumentation/src/opentelemetry/instrumentation/auto_instrumentation/_load.py

@@ -82,6 +82,14 @@ def _load_instrumentors(distro):
                 exc.conflict,
             )
             continue
+        except ModuleNotFoundError as exc:
+            # ModuleNotFoundError is raised when the library is not installed
+            # and the instrumentation is not required to be loaded.
+            # See https://github.com/open-telemetry/opentelemetry-python-contrib/issues/3421
+            _logger.debug(
+                "Skipping instrumentation %s: %s", entry_point.name, exc.msg
+            )
+            continue
         except ImportError:
             # in scenarios using the kubernetes operator to do autoinstrumentation some
             # instrumentors (usually requiring binary extensions) may fail to load

opentelemetry-instrumentation/tests/auto_instrumentation/test_load.py

@@ -326,11 +326,12 @@ def test_load_instrumentors_dep_conflict(self, iter_mock, mock_logger):  # pylin
             ]
         )
 
+    @patch("opentelemetry.instrumentation.auto_instrumentation._load._logger")
     @patch(
         "opentelemetry.instrumentation.auto_instrumentation._load.entry_points"
     )
     def test_load_instrumentors_import_error_does_not_stop_everything(
-        self, iter_mock
+        self, iter_mock, mock_logger
     ):
         ep_mock1 = Mock(name="instr1")
         ep_mock2 = Mock(name="instr2")
@@ -354,6 +355,12 @@ def test_load_instrumentors_import_error_does_not_stop_everything(
             ]
         )
         self.assertEqual(distro_mock.load_instrumentor.call_count, 2)
+        mock_logger.exception.assert_any_call(
+            "Importing of %s failed, skipping it",
+            ep_mock1.name,
+        )
+
+        mock_logger.debug.assert_any_call("Instrumented %s", ep_mock2.name)
 
     @patch(
         "opentelemetry.instrumentation.auto_instrumentation._load.entry_points"
@@ -382,6 +389,46 @@ def test_load_instrumentors_raises_exception(self, iter_mock):
         )
         self.assertEqual(distro_mock.load_instrumentor.call_count, 1)
 
+    @patch("opentelemetry.instrumentation.auto_instrumentation._load._logger")
+    @patch(
+        "opentelemetry.instrumentation.auto_instrumentation._load.entry_points"
+    )
+    def test_load_instrumentors_module_not_found_error(
+        self, iter_mock, mock_logger
+    ):
+        ep_mock1 = Mock()
+        ep_mock1.name = "instr1"
+
+        ep_mock2 = Mock()
+        ep_mock2.name = "instr2"
+
+        distro_mock = Mock()
+
+        distro_mock.load_instrumentor.side_effect = [
+            ModuleNotFoundError("No module named 'fake_module'"),
+            None,
+        ]
+
+        iter_mock.side_effect = [(), (ep_mock1, ep_mock2), ()]
+
+        _load._load_instrumentors(distro_mock)
+
+        distro_mock.load_instrumentor.assert_has_calls(
+            [
+                call(ep_mock1, raise_exception_on_conflict=True),
+                call(ep_mock2, raise_exception_on_conflict=True),
+            ]
+        )
+        self.assertEqual(distro_mock.load_instrumentor.call_count, 2)
+
+        mock_logger.debug.assert_any_call(
+            "Skipping instrumentation %s: %s",
+            "instr1",
+            "No module named 'fake_module'",
+        )
+
+        mock_logger.debug.assert_any_call("Instrumented %s", ep_mock2.name)
+
     def test_load_instrumentors_no_entry_point_mocks(self):
         distro_mock = Mock()
         _load._load_instrumentors(distro_mock)