feat(instrumentation-aiohttp-client): Add support for HTTP metrics (open-telemetry#3517)

* Add metrics support

* Update instrumentation/opentelemetry-instrumentation-aiohttp-client/src/opentelemetry/instrumentation/aiohttp_client/__init__.py

Co-authored-by: Emídio Neto <[email protected]>

* Update instrumentation/opentelemetry-instrumentation-aiohttp-client/src/opentelemetry/instrumentation/aiohttp_client/__init__.py

Co-authored-by: Emídio Neto <[email protected]>

* Update instrumentation/opentelemetry-instrumentation-aiohttp-client/src/opentelemetry/instrumentation/aiohttp_client/__init__.py

Co-authored-by: Emídio Neto <[email protected]>

* Adding tests, updating changelog

* Adding server address and port to both spans and metrics

---------

Co-authored-by: Emídio Neto <[email protected]>
Co-authored-by: Riccardo Magliocchetti <[email protected]>

Author: 3 people

CHANGELOG.md

@@ -45,6 +45,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - `opentelemetry-instrumentation-fastapi`: Drop support for FastAPI versions earlier than `0.92`
   ([#3012](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3012))
 
+### Added
+- `opentelemetry-instrumentation-aiohttp-client` Add support for HTTP metrics
+  ([#3517](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3517))
+
 ### Deprecated
 
 - Drop support for Python 3.8, bump baseline to Python 3.9.

instrumentation/README.md

@@ -2,7 +2,7 @@
 | Instrumentation | Supported Packages | Metrics support | Semconv status |
 | --------------- | ------------------ | --------------- | -------------- |
 | [opentelemetry-instrumentation-aio-pika](./opentelemetry-instrumentation-aio-pika) | aio_pika >= 7.2.0, < 10.0.0 | No | development
-| [opentelemetry-instrumentation-aiohttp-client](./opentelemetry-instrumentation-aiohttp-client) | aiohttp ~= 3.0 | No | migration
+| [opentelemetry-instrumentation-aiohttp-client](./opentelemetry-instrumentation-aiohttp-client) | aiohttp ~= 3.0 | Yes | migration
 | [opentelemetry-instrumentation-aiohttp-server](./opentelemetry-instrumentation-aiohttp-server) | aiohttp ~= 3.0 | Yes | development
 | [opentelemetry-instrumentation-aiokafka](./opentelemetry-instrumentation-aiokafka) | aiokafka >= 0.8, < 1.0 | No | development
 | [opentelemetry-instrumentation-aiopg](./opentelemetry-instrumentation-aiopg) | aiopg >= 0.13.0, < 2.0.0 | No | development

instrumentation/opentelemetry-instrumentation-aiohttp-client/src/opentelemetry/instrumentation/aiohttp_client/__init__.py

@@ -90,7 +90,9 @@ def response_hook(span: Span, params: typing.Union[
 
 import types
 import typing
+from timeit import default_timer
 from typing import Collection
+from urllib.parse import urlparse
 
 import aiohttp
 import wrapt
@@ -99,11 +101,20 @@ def response_hook(span: Span, params: typing.Union[
 from opentelemetry import context as context_api
 from opentelemetry import trace
 from opentelemetry.instrumentation._semconv import (
+    HTTP_DURATION_HISTOGRAM_BUCKETS_NEW,
+    HTTP_DURATION_HISTOGRAM_BUCKETS_OLD,
+    _client_duration_attrs_new,
+    _client_duration_attrs_old,
+    _filter_semconv_duration_attrs,
     _get_schema_url,
     _OpenTelemetrySemanticConventionStability,
     _OpenTelemetryStabilitySignalType,
     _report_new,
+    _report_old,
+    _set_http_host_client,
     _set_http_method,
+    _set_http_net_peer_name_client,
+    _set_http_peer_port_client,
     _set_http_url,
     _set_status,
     _StabilityMode,
@@ -115,8 +126,13 @@ def response_hook(span: Span, params: typing.Union[
     is_instrumentation_enabled,
     unwrap,
 )
+from opentelemetry.metrics import MeterProvider, get_meter
 from opentelemetry.propagate import inject
 from opentelemetry.semconv.attributes.error_attributes import ERROR_TYPE
+from opentelemetry.semconv.metrics import MetricInstruments
+from opentelemetry.semconv.metrics.http_metrics import (
+    HTTP_CLIENT_REQUEST_DURATION,
+)
 from opentelemetry.trace import Span, SpanKind, TracerProvider, get_tracer
 from opentelemetry.trace.status import Status, StatusCode
 from opentelemetry.util.http import remove_url_credentials, sanitize_method
@@ -172,11 +188,14 @@ def _set_http_status_code_attribute(
     )
 
 
+# pylint: disable=too-many-locals
+# pylint: disable=too-many-statements
 def create_trace_config(
     url_filter: _UrlFilterT = None,
     request_hook: _RequestHookT = None,
     response_hook: _ResponseHookT = None,
     tracer_provider: TracerProvider = None,
+    meter_provider: MeterProvider = None,
     sem_conv_opt_in_mode: _StabilityMode = _StabilityMode.DEFAULT,
 ) -> aiohttp.TraceConfig:
     """Create an aiohttp-compatible trace configuration.
@@ -205,6 +224,7 @@ def create_trace_config(
     :param Callable request_hook: Optional callback that can modify span name and request params.
     :param Callable response_hook: Optional callback that can modify span name and response params.
     :param tracer_provider: optional TracerProvider from which to get a Tracer
+    :param meter_provider: optional Meter provider to use
 
     :return: An object suitable for use with :py:class:`aiohttp.ClientSession`.
     :rtype: :py:class:`aiohttp.TraceConfig`
@@ -214,20 +234,70 @@ def create_trace_config(
     # Explicitly specify the type for the `request_hook` and `response_hook` param and rtype to work
     # around this issue.
 
+    schema_url = _get_schema_url(sem_conv_opt_in_mode)
+
     tracer = get_tracer(
         __name__,
         __version__,
         tracer_provider,
-        schema_url=_get_schema_url(sem_conv_opt_in_mode),
+        schema_url=schema_url,
+    )
+
+    meter = get_meter(
+        __name__,
+        __version__,
+        meter_provider,
+        schema_url,
     )
 
-    # TODO: Use this when we have durations for aiohttp-client
+    start_time = 0
+
+    duration_histogram_old = None
+    if _report_old(sem_conv_opt_in_mode):
+        duration_histogram_old = meter.create_histogram(
+            name=MetricInstruments.HTTP_CLIENT_DURATION,
+            unit="ms",
+            description="measures the duration of the outbound HTTP request",
+            explicit_bucket_boundaries_advisory=HTTP_DURATION_HISTOGRAM_BUCKETS_OLD,
+        )
+    duration_histogram_new = None
+    if _report_new(sem_conv_opt_in_mode):
+        duration_histogram_new = meter.create_histogram(
+            name=HTTP_CLIENT_REQUEST_DURATION,
+            unit="s",
+            description="Duration of HTTP client requests.",
+            explicit_bucket_boundaries_advisory=HTTP_DURATION_HISTOGRAM_BUCKETS_NEW,
+        )
+
     metric_attributes = {}
 
     def _end_trace(trace_config_ctx: types.SimpleNamespace):
+        elapsed_time = max(default_timer() - trace_config_ctx.start_time, 0)
         context_api.detach(trace_config_ctx.token)
         trace_config_ctx.span.end()
 
+        if trace_config_ctx.duration_histogram_old is not None:
+            duration_attrs_old = _filter_semconv_duration_attrs(
+                metric_attributes,
+                _client_duration_attrs_old,
+                _client_duration_attrs_new,
+                _StabilityMode.DEFAULT,
+            )
+            trace_config_ctx.duration_histogram_old.record(
+                max(round(elapsed_time * 1000), 0),
+                attributes=duration_attrs_old,
+            )
+        if trace_config_ctx.duration_histogram_new is not None:
+            duration_attrs_new = _filter_semconv_duration_attrs(
+                metric_attributes,
+                _client_duration_attrs_old,
+                _client_duration_attrs_new,
+                _StabilityMode.HTTP,
+            )
+            trace_config_ctx.duration_histogram_new.record(
+                elapsed_time, attributes=duration_attrs_new
+            )
+
     async def on_request_start(
         unused_session: aiohttp.ClientSession,
         trace_config_ctx: types.SimpleNamespace,
@@ -237,6 +307,7 @@ async def on_request_start(
             trace_config_ctx.span = None
             return
 
+        trace_config_ctx.start_time = default_timer()
         method = params.method
         request_span_name = _get_span_name(method)
         request_url = (
@@ -252,8 +323,44 @@ async def on_request_start(
             sanitize_method(method),
             sem_conv_opt_in_mode,
         )
+        _set_http_method(
+            metric_attributes,
+            method,
+            sanitize_method(method),
+            sem_conv_opt_in_mode,
+        )
         _set_http_url(span_attributes, request_url, sem_conv_opt_in_mode)
 
+        try:
+            parsed_url = urlparse(request_url)
+            if parsed_url.hostname:
+                _set_http_host_client(
+                    metric_attributes,
+                    parsed_url.hostname,
+                    sem_conv_opt_in_mode,
+                )
+                _set_http_net_peer_name_client(
+                    metric_attributes,
+                    parsed_url.hostname,
+                    sem_conv_opt_in_mode,
+                )
+                if _report_new(sem_conv_opt_in_mode):
+                    _set_http_host_client(
+                        span_attributes,
+                        parsed_url.hostname,
+                        sem_conv_opt_in_mode,
+                    )
+            if parsed_url.port:
+                _set_http_peer_port_client(
+                    metric_attributes, parsed_url.port, sem_conv_opt_in_mode
+                )
+                if _report_new(sem_conv_opt_in_mode):
+                    _set_http_peer_port_client(
+                        span_attributes, parsed_url.port, sem_conv_opt_in_mode
+                    )
+        except ValueError:
+            pass
+
         trace_config_ctx.span = trace_config_ctx.tracer.start_span(
             request_span_name, kind=SpanKind.CLIENT, attributes=span_attributes
         )
@@ -298,6 +405,7 @@ async def on_request_exception(
             exc_type = type(params.exception).__qualname__
             if _report_new(sem_conv_opt_in_mode):
                 trace_config_ctx.span.set_attribute(ERROR_TYPE, exc_type)
+                metric_attributes[ERROR_TYPE] = exc_type
 
             trace_config_ctx.span.set_status(
                 Status(StatusCode.ERROR, exc_type)
@@ -312,7 +420,12 @@ async def on_request_exception(
     def _trace_config_ctx_factory(**kwargs):
         kwargs.setdefault("trace_request_ctx", {})
         return types.SimpleNamespace(
-            tracer=tracer, url_filter=url_filter, **kwargs
+            tracer=tracer,
+            url_filter=url_filter,
+            start_time=start_time,
+            duration_histogram_old=duration_histogram_old,
+            duration_histogram_new=duration_histogram_new,
+            **kwargs,
         )
 
     trace_config = aiohttp.TraceConfig(
@@ -328,6 +441,7 @@ def _trace_config_ctx_factory(**kwargs):
 
 def _instrument(
     tracer_provider: TracerProvider = None,
+    meter_provider: MeterProvider = None,
     url_filter: _UrlFilterT = None,
     request_hook: _RequestHookT = None,
     response_hook: _ResponseHookT = None,
@@ -357,6 +471,7 @@ def instrumented_init(wrapped, instance, args, kwargs):
             request_hook=request_hook,
             response_hook=response_hook,
             tracer_provider=tracer_provider,
+            meter_provider=meter_provider,
             sem_conv_opt_in_mode=sem_conv_opt_in_mode,
         )
         trace_config._is_instrumented_by_opentelemetry = True
@@ -401,6 +516,7 @@ def _instrument(self, **kwargs):
         Args:
             **kwargs: Optional arguments
                 ``tracer_provider``: a TracerProvider, defaults to global
+                ``meter_provider``: a MeterProvider, defaults to global
                 ``url_filter``: A callback to process the requested URL prior to adding
                     it as a span attribute. This can be useful to remove sensitive data
                     such as API keys or user personal information.
@@ -415,6 +531,7 @@ def _instrument(self, **kwargs):
         )
         _instrument(
             tracer_provider=kwargs.get("tracer_provider"),
+            meter_provider=kwargs.get("meter_provider"),
             url_filter=kwargs.get("url_filter"),
             request_hook=kwargs.get("request_hook"),
             response_hook=kwargs.get("response_hook"),

instrumentation/opentelemetry-instrumentation-aiohttp-client/src/opentelemetry/instrumentation/aiohttp_client/package.py

@@ -15,6 +15,6 @@
 
 _instruments = ("aiohttp ~= 3.0",)
 
-_supports_metrics = False
+_supports_metrics = True
 
 _semconv_status = "migration"