diff --git a/tests/parametric/conftest.py b/tests/parametric/conftest.py index 0c8d37874a2..4aa203eeca1 100644 --- a/tests/parametric/conftest.py +++ b/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. diff --git a/tests/parametric/test_config_consistency.py b/tests/parametric/test_config_consistency.py index 1868052ea2a..ac8823c4f7e 100644 --- a/tests/parametric/test_config_consistency.py +++ b/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", ( diff --git a/tests/test_the_test/test_test_agent.py b/tests/test_the_test/test_test_agent.py new file mode 100644 index 00000000000..35f970829fd --- /dev/null +++ b/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") diff --git a/utils/docker_fixtures/_test_agent.py b/utils/docker_fixtures/_test_agent.py index c6c89355e8b..35c15dab8fb 100644 --- a/utils/docker_fixtures/_test_agent.py +++ b/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]],