Add DB client metrics to aiopg instrumentation

.changelog/4826.added

@@ -0,0 +1 @@
+`opentelemetry-instrumentation-aiopg`: Add DB client operation duration and returned rows metrics.

instrumentation/README.md

@@ -5,7 +5,7 @@
 | [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 | migration
 | [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 | migration
+| [opentelemetry-instrumentation-aiopg](./opentelemetry-instrumentation-aiopg) | aiopg >= 0.13.0, < 2.0.0 | Yes | migration
 | [opentelemetry-instrumentation-asgi](./opentelemetry-instrumentation-asgi) | asgiref ~= 3.0 | Yes | migration
 | [opentelemetry-instrumentation-asyncclick](./opentelemetry-instrumentation-asyncclick) | asyncclick ~= 8.0 | No | development
 | [opentelemetry-instrumentation-asyncio](./opentelemetry-instrumentation-asyncio) | asyncio | No | development

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

@@ -91,13 +91,15 @@ def _instrument(self, **kwargs):
         """
 
         tracer_provider = kwargs.get("tracer_provider")
+        meter_provider = kwargs.get("meter_provider")
 
         wrappers.wrap_connect(
             __name__,
             self._DATABASE_SYSTEM,
             self._CONNECTION_ATTRIBUTES,
             version=__version__,
             tracer_provider=tracer_provider,
+            meter_provider=meter_provider,
         )
 
         wrappers.wrap_create_pool(
@@ -106,6 +108,7 @@ def _instrument(self, **kwargs):
             self._CONNECTION_ATTRIBUTES,
             version=__version__,
             tracer_provider=tracer_provider,
+            meter_provider=meter_provider,
         )
 
     # pylint:disable=no-self-use
@@ -115,13 +118,17 @@ def _uninstrument(self, **kwargs):
         wrappers.unwrap_create_pool()
 
     # pylint:disable=no-self-use
-    def instrument_connection(self, connection, tracer_provider=None):
+    def instrument_connection(
+        self, connection, tracer_provider=None, meter_provider=None
+    ):
         """Enable instrumentation in a aiopg connection.
 
         Args:
             connection: The connection to instrument.
             tracer_provider: The optional tracer provider to use. If omitted
                 the current globally configured one is used.
+            meter_provider: The optional meter provider to use. If omitted the
+                current globally configured one is used.
 
         Returns:
             An instrumented connection.
@@ -133,6 +140,7 @@ def instrument_connection(self, connection, tracer_provider=None):
             self._CONNECTION_ATTRIBUTES,
             version=__version__,
             tracer_provider=tracer_provider,
+            meter_provider=meter_provider,
         )
 
     def uninstrument_connection(self, connection):

instrumentation/opentelemetry-instrumentation-aiopg/src/opentelemetry/instrumentation/aiopg/aiopg_integration.py

@@ -2,6 +2,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import inspect
+import time
 import typing
 from collections.abc import Coroutine
 
@@ -109,30 +110,36 @@ async def traced_execution(
         *args: typing.Tuple[typing.Any, typing.Any],
         **kwargs: typing.Dict[typing.Any, typing.Any],
     ):
-        name = ""
+        operation_name = ""
         if args:
-            name = self.get_operation_name(cursor, args)
+            operation_name = self.get_operation_name(cursor, args)
 
-        if not name:
-            name = (
+        span_name = operation_name
+        if not span_name:
+            span_name = (
                 self._db_api_integration.database
                 if self._db_api_integration.database
                 else self._db_api_integration.name
             )
 
         with self._db_api_integration._tracer.start_as_current_span(
-            name, kind=SpanKind.CLIENT
+            span_name, kind=SpanKind.CLIENT
         ) as span:
             if span.is_recording():
                 self._populate_span(span, cursor, *args)
+            start_time = time.perf_counter()
+            error = None
             try:
                 return await query_method(*args, **kwargs)
             except Exception as exc:
+                error = exc
                 if span.is_recording() and _report_new(
                     self._db_api_integration._sem_conv_opt_in_mode_db
                 ):
                     span.set_attribute(ERROR_TYPE, type(exc).__qualname__)
                 raise
+            finally:
+                self._record_metrics(cursor, operation_name, start_time, error)
 
 
 def get_traced_cursor_proxy(cursor, db_api_integration, *args, **kwargs):

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

@@ -4,4 +4,5 @@
 
 _instruments = ("aiopg >= 0.13.0, < 2.0.0",)
 
+_supports_metrics = True
 _semconv_status = "migration"

instrumentation/opentelemetry-instrumentation-aiopg/src/opentelemetry/instrumentation/aiopg/wrappers.py

@@ -34,6 +34,7 @@
 )
 from opentelemetry.instrumentation.aiopg.version import __version__
 from opentelemetry.instrumentation.utils import unwrap
+from opentelemetry.metrics import MeterProvider
 from opentelemetry.trace import TracerProvider
 
 logger = logging.getLogger(__name__)
@@ -43,6 +44,7 @@ def trace_integration(
     database_system: str,
     connection_attributes: typing.Dict = None,
     tracer_provider: typing.Optional[TracerProvider] = None,
+    meter_provider: typing.Optional[MeterProvider] = None,
 ):
     """Integrate with aiopg library.
     based on dbapi integration, where replaced sync wrap methods to async
@@ -54,6 +56,8 @@ def trace_integration(
             user in Connection object.
         tracer_provider: The :class:`opentelemetry.trace.TracerProvider` to
             use. If omitted the current configured one is used.
+        meter_provider: The :class:`opentelemetry.metrics.MeterProvider` to
+            use. If omitted the current configured one is used.
     """
 
     wrap_connect(
@@ -62,6 +66,7 @@ def trace_integration(
         connection_attributes,
         __version__,
         tracer_provider,
+        meter_provider,
     )
 
 
@@ -71,6 +76,7 @@ def wrap_connect(
     connection_attributes: typing.Dict = None,
     version: str = "",
     tracer_provider: typing.Optional[TracerProvider] = None,
+    meter_provider: typing.Optional[MeterProvider] = None,
 ):
     """Integrate with aiopg library.
     https://github.com/aio-libs/aiopg
@@ -84,6 +90,8 @@ def wrap_connect(
         version: Version of opentelemetry extension for aiopg.
         tracer_provider: The :class:`opentelemetry.trace.TracerProvider` to
             use. If omitted the current configured one is used.
+        meter_provider: The :class:`opentelemetry.metrics.MeterProvider` to
+            use. If omitted the current configured one is used.
     """
 
     # pylint: disable=unused-argument
@@ -99,6 +107,7 @@ def wrap_connect_(
             connection_attributes=connection_attributes,
             version=version,
             tracer_provider=tracer_provider,
+            meter_provider=meter_provider,
         )
         return _ContextManager(  # pylint: disable=no-value-for-parameter
             db_integration.wrapped_connection(wrapped, args, kwargs)
@@ -125,6 +134,7 @@ def instrument_connection(
     connection_attributes: typing.Dict = None,
     version: str = "",
     tracer_provider: typing.Optional[TracerProvider] = None,
+    meter_provider: typing.Optional[MeterProvider] = None,
 ):
     """Enable instrumentation in a database connection.
 
@@ -138,6 +148,8 @@ def instrument_connection(
         version: Version of opentelemetry extension for aiopg.
         tracer_provider: The :class:`opentelemetry.trace.TracerProvider` to
             use. If omitted the current configured one is used.
+        meter_provider: The :class:`opentelemetry.metrics.MeterProvider` to
+            use. If omitted the current configured one is used.
 
     Returns:
         An instrumented connection.
@@ -152,6 +164,7 @@ def instrument_connection(
         connection_attributes=connection_attributes,
         version=version,
         tracer_provider=tracer_provider,
+        meter_provider=meter_provider,
     )
     db_integration.get_connection_attributes(connection)
     return get_traced_connection_proxy(connection, db_integration)
@@ -179,6 +192,7 @@ def wrap_create_pool(
     connection_attributes: typing.Dict = None,
     version: str = "",
     tracer_provider: typing.Optional[TracerProvider] = None,
+    meter_provider: typing.Optional[MeterProvider] = None,
 ):
     # pylint: disable=unused-argument
     def wrap_create_pool_(
@@ -193,6 +207,7 @@ def wrap_create_pool_(
             connection_attributes=connection_attributes,
             version=version,
             tracer_provider=tracer_provider,
+            meter_provider=meter_provider,
         )
         return _PoolContextManager(
             db_integration.wrapped_pool(wrapped, args, kwargs)