opentelemetry-instrumentation-aiopg: fix manual connection attributes

.changelog/4821.fixed

@@ -0,0 +1 @@
+`opentelemetry-instrumentation-aiopg`: fix connection metadata for manual instrumentation

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

@@ -109,9 +109,7 @@ async def traced_execution(
         *args: typing.Tuple[typing.Any, typing.Any],
         **kwargs: typing.Dict[typing.Any, typing.Any],
     ):
-        name = ""
-        if args:
-            name = self.get_operation_name(cursor, args)
+        name = self.get_operation_name(cursor, args)
 
         if not name:
             name = (

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

@@ -153,7 +153,9 @@ def instrument_connection(
         version=version,
         tracer_provider=tracer_provider,
     )
-    db_integration.get_connection_attributes(connection)
+    db_integration.get_connection_attributes(
+        getattr(connection, "_conn", connection)
+    )
     return get_traced_connection_proxy(connection, db_integration)
 
 

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

@@ -247,7 +247,14 @@ def test_custom_tracer_provider_create_pool(self):
         self.assertIs(span.resource, resource)
 
     def test_instrument_connection(self):
-        cnx = async_call(aiopg.connect(database="test"))
+        cnx = async_call(
+            aiopg.connect(
+                database="testdatabase",
+                server_host="testhost",
+                server_port=123,
+                user="testuser",
+            )
+        )
         query = "SELECT * FROM test"
         cursor = async_call(cnx.cursor())
         async_call(cursor.execute(query))
@@ -261,6 +268,11 @@ def test_instrument_connection(self):
 
         spans_list = self.memory_exporter.get_finished_spans()
         self.assertEqual(len(spans_list), 1)
+        span = spans_list[0]
+        self.assertEqual(span.attributes[DB_NAME], "testdatabase")
+        self.assertEqual(span.attributes[DB_USER], "testuser")
+        self.assertEqual(span.attributes[NET_PEER_NAME], "testhost")
+        self.assertEqual(span.attributes[NET_PEER_PORT], 123)
 
     def test_instrument_connection_new_semconv(self):
         with use_semconv_opt_in("database,http"):
@@ -657,6 +669,7 @@ def test_instrument_connection(self):
         connection = mock.Mock()
         # Avoid get_attributes failing because can't concatenate mock
         connection.database = "-"
+        connection._conn = connection
         connection2 = wrappers.instrument_connection(
             self.tracer, connection, "-"
         )
@@ -667,6 +680,7 @@ def test_uninstrument_connection(self):
         # Set connection.database to avoid a failure because mock can't
         # be concatenated
         connection.database = "-"
+        connection._conn = connection
         connection2 = wrappers.instrument_connection(
             self.tracer, connection, "-"
         )

tests/opentelemetry-docker-tests/tests/postgres/test_aiopg_functional.py

@@ -103,6 +103,46 @@ def test_callproc(self):
             self.validate_spans("test")
 
 
+class TestFunctionalAiopgInstrumentConnection(TestBase):
+    def setUp(self):
+        super().setUp()
+
+        async def connect():
+            connection = await aiopg.connect(
+                dbname=POSTGRES_DB_NAME,
+                user=POSTGRES_USER,
+                password=POSTGRES_PASSWORD,
+                host=POSTGRES_HOST,
+                port=POSTGRES_PORT,
+            )
+            return AiopgInstrumentor().instrument_connection(
+                connection, tracer_provider=self.tracer_provider
+            )
+
+        self._connection = async_call(connect())
+        self._cursor = async_call(self._connection.cursor())
+
+    def tearDown(self):
+        self._cursor.close()
+        self._connection.close()
+        super().tearDown()
+
+    def test_execute(self):
+        """Should emit connection attributes for a manually instrumented connection."""
+        async_call(self._cursor.execute("SELECT 1"))
+
+        spans = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans), 1)
+        span = spans[0]
+        self.assertEqual(span.name, "SELECT")
+        self.assertEqual(span.kind, trace_api.SpanKind.CLIENT)
+        self.assertEqual(span.attributes[DB_SYSTEM], "postgresql")
+        self.assertEqual(span.attributes[DB_NAME], POSTGRES_DB_NAME)
+        self.assertEqual(span.attributes[DB_USER], POSTGRES_USER)
+        self.assertEqual(span.attributes[NET_PEER_NAME], POSTGRES_HOST)
+        self.assertEqual(span.attributes[NET_PEER_PORT], POSTGRES_PORT)
+
+
 class TestFunctionalAiopgCreatePool(TestBase):
     def setUp(self):
         super().setUp()