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

CHANGELOG.md

@@ -19,6 +19,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#2082](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/2082))
 - `opentelemetry-instrumentation-redis` Add additional attributes for methods create_index and search, rename those spans
   ([#2635](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/2635))
+- `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,32 @@ 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(?: IF NOT EXISTS)?)\s+(['`]?(\w+)['`]?(?:\s*\.\s*['`]?(\w+)['`]?)?)",
+            statement,
+        )
+        if match:
+            collection_name = match.group(1)
+
+        return collection_name
+
     def get_statement(self, cursor, args):  # pylint: disable=no-self-use
         if not args:
             return ""
@@ -412,7 +445,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), 10)
         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 = {

instrumentation/opentelemetry-instrumentation-sqlite3/tests/test_sqlite3.py

@@ -72,11 +72,11 @@ def test_execute(self):
         stmt = "CREATE TABLE IF NOT EXISTS test (id integer)"
         with self._tracer.start_as_current_span("rootSpan"):
             self._cursor.execute(stmt)
-        self.validate_spans("CREATE")
+        self.validate_spans("CREATE test")
 
         with self._tracer.start_as_current_span("rootSpan"):
             self._cursor2.execute(stmt)
-        self.validate_spans("CREATE")
+        self.validate_spans("CREATE test")
 
     def test_executemany(self):
         """Should create a child span for executemany"""
@@ -87,11 +87,11 @@ def test_executemany(self):
         data = [("1",), ("2",), ("3",)]
         with self._tracer.start_as_current_span("rootSpan"):
             self._cursor.executemany(stmt, data)
-        self.validate_spans("INSERT")
+        self.validate_spans("INSERT test")
 
         with self._tracer.start_as_current_span("rootSpan"):
             self._cursor2.executemany(stmt, data)
-        self.validate_spans("INSERT")
+        self.validate_spans("INSERT test")
 
     def test_callproc(self):
         """Should create a child span for callproc"""

tests/opentelemetry-docker-tests/tests/mysql/test_mysql_functional.py

@@ -84,7 +84,7 @@ def test_execute(self):
         stmt = "CREATE TABLE IF NOT EXISTS test (id INT)"
         with self._tracer.start_as_current_span("rootSpan"):
             self._cursor.execute(stmt)
-        self.validate_spans("CREATE")
+        self.validate_spans("CREATE test")
 
     def test_execute_with_connection_context_manager(self):
         """Should create a child span for execute with connection context"""
@@ -93,23 +93,23 @@ def test_execute_with_connection_context_manager(self):
             with self._connection as conn:
                 cursor = conn.cursor()
                 cursor.execute(stmt)
-        self.validate_spans("CREATE")
+        self.validate_spans("CREATE test")
 
     def test_execute_with_cursor_context_manager(self):
         """Should create a child span for execute with cursor context"""
         stmt = "CREATE TABLE IF NOT EXISTS test (id INT)"
         with self._tracer.start_as_current_span("rootSpan"):
             with self._connection.cursor() as cursor:
                 cursor.execute(stmt)
-        self.validate_spans("CREATE")
+        self.validate_spans("CREATE test")
 
     def test_executemany(self):
         """Should create a child span for executemany"""
         stmt = "INSERT INTO test (id) VALUES (%s)"
         with self._tracer.start_as_current_span("rootSpan"):
             data = (("1",), ("2",), ("3",))
             self._cursor.executemany(stmt, data)
-        self.validate_spans("INSERT")
+        self.validate_spans("INSERT test")
 
     def test_callproc(self):
         """Should create a child span for callproc"""

tests/opentelemetry-docker-tests/tests/pymysql/test_pymysql_functional.py

@@ -84,23 +84,23 @@ def test_execute(self):
         stmt = "CREATE TABLE IF NOT EXISTS test (id INT)"
         with self._tracer.start_as_current_span("rootSpan"):
             self._cursor.execute(stmt)
-        self.validate_spans("CREATE")
+        self.validate_spans("CREATE test")
 
     def test_execute_with_cursor_context_manager(self):
         """Should create a child span for execute with cursor context"""
         stmt = "CREATE TABLE IF NOT EXISTS test (id INT)"
         with self._tracer.start_as_current_span("rootSpan"):
             with self._connection.cursor() as cursor:
                 cursor.execute(stmt)
-        self.validate_spans("CREATE")
+        self.validate_spans("CREATE test")
 
     def test_executemany(self):
         """Should create a child span for executemany"""
         stmt = "INSERT INTO test (id) VALUES (%s)"
         with self._tracer.start_as_current_span("rootSpan"):
             data = (("1",), ("2",), ("3",))
             self._cursor.executemany(stmt, data)
-        self.validate_spans("INSERT")
+        self.validate_spans("INSERT test")
 
     def test_callproc(self):
         """Should create a child span for callproc"""
@@ -116,12 +116,12 @@ def test_commit(self):
             data = (("4",), ("5",), ("6",))
             self._cursor.executemany(stmt, data)
             self._connection.commit()
-        self.validate_spans("INSERT")
+        self.validate_spans("INSERT test")
 
     def test_rollback(self):
         stmt = "INSERT INTO test (id) VALUES (%s)"
         with self._tracer.start_as_current_span("rootSpan"):
             data = (("7",), ("8",), ("9",))
             self._cursor.executemany(stmt, data)
             self._connection.rollback()
-        self.validate_spans("INSERT")
+        self.validate_spans("INSERT test")