Add metric instrumentation for WSGI

.github/workflows/test.yml

@@ -6,7 +6,7 @@ on:
     - 'release/*'
   pull_request:
 env:
-  CORE_REPO_SHA: cad776a2031c84fb3c3a1af90ee2a939f3394b9a
+  CORE_REPO_SHA: d4d7c67663cc22615748d632e1c8c5799e8eacae
 
 jobs:
   build:
@@ -42,7 +42,7 @@ jobs:
           path: |
             .tox
             ~/.cache/pip
-          key: v5-build-tox-cache-${{ env.RUN_MATRIX_COMBINATION }}-${{ hashFiles('tox.ini', 'gen-requirements.txt', 'dev-requirements.txt') }}
+          key: v6-build-tox-cache-${{ env.RUN_MATRIX_COMBINATION }}-${{ hashFiles('tox.ini', 'gen-requirements.txt', 'dev-requirements.txt') }}
       - name: run tox
         run: tox -f ${{ matrix.python-version }}-${{ matrix.package }} -- --benchmark-json=${{ env.RUN_MATRIX_COMBINATION }}-benchmark.json
   #     - name: Find and merge ${{ matrix.package }} benchmarks
@@ -118,7 +118,7 @@ jobs:
           path: |
             .tox
             ~/.cache/pip
-          key: v5-misc-tox-cache-${{ matrix.tox-environment }}-${{ hashFiles('tox.ini', 'dev-requirements.txt', 'gen-requirements.txt', 'docs-requirements.txt') }}
+          key: v6-misc-tox-cache-${{ matrix.tox-environment }}-${{ hashFiles('tox.ini', 'dev-requirements.txt', 'gen-requirements.txt', 'docs-requirements.txt') }}
       - name: run tox
         run: tox -e ${{ matrix.tox-environment }}
       - name: Ensure generated code is up to date

CHANGELOG.md

@@ -33,6 +33,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#1111](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1111))
 - Set otlp-proto-grpc as the default metrics exporter for auto-instrumentation
   ([#1127](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1127))
+- Add metric instrumentation for WSGI
+  ([#1128](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1128))
 
 
 ## [1.12.0rc1-0.31b0](https://github.com/open-telemetry/opentelemetry-python/releases/tag/v1.12.0rc1-0.31b0) - 2022-05-17

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

@@ -158,13 +158,15 @@ def response_hook(span: Span, environ: WSGIEnvironment, status: str, response_he
 import functools
 import typing
 import wsgiref.util as wsgiref_util
+from timeit import default_timer
 
 from opentelemetry import context, trace
 from opentelemetry.instrumentation.utils import (
     _start_internal_or_server_span,
     http_status_to_status_code,
 )
 from opentelemetry.instrumentation.wsgi.version import __version__
+from opentelemetry.metrics import get_meter
 from opentelemetry.propagators.textmap import Getter
 from opentelemetry.semconv.trace import SpanAttributes
 from opentelemetry.trace.status import Status, StatusCode
@@ -181,6 +183,26 @@ def response_hook(span: Span, environ: WSGIEnvironment, status: str, response_he
 _CARRIER_KEY_PREFIX = "HTTP_"
 _CARRIER_KEY_PREFIX_LEN = len(_CARRIER_KEY_PREFIX)
 
+# 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 WSGIGetter(Getter[dict]):
     def get(
@@ -304,6 +326,14 @@ def collect_custom_response_headers_attributes(response_headers):
     return attributes
 
 
+def _parse_status_code(resp_status):
+    status_code, _ = resp_status.split(" ", 1)
+    try:
+        return int(status_code)
+    except ValueError:
+        return None
+
+
 def add_response_attributes(
     span, start_response_status, response_headers
 ):  # pylint: disable=unused-argument
@@ -352,18 +382,39 @@ class OpenTelemetryMiddleware:
     """
 
     def __init__(
-        self, wsgi, request_hook=None, response_hook=None, tracer_provider=None
+        self,
+        wsgi,
+        request_hook=None,
+        response_hook=None,
+        tracer_provider=None,
+        meter_provider=None,
     ):
         self.wsgi = wsgi
         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.request_hook = request_hook
         self.response_hook = response_hook
 
     @staticmethod
-    def _create_start_response(span, start_response, response_hook):
+    def _create_start_response(
+        span, start_response, response_hook, duration_attrs
+    ):
         @functools.wraps(start_response)
         def _start_response(status, response_headers, *args, **kwargs):
             add_response_attributes(span, status, response_headers)
+            status_code = _parse_status_code(status)
+            if status_code is not None:
+                duration_attrs[SpanAttributes.HTTP_STATUS_CODE] = status_code
             if span.is_recording() and span.kind == trace.SpanKind.SERVER:
                 custom_attributes = collect_custom_response_headers_attributes(
                     response_headers
@@ -383,13 +434,24 @@ def __call__(self, environ, start_response):
             environ: A WSGI environment.
             start_response: The WSGI start_response callable.
         """
+        req_attrs = collect_request_attributes(environ)
+        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]
+
+        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]
+
         span, token = _start_internal_or_server_span(
             tracer=self.tracer,
             span_name=get_default_span_name(environ),
             start_time=None,
             context_carrier=environ,
             context_getter=wsgi_getter,
-            attributes=collect_request_attributes(environ),
+            attributes=req_attrs,
         )
         if span.is_recording() and span.kind == trace.SpanKind.SERVER:
             custom_attributes = collect_custom_request_headers_attributes(
@@ -405,28 +467,32 @@ def __call__(self, environ, start_response):
         if response_hook:
             response_hook = functools.partial(response_hook, span, environ)
 
+        start = default_timer()
+        self.active_requests_counter.add(1, active_requests_count_attrs)
         try:
             with trace.use_span(span):
                 start_response = self._create_start_response(
-                    span, start_response, response_hook
+                    span, start_response, response_hook, duration_attrs
                 )
                 iterable = self.wsgi(environ, start_response)
-                return _end_span_after_iterating(
-                    iterable, span, self.tracer, token
-                )
+                return _end_span_after_iterating(iterable, span, token)
         except Exception as ex:
             if span.is_recording():
                 span.set_status(Status(StatusCode.ERROR, str(ex)))
             span.end()
             if token is not None:
                 context.detach(token)
             raise
+        finally:
+            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)
 
 
 # Put this in a subfunction to not delay the call to the wrapped
 # WSGI application (instrumentation should change the application
 # behavior as little as possible).
-def _end_span_after_iterating(iterable, span, tracer, token):
+def _end_span_after_iterating(iterable, span, token):
     try:
         with trace.use_span(span):
             yield from iterable

instrumentation/opentelemetry-instrumentation-wsgi/tests/test_wsgi_middleware.py

@@ -20,6 +20,10 @@
 
 import opentelemetry.instrumentation.wsgi as otel_wsgi
 from opentelemetry import trace as trace_api
+from opentelemetry.sdk.metrics.export import (
+    HistogramDataPoint,
+    NumberDataPoint,
+)
 from opentelemetry.sdk.resources import Resource
 from opentelemetry.semconv.trace import SpanAttributes
 from opentelemetry.test.test_base import TestBase
@@ -99,6 +103,16 @@ def wsgi_with_custom_response_headers(environ, start_response):
     return [b"*"]
 
 
+_expected_metric_names = [
+    "http.server.active_requests",
+    "http.server.duration",
+]
+_recommended_attrs = {
+    "http.server.active_requests": otel_wsgi._active_requests_count_attrs,
+    "http.server.duration": otel_wsgi._duration_attrs,
+}
+
+
 class TestWsgiApplication(WsgiTestBase):
     def validate_response(
         self,
@@ -230,6 +244,36 @@ def test_wsgi_internal_error(self):
             StatusCode.ERROR,
         )
 
+    def test_wsgi_metrics(self):
+        app = otel_wsgi.OpenTelemetryMiddleware(error_wsgi_unhandled)
+        self.assertRaises(ValueError, app, self.environ, self.start_response)
+        self.assertRaises(ValueError, app, self.environ, self.start_response)
+        self.assertRaises(ValueError, app, self.environ, self.start_response)
+        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_default_span_name_missing_request_method(self):
         """Test that default span_names with missing request method."""
         self.environ.pop("REQUEST_METHOD")