feat: add dbm dynamic_service mode

ddtrace/internal/settings/_database_monitoring.py

@@ -10,8 +10,8 @@ class DatabaseMonitoringConfig(DDConfig):
         str,
         "propagation_mode",
         default="disabled",
-        help="Valid Injection Modes: disabled, service, and full",
-        validator=validators.choice(["disabled", "full", "service"]),
+        help="Valid Injection Modes: disabled, service, dynamic_service, and full",
+        validator=validators.choice(["disabled", "full", "service", "dynamic_service"]),
     )
 
     inject_sql_basehash = DDConfig.v(

ddtrace/propagation/_database_monitoring.py

@@ -33,10 +33,19 @@
 DBM_SERVICE_HASH: Literal["ddsh"] = "ddsh"
 DBM_TRACE_PARENT_KEY: Literal["traceparent"] = "traceparent"
 DBM_TRACE_INJECTED_TAG: Literal["_dd.dbm_trace_injected"] = "_dd.dbm_trace_injected"
+DBM_PROPAGATION_MODE_DYNAMIC_SERVICE: Literal["dynamic_service"] = "dynamic_service"
+_DBM_INJECTION_MODES = ("full", "service", DBM_PROPAGATION_MODE_DYNAMIC_SERVICE)
 
 log = get_logger(__name__)
 
 
+def _should_inject_sql_basehash():
+    # type: () -> bool
+    return dbm_config.propagation_mode == DBM_PROPAGATION_MODE_DYNAMIC_SERVICE or (
+        dbm_config.propagation_mode == "service" and dbm_config.inject_sql_basehash
+    )
+
+
 def default_sql_injector(dbm_comment, sql_statement):
     # type: (str, Union[str, bytes]) -> Union[str, bytes]
     try:
@@ -83,7 +92,7 @@ def inject(self, dbspan, args, kwargs):
             return args, kwargs
 
         # the base hash is injected in the comment and on the span tags for correlation purpose
-        if dbm_config.inject_sql_basehash and (base_hash := process_tags.base_hash):
+        if _should_inject_sql_basehash() and (base_hash := process_tags.base_hash):
             dbspan._set_attribute(PROPAGATED_HASH, str(base_hash))
 
         original_sql_statement = get_argument_value(args, kwargs, self.sql_pos, self.sql_kw)
@@ -102,7 +111,7 @@ def _get_dbm_comment(self, db_span):
         if dbm_config.propagation_mode == "disabled":
             return None
 
-        # set the following tags if DBM injection mode is full or service
+        # set the following tags if DBM injection mode is full, service, or dynamic_service
         peer_service_enabled = PeerServiceConfig().set_defaults_enabled
         service_name_key = db_span.service
         if peer_service_enabled:
@@ -132,7 +141,7 @@ def _get_dbm_comment(self, db_span):
             db_span._set_attribute(DBM_TRACE_INJECTED_TAG, "true")
             dbm_tags[DBM_TRACE_PARENT_KEY] = db_span.context._traceparent
 
-        if dbm_config.inject_sql_basehash and (base_hash := process_tags.base_hash):
+        if _should_inject_sql_basehash() and (base_hash := process_tags.base_hash):
             dbm_tags[DBM_SERVICE_HASH] = str(base_hash)
 
         sql_comment = self.comment_generator(**dbm_tags)
@@ -170,7 +179,7 @@ def handle_dbm_injection_asyncpg(int_config, method, span, args, kwargs):
 
 
 def listen():
-    if dbm_config.propagation_mode in ["full", "service"]:
+    if dbm_config.propagation_mode in _DBM_INJECTION_MODES:
         for event in _DBM_STANDARD_EVENTS:
             core.on(event, handle_dbm_injection, "result")
         core.on("asyncpg.execute", handle_dbm_injection_asyncpg, "result")

releasenotes/notes/add-dbm-dynamic-service-propagation-mode-5f0c2a8b7d1e.yaml

@@ -0,0 +1,7 @@
+---
+features:
+  - |
+    Database Monitoring (DBM) propagation supports ``dynamic_service`` as a new
+    ``DD_DBM_PROPAGATION_MODE`` value. Set
+    ``DD_DBM_PROPAGATION_MODE=dynamic_service`` to inject DBM service metadata and
+    the SQL base hash without injecting trace context.

tests/internal/test_database_monitoring.py

@@ -22,13 +22,18 @@ def test_propagation_mode_configuration():
         config = _database_monitoring.DatabaseMonitoringConfig()
         assert config.propagation_mode == "full"
 
+    # Ensure dynamic_service is a valid injection mode
+    with override_env(dict(DD_DBM_PROPAGATION_MODE="dynamic_service")):
+        config = _database_monitoring.DatabaseMonitoringConfig()
+        assert config.propagation_mode == "dynamic_service"
+
     # Ensure an invalid injection mode raises a ValueError
     with override_env(dict(DD_DBM_PROPAGATION_MODE="notaninjectionmode")):
         with pytest.raises(ValueError) as excinfo:
             _database_monitoring.DatabaseMonitoringConfig()
     assert (
         excinfo.value.args[0] == "Invalid value for environment variable DD_DBM_PROPAGATION_MODE: "
-        "value must be one of ['disabled', 'full', 'service']"
+        "value must be one of ['disabled', 'dynamic_service', 'full', 'service']"
     )
 
 
@@ -192,6 +197,76 @@ def test_dbm_propagating_base_hash_when_activated():
         assert ddsh_value == dbspan._get_str_attribute(PROPAGATED_HASH)
 
 
+@pytest.mark.subprocess(
+    env=dict(
+        DD_DBM_PROPAGATION_MODE="full",
+        DD_DBM_INJECT_SQL_BASEHASH="True",
+        DD_SERVICE="orders-app",
+        DD_ENV="staging",
+        DD_VERSION="v7343437-d7ac743",
+    )
+)
+def test_dbm_not_propagating_base_hash_in_full_mode():
+    from ddtrace.internal import process_tags
+    from ddtrace.internal.constants import PROPAGATED_HASH
+    from ddtrace.propagation import _database_monitoring
+    from ddtrace.trace import tracer
+
+    process_tags.compute_base_hash("abc123")
+
+    with tracer.trace("dbspan", service="orders-db") as dbspan:
+        dbm_propagator = _database_monitoring._DBM_Propagator(0, "query")
+
+        original_sql = "SELECT * FROM users"
+        modified_args, _ = dbm_propagator.inject(dbspan, (original_sql,), {})
+        injected_sql = modified_args[0]
+
+        assert "traceparent" in injected_sql
+        assert "ddsh" not in injected_sql
+        assert not dbspan._has_attribute(PROPAGATED_HASH)
+
+
+@pytest.mark.subprocess(
+    env=dict(
+        DD_DBM_PROPAGATION_MODE="dynamic_service",
+        DD_DBM_INJECT_SQL_BASEHASH="False",
+        DD_SERVICE="orders-app",
+        DD_ENV="staging",
+        DD_VERSION="v7343437-d7ac743",
+    )
+)
+def test_dbm_propagation_dynamic_service_mode():
+    import re
+
+    from ddtrace.internal import process_tags
+    from ddtrace.internal.constants import PROPAGATED_HASH
+    from ddtrace.propagation import _database_monitoring
+    from ddtrace.trace import tracer
+
+    process_tags.compute_base_hash("abc123")
+
+    with tracer.trace("dbspan", service="orders-db") as dbspan:
+        dbm_propagator = _database_monitoring._DBM_Propagator(0, "query")
+
+        original_sql = "SELECT * FROM users"
+        modified_args, modified_kwargs = dbm_propagator.inject(dbspan, (original_sql,), {})
+        injected_sql = modified_args[0]
+
+        assert modified_kwargs == {}
+        assert "dddbs='orders-db'" in injected_sql
+        assert "dde='staging'" in injected_sql
+        assert "ddps='orders-app'" in injected_sql
+        assert "ddpv='v7343437-d7ac743'" in injected_sql
+        assert "traceparent" not in injected_sql
+        assert dbspan.get_tag(_database_monitoring.DBM_TRACE_INJECTED_TAG) is None
+
+        match = re.search(r"ddsh='(\d+)'", injected_sql)
+        assert match is not None
+        assert dbspan._has_attribute(PROPAGATED_HASH)
+        assert dbspan._get_str_attribute(PROPAGATED_HASH) == str(process_tags.base_hash)
+        assert match.group(1) == dbspan._get_str_attribute(PROPAGATED_HASH)
+
+
 @pytest.mark.subprocess(
     env=dict(
         DD_DBM_PROPAGATION_MODE="service",