WIP: Support exemplars in WSGI and ASGI histogram

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

@@ -814,25 +814,34 @@ async def __call__(
                 duration_attrs_new = _parse_duration_attrs(
                     attributes, _StabilityMode.HTTP
                 )
+                span_ctx = set_span_in_context(span)
                 if self.duration_histogram_old:
                     self.duration_histogram_old.record(
-                        max(round(duration_s * 1000), 0), duration_attrs_old
+                        max(round(duration_s * 1000), 0),
+                        duration_attrs_old,
+                        context=span_ctx,
                     )
                 if self.duration_histogram_new:
                     self.duration_histogram_new.record(
-                        max(duration_s, 0), duration_attrs_new
+                        max(duration_s, 0),
+                        duration_attrs_new,
+                        context=span_ctx,
                     )
                 self.active_requests_counter.add(
                     -1, active_requests_count_attrs
                 )
                 if self.content_length_header:
                     if self.server_response_size_histogram:
                         self.server_response_size_histogram.record(
-                            self.content_length_header, duration_attrs_old
+                            self.content_length_header,
+                            duration_attrs_old,
+                            context=span_ctx,
                         )
                     if self.server_response_body_size_histogram:
                         self.server_response_body_size_histogram.record(
-                            self.content_length_header, duration_attrs_new
+                            self.content_length_header,
+                            duration_attrs_new,
+                            context=span_ctx,
                         )
 
                 request_size = asgi_getter.get(scope, "content-length")
@@ -844,11 +853,15 @@ async def __call__(
                     else:
                         if self.server_request_size_histogram:
                             self.server_request_size_histogram.record(
-                                request_size_amount, duration_attrs_old
+                                request_size_amount,
+                                duration_attrs_old,
+                                context=span_ctx,
                             )
                         if self.server_request_body_size_histogram:
                             self.server_request_body_size_histogram.record(
-                                request_size_amount, duration_attrs_new
+                                request_size_amount,
+                                duration_attrs_new,
+                                context=span_ctx,
                             )
             if token:
                 context.detach(token)

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

@@ -312,6 +312,46 @@ def setUp(self):
 
         self.env_patch.start()
 
+    # Helper to assert exemplars presence across specified histogram metric names.
+    def _assert_exemplars_present(
+        self, metric_names: set[str], context: str = ""
+    ):
+        metrics_list = self.memory_metrics_reader.get_metrics_data()
+        print(metrics_list)
+        found = {name: 0 for name in metric_names}
+        for resource_metric in (
+            getattr(metrics_list, "resource_metrics", []) or []
+        ):
+            for scope_metric in (
+                getattr(resource_metric, "scope_metrics", []) or []
+            ):
+                for metric in getattr(scope_metric, "metrics", []) or []:
+                    if metric.name not in metric_names:
+                        continue
+                    for point in metric.data.data_points:
+                        found[metric.name] += 1
+                        exemplars = getattr(point, "exemplars", None)
+                        self.assertIsNotNone(
+                            exemplars,
+                            msg=f"Expected exemplars list attribute on histogram data point for {metric.name} ({context})",
+                        )
+                        self.assertGreater(
+                            len(exemplars or []),
+                            0,
+                            msg=f"Expected at least one exemplar on histogram data point for {metric.name} ({context}) but none found.",
+                        )
+                        for ex in exemplars or []:
+                            if hasattr(ex, "span_id"):
+                                self.assertNotEqual(ex.span_id, 0)
+                            if hasattr(ex, "trace_id"):
+                                self.assertNotEqual(ex.trace_id, 0)
+        for name, count in found.items():
+            self.assertGreater(
+                count,
+                0,
+                msg=f"Did not encounter any data points for metric {name} while checking exemplars ({context}).",
+            )
+
     # pylint: disable=too-many-locals
     def validate_outputs(
         self,
@@ -1392,6 +1432,40 @@ async def test_asgi_metrics_both_semconv(self):
                             )
         self.assertTrue(number_data_point_seen and histogram_data_point_seen)
 
+    async def test_asgi_metrics_exemplars_expected_old_semconv(self):
+        """Failing test placeholder asserting exemplars should be present for duration histogram (old semconv)."""
+        app = otel_asgi.OpenTelemetryMiddleware(simple_asgi)
+        for _ in range(5):
+            self.seed_app(app)
+            await self.send_default_request()
+            await self.get_all_output()
+        self._assert_exemplars_present(
+            {"http.server.duration"}, context="old semconv"
+        )
+
+    async def test_asgi_metrics_exemplars_expected_new_semconv(self):
+        """Failing test placeholder asserting exemplars should be present for request duration histogram (new semconv)."""
+        app = otel_asgi.OpenTelemetryMiddleware(simple_asgi)
+        for _ in range(5):
+            self.seed_app(app)
+            await self.send_default_request()
+            await self.get_all_output()
+        self._assert_exemplars_present(
+            {"http.server.request.duration"}, context="new semconv"
+        )
+
+    async def test_asgi_metrics_exemplars_expected_both_semconv(self):
+        """Failing test placeholder asserting exemplars should be present for both duration histograms when both semconv modes enabled."""
+        app = otel_asgi.OpenTelemetryMiddleware(simple_asgi)
+        for _ in range(5):
+            self.seed_app(app)
+            await self.send_default_request()
+            await self.get_all_output()
+        self._assert_exemplars_present(
+            {"http.server.duration", "http.server.request.duration"},
+            context="both semconv",
+        )
+
     async def test_basic_metric_success(self):
         app = otel_asgi.OpenTelemetryMiddleware(simple_asgi)
         self.seed_app(app)

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

@@ -695,41 +695,46 @@ def __call__(
                     self._sem_conv_opt_in_mode,
                 )
                 iterable = self.wsgi(environ, start_response)
-                return _end_span_after_iterating(iterable, span, token)
+                return _iterate_and_close_with_span(iterable, span, token)
         except Exception as ex:
             if _report_new(self._sem_conv_opt_in_mode):
                 req_attrs[ERROR_TYPE] = type(ex).__qualname__
                 if span.is_recording():
                     span.set_attribute(ERROR_TYPE, type(ex).__qualname__)
                 span.set_status(Status(StatusCode.ERROR, str(ex)))
-            span.end()
-            if token is not None:
-                context.detach(token)
             raise
         finally:
             duration_s = default_timer() - start
+            active_metric_ctx = trace.set_span_in_context(span)
             if self.duration_histogram_old:
                 duration_attrs_old = _parse_duration_attrs(
                     req_attrs, _StabilityMode.DEFAULT
                 )
                 self.duration_histogram_old.record(
-                    max(round(duration_s * 1000), 0), duration_attrs_old
+                    max(round(duration_s * 1000), 0),
+                    duration_attrs_old,
+                    context=active_metric_ctx,
                 )
             if self.duration_histogram_new:
                 duration_attrs_new = _parse_duration_attrs(
                     req_attrs, _StabilityMode.HTTP
                 )
                 self.duration_histogram_new.record(
-                    max(duration_s, 0), duration_attrs_new
+                    max(duration_s, 0),
+                    duration_attrs_new,
+                    context=active_metric_ctx,
                 )
             self.active_requests_counter.add(-1, active_requests_count_attrs)
+            span.end()
+            if token is not None:
+                context.detach(token)
 
 
 # Put this in a subfunction to not delay the call to the wrapped
 # WSGI application (instrumentation should change the application
 # behavior as little as possible).
-def _end_span_after_iterating(
-    iterable: Iterable[T], span: trace.Span, token: object
+def _iterate_and_close_with_span(
+    iterable: Iterable[T], span: trace.Span
 ) -> Iterable[T]:
     try:
         with trace.use_span(span):
@@ -738,9 +743,6 @@ def _end_span_after_iterating(
         close = getattr(iterable, "close", None)
         if close:
             close()
-        span.end()
-        if token is not None:
-            context.detach(token)
 
 
 # TODO: inherit from opentelemetry.instrumentation.propagators.Setter

instrumentation/opentelemetry-instrumentation-wsgi/tests/test_wsgi_middleware.py

@@ -291,6 +291,41 @@ def validate_response(
                 expected_attributes[HTTP_REQUEST_METHOD] = http_method
         self.assertEqual(span_list[0].attributes, expected_attributes)
 
+    # Helper modeled after ASGI test suite to assert presence of exemplars on histogram metrics
+    def _assert_exemplars_present(self, metric_names, context=""):
+        metrics_data = self.memory_metrics_reader.get_metrics_data()
+        self.assertTrue(
+            len(metrics_data.resource_metrics) > 0,
+            f"No resource metrics collected while checking exemplars ({context})",
+        )
+        checked = set()
+        for resource_metric in metrics_data.resource_metrics:
+            for scope_metric in resource_metric.scope_metrics:
+                for metric in scope_metric.metrics:
+                    if metric.name not in metric_names:
+                        continue
+                    checked.add(metric.name)
+                    # Expect exactly one datapoint per histogram metric in these tests
+                    data_points = list(metric.data.data_points)
+                    self.assertGreater(
+                        len(data_points),
+                        0,
+                        f"No data points for {metric.name} while checking exemplars ({context})",
+                    )
+                    for point in data_points:
+                        if isinstance(point, HistogramDataPoint):
+                            self.assertGreater(
+                                len(point.exemplars),
+                                0,
+                                f"Expected at least one exemplar on histogram data point for {metric.name} ({context}) but none found.",
+                            )
+        # Ensure we actually saw all targeted metrics
+        self.assertSetEqual(
+            set(metric_names),
+            checked,
+            f"Did not observe all targeted metrics when asserting exemplars ({context}). Expected {metric_names} got {checked}",
+        )
+
     def test_basic_wsgi_call(self):
         app = otel_wsgi.OpenTelemetryMiddleware(simple_wsgi)
         response = app(self.environ, self.start_response)
@@ -415,6 +450,42 @@ def test_wsgi_metrics(self):
                             )
         self.assertTrue(number_data_point_seen and histogram_data_point_seen)
 
+    def test_wsgi_metrics_exemplars_expected_old_semconv(self):  # type: ignore[func-returns-value]
+        """Failing test asserting exemplars should be present for duration histogram (old semconv)."""
+        app = otel_wsgi.OpenTelemetryMiddleware(simple_wsgi)
+        # generate several requests to increase chance of exemplar sampling
+        for _ in range(5):
+            response = app(self.environ, self.start_response)
+            # exhaust response iterable
+            for _ in response:
+                pass
+        self._assert_exemplars_present(
+            {"http.server.duration"}, context="old semconv"
+        )
+
+    def test_wsgi_metrics_exemplars_expected_new_semconv(self):  # type: ignore[func-returns-value]
+        """Failing test asserting exemplars should be present for request duration histogram (new semconv)."""
+        app = otel_wsgi.OpenTelemetryMiddleware(simple_wsgi)
+        for _ in range(5):
+            response = app(self.environ, self.start_response)
+            for _ in response:
+                pass
+        self._assert_exemplars_present(
+            {"http.server.request.duration"}, context="new semconv"
+        )
+
+    def test_wsgi_metrics_exemplars_expected_both_semconv(self):  # type: ignore[func-returns-value]
+        """Failing test asserting exemplars should be present for both duration histograms when both semconv modes enabled."""
+        app = otel_wsgi.OpenTelemetryMiddleware(simple_wsgi)
+        for _ in range(5):
+            response = app(self.environ, self.start_response)
+            for _ in response:
+                pass
+        self._assert_exemplars_present(
+            {"http.server.duration", "http.server.request.duration"},
+            context="both semconv",
+        )
+
     def test_wsgi_metrics_new_semconv(self):
         # pylint: disable=too-many-nested-blocks
         app = otel_wsgi.OpenTelemetryMiddleware(error_wsgi_unhandled)