Merge branch 'main' into fix/fastapi-memory-leak-only

Author: emdneto

CHANGELOG.md

@@ -15,9 +15,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - `opentelemetry-instrumentation-fastapi`: Fix memory leak in `uninstrument_app()` by properly removing apps from the tracking set
   ([#3688](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3688)
+- `opentelemetry-instrumentation-tornado` Fix server (request) duration metric calculation
+  ([#3679](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3679))
 - `opentelemetry-instrumentation`: Avoid calls to `context.detach` with `None` token.
   ([#3673](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3673))
 
+### Added
+
+- `opentelemetry-instrumentation-confluent-kafka` Add support for confluent-kafka <=2.11.0
+  ([#3685](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3685))
+
 ## Version 1.36.0/0.57b0 (2025-07-29)
 
 ### Fixed
@@ -118,7 +125,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Deprecated
 
 - Drop support for Python 3.8, bump baseline to Python 3.9.
-([#3399](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3399))
+  ([#3399](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3399))
 
 ## Version 1.33.0/0.54b0 (2025-05-09)
 

docs-requirements.txt

@@ -21,7 +21,7 @@ botocore~=1.0
 boto3~=1.0
 cassandra-driver~=3.25
 celery>=4.0
-confluent-kafka>= 1.8.2,<= 2.4.0
+confluent-kafka>= 1.8.2,<= 2.11.0
 elasticsearch>=6.0,<9.0
 flask~=2.0
 falcon~=2.0

instrumentation/README.md

@@ -17,7 +17,7 @@
 | [opentelemetry-instrumentation-cassandra](./opentelemetry-instrumentation-cassandra) | cassandra-driver ~= 3.25,scylla-driver ~= 3.25 | No | development
 | [opentelemetry-instrumentation-celery](./opentelemetry-instrumentation-celery) | celery >= 4.0, < 6.0 | No | development
 | [opentelemetry-instrumentation-click](./opentelemetry-instrumentation-click) | click >= 8.1.3, < 9.0.0 | No | development
-| [opentelemetry-instrumentation-confluent-kafka](./opentelemetry-instrumentation-confluent-kafka) | confluent-kafka >= 1.8.2, <= 2.7.0 | No | development
+| [opentelemetry-instrumentation-confluent-kafka](./opentelemetry-instrumentation-confluent-kafka) | confluent-kafka >= 1.8.2, <= 2.11.0 | No | development
 | [opentelemetry-instrumentation-dbapi](./opentelemetry-instrumentation-dbapi) | dbapi | No | development
 | [opentelemetry-instrumentation-django](./opentelemetry-instrumentation-django) | django >= 1.10 | Yes | development
 | [opentelemetry-instrumentation-elasticsearch](./opentelemetry-instrumentation-elasticsearch) | elasticsearch >= 6.0 | No | development

instrumentation/opentelemetry-instrumentation-confluent-kafka/pyproject.toml

@@ -32,7 +32,7 @@ dependencies = [
 
 [project.optional-dependencies]
 instruments = [
-  "confluent-kafka >= 1.8.2, <= 2.7.0",
+  "confluent-kafka >= 1.8.2, <= 2.11.0",
 ]
 
 [project.entry-points.opentelemetry_instrumentor]

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

@@ -13,4 +13,4 @@
 # limitations under the License.
 
 
-_instruments = ("confluent-kafka >= 1.8.2, <= 2.7.0",)
+_instruments = ("confluent-kafka >= 1.8.2, <= 2.11.0",)

instrumentation/opentelemetry-instrumentation-confluent-kafka/test-requirements.txt

@@ -1,5 +1,5 @@
 asgiref==3.8.1
-confluent-kafka==2.6.1
+confluent-kafka==2.11.0
 Deprecated==1.2.14
 iniconfig==2.0.0
 packaging==24.0

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

@@ -211,6 +211,7 @@ def client_response_hook(span, future):
 
 _logger = getLogger(__name__)
 _TraceContext = namedtuple("TraceContext", ["activation", "span", "token"])
+_HANDLER_STATE_KEY = "_otel_state_key"
 _HANDLER_CONTEXT_KEY = "_otel_trace_context_key"
 _OTEL_PATCHED_KEY = "_otel_patched_key"
 
@@ -402,7 +403,8 @@ def _wrap(cls, method_name, wrapper):
 def _prepare(
     tracer, server_histograms, request_hook, func, handler, args, kwargs
 ):
-    server_histograms[_START_TIME] = default_timer()
+    otel_handler_state = {_START_TIME: default_timer()}
+    setattr(handler, _HANDLER_STATE_KEY, otel_handler_state)
 
     request = handler.request
     if _excluded_urls.url_disabled(request.uri):
@@ -622,9 +624,9 @@ def _record_prepare_metrics(server_histograms, handler):
 
 
 def _record_on_finish_metrics(server_histograms, handler, error=None):
-    elapsed_time = round(
-        (default_timer() - server_histograms[_START_TIME]) * 1000
-    )
+    otel_handler_state = getattr(handler, _HANDLER_STATE_KEY, None) or {}
+    start_time = otel_handler_state.get(_START_TIME, None) or default_timer()
+    elapsed_time = round((default_timer() - start_time) * 1000)
 
     response_size = int(handler._headers.get("Content-Length", 0))
     metric_attributes = _create_metric_attributes(handler)

instrumentation/opentelemetry-instrumentation-tornado/tests/test_metrics_instrumentation.py

@@ -13,8 +13,11 @@
 # limitations under the License.
 
 
+import asyncio
 from timeit import default_timer
 
+import tornado.testing
+
 from opentelemetry.instrumentation.tornado import TornadoInstrumentor
 from opentelemetry.sdk.metrics.export import HistogramDataPoint
 
@@ -165,6 +168,78 @@ def test_basic_metrics(self):
             ),
         )
 
+    @tornado.testing.gen_test
+    async def test_metrics_concurrent_requests(self):
+        """
+        Test that metrics can handle concurrent requests and calculate in an async-safe way.
+        """
+        req1 = self.http_client.fetch(self.get_url("/slow?duration=1.0"))
+        req2 = self.http_client.fetch(self.get_url("/async"))
+        await asyncio.gather(req1, req2)
+
+        metrics = self.get_sorted_metrics()
+        self.assertEqual(len(metrics), 7)
+
+        client_duration = metrics[0]
+        server_duration = metrics[4]
+        self.assertEqual(client_duration.name, "http.client.duration")
+        self.assertEqual(server_duration.name, "http.server.duration")
+
+        # Calculating duration requires tracking state via `_HANDLER_STATE_KEY`, so we want to make sure
+        # duration is calculated properly per request, and doesn't affect concurrent requests.
+        req1_client_duration_data_point = next(
+            dp
+            for dp in client_duration.data.data_points
+            if "/slow" in dp.attributes.get("http.url")
+        )
+        req1_server_duration_data_point = next(
+            dp
+            for dp in server_duration.data.data_points
+            if "/slow" in dp.attributes.get("http.target")
+        )
+        req2_client_duration_data_point = next(
+            dp
+            for dp in client_duration.data.data_points
+            if "/async" in dp.attributes.get("http.url")
+        )
+        req2_server_duration_data_point = next(
+            dp
+            for dp in server_duration.data.data_points
+            if "/async" in dp.attributes.get("http.target")
+        )
+
+        # Server and client durations should be similar (adjusting for msecs vs secs)
+        self.assertAlmostEqual(
+            abs(
+                req1_server_duration_data_point.sum / 1000.0
+                - req1_client_duration_data_point.sum
+            ),
+            0.0,
+            delta=0.01,
+        )
+        self.assertAlmostEqual(
+            abs(
+                req2_server_duration_data_point.sum / 1000.0
+                - req2_client_duration_data_point.sum
+            ),
+            0.0,
+            delta=0.01,
+        )
+
+        # Make sure duration is roughly equivalent to expected (req1/slow) should be around 1 second
+        self.assertAlmostEqual(
+            req1_server_duration_data_point.sum / 1000.0,
+            1.0,
+            delta=0.1,
+            msg="Should have been about 1 second",
+        )
+        self.assertAlmostEqual(
+            req2_server_duration_data_point.sum / 1000.0,
+            0.0,
+            delta=0.1,
+            msg="Should have been really short",
+        )
+
     def test_metric_uninstrument(self):
         self.fetch("/")
         TornadoInstrumentor().uninstrument()

instrumentation/opentelemetry-instrumentation-tornado/tests/tornado_test_app.py

@@ -1,4 +1,5 @@
 # pylint: disable=W0223,R0201
+import asyncio
 import time
 
 import tornado.web
@@ -106,6 +107,13 @@ def get(self):
         self.set_status(200)
 
 
+class SlowHandler(tornado.web.RequestHandler):
+    async def get(self):
+        with self.application.tracer.start_as_current_span("slow"):
+            duration = float(self.get_argument("duration", "1.0"))
+            await asyncio.sleep(duration)
+
+
 class RaiseHTTPErrorHandler(tornado.web.RequestHandler):
     def get(self):
         raise tornado.web.HTTPError(403)
@@ -133,6 +141,7 @@ def make_app(tracer):
             (r"/ping", HealthCheckHandler),
             (r"/test_custom_response_headers", CustomResponseHeaderHandler),
             (r"/raise_403", RaiseHTTPErrorHandler),
+            (r"/slow", SlowHandler),
             (r"/echo_socket", EchoWebSocketHandler),
         ]
     )

opentelemetry-instrumentation/src/opentelemetry/instrumentation/bootstrap_gen.py

@@ -85,7 +85,7 @@
         "instrumentation": "opentelemetry-instrumentation-click==0.58b0.dev",
     },
     {
-        "library": "confluent-kafka >= 1.8.2, <= 2.7.0",
+        "library": "confluent-kafka >= 1.8.2, <= 2.11.0",
         "instrumentation": "opentelemetry-instrumentation-confluent-kafka==0.58b0.dev",
     },
     {