fix(parametric): wait for post-restart Node telemetry

tests/parametric/conftest.py

@@ -188,13 +188,15 @@ def nodejs_telemetry_value(test_agent: TestAgentAPI, dd_key: str) -> str | int |
     return entries[0].get("value")
 
 
-def assert_nodejs_telemetry_config(test_agent: TestAgentAPI, expected: dict) -> None:
+def assert_nodejs_telemetry_config(test_agent: TestAgentAPI, expected: dict, *, runtime_id: str | None = None) -> None:
     """Assert expected dd_* config values against the nodejs telemetry configuration."""
-    configuration_by_name = test_agent.wait_for_telemetry_configurations()
+    if runtime_id is None:
+        runtime_id = test_agent.wait_for_telemetry_runtime_id()
+    configurations = test_agent.wait_for_telemetry_configurations(runtime_id=runtime_id)
     for dd_key, expected_value in expected.items():
         name = _telemetry_name(dd_key)
-        entries = configuration_by_name.get(name)
-        assert entries, f"No telemetry configuration '{name}'"
+        entries = configurations.get(name)
+        assert entries, f"No telemetry configuration '{name}' for runtime '{runtime_id}'"
         actual = entries[0].get("value")
         if dd_key == "dd_tags":
             actual_tags = "" if actual is None else str(actual)
@@ -205,6 +207,14 @@ def assert_nodejs_telemetry_config(test_agent: TestAgentAPI, expected: dict) ->
             assert str(actual).lower() == str(expected_value).lower(), f"Expected {name}={expected_value}, got {actual}"
 
 
+def restart_and_get_runtime_id(test_agent: TestAgentAPI, test_library: APMLibrary) -> str | None:
+    previous_runtime_id = test_agent.wait_for_telemetry_runtime_id() if test_library.lang == "nodejs" else None
+    test_library.container_restart()
+    if previous_runtime_id is None:
+        return None
+    return test_agent.wait_for_telemetry_runtime_id(exclude=previous_runtime_id)
+
+
 def nodejs_startup_config(test_library: APMLibrary) -> dict:
     """Parse the tracer's published `DATADOG TRACER CONFIGURATION - {json}` startup line.
 

tests/parametric/test_config_consistency.py

@@ -13,7 +13,13 @@
 )
 from utils.docker_fixtures.spec.trace import find_span_in_traces, find_only_span
 from utils.docker_fixtures import TestAgentAPI
-from .conftest import APMLibrary, StableConfigWriter, assert_nodejs_telemetry_config, nodejs_startup_config
+from .conftest import (
+    APMLibrary,
+    StableConfigWriter,
+    assert_nodejs_telemetry_config,
+    nodejs_startup_config,
+    restart_and_get_runtime_id,
+)
 
 parametrize = pytest.mark.parametrize
 
@@ -537,9 +543,9 @@ def test_default_config(
                 path,
                 test_library,
             )
-            test_library.container_restart()
+            runtime_id = restart_and_get_runtime_id(test_agent, test_library)
             if test_library.lang == "nodejs":
-                assert_nodejs_telemetry_config(test_agent, expected)
+                assert_nodejs_telemetry_config(test_agent, expected, runtime_id=runtime_id)
             else:
                 config = test_library.config()
                 assert expected.items() <= config.items()
@@ -597,9 +603,9 @@ def test_extended_configs(
                 path,
                 test_library,
             )
-            test_library.container_restart()
+            runtime_id = restart_and_get_runtime_id(test_agent, test_library)
             if test_library.lang == "nodejs":
-                assert_nodejs_telemetry_config(test_agent, expected)
+                assert_nodejs_telemetry_config(test_agent, expected, runtime_id=runtime_id)
                 return
 
             config = test_library.config()
@@ -656,9 +662,9 @@ def test_unknown_key_skipped(self, test_agent: TestAgentAPI, test_library: APMLi
                 path,
                 test_library,
             )
-            test_library.container_restart()
+            runtime_id = restart_and_get_runtime_id(test_agent, test_library)
             if test_library.lang == "nodejs":
-                assert_nodejs_telemetry_config(test_agent, test["expected"])
+                assert_nodejs_telemetry_config(test_agent, test["expected"], runtime_id=runtime_id)
             else:
                 config = test_library.config()
                 assert test["expected"].items() <= config.items()
@@ -677,9 +683,9 @@ def test_invalid_files(self, test_agent: TestAgentAPI, test_library: APMLibrary,
                 path,
                 test_library,
             )
-            test_library.container_restart()
+            runtime_id = restart_and_get_runtime_id(test_agent, test_library)
             if test_library.lang == "nodejs":
-                assert_nodejs_telemetry_config(test_agent, SDK_DEFAULT_STABLE_CONFIG)
+                assert_nodejs_telemetry_config(test_agent, SDK_DEFAULT_STABLE_CONFIG, runtime_id=runtime_id)
             else:
                 config = test_library.config()
                 assert SDK_DEFAULT_STABLE_CONFIG.items() <= config.items()
@@ -750,9 +756,9 @@ def test_config_precedence(
                 test_library,
             )
 
-            test_library.container_restart()
+            runtime_id = restart_and_get_runtime_id(test_agent, test_library)
             if test_library.lang == "nodejs":
-                assert_nodejs_telemetry_config(test_agent, expected)
+                assert_nodejs_telemetry_config(test_agent, expected, runtime_id=runtime_id)
                 return
 
             config = test_library.config()
@@ -790,9 +796,9 @@ def test_targeting_rules(self, test_agent: TestAgentAPI, test_library: APMLibrar
                 path,
                 test_library,
             )
-            test_library.container_restart()
+            runtime_id = restart_and_get_runtime_id(test_agent, test_library)
             if test_library.lang == "nodejs":
-                assert_nodejs_telemetry_config(test_agent, {"dd_service": "my-service"})
+                assert_nodejs_telemetry_config(test_agent, {"dd_service": "my-service"}, runtime_id=runtime_id)
             else:
                 config = test_library.config()
                 assert config["dd_service"] == "my-service", (
@@ -826,9 +832,9 @@ def test_process_arguments(self, test_agent: TestAgentAPI, test_library: APMLibr
             # Use custom dumper for this specific test
             stable_config_content = yaml.dump(config, Dumper=CustomDumper)
             self.write_stable_config_content(stable_config_content, path, test_library)
-            test_library.container_restart()
+            runtime_id = restart_and_get_runtime_id(test_agent, test_library)
             if test_library.lang == "nodejs":
-                assert_nodejs_telemetry_config(test_agent, {"dd_service": "value"})
+                assert_nodejs_telemetry_config(test_agent, {"dd_service": "value"}, runtime_id=runtime_id)
             else:
                 lib_config = test_library.config()
                 assert lib_config["dd_service"] == "value", (

tests/test_the_test/test_test_agent.py

@@ -0,0 +1,100 @@
+import pytest
+
+from tests.parametric.conftest import assert_nodejs_telemetry_config
+from utils import scenarios
+from utils.docker_fixtures import TestAgentAPI
+
+
+def _configuration_event(
+    *, runtime_id: str, tracer_time: int, configurations: list[dict[str, object]]
+) -> dict[str, object]:
+    return {
+        "request_type": "app-started",
+        "runtime_id": runtime_id,
+        "tracer_time": tracer_time,
+        "payload": {"configuration": configurations},
+    }
+
+
+def _config(name: str, value: str, *, seq_id: int = 0) -> dict[str, object]:
+    return {"name": name, "seq_id": seq_id, "value": value}
+
+
+def _test_agent(monkeypatch: pytest.MonkeyPatch, responses: list[list[dict[str, object]]]) -> TestAgentAPI:
+    test_agent = object.__new__(TestAgentAPI)
+
+    def telemetry(*, clear: bool = False) -> list[dict[str, object]]:
+        assert clear is False
+        return responses.pop(0) if len(responses) > 1 else responses[0]
+
+    monkeypatch.setattr(test_agent, "telemetry", telemetry)
+    monkeypatch.setattr("utils.docker_fixtures._test_agent.time.sleep", lambda _seconds: None)
+    return test_agent
+
+
+@scenarios.test_the_test
+def test_nodejs_telemetry_assertion_waits_for_post_restart_runtime(monkeypatch: pytest.MonkeyPatch) -> None:
+    stale_event = _configuration_event(
+        runtime_id="before-restart",
+        tracer_time=1,
+        configurations=[_config("DD_TRACE_PROPAGATION_STYLE", "datadog,tracecontext,baggage")],
+    )
+    current_event = _configuration_event(
+        runtime_id="after-restart",
+        tracer_time=2,
+        configurations=[_config("DD_TRACE_PROPAGATION_STYLE", "tracecontext")],
+    )
+    test_agent = _test_agent(monkeypatch, [[stale_event], [stale_event, current_event]])
+    runtime_id = test_agent.wait_for_telemetry_runtime_id(exclude="before-restart")
+
+    assert_nodejs_telemetry_config(
+        test_agent,
+        {"dd_trace_propagation_style": "tracecontext"},
+        runtime_id=runtime_id,
+    )
+
+
+@scenarios.test_the_test
+@pytest.mark.parametrize(
+    ("events", "expected"),
+    [
+        (
+            [
+                _configuration_event(
+                    runtime_id="before-restart",
+                    tracer_time=1,
+                    configurations=[_config("DD_TRACE_PROPAGATION_STYLE", "tracecontext")],
+                ),
+                _configuration_event(
+                    runtime_id="after-restart",
+                    tracer_time=2,
+                    configurations=[_config("DD_TRACE_PROPAGATION_STYLE", "datadog")],
+                ),
+            ],
+            {"dd_trace_propagation_style": "tracecontext"},
+        ),
+        (
+            [
+                _configuration_event(
+                    runtime_id="after-restart",
+                    tracer_time=2,
+                    configurations=[
+                        _config("DD_SERVICE", "expected", seq_id=0),
+                        _config("DD_SERVICE", "wrong", seq_id=1),
+                    ],
+                )
+            ],
+            {"dd_service": "expected"},
+        ),
+    ],
+    ids=["stale-match", "superseded-sequence"],
+)
+def test_nodejs_telemetry_assertion_rejects_invalid_runtime_snapshot(
+    monkeypatch: pytest.MonkeyPatch,
+    events: list[dict[str, object]],
+    expected: dict[str, object],
+) -> None:
+    test_agent = _test_agent(monkeypatch, [events])
+
+    with pytest.raises(AssertionError):
+        assert_nodejs_telemetry_config(test_agent, expected, runtime_id="after-restart")

utils/docker_fixtures/_test_agent.py

@@ -705,21 +705,22 @@ def wait_for_telemetry_event(self, event_name: str, *, clear: bool = False, wait
         raise AssertionError(f"Telemetry event {event_name} not found")
 
     def wait_for_telemetry_configurations(
-        self, *, service: str | None = None, clear: bool = False, wait_loops: int = 100
-    ) -> dict[str, list[dict]]:
+        self,
+        *,
+        service: str | None = None,
+        runtime_id: str | None = None,
+        clear: bool = False,
+        wait_loops: int = 100,
+    ) -> dict[str, list[dict[str, Any]]]:
         """Waits for and returns configurations captured in telemetry events.
 
         Telemetry events can be found in `app-started` or `app-client-configuration-change` events.
-        The function ensures that at least one telemetry event is captured before processing.
         Returns a dictionary where keys are configuration names and values are lists of
         configuration dictionaries, allowing for multiple entries per configuration name
         with different origins.
         """
-        events = []
-        configurations: dict[str, list[dict]] = {}
-        # Poll until telemetry is captured instead of sleeping a fixed delay: returns as soon as
-        # app-started arrives (usually within a heartbeat) and retries the empty-read window that
-        # a single fixed-delay read can land in.
+        events: list[dict[str, Any]] = []
+        configurations: dict[str, list[dict[str, Any]]] = {}
         for _ in range(wait_loops):
             with contextlib.suppress(requests.exceptions.RequestException):
                 events = self.telemetry(clear=False)
@@ -729,33 +730,43 @@ def wait_for_telemetry_configurations(
         else:
             raise AssertionError("No telemetry events were found. Ensure the application is sending telemetry events.")
 
-        # Sort events by tracer_time to ensure configurations are processed in order
-        events.sort(key=lambda r: r["tracer_time"])
-
-        # Extract configuration data from relevant telemetry events
+        events.sort(key=lambda event: event["tracer_time"])
         for event in events:
+            if runtime_id is not None and event.get("runtime_id") != runtime_id:
+                continue
             if service is not None and event["application"]["service_name"] != service:
                 continue
             for event_type in ["app-started", "app-client-configuration-change"]:
                 telemetry_event = self._get_telemetry_event(event, event_type)
                 if telemetry_event:
                     for config in telemetry_event.get("payload", {}).get("configuration", []):
-                        # Store all configurations, allowing multiple entries per name with different origins
-                        config_name = config["name"]
-                        if config_name not in configurations:
-                            configurations[config_name] = []
-                        configurations[config_name].append(config)
-        if configurations:
-            # Checking if we need to sort due to multiple sources being sent for the same config
-            sample_key = next(iter(configurations))
-            if "seq_id" in configurations[sample_key][0] and configurations[sample_key][0]["seq_id"] is not None:
-                # Sort seq_id for each config from highest to lowest
-                for payload in configurations.values():
-                    payload.sort(key=lambda item: item["seq_id"], reverse=True)
+                        configurations.setdefault(config["name"], []).append(config)
+        for payload in configurations.values():
+            payload.sort(key=lambda item: item.get("seq_id") or 0, reverse=True)
         if clear:
             self.clear()
         return configurations
 
+    def wait_for_telemetry_runtime_id(self, *, exclude: str | None = None, wait_loops: int = 200) -> str:
+        observed_runtime_ids: set[str] = set()
+        for _ in range(wait_loops):
+            events: list[dict[str, Any]] = []
+            with contextlib.suppress(requests.exceptions.RequestException):
+                events = self.telemetry(clear=False)
+            events.sort(key=lambda event: event["tracer_time"], reverse=True)
+            for event in events:
+                telemetry_event = self._get_telemetry_event(event, "app-started")
+                if telemetry_event is None:
+                    continue
+                runtime_id = telemetry_event.get("runtime_id") or event.get("runtime_id")
+                if runtime_id is None:
+                    continue
+                observed_runtime_ids.add(runtime_id)
+                if runtime_id != exclude:
+                    return runtime_id
+            time.sleep(0.01)
+        raise AssertionError(f"No telemetry runtime ID found excluding {exclude}. Observed: {observed_runtime_ids}")
+
     def get_telemetry_config_by_origin(
         self,
         configurations: dict[str, list[dict]],