Merge branch 'main' into hectorhdzg/httpxmetrics

Author: xrmx

CHANGELOG.md

@@ -11,6 +11,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+### Added
+
+- `opentelemetry-instrumentation-aiokafka` Add instrumentation of `consumer.getmany` (batch)
+  ([#3257](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3257))
+
 ### Fixed
 
 - `opentelemetry-instrumentation-system-metrics`: fix loading on Google Cloud Run
@@ -19,6 +24,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#3012](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3012))
 - `opentelemetry-instrumentation-starlette` Remove max version constraint on starlette
   ([#3456](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3456))
+- `opentelemetry-instrumentation-starlette` Fix memory leak and double middleware
+  ([#3529](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3529))
 - `opentelemetry-instrumentation-urllib3`: proper bucket boundaries in stable semconv http duration metrics
   ([#3518](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3518))
 - `opentelemetry-instrumentation-urllib`: proper bucket boundaries in stable semconv http duration metrics

instrumentation/opentelemetry-instrumentation-aiokafka/pyproject.toml

@@ -28,6 +28,7 @@ dependencies = [
   "opentelemetry-api ~= 1.27",
   "opentelemetry-instrumentation == 0.55b0.dev",
   "opentelemetry-semantic-conventions == 0.55b0.dev",
+  "typing_extensions ~= 4.1",
 ]
 
 [project.optional-dependencies]

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

@@ -93,15 +93,20 @@ async def produce():
 ___
 """
 
+from __future__ import annotations
+
 from asyncio import iscoroutinefunction
-from typing import Collection
+from typing import TYPE_CHECKING, Collection
 
 import aiokafka
-from wrapt import wrap_function_wrapper
+from wrapt import (
+    wrap_function_wrapper,  # type: ignore[reportUnknownVariableType]
+)
 
 from opentelemetry import trace
 from opentelemetry.instrumentation.aiokafka.package import _instruments
 from opentelemetry.instrumentation.aiokafka.utils import (
+    _wrap_getmany,
     _wrap_getone,
     _wrap_send,
 )
@@ -110,6 +115,21 @@ async def produce():
 from opentelemetry.instrumentation.utils import unwrap
 from opentelemetry.semconv.schemas import Schemas
 
+if TYPE_CHECKING:
+    from typing import TypedDict
+
+    from typing_extensions import Unpack
+
+    from .utils import ConsumeHookT, ProduceHookT
+
+    class InstrumentKwargs(TypedDict, total=False):
+        tracer_provider: trace.TracerProvider
+        async_produce_hook: ProduceHookT
+        async_consume_hook: ConsumeHookT
+
+    class UninstrumentKwargs(TypedDict, total=False):
+        pass
+
 
 class AIOKafkaInstrumentor(BaseInstrumentor):
     """An instrumentor for kafka module
@@ -119,7 +139,7 @@ class AIOKafkaInstrumentor(BaseInstrumentor):
     def instrumentation_dependencies(self) -> Collection[str]:
         return _instruments
 
-    def _instrument(self, **kwargs):
+    def _instrument(self, **kwargs: Unpack[InstrumentKwargs]):
         """Instruments the kafka module
 
         Args:
@@ -155,7 +175,13 @@ def _instrument(self, **kwargs):
             "getone",
             _wrap_getone(tracer, async_consume_hook),
         )
+        wrap_function_wrapper(
+            aiokafka.AIOKafkaConsumer,
+            "getmany",
+            _wrap_getmany(tracer, async_consume_hook),
+        )
 
-    def _uninstrument(self, **kwargs):
+    def _uninstrument(self, **kwargs: Unpack[UninstrumentKwargs]):
         unwrap(aiokafka.AIOKafkaProducer, "send")
         unwrap(aiokafka.AIOKafkaConsumer, "getone")
+        unwrap(aiokafka.AIOKafkaConsumer, "getmany")