Draft: Initial get_connection wrapper for redis metrics.

Author: dbgoytia

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

@@ -91,7 +91,6 @@ def response_hook(span, instance, response):
 """
 import typing
 from typing import Any, Collection
-
 import redis
 from wrapt import wrap_function_wrapper
 
@@ -106,6 +105,7 @@ def response_hook(span, instance, response):
 from opentelemetry.instrumentation.utils import unwrap
 from opentelemetry.semconv.trace import SpanAttributes
 from opentelemetry.trace import Span
+from opentelemetry.metrics import UpDownCounter, get_meter
 
 _DEFAULT_SERVICE = "redis"
 
@@ -119,6 +119,7 @@ def response_hook(span, instance, response):
 ]
 
 _REDIS_ASYNCIO_VERSION = (4, 2, 0)
+
 if redis.VERSION >= _REDIS_ASYNCIO_VERSION:
     import redis.asyncio
 
@@ -137,6 +138,7 @@ def _set_connection_attributes(span, conn):
 
 def _instrument(
     tracer,
+    connections_usage: UpDownCounter,
     request_hook: _RequestHookT = None,
     response_hook: _ResponseHookT = None,
 ):
@@ -147,6 +149,7 @@ def _traced_execute_command(func, instance, args, kwargs):
             name = args[0]
         else:
             name = instance.connection_pool.connection_kwargs.get("db", 0)
+
         with tracer.start_as_current_span(
             name, kind=trace.SpanKind.CLIENT
         ) as span:
@@ -200,13 +203,31 @@ def _traced_execute_pipeline(func, instance, args, kwargs):
                 response_hook(span, instance, response)
             return response
 
+    def _traced_get_connection(func, connection_pool, command_name, *keys, **options):
+        span_name = "random-test"
+        with tracer.start_as_current_span(
+            span_name, kind=trace.SpanKind.CLIENT
+        ) as span:
+            response = func(command_name, *keys, **options)
+            metric_labels = {
+                SpanAttributes.DB_CLIENT_CONNECTIONS_USAGE: connection_pool._created_connections,
+            }
+            connections_usage.add(1, metric_labels)
+            metric_labels[
+                SpanAttributes.DB_CLIENT_CONNECTIONS_USAGE
+            ] = connection_pool._created_connections
+            return response
+
+
     pipeline_class = (
         "BasePipeline" if redis.VERSION < (3, 0, 0) else "Pipeline"
     )
     redis_class = "StrictRedis" if redis.VERSION < (3, 0, 0) else "Redis"
 
     wrap_function_wrapper(
-        "redis", f"{redis_class}.execute_command", _traced_execute_command
+        "redis",
+        f"{redis_class}.execute_command",
+        _traced_execute_command
     )
     wrap_function_wrapper(
         "redis.client",
@@ -229,6 +250,11 @@ def _traced_execute_pipeline(func, instance, args, kwargs):
             "ClusterPipeline.execute",
             _traced_execute_pipeline,
         )
+    wrap_function_wrapper(
+        "redis",
+        "ConnectionPool.get_connection",
+        _traced_get_connection
+    )
     if redis.VERSION >= _REDIS_ASYNCIO_VERSION:
         wrap_function_wrapper(
             "redis.asyncio",
@@ -278,8 +304,20 @@ def _instrument(self, **kwargs):
         tracer = trace.get_tracer(
             __name__, __version__, tracer_provider=tracer_provider
         )
+        meter_provider = kwargs.get("meter_provider")
+        meter = get_meter(
+            __name__,
+            __version__,
+            meter_provider
+        )
+        connections_usage = meter.create_up_down_counter(
+            name="db.client.connections.usage",
+            unit="connections",
+            description="The number of connections that are currently in state described"
+        )
         _instrument(
             tracer,
+            connections_usage,
             request_hook=kwargs.get("request_hook"),
             response_hook=kwargs.get("response_hook"),
         )

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

@@ -14,3 +14,4 @@
 
 
 _instruments = ("redis >= 2.6",)
+_supports_metrics = True

instrumentation/opentelemetry-instrumentation-redis/tests/test_redis.py

@@ -141,3 +141,60 @@ def request_hook(span, conn, args, kwargs):
 
         span = spans[0]
         self.assertEqual(span.attributes.get(custom_attribute_name), "GET")
+
+
+class TestRedisIntegrationMetric(TestBase):
+
+    def setUp(self):
+        super().setUp()
+        RedisInstrumentor().instrument(meter_provider=self.meter_provider)
+
+    def tearDown(self):
+        super().tearDown()
+        RedisInstrumentor().uninstrument()
+
+
+    @staticmethod
+    def redis_get_multiple_connections():
+        pool = redis.ConnectionPool(host='localhost', port=6379, db=0)
+        c1 = pool.get_connection('_')
+        c2 = pool.get_connection('_')
+        c3 = pool.get_connection('_')
+
+    @staticmethod
+    def redis_get():
+        pool = redis.ConnectionPool(host='localhost', port=6379, db=0)
+        redis_client = redis.Redis(connection_pool=pool)
+        redis_client.get('foo')
+
+
+    def test_basic_metric_success_redis(self):
+        self.redis_get()
+        expected_attributes = {
+            "db.client.connections.usage": 1,
+        }
+        for (
+            resource_metrics
+        ) in self.memory_metrics_reader.get_metrics_data().resource_metrics:
+            for scope_metrics in resource_metrics.scope_metrics:
+                for metric in scope_metrics.metrics:
+                    for data_point in metric.data.data_points:
+                        self.assertDictEqual(
+                            expected_attributes, dict(data_point.attributes)
+                        )
+
+    def test_multiple_connections_metric_success_redis(self):
+        self.redis_get_multiple_connections()
+
+        expected_attributes = {
+            "db.client.connections.usage": 3,
+        }
+        for (
+                resource_metrics
+        ) in self.memory_metrics_reader.get_metrics_data().resource_metrics:
+            for scope_metrics in resource_metrics.scope_metrics:
+                for metric in scope_metrics.metrics:
+                    for data_point in metric.data.data_points:
+                        self.assertDictEqual(
+                            expected_attributes, dict(data_point.attributes)
+                        )