Merge branch 'main' into drop-logging-handler-setup-genai

Author: xrmx

CHANGELOG.md

@@ -115,6 +115,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#4175](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4175))
 - `opentelemetry-docker-tests` Fix docker-tests assumption by Postgres-Sqlalchemy case about scope of metrics
   ([#4258](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4258))
+- `opentelemetry-instrumentation-threading`: fix AttributeError when Thread is run without starting
+  ([#4246](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4246))
 
 ### Breaking changes
 

gen-requirements.txt

@@ -7,3 +7,5 @@ requests
 tomli
 tomli_w
 hatch
+# TODO: stick with virtualenv < 21 until a new hatch release
+virtualenv<21

instrumentation-genai/README.md

@@ -3,7 +3,7 @@
 | --------------- | ------------------ | --------------- | -------------- |
 | [opentelemetry-instrumentation-anthropic](./opentelemetry-instrumentation-anthropic) | anthropic >= 0.16.0 | No | development
 | [opentelemetry-instrumentation-claude-agent-sdk](./opentelemetry-instrumentation-claude-agent-sdk) | claude-agent-sdk >= 0.1.14 | No | development
-| [opentelemetry-instrumentation-google-genai](./opentelemetry-instrumentation-google-genai) | google-genai >= 1.0.0 | No | development
+| [opentelemetry-instrumentation-google-genai](./opentelemetry-instrumentation-google-genai) | google-genai >= 1.32.0 | No | development
 | [opentelemetry-instrumentation-langchain](./opentelemetry-instrumentation-langchain) | langchain >= 0.3.21 | No | development
 | [opentelemetry-instrumentation-openai-agents-v2](./opentelemetry-instrumentation-openai-agents-v2) | openai-agents >= 0.3.3 | No | development
 | [opentelemetry-instrumentation-openai-v2](./opentelemetry-instrumentation-openai-v2) | openai >= 1.26.0 | Yes | development

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

@@ -12,4 +12,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-_instruments = ("google-genai >= 1.0.0",)
+_instruments = ("google-genai >= 1.32.0",)

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

@@ -40,6 +40,14 @@
 from google.genai import types
 from vcr.record_mode import RecordMode
 
+try:
+    # These modules are only supported in python >= 3.10
+    from aiohttp.client_exceptions import ClientConnectionError
+    from vcr.stubs import aiohttp_stubs
+except ImportError:
+    ClientConnectionError = None
+    aiohttp_stubs = None
+
 from opentelemetry.instrumentation._semconv import (
     OTEL_SEMCONV_STABILITY_OPT_IN,
     _OpenTelemetrySemanticConventionStability,
@@ -135,6 +143,9 @@ def _redact_headers(headers):
 
 
 def _before_record_request(request):
+    # aiohttp reports the request method in lower case while it is recorded in the cassette in upper case.
+    if request.method:
+        request.method = request.method.upper()
     if request.headers:
         _redact_headers(request.headers)
     uri = request.uri
@@ -316,6 +327,48 @@ def setup_vcr(vcr):
     return vcr
 
 
+@pytest.fixture(name="patch_vcr_aiohttp_stream", scope="module", autouse=True)
+def fixture_patch_vcr_aiohttp_stream():
+    # Allows the async tests to not be stuck in infinite loop when streaming
+    # a VCR cassette with aiohttp stubs.
+    # https://github.com/kevin1024/vcrpy/issues/927
+    if ClientConnectionError is None or aiohttp_stubs is None:
+        return
+
+    class _ReplayMockStream(aiohttp_stubs.MockStream):
+        # Keep vcrpy's stream behavior, but ignore aiohttp's
+        # close-time ClientConnectionError("Connection closed") during
+        # cassette replay, where the full response is already buffered
+        # and this condition should be treated as normal EOF.
+        def set_exception(self, exc):
+            if isinstance(exc, ClientConnectionError) and exc.args == (
+                "Connection closed",
+            ):
+                return
+            super().set_exception(exc)
+
+    class _ReplayMockClientResponse(aiohttp_stubs.MockClientResponse):
+        def __init__(self, *args, **kwargs):
+            super().__init__(*args, **kwargs)
+            self._mock_content_stream = None
+
+        @property
+        def content(self):
+            # vcrpy's aiohttp MockClientResponse.content creates a fresh stream object
+            # on every property access. google-genai async streaming repeatedly reads
+            # response.content.readline() and expects the same stream instance until EOF is
+            # reached.
+            if self._mock_content_stream is None:
+                body = self._body or b""
+                stream = _ReplayMockStream()
+                stream.feed_data(body)
+                stream.feed_eof()
+                self._mock_content_stream = stream
+            return self._mock_content_stream
+
+    aiohttp_stubs.MockClientResponse = _ReplayMockClientResponse
+
+
 @pytest.fixture(name="instrumentor")
 def fixture_instrumentor():
     return GoogleGenAiSdkInstrumentor()

instrumentation-genai/opentelemetry-instrumentation-google-genai/tests/requirements.latest.txt

@@ -40,8 +40,10 @@ pytest==7.4.4
 pytest-asyncio==0.21.0
 pytest-vcr==1.0.2
 
-google-auth==2.38.0
-google-genai==1.32.0
+google-auth==2.47.0
+
+google-genai==1.47.0; python_version < "3.10"
+google-genai==1.64.0; python_version >= "3.10"
 
 # Install locally from the folder. This path is relative to the
 # root directory, given invocation from "tox" at root level.

instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py

@@ -277,6 +277,9 @@ def instrumentation_dependencies(self) -> Collection[str]:
         return _instruments
 
     def _instrument(self, **kwargs):
+        # TODO: should probably wrap methods directly instead of going through
+        # these classes. Hopefully it'll make the patching work if called after
+        # the original classes have already been imported,  #4270
         self._original_kafka_producer = confluent_kafka.Producer
         self._original_kafka_consumer = confluent_kafka.Consumer
 

instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py

@@ -12,11 +12,11 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-# pylint: disable=no-name-in-module
-
-from confluent_kafka import Consumer, Producer
+# pylint: disable=no-name-in-module,import-outside-toplevel
 
 from opentelemetry.instrumentation.confluent_kafka import (
+    AutoInstrumentedConsumer,
+    AutoInstrumentedProducer,
     ConfluentKafkaInstrumentor,
     ProxiedConsumer,
     ProxiedProducer,
@@ -41,6 +41,8 @@
 
 class TestConfluentKafka(TestBase):
     def test_instrument_api(self) -> None:
+        from confluent_kafka import Consumer, Producer  # noqa: PLC0415
+
         instrumentation = ConfluentKafkaInstrumentor()
 
         producer = Producer({"bootstrap.servers": "localhost:29092"})
@@ -51,16 +53,22 @@ def test_instrument_api(self) -> None:
         producer = instrumentation.uninstrument_producer(producer)
         self.assertEqual(producer.__class__, Producer)
 
-        producer = Producer({"bootstrap.servers": "localhost:29092"})
-        producer = instrumentation.instrument_producer(producer)
+        consumer = Consumer(
+            {
+                "bootstrap.servers": "localhost:29092",
+                "group.id": "mygroup",
+                "auto.offset.reset": "earliest",
+            }
+        )
 
-        self.assertEqual(producer.__class__, ProxiedProducer)
+        consumer = instrumentation.instrument_consumer(consumer)
+        self.assertEqual(consumer.__class__, ProxiedConsumer)
 
-        producer = instrumentation.uninstrument_producer(producer)
-        self.assertEqual(producer.__class__, Producer)
+        consumer = instrumentation.uninstrument_consumer(consumer)
+        self.assertEqual(consumer.__class__, Consumer)
 
         consumer = Consumer(
-            {
+            **{
                 "bootstrap.servers": "localhost:29092",
                 "group.id": "mygroup",
                 "auto.offset.reset": "earliest",
@@ -73,7 +81,37 @@ def test_instrument_api(self) -> None:
         consumer = instrumentation.uninstrument_consumer(consumer)
         self.assertEqual(consumer.__class__, Consumer)
 
+    def test_instrument_api_with_instrument(self) -> None:
+        ConfluentKafkaInstrumentor().instrument()
+
+        from confluent_kafka import Consumer, Producer  # noqa: PLC0415
+
+        producer = Producer({"bootstrap.servers": "localhost:29092"})
+        self.assertEqual(producer.__class__, AutoInstrumentedProducer)
+
+        consumer = Consumer(
+            {
+                "bootstrap.servers": "localhost:29092",
+                "group.id": "mygroup",
+                "auto.offset.reset": "earliest",
+            }
+        )
+        self.assertEqual(consumer.__class__, AutoInstrumentedConsumer)
+
+        consumer = Consumer(
+            **{
+                "bootstrap.servers": "localhost:29092",
+                "group.id": "mygroup",
+                "auto.offset.reset": "earliest",
+            }
+        )
+        self.assertEqual(consumer.__class__, AutoInstrumentedConsumer)
+
+        ConfluentKafkaInstrumentor().uninstrument()
+
     def test_consumer_commit_method_exists(self) -> None:
+        from confluent_kafka import Consumer  # noqa: PLC0415
+
         instrumentation = ConfluentKafkaInstrumentor()
 
         consumer = Consumer(

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

@@ -147,7 +147,8 @@ def __wrap_threading_run(
     ) -> R:
         token = None
         try:
-            token = context.attach(instance._otel_context)
+            if hasattr(instance, "_otel_context"):
+                token = context.attach(instance._otel_context)
             return call_wrapped(*args, **kwargs)
         finally:
             if token is not None:

instrumentation/opentelemetry-instrumentation-threading/tests/test_threading.py

@@ -12,8 +12,11 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from __future__ import annotations
+
 import threading
 from concurrent.futures import (  # pylint: disable=no-name-in-module; TODO #4199
+    Future,
     ThreadPoolExecutor,
 )
 from typing import List
@@ -66,7 +69,7 @@ def test_trace_context_propagation_in_thread_pool_with_multiple_workers(
         executor = ThreadPoolExecutor(max_workers=max_workers)
 
         expected_span_contexts: List[trace.SpanContext] = []
-        futures_list = []
+        futures_list: List[Future[trace.SpanContext]] = []
         for num in range(max_workers):
             with self._tracer.start_as_current_span(f"trace_{num}") as span:
                 expected_span_context = span.get_span_context()
@@ -125,23 +128,23 @@ def fake_func(self):
     def get_current_span_context_for_test() -> trace.SpanContext:
         return trace.get_current_span().get_span_context()
 
-    def print_square(self, num):
+    def print_square(self, num: int | float) -> int | float:
         with self._tracer.start_as_current_span("square"):
             return num * num
 
-    def print_cube(self, num):
+    def print_cube(self, num: int | float) -> int | float:
         with self._tracer.start_as_current_span("cube"):
             return num * num * num
 
-    def print_square_with_thread(self, num):
+    def print_square_with_thread(self, num: int | float) -> int | float:
         with self._tracer.start_as_current_span("square"):
             cube_thread = threading.Thread(target=self.print_cube, args=(10,))
 
             cube_thread.start()
             cube_thread.join()
             return num * num
 
-    def calculate(self, num):
+    def calculate(self, num: int | float) -> None:
         with self._tracer.start_as_current_span("calculate"):
             square_thread = threading.Thread(
                 target=self.print_square, args=(num,)
@@ -294,3 +297,48 @@ def test_threadpool_with_valid_context_token(self, mock_detach: MagicMock):
             future = executor.submit(self.get_current_span_context_for_test)
             future.result()
             mock_detach.assert_called_once()
+
+    def test_threading_run_without_start(self):
+        square_thread = threading.Thread(target=self.print_square, args=(10,))
+        with self._tracer.start_as_current_span("root"):
+            square_thread.run()
+
+        spans = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans), 2)
+        root_span = next(span for span in spans if span.name == "root")
+        self.assertIsNotNone(root_span)
+        self.assertIsNone(root_span.parent)
+        square_span = next(span for span in spans if span.name == "square")
+        self.assertIsNotNone(square_span)
+        self.assertIs(square_span.parent, root_span.get_span_context())
+
+    def test_threading_run_with_custom_run(self):
+        _tracer = self._tracer
+
+        class ThreadWithCustomRun(threading.Thread):
+            def run(self):
+                # don't call super().run() on purpose
+                # Thread.run() cannot be called twice
+                with _tracer.start_as_current_span("square"):
+                    pass
+
+        square_thread = ThreadWithCustomRun(
+            target=self.print_square, args=(10,)
+        )
+        with self._tracer.start_as_current_span("run_1"):
+            square_thread.run()
+        with self._tracer.start_as_current_span("run_2"):
+            square_thread.run()
+
+        spans = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans), 4)
+        run_1_span = next(span for span in spans if span.name == "run_1")
+        run_2_span = next(span for span in spans if span.name == "run_2")
+        square_spans = [span for span in spans if span.name == "square"]
+        square_spans.sort(key=lambda x: x.start_time or 0)
+        run_1_child_span = square_spans[0]
+        run_2_child_span = square_spans[1]
+        self.assertIs(run_1_child_span.parent, run_1_span.get_span_context())
+        self.assertIs(run_2_child_span.parent, run_2_span.get_span_context())
+        self.assertIsNone(run_1_span.parent)
+        self.assertIsNone(run_2_span.parent)