opentelemetry-instrumentation-dbapi: Add Database client operation duration and returned rows metrics (#4481)

* opentelemetry-instrumentation-dbapi: Add Database client operation duration and returned rows metrics

* update CHANGELOG.md

* add changelog fragment

* Update changelog for opentelemetry-instrumentation-dbapi

---------

Co-authored-by: Mike Goldsmith <goldsmith.mike@gmail.com>
Co-authored-by: Tammy Baylis <96076570+tammy-baylis-swi@users.noreply.github.com>
Co-authored-by: Riccardo Magliocchetti <riccardo.magliocchetti@gmail.com>

Author: 4 people

.changelog/4481.added

@@ -0,0 +1 @@
+`opentelemetry-instrumentation-dbapi`: Add Database client operation duration and returned rows metrics

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

@@ -160,6 +160,7 @@
 import functools
 import logging
 import re
+import time
 from typing import Any, Awaitable, Callable, Generic, TypeVar
 
 from wrapt import wrap_function_wrapper
@@ -179,6 +180,7 @@
     _get_schema_url_for_signal_types,
     _OpenTelemetrySemanticConventionStability,
     _OpenTelemetryStabilitySignalType,
+    _report_new,
     _set_db_name,
     _set_db_statement,
     _set_db_system,
@@ -193,6 +195,21 @@
     is_instrumentation_enabled,
     unwrap,
 )
+from opentelemetry.metrics import MeterProvider, get_meter
+from opentelemetry.semconv._incubating.metrics.db_metrics import (
+    create_db_client_operation_duration,
+    create_db_client_response_returned_rows,
+)
+from opentelemetry.semconv.attributes.db_attributes import (
+    DB_NAMESPACE,
+    DB_OPERATION_NAME,
+    DB_SYSTEM_NAME,
+)
+from opentelemetry.semconv.attributes.error_attributes import ERROR_TYPE
+from opentelemetry.semconv.attributes.server_attributes import (
+    SERVER_ADDRESS,
+    SERVER_PORT,
+)
 from opentelemetry.trace import SpanKind, TracerProvider, get_tracer
 from opentelemetry.util._importlib_metadata import version as util_version
 
@@ -217,6 +234,7 @@ def trace_integration(
     db_api_integration_factory: type[DatabaseApiIntegration] | None = None,
     enable_attribute_commenter: bool = False,
     commenter_options: dict[str, Any] | None = None,
+    meter_provider: MeterProvider | None = None,
 ):
     """Integrate with DB API library.
     https://www.python.org/dev/peps/pep-0249/
@@ -236,6 +254,8 @@ def trace_integration(
             default one is used.
         enable_attribute_commenter: Flag to enable/disable sqlcomment inclusion in `db.statement` and/or `db.query.text` span attribute. Only available if enable_commenter=True.
         commenter_options: Configurations for tags to be appended at the sql query.
+        meter_provider: The :class:`opentelemetry.metrics.MeterProvider` to
+            use. If omitted the current configured one is used.
     """
     wrap_connect(
         __name__,
@@ -250,9 +270,11 @@ def trace_integration(
         db_api_integration_factory=db_api_integration_factory,
         enable_attribute_commenter=enable_attribute_commenter,
         commenter_options=commenter_options,
+        meter_provider=meter_provider,
     )
 
 
+# pylint: disable-next=too-many-positional-arguments
 def wrap_connect(
     name: str,
     connect_module: Callable[..., Any],
@@ -266,6 +288,7 @@ def wrap_connect(
     db_api_integration_factory: type[DatabaseApiIntegration] | None = None,
     commenter_options: dict[str, Any] | None = None,
     enable_attribute_commenter: bool = False,
+    meter_provider: MeterProvider | None = None,
 ):
     """Integrate with DB API library.
     https://www.python.org/dev/peps/pep-0249/
@@ -285,6 +308,8 @@ def wrap_connect(
             default one is used.
         commenter_options: Configurations for tags to be appended at the sql query.
         enable_attribute_commenter: Flag to enable/disable sqlcomment inclusion in `db.statement` and/or `db.query.text` span attribute. Only available if enable_commenter=True.
+        meter_provider: The :class:`opentelemetry.metrics.MeterProvider` to
+            use. If omitted the current configured one is used.
 
     """
     db_api_integration_factory = (
@@ -309,6 +334,7 @@ def wrap_connect_(
             commenter_options=commenter_options,
             connect_module=connect_module,
             enable_attribute_commenter=enable_attribute_commenter,
+            meter_provider=meter_provider,
         )
         return db_integration.wrapped_connection(wrapped, args, kwargs)
 
@@ -333,6 +359,7 @@ def unwrap_connect(
     unwrap(connect_module, connect_method_name)
 
 
+# pylint: disable-next=too-many-positional-arguments
 def instrument_connection(
     name: str,
     connection: ConnectionT | TracedConnectionProxy[ConnectionT],
@@ -346,6 +373,7 @@ def instrument_connection(
     connect_module: Callable[..., Any] | None = None,
     enable_attribute_commenter: bool = False,
     db_api_integration_factory: type[DatabaseApiIntegration] | None = None,
+    meter_provider: MeterProvider | None = None,
 ) -> TracedConnectionProxy[ConnectionT]:
     """Enable instrumentation in a database connection.
 
@@ -367,6 +395,8 @@ def instrument_connection(
             replacement for :class:`DatabaseApiIntegration`. Can be used to
             obtain connection attributes from the connect method instead of
             from the connection itself (as done by the pymssql intrumentor).
+        meter_provider: The :class:`opentelemetry.metrics.MeterProvider` to
+            use. If omitted the current configured one is used.
 
     Returns:
         An instrumented connection.
@@ -390,6 +420,7 @@ def instrument_connection(
         commenter_options=commenter_options,
         connect_module=connect_module,
         enable_attribute_commenter=enable_attribute_commenter,
+        meter_provider=meter_provider,
     )
     db_integration.get_connection_attributes(connection)
     return get_traced_connection_proxy(connection, db_integration)
@@ -426,6 +457,7 @@ def __init__(
         commenter_options: dict[str, Any] | None = None,
         connect_module: Callable[..., Any] | None = None,
         enable_attribute_commenter: bool = False,
+        meter_provider: MeterProvider | None = None,
     ):
         # Initialize semantic conventions opt-in if needed
         _OpenTelemetrySemanticConventionStability._initialize()
@@ -458,6 +490,24 @@ def __init__(
                 ]
             ),
         )
+        self._meter = None
+        self._duration_histogram = None
+        self._returned_rows_histogram = None
+        if _report_new(self._sem_conv_opt_in_mode_db):
+            self._meter = get_meter(
+                self._name,
+                self._version,
+                meter_provider,
+                schema_url=_get_schema_url_for_signal_types(
+                    [_OpenTelemetryStabilitySignalType.DATABASE]
+                ),
+            )
+            self._duration_histogram = create_db_client_operation_duration(
+                self._meter
+            )
+            self._returned_rows_histogram = (
+                create_db_client_response_returned_rows(self._meter)
+            )
         self.capture_parameters = capture_parameters
         self.enable_commenter = enable_commenter
         self.commenter_options = commenter_options
@@ -467,6 +517,8 @@ def __init__(
         self.span_attributes: dict[str, Any] = {}
         self.name = ""
         self.database = ""
+        self._server_address: str | None = None
+        self._server_port: int | None = None
         self.connect_module = connect_module
         self.commenter_data = self.calculate_commenter_data()
 
@@ -579,11 +631,13 @@ def get_connection_attributes(self, connection: object) -> None:
                 host,
                 self._sem_conv_opt_in_mode_http,
             )
+            self._server_address = host
         port = self.connection_props.get("port")
         if port is not None:
             _set_http_peer_port_client(
                 self.span_attributes, port, self._sem_conv_opt_in_mode_http
             )
+            self._server_port = port
 
 
 # pylint: disable=abstract-method,no-member
@@ -754,6 +808,50 @@ def get_statement(self, cursor: CursorT, args: tuple[Any, ...]):  # pylint: disa
             return statement.decode("utf8", "replace")
         return statement
 
+    def _get_metric_attributes(
+        self,
+        operation_name: str,
+        error: Exception | None,
+    ) -> dict[str, Any]:
+        attributes: dict[str, Any] = {
+            DB_SYSTEM_NAME: self._db_api_integration.database_system,
+        }
+        if self._db_api_integration.database:
+            attributes[DB_NAMESPACE] = self._db_api_integration.database
+        if operation_name:
+            attributes[DB_OPERATION_NAME] = operation_name
+        if self._db_api_integration._server_address is not None:
+            attributes[SERVER_ADDRESS] = (
+                self._db_api_integration._server_address
+            )
+        if self._db_api_integration._server_port is not None:
+            attributes[SERVER_PORT] = self._db_api_integration._server_port
+        if error is not None:
+            attributes[ERROR_TYPE] = type(error).__qualname__
+        return attributes
+
+    def _record_metrics(
+        self,
+        cursor: CursorT,
+        operation_name: str,
+        start_time: float,
+        error: Exception | None,
+    ) -> None:
+        if not _report_new(self._db_api_integration._sem_conv_opt_in_mode_db):
+            # DB Metrics are not supported without Database semconv opt-in
+            return
+        elapsed = time.perf_counter() - start_time
+        attributes = self._get_metric_attributes(operation_name, error)
+        self._db_api_integration._duration_histogram.record(
+            elapsed, attributes=attributes
+        )
+        if error is None:
+            rowcount = getattr(cursor, "rowcount", None)
+            if isinstance(rowcount, int) and rowcount >= 0:
+                self._db_api_integration._returned_rows_histogram.record(
+                    rowcount, attributes=attributes
+                )
+
     def traced_execution(
         self,
         cursor: CursorT,
@@ -764,7 +862,8 @@ def traced_execution(
         if not is_instrumentation_enabled():
             return query_method(*args, **kwargs)
 
-        name = self.get_operation_name(cursor, args)
+        operation_name = self.get_operation_name(cursor, args)
+        name = operation_name
         if not name:
             name = (
                 self._db_api_integration.database
@@ -793,7 +892,15 @@ def traced_execution(
                 else:
                     # no sqlcomment anywhere
                     self._populate_span(span, cursor, *args)
-            return query_method(*args, **kwargs)
+            start_time = time.perf_counter()
+            error: Exception | None = None
+            try:
+                return query_method(*args, **kwargs)
+            except Exception as exc:
+                error = exc
+                raise
+            finally:
+                self._record_metrics(cursor, operation_name, start_time, error)
 
     async def traced_execution_async(
         self,
@@ -802,7 +909,8 @@ async def traced_execution_async(
         *args: tuple[Any, ...],
         **kwargs: dict[Any, Any],
     ):
-        name = self.get_operation_name(cursor, args)
+        operation_name = self.get_operation_name(cursor, args)
+        name = operation_name
         if not name:
             name = (
                 self._db_api_integration.database
@@ -831,7 +939,15 @@ async def traced_execution_async(
                 else:
                     # no sqlcomment anywhere
                     self._populate_span(span, cursor, *args)
-            return await query_method(*args, **kwargs)
+            start_time = time.perf_counter()
+            error: Exception | None = None
+            try:
+                return await query_method(*args, **kwargs)
+            except Exception as exc:
+                error = exc
+                raise
+            finally:
+                self._record_metrics(cursor, operation_name, start_time, error)
 
 
 # pylint: disable=abstract-method,no-member