pymssql: add capture_parameters option

.changelog/4792.added

@@ -0,0 +1 @@
+`opentelemetry-instrumentation-pymssql`: add `capture_parameters` option to capture query parameters as the `db.statement.parameters` span attribute

.changelog/5362.added

@@ -0,0 +1 @@
+`opentelemetry-instrumentation-dbapi`: capture prepared statement parameters as `db.query.parameter.<key>` when the database semantic conventions are opted in and `capture_parameters` is enabled. In the new semconv the legacy `db.statement.parameters` attribute is replaced by one `db.query.parameter.<key>` attribute per parameter, and parameters are not captured on batch operations

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

@@ -184,7 +184,9 @@
     _OpenTelemetrySemanticConventionStability,
     _OpenTelemetryStabilitySignalType,
     _report_new,
+    _report_old,
     _set_db_name,
+    _set_db_query_parameters,
     _set_db_statement,
     _set_db_system,
     _set_db_user,
@@ -789,6 +791,7 @@ def _populate_span(
         span: trace_api.Span,
         cursor: CursorT,
         *args: tuple[Any, ...],
+        is_batch: bool = False,
     ):
         if not span.is_recording():
             return
@@ -818,7 +821,17 @@ def _populate_span(
             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]))
+            parameters = args[1]
+            if _report_old(sem_conv_mode):
+                span.set_attribute("db.statement.parameters", str(parameters))
+            # db.query.parameter.<key> SHOULD NOT be captured on batch
+            # operations (e.g. executemany).
+            if not is_batch:
+                query_parameter_attributes = {}
+                _set_db_query_parameters(
+                    query_parameter_attributes, parameters, sem_conv_mode
+                )
+                span.set_attributes(query_parameter_attributes)
 
     def get_operation_name(self, cursor: CursorT, args: Sequence[Any]) -> str:  # pylint: disable=no-self-use
         if not args:
@@ -904,6 +917,9 @@ def traced_execution(
             return query_method(*args, **kwargs)
 
         operation_name = self.get_operation_name(cursor, args)
+        # Query parameters must not be captured for batch operations, which are
+        # executed through the cursor's executemany method.
+        is_batch = getattr(query_method, "__name__", None) == "executemany"
         name = operation_name
         if not name:
             name = (
@@ -922,17 +938,21 @@ def traced_execution(
                         args = self._update_args_with_added_sql_comment(
                             args, cursor
                         )
-                        self._populate_span(span, cursor, *args)
+                        self._populate_span(
+                            span, cursor, *args, is_batch=is_batch
+                        )
                     else:
                         # sqlcomment is only added to executed query
                         # so db.statement and/or db.query.text are set before add_sql_comment
-                        self._populate_span(span, cursor, *args)
+                        self._populate_span(
+                            span, cursor, *args, is_batch=is_batch
+                        )
                         args = self._update_args_with_added_sql_comment(
                             args, cursor
                         )
                 else:
                     # no sqlcomment anywhere
-                    self._populate_span(span, cursor, *args)
+                    self._populate_span(span, cursor, *args, is_batch=is_batch)
             start_time = time.perf_counter()
             error: Exception | None = None
             try:
@@ -954,6 +974,9 @@ async def traced_execution_async(
             return await query_method(*args, **kwargs)
 
         operation_name = self.get_operation_name(cursor, args)
+        # Query parameters must not be captured for batch operations, which are
+        # executed through the cursor's executemany method.
+        is_batch = getattr(query_method, "__name__", None) == "executemany"
         name = operation_name
         if not name:
             name = (
@@ -972,17 +995,21 @@ async def traced_execution_async(
                         args = self._update_args_with_added_sql_comment(
                             args, cursor
                         )
-                        self._populate_span(span, cursor, *args)
+                        self._populate_span(
+                            span, cursor, *args, is_batch=is_batch
+                        )
                     else:
                         # sqlcomment is only added to executed query
                         # so db.statement and/or db.query.text are set before add_sql_comment
-                        self._populate_span(span, cursor, *args)
+                        self._populate_span(
+                            span, cursor, *args, is_batch=is_batch
+                        )
                         args = self._update_args_with_added_sql_comment(
                             args, cursor
                         )
                 else:
                     # no sqlcomment anywhere
-                    self._populate_span(span, cursor, *args)
+                    self._populate_span(span, cursor, *args, is_batch=is_batch)
             start_time = time.perf_counter()
             error: Exception | None = None
             try:

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

@@ -24,6 +24,7 @@
 from opentelemetry.semconv._incubating.attributes import net_attributes
 from opentelemetry.semconv._incubating.attributes.db_attributes import (
     DB_NAME,
+    DB_QUERY_PARAMETER_TEMPLATE,
     DB_STATEMENT,
     DB_SYSTEM,
     DB_USER,
@@ -354,6 +355,13 @@ def test_span_succeeded_with_capture_of_statement_parameters(self):
             span.attributes["db.statement.parameters"],
             "('param1Value', False)",
         )
+        # db.query.parameter.<key> belongs to the new semconv only.
+        self.assertFalse(
+            any(
+                key.startswith(DB_QUERY_PARAMETER_TEMPLATE)
+                for key in span.attributes
+            )
+        )
         self.assertEqual(span.attributes[DB_USER], "testuser")
         self.assertEqual(
             span.attributes[net_attributes.NET_PEER_NAME], "testhost"
@@ -388,10 +396,18 @@ def test_span_succeeded_with_capture_of_statement_parameters_new_semconv(
             self.assertEqual(span.attributes[DB_SYSTEM_NAME], "testcomponent")
             self.assertEqual(span.attributes[DB_NAMESPACE], "testdatabase")
             self.assertEqual(span.attributes[DB_QUERY_TEXT], "Test query")
+            # Positional parameters are keyed by their 0-based index and their
+            # values are captured as strings.
             self.assertEqual(
-                span.attributes["db.statement.parameters"],
-                "('param1Value', False)",
+                span.attributes[f"{DB_QUERY_PARAMETER_TEMPLATE}.0"],
+                "param1Value",
+            )
+            self.assertEqual(
+                span.attributes[f"{DB_QUERY_PARAMETER_TEMPLATE}.1"], "False"
             )
+            # The legacy db.statement.parameters is replaced by
+            # db.query.parameter.<key> in the stable semconv.
+            self.assertFalse("db.statement.parameters" in span.attributes)
             # db.user removed in stable - no replacement
             self.assertFalse(DB_USER in span.attributes)
             self.assertEqual(span.attributes[SERVER_ADDRESS], "testhost")
@@ -450,11 +466,79 @@ def test_span_succeeded_with_capture_of_statement_parameters_both_semconv(
             self.assertEqual(span.attributes[DB_SYSTEM_NAME], "testcomponent")
             self.assertEqual(span.attributes[DB_NAMESPACE], "testdatabase")
             self.assertEqual(span.attributes[DB_QUERY_TEXT], "Test query")
+            self.assertEqual(
+                span.attributes[f"{DB_QUERY_PARAMETER_TEMPLATE}.0"],
+                "param1Value",
+            )
+            self.assertEqual(
+                span.attributes[f"{DB_QUERY_PARAMETER_TEMPLATE}.1"], "False"
+            )
             self.assertEqual(span.attributes[SERVER_ADDRESS], "testhost")
             self.assertEqual(span.attributes[SERVER_PORT], 123)
 
             self.assertIs(span.status.status_code, trace_api.StatusCode.UNSET)
 
+    def test_span_succeeded_with_capture_of_named_query_parameters(self):
+        with use_semconv_opt_in("database"):
+            connection_props = _get_default_connection_props()
+            connection_attributes = _get_default_connection_attributes()
+            db_integration = dbapi.DatabaseApiIntegration(
+                "instrumenting_module_test_name",
+                "testcomponent",
+                connection_attributes,
+                capture_parameters=True,
+            )
+            mock_connection = db_integration.wrapped_connection(
+                mock_connect, {}, connection_props
+            )
+            cursor = mock_connection.cursor()
+            cursor.execute(
+                "SELECT * FROM users WHERE name = %(userName)s",
+                {"userName": "jdoe"},
+            )
+            spans_list = self.memory_exporter.get_finished_spans()
+            self.assertEqual(len(spans_list), 1)
+            span = spans_list[0]
+            # Named parameters are keyed by their name.
+            self.assertEqual(
+                span.attributes[f"{DB_QUERY_PARAMETER_TEMPLATE}.userName"],
+                "jdoe",
+            )
+
+    def test_query_parameters_not_captured_for_batch_operations(self):
+        with use_semconv_opt_in("database/dup"):
+            connection_props = _get_default_connection_props()
+            connection_attributes = _get_default_connection_attributes()
+            db_integration = dbapi.DatabaseApiIntegration(
+                "instrumenting_module_test_name",
+                "testcomponent",
+                connection_attributes,
+                capture_parameters=True,
+            )
+            mock_connection = db_integration.wrapped_connection(
+                mock_connect, {}, connection_props
+            )
+            cursor = mock_connection.cursor()
+            cursor.executemany(
+                "INSERT INTO users VALUES (%s)",
+                [("param1Value",), ("param2Value",)],
+            )
+            spans_list = self.memory_exporter.get_finished_spans()
+            self.assertEqual(len(spans_list), 1)
+            span = spans_list[0]
+            # db.query.parameter.<key> SHOULD NOT be captured on batch
+            # operations, but the legacy attribute is still captured.
+            self.assertFalse(
+                any(
+                    key.startswith(DB_QUERY_PARAMETER_TEMPLATE)
+                    for key in span.attributes
+                )
+            )
+            self.assertEqual(
+                span.attributes["db.statement.parameters"],
+                "[('param1Value',), ('param2Value',)]",
+            )
+
     def test_span_not_recording(self):
         connection_props = _get_default_connection_props()
         connection_attributes = _get_default_connection_attributes()

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

@@ -45,6 +45,9 @@
 The `instrument` method accepts the following keyword args:
 
 * tracer_provider (``TracerProvider``) - an optional tracer provider
+* capture_parameters (``bool``) - an optional flag to enable capturing of
+  query parameters as the ``db.statement.parameters`` span attribute
+  (default ``False``)
 
 For example:
 
@@ -157,6 +160,7 @@ def _instrument(self, **kwargs):
         https://github.com/pymssql/pymssql/
         """
         tracer_provider = kwargs.get("tracer_provider")
+        capture_parameters = kwargs.get("capture_parameters", False)
 
         dbapi.wrap_connect(
             __name__,
@@ -165,6 +169,7 @@ def _instrument(self, **kwargs):
             _DATABASE_SYSTEM,
             version=__version__,
             tracer_provider=tracer_provider,
+            capture_parameters=capture_parameters,
             # pymssql does not keep the connection attributes in its connection object;
             # instead, we get the attributes from the connect method (which is done
             # via PyMSSQLDatabaseApiIntegration.wrapped_connection)
@@ -176,13 +181,20 @@ def _uninstrument(self, **kwargs):
         dbapi.unwrap_connect(pymssql, "connect")
 
     @staticmethod
-    def instrument_connection(connection, tracer_provider=None):
+    def instrument_connection(
+        connection,
+        tracer_provider=None,
+        capture_parameters: bool = 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.
+            capture_parameters: Optional flag to enable capturing of query
+                parameters as the ``db.statement.parameters`` span attribute
+                (default False).
 
         Returns:
             An instrumented connection.
@@ -194,6 +206,7 @@ def instrument_connection(connection, tracer_provider=None):
             _DATABASE_SYSTEM,
             version=__version__,
             tracer_provider=tracer_provider,
+            capture_parameters=capture_parameters,
             db_api_integration_factory=_PyMSSQLDatabaseApiIntegration,
         )
 

instrumentation/opentelemetry-instrumentation-pymssql/tests/test_pymssql_integration.py

@@ -271,3 +271,35 @@ def test_semconv_dup(self):
             )
             self.assertEqual(span.attributes["net.peer.port"], 1433)
             self.assertEqual(span.attributes["server.port"], 1433)
+
+    @patch("opentelemetry.instrumentation.pymssql.dbapi.wrap_connect")
+    def test_instrument_capture_parameters_default(self, mock_wrap_connect):
+        PyMSSQLInstrumentor()._instrument()
+        _, kwargs = mock_wrap_connect.call_args
+        self.assertFalse(kwargs["capture_parameters"])
+
+    @patch("opentelemetry.instrumentation.pymssql.dbapi.wrap_connect")
+    def test_instrument_capture_parameters_enabled(self, mock_wrap_connect):
+        PyMSSQLInstrumentor()._instrument(capture_parameters=True)
+        _, kwargs = mock_wrap_connect.call_args
+        self.assertTrue(kwargs["capture_parameters"])
+
+    @patch("opentelemetry.instrumentation.pymssql.dbapi.instrument_connection")
+    def test_instrument_connection_capture_parameters_default(
+        self, mock_instrument_connection
+    ):
+        connection = Mock()
+        PyMSSQLInstrumentor().instrument_connection(connection)
+        _, kwargs = mock_instrument_connection.call_args
+        self.assertFalse(kwargs["capture_parameters"])
+
+    @patch("opentelemetry.instrumentation.pymssql.dbapi.instrument_connection")
+    def test_instrument_connection_capture_parameters_enabled(
+        self, mock_instrument_connection
+    ):
+        connection = Mock()
+        PyMSSQLInstrumentor().instrument_connection(
+            connection, capture_parameters=True
+        )
+        _, kwargs = mock_instrument_connection.call_args
+        self.assertTrue(kwargs["capture_parameters"])

opentelemetry-instrumentation/src/opentelemetry/instrumentation/_semconv.py

@@ -5,8 +5,9 @@
 
 import os
 import threading
+from collections.abc import Container, Mapping, MutableMapping, Sequence
 from enum import Enum
-from typing import Container, Mapping, MutableMapping
+from typing import Any
 from urllib.parse import urlparse
 
 from packaging import version as package_version
@@ -15,6 +16,7 @@
 from opentelemetry.semconv._incubating.attributes.db_attributes import (
     DB_NAME,
     DB_OPERATION,
+    DB_QUERY_PARAMETER_TEMPLATE,
     DB_REDIS_DATABASE_INDEX,
     DB_STATEMENT,
     DB_SYSTEM,
@@ -599,6 +601,32 @@ def _set_db_statement(
         result[DB_QUERY_TEXT] = statement
 
 
+def _set_db_query_parameters(
+    result: MutableMapping[str, AttributeValue],
+    parameters: Sequence[Any] | Mapping[str, Any] | None,
+    sem_conv_opt_in_mode: _StabilityMode,
+) -> None:
+    # db.query.parameter.<key> is only defined in the new (stable) database
+    # semantic conventions.
+    if not _report_new(sem_conv_opt_in_mode) or not parameters:
+        return
+    # Named parameters are keyed by their name; positional parameters use their
+    # 0-based index, both matching the placeholders in db.query.text. A string
+    # or bytes value is a single scalar parameter, not a sequence of parameters.
+    if isinstance(parameters, Mapping):
+        items = parameters.items()
+    elif isinstance(parameters, (str, bytes, bytearray)):
+        return
+    elif isinstance(parameters, Sequence):
+        items = enumerate(parameters)
+    else:
+        return
+    for key, value in items:
+        # Assign directly instead of using set_string_attribute so that falsy
+        # values (empty string, 0, False, None) are still captured.
+        result[f"{DB_QUERY_PARAMETER_TEMPLATE}.{key}"] = str(value)
+
+
 def _set_db_user(
     result: MutableMapping[str, AttributeValue],
     user: str,