diff --git a/CHANGELOG.md b/CHANGELOG.md index 46f505e998..257510fec0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ([#4212](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4212)) - `opentelemetry-instrumentation-botocore`: Add support for instrumenting `aiobotocore` ([#4049](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4049)) +- `opentelemetry-instrumentation-sqlalchemy`: implement new semantic convention opt-in migration + ([#4110](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4110)) ### Fixed diff --git a/instrumentation/README.md b/instrumentation/README.md index 2802d8a872..8d9a247945 100644 --- a/instrumentation/README.md +++ b/instrumentation/README.md @@ -41,7 +41,7 @@ | [opentelemetry-instrumentation-redis](./opentelemetry-instrumentation-redis) | redis >= 2.6 | No | development | [opentelemetry-instrumentation-remoulade](./opentelemetry-instrumentation-remoulade) | remoulade >= 0.50 | No | development | [opentelemetry-instrumentation-requests](./opentelemetry-instrumentation-requests) | requests ~= 2.0 | Yes | migration -| [opentelemetry-instrumentation-sqlalchemy](./opentelemetry-instrumentation-sqlalchemy) | sqlalchemy >= 1.0.0, < 2.1.0 | Yes | development +| [opentelemetry-instrumentation-sqlalchemy](./opentelemetry-instrumentation-sqlalchemy) | sqlalchemy >= 1.0.0, < 2.1.0 | Yes | migration | [opentelemetry-instrumentation-sqlite3](./opentelemetry-instrumentation-sqlite3) | sqlite3 | No | development | [opentelemetry-instrumentation-starlette](./opentelemetry-instrumentation-starlette) | starlette >= 0.13 | Yes | development | [opentelemetry-instrumentation-system-metrics](./opentelemetry-instrumentation-system-metrics) | psutil >= 5 | No | development diff --git a/instrumentation/opentelemetry-instrumentation-sqlalchemy/src/opentelemetry/instrumentation/sqlalchemy/__init__.py b/instrumentation/opentelemetry-instrumentation-sqlalchemy/src/opentelemetry/instrumentation/sqlalchemy/__init__.py index a95266351f..dbcffc5ecf 100644 --- a/instrumentation/opentelemetry-instrumentation-sqlalchemy/src/opentelemetry/instrumentation/sqlalchemy/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-sqlalchemy/src/opentelemetry/instrumentation/sqlalchemy/__init__.py @@ -110,16 +110,16 @@ SQLComment in span attribute **************************** If sqlcommenter is enabled, you can opt into the inclusion of sqlcomment in -the query span ``db.statement`` attribute for your needs. If ``commenter_options`` -have been set, the span attribute comment will also be configured by this -setting. +the query span ``db.statement`` and/or ``db.query.text`` attribute for your +needs. If ``commenter_options`` have been set, the span attribute comment +will also be configured by this setting. .. code:: python from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor # Opts into sqlcomment for SQLAlchemy trace integration. - # Opts into sqlcomment for `db.statement` span attribute. + # Opts into sqlcomment for `db.statement` and/or `db.query.text` span attribute. SQLAlchemyInstrumentor().instrument( enable_commenter=True, commenter_options={}, @@ -127,7 +127,7 @@ ) Warning: - Capture of sqlcomment in ``db.statement`` may have high cardinality without platform normalization. See `Semantic Conventions for database spans `_ for more information. + Capture of sqlcomment in ``db.statement``/``db.query.text`` may have high cardinality without platform normalization. See `Semantic Conventions for database spans `_ for more information. API --- @@ -141,6 +141,11 @@ from sqlalchemy.engine.base import Engine from wrapt import wrap_function_wrapper as _w +from opentelemetry.instrumentation._semconv import ( + _get_schema_url_for_signal_types, + _OpenTelemetrySemanticConventionStability, + _OpenTelemetryStabilitySignalType, +) from opentelemetry.instrumentation.instrumentor import BaseInstrumentor from opentelemetry.instrumentation.sqlalchemy.engine import ( EngineTracer, @@ -181,12 +186,24 @@ def _instrument(self, **kwargs): Returns: An instrumented engine if passed in as an argument or list of instrumented engines, None otherwise. """ + # Initialize semantic conventions opt-in if needed + _OpenTelemetrySemanticConventionStability._initialize() + + # Determine schema URL based on both DATABASE and HTTP signal types + # and semconv opt-in mode + schema_url = _get_schema_url_for_signal_types( + [ + _OpenTelemetryStabilitySignalType.DATABASE, + _OpenTelemetryStabilitySignalType.HTTP, + ] + ) + tracer_provider = kwargs.get("tracer_provider") tracer = get_tracer( __name__, __version__, tracer_provider, - schema_url="https://opentelemetry.io/schemas/1.11.0", + schema_url=schema_url, ) meter_provider = kwargs.get("meter_provider") @@ -194,7 +211,7 @@ def _instrument(self, **kwargs): __name__, __version__, meter_provider, - schema_url="https://opentelemetry.io/schemas/1.11.0", + schema_url=schema_url, ) connections_usage = meter.create_up_down_counter( diff --git a/instrumentation/opentelemetry-instrumentation-sqlalchemy/src/opentelemetry/instrumentation/sqlalchemy/engine.py b/instrumentation/opentelemetry-instrumentation-sqlalchemy/src/opentelemetry/instrumentation/sqlalchemy/engine.py index 6a380a3221..d357798ffa 100644 --- a/instrumentation/opentelemetry-instrumentation-sqlalchemy/src/opentelemetry/instrumentation/sqlalchemy/engine.py +++ b/instrumentation/opentelemetry-instrumentation-sqlalchemy/src/opentelemetry/instrumentation/sqlalchemy/engine.py @@ -22,26 +22,91 @@ ) from opentelemetry import trace +from opentelemetry.instrumentation._semconv import ( + _OpenTelemetrySemanticConventionStability, + _OpenTelemetryStabilitySignalType, + _set_db_name, + _set_db_operation, + _set_db_statement, + _set_db_system, + _set_db_user, + _set_http_net_peer_name_client, + _set_http_peer_port_client, +) from opentelemetry.instrumentation.sqlcommenter_utils import _add_sql_comment from opentelemetry.instrumentation.utils import ( _get_opentelemetry_values, is_instrumentation_enabled, ) -from opentelemetry.semconv._incubating.attributes.db_attributes import ( - DB_NAME, - DB_STATEMENT, - DB_SYSTEM, - DB_USER, -) from opentelemetry.semconv._incubating.attributes.net_attributes import ( - NET_PEER_NAME, - NET_PEER_PORT, NET_TRANSPORT, NetTransportValues, ) from opentelemetry.trace.status import Status, StatusCode +def _get_db_name_from_cursor_or_conn(vendor, conn, cursor): + """Return DB name from cursor or connection when available -- else None.""" + if not vendor: + return None + + vendor = vendor.lower() + db_name = None + if "postgres" in vendor: + info = getattr(getattr(cursor, "connection", None), "info", None) + if info and hasattr(info, "dbname"): + db_name = info.dbname + elif "mysql" in vendor: + db_name = _get_mysql_db_name(cursor) + elif "mssql" in vendor or "sqlserver" in vendor: + db_name = _get_mssql_db_name(cursor) + if not db_name: + engine = getattr(conn, "engine", None) + url = getattr(engine, "url", None) + db_name = getattr(url, "database", None) + else: + # Try connection for sqlite and others + engine = getattr(conn, "engine", None) + url = getattr(engine, "url", None) + db_name = getattr(url, "database", None) + return db_name + + +def _get_mysql_db_name(cursor): + """Extract database name from MySQL cursor.""" + # mysql-connector with c-extension uses _cnx + connection = getattr(cursor, "connection", None) or getattr( + cursor, "_cnx", None + ) + if not connection: + return None + if hasattr(connection, "database"): + return connection.database + if hasattr(connection, "db"): + raw_db_name = connection.db + return ( + raw_db_name.decode("utf-8") + if isinstance(raw_db_name, bytes) + else raw_db_name + ) + return None + + +def _get_mssql_db_name(cursor): + """Extract database name from MSSQL cursor.""" + connection = getattr(cursor, "connection", None) + if not connection: + return None + if hasattr(connection, "database"): + return connection.database + if hasattr(connection, "db"): + return connection.db + info = getattr(connection, "info", None) + if info and hasattr(info, "database"): + return info.database + return None + + def _normalize_vendor(vendor): """Return a canonical name for a type of database.""" if not vendor: @@ -119,13 +184,30 @@ def _wrap_connect_internal(func, module, args, kwargs): if not is_instrumentation_enabled(): return func(*args, **kwargs) + # Initialize semantic conventions opt-in if needed + _OpenTelemetrySemanticConventionStability._initialize() + sem_conv_opt_in_mode_db = _OpenTelemetrySemanticConventionStability._get_opentelemetry_stability_opt_in_mode( + _OpenTelemetryStabilitySignalType.DATABASE, + ) + sem_conv_opt_in_mode_http = _OpenTelemetrySemanticConventionStability._get_opentelemetry_stability_opt_in_mode( + _OpenTelemetryStabilitySignalType.HTTP, + ) + with tracer.start_as_current_span( "connect", kind=trace.SpanKind.CLIENT ) as span: if span.is_recording(): - attrs, _ = _get_attributes_from_url(module.url) + attrs, _ = _get_attributes_from_url( + module.url, + sem_conv_opt_in_mode_db, + sem_conv_opt_in_mode_http, + ) + _set_db_system( + attrs, + _normalize_vendor(module.name), + sem_conv_opt_in_mode_db, + ) span.set_attributes(attrs) - span.set_attribute(DB_SYSTEM, _normalize_vendor(module.name)) return func(*args, **kwargs) return _wrap_connect_internal @@ -143,6 +225,15 @@ def __init__( commenter_options=None, enable_attribute_commenter=False, ): + # Initialize semantic conventions opt-in if needed + _OpenTelemetrySemanticConventionStability._initialize() + self._sem_conv_opt_in_mode_db = _OpenTelemetrySemanticConventionStability._get_opentelemetry_stability_opt_in_mode( + _OpenTelemetryStabilitySignalType.DATABASE, + ) + self._sem_conv_opt_in_mode_http = _OpenTelemetrySemanticConventionStability._get_opentelemetry_stability_opt_in_mode( + _OpenTelemetryStabilitySignalType.HTTP, + ) + self.tracer = tracer self.connections_usage = connections_usage self.vendor = _normalize_vendor(engine.name) @@ -276,11 +367,19 @@ def _get_commenter_data(self, conn) -> dict: } return commenter_data - def _set_db_client_span_attributes(self, span, statement, attrs) -> None: - """Uses statement and attrs to set attributes of provided Otel span""" - span.set_attribute(DB_STATEMENT, statement) - span.set_attribute(DB_SYSTEM, self.vendor) - for key, value in attrs.items(): + def _set_db_client_span_attributes( + self, span, statement, db_name, attrs + ) -> None: + """Uses statement, db_name, and attrs to set attributes of provided Otel span""" + span_attrs = dict(attrs) + _set_db_statement(span_attrs, statement, self._sem_conv_opt_in_mode_db) + _set_db_system(span_attrs, self.vendor, self._sem_conv_opt_in_mode_db) + _set_db_operation( + span_attrs, + self._operation_name(db_name, statement), + self._sem_conv_opt_in_mode_db, + ) + for key, value in span_attrs.items(): span.set_attribute(key, value) def _before_cur_exec( @@ -289,11 +388,24 @@ def _before_cur_exec( if not is_instrumentation_enabled(): return statement, params - attrs, found = _get_attributes_from_url(conn.engine.url) + attrs, found = _get_attributes_from_url( + conn.engine.url, + self._sem_conv_opt_in_mode_db, + self._sem_conv_opt_in_mode_http, + ) if not found: - attrs = _get_attributes_from_cursor(self.vendor, cursor, attrs) + attrs = _get_attributes_from_cursor_or_conn( + self.vendor, + conn, + cursor, + attrs, + self._sem_conv_opt_in_mode_db, + self._sem_conv_opt_in_mode_http, + ) + + # Extract db_name for operation name + db_name = _get_db_name_from_cursor_or_conn(self.vendor, conn, cursor) - db_name = attrs.get(DB_NAME, "") span = self.tracer.start_span( self._operation_name(db_name, statement), kind=trace.SpanKind.CLIENT, @@ -307,19 +419,19 @@ def _before_cur_exec( # just to handle type safety statement = str(statement) - # sqlcomment is added to executed query and db.statement span attribute + # sqlcomment is added to executed query and db.statement and/or db.query.text span attribute statement = _add_sql_comment( statement, **commenter_data ) self._set_db_client_span_attributes( - span, statement, attrs + span, statement, db_name, attrs ) else: # sqlcomment is only added to executed query - # so db.statement is set before add_sql_comment + # so db.statement and/or db.query.text is set before add_sql_comment self._set_db_client_span_attributes( - span, statement, attrs + span, statement, db_name, attrs ) statement = _add_sql_comment( statement, **commenter_data @@ -327,7 +439,9 @@ def _before_cur_exec( else: # no sqlcomment anywhere - self._set_db_client_span_attributes(span, statement, attrs) + self._set_db_client_span_attributes( + span, statement, db_name, attrs + ) context._otel_span = span @@ -358,42 +472,64 @@ def _handle_error(context): span.end() -def _get_attributes_from_url(url): +def _get_attributes_from_url( + url, sem_conv_opt_in_mode_db, sem_conv_opt_in_mode_http +): """Set connection tags from the url. return true if successful.""" attrs = {} if url.host: - attrs[NET_PEER_NAME] = url.host + _set_http_net_peer_name_client( + attrs, url.host, sem_conv_opt_in_mode_http + ) if url.port: - attrs[NET_PEER_PORT] = url.port + _set_http_peer_port_client(attrs, url.port, sem_conv_opt_in_mode_http) if url.database: - attrs[DB_NAME] = url.database + _set_db_name(attrs, url.database, sem_conv_opt_in_mode_db) if url.username: - attrs[DB_USER] = url.username + _set_db_user(attrs, url.username, sem_conv_opt_in_mode_db) return attrs, bool(url.host) -def _get_attributes_from_cursor(vendor, cursor, attrs): +def _get_attributes_from_cursor_or_conn( + vendor, + conn, + cursor, + attrs, + sem_conv_opt_in_mode_db, + sem_conv_opt_in_mode_http, +): """Attempt to set db connection attributes by introspecting the cursor.""" if vendor == "postgresql": info = getattr(getattr(cursor, "connection", None), "info", None) if not info: return attrs - attrs[DB_NAME] = info.dbname + db_name = _get_db_name_from_cursor_or_conn(vendor, conn, cursor) + _set_db_name(attrs, db_name, sem_conv_opt_in_mode_db) is_unix_socket = info.host and info.host.startswith("/") if is_unix_socket: attrs[NET_TRANSPORT] = NetTransportValues.OTHER.value if info.port: # postgresql enforces this pattern on all socket names - attrs[NET_PEER_NAME] = os.path.join( - info.host, f".s.PGSQL.{info.port}" + _set_http_net_peer_name_client( + attrs, + os.path.join(info.host, f".s.PGSQL.{info.port}"), + sem_conv_opt_in_mode_http, ) else: attrs[NET_TRANSPORT] = NetTransportValues.IP_TCP.value - attrs[NET_PEER_NAME] = info.host + _set_http_net_peer_name_client( + attrs, info.host, sem_conv_opt_in_mode_http + ) if info.port: - attrs[NET_PEER_PORT] = int(info.port) + _set_http_peer_port_client( + attrs, int(info.port), sem_conv_opt_in_mode_http + ) + elif vendor == "sqlite": + db_name = _get_db_name_from_cursor_or_conn(vendor, conn, cursor) + _set_db_name(attrs, db_name, sem_conv_opt_in_mode_db) + # SQLite has no network attributes return attrs diff --git a/instrumentation/opentelemetry-instrumentation-sqlalchemy/src/opentelemetry/instrumentation/sqlalchemy/package.py b/instrumentation/opentelemetry-instrumentation-sqlalchemy/src/opentelemetry/instrumentation/sqlalchemy/package.py index 77b148e51c..a95b8a6c76 100644 --- a/instrumentation/opentelemetry-instrumentation-sqlalchemy/src/opentelemetry/instrumentation/sqlalchemy/package.py +++ b/instrumentation/opentelemetry-instrumentation-sqlalchemy/src/opentelemetry/instrumentation/sqlalchemy/package.py @@ -15,3 +15,5 @@ _instruments = ("sqlalchemy >= 1.0.0, < 2.1.0",) _supports_metrics = True + +_semconv_status = "migration" diff --git a/instrumentation/opentelemetry-instrumentation-sqlalchemy/tests/test_sqlalchemy.py b/instrumentation/opentelemetry-instrumentation-sqlalchemy/tests/test_sqlalchemy.py index 882659daf7..36257d9d32 100644 --- a/instrumentation/opentelemetry-instrumentation-sqlalchemy/tests/test_sqlalchemy.py +++ b/instrumentation/opentelemetry-instrumentation-sqlalchemy/tests/test_sqlalchemy.py @@ -23,18 +23,35 @@ ) from opentelemetry import trace +from opentelemetry.instrumentation._semconv import ( + OTEL_SEMCONV_STABILITY_OPT_IN, + _OpenTelemetrySemanticConventionStability, +) from opentelemetry.instrumentation.sqlalchemy import ( EngineTracer, SQLAlchemyInstrumentor, ) +from opentelemetry.instrumentation.sqlalchemy.engine import ( + _get_db_name_from_cursor_or_conn, +) from opentelemetry.instrumentation.utils import suppress_instrumentation from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import TracerProvider, export from opentelemetry.semconv._incubating.attributes.db_attributes import ( DB_NAME, + DB_OPERATION, DB_STATEMENT, DB_SYSTEM, ) +from opentelemetry.semconv._incubating.attributes.net_attributes import ( + NET_PEER_NAME, +) +from opentelemetry.semconv.attributes.db_attributes import ( + DB_NAMESPACE, + DB_OPERATION_NAME, + DB_QUERY_TEXT, + DB_SYSTEM_NAME, +) from opentelemetry.test.test_base import TestBase @@ -43,6 +60,12 @@ class TestSqlalchemyInstrumentation(TestBase): def inject_fixtures(self, caplog): self.caplog = caplog # pylint: disable=attribute-defined-outside-init + def setUp(self): + super().setUp() + # Reset semconv state before each test for reproducibility + # Tests using @mock.patch.dict will re-initialize again after the mock is applied + _OpenTelemetrySemanticConventionStability._initialized = False + def tearDown(self): super().tearDown() SQLAlchemyInstrumentor().uninstrument() @@ -687,6 +710,276 @@ def test_suppress_instrumentation_connect(self): spans_list = self.memory_exporter.get_finished_spans() self.assertEqual(len(spans_list), 0) + def test_semconv_default_mode(self): + SQLAlchemyInstrumentor().uninstrument() + engine = create_engine("sqlite:///:memory:") + SQLAlchemyInstrumentor().instrument( + engine=engine, + tracer_provider=self.tracer_provider, + ) + cnx = engine.connect() + cnx.execute(text("SELECT 1 + 1;")).fetchall() + spans = self.memory_exporter.get_finished_spans() + + connect_span = spans[0] + self.assertEqual( + connect_span.instrumentation_scope.schema_url, + "https://opentelemetry.io/schemas/1.11.0", + ) + self.assertIn(DB_SYSTEM, connect_span.attributes) + self.assertEqual(connect_span.attributes[DB_SYSTEM], "sqlite") + self.assertIn(DB_NAME, connect_span.attributes) + self.assertEqual(connect_span.attributes[DB_NAME], ":memory:") + # Verify new conventions are NOT present + self.assertNotIn(DB_SYSTEM_NAME, connect_span.attributes) + self.assertNotIn(DB_NAMESPACE, connect_span.attributes) + + query_span = spans[1] + self.assertIn(DB_STATEMENT, query_span.attributes) + self.assertIn(DB_SYSTEM, query_span.attributes) + self.assertEqual(query_span.attributes[DB_SYSTEM], "sqlite") + self.assertIn(DB_OPERATION, query_span.attributes) + self.assertEqual( + query_span.attributes[DB_OPERATION], "SELECT :memory:" + ) + # Verify new conventions are NOT present + self.assertNotIn(DB_QUERY_TEXT, query_span.attributes) + self.assertNotIn(DB_SYSTEM_NAME, query_span.attributes) + + @mock.patch.dict("os.environ", {OTEL_SEMCONV_STABILITY_OPT_IN: "http"}) + def test_semconv_http_mode(self): + SQLAlchemyInstrumentor().uninstrument() + _OpenTelemetrySemanticConventionStability._initialized = False + _OpenTelemetrySemanticConventionStability._initialize() + engine = create_engine("sqlite:///:memory:") + SQLAlchemyInstrumentor().instrument( + engine=engine, + tracer_provider=self.tracer_provider, + ) + cnx = engine.connect() + cnx.execute(text("SELECT 1 + 1;")).fetchall() + spans = self.memory_exporter.get_finished_spans() + + connect_span = spans[0] + self.assertEqual( + connect_span.instrumentation_scope.schema_url, + "https://opentelemetry.io/schemas/1.21.0", + ) + # HTTP attributes should use new semconv + self.assertNotIn(NET_PEER_NAME, connect_span.attributes) + # DB attributes should still use old semconv + self.assertIn(DB_SYSTEM, connect_span.attributes) + self.assertEqual(connect_span.attributes[DB_SYSTEM], "sqlite") + self.assertIn(DB_NAME, connect_span.attributes) + self.assertEqual(connect_span.attributes[DB_NAME], ":memory:") + # Verify new DB conventions are NOT present + self.assertNotIn(DB_SYSTEM_NAME, connect_span.attributes) + self.assertNotIn(DB_NAMESPACE, connect_span.attributes) + + query_span = spans[1] + # DB attributes should still use old semconv + self.assertIn(DB_STATEMENT, query_span.attributes) + self.assertIn(DB_SYSTEM, query_span.attributes) + self.assertEqual(query_span.attributes[DB_SYSTEM], "sqlite") + # Verify new DB conventions are NOT present + self.assertNotIn(DB_QUERY_TEXT, query_span.attributes) + self.assertNotIn(DB_SYSTEM_NAME, query_span.attributes) + + @mock.patch.dict("os.environ", {OTEL_SEMCONV_STABILITY_OPT_IN: "database"}) + def test_semconv_database_mode(self): + SQLAlchemyInstrumentor().uninstrument() + _OpenTelemetrySemanticConventionStability._initialized = False + _OpenTelemetrySemanticConventionStability._initialize() + engine = create_engine("sqlite:///:memory:") + SQLAlchemyInstrumentor().instrument( + engine=engine, + tracer_provider=self.tracer_provider, + ) + cnx = engine.connect() + cnx.execute(text("SELECT 1 + 1;")).fetchall() + spans = self.memory_exporter.get_finished_spans() + + connect_span = spans[0] + self.assertEqual( + connect_span.instrumentation_scope.schema_url, + "https://opentelemetry.io/schemas/1.25.0", + ) + self.assertIn(DB_SYSTEM_NAME, connect_span.attributes) + self.assertEqual(connect_span.attributes[DB_SYSTEM_NAME], "sqlite") + self.assertIn(DB_NAMESPACE, connect_span.attributes) + self.assertEqual(connect_span.attributes[DB_NAMESPACE], ":memory:") + # Verify old conventions are NOT present + self.assertNotIn(DB_SYSTEM, connect_span.attributes) + self.assertNotIn(DB_NAME, connect_span.attributes) + + query_span = spans[1] + self.assertIn(DB_QUERY_TEXT, query_span.attributes) + self.assertIn(DB_SYSTEM_NAME, query_span.attributes) + self.assertEqual(query_span.attributes[DB_SYSTEM_NAME], "sqlite") + self.assertIn(DB_OPERATION_NAME, query_span.attributes) + self.assertEqual( + query_span.attributes[DB_OPERATION_NAME], "SELECT :memory:" + ) + # Verify old conventions are NOT present + self.assertNotIn(DB_STATEMENT, query_span.attributes) + self.assertNotIn(DB_SYSTEM, query_span.attributes) + + @mock.patch.dict( + "os.environ", {OTEL_SEMCONV_STABILITY_OPT_IN: "http,database"} + ) + def test_semconv_http_and_database_mode(self): + SQLAlchemyInstrumentor().uninstrument() + _OpenTelemetrySemanticConventionStability._initialized = False + _OpenTelemetrySemanticConventionStability._initialize() + engine = create_engine("sqlite:///:memory:") + SQLAlchemyInstrumentor().instrument( + engine=engine, + tracer_provider=self.tracer_provider, + ) + cnx = engine.connect() + cnx.execute(text("SELECT 1 + 1;")).fetchall() + spans = self.memory_exporter.get_finished_spans() + + connect_span = spans[0] + self.assertEqual( + connect_span.instrumentation_scope.schema_url, + "https://opentelemetry.io/schemas/1.25.0", + ) + # New DB conventions should be present + self.assertIn(DB_SYSTEM_NAME, connect_span.attributes) + self.assertEqual(connect_span.attributes[DB_SYSTEM_NAME], "sqlite") + self.assertIn(DB_NAMESPACE, connect_span.attributes) + self.assertEqual(connect_span.attributes[DB_NAMESPACE], ":memory:") + # Old DB conventions should NOT be present + self.assertNotIn(DB_SYSTEM, connect_span.attributes) + self.assertNotIn(DB_NAME, connect_span.attributes) + + query_span = spans[1] + # New DB conventions should be present + self.assertIn(DB_QUERY_TEXT, query_span.attributes) + self.assertIn(DB_SYSTEM_NAME, query_span.attributes) + self.assertEqual(query_span.attributes[DB_SYSTEM_NAME], "sqlite") + # Old DB conventions should NOT be present + self.assertNotIn(DB_STATEMENT, query_span.attributes) + self.assertNotIn(DB_SYSTEM, query_span.attributes) + + @mock.patch.dict("os.environ", {OTEL_SEMCONV_STABILITY_OPT_IN: "http/dup"}) + def test_semconv_http_dup_mode(self): + SQLAlchemyInstrumentor().uninstrument() + _OpenTelemetrySemanticConventionStability._initialized = False + _OpenTelemetrySemanticConventionStability._initialize() + engine = create_engine("sqlite:///:memory:") + SQLAlchemyInstrumentor().instrument( + engine=engine, + tracer_provider=self.tracer_provider, + ) + cnx = engine.connect() + cnx.execute(text("SELECT 1 + 1;")).fetchall() + spans = self.memory_exporter.get_finished_spans() + + connect_span = spans[0] + self.assertEqual( + connect_span.instrumentation_scope.schema_url, + "https://opentelemetry.io/schemas/1.21.0", + ) + # DB attributes should use old semconv only + self.assertIn(DB_SYSTEM, connect_span.attributes) + self.assertEqual(connect_span.attributes[DB_SYSTEM], "sqlite") + self.assertIn(DB_NAME, connect_span.attributes) + self.assertEqual(connect_span.attributes[DB_NAME], ":memory:") + # Verify new DB conventions are NOT present + self.assertNotIn(DB_SYSTEM_NAME, connect_span.attributes) + self.assertNotIn(DB_NAMESPACE, connect_span.attributes) + + query_span = spans[1] + # DB attributes should use old semconv + self.assertIn(DB_STATEMENT, query_span.attributes) + self.assertIn(DB_SYSTEM, query_span.attributes) + self.assertEqual(query_span.attributes[DB_SYSTEM], "sqlite") + # Verify new DB conventions are NOT present + self.assertNotIn(DB_QUERY_TEXT, query_span.attributes) + self.assertNotIn(DB_SYSTEM_NAME, query_span.attributes) + + @mock.patch.dict( + "os.environ", {OTEL_SEMCONV_STABILITY_OPT_IN: "database/dup"} + ) + def test_semconv_database_dup_mode(self): + SQLAlchemyInstrumentor().uninstrument() + engine = create_engine("sqlite:///:memory:") + SQLAlchemyInstrumentor().instrument( + engine=engine, + tracer_provider=self.tracer_provider, + ) + cnx = engine.connect() + cnx.execute(text("SELECT 1 + 1;")).fetchall() + spans = self.memory_exporter.get_finished_spans() + + connect_span = spans[0] + self.assertEqual( + connect_span.instrumentation_scope.schema_url, + "https://opentelemetry.io/schemas/1.25.0", + ) + # Old conventions + self.assertIn(DB_SYSTEM, connect_span.attributes) + self.assertEqual(connect_span.attributes[DB_SYSTEM], "sqlite") + self.assertIn(DB_NAME, connect_span.attributes) + self.assertEqual(connect_span.attributes[DB_NAME], ":memory:") + # New conventions + self.assertIn(DB_SYSTEM_NAME, connect_span.attributes) + self.assertEqual(connect_span.attributes[DB_SYSTEM_NAME], "sqlite") + self.assertIn(DB_NAMESPACE, connect_span.attributes) + self.assertEqual(connect_span.attributes[DB_NAMESPACE], ":memory:") + + query_span = spans[1] + # Old conventions + self.assertIn(DB_STATEMENT, query_span.attributes) + self.assertIn(DB_SYSTEM, query_span.attributes) + self.assertEqual(query_span.attributes[DB_SYSTEM], "sqlite") + # New conventions + self.assertIn(DB_QUERY_TEXT, query_span.attributes) + self.assertIn(DB_SYSTEM_NAME, query_span.attributes) + self.assertEqual(query_span.attributes[DB_SYSTEM_NAME], "sqlite") + + @mock.patch.dict( + "os.environ", {OTEL_SEMCONV_STABILITY_OPT_IN: "http/dup,database/dup"} + ) + def test_semconv_http_and_database_dup_mode(self): + SQLAlchemyInstrumentor().uninstrument() + _OpenTelemetrySemanticConventionStability._initialized = False + _OpenTelemetrySemanticConventionStability._initialize() + engine = create_engine("sqlite:///:memory:") + SQLAlchemyInstrumentor().instrument( + engine=engine, + tracer_provider=self.tracer_provider, + ) + cnx = engine.connect() + cnx.execute(text("SELECT 1 + 1;")).fetchall() + spans = self.memory_exporter.get_finished_spans() + + connect_span = spans[0] + self.assertEqual( + connect_span.instrumentation_scope.schema_url, + "https://opentelemetry.io/schemas/1.25.0", + ) + # Both old and new DB conventions + self.assertIn(DB_SYSTEM, connect_span.attributes) + self.assertEqual(connect_span.attributes[DB_SYSTEM], "sqlite") + self.assertIn(DB_NAME, connect_span.attributes) + self.assertEqual(connect_span.attributes[DB_NAME], ":memory:") + self.assertIn(DB_SYSTEM_NAME, connect_span.attributes) + self.assertEqual(connect_span.attributes[DB_SYSTEM_NAME], "sqlite") + self.assertIn(DB_NAMESPACE, connect_span.attributes) + self.assertEqual(connect_span.attributes[DB_NAMESPACE], ":memory:") + + query_span = spans[1] + # Both old and new DB conventions + self.assertIn(DB_STATEMENT, query_span.attributes) + self.assertIn(DB_SYSTEM, query_span.attributes) + self.assertEqual(query_span.attributes[DB_SYSTEM], "sqlite") + self.assertIn(DB_QUERY_TEXT, query_span.attributes) + self.assertIn(DB_SYSTEM_NAME, query_span.attributes) + self.assertEqual(query_span.attributes[DB_SYSTEM_NAME], "sqlite") + def test_suppress_instrumentation_cursor_and_metric(self): engine = create_engine("sqlite:///:memory:") SQLAlchemyInstrumentor().instrument( @@ -704,3 +997,145 @@ def test_suppress_instrumentation_cursor_and_metric(self): metric_list = self.get_sorted_metrics() self.assertEqual(len(metric_list), 0) + + def test_get_db_name_from_cursor_or_conn_postgresql_success(self): + mock_cursor = mock.Mock() + mock_connection = mock.Mock() + mock_info = mock.Mock() + mock_info.dbname = "test_database" + mock_connection.info = mock_info + mock_cursor.connection = mock_connection + result = _get_db_name_from_cursor_or_conn( + "postgresql", mock_connection, mock_cursor + ) + self.assertEqual(result, "test_database") + + def test_get_db_name_from_cursor_or_conn_postgresql_no_connection(self): + mock_cursor = mock.Mock() + mock_cursor.connection = None + mock_connection = mock.Mock() + result = _get_db_name_from_cursor_or_conn( + "postgresql", mock_connection, mock_cursor + ) + self.assertIsNone(result) + + def test_get_db_name_from_cursor_or_conn_postgresql_no_dbname(self): + mock_cursor = mock.Mock() + mock_connection = mock.Mock() + mock_info = mock.Mock() + mock_info.dbname = None + mock_connection.info = mock_info + mock_cursor.connection = mock_connection + result = _get_db_name_from_cursor_or_conn( + "postgresql", mock_connection, mock_cursor + ) + self.assertIsNone(result) + + def test_get_db_name_from_cursor_or_conn_unknown_vendor_with_url(self): + mock_cursor = mock.Mock() + mock_connection = mock.Mock() + mock_connection.engine.url.database = "foo" + mock_cursor.connection = mock_connection + result = _get_db_name_from_cursor_or_conn( + "unknown_db", mock_connection, mock_cursor + ) + self.assertEqual(result, "foo") + + def test_get_db_name_from_cursor_or_conn_unknown_vendor_without_url(self): + mock_cursor = mock.Mock() + mock_connection = mock.Mock() + mock_connection.engine.url.database = None + mock_cursor.connection = mock_connection + result = _get_db_name_from_cursor_or_conn( + "unknown_db", mock_connection, mock_cursor + ) + self.assertIsNone(result) + + def test_get_db_name_from_cursor_or_conn_mysql_with_database_attr(self): + mock_cursor = mock.Mock() + mock_connection = mock.Mock() + mock_connection.database = "mysql_test_db" + mock_cursor.connection = mock_connection + result = _get_db_name_from_cursor_or_conn( + "mysql", mock_connection, mock_cursor + ) + self.assertEqual(result, "mysql_test_db") + + def test_get_db_name_from_cursor_or_conn_mysql_with_db_attr_string(self): + mock_cursor = mock.Mock() + mock_connection = mock.Mock() + mock_connection.db = "mysql_test_db" + del mock_connection.database # Remove database attribute + mock_cursor.connection = mock_connection + result = _get_db_name_from_cursor_or_conn( + "mysql", mock_connection, mock_cursor + ) + self.assertEqual(result, "mysql_test_db") + + def test_get_db_name_from_cursor_or_conn_mysql_with_db_attr_bytes(self): + mock_cursor = mock.Mock() + mock_connection = mock.Mock() + mock_connection.db = b"mysql_test_db" + del mock_connection.database # Remove database attribute + mock_cursor.connection = mock_connection + result = _get_db_name_from_cursor_or_conn( + "mysql", mock_connection, mock_cursor + ) + self.assertEqual(result, "mysql_test_db") + + def test_get_db_name_from_cursor_or_conn_mysql_with_cnx_attr(self): + mock_cursor = mock.Mock() + mock_connection = mock.Mock() + mock_cnx = mock.Mock() + mock_cnx.database = "mysql_cnx_db" + mock_cursor.connection = None + mock_cursor._cnx = mock_cnx + result = _get_db_name_from_cursor_or_conn( + "mysql", mock_connection, mock_cursor + ) + self.assertEqual(result, "mysql_cnx_db") + + def test_get_db_name_from_cursor_or_conn_mssql_with_database_attr(self): + mock_cursor = mock.Mock() + mock_connection = mock.Mock() + mock_connection.database = "mssql_test_db" + mock_cursor.connection = mock_connection + result = _get_db_name_from_cursor_or_conn( + "mssql", mock_connection, mock_cursor + ) + self.assertEqual(result, "mssql_test_db") + + def test_get_db_name_from_cursor_or_conn_mysql_variant_names(self): + mock_cursor = mock.Mock() + mock_connection = mock.Mock() + mock_connection.database = "test_db" + mock_cursor.connection = mock_connection + for vendor in ["mysql", "pymysql", "mysqlclient", "mysql+pymysql"]: + result = _get_db_name_from_cursor_or_conn( + vendor, mock_connection, mock_cursor + ) + self.assertEqual(result, "test_db", f"Failed for vendor: {vendor}") + + def test_get_db_name_from_cursor_or_conn_mssql_variant_names(self): + mock_cursor = mock.Mock() + mock_connection = mock.Mock() + mock_connection.database = "test_db" + mock_cursor.connection = mock_connection + for vendor in ["mssql", "mssql+pyodbc", "sqlserver"]: + result = _get_db_name_from_cursor_or_conn( + vendor, mock_connection, mock_cursor + ) + self.assertEqual(result, "test_db", f"Failed for vendor: {vendor}") + + def test_get_db_name_from_cursor_or_conn_postgresql_variant_names(self): + mock_cursor = mock.Mock() + mock_connection = mock.Mock() + mock_info = mock.Mock() + mock_info.dbname = "test_db" + mock_connection.info = mock_info + mock_cursor.connection = mock_connection + for vendor in ["postgresql", "postgres", "postgresql+psycopg2"]: + result = _get_db_name_from_cursor_or_conn( + vendor, mock_connection, mock_cursor + ) + self.assertEqual(result, "test_db", f"Failed for vendor: {vendor}") diff --git a/instrumentation/opentelemetry-instrumentation-sqlalchemy/tests/test_sqlcommenter.py b/instrumentation/opentelemetry-instrumentation-sqlalchemy/tests/test_sqlcommenter.py index 946be53b3c..aeb9f74081 100644 --- a/instrumentation/opentelemetry-instrumentation-sqlalchemy/tests/test_sqlcommenter.py +++ b/instrumentation/opentelemetry-instrumentation-sqlalchemy/tests/test_sqlcommenter.py @@ -13,6 +13,7 @@ # limitations under the License. import logging import re +from unittest import mock import pytest from sqlalchemy import ( @@ -21,10 +22,17 @@ ) from opentelemetry import context +from opentelemetry.instrumentation._semconv import ( + OTEL_SEMCONV_STABILITY_OPT_IN, + _OpenTelemetrySemanticConventionStability, +) from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor from opentelemetry.semconv._incubating.attributes.db_attributes import ( DB_STATEMENT, ) +from opentelemetry.semconv.attributes.db_attributes import ( + DB_QUERY_TEXT, +) from opentelemetry.test.test_base import TestBase @@ -33,6 +41,10 @@ class TestSqlalchemyInstrumentationWithSQLCommenter(TestBase): def inject_fixtures(self, caplog): self.caplog = caplog # pylint: disable=attribute-defined-outside-init + def setUp(self): + super().setUp() + _OpenTelemetrySemanticConventionStability._initialized = False + def tearDown(self): super().tearDown() SQLAlchemyInstrumentor().uninstrument() @@ -358,6 +370,96 @@ def test_sqlcommenter_enabled_stmt_enabled_create_engine_after_instrumentation( r"SELECT 1 /\*db_driver='(.*)',traceparent='\d{1,2}-[a-zA-Z0-9_]{32}-[a-zA-Z0-9_]{16}-\d{1,2}'\*/;", ) + @mock.patch.dict("os.environ", {OTEL_SEMCONV_STABILITY_OPT_IN: "database"}) + def test_sqlcommenter_enabled_database_mode(self): + _OpenTelemetrySemanticConventionStability._initialized = False + _OpenTelemetrySemanticConventionStability._initialize() + logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO) + engine = create_engine("sqlite:///:memory:") + SQLAlchemyInstrumentor().instrument( + engine=engine, + tracer_provider=self.tracer_provider, + enable_commenter=True, + commenter_options={"db_framework": False}, + ) + cnx = engine.connect() + cnx.execute(text("SELECT 1;")).fetchall() + # Query log should have sqlcommenter + self.assertRegex( + self.caplog.records[-2].getMessage(), + r"SELECT 1 /\*db_driver='(.*)',traceparent='\d{1,2}-[a-zA-Z0-9_]{32}-[a-zA-Z0-9_]{16}-\d{1,2}'\*/;", + ) + spans = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans), 2) + query_span = spans[1] + # Should use new semconv attribute (db.query.text not db.statement) + self.assertNotIn(DB_STATEMENT, query_span.attributes) + self.assertIn(DB_QUERY_TEXT, query_span.attributes) + self.assertEqual(query_span.attributes[DB_QUERY_TEXT], "SELECT 1;") + + @mock.patch.dict("os.environ", {OTEL_SEMCONV_STABILITY_OPT_IN: "database"}) + def test_sqlcommenter_enabled_stmt_enabled_database_mode(self): + _OpenTelemetrySemanticConventionStability._initialized = False + _OpenTelemetrySemanticConventionStability._initialize() + logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO) + engine = create_engine("sqlite:///:memory:") + SQLAlchemyInstrumentor().instrument( + engine=engine, + tracer_provider=self.tracer_provider, + enable_commenter=True, + commenter_options={"db_framework": False}, + enable_attribute_commenter=True, + ) + cnx = engine.connect() + cnx.execute(text("SELECT 1;")).fetchall() + query_log = self.caplog.records[-2].getMessage() + self.assertRegex( + query_log, + r"SELECT 1 /\*db_driver='(.*)',traceparent='\d{1,2}-[a-zA-Z0-9_]{32}-[a-zA-Z0-9_]{16}-\d{1,2}'\*/;", + ) + spans = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans), 2) + query_span = spans[1] + # Should use new semconv attribute with comment + self.assertNotIn(DB_STATEMENT, query_span.attributes) + self.assertIn(DB_QUERY_TEXT, query_span.attributes) + self.assertRegex( + query_span.attributes[DB_QUERY_TEXT], + r"SELECT 1 /\*db_driver='(.*)',traceparent='\d{1,2}-[a-zA-Z0-9_]{32}-[a-zA-Z0-9_]{16}-\d{1,2}'\*/;", + ) + + @mock.patch.dict( + "os.environ", {OTEL_SEMCONV_STABILITY_OPT_IN: "database/dup"} + ) + def test_sqlcommenter_enabled_database_dup_mode(self): + _OpenTelemetrySemanticConventionStability._initialized = False + _OpenTelemetrySemanticConventionStability._initialize() + logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO) + engine = create_engine("sqlite:///:memory:") + SQLAlchemyInstrumentor().instrument( + engine=engine, + tracer_provider=self.tracer_provider, + enable_commenter=True, + commenter_options={"db_framework": False}, + enable_attribute_commenter=True, + ) + cnx = engine.connect() + cnx.execute(text("SELECT 1;")).fetchall() + spans = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans), 2) + query_span = spans[1] + # Should have both old and new semconv attributes with comment + self.assertIn(DB_STATEMENT, query_span.attributes) + self.assertRegex( + query_span.attributes[DB_STATEMENT], + r"SELECT 1 /\*db_driver='(.*)',traceparent='\d{1,2}-[a-zA-Z0-9_]{32}-[a-zA-Z0-9_]{16}-\d{1,2}'\*/;", + ) + self.assertIn(DB_QUERY_TEXT, query_span.attributes) + self.assertRegex( + query_span.attributes[DB_QUERY_TEXT], + r"SELECT 1 /\*db_driver='(.*)',traceparent='\d{1,2}-[a-zA-Z0-9_]{32}-[a-zA-Z0-9_]{16}-\d{1,2}'\*/;", + ) + def test_sqlcommenter_disabled_create_engine_after_instrumentation(self): SQLAlchemyInstrumentor().instrument( tracer_provider=self.tracer_provider, diff --git a/opentelemetry-instrumentation/src/opentelemetry/instrumentation/_semconv.py b/opentelemetry-instrumentation/src/opentelemetry/instrumentation/_semconv.py index ccf1f39202..1edd18d038 100644 --- a/opentelemetry-instrumentation/src/opentelemetry/instrumentation/_semconv.py +++ b/opentelemetry-instrumentation/src/opentelemetry/instrumentation/_semconv.py @@ -18,10 +18,14 @@ import threading from enum import Enum from typing import Container, Mapping, MutableMapping +from urllib.parse import urlparse + +from packaging import version as package_version from opentelemetry.instrumentation.utils import http_status_to_status_code from opentelemetry.semconv._incubating.attributes.db_attributes import ( DB_NAME, + DB_OPERATION, DB_STATEMENT, DB_SYSTEM, DB_USER, @@ -50,6 +54,7 @@ ) from opentelemetry.semconv.attributes.db_attributes import ( DB_NAMESPACE, + DB_OPERATION_NAME, DB_QUERY_TEXT, DB_SYSTEM_NAME, ) @@ -177,6 +182,9 @@ OTEL_SEMCONV_STABILITY_OPT_IN = "OTEL_SEMCONV_STABILITY_OPT_IN" +# Legacy/default schema version when schema_url was first introduced +_LEGACY_SCHEMA_VERSION = "1.11.0" + class _OpenTelemetryStabilitySignalType(Enum): HTTP = "http" @@ -590,6 +598,17 @@ def _set_db_user( # No new attribute - db.user was removed with no replacement +def _set_db_operation( + result: MutableMapping[str, AttributeValue], + operation: str, + sem_conv_opt_in_mode: _StabilityMode, +) -> None: + if _report_old(sem_conv_opt_in_mode): + set_string_attribute(result, DB_OPERATION, operation) + if _report_new(sem_conv_opt_in_mode): + set_string_attribute(result, DB_OPERATION_NAME, operation) + + # General @@ -634,8 +653,63 @@ def _set_status( span.set_status(Status(status)) -# Get schema version based off of opt-in mode def _get_schema_url(mode: _StabilityMode) -> str: + """Get schema version URL for a single signal type's opt-in mode (backwards compatible). + + For new instrumentations using multiple signal types, use + _get_schema_url_for_signal_types() + """ if mode is _StabilityMode.DEFAULT: - return "https://opentelemetry.io/schemas/1.11.0" + return f"https://opentelemetry.io/schemas/{_LEGACY_SCHEMA_VERSION}" return Schemas.V1_21_0.value + + +def _get_schema_version_for_opt_in_mode( + signal_type: _OpenTelemetryStabilitySignalType, + mode: _StabilityMode, +) -> str: + """Get the schema version for a specific signal type and opt-in mode.""" + if mode == _StabilityMode.DEFAULT: + return _LEGACY_SCHEMA_VERSION + + signal_versions = { + _OpenTelemetryStabilitySignalType.HTTP: Schemas.V1_21_0.value, + _OpenTelemetryStabilitySignalType.DATABASE: Schemas.V1_25_0.value, + _OpenTelemetryStabilitySignalType.GEN_AI: Schemas.V1_26_0.value, + } + schema_url = signal_versions.get(signal_type) + if not schema_url: + return _LEGACY_SCHEMA_VERSION + + path = urlparse(schema_url).path + schema_version = path.rstrip("/").split("/")[-1] + return schema_version or _LEGACY_SCHEMA_VERSION + + +def _get_schema_url_for_signal_types( + signal_types: list[_OpenTelemetryStabilitySignalType], +) -> str: + """Get the highest applicable schema URL for multiple signal types. + + Note: + Instrumentors should call _OpenTelemetrySemanticConventionStability._initialize() + before using this function to ensure proper initialization of stability modes. + + Args: + signal_types: List of signal types used by the instrumentation + + Returns: + Schema URL string representing the highest applicable semconv version + """ + highest_schema_version = _LEGACY_SCHEMA_VERSION + for signal_type in signal_types: + mode = _OpenTelemetrySemanticConventionStability._get_opentelemetry_stability_opt_in_mode( + signal_type + ) + schema_version = _get_schema_version_for_opt_in_mode(signal_type, mode) + # Keep the highest for all signals + if package_version.Version(schema_version) > package_version.Version( + highest_schema_version + ): + highest_schema_version = schema_version + return f"https://opentelemetry.io/schemas/{highest_schema_version}" diff --git a/opentelemetry-instrumentation/tests/test_semconv.py b/opentelemetry-instrumentation/tests/test_semconv.py index 4560e28101..f160c51286 100644 --- a/opentelemetry-instrumentation/tests/test_semconv.py +++ b/opentelemetry-instrumentation/tests/test_semconv.py @@ -17,10 +17,14 @@ from unittest.mock import Mock, patch from opentelemetry.instrumentation._semconv import ( + _LEGACY_SCHEMA_VERSION, OTEL_SEMCONV_STABILITY_OPT_IN, + _get_schema_url_for_signal_types, + _get_schema_version_for_opt_in_mode, _OpenTelemetrySemanticConventionStability, _OpenTelemetryStabilitySignalType, _set_db_name, + _set_db_operation, _set_db_statement, _set_db_system, _set_db_user, @@ -29,12 +33,14 @@ ) from opentelemetry.semconv._incubating.attributes.db_attributes import ( DB_NAME, + DB_OPERATION, DB_STATEMENT, DB_SYSTEM, DB_USER, ) from opentelemetry.semconv.attributes.db_attributes import ( DB_NAMESPACE, + DB_OPERATION_NAME, DB_QUERY_TEXT, DB_SYSTEM_NAME, ) @@ -188,6 +194,134 @@ def test_stability_mode_dup_precedence(self): ) +class TestOpenTelemetrySemConvSchemaUrl(TestCase): + @stability_mode("") + def test_get_schema_version_for_opt_in_mode_default(self): + version = _get_schema_version_for_opt_in_mode( + _OpenTelemetryStabilitySignalType.HTTP, _StabilityMode.DEFAULT + ) + self.assertEqual(version, _LEGACY_SCHEMA_VERSION) + + version = _get_schema_version_for_opt_in_mode( + _OpenTelemetryStabilitySignalType.DATABASE, _StabilityMode.DEFAULT + ) + self.assertEqual(version, _LEGACY_SCHEMA_VERSION) + + version = _get_schema_version_for_opt_in_mode( + _OpenTelemetryStabilitySignalType.GEN_AI, _StabilityMode.DEFAULT + ) + self.assertEqual(version, _LEGACY_SCHEMA_VERSION) + + @stability_mode("") + def test_get_schema_version_for_opt_in_mode_http_stable(self): + version = _get_schema_version_for_opt_in_mode( + _OpenTelemetryStabilitySignalType.HTTP, _StabilityMode.HTTP + ) + self.assertEqual(version, "1.21.0") + + @stability_mode("") + def test_get_schema_version_for_opt_in_mode_database_stable(self): + version = _get_schema_version_for_opt_in_mode( + _OpenTelemetryStabilitySignalType.DATABASE, _StabilityMode.DATABASE + ) + self.assertEqual(version, "1.25.0") + + @stability_mode("") + def test_get_schema_version_for_opt_in_mode_gen_ai_stable(self): + version = _get_schema_version_for_opt_in_mode( + _OpenTelemetryStabilitySignalType.GEN_AI, + _StabilityMode.GEN_AI_LATEST_EXPERIMENTAL, + ) + self.assertEqual(version, "1.26.0") + + @stability_mode("") + def test_get_schema_url_for_signal_types_single_http_default(self): + url = _get_schema_url_for_signal_types( + [_OpenTelemetryStabilitySignalType.HTTP] + ) + self.assertEqual( + url, f"https://opentelemetry.io/schemas/{_LEGACY_SCHEMA_VERSION}" + ) + + @stability_mode("http") + def test_get_schema_url_for_signal_types_single_http_stable(self): + url = _get_schema_url_for_signal_types( + [_OpenTelemetryStabilitySignalType.HTTP] + ) + self.assertEqual(url, "https://opentelemetry.io/schemas/1.21.0") + + @stability_mode("database") + def test_get_schema_url_for_signal_types_single_database_stable(self): + url = _get_schema_url_for_signal_types( + [_OpenTelemetryStabilitySignalType.DATABASE] + ) + self.assertEqual(url, "https://opentelemetry.io/schemas/1.25.0") + + @stability_mode("http,database") + def test_get_schema_url_for_signal_types_multiple_both_stable(self): + # DATABASE has higher version (1.25.0) than HTTP (1.21.0) + url = _get_schema_url_for_signal_types( + [ + _OpenTelemetryStabilitySignalType.HTTP, + _OpenTelemetryStabilitySignalType.DATABASE, + ] + ) + self.assertEqual(url, "https://opentelemetry.io/schemas/1.25.0") + + @stability_mode("http") + def test_get_schema_url_for_signal_types_mixed_modes(self): + # HTTP is stable (1.21.0), DATABASE is default (1.11.0) + # Should return HTTP version as it's higher + url = _get_schema_url_for_signal_types( + [ + _OpenTelemetryStabilitySignalType.HTTP, + _OpenTelemetryStabilitySignalType.DATABASE, + ] + ) + self.assertEqual(url, "https://opentelemetry.io/schemas/1.21.0") + + @stability_mode("database") + def test_get_schema_url_for_signal_types_database_only_stable(self): + # DATABASE is stable (1.25.0), HTTP is default (1.11.0) + # Should return DATABASE version as it's highest + url = _get_schema_url_for_signal_types( + [ + _OpenTelemetryStabilitySignalType.HTTP, + _OpenTelemetryStabilitySignalType.DATABASE, + ] + ) + self.assertEqual(url, "https://opentelemetry.io/schemas/1.25.0") + + @stability_mode("") + def test_get_schema_url_for_signal_types_empty_list(self): + url = _get_schema_url_for_signal_types([]) + self.assertEqual( + url, f"https://opentelemetry.io/schemas/{_LEGACY_SCHEMA_VERSION}" + ) + + @stability_mode("http/dup,database/dup") + def test_get_schema_url_for_signal_types_dup_modes(self): + url = _get_schema_url_for_signal_types( + [ + _OpenTelemetryStabilitySignalType.HTTP, + _OpenTelemetryStabilitySignalType.DATABASE, + ] + ) + self.assertEqual(url, "https://opentelemetry.io/schemas/1.25.0") + + @stability_mode("http,database,gen_ai_latest_experimental") + def test_get_schema_url_for_signal_types_with_gen_ai(self): + # GEN_AI should be highest at 1.26.0 + url = _get_schema_url_for_signal_types( + [ + _OpenTelemetryStabilitySignalType.HTTP, + _OpenTelemetryStabilitySignalType.DATABASE, + _OpenTelemetryStabilitySignalType.GEN_AI, + ] + ) + self.assertEqual(url, "https://opentelemetry.io/schemas/1.26.0") + + class TestOpenTelemetrySemConvStabilityHTTP(TestCase): def test_set_status_for_non_http_code_with_recording_span(self): span = Mock() @@ -443,3 +577,45 @@ def test_db_user_none_value(self): result = {} _set_db_user(result, None, sem_conv_opt_in_mode=_StabilityMode.DEFAULT) self.assertNotIn(DB_USER, result) + + def test_db_operation_default(self): + result = {} + _set_db_operation( + result, + "SELECT", + sem_conv_opt_in_mode=_StabilityMode.DEFAULT, + ) + self.assertIn(DB_OPERATION, result) + self.assertEqual(result[DB_OPERATION], "SELECT") + self.assertNotIn(DB_OPERATION_NAME, result) + + def test_db_operation_database_stable(self): + result = {} + _set_db_operation( + result, + "SELECT", + sem_conv_opt_in_mode=_StabilityMode.DATABASE, + ) + self.assertNotIn(DB_OPERATION, result) + self.assertIn(DB_OPERATION_NAME, result) + self.assertEqual(result[DB_OPERATION_NAME], "SELECT") + + def test_db_operation_database_dup(self): + result = {} + _set_db_operation( + result, + "SELECT", + sem_conv_opt_in_mode=_StabilityMode.DATABASE_DUP, + ) + self.assertIn(DB_OPERATION, result) + self.assertEqual(result[DB_OPERATION], "SELECT") + self.assertIn(DB_OPERATION_NAME, result) + self.assertEqual(result[DB_OPERATION_NAME], "SELECT") + + def test_db_operation_none_value(self): + result = {} + _set_db_operation( + result, None, sem_conv_opt_in_mode=_StabilityMode.DEFAULT + ) + self.assertNotIn(DB_OPERATION, result) + self.assertNotIn(DB_OPERATION_NAME, result)