[WIP] Add Flask enrich_metric_attributes if Labeler contains custom attributes

CHANGELOG.md

@@ -26,6 +26,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - `opentelemetry-instrumentation-sqlalchemy`: implement new semantic convention opt-in migration
   ([#4110](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4110))
 
+### Added
+
+- `opentelemetry-instrumentation`: Add experimental metrics attributes Labeler utility
+  ([#4288](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4288))
+- ``opentelemetry-instrumentation-wsgi`, `opentelemetry-instrumentation-asgi`: `enrich_metric_attributes` with any Labeler-stored attributes in Context for HTTP server duration metrics
+  ([#4300](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4300))
+- `opentelemetry-instrumentation-flask`: `enrich_metric_attributes` with any Labeler-stored attributes in Context
+  ([#4307](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4307))
+
 ### Fixed
 
 - `opentelemetry-docker-tests`: Replace deprecated `SpanAttributes` from `opentelemetry.semconv.trace` with `opentelemetry.semconv._incubating.attributes`

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

@@ -204,6 +204,54 @@ def client_response_hook(span: Span, scope: Scope, message: dict[str, Any]):
 Note:
     The environment variable names used to capture HTTP headers are still experimental, and thus are subject to change.
 
+Custom Metrics Attributes using Labeler
+***************************************
+The ASGI instrumentation reads custom attributes from the labeler (when present)
+and applies them to all HTTP server metric points emitted by the middleware:
+
+- Active requests counter (`http.server.active_requests`)
+- Duration histograms (`http.server.duration` and/or
+    `http.server.request.duration` depending on semantic convention mode)
+- Response size histograms (`http.server.response.size` and/or
+    `http.server.response.body.size`)
+- Request size histograms (`http.server.request.size` and/or
+    `http.server.request.body.size`)
+
+Labeler attributes are request-scoped and merged without overriding base metric
+attributes at the same keys.
+
+
+.. code-block:: python
+
+    .. code-block:: python
+
+    from quart import Quart
+    from opentelemetry.instrumentation._labeler import get_labeler
+    from opentelemetry.instrumentation.asgi import OpenTelemetryMiddleware
+
+    app = Quart(__name__)
+    app.asgi_app = OpenTelemetryMiddleware(app.asgi_app)
+
+    @app.route("/users/<user_id>/")
+    async def user_profile(user_id):
+        # Get the labeler for the current request
+        labeler = get_labeler()
+
+        # Add custom attributes to ASGI instrumentation metrics
+        labeler.add("user_id", user_id)
+        labeler.add("user_type", "registered")
+
+        # Or, add multiple attributes at once
+        labeler.add_attributes({
+            "feature_flag": "new_ui",
+            "experiment_group": "control"
+        })
+
+        return f"User profile for {user_id}"
+
+    if __name__ == "__main__":
+        app.run(debug=True)
+
 API
 ---
 """
@@ -220,6 +268,11 @@ def client_response_hook(span: Span, scope: Scope, message: dict[str, Any]):
 from asgiref.compatibility import guarantee_single_callable
 
 from opentelemetry import context, trace
+from opentelemetry.instrumentation._labeler import (
+    enrich_metric_attributes,
+    get_labeler,
+    get_labeler_attributes,
+)
 from opentelemetry.instrumentation._semconv import (
     HTTP_DURATION_HISTOGRAM_BUCKETS_NEW,
     _filter_semconv_active_request_count_attr,
@@ -758,6 +811,9 @@ async def __call__(
         if self.excluded_urls and self.excluded_urls.url_disabled(url):
             return await self.app(scope, receive, send)
 
+        # Required to create new instance for custom attributes in async context
+        _ = get_labeler()
+
         span_name, additional_attributes = self.default_span_details(scope)
 
         attributes = collect_request_attributes(
@@ -776,6 +832,9 @@ async def __call__(
             attributes,
             self._sem_conv_opt_in_mode,
         )
+        active_requests_count_attrs = enrich_metric_attributes(
+            active_requests_count_attrs
+        )
 
         if scope["type"] == "http":
             self.active_requests_counter.add(1, active_requests_count_attrs)
@@ -806,12 +865,14 @@ async def __call__(
                     span_name, scope, receive
                 )
 
+                labeler_metric_attributes = {}
                 otel_send = self._get_otel_send(
                     current_span,
                     span_name,
                     scope,
                     send,
                     attributes,
+                    labeler_metric_attributes,
                 )
 
                 await self.app(scope, otel_receive, otel_send)
@@ -833,9 +894,21 @@ async def __call__(
                 )
                 if target:
                     duration_attrs_old[HTTP_TARGET] = target
+                duration_attrs_old = enrich_metric_attributes(
+                    duration_attrs_old
+                )
+                for key, value in labeler_metric_attributes.items():
+                    if key not in duration_attrs_old:
+                        duration_attrs_old[key] = value
                 duration_attrs_new = _parse_duration_attrs(
                     attributes, _StabilityMode.HTTP
                 )
+                duration_attrs_new = enrich_metric_attributes(
+                    duration_attrs_new
+                )
+                for key, value in labeler_metric_attributes.items():
+                    if key not in duration_attrs_new:
+                        duration_attrs_new[key] = value
                 span_ctx = set_span_in_context(span)
                 if self.duration_histogram_old:
                     self.duration_histogram_old.record(
@@ -985,13 +1058,17 @@ def _get_otel_send(
         scope,
         send,
         duration_attrs,
+        labeler_metric_attributes,
     ):
         expecting_trailers = False
 
         @wraps(send)
         async def otel_send(message: dict[str, Any]):
             nonlocal expecting_trailers
 
+            if not labeler_metric_attributes:
+                labeler_metric_attributes.update(get_labeler_attributes())
+
             status_code = None
             if message["type"] == "http.response.start":
                 status_code = message["status"]

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

@@ -22,6 +22,7 @@
 
 import opentelemetry.instrumentation.asgi as otel_asgi
 from opentelemetry import trace as trace_api
+from opentelemetry.instrumentation._labeler import clear_labeler, get_labeler
 from opentelemetry.instrumentation._semconv import (
     HTTP_DURATION_HISTOGRAM_BUCKETS_NEW,
     OTEL_SEMCONV_STABILITY_OPT_IN,
@@ -273,6 +274,50 @@ async def background_execution_trailers_asgi(scope, receive, send):
         time.sleep(_SIMULATED_BACKGROUND_TASK_EXECUTION_TIME_S)
 
 
+async def custom_attrs_asgi(scope, receive, send):
+    assert isinstance(scope, dict)
+    assert scope["type"] == "http"
+    labeler = get_labeler()
+    labeler.add("custom_attr", "test_value")
+    labeler.add("http.method", "POST")
+    message = await receive()
+    scope["headers"] = [(b"content-length", b"128")]
+    if message.get("type") == "http.request":
+        await send(
+            {
+                "type": "http.response.start",
+                "status": 200,
+                "headers": [
+                    [b"Content-Type", b"text/plain"],
+                    [b"content-length", b"1024"],
+                ],
+            }
+        )
+        await send({"type": "http.response.body", "body": b"*"})
+
+
+async def custom_attrs_asgi_new_semconv(scope, receive, send):
+    assert isinstance(scope, dict)
+    assert scope["type"] == "http"
+    labeler = get_labeler()
+    labeler.add("custom_attr", "test_value")
+    labeler.add("http.request.method", "POST")
+    message = await receive()
+    scope["headers"] = [(b"content-length", b"128")]
+    if message.get("type") == "http.request":
+        await send(
+            {
+                "type": "http.response.start",
+                "status": 200,
+                "headers": [
+                    [b"Content-Type", b"text/plain"],
+                    [b"content-length", b"1024"],
+                ],
+            }
+        )
+        await send({"type": "http.response.body", "body": b"*"})
+
+
 async def error_asgi(scope, receive, send):
     assert isinstance(scope, dict)
     assert scope["type"] == "http"
@@ -314,6 +359,7 @@ def hook(*_):
 class TestAsgiApplication(AsyncAsgiTestBase):
     def setUp(self):
         super().setUp()
+        clear_labeler()
 
         test_name = ""
         if hasattr(self, "_testMethodName"):
@@ -1554,6 +1600,135 @@ async def test_asgi_metrics_both_semconv(self):
                     self.assertIn(attr, _recommended_attrs_both[metric.name])
         self.assertTrue(number_data_point_seen and histogram_data_point_seen)
 
+    async def test_asgi_metrics_custom_attributes_skip_override(self):
+        app = otel_asgi.OpenTelemetryMiddleware(custom_attrs_asgi)
+        self.seed_app(app)
+        await self.send_default_request()
+        await self.get_all_output()
+
+        metrics = self.get_sorted_metrics(SCOPE)
+        enriched_histogram_metric_names = {
+            "http.server.duration",
+            "http.server.response.size",
+            "http.server.request.size",
+        }
+        active_requests_point_seen = False
+        enriched_histograms_seen = set()
+        for metric in metrics:
+            if metric.name == "http.server.active_requests":
+                data_points = list(metric.data.data_points)
+                self.assertEqual(len(data_points), 1)
+                point = data_points[0]
+                self.assertIsInstance(point, NumberDataPoint)
+                self.assertEqual(point.attributes[HTTP_METHOD], "GET")
+                self.assertNotIn("custom_attr", point.attributes)
+                active_requests_point_seen = True
+                continue
+
+            if metric.name not in enriched_histogram_metric_names:
+                continue
+            data_points = list(metric.data.data_points)
+            self.assertEqual(len(data_points), 1)
+            point = data_points[0]
+            self.assertIsInstance(point, HistogramDataPoint)
+            self.assertEqual(point.attributes[HTTP_METHOD], "GET")
+            self.assertEqual(point.attributes["custom_attr"], "test_value")
+            enriched_histograms_seen.add(metric.name)
+
+        self.assertTrue(active_requests_point_seen)
+        self.assertSetEqual(
+            enriched_histogram_metric_names,
+            enriched_histograms_seen,
+        )
+
+    async def test_asgi_metrics_custom_attributes_skip_override_new_semconv(
+        self,
+    ):
+        app = otel_asgi.OpenTelemetryMiddleware(custom_attrs_asgi_new_semconv)
+        self.seed_app(app)
+        await self.send_default_request()
+        await self.get_all_output()
+
+        metrics = self.get_sorted_metrics(SCOPE)
+        enriched_histogram_metric_names = {
+            "http.server.request.duration",
+            "http.server.response.body.size",
+            "http.server.request.body.size",
+        }
+        active_requests_point_seen = False
+        enriched_histograms_seen = set()
+        for metric in metrics:
+            if metric.name == "http.server.active_requests":
+                data_points = list(metric.data.data_points)
+                self.assertEqual(len(data_points), 1)
+                point = data_points[0]
+                self.assertIsInstance(point, NumberDataPoint)
+                self.assertEqual(point.attributes[HTTP_REQUEST_METHOD], "GET")
+                self.assertNotIn("custom_attr", point.attributes)
+                active_requests_point_seen = True
+                continue
+
+            if metric.name not in enriched_histogram_metric_names:
+                continue
+            data_points = list(metric.data.data_points)
+            self.assertEqual(len(data_points), 1)
+            point = data_points[0]
+            self.assertIsInstance(point, HistogramDataPoint)
+            self.assertEqual(point.attributes[HTTP_REQUEST_METHOD], "GET")
+            self.assertEqual(point.attributes["custom_attr"], "test_value")
+            enriched_histograms_seen.add(metric.name)
+
+        self.assertTrue(active_requests_point_seen)
+        self.assertSetEqual(
+            enriched_histogram_metric_names,
+            enriched_histograms_seen,
+        )
+
+    async def test_asgi_active_requests_attrs_use_enrich_old_semconv(self):
+        with mock.patch(
+            "opentelemetry.instrumentation.asgi.enrich_metric_attributes",
+            wraps=otel_asgi.enrich_metric_attributes,
+        ) as mock_enrich:
+            app = otel_asgi.OpenTelemetryMiddleware(simple_asgi)
+            self.seed_app(app)
+            await self.send_default_request()
+            await self.get_all_output()
+
+        enriched_active_attrs_seen = False
+        for call in mock_enrich.call_args_list:
+            if not call.args:
+                continue
+            attrs = call.args[0]
+            if HTTP_METHOD in attrs and HTTP_STATUS_CODE not in attrs:
+                enriched_active_attrs_seen = True
+                break
+
+        self.assertTrue(enriched_active_attrs_seen)
+
+    async def test_asgi_active_requests_attrs_use_enrich_new_semconv(self):
+        with mock.patch(
+            "opentelemetry.instrumentation.asgi.enrich_metric_attributes",
+            wraps=otel_asgi.enrich_metric_attributes,
+        ) as mock_enrich:
+            app = otel_asgi.OpenTelemetryMiddleware(simple_asgi)
+            self.seed_app(app)
+            await self.send_default_request()
+            await self.get_all_output()
+
+        enriched_active_attrs_seen = False
+        for call in mock_enrich.call_args_list:
+            if not call.args:
+                continue
+            attrs = call.args[0]
+            if (
+                HTTP_REQUEST_METHOD in attrs
+                and HTTP_RESPONSE_STATUS_CODE not in attrs
+            ):
+                enriched_active_attrs_seen = True
+                break
+
+        self.assertTrue(enriched_active_attrs_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)