Fix FlaskInstrumentor exemplars generation for http.server.(request.)duration (#3912)

* Fix FlaskInstrumentor http.server.request.duration exemplars

* Fix FlaskInstrumentor http.server.duration exemplars

* Add unit test

* Changelog

* lint

* Add flask functional test for exemplars

* lint

* More docker-test deps

* More concise

* Refactor metrics_context assignment, update test

---------

Co-authored-by: Riccardo Magliocchetti <[email protected]>

Author: tammy-baylis-swi

CHANGELOG.md

@@ -74,6 +74,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#3796](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3796))
 - `opentelemetry-instrumentation-fastapi`: Fix handling of APIRoute subclasses
   ([#3681](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3681))
+- `opentelemetry-instrumentation-flask`: Fix exemplars generation for `http.server.request.duration` and `http.server.duration` metrics
+  ([#3912](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3912))
 
 ### Added
 

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

@@ -342,6 +342,7 @@ def _rewrapped_app(
     sem_conv_opt_in_mode=_StabilityMode.DEFAULT,
     duration_histogram_new=None,
 ):
+    # pylint: disable=too-many-statements
     def _wrapped_app(wrapped_app_environ, start_response):
         # We want to measure the time for route matching, etc.
         # In theory, we could start the span here and use
@@ -410,6 +411,11 @@ def _start_response(status, response_headers, *args, **kwargs):
         result = wsgi_app(wrapped_app_environ, _start_response)
         if should_trace:
             duration_s = default_timer() - start
+            # Get the span from wrapped_app_environ and re-create context manually
+            # to pass to histogram for exemplars generation
+            span = wrapped_app_environ.get(_ENVIRON_SPAN_KEY)
+            metrics_context = trace.set_span_in_context(span)
+
             if duration_histogram_old:
                 duration_attrs_old = otel_wsgi._parse_duration_attrs(
                     attributes, _StabilityMode.DEFAULT
@@ -418,9 +424,10 @@ def _start_response(status, response_headers, *args, **kwargs):
                 if request_route:
                     # http.target to be included in old semantic conventions
                     duration_attrs_old[HTTP_TARGET] = str(request_route)
-
                 duration_histogram_old.record(
-                    max(round(duration_s * 1000), 0), duration_attrs_old
+                    max(round(duration_s * 1000), 0),
+                    duration_attrs_old,
+                    context=metrics_context,
                 )
             if duration_histogram_new:
                 duration_attrs_new = otel_wsgi._parse_duration_attrs(
@@ -431,7 +438,9 @@ def _start_response(status, response_headers, *args, **kwargs):
                     duration_attrs_new[HTTP_ROUTE] = str(request_route)
 
                 duration_histogram_new.record(
-                    max(duration_s, 0), duration_attrs_new
+                    max(duration_s, 0),
+                    duration_attrs_new,
+                    context=metrics_context,
                 )
         active_requests_counter.add(-1, active_requests_count_attrs)
         return result

instrumentation/opentelemetry-instrumentation-flask/tests/test_programmatic.py

@@ -810,6 +810,48 @@ def test_flask_metrics_excluded_urls_new_semconv(self):
         self.assertTrue(number_data_point_seen)
         self.assertFalse(histogram_data_point_seen)
 
+    def test_duration_histogram_old_record_with_context(self):
+        with patch("opentelemetry.trace.set_span_in_context") as mock_set_span:
+            self.client.get("/hello/123")
+
+            # Verify that trace.set_span_in_context was called for metrics exemplar context
+            # with same trace, span ID as trace
+            mock_set_span.assert_called()
+            call_args = mock_set_span.call_args
+            self.assertEqual(len(call_args[0]), 1)
+            span_arg = call_args[0][0]
+            self.assertIsNotNone(span_arg)
+            finished_spans = self.memory_exporter.get_finished_spans()
+            self.assertEqual(len(finished_spans), 1)
+            finished_span = finished_spans[0]
+            self.assertEqual(
+                span_arg.context.trace_id, finished_span.context.trace_id
+            )
+            self.assertEqual(
+                span_arg.context.span_id, finished_span.context.span_id
+            )
+
+    def test_duration_histogram_new_record_with_context_new_semconv(self):
+        with patch("opentelemetry.trace.set_span_in_context") as mock_set_span:
+            self.client.get("/hello/123")
+
+            # Verify that trace.set_span_in_context was called for metrics exemplar context
+            # with same trace, span ID as trace
+            mock_set_span.assert_called()
+            call_args = mock_set_span.call_args
+            self.assertEqual(len(call_args[0]), 1)
+            span_arg = call_args[0][0]
+            self.assertIsNotNone(span_arg)
+            finished_spans = self.memory_exporter.get_finished_spans()
+            self.assertEqual(len(finished_spans), 1)
+            finished_span = finished_spans[0]
+            self.assertEqual(
+                span_arg.context.trace_id, finished_span.context.trace_id
+            )
+            self.assertEqual(
+                span_arg.context.span_id, finished_span.context.span_id
+            )
+
 
 class TestProgrammaticHooks(InstrumentationTest, WsgiTestBase):
     def setUp(self):

tests/opentelemetry-docker-tests/tests/flask/test_flask_functional.py

@@ -0,0 +1,118 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from flask import Flask
+
+from opentelemetry import metrics as metrics_api
+from opentelemetry.instrumentation.flask import FlaskInstrumentor
+from opentelemetry.sdk.metrics import AlwaysOnExemplarFilter
+from opentelemetry.test.globals_test import (
+    reset_metrics_globals,
+)
+from opentelemetry.test.test_base import TestBase
+from opentelemetry.trace import (
+    INVALID_SPAN_ID,
+    INVALID_TRACE_ID,
+)
+
+
+class TestFunctionalFlask(TestBase):
+    def setUp(self):
+        super().setUp()
+        self.memory_exporter.clear()
+        # This is done because set_meter_provider cannot override the
+        # current meter provider.
+        reset_metrics_globals()
+        (
+            self.meter_provider,
+            self.memory_metrics_reader,
+        ) = self.create_meter_provider(
+            exemplar_filter=AlwaysOnExemplarFilter(),
+        )
+        metrics_api.set_meter_provider(self.meter_provider)
+
+        self._app = Flask(__name__)
+
+        @self._app.route("/test/")
+        def test_route():
+            return "Test response"
+
+        self._client = self._app.test_client()
+
+        FlaskInstrumentor().instrument_app(
+            self._app,
+            meter_provider=self.meter_provider,
+        )
+
+    def tearDown(self):
+        FlaskInstrumentor().uninstrument()
+        super().tearDown()
+
+    def test_duration_metrics_exemplars(self):
+        """Should generate exemplars with trace and span IDs for Flask HTTP requests."""
+        self._client.get("/test/")
+        self._client.get("/test/")
+        self._client.get("/test/")
+
+        metrics_data = self.memory_metrics_reader.get_metrics_data()
+        self.assertIsNotNone(metrics_data)
+        self.assertTrue(len(metrics_data.resource_metrics) > 0)
+
+        duration_metric = None
+        metric_names = []
+        for resource_metric in metrics_data.resource_metrics:
+            for scope_metric in resource_metric.scope_metrics:
+                for metric in scope_metric.metrics:
+                    metric_names.append(metric.name)
+                    if metric.name in [
+                        "http.server.request.duration",
+                        "http.server.duration",
+                    ]:
+                        duration_metric = metric
+                        break
+                if duration_metric:
+                    break
+            if duration_metric:
+                break
+
+        self.assertIsNotNone(duration_metric)
+        data_points = list(duration_metric.data.data_points)
+        self.assertTrue(len(data_points) > 0)
+
+        exemplar_count = 0
+        for data_point in data_points:
+            if hasattr(data_point, "exemplars") and data_point.exemplars:
+                for exemplar in data_point.exemplars:
+                    exemplar_count += 1
+                    # Exemplar has required fields and valid span context
+                    self.assertIsNotNone(exemplar.value)
+                    self.assertIsNotNone(exemplar.time_unix_nano)
+                    self.assertIsNotNone(exemplar.span_id)
+                    self.assertNotEqual(exemplar.span_id, INVALID_SPAN_ID)
+                    self.assertIsNotNone(exemplar.trace_id)
+                    self.assertNotEqual(exemplar.trace_id, INVALID_TRACE_ID)
+
+                    # Trace and span ID of exemplar are part of finished spans
+                    finished_spans = self.memory_exporter.get_finished_spans()
+                    finished_span_ids = [
+                        span.context.span_id for span in finished_spans
+                    ]
+                    finished_trace_ids = [
+                        span.context.trace_id for span in finished_spans
+                    ]
+                    self.assertIn(exemplar.span_id, finished_span_ids)
+                    self.assertIn(exemplar.trace_id, finished_trace_ids)
+
+        # At least one exemplar was generated
+        self.assertGreater(exemplar_count, 0)

tests/opentelemetry-docker-tests/tests/test-requirements.txt

@@ -24,6 +24,7 @@ dockerpty==0.4.1
 docopt==0.6.2
 exceptiongroup==1.2.0
 flaky==3.7.0
+flask==3.0.2
 greenlet==3.0.3
 grpcio==1.63.2
 idna==2.10

tox.ini

@@ -1008,6 +1008,7 @@ deps =
   -e {toxinidir}/instrumentation/opentelemetry-instrumentation-kafka-python
   -e {toxinidir}/instrumentation/opentelemetry-instrumentation-confluent-kafka
   -e {toxinidir}/instrumentation/opentelemetry-instrumentation-dbapi
+  -e {toxinidir}/instrumentation/opentelemetry-instrumentation-flask
   -e {toxinidir}/instrumentation/opentelemetry-instrumentation-mysql
   -e {toxinidir}/instrumentation/opentelemetry-instrumentation-mysqlclient
   -e {toxinidir}/instrumentation/opentelemetry-instrumentation-psycopg
@@ -1019,6 +1020,8 @@ deps =
   -e {toxinidir}/instrumentation/opentelemetry-instrumentation-aiopg
   -e {toxinidir}/instrumentation/opentelemetry-instrumentation-redis
   -e {toxinidir}/instrumentation/opentelemetry-instrumentation-remoulade
+  -e {toxinidir}/instrumentation/opentelemetry-instrumentation-wsgi
+  -e {toxinidir}/util/opentelemetry-util-http
   opentelemetry-exporter-opencensus@{env:CORE_REPO}\#egg=opentelemetry-exporter-opencensus&subdirectory=exporter/opentelemetry-exporter-opencensus
 
 changedir =