dbapi: Add db.collection.name, use connection kwargs for connection attributes

CHANGELOG.md

@@ -13,6 +13,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#2860](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/2860))
 - `opentelemetry-instrumentation-aiokafka` Add instrumentor and auto instrumentation support for aiokafka
   ([#2082](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/2082))
+- `opentelemetry-instrumentation-dbapi` Add db.collection.name, use connection kwargs for connection attributes
+  ([#2869](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/2869))
 
 
 ### Fixed

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

@@ -51,6 +51,9 @@
     _get_opentelemetry_values,
     unwrap,
 )
+from opentelemetry.semconv._incubating.attributes.db_attributes import (
+    DB_COLLECTION_NAME,
+)
 from opentelemetry.semconv.trace import SpanAttributes
 from opentelemetry.trace import SpanKind, TracerProvider, get_tracer
 
@@ -217,7 +220,7 @@ def instrument_connection(
         enable_commenter=enable_commenter,
         commenter_options=commenter_options,
     )
-    db_integration.get_connection_attributes(connection)
+    db_integration.get_connection_attributes(connection=connection)
     return get_traced_connection_proxy(connection, db_integration)
 
 
@@ -284,12 +287,17 @@ def wrapped_connection(
     ):
         """Add object proxy to connection object."""
         connection = connect_method(*args, **kwargs)
-        self.get_connection_attributes(connection)
+        self.get_connection_attributes(connection=connection, kwargs=kwargs)
         return get_traced_connection_proxy(connection, self)
 
-    def get_connection_attributes(self, connection):
-        # Populate span fields using connection
+    def get_connection_attributes(self, connection, kwargs=None):
+        # Populate span fields using kwargs and connection
         for key, value in self.connection_attributes.items():
+            # First set from kwargs
+            if kwargs and value in kwargs:
+                self.connection_props[key] = kwargs.get(value)
+
+            # Then override from connection object
             # Allow attributes nested in connection object
             attribute = functools.reduce(
                 lambda attribute, attribute_value: getattr(
@@ -373,14 +381,19 @@ def _populate_span(
     ):
         if not span.is_recording():
             return
+
         statement = self.get_statement(cursor, args)
+        collection_name = self.get_collection_name(statement)
+
         span.set_attribute(
             SpanAttributes.DB_SYSTEM, self._db_api_integration.database_system
         )
         span.set_attribute(
             SpanAttributes.DB_NAME, self._db_api_integration.database
         )
         span.set_attribute(SpanAttributes.DB_STATEMENT, statement)
+        if collection_name:
+            span.set_attribute(DB_COLLECTION_NAME, collection_name)
 
         for (
             attribute_key,
@@ -391,12 +404,29 @@ def _populate_span(
         if self._db_api_integration.capture_parameters and len(args) > 1:
             span.set_attribute("db.statement.parameters", str(args[1]))
 
-    def get_operation_name(self, cursor, args):  # pylint: disable=no-self-use
+    def get_span_name(self, cursor, args):
+        operation_name = self.get_operation_name(cursor, args)
+        statement = self.get_statement(cursor, args)
+        collection_name = CursorTracer.get_collection_name(statement)
+        return " ".join(
+            name for name in (operation_name, collection_name) if name
+        )
+
+    def get_operation_name(self, cursor, args):
         if args and isinstance(args[0], str):
             # Strip leading comments so we get the operation name.
             return self._leading_comment_remover.sub("", args[0]).split()[0]
         return ""
 
+    @staticmethod
+    def get_collection_name(statement):
+        collection_name = ""
+        match = re.search(r"\b(?:FROM|JOIN|INTO|UPDATE|TABLE)\s+(['`]?(\w+)['`]?(?:\s*\.\s*['`]?(\w+)['`]?)?)", statement)
+        if match:
+            collection_name = match.group(1).replace("'`", "")
+
+        return collection_name
+
     def get_statement(self, cursor, args):  # pylint: disable=no-self-use
         if not args:
             return ""
@@ -412,7 +442,7 @@ def traced_execution(
         *args: typing.Tuple[typing.Any, typing.Any],
         **kwargs: typing.Dict[typing.Any, typing.Any],
     ):
-        name = self.get_operation_name(cursor, args)
+        name = self.get_span_name(cursor, args)
         if not name:
             name = (
                 self._db_api_integration.database

instrumentation/opentelemetry-instrumentation-dbapi/tests/test_dbapi_integration.py

@@ -20,6 +20,9 @@
 from opentelemetry import trace as trace_api
 from opentelemetry.instrumentation import dbapi
 from opentelemetry.sdk import resources
+from opentelemetry.semconv._incubating.attributes.db_attributes import (
+    DB_COLLECTION_NAME,
+)
 from opentelemetry.semconv.trace import SpanAttributes
 from opentelemetry.test.test_base import TestBase
 
@@ -49,11 +52,12 @@ def test_span_succeeded(self):
             mock_connect, {}, connection_props
         )
         cursor = mock_connection.cursor()
-        cursor.execute("Test query", ("param1Value", False))
+        expected_query = "Test query FROM test_table"
+        cursor.execute(expected_query, ("param1Value", False))
         spans_list = self.memory_exporter.get_finished_spans()
         self.assertEqual(len(spans_list), 1)
         span = spans_list[0]
-        self.assertEqual(span.name, "Test")
+        self.assertEqual(span.name, "Test test_table")
         self.assertIs(span.kind, trace_api.SpanKind.CLIENT)
 
         self.assertEqual(
@@ -63,8 +67,9 @@ def test_span_succeeded(self):
             span.attributes[SpanAttributes.DB_NAME], "testdatabase"
         )
         self.assertEqual(
-            span.attributes[SpanAttributes.DB_STATEMENT], "Test query"
+            span.attributes[SpanAttributes.DB_STATEMENT], expected_query
         )
+        self.assertEqual(span.attributes[DB_COLLECTION_NAME], "test_table")
         self.assertFalse("db.statement.parameters" in span.attributes)
         self.assertEqual(span.attributes[SpanAttributes.DB_USER], "testuser")
         self.assertEqual(
@@ -91,14 +96,22 @@ def test_span_name(self):
         cursor.execute("/* leading comment */ query")
         cursor.execute("/* leading comment */ query /* trailing comment */")
         cursor.execute("query /* trailing comment */")
+        cursor.execute("SELECT * FROM test_table")
+        cursor.execute("SELECT * FROM schema.test_table")
+        cursor.execute("SELECT * FROM `schema`.`test_table`")
+        cursor.execute("SELECT * FROM 'schema'.'test_table'")
         spans_list = self.memory_exporter.get_finished_spans()
-        self.assertEqual(len(spans_list), 6)
+        self.assertEqual(len(spans_list), 9)
         self.assertEqual(spans_list[0].name, "Test")
         self.assertEqual(spans_list[1].name, "multi")
         self.assertEqual(spans_list[2].name, "tab")
         self.assertEqual(spans_list[3].name, "query")
         self.assertEqual(spans_list[4].name, "query")
         self.assertEqual(spans_list[5].name, "query")
+        self.assertEqual(spans_list[6].name, "SELECT test_table")
+        self.assertEqual(spans_list[7].name, "SELECT schema.test_table")
+        self.assertEqual(spans_list[8].name, "SELECT schema.test_table")
+        self.assertEqual(spans_list[9].name, "SELECT schema.test_table")
 
     def test_span_succeeded_with_capture_of_statement_parameters(self):
         connection_props = {