chore(tracer): report changes to config values (#7909)

A feature is being built to report configuration in the Datadog UI based
on the telemetry data. In order to provide the most up to date
configuration data in the UI we need to report configuration change
events when settings are updated. The feature also supports reporting
the origin of a setting at the granularity of ("code", "env_var",
"remote_config" or "default") which will help customers debug where a
setting for their service might be coming from.

I noticed that the telemetry config changed code doesn't do any batching
of configuration updates so this would be a potential performance
improvement in the future. There is the risk that if a customer is
frequently configuring a setting that many telemetry events will be
generated which could lead to performance regressions. I imagine this
case is small since up until
#7489 global config updates
to sample rate would not have any effect.

Tests are added to cover the different origins.


![image](https://github.com/DataDog/dd-trace-py/assets/6321485/8cbfbb1e-c2ad-4c1e-984b-2170822d2264)


The implementation should be consistent with
DataDog/system-tests#1734.

Author: Kyle-Verhoog

ddtrace/internal/telemetry/constants.py

@@ -22,7 +22,6 @@
 TELEMETRY_ANALYTICS_ENABLED = "DD_TRACE_ANALYTICS_ENABLED"
 TELEMETRY_STARTUP_LOGS_ENABLED = "DD_TRACE_STARTUP_LOGS"
 TELEMETRY_CLIENT_IP_ENABLED = "DD_TRACE_CLIENT_IP_ENABLED"
-TELEMETRY_LOGS_INJECTION_ENABLED = "DD_LOGS_INJECTION"
 TELEMETRY_128_BIT_TRACEID_GENERATION_ENABLED = "DD_TRACE_128_BIT_TRACEID_GENERATION_ENABLED"
 TELEMETRY_128_BIT_TRACEID_LOGGING_ENABLED = "DD_TRACE_128_BIT_TRACEID_LOGGING_ENABLED"
 TELEMETRY_TRACE_COMPUTE_STATS = "DD_TRACE_COMPUTE_STATS"
@@ -38,7 +37,6 @@
 TELEMETRY_SPAN_SAMPLING_RULES_FILE = "DD_SPAN_SAMPLING_RULES_FILE"
 TELEMETRY_PROPAGATION_STYLE_INJECT = "DD_TRACE_PROPAGATION_STYLE_INJECT"
 TELEMETRY_PROPAGATION_STYLE_EXTRACT = "DD_TRACE_PROPAGATION_STYLE_EXTRACT"
-TELEMETRY_TRACE_SAMPLING_RATE = "DD_TRACE_SAMPLE_RATE"
 TELEMETRY_TRACE_SAMPLING_RULES = "DD_TRACE_SAMPLING_RULES"
 TELEMETRY_TRACE_SAMPLING_LIMIT = "DD_TRACE_RATE_LIMIT"
 TELEMETRY_PRIORITY_SAMPLING = "DD_PRIORITY_SAMPLING"

ddtrace/internal/telemetry/writer.py

@@ -19,6 +19,7 @@
 from ...internal.schema import _remove_client_service_names
 from ...settings import _config as config
 from ...settings.asm import config as asm_config
+from ...settings.config import _ConfigSource
 from ...settings.dynamic_instrumentation import config as di_config
 from ...settings.exception_debugging import config as ed_config
 from ...settings.peer_service import _ps_config
@@ -50,7 +51,6 @@
 from .constants import TELEMETRY_DYNAMIC_INSTRUMENTATION_ENABLED
 from .constants import TELEMETRY_ENABLED
 from .constants import TELEMETRY_EXCEPTION_DEBUGGING_ENABLED
-from .constants import TELEMETRY_LOGS_INJECTION_ENABLED
 from .constants import TELEMETRY_OBFUSCATION_QUERY_STRING_PATTERN
 from .constants import TELEMETRY_OTEL_ENABLED
 from .constants import TELEMETRY_PARTIAL_FLUSH_ENABLED
@@ -75,7 +75,6 @@
 from .constants import TELEMETRY_TRACE_PEER_SERVICE_MAPPING
 from .constants import TELEMETRY_TRACE_REMOVE_INTEGRATION_SERVICE_NAMES_ENABLED
 from .constants import TELEMETRY_TRACE_SAMPLING_LIMIT
-from .constants import TELEMETRY_TRACE_SAMPLING_RATE
 from .constants import TELEMETRY_TRACE_SAMPLING_RULES
 from .constants import TELEMETRY_TRACE_SPAN_ATTRIBUTE_SCHEMA
 from .constants import TELEMETRY_TRACE_WRITER_BUFFER_SIZE_BYTES
@@ -294,6 +293,25 @@ def add_error(self, code, msg, filename, line_number):
             msg = "%s:%s: %s" % (filename, line_number, msg)
         self._error = (code, msg)
 
+    def add_configs_changed(self, cfg_names):
+        cs = [{"name": n, "value": v, "origin": o} for n, v, o in [self._telemetry_entry(n) for n in cfg_names]]
+        self._app_client_configuration_changed_event(cs)
+
+    def _telemetry_entry(self, cfg_name: str) -> Tuple[str, str, _ConfigSource]:
+        item = config._config[cfg_name]
+        if cfg_name == "_trace_sample_rate":
+            name = "trace_sample_rate"
+            value = str(item.value())
+        elif cfg_name == "logs_injection":
+            name = "logs_injection_enabled"
+            value = "true" if item.value() else "false"
+        elif cfg_name == "trace_http_header_tags":
+            name = "trace_header_tags"
+            value = ",".join(":".join(x) for x in item.value().items())
+        else:
+            raise ValueError("Unknown configuration item: %s" % cfg_name)
+        return name, value, item.source()
+
     def _app_started_event(self, register_app_shutdown=True):
         # type: (bool) -> None
         """Sent when TelemetryWriter is enabled or forks"""
@@ -308,6 +326,9 @@ def _app_started_event(self, register_app_shutdown=True):
 
         self.add_configurations(
             [
+                self._telemetry_entry("_trace_sample_rate"),
+                self._telemetry_entry("logs_injection"),
+                self._telemetry_entry("trace_http_header_tags"),
                 (TELEMETRY_TRACING_ENABLED, config._tracing_enabled, "unknown"),
                 (TELEMETRY_STARTUP_LOGS_ENABLED, config._startup_logs_enabled, "unknown"),
                 (TELEMETRY_DSM_ENABLED, config._data_streams_enabled, "unknown"),
@@ -324,7 +345,6 @@ def _app_started_event(self, register_app_shutdown=True):
                 (TELEMETRY_ENABLED, config._telemetry_enabled, "unknown"),
                 (TELEMETRY_ANALYTICS_ENABLED, config.analytics_enabled, "unknown"),
                 (TELEMETRY_CLIENT_IP_ENABLED, config.client_ip_header, "unknown"),
-                (TELEMETRY_LOGS_INJECTION_ENABLED, config.logs_injection, "unknown"),
                 (TELEMETRY_128_BIT_TRACEID_GENERATION_ENABLED, config._128_bit_trace_id_enabled, "unknown"),
                 (TELEMETRY_128_BIT_TRACEID_LOGGING_ENABLED, config._128_bit_trace_id_logging_enabled, "unknown"),
                 (TELEMETRY_TRACE_COMPUTE_STATS, config._trace_compute_stats, "unknown"),
@@ -340,7 +360,6 @@ def _app_started_event(self, register_app_shutdown=True):
                 (TELEMETRY_RUNTIMEMETRICS_ENABLED, config._runtime_metrics_enabled, "unknown"),
                 (TELEMETRY_REMOTE_CONFIGURATION_ENABLED, config._remote_config_enabled, "unknown"),
                 (TELEMETRY_REMOTE_CONFIGURATION_INTERVAL, config._remote_config_poll_interval, "unknown"),
-                (TELEMETRY_TRACE_SAMPLING_RATE, config._trace_sample_rate, "unknown"),
                 (TELEMETRY_TRACE_SAMPLING_LIMIT, config._trace_rate_limit, "unknown"),
                 (TELEMETRY_SPAN_SAMPLING_RULES, config._sampling_rules, "unknown"),
                 (TELEMETRY_SPAN_SAMPLING_RULES_FILE, config._sampling_rules_file, "unknown"),

ddtrace/settings/config.py

@@ -193,7 +193,7 @@ def get_error_ranges(error_range_str):
     return error_ranges  # type: ignore[return-value]
 
 
-_ConfigSource = Literal["default", "env", "code", "remote_config"]
+_ConfigSource = Literal["default", "env_var", "code", "remote_config"]
 _JSONType = Union[None, int, float, str, bool, List["_JSONType"], Dict[str, "_JSONType"]]
 
 
@@ -247,7 +247,7 @@ def source(self):
         if self._code_value is not None:
             return "code"
         if self._env_value is not None:
-            return "env"
+            return "env_var"
         return "default"
 
     def __repr__(self):
@@ -653,14 +653,15 @@ def __setattr__(self, key, value):
 
     def _set_config_items(self, items):
         # type: (List[Tuple[str, Any, _ConfigSource]]) -> None
+        item_names = []
         for key, value, origin in items:
+            item_names.append(key)
             self._config[key].set_value_source(value, origin)
 
         from ..internal.telemetry import telemetry_writer
 
-        cs = [{"name": k, "value": self._config[k].value(), "origin": self._get_source(k)} for k, _, _ in items]
-        telemetry_writer.add_event({"configuration": cs}, "app-client-configuration-change")
-        self._notify_subscribers([i[0] for i in items])
+        telemetry_writer.add_configs_changed(item_names)
+        self._notify_subscribers(item_names)
 
     def _reset(self):
         # type: () -> None

tests/integration/test_settings.py

@@ -1,3 +1,5 @@
+import os
+
 import pytest
 
 from .test_integration import AGENT_VERSION
@@ -11,6 +13,85 @@ def _get_latest_telemetry_config_item(events, item_name):
     return None
 
 
+@pytest.mark.skipif(AGENT_VERSION != "testagent", reason="Tests only compatible with a testagent")
+def test_setting_origin_environment(test_agent_session, run_python_code_in_subprocess):
+    env = os.environ.copy()
+    env.update(
+        {
+            "DD_TRACE_SAMPLE_RATE": "0.1",
+            "DD_LOGS_INJECTION": "true",
+            "DD_TRACE_HEADER_TAGS": "X-Header-Tag-1:header_tag_1,X-Header-Tag-2:header_tag_2",
+        }
+    )
+    out, err, status, _ = run_python_code_in_subprocess(
+        """
+from ddtrace import config, tracer
+with tracer.trace("test") as span:
+    pass
+        """,
+        env=env,
+    )
+    assert status == 0, err
+
+    events = test_agent_session.get_events()
+    assert _get_latest_telemetry_config_item(events, "trace_sample_rate") == {
+        "name": "trace_sample_rate",
+        "value": "0.1",
+        "origin": "env_var",
+    }
+    assert _get_latest_telemetry_config_item(events, "logs_injection_enabled") == {
+        "name": "logs_injection_enabled",
+        "value": "true",
+        "origin": "env_var",
+    }
+    assert _get_latest_telemetry_config_item(events, "trace_header_tags") == {
+        "name": "trace_header_tags",
+        "value": "X-Header-Tag-1:header_tag_1,X-Header-Tag-2:header_tag_2",
+        "origin": "env_var",
+    }
+
+
+@pytest.mark.skipif(AGENT_VERSION != "testagent", reason="Tests only compatible with a testagent")
+def test_setting_origin_code(test_agent_session, run_python_code_in_subprocess):
+    env = os.environ.copy()
+    env.update(
+        {
+            "DD_TRACE_SAMPLE_RATE": "0.1",
+            "DD_LOGS_INJECTION": "true",
+            "DD_TRACE_HEADER_TAGS": "X-Header-Tag-1:header_tag_1,X-Header-Tag-2:header_tag_2",
+        }
+    )
+    out, err, status, _ = run_python_code_in_subprocess(
+        """
+from ddtrace import config, tracer
+config._trace_sample_rate = 0.2
+config.logs_injection = False
+config.trace_http_header_tags = {"header": "value"}
+with tracer.trace("test") as span:
+    pass
+        """,
+        env=env,
+    )
+    assert status == 0, err
+
+    events = test_agent_session.get_events()
+    assert _get_latest_telemetry_config_item(events, "trace_sample_rate") == {
+        "name": "trace_sample_rate",
+        "value": "0.2",
+        "origin": "code",
+    }
+    assert _get_latest_telemetry_config_item(events, "logs_injection_enabled") == {
+        "name": "logs_injection_enabled",
+        "value": "false",
+        "origin": "code",
+    }
+    assert _get_latest_telemetry_config_item(events, "trace_header_tags") == {
+        "name": "trace_header_tags",
+        "value": "header:value",
+        "origin": "code",
+    }
+
+
 @pytest.mark.skipif(AGENT_VERSION != "testagent", reason="Tests only compatible with a testagent")
 def test_remoteconfig_sampling_rate_default(test_agent_session, run_python_code_in_subprocess):
     out, err, status, _ = run_python_code_in_subprocess(
@@ -46,9 +127,9 @@ def test_remoteconfig_sampling_rate_default(test_agent_session, run_python_code_
     assert status == 0, err
 
     events = test_agent_session.get_events()
-    assert _get_latest_telemetry_config_item(events, "_trace_sample_rate") == {
-        "name": "_trace_sample_rate",
-        "value": 1.0,
+    assert _get_latest_telemetry_config_item(events, "trace_sample_rate") == {
+        "name": "trace_sample_rate",
+        "value": "1.0",
         "origin": "default",
     }
 
@@ -69,8 +150,8 @@ def test_remoteconfig_sampling_rate_telemetry(test_agent_session, run_python_cod
     assert status == 0, err
 
     events = test_agent_session.get_events()
-    assert _get_latest_telemetry_config_item(events, "_trace_sample_rate") == {
-        "name": "_trace_sample_rate",
-        "value": 0.5,
+    assert _get_latest_telemetry_config_item(events, "trace_sample_rate") == {
+        "name": "trace_sample_rate",
+        "value": "0.5",
         "origin": "remote_config",
     }

tests/internal/test_settings.py

@@ -50,7 +50,7 @@ def _base_rc_config(cfg):
         {
             "env": {"DD_TRACE_SAMPLE_RATE": "0.9"},
             "expected": {"_trace_sample_rate": 0.9},
-            "expected_source": {"_trace_sample_rate": "env"},
+            "expected_source": {"_trace_sample_rate": "env_var"},
         },
         {
             "env": {"DD_TRACE_SAMPLE_RATE": "0.9"},
@@ -68,7 +68,7 @@ def _base_rc_config(cfg):
         {
             "env": {"DD_LOGS_INJECTION": "true"},
             "expected": {"logs_injection": True},
-            "expected_source": {"logs_injection": "env"},
+            "expected_source": {"logs_injection": "env_var"},
         },
         {
             "env": {"DD_LOGS_INJECTION": "true"},
@@ -81,7 +81,7 @@ def _base_rc_config(cfg):
             "expected": {
                 "trace_http_header_tags": {"X-Header-Tag-1": "header_tag_1", "X-Header-Tag-2": "header_tag_2"}
             },
-            "expected_source": {"trace_http_header_tags": "env"},
+            "expected_source": {"trace_http_header_tags": "env_var"},
         },
         {
             "env": {"DD_TRACE_HEADER_TAGS": "X-Header-Tag-1:header_tag_1,X-Header-Tag-2:header_tag_2"},