Updated dbapi and psycopg2 instrumentations.

Changes:

- Update dbapi instrumentation to use the SQL statement name as the span
instead of the entire SQL query.
- Renamed TracedCursor with CursorTracing. The class was not a valid
Cursor so the name was confusing.
- Updated CursorTracing's (previously TracedCursor) traced_execution
method to accept the cursor instance as the first argument. This is
required as for some dbapi implementations, we need a reference to the
cursor in order to correctly format the SQL query.
- Updated psycopg2 instrumentation to leverage dbapi's `cursor_factory`
mechanism instead of wrapping the cursor with wrapt. This results in a
simpler instrumentation without monkey patching objects at runtime and
allows psycopg2's type registration system to work. This should make it
possible to use psycopg2 instrumentation when using the JSONB feature or
with frameworks like Django.

Author: owais

.gitignore

@@ -22,7 +22,7 @@ lib64
 __pycache__
 venv*/
 .venv*/
-opentelemetry-python-core*/
+/opentelemetry-python-core*/
 /opentelemetry-python-core
 
 # Installer logs

CHANGELOG.md

@@ -41,6 +41,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#253](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/253))
 - `opentelemetry-instrumentation-requests`, `opentelemetry-instrumentation-urllib` Fix span name callback parameters
 - ([#259](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/259))
+- `opentelemetry-instrumentation-dbapi`, `TracedCursor` replaced by `CursorTracer`
+  ([#246](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/246))
+- `opentelemetry-instrumentation-psycopg2`, Added support for psycopg2 registered types.
+  ([#246](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/246))
+- `opentelemetry-instrumentation-dbapi`, `opentelemetry-instrumentation-psycopg2`, `opentelemetry-instrumentation-mysql`, `opentelemetry-instrumentation-pymysql`, `opentelemetry-instrumentation-aiopg` Use SQL command name as the span operation name instead of the entire query.
+  ([#246](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/246))
 
 ## [0.16b1](https://github.com/open-telemetry/opentelemetry-python-contrib/releases/tag/v0.16b1) - 2020-11-26
 

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

@@ -4,8 +4,8 @@
 from aiopg.utils import _ContextManager, _PoolAcquireContextManager
 
 from opentelemetry.instrumentation.dbapi import (
+    CursorTracer,
     DatabaseApiIntegration,
-    TracedCursor,
 )
 from opentelemetry.trace import SpanKind
 from opentelemetry.trace.status import Status, StatusCode
@@ -94,25 +94,29 @@ async def _acquire(self):
     return TracedPoolProxy(pool, *args, **kwargs)
 
 
-class AsyncTracedCursor(TracedCursor):
+class AsyncCursorTracer(CursorTracer):
     async def traced_execution(
         self,
+        cursor,
         query_method: typing.Callable[..., typing.Any],
         *args: typing.Tuple[typing.Any, typing.Any],
         **kwargs: typing.Dict[typing.Any, typing.Any]
     ):
         name = ""
-        if len(args) > 0 and args[0]:
-            name = args[0]
-        elif self._db_api_integration.database:
-            name = self._db_api_integration.database
-        else:
-            name = self._db_api_integration.name
+        if args:
+            name = self.get_operation_name(cursor, args)
+
+        if not name:
+            name = (
+                self._db_api_integration.database
+                if self._db_api_integration.database
+                else self._db_api_integration.name
+            )
 
         with self._db_api_integration.get_tracer().start_as_current_span(
             name, kind=SpanKind.CLIENT
         ) as span:
-            self._populate_span(span, *args)
+            self._populate_span(cursor, span, *args)
             try:
                 result = await query_method(*args, **kwargs)
                 return result
@@ -123,7 +127,7 @@ async def traced_execution(
 
 
 def get_traced_cursor_proxy(cursor, db_api_integration, *args, **kwargs):
-    _traced_cursor = AsyncTracedCursor(db_api_integration)
+    _traced_cursor = AsyncCursorTracer(db_api_integration)
 
     # pylint: disable=abstract-method
     class AsyncTracedCursorProxy(AsyncProxyObject):
@@ -134,19 +138,19 @@ def __init__(self, cursor, *args, **kwargs):
 
         async def execute(self, *args, **kwargs):
             result = await _traced_cursor.traced_execution(
-                self.__wrapped__.execute, *args, **kwargs
+                self, self.__wrapped__.execute, *args, **kwargs
             )
             return result
 
         async def executemany(self, *args, **kwargs):
             result = await _traced_cursor.traced_execution(
-                self.__wrapped__.executemany, *args, **kwargs
+                self, self.__wrapped__.executemany, *args, **kwargs
             )
             return result
 
         async def callproc(self, *args, **kwargs):
             result = await _traced_cursor.traced_execution(
-                self.__wrapped__.callproc, *args, **kwargs
+                self, self.__wrapped__.callproc, *args, **kwargs
             )
             return result
 

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

@@ -215,7 +215,7 @@ def test_span_succeeded(self):
         spans_list = self.memory_exporter.get_finished_spans()
         self.assertEqual(len(spans_list), 1)
         span = spans_list[0]
-        self.assertEqual(span.name, "Test query")
+        self.assertEqual(span.name, "Test")
         self.assertIs(span.kind, trace_api.SpanKind.CLIENT)
 
         self.assertEqual(span.attributes["component"], "testcomponent")

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

@@ -60,6 +60,7 @@ def trace_integration(
     connection_attributes: typing.Dict = None,
     tracer_provider: typing.Optional[TracerProvider] = None,
     capture_parameters: bool = False,
+    db_api_integration_factory=None,
 ):
     """Integrate with DB API library.
     https://www.python.org/dev/peps/pep-0249/
@@ -86,6 +87,7 @@ def trace_integration(
         version=__version__,
         tracer_provider=tracer_provider,
         capture_parameters=capture_parameters,
+        db_api_integration_factory=db_api_integration_factory,
     )
 
 
@@ -99,6 +101,7 @@ def wrap_connect(
     version: str = "",
     tracer_provider: typing.Optional[TracerProvider] = None,
     capture_parameters: bool = False,
+    db_api_integration_factory=None,
 ):
     """Integrate with DB API library.
     https://www.python.org/dev/peps/pep-0249/
@@ -115,6 +118,9 @@ def wrap_connect(
         capture_parameters: Configure if db.statement.parameters should be captured.
 
     """
+    db_api_integration_factory = (
+        db_api_integration_factory or DatabaseApiIntegration
+    )
 
     # pylint: disable=unused-argument
     def wrap_connect_(
@@ -123,7 +129,7 @@ def wrap_connect_(
         args: typing.Tuple[typing.Any, typing.Any],
         kwargs: typing.Dict[typing.Any, typing.Any],
     ):
-        db_integration = DatabaseApiIntegration(
+        db_integration = db_api_integration_factory(
             name,
             database_component,
             database_type=database_type,
@@ -314,16 +320,19 @@ def __exit__(self, *args, **kwargs):
     return TracedConnectionProxy(connection, *args, **kwargs)
 
 
-class TracedCursor:
+class CursorTracer:
     def __init__(self, db_api_integration: DatabaseApiIntegration):
         self._db_api_integration = db_api_integration
 
     def _populate_span(
-        self, span: trace_api.Span, *args: typing.Tuple[typing.Any, typing.Any]
+        self,
+        span: trace_api.Span,
+        cursor,
+        *args: typing.Tuple[typing.Any, typing.Any]
     ):
         if not span.is_recording():
             return
-        statement = args[0] if args else ""
+        statement = self.get_statement(cursor, args)
         span.set_attribute(
             "component", self._db_api_integration.database_component
         )
@@ -342,24 +351,38 @@ 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):
+        if args and isinstance(args[0], str):
+            return args[0].split(" ")[0]
+        return ""
+
+    def get_statement(self, cursor, args):
+        if not args:
+            return ""
+        statement = args[0]
+        if isinstance(statement, bytes):
+            return statement.decode("utf8", "replace")
+        return statement
+
     def traced_execution(
         self,
+        cursor,
         query_method: typing.Callable[..., typing.Any],
         *args: typing.Tuple[typing.Any, typing.Any],
         **kwargs: typing.Dict[typing.Any, typing.Any]
     ):
-        name = ""
-        if args:
-            name = args[0]
-        elif self._db_api_integration.database:
-            name = self._db_api_integration.database
-        else:
-            name = self._db_api_integration.name
+        name = self.get_operation_name(cursor, args)
+        if not name:
+            name = (
+                self._db_api_integration.database
+                if self._db_api_integration.database
+                else self._db_api_integration.name
+            )
 
         with self._db_api_integration.get_tracer().start_as_current_span(
             name, kind=SpanKind.CLIENT
         ) as span:
-            self._populate_span(span, *args)
+            self._populate_span(span, cursor, *args)
             try:
                 result = query_method(*args, **kwargs)
                 return result
@@ -370,7 +393,7 @@ def traced_execution(
 
 
 def get_traced_cursor_proxy(cursor, db_api_integration, *args, **kwargs):
-    _traced_cursor = TracedCursor(db_api_integration)
+    _cursor_tracer = CursorTracer(db_api_integration)
 
     # pylint: disable=abstract-method
     class TracedCursorProxy(wrapt.ObjectProxy):
@@ -380,18 +403,18 @@ def __init__(self, cursor, *args, **kwargs):
             wrapt.ObjectProxy.__init__(self, cursor)
 
         def execute(self, *args, **kwargs):
-            return _traced_cursor.traced_execution(
-                self.__wrapped__.execute, *args, **kwargs
+            return _cursor_tracer.traced_execution(
+                self.__wrapped__, self.__wrapped__.execute, *args, **kwargs
             )
 
         def executemany(self, *args, **kwargs):
-            return _traced_cursor.traced_execution(
-                self.__wrapped__.executemany, *args, **kwargs
+            return _cursor_tracer.traced_execution(
+                self.__wrapped__, self.__wrapped__.executemany, *args, **kwargs
             )
 
         def callproc(self, *args, **kwargs):
-            return _traced_cursor.traced_execution(
-                self.__wrapped__.callproc, *args, **kwargs
+            return _cursor_tracer.traced_execution(
+                self.__wrapped__, self.__wrapped__.callproc, *args, **kwargs
             )
 
         def __enter__(self):

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

@@ -50,7 +50,7 @@ def test_span_succeeded(self):
         spans_list = self.memory_exporter.get_finished_spans()
         self.assertEqual(len(spans_list), 1)
         span = spans_list[0]
-        self.assertEqual(span.name, "Test query")
+        self.assertEqual(span.name, "Test")
         self.assertIs(span.kind, trace_api.SpanKind.CLIENT)
 
         self.assertEqual(span.attributes["component"], "testcomponent")
@@ -93,7 +93,7 @@ def test_span_succeeded_with_capture_of_statement_parameters(self):
         spans_list = self.memory_exporter.get_finished_spans()
         self.assertEqual(len(spans_list), 1)
         span = spans_list[0]
-        self.assertEqual(span.name, "Test query")
+        self.assertEqual(span.name, "Test")
         self.assertIs(span.kind, trace_api.SpanKind.CLIENT)
 
         self.assertEqual(span.attributes["component"], "testcomponent")

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

@@ -39,12 +39,17 @@
 ---
 """
 
+import typing
+
 import psycopg2
+from psycopg2.extensions import cursor
+from psycopg2.sql import Composed
 
 from opentelemetry.instrumentation import dbapi
 from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
 from opentelemetry.instrumentation.psycopg2.version import __version__
-from opentelemetry.trace import get_tracer
+
+OTEL_CURSOR_FACTORY_KEY = "_otel_orig_cursor_factory"
 
 
 class Psycopg2Instrumentor(BaseInstrumentor):
@@ -62,7 +67,6 @@ def _instrument(self, **kwargs):
         """Integrate with PostgreSQL Psycopg library.
            Psycopg: http://initd.org/psycopg/
         """
-
         tracer_provider = kwargs.get("tracer_provider")
 
         dbapi.wrap_connect(
@@ -74,39 +78,100 @@ def _instrument(self, **kwargs):
             self._CONNECTION_ATTRIBUTES,
             version=__version__,
             tracer_provider=tracer_provider,
+            db_api_integration_factory=DatabaseApiIntegration,
         )
 
     def _uninstrument(self, **kwargs):
         """"Disable Psycopg2 instrumentation"""
         dbapi.unwrap_connect(psycopg2, "connect")
 
-    # pylint:disable=no-self-use
+    # TODO(owais): check if core dbapi can do this for all dbapi implementations e.g, pymysql and mysql
     def instrument_connection(self, connection):
-        """Enable instrumentation in a Psycopg2 connection.
+        setattr(connection, OTEL_CURSOR_FACTORY_KEY, connection.cursor_factory)
+        connection.cursor_factory = _new_cursor_factory()
+        return connection
 
-        Args:
-            connection: The connection to instrument.
+    # TODO(owais): check if core dbapi can do this for all dbapi implementations e.g, pymysql and mysql
+    def uninstrument_connection(self, connection):
+        connection.cursor_factory = getattr(
+            connection, OTEL_CURSOR_FACTORY_KEY, None
+        )
+        return connection
+
+
+# TODO(owais): check if core dbapi can do this for all dbapi implementations e.g, pymysql and mysql
+class DatabaseApiIntegration(dbapi.DatabaseApiIntegration):
+    def wrapped_connection(
+        self,
+        connect_method: typing.Callable[..., typing.Any],
+        args: typing.Tuple[typing.Any, typing.Any],
+        kwargs: typing.Dict[typing.Any, typing.Any],
+    ):
+        """Add object proxy to connection object."""
+        base_cursor_factory = kwargs.pop("cursor_factory", None)
+        new_factory_kwargs = {"dbapi": self}
+        if base_cursor_factory:
+            new_factory_kwargs["base_factory"] = base_cursor_factory
+        kwargs["cursor_factory"] = _new_cursor_factory(**new_factory_kwargs)
+        connection = connect_method(*args, **kwargs)
+        self.get_connection_attributes(connection)
+        return connection
+
+
+class CursorTracer(dbapi.CursorTracer):
+    def get_operation_name(self, cursor, args):
+        if not args:
+            return ""
+
+        statement = args[0]
+        if isinstance(statement, Composed):
+            print('-=---------- composed')
+            print(statement)
+            print('-=----------')
+            statement = statement.as_string(cursor)
+
+        if isinstance(statement, str):
+            return statement.split(" ")[0]
+
+        return ""
+
+    def get_statement(self, cursor, args):
+        if not args:
+            return ""
+
+        statement = args[0]
+        if isinstance(statement, Composed):
+            statement = statement.as_string(cursor)
+        return statement
+
+
+def _new_cursor_factory(dbapi=None, base_factory=None):
+    if not dbapi:
+        dbapi = DatabaseApiIntegration(
+            __name__,
+            Psycopg2Instrumentor._DATABASE_COMPONENT,
+            database_type=Psycopg2Instrumentor._DATABASE_TYPE,
+            connection_attributes=Psycopg2Instrumentor._CONNECTION_ATTRIBUTES,
+            version=__version__,
+        )
 
-        Returns:
-            An instrumented connection.
-        """
-        tracer = get_tracer(__name__, __version__)
+    base_factory = base_factory or cursor
+    _cursor_tracer = CursorTracer(dbapi)
 
-        return dbapi.instrument_connection(
-            tracer,
-            connection,
-            self._DATABASE_COMPONENT,
-            self._DATABASE_TYPE,
-            self._CONNECTION_ATTRIBUTES,
-        )
+    class TracedCursorFactory(base_factory):
+        def execute(self, *args, **kwargs):
+            return _cursor_tracer.traced_execution(
+                self, super().execute, *args, **kwargs
+            )
 
-    def uninstrument_connection(self, connection):
-        """Disable instrumentation in a Psycopg2 connection.
+        def executemany(self, *args, **kwargs):
+            return _cursor_tracer.traced_execution(
+                self, super().executemany, *args, **kwargs
+            )
 
-        Args:
-            connection: The connection to uninstrument.
+        def callproc(self, *args, **kwargs):
+            return _cursor_tracer.traced_execution(
+                self, super().callproc, *args, **kwargs
+            )
 
-        Returns:
-            An uninstrumented connection.
-        """
-        return dbapi.uninstrument_connection(connection)
+    return TracedCursorFactory

instrumentation/opentelemetry-instrumentation-psycopg2/tests/test_psycopg2_integration.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import types
 from unittest import mock
 
 import psycopg2
@@ -22,15 +23,69 @@
 from opentelemetry.test.test_base import TestBase
 
 
+class MockCursor:
+
+    execute = mock.MagicMock(spec=types.MethodType)
+    execute.__name__ = "execute"
+
+    executemany = mock.MagicMock(spec=types.MethodType)
+    executemany.__name__ = "executemany"
+
+    callproc = mock.MagicMock(spec=types.MethodType)
+    callproc.__name__ = "callproc"
+
+    rowcount = "SomeRowCount"
+
+    def __init__(self, *args, **kwargs):
+        pass
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *args):
+        return self
+
+
+class MockConnection(object):
+
+    commit = mock.MagicMock(spec=types.MethodType)
+    commit.__name__ = "commit"
+
+    rollback = mock.MagicMock(spec=types.MethodType)
+    rollback.__name__ = "rollback"
+
+    def __init__(self, *args, **kwargs):
+        self.cursor_factory = kwargs.pop("cursor_factory", None)
+
+    def cursor(self):
+        if self.cursor_factory:
+            return self.cursor_factory(self)
+        return MockCursor()
+
+    def get_dsn_parameters(self):
+        return dict(dbname="test")
+
+
 class TestPostgresqlIntegration(TestBase):
+    def setUp(self):
+        self.cursor_mock = mock.patch(
+            "opentelemetry.instrumentation.psycopg2.cursor", MockCursor
+        )
+        self.connection_mock = mock.patch("psycopg2.connect", MockConnection)
+
+        self.cursor_mock.start()
+        self.connection_mock.start()
+
     def tearDown(self):
         super().tearDown()
+        self.memory_exporter.clear()
+        self.cursor_mock.stop()
+        self.connection_mock.stop()
         with self.disable_logging():
             Psycopg2Instrumentor().uninstrument()
 
-    @mock.patch("psycopg2.connect")
     # pylint: disable=unused-argument
-    def test_instrumentor(self, mock_connect):
+    def test_instrumentor(self):
         Psycopg2Instrumentor().instrument()
 
         cnx = psycopg2.connect(database="test")
@@ -60,9 +115,39 @@ def test_instrumentor(self, mock_connect):
         spans_list = self.memory_exporter.get_finished_spans()
         self.assertEqual(len(spans_list), 1)
 
-    @mock.patch("psycopg2.connect")
+    def test_composed(self):
+        Psycopg2Instrumentor().instrument()
+
+        cnx = psycopg2.connect(database="test")
+
+        cursor = cnx.cursor()
+
+        query = "SELECT * FROM test"
+        cursor.execute(query)
+
+        spans_list = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans_list), 1)
+        span = spans_list[0]
+
+        # Check version and name in span's instrumentation info
+        self.check_span_instrumentation_info(
+            span, opentelemetry.instrumentation.psycopg2
+        )
+
+        # check that no spans are generated after uninstrument
+        Psycopg2Instrumentor().uninstrument()
+
+        cnx = psycopg2.connect(database="test")
+        cursor = cnx.cursor()
+        query = "SELECT * FROM test"
+        cursor.execute(query)
+
+        spans_list = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans_list), 1)
+
+
     # pylint: disable=unused-argument
-    def test_not_recording(self, mock_connect):
+    def test_not_recording(self):
         mock_tracer = mock.Mock()
         mock_span = mock.Mock()
         mock_span.is_recording.return_value = False
@@ -83,9 +168,8 @@ def test_not_recording(self, mock_connect):
 
         Psycopg2Instrumentor().uninstrument()
 
-    @mock.patch("psycopg2.connect")
     # pylint: disable=unused-argument
-    def test_custom_tracer_provider(self, mock_connect):
+    def test_custom_tracer_provider(self):
         resource = resources.Resource.create({})
         result = self.create_tracer_provider(resource=resource)
         tracer_provider, exporter = result
@@ -103,9 +187,8 @@ def test_custom_tracer_provider(self, mock_connect):
 
         self.assertIs(span.resource, resource)
 
-    @mock.patch("psycopg2.connect")
     # pylint: disable=unused-argument
-    def test_instrument_connection(self, mock_connect):
+    def test_instrument_connection(self):
         cnx = psycopg2.connect(database="test")
         query = "SELECT * FROM test"
         cursor = cnx.cursor()
@@ -121,9 +204,8 @@ def test_instrument_connection(self, mock_connect):
         spans_list = self.memory_exporter.get_finished_spans()
         self.assertEqual(len(spans_list), 1)
 
-    @mock.patch("psycopg2.connect")
     # pylint: disable=unused-argument
-    def test_uninstrument_connection(self, mock_connect):
+    def test_uninstrument_connection(self):
         Psycopg2Instrumentor().instrument()
         cnx = psycopg2.connect(database="test")
         query = "SELECT * FROM test"

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

@@ -60,15 +60,15 @@ 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(stmt)
+        self.validate_spans("CREATE")
 
     def test_executemany(self):
         """Should create a child span for executemany"""
         stmt = "INSERT INTO test (id) VALUES (?)"
         with self._tracer.start_as_current_span("rootSpan"):
             data = [("1",), ("2",), ("3",)]
             self._cursor.executemany(stmt, data)
-        self.validate_spans(stmt)
+        self.validate_spans("INSERT")
 
     def test_callproc(self):
         """Should create a child span for callproc"""

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

@@ -81,7 +81,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(stmt)
+        self.validate_spans("CREATE")
 
     def test_execute_with_connection_context_manager(self):
         """Should create a child span for execute with connection context"""
@@ -90,23 +90,23 @@ def test_execute_with_connection_context_manager(self):
             with self._connection as conn:
                 cursor = conn.cursor()
                 cursor.execute(stmt)
-        self.validate_spans(stmt)
+        self.validate_spans("CREATE")
 
     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(stmt)
+        self.validate_spans("CREATE")
 
     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(stmt)
+        self.validate_spans("INSERT")
 
     def test_callproc(self):
         """Should create a child span for callproc"""

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

@@ -89,7 +89,7 @@ def test_execute(self):
         stmt = "CREATE TABLE IF NOT EXISTS test (id integer)"
         with self._tracer.start_as_current_span("rootSpan"):
             async_call(self._cursor.execute(stmt))
-        self.validate_spans(stmt)
+        self.validate_spans("CREATE")
 
     def test_executemany(self):
         """Should create a child span for executemany"""
@@ -98,7 +98,7 @@ def test_executemany(self):
             with self._tracer.start_as_current_span("rootSpan"):
                 data = (("1",), ("2",), ("3",))
                 async_call(self._cursor.executemany(stmt, data))
-            self.validate_spans(stmt)
+            self.validate_spans("INSERT")
 
     def test_callproc(self):
         """Should create a child span for callproc"""
@@ -167,7 +167,7 @@ def test_execute(self):
         stmt = "CREATE TABLE IF NOT EXISTS test (id integer)"
         with self._tracer.start_as_current_span("rootSpan"):
             async_call(self._cursor.execute(stmt))
-        self.validate_spans(stmt)
+        self.validate_spans("CREATE")
 
     def test_executemany(self):
         """Should create a child span for executemany"""
@@ -176,7 +176,7 @@ def test_executemany(self):
             with self._tracer.start_as_current_span("rootSpan"):
                 data = (("1",), ("2",), ("3",))
                 async_call(self._cursor.executemany(stmt, data))
-            self.validate_spans(stmt)
+            self.validate_spans("INSERT")
 
     def test_callproc(self):
         """Should create a child span for callproc"""

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

@@ -81,7 +81,7 @@ 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(stmt)
+        self.validate_spans("CREATE")
 
     def test_execute_with_connection_context_manager(self):
         """Should create a child span for execute with connection context"""
@@ -90,15 +90,15 @@ def test_execute_with_connection_context_manager(self):
             with self._connection as conn:
                 cursor = conn.cursor()
                 cursor.execute(stmt)
-        self.validate_spans(stmt)
+        self.validate_spans("CREATE")
 
     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(stmt)
+        self.validate_spans("CREATE")
         self.assertTrue(cursor.closed)
 
     def test_executemany(self):
@@ -107,7 +107,7 @@ def test_executemany(self):
         with self._tracer.start_as_current_span("rootSpan"):
             data = (("1",), ("2",), ("3",))
             self._cursor.executemany(stmt, data)
-        self.validate_spans(stmt)
+        self.validate_spans("INSERT")
 
     def test_callproc(self):
         """Should create a child span for callproc"""

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

@@ -78,23 +78,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(stmt)
+        self.validate_spans("CREATE")
 
     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(stmt)
+        self.validate_spans("CREATE")
 
     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(stmt)
+        self.validate_spans("INSERT")
 
     def test_callproc(self):
         """Should create a child span for callproc"""