chore(telemetry): report all configuration sources with sequence tracking [config chaining] (#14122)

## Description

This PR updates the behavior of the telemetry writer to ensure that all
discovered configuration sources are submitted to the instrumentation
platform. Previously, only the effective (final) configuration value was
sent—typically the last value set in code within 60 seconds.

Now, the telemetry writer will queue and send every configuration object
it observes. Each configuration entry will include a seq_id field to
indicate the order in which configurations were applied. This allows the
telemetry intake system to reconstruct the full configuration history
and understand how a given value was derived.

## Motivation

Accurate tracking of configuration state is essential for debugging,
auditing, and understanding application behavior across environments. By
reporting every configuration source (e.g., environment variable, fleet
config, in-code setting), we provide full visibility into how and when
each setting was applied. The addition of seq_id enables the intake
pipeline to reliably determine the configuration precedence and
reconstruct state transitions over time.

Note: should follow this spec
https://docs.google.com/document/d/1vhIimn2vt4tDRSxsHn6vWSc8zYHl0Lv0Fk7CQps04C4/edit?tab=t.0

## Next Steps

- Update the get_config(...) and report_telemetry methods to:
- Queue a configuration event whenever a new configuration source is
discovered.
- Ensure that multiple sources for the same config key (e.g.,
environment variable, fleet config, code) are reported, each with its
own seq_id to reflect the correct order of precedence.

## Checklist
- [ ] PR author has checked that all the criteria below are met
- The PR description includes an overview of the change
- The PR description articulates the motivation for the change
- The change includes tests OR the PR description describes a testing
strategy
- The PR description notes risks associated with the change, if any
- Newly-added code is easy to change
- The change follows the [library release note
guidelines](https://ddtrace.readthedocs.io/en/stable/releasenotes.html)
- The change includes or references documentation updates if necessary
- Backport labels are set (if
[applicable](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting))

## Reviewer Checklist
- [ ] Reviewer has checked that all the criteria below are met 
- Title is accurate
- All changes are related to the pull request's stated goal
- Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes
- Testing strategy adequately addresses listed risks
- Newly-added code is easy to change
- Release note makes sense to a user of the library
- If necessary, author has acknowledged and discussed the performance
implications of this PR as reported in the benchmarks PR comment
- Backport labels are set in a manner that is consistent with the
[release branch maintenance
policy](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting)

Author: mabdinur

ddtrace/internal/telemetry/writer.py

@@ -145,7 +145,8 @@ class TelemetryWriter(PeriodicService):
     # Counter representing the number of events sent by the writer. Here we are relying on the atomicity
     # of `itertools.count()` which is a CPython implementation detail. The sequence field in telemetry
     # payloads is only used in tests and is not required to process Telemetry events.
-    _sequence = itertools.count(1)
+    _sequence_payloads = itertools.count(1)
+    _sequence_configurations = itertools.count(1)
     _ORIGINAL_EXCEPTHOOK = staticmethod(sys.excepthook)
     CWD = os.getcwd()
 
@@ -168,7 +169,7 @@ def __init__(self, is_periodic=True, agentless=None):
         self._logs = set()  # type: Set[Dict[str, Any]]
         self._forked = False  # type: bool
         self._events_queue = []  # type: List[Dict]
-        self._configuration_queue = {}  # type: Dict[str, Dict]
+        self._configuration_queue = []  # type: List[Dict]
         self._imported_dependencies: Dict[str, str] = dict()
         self._modules_already_imported: Set[str] = set()
         self._product_enablement = {product.value: False for product in TELEMETRY_APM_PRODUCT}
@@ -266,7 +267,7 @@ def add_event(self, payload, payload_type):
                 "tracer_time": int(time.time()),
                 "runtime_id": get_runtime_id(),
                 "api_version": "v2",
-                "seq_id": next(self._sequence),
+                "seq_id": next(self._sequence_payloads),
                 "debug": self._debug,
                 "application": get_application(config.SERVICE, config.VERSION, config.ENV),
                 "host": get_host_info(),
@@ -388,8 +389,8 @@ def _flush_configuration_queue(self):
         # type: () -> List[Dict]
         """Flushes and returns a list of all queued configurations"""
         with self._service_lock:
-            configurations = list(self._configuration_queue.values())
-            self._configuration_queue = {}
+            configurations = self._configuration_queue
+            self._configuration_queue = []
         return configurations
 
     def _app_client_configuration_changed_event(self, configurations):
@@ -464,10 +465,6 @@ def product_activated(self, product, enabled):
         if self.started:
             self._send_product_change_updates = True
 
-    def remove_configuration(self, configuration_name):
-        with self._service_lock:
-            del self._configuration_queue[configuration_name]
-
     def add_configuration(self, configuration_name, configuration_value, origin="unknown", config_id=None):
         # type: (str, Any, str, Optional[str]) -> None
         """Creates and queues the name, origin, value of a configuration"""
@@ -488,17 +485,21 @@ def add_configuration(self, configuration_name, configuration_value, origin="unk
             config["config_id"] = config_id
 
         with self._service_lock:
-            self._configuration_queue[configuration_name] = config
+            config["seq_id"] = next(self._sequence_configurations)
+            self._configuration_queue.append(config)
 
     def add_configurations(self, configuration_list):
         """Creates and queues a list of configurations"""
         with self._service_lock:
-            for name, value, _origin in configuration_list:
-                self._configuration_queue[name] = {
-                    "name": name,
-                    "origin": _origin,
-                    "value": value,
-                }
+            for name, value, origin in configuration_list:
+                self._configuration_queue.append(
+                    {
+                        "name": name,
+                        "origin": origin,
+                        "value": value,
+                        "seq_id": next(self._sequence_configurations),
+                    }
+                )
 
     def add_log(self, level, message, stack_trace="", tags=None):
         """
@@ -701,7 +702,7 @@ def reset_queues(self):
         self._namespace.flush()
         self._logs = set()
         self._imported_dependencies = {}
-        self._configuration_queue = {}
+        self._configuration_queue = []
 
     def _flush_events_queue(self):
         # type: () -> List[Dict]
@@ -727,7 +728,8 @@ def _fork_writer(self):
         self.enable()
 
     def _restart_sequence(self):
-        self._sequence = itertools.count(1)
+        self._sequence_payloads = itertools.count(1)
+        self._sequence_configurations = itertools.count(1)
 
     def _stop_service(self, join=True, *args, **kwargs):
         # type: (...) -> None

tests/conftest.py

@@ -621,7 +621,7 @@ def get_metrics(self, name=None):
             for series in event["payload"]["series"]:
                 if name is None or series["metric"] == name:
                     metrics.append(series)
-        metrics.sort(key=lambda x: (x["metric"], x["tags"]), reverse=False)
+        metrics.sort(key=lambda x: (x["metric"], x["tags"]))
         return metrics
 
     def get_dependencies(self, name=None):
@@ -630,18 +630,28 @@ def get_dependencies(self, name=None):
             for dep in event["payload"]["dependencies"]:
                 if name is None or dep["name"] == name:
                     deps.append(dep)
-        deps.sort(key=lambda x: x["name"], reverse=False)
+        deps.sort(key=lambda x: x["name"])
         return deps
 
-    def get_configurations(self, name=None, ignores=None):
+    def get_configurations(self, name=None, ignores=None, remove_seq_id=False, effective=False):
         ignores = ignores or []
         configurations = []
         events_with_configs = self.get_events("app-started") + self.get_events("app-client-configuration-change")
         for event in events_with_configs:
             for c in event["payload"]["configuration"]:
-                if c["name"] == name or (name is None and c["name"] not in ignores):
-                    configurations.append(c)
-        configurations.sort(key=lambda x: x["name"], reverse=False)
+                config = c.copy()
+                if remove_seq_id:
+                    config.pop("seq_id")
+                if config["name"] == name or (name is None and config["name"] not in ignores):
+                    configurations.append(config)
+
+        if effective:
+            config_map = {}
+            for c in configurations:
+                config_map[c["name"]] = c
+            configurations = list(config_map.values())
+
+        configurations.sort(key=lambda x: x["name"])
         return configurations
 
 

tests/integration/test_settings.py

@@ -5,16 +5,6 @@
 from tests.integration.utils import AGENT_VERSION
 
 
-def _get_telemetry_config_items(events, item_name):
-    items = []
-    for event in reversed(sorted(events, key=lambda e: (e["tracer_time"], e["seq_id"]))):
-        for item in reversed(event.get("payload", {}).get("configuration", [])):
-            if item_name == item["name"]:
-                items.append(item)
-    if items:
-        return items or 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()
@@ -38,30 +28,25 @@ def test_setting_origin_environment(test_agent_session, run_python_code_in_subpr
     )
     assert status == 0, err
 
-    events = test_agent_session.get_events(subprocess=True)
-    events_trace_sample_rate = _get_telemetry_config_items(events, "DD_TRACE_SAMPLING_RULES")
+    configurations = test_agent_session.get_configurations(effective=True, remove_seq_id=True)
 
     assert {
         "name": "DD_TRACE_SAMPLING_RULES",
         "value": '[{"sample_rate":0.1}]',
         "origin": "env_var",
-    } in events_trace_sample_rate
+    } in configurations
 
-    events_logs_injection_enabled = _get_telemetry_config_items(events, "DD_LOGS_INJECTION")
-    assert {"name": "DD_LOGS_INJECTION", "value": True, "origin": "env_var"} in events_logs_injection_enabled
+    assert {"name": "DD_LOGS_INJECTION", "value": True, "origin": "env_var"} in configurations
 
-    events_trace_header_tags = _get_telemetry_config_items(events, "DD_TRACE_HEADER_TAGS")
     assert {
         "name": "DD_TRACE_HEADER_TAGS",
         "value": "X-Header-Tag-1:header_tag_1,X-Header-Tag-2:header_tag_2",
         "origin": "env_var",
-    } in events_trace_header_tags
+    } in configurations
 
-    events_trace_tags = _get_telemetry_config_items(events, "DD_TAGS")
-    assert {"name": "DD_TAGS", "value": "team:apm,component:web", "origin": "env_var"} in events_trace_tags
+    assert {"name": "DD_TAGS", "value": "team:apm,component:web", "origin": "env_var"} in configurations
 
-    events_tracing_enabled = _get_telemetry_config_items(events, "DD_TRACE_ENABLED")
-    assert {"name": "DD_TRACE_ENABLED", "value": True, "origin": "env_var"} in events_tracing_enabled
+    assert {"name": "DD_TRACE_ENABLED", "value": True, "origin": "env_var"} in configurations
 
 
 @pytest.mark.skipif(AGENT_VERSION != "testagent", reason="Tests only compatible with a testagent")
@@ -93,35 +78,31 @@ def test_setting_origin_code(test_agent_session, run_python_code_in_subprocess):
     )
     assert status == 0, err
 
-    events = test_agent_session.get_events(subprocess=True)
+    configurations = test_agent_session.get_configurations(effective=True, remove_seq_id=True)
 
-    events_logs_injection_enabled = _get_telemetry_config_items(events, "DD_LOGS_INJECTION")
     assert {
         "name": "DD_LOGS_INJECTION",
         "value": False,
         "origin": "code",
-    } in events_logs_injection_enabled
+    } in configurations
 
-    events_trace_header_tags = _get_telemetry_config_items(events, "DD_TRACE_HEADER_TAGS")
     assert {
         "name": "DD_TRACE_HEADER_TAGS",
         "value": "header:value",
         "origin": "code",
-    } in events_trace_header_tags
+    } in configurations
 
-    events_trace_tags = _get_telemetry_config_items(events, "DD_TAGS")
     assert {
         "name": "DD_TAGS",
         "value": "header:value",
         "origin": "code",
-    } in events_trace_tags
+    } in configurations
 
-    events_tracing_enabled = _get_telemetry_config_items(events, "DD_TRACE_ENABLED")
     assert {
         "name": "DD_TRACE_ENABLED",
         "value": False,
         "origin": "code",
-    } in events_tracing_enabled
+    } in configurations
 
 
 @pytest.mark.skipif(AGENT_VERSION != "testagent", reason="Tests only compatible with a testagent")
@@ -166,9 +147,10 @@ def test_remoteconfig_sampling_rate_default(test_agent_session, ddtrace_run_pyth
     )
     assert status == 0, err
 
-    events = test_agent_session.get_events(subprocess=True)
-    events_trace_sample_rate = _get_telemetry_config_items(events, "DD_TRACE_SAMPLING_RULES")
-    assert {"name": "DD_TRACE_SAMPLING_RULES", "value": "", "origin": "default"} in events_trace_sample_rate
+    configurations = test_agent_session.get_configurations(name="DD_TRACE_SAMPLING_RULES", effective=True)
+    assert len(configurations) == 1
+    assert configurations[0]["value"] == ""
+    assert configurations[0]["origin"] == "default"
 
 
 @pytest.mark.skipif(AGENT_VERSION != "testagent", reason="Tests only compatible with a testagent")
@@ -213,14 +195,17 @@ def test_remoteconfig_sampling_rate_telemetry(test_agent_session, run_python_cod
     )
     assert status == 0, err
 
-    events = test_agent_session.get_events(subprocess=True)
-    events_trace_sample_rate = _get_telemetry_config_items(events, "DD_TRACE_SAMPLING_RULES")
-    assert {
-        "name": "DD_TRACE_SAMPLING_RULES",
-        "origin": "remote_config",
-        "value": '[{"sample_rate": 0.5, "service": "*", "name": "*", "resource": "*", '
-        '"tags": {}, "provenance": "customer"}]',
-    } in events_trace_sample_rate
+    configurations = test_agent_session.get_configurations(
+        name="DD_TRACE_SAMPLING_RULES", effective=True, remove_seq_id=True
+    )
+    assert configurations == [
+        {
+            "name": "DD_TRACE_SAMPLING_RULES",
+            "origin": "remote_config",
+            "value": '[{"sample_rate": 0.5, "service": "*", "name": "*", "resource": "*", '
+            '"tags": {}, "provenance": "customer"}]',
+        }
+    ]
 
 
 @pytest.mark.skipif(AGENT_VERSION != "testagent", reason="Tests only compatible with a testagent")
@@ -258,10 +243,13 @@ def test_remoteconfig_header_tags_telemetry(test_agent_session, ddtrace_run_pyth
     )
     assert status == 0, err
 
-    events = test_agent_session.get_events(subprocess=True)
-    events_trace_header_tags = _get_telemetry_config_items(events, "DD_TRACE_HEADER_TAGS")
-    assert {
-        "name": "DD_TRACE_HEADER_TAGS",
-        "value": "used:header_tag_69,unused:header_tag_70,used-with-default:",
-        "origin": "remote_config",
-    } in events_trace_header_tags
+    configurations = test_agent_session.get_configurations(
+        name="DD_TRACE_HEADER_TAGS", effective=True, remove_seq_id=True
+    )
+    assert configurations == [
+        {
+            "name": "DD_TRACE_HEADER_TAGS",
+            "value": "used:header_tag_69,unused:header_tag_70,used-with-default:",
+            "origin": "remote_config",
+        }
+    ]

tests/telemetry/test_writer.py

@@ -79,7 +79,7 @@ def test_app_started_event_configuration_override_asm(
     _, stderr, status, _ = run_python_code_in_subprocess("import ddtrace.auto", env=env)
     assert status == 0, stderr
 
-    configuration = test_agent_session.get_configurations(name=env_var)
+    configuration = test_agent_session.get_configurations(name=env_var, remove_seq_id=True)
     assert len(configuration) == 1, configuration
     assert configuration[0] == {"name": env_var, "origin": "env_var", "value": expected_value}
 
@@ -296,12 +296,13 @@ def test_app_started_event_configuration_override(test_agent_session, run_python
 
     # DD_TRACE_AGENT_URL in gitlab is different from CI, to keep things simple we will
     # skip validating this config
-    configurations = test_agent_session.get_configurations(ignores=["DD_TRACE_AGENT_URL"])
+    configurations = test_agent_session.get_configurations(
+        ignores=["DD_TRACE_AGENT_URL", "DD_AGENT_PORT", "DD_TRACE_AGENT_PORT"], remove_seq_id=True, effective=True
+    )
     assert configurations
 
     expected = [
         {"name": "DD_AGENT_HOST", "origin": "default", "value": None},
-        {"name": "DD_AGENT_PORT", "origin": "default", "value": None},
         {"name": "DD_API_KEY", "origin": "default", "value": None},
         {"name": "DD_API_SECURITY_ENABLED", "origin": "env_var", "value": False},
         {"name": "DD_API_SECURITY_ENDPOINT_COLLECTION_ENABLED", "origin": "default", "value": True},
@@ -471,7 +472,6 @@ def test_app_started_event_configuration_override(test_agent_session, run_python
         {"name": "DD_TRACE_128_BIT_TRACEID_GENERATION_ENABLED", "origin": "env_var", "value": True},
         {"name": "DD_TRACE_128_BIT_TRACEID_LOGGING_ENABLED", "origin": "default", "value": False},
         {"name": "DD_TRACE_AGENT_HOSTNAME", "origin": "default", "value": None},
-        {"name": "DD_TRACE_AGENT_PORT", "origin": "default", "value": None},
         {"name": "DD_TRACE_AGENT_TIMEOUT_SECONDS", "origin": "default", "value": 2.0},
         {"name": "DD_TRACE_API_VERSION", "origin": "env_var", "value": "v0.5"},
         {"name": "DD_TRACE_BAGGAGE_TAG_KEYS", "origin": "default", "value": "user.id,account.id,session.id"},
@@ -743,28 +743,30 @@ def test_app_client_configuration_changed_event(telemetry_writer, test_agent_ses
     telemetry_writer.periodic(force_flush=True)
     """asserts that queuing a configuration sends a valid telemetry request"""
     with override_global_config(dict()):
-        telemetry_writer.add_configuration("appsec_enabled", True)
-        telemetry_writer.add_configuration("DD_TRACE_PROPAGATION_STYLE_EXTRACT", "datadog")
-        telemetry_writer.add_configuration("appsec_enabled", False, "env_var")
+        telemetry_writer.add_configuration("appsec_enabled", True, "env_var")
+        telemetry_writer.add_configuration("DD_TRACE_PROPAGATION_STYLE_EXTRACT", "datadog", "default")
+        telemetry_writer.add_configuration("appsec_enabled", False, "code")
 
         telemetry_writer.periodic(force_flush=True)
 
         events = test_agent_session.get_events("app-client-configuration-change")
         received_configurations = [c for event in events for c in event["payload"]["configuration"]]
-        received_configurations.sort(key=lambda c: c["name"])
-        # assert the latest configuration value is send to the agent
-        assert received_configurations == [
-            {
-                "name": "DD_TRACE_PROPAGATION_STYLE_EXTRACT",
-                "origin": "unknown",
-                "value": "datadog",
-            },
-            {
-                "name": "appsec_enabled",
-                "origin": "env_var",
-                "value": False,
-            },
-        ]
+        received_configurations.sort(key=lambda c: c["seq_id"])
+        assert (
+            received_configurations[0]["seq_id"]
+            < received_configurations[1]["seq_id"]
+            < received_configurations[2]["seq_id"]
+        )
+        # assert that all configuration values are sent to the agent in the order they were added (by seq_id)
+        assert received_configurations[0]["name"] == "appsec_enabled"
+        assert received_configurations[0]["origin"] == "env_var"
+        assert received_configurations[0]["value"] is True
+        assert received_configurations[1]["name"] == "DD_TRACE_PROPAGATION_STYLE_EXTRACT"
+        assert received_configurations[1]["origin"] == "default"
+        assert received_configurations[1]["value"] == "datadog"
+        assert received_configurations[2]["name"] == "appsec_enabled"
+        assert received_configurations[2]["origin"] == "code"
+        assert received_configurations[2]["value"] is False
 
 
 def test_add_integration_disabled_writer(telemetry_writer, test_agent_session):
@@ -992,7 +994,7 @@ def test_otel_config_telemetry(test_agent_session, run_python_code_in_subprocess
     _, stderr, status, _ = run_python_code_in_subprocess("import ddtrace", env=env)
     assert status == 0, stderr
 
-    configurations = {c["name"]: c for c in test_agent_session.get_configurations()}
+    configurations = {c["name"]: c for c in test_agent_session.get_configurations(remove_seq_id=True)}
 
     assert configurations["DD_SERVICE"] == {"name": "DD_SERVICE", "origin": "env_var", "value": "dd_service"}
     assert configurations["OTEL_LOG_LEVEL"] == {"name": "OTEL_LOG_LEVEL", "origin": "env_var", "value": "debug"}