Metric instrumentation fastapi

CHANGELOG.md

@@ -17,6 +17,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - `opentelemetry-instrumentation-redis` add support to instrument RedisCluster clients
   ([#1177](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1177))
 - `opentelemetry-instrumentation-sqlalchemy` Added span for the connection phase ([#1133](https://github.com/open-telemetry/opentelemetry-python-contrib/issues/1133))
+- Add metric instrumentation in asgi
+  ([#1197](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1197))
+- Add metric instrumentation in fastapi
+  ([#1199](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1199))
 
 ## [1.12.0rc2-0.32b0](https://github.com/open-telemetry/opentelemetry-python/releases/tag/v1.12.0rc2-0.32b0) - 2022-07-01
 

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

@@ -149,6 +149,7 @@ def client_response_hook(span: Span, message: dict):
 import typing
 import urllib
 from functools import wraps
+from timeit import default_timer
 from typing import Tuple
 
 from asgiref.compatibility import guarantee_single_callable
@@ -162,6 +163,7 @@ def client_response_hook(span: Span, message: dict):
     _start_internal_or_server_span,
     http_status_to_status_code,
 )
+from opentelemetry.metrics import get_meter
 from opentelemetry.propagators.textmap import Getter, Setter
 from opentelemetry.semconv.trace import SpanAttributes
 from opentelemetry.trace import Span, set_span_in_context
@@ -179,6 +181,26 @@ def client_response_hook(span: Span, message: dict):
 _ClientRequestHookT = typing.Optional[typing.Callable[[Span, dict], None]]
 _ClientResponseHookT = typing.Optional[typing.Callable[[Span, dict], None]]
 
+# List of recommended attributes
+_duration_attrs = [
+    SpanAttributes.HTTP_METHOD,
+    SpanAttributes.HTTP_HOST,
+    SpanAttributes.HTTP_SCHEME,
+    SpanAttributes.HTTP_STATUS_CODE,
+    SpanAttributes.HTTP_FLAVOR,
+    SpanAttributes.HTTP_SERVER_NAME,
+    SpanAttributes.NET_HOST_NAME,
+    SpanAttributes.NET_HOST_PORT,
+]
+
+_active_requests_count_attrs = [
+    SpanAttributes.HTTP_METHOD,
+    SpanAttributes.HTTP_HOST,
+    SpanAttributes.HTTP_SCHEME,
+    SpanAttributes.HTTP_FLAVOR,
+    SpanAttributes.HTTP_SERVER_NAME,
+]
+
 
 class ASGIGetter(Getter[dict]):
     def get(
@@ -361,6 +383,22 @@ def get_default_span_details(scope: dict) -> Tuple[str, dict]:
     return span_name, {}
 
 
+def _parse_active_request_count_attrs(req_attrs):
+    active_requests_count_attrs = {}
+    for attr_key in _active_requests_count_attrs:
+        if req_attrs.get(attr_key) is not None:
+            active_requests_count_attrs[attr_key] = req_attrs[attr_key]
+    return active_requests_count_attrs
+
+
+def _parse_duration_attrs(req_attrs):
+    duration_attrs = {}
+    for attr_key in _duration_attrs:
+        if req_attrs.get(attr_key) is not None:
+            duration_attrs[attr_key] = req_attrs[attr_key]
+    return duration_attrs
+
+
 class OpenTelemetryMiddleware:
     """The ASGI application middleware.
 
@@ -391,9 +429,21 @@ def __init__(
         client_request_hook: _ClientRequestHookT = None,
         client_response_hook: _ClientResponseHookT = None,
         tracer_provider=None,
+        meter_provider=None,
     ):
         self.app = guarantee_single_callable(app)
         self.tracer = trace.get_tracer(__name__, __version__, tracer_provider)
+        self.meter = get_meter(__name__, __version__, meter_provider)
+        self.duration_histogram = self.meter.create_histogram(
+            name="http.server.duration",
+            unit="ms",
+            description="measures the duration of the inbound HTTP request",
+        )
+        self.active_requests_counter = self.meter.create_up_down_counter(
+            name="http.server.active_requests",
+            unit="requests",
+            description="measures the number of concurrent HTTP requests that are currently in-flight",
+        )
         self.excluded_urls = excluded_urls
         self.default_span_details = (
             default_span_details or get_default_span_details
@@ -426,12 +476,17 @@ async def __call__(self, scope, receive, send):
             context_carrier=scope,
             context_getter=asgi_getter,
         )
-
+        attributes = collect_request_attributes(scope)
+        attributes.update(additional_attributes)
+        active_requests_count_attrs = _parse_active_request_count_attrs(
+            attributes
+        )
+        duration_attrs = _parse_duration_attrs(attributes)
+        if scope["type"] == "http":
+            self.active_requests_counter.add(1, active_requests_count_attrs)
         try:
             with trace.use_span(span, end_on_exit=True) as current_span:
                 if current_span.is_recording():
-                    attributes = collect_request_attributes(scope)
-                    attributes.update(additional_attributes)
                     for key, value in attributes.items():
                         current_span.set_attribute(key, value)
 
@@ -454,10 +509,18 @@ async def __call__(self, scope, receive, send):
                     span_name,
                     scope,
                     send,
+                    duration_attrs,
                 )
+                start = default_timer()
 
                 await self.app(scope, otel_receive, otel_send)
         finally:
+            if scope["type"] == "http":
+                duration = max(round((default_timer() - start) * 1000), 0)
+                self.duration_histogram.record(duration, duration_attrs)
+                self.active_requests_counter.add(
+                    -1, active_requests_count_attrs
+                )
             if token:
                 context.detach(token)
 
@@ -478,7 +541,9 @@ async def otel_receive():
 
         return otel_receive
 
-    def _get_otel_send(self, server_span, server_span_name, scope, send):
+    def _get_otel_send(
+        self, server_span, server_span_name, scope, send, duration_attrs
+    ):
         @wraps(send)
         async def otel_send(message):
             with self.tracer.start_as_current_span(
@@ -489,6 +554,9 @@ async def otel_send(message):
                 if send_span.is_recording():
                     if message["type"] == "http.response.start":
                         status_code = message["status"]
+                        duration_attrs[
+                            SpanAttributes.HTTP_STATUS_CODE
+                        ] = status_code
                         set_status_code(server_span, status_code)
                         set_status_code(send_span, status_code)
                     elif message["type"] == "websocket.send":

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

@@ -14,6 +14,7 @@
 
 import sys
 import unittest
+from timeit import default_timer
 from unittest import mock
 
 import opentelemetry.instrumentation.asgi as otel_asgi
@@ -24,6 +25,10 @@
     set_global_response_propagator,
 )
 from opentelemetry.sdk import resources
+from opentelemetry.sdk.metrics.export import (
+    HistogramDataPoint,
+    NumberDataPoint,
+)
 from opentelemetry.semconv.trace import SpanAttributes
 from opentelemetry.test.asgitestutil import (
     AsgiTestBase,
@@ -36,6 +41,15 @@
     OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_RESPONSE,
 )
 
+_expected_metric_names = [
+    "http.server.active_requests",
+    "http.server.duration",
+]
+_recommended_attrs = {
+    "http.server.active_requests": otel_asgi._active_requests_count_attrs,
+    "http.server.duration": otel_asgi._duration_attrs,
+}
+
 
 async def http_app(scope, receive, send):
     message = await receive()
@@ -523,6 +537,101 @@ def update_expected_hook_results(expected):
             outputs, modifiers=[update_expected_hook_results]
         )
 
+    def test_asgi_metrics(self):
+        app = otel_asgi.OpenTelemetryMiddleware(simple_asgi)
+        self.seed_app(app)
+        self.send_default_request()
+        self.seed_app(app)
+        self.send_default_request()
+        self.seed_app(app)
+        self.send_default_request()
+        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)
+                for metric in scope_metric.metrics:
+                    self.assertIn(metric.name, _expected_metric_names)
+                    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
+                        if isinstance(point, NumberDataPoint):
+                            number_data_point_seen = True
+                        for attr in point.attributes:
+                            self.assertIn(
+                                attr, _recommended_attrs[metric.name]
+                            )
+        self.assertTrue(number_data_point_seen and histogram_data_point_seen)
+
+    def test_basic_metric_success(self):
+        app = otel_asgi.OpenTelemetryMiddleware(simple_asgi)
+        self.seed_app(app)
+        start = default_timer()
+        self.send_default_request()
+        duration = max(round((default_timer() - start) * 1000), 0)
+        expected_duration_attributes = {
+            "http.method": "GET",
+            "http.host": "127.0.0.1",
+            "http.scheme": "http",
+            "http.flavor": "1.0",
+            "net.host.port": 80,
+            "http.status_code": 200,
+        }
+        expected_requests_count_attributes = {
+            "http.method": "GET",
+            "http.host": "127.0.0.1",
+            "http.scheme": "http",
+            "http.flavor": "1.0",
+        }
+        metrics_list = self.memory_metrics_reader.get_metrics_data()
+        for resource_metric in metrics_list.resource_metrics:
+            for scope_metrics in resource_metric.scope_metrics:
+                for metric in scope_metrics.metrics:
+                    for point in list(metric.data.data_points):
+                        if isinstance(point, HistogramDataPoint):
+                            self.assertDictEqual(
+                                expected_duration_attributes,
+                                dict(point.attributes),
+                            )
+                            self.assertEqual(point.count, 1)
+                            self.assertAlmostEqual(
+                                duration, point.sum, delta=5
+                            )
+                        elif isinstance(point, NumberDataPoint):
+                            self.assertDictEqual(
+                                expected_requests_count_attributes,
+                                dict(point.attributes),
+                            )
+                            self.assertEqual(point.value, 0)
+
+    def test_no_metric_for_websockets(self):
+        self.scope = {
+            "type": "websocket",
+            "http_version": "1.1",
+            "scheme": "ws",
+            "path": "/",
+            "query_string": b"",
+            "headers": [],
+            "client": ("127.0.0.1", 32767),
+            "server": ("127.0.0.1", 80),
+        }
+        app = otel_asgi.OpenTelemetryMiddleware(simple_asgi)
+        self.seed_app(app)
+        self.send_input({"type": "websocket.connect"})
+        self.send_input({"type": "websocket.receive", "text": "ping"})
+        self.send_input({"type": "websocket.disconnect"})
+        self.get_all_output()
+        metrics_list = self.memory_metrics_reader.get_metrics_data()
+        self.assertEqual(
+            len(metrics_list.resource_metrics[0].scope_metrics), 0
+        )
+
 
 class TestAsgiAttributes(unittest.TestCase):
     def setUp(self):

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

@@ -165,6 +165,7 @@ def instrument_app(
         client_request_hook: _ClientRequestHookT = None,
         client_response_hook: _ClientResponseHookT = None,
         tracer_provider=None,
+        meter_provider=None,
         excluded_urls=None,
     ):
         """Instrument an uninstrumented FastAPI application."""
@@ -185,6 +186,7 @@ def instrument_app(
                 client_request_hook=client_request_hook,
                 client_response_hook=client_response_hook,
                 tracer_provider=tracer_provider,
+                meter_provider=meter_provider,
             )
             app._is_instrumented_by_opentelemetry = True
         else:
@@ -223,6 +225,7 @@ def _instrument(self, **kwargs):
             if _excluded_urls is None
             else parse_excluded_urls(_excluded_urls)
         )
+        _InstrumentedFastAPI._meter_provider = kwargs.get("meter_provider")
         fastapi.FastAPI = _InstrumentedFastAPI
 
     def _uninstrument(self, **kwargs):
@@ -231,6 +234,7 @@ def _uninstrument(self, **kwargs):
 
 class _InstrumentedFastAPI(fastapi.FastAPI):
     _tracer_provider = None
+    _meter_provider = None
     _excluded_urls = None
     _server_request_hook: _ServerRequestHookT = None
     _client_request_hook: _ClientRequestHookT = None
@@ -246,6 +250,7 @@ def __init__(self, *args, **kwargs):
             client_request_hook=_InstrumentedFastAPI._client_request_hook,
             client_response_hook=_InstrumentedFastAPI._client_response_hook,
             tracer_provider=_InstrumentedFastAPI._tracer_provider,
+            meter_provider=_InstrumentedFastAPI._meter_provider,
         )