Feature/metrics instrumentation urllib3 (#1198)

CHANGELOG.md

@@ -59,6 +59,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#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))
+- Add metric instrumentation for Urllib3
+  ([#1198](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1198))
 - `opentelemetry-instrumentation-aio-pika` added RabbitMQ aio-pika module instrumentation.
   ([#1095](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1095))
 - `opentelemetry-instrumentation-requests` Restoring metrics in requests

instrumentation/README.md

@@ -40,5 +40,5 @@
 | [opentelemetry-instrumentation-system-metrics](./opentelemetry-instrumentation-system-metrics) | psutil >= 5 | No
 | [opentelemetry-instrumentation-tornado](./opentelemetry-instrumentation-tornado) | tornado >= 5.1.1 | No
 | [opentelemetry-instrumentation-urllib](./opentelemetry-instrumentation-urllib) | urllib | No
-| [opentelemetry-instrumentation-urllib3](./opentelemetry-instrumentation-urllib3) | urllib3 >= 1.0.0, < 2.0.0 | No
+| [opentelemetry-instrumentation-urllib3](./opentelemetry-instrumentation-urllib3) | urllib3 >= 1.0.0, < 2.0.0 | Yes
 | [opentelemetry-instrumentation-wsgi](./opentelemetry-instrumentation-wsgi) | wsgi | Yes

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

@@ -66,6 +66,7 @@ API
 
 import contextlib
 import typing
+from timeit import default_timer
 from typing import Collection
 
 import urllib3.connectionpool
@@ -83,9 +84,10 @@ from opentelemetry.instrumentation.utils import (
     http_status_to_status_code,
     unwrap,
 )
+from opentelemetry.metrics import Histogram, get_meter
 from opentelemetry.propagate import inject
 from opentelemetry.semconv.trace import SpanAttributes
-from opentelemetry.trace import Span, SpanKind, get_tracer
+from opentelemetry.trace import Span, SpanKind, Tracer, get_tracer
 from opentelemetry.trace.status import Status
 from opentelemetry.util.http.httplib import set_ip_on_next_http_connection
 
@@ -135,8 +137,31 @@ class URLLib3Instrumentor(BaseInstrumentor):
         """
         tracer_provider = kwargs.get("tracer_provider")
         tracer = get_tracer(__name__, __version__, tracer_provider)
+
+        meter_provider = kwargs.get("meter_provider")
+        meter = get_meter(__name__, __version__, meter_provider)
+
+        duration_histogram = meter.create_histogram(
+            name="http.client.duration",
+            unit="ms",
+            description="measures the duration outbound HTTP requests",
+        )
+        request_size_histogram = meter.create_histogram(
+            name="http.client.request.size",
+            unit="By",
+            description="measures the size of HTTP request messages (compressed)",
+        )
+        response_size_histogram = meter.create_histogram(
+            name="http.client.response.size",
+            unit="By",
+            description="measures the size of HTTP response messages (compressed)",
+        )
+
         _instrument(
             tracer,
+            duration_histogram,
+            request_size_histogram,
+            response_size_histogram,
             request_hook=kwargs.get("request_hook"),
             response_hook=kwargs.get("response_hook"),
             url_filter=kwargs.get("url_filter"),
@@ -147,7 +172,10 @@ class URLLib3Instrumentor(BaseInstrumentor):
 
 
 def _instrument(
-    tracer,
+    tracer: Tracer,
+    duration_histogram: Histogram,
+    request_size_histogram: Histogram,
+    response_size_histogram: Histogram,
     request_hook: _RequestHookT = None,
     response_hook: _ResponseHookT = None,
     url_filter: _UrlFilterT = None,
@@ -175,11 +203,30 @@ def _instrument(
             inject(headers)
 
             with _suppress_further_instrumentation():
+                start_time = default_timer()
                 response = wrapped(*args, **kwargs)
+                elapsed_time = round((default_timer() - start_time) * 1000)
 
             _apply_response(span, response)
             if callable(response_hook):
                 response_hook(span, instance, response)
+
+            request_size = 0 if body is None else len(body)
+            response_size = int(response.headers.get("Content-Length", 0))
+            metric_attributes = _create_metric_attributes(
+                instance, response, method
+            )
+
+            duration_histogram.record(
+                elapsed_time, attributes=metric_attributes
+            )
+            request_size_histogram.record(
+                request_size, attributes=metric_attributes
+            )
+            response_size_histogram.record(
+                response_size, attributes=metric_attributes
+            )
+
             return response
 
     wrapt.wrap_function_wrapper(
@@ -254,6 +301,29 @@ def _is_instrumentation_suppressed() -> bool:
     )
 
 
+def _create_metric_attributes(
+    instance: urllib3.connectionpool.HTTPConnectionPool,
+    response: urllib3.response.HTTPResponse,
+    method: str,
+) -> dict:
+    metric_attributes = {
+        SpanAttributes.HTTP_METHOD: method,
+        SpanAttributes.HTTP_HOST: instance.host,
+        SpanAttributes.HTTP_SCHEME: instance.scheme,
+        SpanAttributes.HTTP_STATUS_CODE: response.status,
+        SpanAttributes.NET_PEER_NAME: instance.host,
+        SpanAttributes.NET_PEER_PORT: instance.port,
+    }
+
+    version = getattr(response, "version")
+    if version:
+        metric_attributes[SpanAttributes.HTTP_FLAVOR] = (
+            "1.1" if version == 11 else "1.0"
+        )
+
+    return metric_attributes
+
+
 @contextlib.contextmanager
 def _suppress_further_instrumentation():
     token = context.attach(

instrumentation/opentelemetry-instrumentation-urllib3/src/opentelemetry/instrumentation/urllib3/package.py

@@ -14,3 +14,5 @@
 
 
 _instruments = ("urllib3 >= 1.0.0, < 2.0.0",)
+
+_supports_metrics = True

instrumentation/opentelemetry-instrumentation-urllib3/tests/test_urllib3_ip_support.py

@@ -12,8 +12,11 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from timeit import default_timer
+
 import urllib3
 import urllib3.exceptions
+from urllib3.request import encode_multipart_formdata
 
 from opentelemetry import trace
 from opentelemetry.instrumentation.urllib3 import URLLib3Instrumentor
@@ -84,3 +87,136 @@ class TestURLLib3InstrumentorWithRealSocket(HttpTestBase, TestBase):
             "net.peer.ip": self.assert_ip,
         }
         self.assertGreaterEqual(span.attributes.items(), attributes.items())
+
+
+class TestURLLib3InstrumentorMetric(HttpTestBase, TestBase):
+    def setUp(self):
+        super().setUp()
+        self.assert_ip = self.server.server_address[0]
+        self.assert_port = self.server.server_address[1]
+        self.http_host = ":".join(map(str, self.server.server_address[:2]))
+        self.http_url_base = "http://" + self.http_host
+        self.http_url = self.http_url_base + "/status/200"
+        URLLib3Instrumentor().instrument(meter_provider=self.meter_provider)
+
+    def tearDown(self):
+        super().tearDown()
+        URLLib3Instrumentor().uninstrument()
+
+    def test_metric_uninstrument(self):
+        with urllib3.PoolManager() as pool:
+            pool.request("GET", self.http_url)
+            URLLib3Instrumentor().uninstrument()
+            pool.request("GET", self.http_url)
+
+            metrics_list = self.memory_metrics_reader.get_metrics_data()
+            for resource_metric in metrics_list.resource_metrics:
+                for scope_metric in resource_metric.scope_metrics:
+                    for metric in scope_metric.metrics:
+                        for point in list(metric.data.data_points):
+                            self.assertEqual(point.count, 1)
+
+    def test_basic_metric_check_client_size_get(self):
+        with urllib3.PoolManager() as pool:
+            start_time = default_timer()
+            response = pool.request("GET", self.http_url)
+            client_duration_estimated = (default_timer() - start_time) * 1000
+
+            expected_attributes = {
+                "http.status_code": 200,
+                "http.host": self.assert_ip,
+                "http.method": "GET",
+                "http.flavor": "1.1",
+                "http.scheme": "http",
+                "net.peer.name": self.assert_ip,
+                "net.peer.port": self.assert_port,
+            }
+            expected_data = {
+                "http.client.request.size": 0,
+                "http.client.response.size": len(response.data),
+            }
+            expected_metrics = [
+                "http.client.duration",
+                "http.client.request.size",
+                "http.client.response.size",
+            ]
+
+            resource_metrics = (
+                self.memory_metrics_reader.get_metrics_data().resource_metrics
+            )
+            for metrics in resource_metrics:
+                for scope_metrics in metrics.scope_metrics:
+                    self.assertEqual(len(scope_metrics.metrics), 3)
+                    for metric in scope_metrics.metrics:
+                        for data_point in metric.data.data_points:
+                            if metric.name in expected_data:
+                                self.assertEqual(
+                                    data_point.sum, expected_data[metric.name]
+                                )
+                            if metric.name == "http.client.duration":
+                                self.assertAlmostEqual(
+                                    data_point.sum,
+                                    client_duration_estimated,
+                                    delta=1000,
+                                )
+                            self.assertIn(metric.name, expected_metrics)
+                            self.assertDictEqual(
+                                expected_attributes,
+                                dict(data_point.attributes),
+                            )
+                            self.assertEqual(data_point.count, 1)
+
+    def test_basic_metric_check_client_size_post(self):
+        with urllib3.PoolManager() as pool:
+            start_time = default_timer()
+            data_fields = {"data": "test"}
+            response = pool.request("POST", self.http_url, fields=data_fields)
+            client_duration_estimated = (default_timer() - start_time) * 1000
+
+            expected_attributes = {
+                "http.status_code": 501,
+                "http.host": self.assert_ip,
+                "http.method": "POST",
+                "http.flavor": "1.1",
+                "http.scheme": "http",
+                "net.peer.name": self.assert_ip,
+                "net.peer.port": self.assert_port,
+            }
+
+            body = encode_multipart_formdata(data_fields)[0]
+
+            expected_data = {
+                "http.client.request.size": len(body),
+                "http.client.response.size": len(response.data),
+            }
+            expected_metrics = [
+                "http.client.duration",
+                "http.client.request.size",
+                "http.client.response.size",
+            ]
+
+            resource_metrics = (
+                self.memory_metrics_reader.get_metrics_data().resource_metrics
+            )
+            for metrics in resource_metrics:
+                for scope_metrics in metrics.scope_metrics:
+                    self.assertEqual(len(scope_metrics.metrics), 3)
+                    for metric in scope_metrics.metrics:
+                        for data_point in metric.data.data_points:
+                            if metric.name in expected_data:
+                                self.assertEqual(
+                                    data_point.sum, expected_data[metric.name]
+                                )
+                            if metric.name == "http.client.duration":
+                                self.assertAlmostEqual(
+                                    data_point.sum,
+                                    client_duration_estimated,
+                                    delta=1000,
+                                )
+                            self.assertIn(metric.name, expected_metrics)
+
+                            self.assertDictEqual(
+                                expected_attributes,
+                                dict(data_point.attributes),
+                            )
+                            self.assertEqual(data_point.count, 1)