opentelemetry-instrumentation-aiopg: semantic convention stability migration (#1636) (#4820)

Co-authored-by: Riccardo Magliocchetti <riccardo.magliocchetti@gmail.com>

Author: henry3260

.changelog/4820.added

@@ -0,0 +1 @@
+`opentelemetry-instrumentation-aiopg`: add database semantic convention stability support through `OTEL_SEMCONV_STABILITY_OPT_IN`

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 | development
+| [opentelemetry-instrumentation-aiopg](./opentelemetry-instrumentation-aiopg) | aiopg >= 0.13.0, < 2.0.0 | No | 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/aiopg_integration.py

@@ -7,10 +7,12 @@
 
 import wrapt
 
+from opentelemetry.instrumentation._semconv import _report_new
 from opentelemetry.instrumentation.dbapi import (
     CursorTracer,
     DatabaseApiIntegration,
 )
+from opentelemetry.semconv.attributes.error_attributes import ERROR_TYPE
 from opentelemetry.trace import SpanKind
 
 
@@ -121,8 +123,16 @@ async def traced_execution(
         with self._db_api_integration._tracer.start_as_current_span(
             name, kind=SpanKind.CLIENT
         ) as span:
-            self._populate_span(span, cursor, *args)
-            return await query_method(*args, **kwargs)
+            if span.is_recording():
+                self._populate_span(span, cursor, *args)
+            try:
+                return await query_method(*args, **kwargs)
+            except Exception as 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
 
 
 def get_traced_cursor_proxy(cursor, db_api_integration, *args, **kwargs):

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

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

instrumentation/opentelemetry-instrumentation-aiopg/tests/test_aiopg_integration.py

@@ -1,7 +1,9 @@
 # Copyright The OpenTelemetry Authors
 # SPDX-License-Identifier: Apache-2.0
 import asyncio
+import contextlib
 import logging
+from types import SimpleNamespace
 from unittest import mock
 from unittest.mock import MagicMock
 
@@ -10,6 +12,10 @@
 
 import opentelemetry.instrumentation.aiopg
 from opentelemetry import trace as trace_api
+from opentelemetry.instrumentation._semconv import (
+    OTEL_SEMCONV_STABILITY_OPT_IN,
+    _OpenTelemetrySemanticConventionStability,
+)
 from opentelemetry.instrumentation.aiopg import AiopgInstrumentor, wrappers
 from opentelemetry.instrumentation.aiopg.aiopg_integration import (
     AiopgIntegration,
@@ -27,13 +33,38 @@
     NET_PEER_NAME,
     NET_PEER_PORT,
 )
+from opentelemetry.semconv.attributes.db_attributes import (
+    DB_NAMESPACE,
+    DB_QUERY_TEXT,
+    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.test.test_base import TestBase
 
 
 def async_call(coro):
     return asyncio.run(coro)
 
 
+@contextlib.contextmanager
+def use_semconv_opt_in(sem_conv_mode):
+    env_patch = mock.patch.dict(
+        "os.environ",
+        {OTEL_SEMCONV_STABILITY_OPT_IN: sem_conv_mode},
+    )
+    _OpenTelemetrySemanticConventionStability._initialized = False
+    env_patch.start()
+    try:
+        yield
+    finally:
+        env_patch.stop()
+        _OpenTelemetrySemanticConventionStability._initialized = False
+
+
 class TestAiopgInstrumentor(TestBase):
     def setUp(self):
         super().setUp()
@@ -79,6 +110,34 @@ def test_instrumentor_connect(self):
         spans_list = self.memory_exporter.get_finished_spans()
         self.assertEqual(len(spans_list), 1)
 
+    def test_instrumentor_connect_new_semconv(self):
+        with use_semconv_opt_in("database,http"):
+            AiopgInstrumentor().instrument()
+
+            cnx = async_call(
+                aiopg.connect(
+                    database="testdatabase",
+                    server_host="testhost",
+                    server_port=123,
+                )
+            )
+            cursor = async_call(cnx.cursor())
+            async_call(cursor.execute("SELECT * FROM test"))
+
+        spans_list = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans_list), 1)
+        span = spans_list[0]
+        self.assertEqual(span.attributes[DB_SYSTEM_NAME], "postgresql")
+        self.assertEqual(span.attributes[DB_NAMESPACE], "testdatabase")
+        self.assertEqual(span.attributes[DB_QUERY_TEXT], "SELECT * FROM test")
+        self.assertEqual(span.attributes[SERVER_ADDRESS], "testhost")
+        self.assertEqual(span.attributes[SERVER_PORT], 123)
+        self.assertNotIn(DB_SYSTEM, span.attributes)
+        self.assertNotIn(DB_NAME, span.attributes)
+        self.assertNotIn(DB_STATEMENT, span.attributes)
+        self.assertNotIn(NET_PEER_NAME, span.attributes)
+        self.assertNotIn(NET_PEER_PORT, span.attributes)
+
     def test_instrumentor_connect_ctx_manager(self):
         async def _ctx_manager_connect():
             AiopgInstrumentor().instrument()
@@ -203,6 +262,27 @@ def test_instrument_connection(self):
         spans_list = self.memory_exporter.get_finished_spans()
         self.assertEqual(len(spans_list), 1)
 
+    def test_instrument_connection_new_semconv(self):
+        with use_semconv_opt_in("database,http"):
+            cnx = async_call(
+                aiopg.connect(
+                    database="testdatabase",
+                    server_host="testhost",
+                    server_port=123,
+                )
+            )
+            cnx = AiopgInstrumentor().instrument_connection(cnx)
+            cursor = async_call(cnx.cursor())
+            async_call(cursor.execute("SELECT * FROM test"))
+
+        spans_list = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans_list), 1)
+        span = spans_list[0]
+        self.assertEqual(span.attributes[DB_SYSTEM_NAME], "postgresql")
+        self.assertEqual(span.attributes[DB_QUERY_TEXT], "SELECT * FROM test")
+        self.assertNotIn(DB_SYSTEM, span.attributes)
+        self.assertNotIn(DB_STATEMENT, span.attributes)
+
     def test_instrument_connection_after_instrument(self):
         cnx = async_call(aiopg.connect(database="test"))
         query = "SELECT * FROM test"
@@ -280,8 +360,13 @@ def test_uninstrument_connection(self):
 class TestAiopgIntegration(TestBase):
     def setUp(self):
         super().setUp()
+        _OpenTelemetrySemanticConventionStability._initialized = False
         self.tracer = self.tracer_provider.get_tracer(__name__)
 
+    def tearDown(self):
+        _OpenTelemetrySemanticConventionStability._initialized = False
+        super().tearDown()
+
     def test_span_succeeded(self):
         connection_props = {
             "database": "testdatabase",
@@ -339,12 +424,11 @@ def test_span_not_recording(self):
             "host": "server_host",
             "user": "user",
         }
-        mock_tracer = mock.Mock()
-        mock_span = mock.Mock()
-        mock_span.is_recording.return_value = False
-        mock_tracer.start_span.return_value = mock_span
         db_integration = AiopgIntegration(
-            mock_tracer, "testcomponent", connection_attributes
+            __name__,
+            "testcomponent",
+            connection_attributes,
+            tracer_provider=trace_api.NoOpTracerProvider(),
         )
         mock_connection = async_call(
             db_integration.wrapped_connection(
@@ -353,10 +437,108 @@ def test_span_not_recording(self):
         )
         cursor = async_call(mock_connection.cursor())
         async_call(cursor.execute("Test query", ("param1Value", False)))
-        self.assertFalse(mock_span.is_recording())
-        self.assertTrue(mock_span.is_recording.called)
-        self.assertFalse(mock_span.set_attribute.called)
-        self.assertFalse(mock_span.set_status.called)
+        self.assertEqual(len(self.memory_exporter.get_finished_spans()), 0)
+
+    def test_span_succeeded_new_semconv(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",
+        }
+        with use_semconv_opt_in("database,http"):
+            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("Test query"))
+
+        spans_list = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans_list), 1)
+        span = spans_list[0]
+        self.assertEqual(
+            span.instrumentation_scope.schema_url,
+            "https://opentelemetry.io/schemas/1.25.0",
+        )
+        self.assertEqual(span.attributes[DB_SYSTEM_NAME], "testcomponent")
+        self.assertEqual(span.attributes[DB_NAMESPACE], "testdatabase")
+        self.assertEqual(span.attributes[DB_QUERY_TEXT], "Test query")
+        self.assertEqual(span.attributes[SERVER_ADDRESS], "testhost")
+        self.assertEqual(span.attributes[SERVER_PORT], 123)
+        self.assertNotIn(DB_SYSTEM, span.attributes)
+        self.assertNotIn(DB_NAME, span.attributes)
+        self.assertNotIn(DB_STATEMENT, span.attributes)
+        self.assertNotIn(DB_USER, span.attributes)
+        self.assertNotIn(NET_PEER_NAME, span.attributes)
+        self.assertNotIn(NET_PEER_PORT, span.attributes)
+
+    def test_span_succeeded_both_semconv(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",
+        }
+        with use_semconv_opt_in("database/dup,http/dup"):
+            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("Test query"))
+
+        spans_list = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans_list), 1)
+        span = spans_list[0]
+        self.assertEqual(span.attributes[DB_SYSTEM], "testcomponent")
+        self.assertEqual(span.attributes[DB_NAME], "testdatabase")
+        self.assertEqual(span.attributes[DB_STATEMENT], "Test query")
+        self.assertEqual(span.attributes[DB_USER], "testuser")
+        self.assertEqual(span.attributes[NET_PEER_NAME], "testhost")
+        self.assertEqual(span.attributes[NET_PEER_PORT], 123)
+        self.assertEqual(span.attributes[DB_SYSTEM_NAME], "testcomponent")
+        self.assertEqual(span.attributes[DB_NAMESPACE], "testdatabase")
+        self.assertEqual(span.attributes[DB_QUERY_TEXT], "Test query")
+        self.assertEqual(span.attributes[SERVER_ADDRESS], "testhost")
+        self.assertEqual(span.attributes[SERVER_PORT], 123)
+
+    def test_executemany_new_semconv(self):
+        with use_semconv_opt_in("database"):
+            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("Test query"))
+
+        spans_list = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans_list), 1)
+        span = spans_list[0]
+        self.assertEqual(span.attributes[DB_SYSTEM_NAME], "testcomponent")
+        self.assertEqual(span.attributes[DB_QUERY_TEXT], "Test query")
+        self.assertNotIn(DB_SYSTEM, span.attributes)
+        self.assertNotIn(DB_STATEMENT, span.attributes)
 
     def test_span_failed(self):
         db_integration = AiopgIntegration(self.tracer, "testcomponent")
@@ -376,6 +558,23 @@ def test_span_failed(self):
             span.status.description, "ProgrammingError: Test Exception"
         )
 
+    def test_span_failed_new_semconv(self):
+        with use_semconv_opt_in("database"):
+            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("Test query", throw_exception=True))
+
+        spans_list = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans_list), 1)
+        span = spans_list[0]
+        self.assertEqual(span.attributes[DB_QUERY_TEXT], "Test query")
+        self.assertEqual(span.attributes[ERROR_TYPE], "ProgrammingError")
+        self.assertIs(span.status.status_code, trace_api.StatusCode.ERROR)
+
     def test_executemany(self):
         db_integration = AiopgIntegration(self.tracer, "testcomponent")
         mock_connection = async_call(
@@ -534,6 +733,12 @@ def __init__(self, database, server_port, server_host, user):
         self.server_port = server_port
         self.server_host = server_host
         self.user = user
+        self.info = SimpleNamespace(
+            dbname=database,
+            port=server_port,
+            host=server_host,
+            user=user,
+        )
 
 
 class MockConnection: