diff --git a/.changelog/4826.added b/.changelog/4826.added new file mode 100644 index 0000000000..e6f4bf66b3 --- /dev/null +++ b/.changelog/4826.added @@ -0,0 +1 @@ +`opentelemetry-instrumentation-aiopg`: Add DB client operation duration and returned rows metrics. diff --git a/instrumentation/README.md b/instrumentation/README.md index 4c682e78ef..8432356600 100644 --- a/instrumentation/README.md +++ b/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 diff --git a/instrumentation/opentelemetry-instrumentation-aiopg/src/opentelemetry/instrumentation/aiopg/__init__.py b/instrumentation/opentelemetry-instrumentation-aiopg/src/opentelemetry/instrumentation/aiopg/__init__.py index 8167847002..441ddd1ad8 100644 --- a/instrumentation/opentelemetry-instrumentation-aiopg/src/opentelemetry/instrumentation/aiopg/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-aiopg/src/opentelemetry/instrumentation/aiopg/__init__.py @@ -91,6 +91,7 @@ def _instrument(self, **kwargs): """ tracer_provider = kwargs.get("tracer_provider") + meter_provider = kwargs.get("meter_provider") wrappers.wrap_connect( __name__, @@ -98,6 +99,7 @@ def _instrument(self, **kwargs): 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): diff --git a/instrumentation/opentelemetry-instrumentation-aiopg/src/opentelemetry/instrumentation/aiopg/aiopg_integration.py b/instrumentation/opentelemetry-instrumentation-aiopg/src/opentelemetry/instrumentation/aiopg/aiopg_integration.py index 68a1261f0b..0aa5b0afd7 100644 --- a/instrumentation/opentelemetry-instrumentation-aiopg/src/opentelemetry/instrumentation/aiopg/aiopg_integration.py +++ b/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): diff --git a/instrumentation/opentelemetry-instrumentation-aiopg/src/opentelemetry/instrumentation/aiopg/package.py b/instrumentation/opentelemetry-instrumentation-aiopg/src/opentelemetry/instrumentation/aiopg/package.py index 6fd73d515e..ac6297478a 100644 --- a/instrumentation/opentelemetry-instrumentation-aiopg/src/opentelemetry/instrumentation/aiopg/package.py +++ b/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" diff --git a/instrumentation/opentelemetry-instrumentation-aiopg/src/opentelemetry/instrumentation/aiopg/wrappers.py b/instrumentation/opentelemetry-instrumentation-aiopg/src/opentelemetry/instrumentation/aiopg/wrappers.py index 76e5dffcff..12d3a58376 100644 --- a/instrumentation/opentelemetry-instrumentation-aiopg/src/opentelemetry/instrumentation/aiopg/wrappers.py +++ b/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) diff --git a/instrumentation/opentelemetry-instrumentation-aiopg/tests/test_aiopg_integration.py b/instrumentation/opentelemetry-instrumentation-aiopg/tests/test_aiopg_integration.py index a0aa2f85dd..28e932add5 100644 --- a/instrumentation/opentelemetry-instrumentation-aiopg/tests/test_aiopg_integration.py +++ b/instrumentation/opentelemetry-instrumentation-aiopg/tests/test_aiopg_integration.py @@ -33,8 +33,13 @@ NET_PEER_NAME, NET_PEER_PORT, ) +from opentelemetry.semconv._incubating.metrics.db_metrics import ( + DB_CLIENT_OPERATION_DURATION, + DB_CLIENT_RESPONSE_RETURNED_ROWS, +) from opentelemetry.semconv.attributes.db_attributes import ( DB_NAMESPACE, + DB_OPERATION_NAME, DB_QUERY_TEXT, DB_SYSTEM_NAME, ) @@ -68,12 +73,14 @@ def use_semconv_opt_in(sem_conv_mode): class TestAiopgInstrumentor(TestBase): def setUp(self): super().setUp() + _OpenTelemetrySemanticConventionStability._initialized = False self.origin_aiopg_connect = aiopg.connect self.origin_aiopg_create_pool = aiopg.create_pool aiopg.connect = mock_connect aiopg.create_pool = mock_create_pool def tearDown(self): + _OpenTelemetrySemanticConventionStability._initialized = False super().tearDown() aiopg.connect = self.origin_aiopg_connect aiopg.create_pool = self.origin_aiopg_create_pool @@ -246,6 +253,24 @@ def test_custom_tracer_provider_create_pool(self): self.assertIs(span.resource, resource) + @mock.patch.dict("os.environ", {OTEL_SEMCONV_STABILITY_OPT_IN: "database"}) + def test_custom_meter_provider_connect(self): + meter_provider, metrics_reader = self.create_meter_provider() + AiopgInstrumentor().instrument(meter_provider=meter_provider) + + cnx = async_call(aiopg.connect(database="test")) + cursor = async_call(cnx.cursor()) + async_call(cursor.execute("SELECT * FROM test", rowcount=3)) + + metric_names = { + metric.name + for resource_metrics in metrics_reader.get_metrics_data().resource_metrics + for scope_metrics in resource_metrics.scope_metrics + for metric in scope_metrics.metrics + } + self.assertIn(DB_CLIENT_OPERATION_DURATION, metric_names) + self.assertIn(DB_CLIENT_RESPONSE_RETURNED_ROWS, metric_names) + def test_instrument_connection(self): cnx = async_call(aiopg.connect(database="test")) query = "SELECT * FROM test" @@ -680,6 +705,127 @@ def test_uninstrument_connection(self): self.assertIs(connection4, connection) +class TestAiopgMetrics(TestBase): + def setUp(self): + super().setUp() + _OpenTelemetrySemanticConventionStability._initialized = False + + def tearDown(self): + _OpenTelemetrySemanticConventionStability._initialized = False + super().tearDown() + + def _get_metric(self, name): + return next( + ( + metric + for metric in self.get_sorted_metrics() + if metric.name == name + ), + None, + ) + + def test_metrics_not_emitted_in_default_mode(self): + db_integration = AiopgIntegration(__name__, "testcomponent") + mock_connection = async_call( + db_integration.wrapped_connection(mock_connect, (), {}) + ) + cursor = async_call(mock_connection.cursor()) + async_call(cursor.execute("SELECT 1", rowcount=3)) + + self.assertIsNone(self._get_metric(DB_CLIENT_OPERATION_DURATION)) + self.assertIsNone(self._get_metric(DB_CLIENT_RESPONSE_RETURNED_ROWS)) + + @mock.patch.dict("os.environ", {OTEL_SEMCONV_STABILITY_OPT_IN: "database"}) + def test_operation_duration_recorded(self): + connection_props = { + "database": "testdatabase", + "server_host": "testhost", + "server_port": 123, + "user": "testuser", + } + connection_attributes = { + "database": "database", + "port": "server_port", + "host": "server_host", + "user": "user", + } + db_integration = AiopgIntegration( + __name__, "testcomponent", connection_attributes + ) + mock_connection = async_call( + db_integration.wrapped_connection( + mock_connect, (), connection_props + ) + ) + cursor = async_call(mock_connection.cursor()) + async_call(cursor.execute("SELECT * FROM users")) + + duration_metric = self._get_metric(DB_CLIENT_OPERATION_DURATION) + self.assertIsNotNone(duration_metric) + self.assertEqual(duration_metric.unit, "s") + points = list(duration_metric.data.data_points) + self.assertEqual(len(points), 1) + attributes = dict(points[0].attributes) + self.assertEqual(attributes[DB_SYSTEM_NAME], "testcomponent") + self.assertEqual(attributes[DB_NAMESPACE], "testdatabase") + self.assertEqual(attributes[DB_OPERATION_NAME], "SELECT") + self.assertEqual(attributes[SERVER_ADDRESS], "testhost") + self.assertEqual(attributes[SERVER_PORT], 123) + self.assertNotIn(ERROR_TYPE, attributes) + self.assertEqual(points[0].count, 1) + self.assertGreaterEqual(points[0].sum, 0.0) + + @mock.patch.dict("os.environ", {OTEL_SEMCONV_STABILITY_OPT_IN: "database"}) + def test_operation_duration_error_type(self): + db_integration = AiopgIntegration(__name__, "testcomponent") + mock_connection = async_call( + db_integration.wrapped_connection(mock_connect, (), {}) + ) + cursor = async_call(mock_connection.cursor()) + with self.assertRaises(psycopg2.ProgrammingError): + async_call(cursor.execute("SELECT 1", throw_exception=True)) + + duration_metric = self._get_metric(DB_CLIENT_OPERATION_DURATION) + self.assertIsNotNone(duration_metric) + points = list(duration_metric.data.data_points) + self.assertEqual(len(points), 1) + attributes = dict(points[0].attributes) + self.assertEqual(attributes[ERROR_TYPE], "ProgrammingError") + self.assertEqual(attributes[DB_OPERATION_NAME], "SELECT") + self.assertIsNone(self._get_metric(DB_CLIENT_RESPONSE_RETURNED_ROWS)) + + @mock.patch.dict("os.environ", {OTEL_SEMCONV_STABILITY_OPT_IN: "database"}) + def test_returned_rows_recorded_for_executemany(self): + db_integration = AiopgIntegration(__name__, "testcomponent") + mock_connection = async_call( + db_integration.wrapped_connection(mock_connect, (), {}) + ) + cursor = async_call(mock_connection.cursor()) + async_call( + cursor.executemany("INSERT INTO users VALUES (1)", rowcount=3) + ) + + rows_metric = self._get_metric(DB_CLIENT_RESPONSE_RETURNED_ROWS) + self.assertIsNotNone(rows_metric) + self.assertEqual(rows_metric.unit, "{row}") + points = list(rows_metric.data.data_points) + self.assertEqual(len(points), 1) + self.assertEqual(points[0].sum, 3) + self.assertEqual(points[0].count, 1) + + @mock.patch.dict("os.environ", {OTEL_SEMCONV_STABILITY_OPT_IN: "database"}) + def test_returned_rows_skipped_when_rowcount_unknown(self): + db_integration = AiopgIntegration(__name__, "testcomponent") + mock_connection = async_call( + db_integration.wrapped_connection(mock_connect, (), {}) + ) + cursor = async_call(mock_connection.cursor()) + async_call(cursor.execute("CREATE TABLE users (id INT)")) + + self.assertIsNone(self._get_metric(DB_CLIENT_RESPONSE_RETURNED_ROWS)) + self.assertIsNotNone(self._get_metric(DB_CLIENT_OPERATION_DURATION)) + + # pylint: disable=unused-argument async def mock_connect(*args, **kwargs): database = kwargs.get("database") @@ -760,18 +906,30 @@ def close(self): class MockCursor: - # pylint: disable=unused-argument, no-self-use - async def execute(self, query, params=None, throw_exception=False): + def __init__(self): + self.rowcount = -1 + + # pylint: disable=unused-argument + async def execute( + self, query, params=None, throw_exception=False, rowcount=-1 + ): + self.rowcount = rowcount if throw_exception: raise psycopg2.ProgrammingError("Test Exception") - # pylint: disable=unused-argument, no-self-use - async def executemany(self, query, params=None, throw_exception=False): + # pylint: disable=unused-argument + async def executemany( + self, query, params=None, throw_exception=False, rowcount=-1 + ): + self.rowcount = rowcount if throw_exception: raise psycopg2.ProgrammingError("Test Exception") - # pylint: disable=unused-argument, no-self-use - async def callproc(self, query, params=None, throw_exception=False): + # pylint: disable=unused-argument + async def callproc( + self, query, params=None, throw_exception=False, rowcount=-1 + ): + self.rowcount = rowcount if throw_exception: raise psycopg2.ProgrammingError("Test Exception")