Add AsgiInstrumentor labeler custom attrs metrics support

Author: tammy-baylis-swi

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

@@ -204,6 +204,39 @@ 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 the HTTP duration metrics recorded by the instrumentation.
+
+.. 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("/user/<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 +253,7 @@ 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 enhance_metric_attributes
 from opentelemetry.instrumentation._semconv import (
     HTTP_DURATION_HISTOGRAM_BUCKETS_NEW,
     _filter_semconv_active_request_count_attr,
@@ -782,11 +816,19 @@ async def __call__(
                 duration_attrs_old = _parse_duration_attrs(
                     attributes, _StabilityMode.DEFAULT
                 )
+                # Enhance attributes with any custom labeler attributes
+                duration_attrs_old = enhance_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 = enhance_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"
@@ -281,6 +341,7 @@ async def error_asgi(scope, receive, send):
 class TestAsgiApplication(AsyncAsgiTestBase):
     def setUp(self):
         super().setUp()
+        clear_labeler()
 
         test_name = ""
         if hasattr(self, "_testMethodName"):
@@ -1245,6 +1306,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)
@@ -1290,6 +1402,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)
@@ -1335,6 +1495,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)