diff --git a/.changelog/4906.added b/.changelog/4906.added new file mode 100644 index 0000000000..7ca83b58cb --- /dev/null +++ b/.changelog/4906.added @@ -0,0 +1 @@ +`opentelemetry-instrumentation-dbapi`, `opentelemetry-instrumentation-pymysql`, `opentelemetry-instrumentation-mysql`, `opentelemetry-instrumentation-mysqlclient`, `opentelemetry-instrumentation-psycopg`, `opentelemetry-instrumentation-psycopg2`, `opentelemetry-instrumentation-sqlite3`, `opentelemetry-instrumentation-pymssql`: add experimental instrumentation for `commit()` and `rollback()` transaction operations behind the `enable_transaction_spans` flag (default: `False`). diff --git a/instrumentation/opentelemetry-instrumentation-dbapi/src/opentelemetry/instrumentation/dbapi/__init__.py b/instrumentation/opentelemetry-instrumentation-dbapi/src/opentelemetry/instrumentation/dbapi/__init__.py index 9df47e6ad0..36b289ec39 100644 --- a/instrumentation/opentelemetry-instrumentation-dbapi/src/opentelemetry/instrumentation/dbapi/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-dbapi/src/opentelemetry/instrumentation/dbapi/__init__.py @@ -1,5 +1,6 @@ # Copyright The OpenTelemetry Authors # SPDX-License-Identifier: Apache-2.0 +# pylint: disable=too-many-lines """ The trace integration with Database API supports libraries that follow the @@ -182,6 +183,7 @@ _OpenTelemetryStabilitySignalType, _report_new, _set_db_name, + _set_db_operation, _set_db_statement, _set_db_system, _set_db_user, @@ -235,6 +237,7 @@ def trace_integration( enable_attribute_commenter: bool = False, commenter_options: dict[str, Any] | None = None, meter_provider: MeterProvider | None = None, + enable_transaction_spans: bool = False, ): """Integrate with DB API library. https://www.python.org/dev/peps/pep-0249/ @@ -256,6 +259,7 @@ def trace_integration( commenter_options: Configurations for tags to be appended at the sql query. meter_provider: The :class:`opentelemetry.metrics.MeterProvider` to use. If omitted the current configured one is used. + enable_transaction_spans: Experimental flag to enable transaction (commit/rollback) spans. Defaults to False. """ wrap_connect( __name__, @@ -271,10 +275,11 @@ def trace_integration( enable_attribute_commenter=enable_attribute_commenter, commenter_options=commenter_options, meter_provider=meter_provider, + enable_transaction_spans=enable_transaction_spans, ) -# pylint: disable-next=too-many-positional-arguments +# pylint: disable-next=too-many-positional-arguments,too-many-locals def wrap_connect( name: str, connect_module: Callable[..., Any], @@ -289,6 +294,7 @@ def wrap_connect( commenter_options: dict[str, Any] | None = None, enable_attribute_commenter: bool = False, meter_provider: MeterProvider | None = None, + enable_transaction_spans: bool = False, ): """Integrate with DB API library. https://www.python.org/dev/peps/pep-0249/ @@ -310,6 +316,7 @@ def wrap_connect( enable_attribute_commenter: Flag to enable/disable sqlcomment inclusion in `db.statement` and/or `db.query.text` span attribute. Only available if enable_commenter=True. meter_provider: The :class:`opentelemetry.metrics.MeterProvider` to use. If omitted the current configured one is used. + enable_transaction_spans: Experimental flag to enable transaction (commit/rollback) spans. Defaults to False. """ db_api_integration_factory = ( @@ -335,6 +342,7 @@ def wrap_connect_( connect_module=connect_module, enable_attribute_commenter=enable_attribute_commenter, meter_provider=meter_provider, + enable_transaction_spans=enable_transaction_spans, ) return db_integration.wrapped_connection(wrapped, args, kwargs) @@ -374,6 +382,7 @@ def instrument_connection( enable_attribute_commenter: bool = False, db_api_integration_factory: type[DatabaseApiIntegration] | None = None, meter_provider: MeterProvider | None = None, + enable_transaction_spans: bool = False, ) -> TracedConnectionProxy[ConnectionT]: """Enable instrumentation in a database connection. @@ -397,6 +406,7 @@ def instrument_connection( from the connection itself (as done by the pymssql intrumentor). meter_provider: The :class:`opentelemetry.metrics.MeterProvider` to use. If omitted the current configured one is used. + enable_transaction_spans: Experimental flag to enable transaction (commit/rollback) spans. Defaults to False. Returns: An instrumented connection. @@ -421,6 +431,7 @@ def instrument_connection( connect_module=connect_module, enable_attribute_commenter=enable_attribute_commenter, meter_provider=meter_provider, + enable_transaction_spans=enable_transaction_spans, ) db_integration.get_connection_attributes(connection) return get_traced_connection_proxy(connection, db_integration) @@ -445,6 +456,7 @@ def uninstrument_connection( class DatabaseApiIntegration: + # pylint: disable-next=too-many-positional-arguments def __init__( self, name: str, @@ -458,6 +470,7 @@ def __init__( connect_module: Callable[..., Any] | None = None, enable_attribute_commenter: bool = False, meter_provider: MeterProvider | None = None, + enable_transaction_spans: bool = False, ): # Initialize semantic conventions opt-in if needed _OpenTelemetrySemanticConventionStability._initialize() @@ -512,6 +525,7 @@ def __init__( self.enable_commenter = enable_commenter self.commenter_options = commenter_options self.enable_attribute_commenter = enable_attribute_commenter + self.enable_transaction_spans = enable_transaction_spans self.database_system = database_system self.connection_props: dict[str, Any] = {} self.span_attributes: dict[str, Any] = {} @@ -639,6 +653,15 @@ def get_connection_attributes(self, connection: object) -> None: ) self._server_port = port + def common_span_attributes(self) -> dict[str, Any]: + """Build common database connection attributes for a client span.""" + sem_conv_mode = self._sem_conv_opt_in_mode_db + span_attrs = {} + _set_db_system(span_attrs, self.database_system, sem_conv_mode) + _set_db_name(span_attrs, self.database, sem_conv_mode) + span_attrs.update(self.span_attributes) + return span_attrs + # pylint: disable=abstract-method,no-member class TracedConnectionProxy(BaseObjectProxy, Generic[ConnectionT]): @@ -647,22 +670,71 @@ def __init__( self, connection: ConnectionT, db_api_integration: DatabaseApiIntegration | None = None, + wrap_cursors: bool = True, ): BaseObjectProxy.__init__(self, connection) self._self_db_api_integration = db_api_integration + self._self_wrap_cursors = wrap_cursors def __getattribute__(self, name: str): - if object.__getattribute__(self, name): + # Try to get the attribute from the proxy first + try: return object.__getattribute__(self, name) + except AttributeError: + # If not found on proxy, try the wrapped connection + return object.__getattribute__( + object.__getattribute__(self, "__wrapped__"), name + ) + + def cursor(self, *args: Any, **kwargs: Any): + cursor = self.__wrapped__.cursor(*args, **kwargs) + + # For databases like psycopg/psycopg2 that use cursor_factory, + # cursor tracing is already handled by the factory, so skip wrapping + if not self._self_wrap_cursors: + return cursor + + # For standard dbapi connections, wrap the cursor + return get_traced_cursor_proxy(cursor, self._self_db_api_integration) + + def _traced_tx_operation( + self, + operation_name: str, + operation_method: Callable[..., None], + *args: Any, + **kwargs: Any, + ) -> None: + """Execute a traced transaction operation (commit, rollback).""" + if not is_instrumentation_enabled(): + return operation_method(*args, **kwargs) + + if not self._self_db_api_integration.enable_transaction_spans: + return operation_method(*args, **kwargs) - return object.__getattribute__( - object.__getattribute__(self, "_connection"), name + integration = self._self_db_api_integration + sem_conv_mode = integration._sem_conv_opt_in_mode_db + with integration._tracer.start_as_current_span( + operation_name, kind=trace_api.SpanKind.CLIENT + ) as span: + if span.is_recording(): + span_attrs = integration.common_span_attributes() + _set_db_operation(span_attrs, operation_name, sem_conv_mode) + span.set_attributes(span_attrs) + try: + return operation_method(*args, **kwargs) + except Exception as exc: + if span.is_recording() and _report_new(sem_conv_mode): + span.set_attribute(ERROR_TYPE, type(exc).__qualname__) + raise + + def commit(self, *args: Any, **kwargs: Any): + return self._traced_tx_operation( + "COMMIT", self.__wrapped__.commit, *args, **kwargs ) - def cursor(self, *args: Any, **kwargs: Any): - return get_traced_cursor_proxy( - self.__wrapped__.cursor(*args, **kwargs), - self._self_db_api_integration, + def rollback(self, *args: Any, **kwargs: Any): + return self._traced_tx_operation( + "ROLLBACK", self.__wrapped__.rollback, *args, **kwargs ) def __enter__(self): @@ -673,13 +745,98 @@ def __exit__(self, *args: Any, **kwargs: Any): self.__wrapped__.__exit__(*args, **kwargs) +class AsyncTracedConnectionProxy(TracedConnectionProxy[ConnectionT]): + async def _traced_tx_operation_async( + self, + operation_name: str, + operation_method: Callable[..., Awaitable[None]], + *args: Any, + **kwargs: Any, + ) -> None: + """Execute a traced async transaction operation (commit, rollback).""" + if not is_instrumentation_enabled(): + return await operation_method(*args, **kwargs) + + if not self._self_db_api_integration.enable_transaction_spans: + return await operation_method(*args, **kwargs) + + integration = self._self_db_api_integration + sem_conv_mode = integration._sem_conv_opt_in_mode_db + with integration._tracer.start_as_current_span( + operation_name, kind=trace_api.SpanKind.CLIENT + ) as span: + if span.is_recording(): + span_attrs = integration.common_span_attributes() + _set_db_operation(span_attrs, operation_name, sem_conv_mode) + span.set_attributes(span_attrs) + try: + return await operation_method(*args, **kwargs) + except Exception as exc: + if span.is_recording() and _report_new(sem_conv_mode): + span.set_attribute(ERROR_TYPE, type(exc).__qualname__) + raise + + async def commit(self, *args: Any, **kwargs: Any): + """Async commit for async connections (e.g., psycopg.AsyncConnection).""" + return await self._traced_tx_operation_async( + "COMMIT", self.__wrapped__.commit, *args, **kwargs + ) + + async def rollback(self, *args: Any, **kwargs: Any): + """Async rollback for async connections (e.g., psycopg.AsyncConnection).""" + return await self._traced_tx_operation_async( + "ROLLBACK", self.__wrapped__.rollback, *args, **kwargs + ) + + # Async context manager support + async def __aenter__(self): + if hasattr(self.__wrapped__, "__aenter__"): + await self.__wrapped__.__aenter__() + return self + + async def __aexit__(self, *args: Any, **kwargs: Any): + if hasattr(self.__wrapped__, "__aexit__"): + return await self.__wrapped__.__aexit__(*args, **kwargs) + + def get_traced_connection_proxy( connection: ConnectionT, db_api_integration: DatabaseApiIntegration | None, *args: Any, + wrap_cursors: bool = True, **kwargs: Any, ) -> TracedConnectionProxy[ConnectionT]: - return TracedConnectionProxy(connection, db_api_integration) + """Get a traced connection proxy for sync connections. + + Args: + connection: The database connection to wrap. + db_api_integration: The database API integration instance. + wrap_cursors: Whether to wrap cursors returned by connection.cursor(). + Set to False for databases like psycopg/psycopg2 that handle cursor + tracing via cursor_factory. Defaults to True. + """ + return TracedConnectionProxy(connection, db_api_integration, wrap_cursors) + + +def get_traced_async_connection_proxy( + connection: ConnectionT, + db_api_integration: DatabaseApiIntegration | None, + *args: Any, + wrap_cursors: bool = True, + **kwargs: Any, +) -> AsyncTracedConnectionProxy[ConnectionT]: + """Get a traced connection proxy for async connections. + + Args: + connection: The async database connection to wrap. + db_api_integration: The database API integration instance. + wrap_cursors: Whether to wrap cursors returned by connection.cursor(). + Set to False for databases like psycopg/psycopg2 that handle cursor + tracing via cursor_factory. Defaults to True. + """ + return AsyncTracedConnectionProxy( + connection, db_api_integration, wrap_cursors + ) class CursorTracer(Generic[CursorT]): @@ -764,31 +921,13 @@ def _populate_span( ): if not span.is_recording(): return + statement = self.get_statement(cursor, args) sem_conv_mode = self._db_api_integration._sem_conv_opt_in_mode_db - span_attrs = {} - - _set_db_system( - span_attrs, - self._db_api_integration.database_system, - sem_conv_mode, - ) - _set_db_name( - span_attrs, - self._db_api_integration.database, - sem_conv_mode, - ) + span_attrs = self._db_api_integration.common_span_attributes() _set_db_statement(span_attrs, statement, sem_conv_mode) - - # Set all collected attributes span.set_attributes(span_attrs) - for ( - attribute_key, - attribute_value, - ) in self._db_api_integration.span_attributes.items(): - span.set_attribute(attribute_key, attribute_value) - if self._db_api_integration.capture_parameters and len(args) > 1: span.set_attribute("db.statement.parameters", str(args[1])) diff --git a/instrumentation/opentelemetry-instrumentation-dbapi/tests/test_dbapi_integration.py b/instrumentation/opentelemetry-instrumentation-dbapi/tests/test_dbapi_integration.py index 07f1d80345..fd1b9eef97 100644 --- a/instrumentation/opentelemetry-instrumentation-dbapi/tests/test_dbapi_integration.py +++ b/instrumentation/opentelemetry-instrumentation-dbapi/tests/test_dbapi_integration.py @@ -21,6 +21,7 @@ from opentelemetry.semconv._incubating.attributes import net_attributes from opentelemetry.semconv._incubating.attributes.db_attributes import ( DB_NAME, + DB_OPERATION, DB_STATEMENT, DB_SYSTEM, DB_USER, @@ -1512,6 +1513,184 @@ def test_callproc(self): "Test stored procedure", ) + def test_commit(self): + db_integration = dbapi.DatabaseApiIntegration( + "instrumenting_module_test_name", + "testcomponent", + enable_transaction_spans=True, + ) + mock_connection = db_integration.wrapped_connection( + mock_connect, {}, {} + ) + mock_connection.commit() + spans_list = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans_list), 1) + span = spans_list[0] + self.assertEqual(span.name, "COMMIT") + self.assertEqual(span.attributes[DB_OPERATION], "COMMIT") + + def test_commit_new_semconv(self): + with use_semconv_opt_in("database"): + db_integration = dbapi.DatabaseApiIntegration( + "instrumenting_module_test_name", + "testcomponent", + enable_transaction_spans=True, + ) + mock_connection = db_integration.wrapped_connection( + mock_connect, {}, {} + ) + mock_connection.commit() + spans_list = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans_list), 1) + span = spans_list[0] + self.assertEqual(span.name, "COMMIT") + self.assertEqual(span.attributes[DB_OPERATION_NAME], "COMMIT") + + def test_commit_both_semconv(self): + with use_semconv_opt_in("database/dup"): + db_integration = dbapi.DatabaseApiIntegration( + "instrumenting_module_test_name", + "testcomponent", + enable_transaction_spans=True, + ) + mock_connection = db_integration.wrapped_connection( + mock_connect, {}, {} + ) + mock_connection.commit() + spans_list = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans_list), 1) + span = spans_list[0] + self.assertEqual(span.name, "COMMIT") + self.assertEqual(span.attributes[DB_OPERATION], "COMMIT") + self.assertEqual(span.attributes[DB_OPERATION_NAME], "COMMIT") + + def test_rollback(self): + db_integration = dbapi.DatabaseApiIntegration( + "instrumenting_module_test_name", + "testcomponent", + enable_transaction_spans=True, + ) + mock_connection = db_integration.wrapped_connection( + mock_connect, {}, {} + ) + mock_connection.rollback() + spans_list = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans_list), 1) + span = spans_list[0] + self.assertEqual(span.name, "ROLLBACK") + self.assertEqual(span.attributes[DB_OPERATION], "ROLLBACK") + + def test_rollback_new_semconv(self): + with use_semconv_opt_in("database"): + db_integration = dbapi.DatabaseApiIntegration( + "instrumenting_module_test_name", + "testcomponent", + enable_transaction_spans=True, + ) + mock_connection = db_integration.wrapped_connection( + mock_connect, {}, {} + ) + mock_connection.rollback() + spans_list = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans_list), 1) + span = spans_list[0] + self.assertEqual(span.name, "ROLLBACK") + self.assertEqual(span.attributes[DB_OPERATION_NAME], "ROLLBACK") + + def test_rollback_both_semconv(self): + with use_semconv_opt_in("database/dup"): + db_integration = dbapi.DatabaseApiIntegration( + "instrumenting_module_test_name", + "testcomponent", + enable_transaction_spans=True, + ) + mock_connection = db_integration.wrapped_connection( + mock_connect, {}, {} + ) + mock_connection.rollback() + spans_list = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans_list), 1) + span = spans_list[0] + self.assertEqual(span.name, "ROLLBACK") + self.assertEqual(span.attributes[DB_OPERATION], "ROLLBACK") + self.assertEqual(span.attributes[DB_OPERATION_NAME], "ROLLBACK") + + def test_commit_with_suppress_instrumentation(self): + """Test that commit doesn't create a span when instrumentation is suppressed""" + db_integration = dbapi.DatabaseApiIntegration( + "instrumenting_module_test_name", + "testcomponent", + enable_transaction_spans=True, + ) + mock_connection = db_integration.wrapped_connection( + mock_connect, {}, {} + ) + with suppress_instrumentation(): + mock_connection.commit() + + spans_list = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans_list), 0) + + def test_rollback_with_suppress_instrumentation(self): + """Test that rollback doesn't create a span when instrumentation is suppressed""" + db_integration = dbapi.DatabaseApiIntegration( + "instrumenting_module_test_name", + "testcomponent", + enable_transaction_spans=True, + ) + mock_connection = db_integration.wrapped_connection( + mock_connect, {}, {} + ) + with suppress_instrumentation(): + mock_connection.rollback() + + spans_list = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans_list), 0) + + def test_commit_failed_new_semconv(self): + with use_semconv_opt_in("database"): + db_integration = dbapi.DatabaseApiIntegration( + "instrumenting_module_test_name", + "testcomponent", + enable_transaction_spans=True, + ) + mock_connection = db_integration.wrapped_connection( + mock_connect, {}, {} + ) + with self.assertRaises(Exception): + mock_connection.commit(throw_exception=True) + + spans_list = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans_list), 1) + span = spans_list[0] + self.assertEqual(span.name, "COMMIT") + self.assertIs(span.status.status_code, trace_api.StatusCode.ERROR) + self.assertEqual(span.attributes[ERROR_TYPE], "Exception") + self.assertEqual(len(span.events), 1) + self.assertEqual(span.events[0].name, "exception") + + def test_rollback_failed_new_semconv(self): + with use_semconv_opt_in("database"): + db_integration = dbapi.DatabaseApiIntegration( + "instrumenting_module_test_name", + "testcomponent", + enable_transaction_spans=True, + ) + mock_connection = db_integration.wrapped_connection( + mock_connect, {}, {} + ) + with self.assertRaises(Exception): + mock_connection.rollback(throw_exception=True) + + spans_list = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans_list), 1) + span = spans_list[0] + self.assertEqual(span.name, "ROLLBACK") + self.assertIs(span.status.status_code, trace_api.StatusCode.ERROR) + self.assertEqual(span.attributes[ERROR_TYPE], "Exception") + self.assertEqual(len(span.events), 1) + self.assertEqual(span.events[0].name, "exception") + @mock.patch("opentelemetry.instrumentation.dbapi") def test_wrap_connect(self, mock_dbapi): dbapi.wrap_connect(self.tracer, mock_dbapi, "connect", "-") @@ -1749,6 +1928,18 @@ def __init__(self, database, server_port, server_host, user): def cursor(self): return MockCursor() + # pylint: disable=no-self-use + def commit(self, throw_exception=False): + if throw_exception: + # pylint: disable=broad-exception-raised + raise Exception("Test Exception") + + # pylint: disable=no-self-use + def rollback(self, throw_exception=False): + if throw_exception: + # pylint: disable=broad-exception-raised + raise Exception("Test Exception") + class MockCursor: def __init__(self) -> None: diff --git a/instrumentation/opentelemetry-instrumentation-mysql/src/opentelemetry/instrumentation/mysql/__init__.py b/instrumentation/opentelemetry-instrumentation-mysql/src/opentelemetry/instrumentation/mysql/__init__.py index 8de6402aeb..f207b36bd9 100644 --- a/instrumentation/opentelemetry-instrumentation-mysql/src/opentelemetry/instrumentation/mysql/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-mysql/src/opentelemetry/instrumentation/mysql/__init__.py @@ -171,6 +171,9 @@ def _instrument(self, **kwargs): enable_attribute_commenter = kwargs.get( "enable_attribute_commenter", False ) + enable_transaction_spans = kwargs.get( + "enable_transaction_spans", False + ) dbapi.wrap_connect( __name__, @@ -183,6 +186,7 @@ def _instrument(self, **kwargs): enable_commenter=enable_sqlcommenter, commenter_options=commenter_options, enable_attribute_commenter=enable_attribute_commenter, + enable_transaction_spans=enable_transaction_spans, ) def _uninstrument(self, **kwargs): @@ -197,6 +201,7 @@ def instrument_connection( enable_commenter=None, commenter_options=None, enable_attribute_commenter=None, + enable_transaction_spans=False, ): """Enable instrumentation in a MySQL connection. @@ -214,6 +219,8 @@ def instrument_connection( Optional configurations for tags to be appended at the sql query. enable_attribute_commenter: Optional flag to enable/disable addition of sqlcomment to span attribute (default False). Requires enable_commenter=True. + enable_transaction_spans: + Experimental flag to enable transaction (commit/rollback) spans. Defaults to False. Returns: An instrumented MySQL connection with OpenTelemetry tracing enabled. @@ -229,6 +236,7 @@ def instrument_connection( commenter_options=commenter_options, connect_module=mysql.connector, enable_attribute_commenter=enable_attribute_commenter, + enable_transaction_spans=enable_transaction_spans, ) def uninstrument_connection(self, connection): diff --git a/instrumentation/opentelemetry-instrumentation-mysqlclient/src/opentelemetry/instrumentation/mysqlclient/__init__.py b/instrumentation/opentelemetry-instrumentation-mysqlclient/src/opentelemetry/instrumentation/mysqlclient/__init__.py index 98127ec3db..e48b648786 100644 --- a/instrumentation/opentelemetry-instrumentation-mysqlclient/src/opentelemetry/instrumentation/mysqlclient/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-mysqlclient/src/opentelemetry/instrumentation/mysqlclient/__init__.py @@ -155,6 +155,9 @@ def _instrument(self, **kwargs): # pylint: disable=no-self-use enable_attribute_commenter = kwargs.get( "enable_attribute_commenter", False ) + enable_transaction_spans = kwargs.get( + "enable_transaction_spans", False + ) dbapi.wrap_connect( __name__, @@ -167,6 +170,7 @@ def _instrument(self, **kwargs): # pylint: disable=no-self-use enable_commenter=enable_sqlcommenter, commenter_options=commenter_options, enable_attribute_commenter=enable_attribute_commenter, + enable_transaction_spans=enable_transaction_spans, ) def _uninstrument(self, **kwargs): # pylint: disable=no-self-use @@ -180,6 +184,7 @@ def instrument_connection( enable_commenter=None, commenter_options=None, enable_attribute_commenter=None, + enable_transaction_spans=False, ): """Enable instrumentation in a mysqlclient connection. @@ -204,6 +209,8 @@ def instrument_connection( - `mysql_client_version`: Adds the MySQL client version. - `driver_paramstyle`: Adds the parameter style. - `opentelemetry_values`: Includes traceparent values. + enable_transaction_spans: + Experimental flag to enable transaction (commit/rollback) spans. Defaults to False. Returns: An instrumented MySQL connection with OpenTelemetry support enabled. """ @@ -219,6 +226,7 @@ def instrument_connection( commenter_options=commenter_options, connect_module=MySQLdb, enable_attribute_commenter=enable_attribute_commenter, + enable_transaction_spans=enable_transaction_spans, ) @staticmethod diff --git a/instrumentation/opentelemetry-instrumentation-psycopg/src/opentelemetry/instrumentation/psycopg/__init__.py b/instrumentation/opentelemetry-instrumentation-psycopg/src/opentelemetry/instrumentation/psycopg/__init__.py index 5617335f7a..ae4a3aa2d8 100644 --- a/instrumentation/opentelemetry-instrumentation-psycopg/src/opentelemetry/instrumentation/psycopg/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-psycopg/src/opentelemetry/instrumentation/psycopg/__init__.py @@ -177,6 +177,9 @@ def _instrument(self, **kwargs: Any): "enable_attribute_commenter", False ) capture_parameters = kwargs.get("capture_parameters", False) + enable_transaction_spans = kwargs.get( + "enable_transaction_spans", False + ) dbapi.wrap_connect( __name__, psycopg, @@ -190,6 +193,7 @@ def _instrument(self, **kwargs: Any): commenter_options=commenter_options, enable_attribute_commenter=enable_attribute_commenter, capture_parameters=capture_parameters, + enable_transaction_spans=enable_transaction_spans, ) dbapi.wrap_connect( @@ -205,6 +209,7 @@ def _instrument(self, **kwargs: Any): commenter_options=commenter_options, enable_attribute_commenter=enable_attribute_commenter, capture_parameters=capture_parameters, + enable_transaction_spans=enable_transaction_spans, ) dbapi.wrap_connect( __name__, @@ -219,6 +224,7 @@ def _instrument(self, **kwargs: Any): commenter_options=commenter_options, enable_attribute_commenter=enable_attribute_commenter, capture_parameters=capture_parameters, + enable_transaction_spans=enable_transaction_spans, ) def _uninstrument(self, **kwargs: Any): @@ -298,7 +304,10 @@ def wrapped_connection( kwargs["cursor_factory"] = _new_cursor_factory(**new_factory_kwargs) connection = connect_method(*args, **kwargs) self.get_connection_attributes(connection) - return connection + # psycopg uses cursor_factory for cursor tracing, so disable cursor wrapping + return dbapi.get_traced_connection_proxy( + connection, self, wrap_cursors=False + ) class DatabaseApiAsyncIntegration(dbapi.DatabaseApiIntegration): @@ -318,7 +327,10 @@ async def wrapped_connection( ) connection = await connect_method(*args, **kwargs) self.get_connection_attributes(connection) - return connection + # psycopg uses cursor_factory for cursor tracing, so disable cursor wrapping + return dbapi.get_traced_async_connection_proxy( + connection, self, wrap_cursors=False + ) class CursorTracer(dbapi.CursorTracer): diff --git a/instrumentation/opentelemetry-instrumentation-psycopg/tests/test_psycopg_integration.py b/instrumentation/opentelemetry-instrumentation-psycopg/tests/test_psycopg_integration.py index 44dea236c8..78e7c1c600 100644 --- a/instrumentation/opentelemetry-instrumentation-psycopg/tests/test_psycopg_integration.py +++ b/instrumentation/opentelemetry-instrumentation-psycopg/tests/test_psycopg_integration.py @@ -9,8 +9,12 @@ from psycopg.sql import SQL, Composed import opentelemetry.instrumentation.psycopg +from opentelemetry import trace as trace_api from opentelemetry.instrumentation.psycopg import PsycopgInstrumentor from opentelemetry.sdk import resources +from opentelemetry.semconv._incubating.attributes.db_attributes import ( + DB_OPERATION, +) from opentelemetry.test.test_base import TestBase @@ -73,12 +77,6 @@ def close(self): class MockConnection: - 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) @@ -87,17 +85,19 @@ def cursor(self): return self.cursor_factory(self) return MockCursor() + def commit(self, throw_exception=False): # pylint: disable=no-self-use + if throw_exception: + raise psycopg.Error("Test Exception") + + def rollback(self, throw_exception=False): # pylint: disable=no-self-use + if throw_exception: + raise psycopg.Error("Test Exception") + def get_dsn_parameters(self): # pylint: disable=no-self-use return {"dbname": "test"} class MockAsyncConnection(psycopg.AsyncConnection): - 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) @@ -105,6 +105,14 @@ def __init__(self, *args, **kwargs): async def connect(*args, **kwargs): return MockAsyncConnection(**kwargs) + async def commit(self, throw_exception=False): + if throw_exception: + raise psycopg.Error("Test Exception") + + async def rollback(self, throw_exception=False): + if throw_exception: + raise psycopg.Error("Test Exception") + def cursor(self, *args, **kwargs): if self.cursor_factory: cur = self.cursor_factory(self) @@ -430,6 +438,30 @@ def test_uninstrument_connection_with_instrument_connection(self): spans_list = self.memory_exporter.get_finished_spans() self.assertEqual(len(spans_list), 1) + def test_commit(self): + PsycopgInstrumentor().instrument(enable_transaction_spans=True) + + cnx = psycopg.connect(database="test") + cnx.commit() + + spans_list = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans_list), 1) + span = spans_list[0] + self.assertEqual(span.name, "COMMIT") + self.assertEqual(span.attributes[DB_OPERATION], "COMMIT") + + def test_rollback(self): + PsycopgInstrumentor().instrument(enable_transaction_spans=True) + + cnx = psycopg.connect(database="test") + cnx.rollback() + + spans_list = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans_list), 1) + span = spans_list[0] + self.assertEqual(span.name, "ROLLBACK") + self.assertEqual(span.attributes[DB_OPERATION], "ROLLBACK") + @mock.patch("opentelemetry.instrumentation.dbapi.wrap_connect") def test_sqlcommenter_enabled(self, event_mocked): cnx = psycopg.connect(database="test") @@ -580,6 +612,68 @@ async def test_not_recording_async(self): PsycopgInstrumentor().uninstrument() + async def test_async_commit(self): + PsycopgInstrumentor().instrument(enable_transaction_spans=True) + + cnx = await psycopg.AsyncConnection.connect("test") + await cnx.commit() + + spans_list = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans_list), 1) + span = spans_list[0] + self.assertEqual(span.name, "COMMIT") + self.assertEqual(span.attributes[DB_OPERATION], "COMMIT") + + PsycopgInstrumentor().uninstrument() + + async def test_async_rollback(self): + PsycopgInstrumentor().instrument(enable_transaction_spans=True) + + cnx = await psycopg.AsyncConnection.connect("test") + await cnx.rollback() + + spans_list = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans_list), 1) + span = spans_list[0] + self.assertEqual(span.name, "ROLLBACK") + self.assertEqual(span.attributes[DB_OPERATION], "ROLLBACK") + + PsycopgInstrumentor().uninstrument() + + async def test_async_commit_failed(self): + PsycopgInstrumentor().instrument(enable_transaction_spans=True) + + cnx = await psycopg.AsyncConnection.connect("test") + with self.assertRaises(psycopg.Error): + await cnx.commit(throw_exception=True) + + spans_list = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans_list), 1) + span = spans_list[0] + self.assertEqual(span.name, "COMMIT") + self.assertIs(span.status.status_code, trace_api.StatusCode.ERROR) + self.assertEqual(len(span.events), 1) + self.assertEqual(span.events[0].name, "exception") + + PsycopgInstrumentor().uninstrument() + + async def test_async_rollback_failed(self): + PsycopgInstrumentor().instrument(enable_transaction_spans=True) + + cnx = await psycopg.AsyncConnection.connect("test") + with self.assertRaises(psycopg.Error): + await cnx.rollback(throw_exception=True) + + spans_list = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans_list), 1) + span = spans_list[0] + self.assertEqual(span.name, "ROLLBACK") + self.assertIs(span.status.status_code, trace_api.StatusCode.ERROR) + self.assertEqual(len(span.events), 1) + self.assertEqual(span.events[0].name, "exception") + + PsycopgInstrumentor().uninstrument() + async def test_tracing_is_async(self): PsycopgInstrumentor().instrument() diff --git a/instrumentation/opentelemetry-instrumentation-psycopg2/src/opentelemetry/instrumentation/psycopg2/__init__.py b/instrumentation/opentelemetry-instrumentation-psycopg2/src/opentelemetry/instrumentation/psycopg2/__init__.py index b7bf2c9601..f646f3d770 100644 --- a/instrumentation/opentelemetry-instrumentation-psycopg2/src/opentelemetry/instrumentation/psycopg2/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-psycopg2/src/opentelemetry/instrumentation/psycopg2/__init__.py @@ -216,6 +216,9 @@ def _instrument(self, **kwargs): "enable_attribute_commenter", False ) capture_parameters = kwargs.get("capture_parameters", False) + enable_transaction_spans = kwargs.get( + "enable_transaction_spans", False + ) dbapi.wrap_connect( __name__, psycopg2, @@ -229,6 +232,7 @@ def _instrument(self, **kwargs): commenter_options=commenter_options, enable_attribute_commenter=enable_attribute_commenter, capture_parameters=capture_parameters, + enable_transaction_spans=enable_transaction_spans, ) def _uninstrument(self, **kwargs): @@ -309,7 +313,10 @@ def wrapped_connection( kwargs["cursor_factory"] = _new_cursor_factory(**new_factory_kwargs) connection = connect_method(*args, **kwargs) self.get_connection_attributes(connection) - return connection + # psycopg2 uses cursor_factory for cursor tracing, so disable cursor wrapping + return dbapi.get_traced_connection_proxy( + connection, self, wrap_cursors=False + ) class CursorTracer(dbapi.CursorTracer): diff --git a/instrumentation/opentelemetry-instrumentation-psycopg2/tests/test_psycopg2_instrumentation.py b/instrumentation/opentelemetry-instrumentation-psycopg2/tests/test_psycopg2_instrumentation.py index 0f7785f6aa..2f2508bfa4 100644 --- a/instrumentation/opentelemetry-instrumentation-psycopg2/tests/test_psycopg2_instrumentation.py +++ b/instrumentation/opentelemetry-instrumentation-psycopg2/tests/test_psycopg2_instrumentation.py @@ -186,6 +186,7 @@ def test_instrument_defaults(self, mock_dbapi): # pylint: disable=no-self-use commenter_options={}, enable_attribute_commenter=False, capture_parameters=False, + enable_transaction_spans=False, ) def test_instrument_capture_parameters(self, mock_dbapi): diff --git a/instrumentation/opentelemetry-instrumentation-pymssql/src/opentelemetry/instrumentation/pymssql/__init__.py b/instrumentation/opentelemetry-instrumentation-pymssql/src/opentelemetry/instrumentation/pymssql/__init__.py index ea85421e35..0ed6a43af8 100644 --- a/instrumentation/opentelemetry-instrumentation-pymssql/src/opentelemetry/instrumentation/pymssql/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-pymssql/src/opentelemetry/instrumentation/pymssql/__init__.py @@ -157,6 +157,9 @@ def _instrument(self, **kwargs): https://github.com/pymssql/pymssql/ """ tracer_provider = kwargs.get("tracer_provider") + enable_transaction_spans = kwargs.get( + "enable_transaction_spans", False + ) dbapi.wrap_connect( __name__, @@ -169,6 +172,7 @@ def _instrument(self, **kwargs): # instead, we get the attributes from the connect method (which is done # via PyMSSQLDatabaseApiIntegration.wrapped_connection) db_api_integration_factory=_PyMSSQLDatabaseApiIntegration, + enable_transaction_spans=enable_transaction_spans, ) def _uninstrument(self, **kwargs): @@ -176,13 +180,17 @@ def _uninstrument(self, **kwargs): dbapi.unwrap_connect(pymssql, "connect") @staticmethod - def instrument_connection(connection, tracer_provider=None): + def instrument_connection( + connection, tracer_provider=None, enable_transaction_spans=False + ): """Enable instrumentation in a pymssql connection. Args: connection: The connection to instrument. tracer_provider: The optional tracer provider to use. If omitted the current globally configured one is used. + enable_transaction_spans: Experimental flag to enable transaction + (commit/rollback) spans. Defaults to False. Returns: An instrumented connection. @@ -195,6 +203,7 @@ def instrument_connection(connection, tracer_provider=None): version=__version__, tracer_provider=tracer_provider, db_api_integration_factory=_PyMSSQLDatabaseApiIntegration, + enable_transaction_spans=enable_transaction_spans, ) @staticmethod diff --git a/instrumentation/opentelemetry-instrumentation-pymysql/src/opentelemetry/instrumentation/pymysql/__init__.py b/instrumentation/opentelemetry-instrumentation-pymysql/src/opentelemetry/instrumentation/pymysql/__init__.py index 0b9eb4b3b9..507730a9f4 100644 --- a/instrumentation/opentelemetry-instrumentation-pymysql/src/opentelemetry/instrumentation/pymysql/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-pymysql/src/opentelemetry/instrumentation/pymysql/__init__.py @@ -178,6 +178,9 @@ def _instrument(self, **kwargs): # pylint: disable=no-self-use enable_attribute_commenter = kwargs.get( "enable_attribute_commenter", False ) + enable_transaction_spans = kwargs.get( + "enable_transaction_spans", False + ) dbapi.wrap_connect( __name__, @@ -190,6 +193,7 @@ def _instrument(self, **kwargs): # pylint: disable=no-self-use enable_commenter=enable_sqlcommenter, commenter_options=commenter_options, enable_attribute_commenter=enable_attribute_commenter, + enable_transaction_spans=enable_transaction_spans, ) def _uninstrument(self, **kwargs): # pylint: disable=no-self-use @@ -203,6 +207,7 @@ def instrument_connection( enable_commenter=None, commenter_options=None, enable_attribute_commenter=None, + enable_transaction_spans=False, ): """Enable instrumentation in a PyMySQL connection. @@ -221,6 +226,8 @@ def instrument_connection( You can specify various options, such as enabling driver information, database version logging, traceparent propagation, and other customizable metadata enhancements. See *SQLCommenter Configurations* above for more information. + enable_transaction_spans: + Experimental flag to enable transaction (commit/rollback) spans. Defaults to False. Returns: An instrumented connection. """ @@ -236,6 +243,7 @@ def instrument_connection( commenter_options=commenter_options, connect_module=pymysql, enable_attribute_commenter=enable_attribute_commenter, + enable_transaction_spans=enable_transaction_spans, ) @staticmethod diff --git a/instrumentation/opentelemetry-instrumentation-pymysql/tests/test_pymysql_integration.py b/instrumentation/opentelemetry-instrumentation-pymysql/tests/test_pymysql_integration.py index 2abfbbe97f..c953ab9980 100644 --- a/instrumentation/opentelemetry-instrumentation-pymysql/tests/test_pymysql_integration.py +++ b/instrumentation/opentelemetry-instrumentation-pymysql/tests/test_pymysql_integration.py @@ -10,7 +10,9 @@ from opentelemetry.instrumentation.pymysql import PyMySQLInstrumentor from opentelemetry.sdk import resources from opentelemetry.semconv._incubating.attributes.db_attributes import ( + DB_OPERATION, DB_STATEMENT, + DB_SYSTEM, ) from opentelemetry.test.test_base import TestBase @@ -477,3 +479,58 @@ def test_uninstrument_connection(self, mock_connect): spans_list = self.memory_exporter.get_finished_spans() self.assertEqual(len(spans_list), 1) + + @mock.patch("pymysql.connect") + # pylint: disable=unused-argument + def test_commit(self, mock_connect): + """Test that commit creates a span""" + PyMySQLInstrumentor().instrument(enable_transaction_spans=True) + cnx = pymysql.connect(database="test") + cnx.commit() + + spans_list = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans_list), 1) + span = spans_list[0] + self.assertEqual(span.name, "COMMIT") + self.assertIs(span.kind, trace_api.SpanKind.CLIENT) + self.assertEqual(span.attributes[DB_SYSTEM], "mysql") + self.assertEqual(span.attributes[DB_OPERATION], "COMMIT") + + @mock.patch("pymysql.connect") + # pylint: disable=unused-argument + def test_rollback(self, mock_connect): + """Test that rollback creates a span""" + PyMySQLInstrumentor().instrument(enable_transaction_spans=True) + cnx = pymysql.connect(database="test") + cnx.rollback() + + spans_list = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans_list), 1) + span = spans_list[0] + self.assertEqual(span.name, "ROLLBACK") + self.assertIs(span.kind, trace_api.SpanKind.CLIENT) + self.assertEqual(span.attributes[DB_SYSTEM], "mysql") + self.assertEqual(span.attributes[DB_OPERATION], "ROLLBACK") + + @mock.patch("pymysql.connect") + # pylint: disable=unused-argument + def test_commit_and_query(self, mock_connect): + """Test that both execute and commit create spans""" + PyMySQLInstrumentor().instrument(enable_transaction_spans=True) + cnx = pymysql.connect(database="test") + cursor = cnx.cursor() + cursor.execute("SELECT * FROM test") + cnx.commit() + + spans_list = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans_list), 2) + + # First span should be the SELECT + select_span = spans_list[0] + self.assertEqual(select_span.name, "SELECT") + self.assertIs(select_span.kind, trace_api.SpanKind.CLIENT) + + # Second span should be the COMMIT + commit_span = spans_list[1] + self.assertEqual(commit_span.name, "COMMIT") + self.assertIs(commit_span.kind, trace_api.SpanKind.CLIENT) diff --git a/instrumentation/opentelemetry-instrumentation-sqlite3/src/opentelemetry/instrumentation/sqlite3/__init__.py b/instrumentation/opentelemetry-instrumentation-sqlite3/src/opentelemetry/instrumentation/sqlite3/__init__.py index 8075ef2c31..e4012083c6 100644 --- a/instrumentation/opentelemetry-instrumentation-sqlite3/src/opentelemetry/instrumentation/sqlite3/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-sqlite3/src/opentelemetry/instrumentation/sqlite3/__init__.py @@ -78,6 +78,9 @@ def _instrument(self, **kwargs: Any) -> None: https://docs.python.org/3/library/sqlite3.html """ tracer_provider = kwargs.get("tracer_provider") + enable_transaction_spans = kwargs.get( + "enable_transaction_spans", False + ) for module in self._TO_WRAP: dbapi.wrap_connect( @@ -88,6 +91,7 @@ def _instrument(self, **kwargs: Any) -> None: _CONNECTION_ATTRIBUTES, version=__version__, tracer_provider=tracer_provider, + enable_transaction_spans=enable_transaction_spans, ) def _uninstrument(self, **kwargs: Any) -> None: @@ -99,6 +103,7 @@ def _uninstrument(self, **kwargs: Any) -> None: def instrument_connection( connection: SQLite3Connection, tracer_provider: TracerProvider | None = None, + enable_transaction_spans: bool = False, ) -> SQLite3Connection: """Enable instrumentation in a SQLite connection. @@ -106,6 +111,8 @@ def instrument_connection( connection: The connection to instrument. tracer_provider: The optional tracer provider to use. If omitted the current globally configured one is used. + enable_transaction_spans: Experimental flag to enable transaction + (commit/rollback) spans. Defaults to False. Returns: An instrumented SQLite connection that supports @@ -119,6 +126,7 @@ def instrument_connection( _CONNECTION_ATTRIBUTES, version=__version__, tracer_provider=tracer_provider, + enable_transaction_spans=enable_transaction_spans, ) @staticmethod diff --git a/tests/opentelemetry-docker-tests/tests/pymysql/test_pymysql_functional.py b/tests/opentelemetry-docker-tests/tests/pymysql/test_pymysql_functional.py index fb9dacf7c2..0ebb901567 100644 --- a/tests/opentelemetry-docker-tests/tests/pymysql/test_pymysql_functional.py +++ b/tests/opentelemetry-docker-tests/tests/pymysql/test_pymysql_functional.py @@ -29,7 +29,7 @@ class TestFunctionalPyMysql(TestBase): def setUp(self): super().setUp() self._tracer = self.tracer_provider.get_tracer(__name__) - PyMySQLInstrumentor().instrument() + PyMySQLInstrumentor().instrument(enable_transaction_spans=True) self._connection = pymy.connect( user=MYSQL_USER, password=MYSQL_PASSWORD, @@ -45,28 +45,35 @@ def tearDown(self): PyMySQLInstrumentor().uninstrument() super().tearDown() - def validate_spans(self, span_name): + def validate_spans(self, *span_names): spans = self.memory_exporter.get_finished_spans() - self.assertEqual(len(spans), 2) + self.assertEqual(len(spans), len(span_names) + 1) # +1 for rootSpan + + root_span = None + db_spans = [] + for span in spans: if span.name == "rootSpan": root_span = span else: - db_span = span + db_spans.append(span) self.assertIsInstance(span.start_time, int) self.assertIsInstance(span.end_time, int) + self.assertIsNotNone(root_span) - self.assertIsNotNone(db_span) self.assertEqual(root_span.name, "rootSpan") - self.assertEqual(db_span.name, span_name) - self.assertIsNotNone(db_span.parent) - self.assertIs(db_span.parent, root_span.get_span_context()) - self.assertIs(db_span.kind, trace_api.SpanKind.CLIENT) - self.assertEqual(db_span.attributes[DB_SYSTEM], "mysql") - self.assertEqual(db_span.attributes[DB_NAME], MYSQL_DB_NAME) - self.assertEqual(db_span.attributes[DB_USER], MYSQL_USER) - self.assertEqual(db_span.attributes[NET_PEER_NAME], MYSQL_HOST) - self.assertEqual(db_span.attributes[NET_PEER_PORT], MYSQL_PORT) + self.assertEqual(len(db_spans), len(span_names)) + + for db_span, expected_name in zip(db_spans, span_names): + self.assertEqual(db_span.name, expected_name) + self.assertIsNotNone(db_span.parent) + self.assertIs(db_span.parent, root_span.get_span_context()) + self.assertIs(db_span.kind, trace_api.SpanKind.CLIENT) + self.assertEqual(db_span.attributes[DB_SYSTEM], "mysql") + self.assertEqual(db_span.attributes[DB_NAME], MYSQL_DB_NAME) + self.assertEqual(db_span.attributes[DB_USER], MYSQL_USER) + self.assertEqual(db_span.attributes[NET_PEER_NAME], MYSQL_HOST) + self.assertEqual(db_span.attributes[NET_PEER_PORT], MYSQL_PORT) def test_execute(self): """Should create a child span for execute""" @@ -111,17 +118,31 @@ def test_callproc(self): self.validate_spans("test") def test_commit(self): + """Should create spans for both INSERT and COMMIT""" stmt = "INSERT INTO test (id) VALUES (%s)" with self._tracer.start_as_current_span("rootSpan"): data = (("4",), ("5",), ("6",)) self._cursor.executemany(stmt, data) self._connection.commit() - self.validate_spans("INSERT") + self.validate_spans("INSERT", "COMMIT") def test_rollback(self): + """Should create spans for both INSERT and ROLLBACK""" 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", "ROLLBACK") + + def test_commit_only(self): + """Should create a span for standalone COMMIT""" + with self._tracer.start_as_current_span("rootSpan"): + self._connection.commit() + self.validate_spans("COMMIT") + + def test_rollback_only(self): + """Should create a span for standalone ROLLBACK""" + with self._tracer.start_as_current_span("rootSpan"): + self._connection.rollback() + self.validate_spans("ROLLBACK")