Merge branch 'main' into genai-utils-api-cleanup

Author: lmolkova

CHANGELOG.md

@@ -13,6 +13,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- `opentelemetry-instrumentation-asgi`: Respect `suppress_http_instrumentation` context in ASGI middleware to skip server span creation when HTTP instrumentation is suppressed
+  ([#4375](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4375))
 - `opentelemetry-instrumentation-confluent-kafka`: Loosen confluent-kafka upper bound to <3.0.0
   ([#4289](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4289))
 - `opentelemetry-instrumentation`: Add support for wrapt 2.x
@@ -40,6 +42,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#4302](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4302))
 - `opentelemetry-instrumentation-grpc`: Fix bidirectional streaming RPCs raising `AttributeError: 'generator' object has no attribute 'add_done_callback'`
   ([#4259](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4259))
+- `opentelemetry-instrumentation-aiokafka`: fix `Unclosed AIOKafkaProducer` warning and `RuntimeWarning: coroutine was never awaited` in tests
+  ([#4384](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4384)) 
+- `opentelemetry-instrumentation-aiokafka`: Fix compatibility with aiokafka 0.13 by calling
+  `_key_serializer`/`_value_serializer` directly instead of the internal `_serialize` method
+  whose signature changed in 0.13 from `(topic, key, value)` to `(key, value, headers)`
+  ([#4379](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4379))
 
 ### Breaking changes
 

instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py

@@ -162,9 +162,17 @@ async def _extract_send_partition(
         key = _extract_send_key(args, kwargs)
         value = _extract_send_value(args, kwargs)
         partition = _extract_argument("partition", 3, None, args, kwargs)
-        key_bytes, value_bytes = cast(
-            "tuple[bytes | None, bytes | None]",
-            instance._serialize(topic, key, value),  # type: ignore[reportUnknownMemberType]
+        key_bytes = cast(
+            "bytes | None",
+            instance._key_serializer(key)  # type: ignore[reportUnknownMemberType]
+            if instance._key_serializer  # type: ignore[reportUnknownMemberType]
+            else key,
+        )
+        value_bytes = cast(
+            "bytes | None",
+            instance._value_serializer(value)  # type: ignore[reportUnknownMemberType]
+            if instance._value_serializer  # type: ignore[reportUnknownMemberType]
+            else value,
         )
         valid_types = (bytes, bytearray, memoryview, type(None))
         if (

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

@@ -1,4 +1,5 @@
-aiokafka==0.11.0
+aiokafka==0.13.0; python_version >= "3.10"
+aiokafka==0.12.0; python_version < "3.10"
 pytest==7.4.4
 -e opentelemetry-instrumentation
 -e instrumentation/opentelemetry-instrumentation-aiokafka

instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_instrumentation.py

@@ -14,9 +14,10 @@
 
 from __future__ import annotations
 
+import sys
 import uuid
 from typing import Any, Sequence, cast
-from unittest import IsolatedAsyncioTestCase, TestCase, mock
+from unittest import IsolatedAsyncioTestCase, TestCase, mock, skipIf
 
 import aiokafka
 from aiokafka import (
@@ -63,6 +64,10 @@ def test_instrument_api(self) -> None:
         )
 
 
+@skipIf(
+    sys.version_info < (3, 10),
+    "aiokafka >= 0.13 requires Python 3.10+",
+)
 class TestAIOKafkaInstrumentation(TestBase, IsolatedAsyncioTestCase):
     @staticmethod
     def consumer_record_factory(
@@ -126,7 +131,7 @@ async def consumer_factory(**consumer_kwargs: Any) -> AIOKafkaConsumer:
 
     @staticmethod
     async def producer_factory() -> AIOKafkaProducer:
-        producer = AIOKafkaProducer(api_version="1.0")
+        producer = AIOKafkaProducer()
 
         producer.client._wait_on_metadata = mock.AsyncMock()
         producer.client.bootstrap = mock.AsyncMock()
@@ -498,37 +503,43 @@ async def test_send_and_wait(self) -> None:
         AIOKafkaInstrumentor().instrument(tracer_provider=self.tracer_provider)
 
         producer = await self.producer_factory()
-        add_message_mock: mock.AsyncMock = (
-            producer._message_accumulator.add_message
-        )
-        add_message_mock.side_effect = [mock.AsyncMock()(), mock.AsyncMock()()]
-
-        tracer = self.tracer_provider.get_tracer(__name__)
-        with tracer.start_as_current_span("test_span") as span:
-            await producer.send_and_wait("topic_1", b"value_1")
+        try:
+            add_message_mock: mock.AsyncMock = (
+                producer._message_accumulator.add_message
+            )
+            add_message_mock.side_effect = [
+                mock.AsyncMock()(),
+                mock.AsyncMock()(),
+            ]
 
-        add_message_mock.assert_awaited_with(
-            TopicPartition(topic="topic_1", partition=1),
-            None,
-            b"value_1",
-            40.0,
-            timestamp_ms=None,
-            headers=[("traceparent", mock.ANY)],
-        )
-        assert (
-            add_message_mock.call_args_list[0]
-            .kwargs["headers"][0][1]
-            .startswith(
-                f"00-{format_trace_id(span.get_span_context().trace_id)}-".encode()
+            tracer = self.tracer_provider.get_tracer(__name__)
+            with tracer.start_as_current_span("test_span") as span:
+                await producer.send_and_wait("topic_1", b"value_1")
+
+            add_message_mock.assert_awaited_with(
+                TopicPartition(topic="topic_1", partition=1),
+                None,
+                b"value_1",
+                40.0,
+                timestamp_ms=None,
+                headers=[("traceparent", mock.ANY)],
+            )
+            assert (
+                add_message_mock.call_args_list[0]
+                .kwargs["headers"][0][1]
+                .startswith(
+                    f"00-{format_trace_id(span.get_span_context().trace_id)}-".encode()
+                )
             )
-        )
 
-        await producer.send_and_wait("topic_2", b"value_2")
-        add_message_mock.assert_awaited_with(
-            TopicPartition(topic="topic_2", partition=1),
-            None,
-            b"value_2",
-            40.0,
-            timestamp_ms=None,
-            headers=[("traceparent", mock.ANY)],
-        )
+            await producer.send_and_wait("topic_2", b"value_2")
+            add_message_mock.assert_awaited_with(
+                TopicPartition(topic="topic_2", partition=1),
+                None,
+                b"value_2",
+                40.0,
+                timestamp_ms=None,
+                headers=[("traceparent", mock.ANY)],
+            )
+        finally:
+            await producer.stop()

instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py

@@ -256,7 +256,7 @@ async def test_wrap_getmany(
         extract_bootstrap_servers: mock.MagicMock,
         _enrich_getmany_poll_span: mock.MagicMock,
         _enrich_getmany_topic_span: mock.MagicMock,
-        _create_consumer_span: mock.MagicMock,
+        _create_consumer_span: mock.AsyncMock,
         extract: mock.MagicMock,
     ) -> None:
         tracer = mock.MagicMock()
@@ -270,6 +270,7 @@ async def test_wrap_getmany(
             }
         )
         kafka_consumer = mock.MagicMock()
+        _create_consumer_span.return_value = mock.MagicMock()
 
         wrapped_getmany = _wrap_getmany(tracer, consume_hook)
         records = await wrapped_getmany(
@@ -370,7 +371,8 @@ async def test_create_consumer_span(
 
     async def test_kafka_properties_extractor(self):
         aiokafka_instance_mock = mock.Mock()
-        aiokafka_instance_mock._serialize.return_value = None, None
+        aiokafka_instance_mock._key_serializer = None
+        aiokafka_instance_mock._value_serializer = None
         aiokafka_instance_mock._partition.return_value = "partition"
         aiokafka_instance_mock.client._wait_on_metadata = mock.AsyncMock()
         assert (

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

@@ -255,7 +255,10 @@ def client_response_hook(span: Span, scope: Scope, message: dict[str, Any]):
 from opentelemetry.instrumentation.propagators import (
     get_global_response_propagator,
 )
-from opentelemetry.instrumentation.utils import _start_internal_or_server_span
+from opentelemetry.instrumentation.utils import (
+    _start_internal_or_server_span,
+    is_http_instrumentation_enabled,
+)
 from opentelemetry.metrics import get_meter
 from opentelemetry.propagators.textmap import Getter, Setter
 from opentelemetry.semconv._incubating.attributes.http_attributes import (
@@ -745,7 +748,10 @@ async def __call__(
             send: An awaitable callable taking a single dictionary as argument.
         """
         start = default_timer()
-        if scope["type"] not in ("http", "websocket"):
+        if not is_http_instrumentation_enabled() or scope["type"] not in (
+            "http",
+            "websocket",
+        ):
             return await self.app(scope, receive, send)
 
         _, _, url = get_host_port_url_tuple(scope)

instrumentation/opentelemetry-instrumentation-asgi/tests/test_asgi_middleware.py

@@ -37,6 +37,7 @@
     get_global_response_propagator,
     set_global_response_propagator,
 )
+from opentelemetry.instrumentation.utils import suppress_http_instrumentation
 from opentelemetry.sdk import resources
 from opentelemetry.sdk.metrics.export import (
     HistogramDataPoint,
@@ -1880,6 +1881,19 @@ async def test_no_excluded_urls(self):
         spans = self.get_finished_spans()
         self.assertGreater(len(spans), 0)
 
+    async def test_suppress_http_instrumentation(self):
+        app = otel_asgi.OpenTelemetryMiddleware(simple_asgi)
+
+        async def suppression_wrapper(scope, receive, send):
+            with suppress_http_instrumentation():
+                await app(scope, receive, send)
+
+        self.seed_app(suppression_wrapper)
+        await self.send_default_request()
+        await self.get_all_output()
+        spans = self.get_finished_spans()
+        self.assertEqual(len(spans), 0)
+
 
 class TestAsgiAttributes(unittest.TestCase):
     def setUp(self):