Add Labeler and instrumentation integration for custom attribute addition to HTTP request duration metrics (Flask, Django, Falcon, WSGI, ASGI)

CHANGELOG.md

@@ -11,6 +11,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+### Added
+
+- `opentelemetry-instrumentation`, `opentelemetry-instrumentation-flask`, `opentelemetry-instrumentation-wsgi`, `opentelemetry-instrumentation-django`, `opentelemetry-instrumentation-falcon`, `opentelemetry-instrumentation-asgi`: Add Labeler utility. Add FalconInstrumentor, FlaskInstrumentor, DjangoInstrumentor, WsgiInstrumentor, AsgiInstrumentor support of custom attributes merging for HTTP duration metrics.
+  ([#3689](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3689))
+
+
 ## Version 1.37.0/0.58b0 (2025-09-11)
 
 ### Fixed

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

@@ -204,6 +204,46 @@ 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 from a labeler utility that supports adding custom
+attributes to HTTP duration metrics at record time. The custom attributes are
+stored only within the context of an instrumented request or operation. The
+instrumentor does not overwrite base attributes that exist at the same keys as
+any custom attributes.
+
+
+.. 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 +260,10 @@ 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,
+)
 from opentelemetry.instrumentation._semconv import (
     HTTP_DURATION_HISTOGRAM_BUCKETS_NEW,
     _filter_semconv_active_request_count_attr,
@@ -728,6 +772,9 @@ async def __call__(
             receive: An awaitable callable yielding dictionaries
             send: An awaitable callable taking a single dictionary as argument.
         """
+        # Required to create new instance for custom attributes in async context
+        _ = get_labeler()
+
         start = default_timer()
         if scope["type"] not in ("http", "websocket"):
             return await self.app(scope, receive, send)
@@ -809,11 +856,19 @@ async def __call__(
                 duration_attrs_old = _parse_duration_attrs(
                     attributes, _StabilityMode.DEFAULT
                 )
+                # Enhance attributes with any custom labeler attributes
+                duration_attrs_old = enrich_metric_attributes(
+                    duration_attrs_old
+                )
                 if target:
                     duration_attrs_old[SpanAttributes.HTTP_TARGET] = target
                 duration_attrs_new = _parse_duration_attrs(
                     attributes, _StabilityMode.HTTP
                 )
+                # Enhance attributes with any custom labeler attributes
+                duration_attrs_new = enrich_metric_attributes(
+                    duration_attrs_new
+                )
                 if self.duration_histogram_old:
                     self.duration_histogram_old.record(
                         max(round(duration_s * 1000), 0), duration_attrs_old

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,
@@ -108,6 +109,43 @@
     _server_active_requests_count_attrs_old
 )
 
+_server_active_requests_count_attrs_both = (
+    _server_active_requests_count_attrs_old
+)
+_server_active_requests_count_attrs_both.extend(
+    _server_active_requests_count_attrs_new
+)
+
+_custom_attributes = ["custom_attr", "endpoint_type", "feature_flag"]
+_server_duration_attrs_old_with_custom = _server_duration_attrs_old.copy()
+_server_duration_attrs_old_with_custom.append("http.target")
+_server_duration_attrs_old_with_custom.extend(_custom_attributes)
+_server_duration_attrs_new_with_custom = _server_duration_attrs_new.copy()
+_server_duration_attrs_new_with_custom.append("http.route")
+_server_duration_attrs_new_with_custom.extend(_custom_attributes)
+
+_recommended_metrics_attrs_old_with_custom = {
+    "http.server.active_requests": _server_active_requests_count_attrs_old,
+    "http.server.duration": _server_duration_attrs_old_with_custom,
+    "http.server.request.size": _server_duration_attrs_old_with_custom,
+    "http.server.response.size": _server_duration_attrs_old_with_custom,
+}
+_recommended_metrics_attrs_new_with_custom = {
+    "http.server.active_requests": _server_active_requests_count_attrs_new,
+    "http.server.request.duration": _server_duration_attrs_new_with_custom,
+    "http.server.request.body.size": _server_duration_attrs_new_with_custom,
+    "http.server.response.body.size": _server_duration_attrs_new_with_custom,
+}
+_recommended_metrics_attrs_both_with_custom = {
+    "http.server.active_requests": _server_active_requests_count_attrs_both,
+    "http.server.duration": _server_duration_attrs_old_with_custom,
+    "http.server.request.duration": _server_duration_attrs_new_with_custom,
+    "http.server.request.size": _server_duration_attrs_old_with_custom,
+    "http.server.request.body.size": _server_duration_attrs_new_with_custom,
+    "http.server.response.size": _server_duration_attrs_old_with_custom,
+    "http.server.response.body.size": _server_duration_attrs_new_with_custom,
+}
+
 _SIMULATED_BACKGROUND_TASK_EXECUTION_TIME_S = 0.01
 
 
@@ -254,6 +292,28 @@ 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_attributes({"endpoint_type": "test", "feature_flag": True})
+    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"
@@ -292,6 +352,7 @@ def hook(*_):
 class TestAsgiApplication(AsyncAsgiTestBase):
     def setUp(self):
         super().setUp()
+        clear_labeler()
 
         test_name = ""
         if hasattr(self, "_testMethodName"):
@@ -1302,6 +1363,57 @@ async def test_asgi_metrics(self):
                             )
         self.assertTrue(number_data_point_seen and histogram_data_point_seen)
 
+    # pylint: disable=too-many-nested-blocks
+    async def test_asgi_metrics_custom_attributes(self):
+        app = otel_asgi.OpenTelemetryMiddleware(custom_attrs_asgi)
+        self.seed_app(app)
+        await self.send_default_request()
+        await self.get_all_output()
+        self.seed_app(app)
+        await self.send_default_request()
+        await self.get_all_output()
+        self.seed_app(app)
+        await self.send_default_request()
+        await self.get_all_output()
+        metrics_list = self.memory_metrics_reader.get_metrics_data()
+        number_data_point_seen = False
+        histogram_data_point_seen = False
+
+        self.assertTrue(len(metrics_list.resource_metrics) != 0)
+        for resource_metric in metrics_list.resource_metrics:
+            self.assertTrue(len(resource_metric.scope_metrics) != 0)
+            for scope_metric in resource_metric.scope_metrics:
+                self.assertTrue(len(scope_metric.metrics) != 0)
+                self.assertEqual(
+                    scope_metric.scope.name,
+                    "opentelemetry.instrumentation.asgi",
+                )
+                for metric in scope_metric.metrics:
+                    self.assertIn(metric.name, _expected_metric_names_old)
+                    data_points = list(metric.data.data_points)
+                    self.assertEqual(len(data_points), 1)
+                    for point in data_points:
+                        if isinstance(point, HistogramDataPoint):
+                            self.assertEqual(point.count, 3)
+                            histogram_data_point_seen = True
+
+                            for attr in point.attributes:
+                                self.assertIn(
+                                    attr,
+                                    _recommended_metrics_attrs_old_with_custom[
+                                        metric.name
+                                    ],
+                                )
+
+                        if isinstance(point, NumberDataPoint):
+                            number_data_point_seen = True
+
+                            for attr in point.attributes:
+                                self.assertIn(
+                                    attr, _recommended_attrs_old[metric.name]
+                                )
+        self.assertTrue(number_data_point_seen and histogram_data_point_seen)
+
     async def test_asgi_metrics_new_semconv(self):
         # pylint: disable=too-many-nested-blocks
         app = otel_asgi.OpenTelemetryMiddleware(simple_asgi)
@@ -1347,6 +1459,54 @@ async def test_asgi_metrics_new_semconv(self):
                             )
         self.assertTrue(number_data_point_seen and histogram_data_point_seen)
 
+    async def test_asgi_metrics_new_semconv_custom_attributes(self):
+        # pylint: disable=too-many-nested-blocks
+        app = otel_asgi.OpenTelemetryMiddleware(custom_attrs_asgi)
+        self.seed_app(app)
+        await self.send_default_request()
+        await self.get_all_output()
+        self.seed_app(app)
+        await self.send_default_request()
+        await self.get_all_output()
+        self.seed_app(app)
+        await self.send_default_request()
+        await self.get_all_output()
+        metrics_list = self.memory_metrics_reader.get_metrics_data()
+        number_data_point_seen = False
+        histogram_data_point_seen = False
+        self.assertTrue(len(metrics_list.resource_metrics) != 0)
+        for resource_metric in metrics_list.resource_metrics:
+            self.assertTrue(len(resource_metric.scope_metrics) != 0)
+            for scope_metric in resource_metric.scope_metrics:
+                self.assertTrue(len(scope_metric.metrics) != 0)
+                self.assertEqual(
+                    scope_metric.scope.name,
+                    "opentelemetry.instrumentation.asgi",
+                )
+                for metric in scope_metric.metrics:
+                    self.assertIn(metric.name, _expected_metric_names_new)
+                    data_points = list(metric.data.data_points)
+                    self.assertEqual(len(data_points), 1)
+                    for point in data_points:
+                        if isinstance(point, HistogramDataPoint):
+                            self.assertEqual(point.count, 3)
+                            if metric.name == "http.server.request.duration":
+                                self.assertEqual(
+                                    point.explicit_bounds,
+                                    HTTP_DURATION_HISTOGRAM_BUCKETS_NEW,
+                                )
+                            histogram_data_point_seen = True
+                        if isinstance(point, NumberDataPoint):
+                            number_data_point_seen = True
+                        for attr in point.attributes:
+                            self.assertIn(
+                                attr,
+                                _recommended_metrics_attrs_new_with_custom[
+                                    metric.name
+                                ],
+                            )
+        self.assertTrue(number_data_point_seen and histogram_data_point_seen)
+
     async def test_asgi_metrics_both_semconv(self):
         # pylint: disable=too-many-nested-blocks
         app = otel_asgi.OpenTelemetryMiddleware(simple_asgi)
@@ -1392,6 +1552,54 @@ async def test_asgi_metrics_both_semconv(self):
                             )
         self.assertTrue(number_data_point_seen and histogram_data_point_seen)
 
+    async def test_asgi_metrics_both_semconv_custom_attributes(self):
+        # pylint: disable=too-many-nested-blocks
+        app = otel_asgi.OpenTelemetryMiddleware(custom_attrs_asgi)
+        self.seed_app(app)
+        await self.send_default_request()
+        await self.get_all_output()
+        self.seed_app(app)
+        await self.send_default_request()
+        await self.get_all_output()
+        self.seed_app(app)
+        await self.send_default_request()
+        await self.get_all_output()
+        metrics_list = self.memory_metrics_reader.get_metrics_data()
+        number_data_point_seen = False
+        histogram_data_point_seen = False
+        self.assertTrue(len(metrics_list.resource_metrics) != 0)
+        for resource_metric in metrics_list.resource_metrics:
+            self.assertTrue(len(resource_metric.scope_metrics) != 0)
+            for scope_metric in resource_metric.scope_metrics:
+                self.assertTrue(len(scope_metric.metrics) != 0)
+                self.assertEqual(
+                    scope_metric.scope.name,
+                    "opentelemetry.instrumentation.asgi",
+                )
+                for metric in scope_metric.metrics:
+                    self.assertIn(metric.name, _expected_metric_names_both)
+                    data_points = list(metric.data.data_points)
+                    self.assertEqual(len(data_points), 1)
+                    for point in data_points:
+                        if isinstance(point, HistogramDataPoint):
+                            self.assertEqual(point.count, 3)
+                            if metric.name == "http.server.request.duration":
+                                self.assertEqual(
+                                    point.explicit_bounds,
+                                    HTTP_DURATION_HISTOGRAM_BUCKETS_NEW,
+                                )
+                            histogram_data_point_seen = True
+                        if isinstance(point, NumberDataPoint):
+                            number_data_point_seen = True
+                        for attr in point.attributes:
+                            self.assertIn(
+                                attr,
+                                _recommended_metrics_attrs_both_with_custom[
+                                    metric.name
+                                ],
+                            )
+        self.assertTrue(number_data_point_seen and histogram_data_point_seen)
+
     async def test_basic_metric_success(self):
         app = otel_asgi.OpenTelemetryMiddleware(simple_asgi)
         self.seed_app(app)

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

@@ -229,6 +229,40 @@ def response_hook(span, request, response):
 ::
 Enabling this flag will add name of the db driver /*db_driver='django.db.backends.postgresql'*/
 
+
+Custom Metrics Attributes using Labeler
+***************************************
+The Django instrumentation reads from a labeler utility that supports adding custom
+attributes to HTTP duration metrics at record time. The custom attributes are
+stored only within the context of an instrumented request or operation. The
+instrumentor does not overwrite base attributes that exist at the same keys as
+any custom attributes.
+
+
+.. code:: python
+
+    from django.http import HttpResponse
+    from opentelemetry.instrumentation._labeler import get_labeler
+    from opentelemetry.instrumentation.django import DjangoInstrumentor
+
+    DjangoInstrumentor().instrument()
+
+    # Note: urlpattern `/users/<user_id>/` mapped elsewhere
+    def my_user_view(request, user_id):
+        # Get the labeler for the current request
+        labeler = get_labeler()
+
+        # Add custom attributes to Flask 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 HttpResponse("Done!")
+
 API
 ---