telemetry: fix flaky configuration tests (#4283)

Author: mabdinur

tests/parametric/conftest.py

@@ -376,21 +376,61 @@ def wait_for_telemetry_event(self, event_name: str, *, clear: bool = False, wait
                 pass
             else:
                 for event in events:
-                    if event["request_type"] == "message-batch":
-                        for message in event["payload"]:
-                            if message["request_type"] == event_name:
-                                if message.get("application", {}).get("language_version") != "SIDECAR":
-                                    if clear:
-                                        self.clear()
-                                    return message
-                    elif event["request_type"] == event_name:
-                        if event.get("application", {}).get("language_version") != "SIDECAR":
-                            if clear:
-                                self.clear()
-                            return event
+                    e = self._get_telemetry_event(event, event_name)
+                    if e:
+                        if clear:
+                            self.clear()
+                        return e
             time.sleep(0.01)
         raise AssertionError(f"Telemetry event {event_name} not found")
 
+    def wait_for_telemetry_configurations(self, *, clear: bool = False) -> dict[str, str]:
+        """Waits for and returns the latest 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.
+        """
+        events = []
+        configurations = {}
+        # Allow time for telemetry events to be captured
+        time.sleep(1)
+        # Attempt to retrieve telemetry events, suppressing request-related exceptions
+        with contextlib.suppress(requests.exceptions.RequestException):
+            events += self.telemetry(clear=False)
+        if not events:
+            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
+        for event in events:
+            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 only the latest configuration for each name. This is the configuration
+                        # that should be used by the application.
+                        configurations[config["name"]] = config
+        if clear:
+            self.clear()
+        return configurations
+
+    def _get_telemetry_event(self, event, request_type):
+        """Extracts telemetry events from a message batch or returns the telemetry event if it
+        matches the expected request_type and was not emitted from a sidecar.
+        """
+        if not event:
+            return None
+        elif event["request_type"] == "message-batch":
+            for message in event["payload"]:
+                if message["request_type"] == request_type:
+                    if message.get("application", {}).get("language_version") != "SIDECAR":
+                        return message
+        elif event["request_type"] == request_type:
+            if event.get("application", {}).get("language_version") != "SIDECAR":
+                return event
+        return None
+
     def wait_for_rc_apply_state(
         self,
         product: str,

tests/parametric/test_telemetry.py

@@ -78,10 +78,8 @@ class Test_Defaults:
     def test_library_settings(self, library_env, test_agent, test_library):
         with test_library.dd_start_span("test"):
             pass
-        event = test_agent.wait_for_telemetry_event("app-started", wait_loops=400)
-        configuration = event["payload"]["configuration"]
 
-        configuration_by_name = {item["name"]: item for item in configuration}
+        configuration_by_name = test_agent.wait_for_telemetry_configurations()
         for apm_telemetry_name, value in [
             ("trace_sample_rate", (1.0, None, "1.0")),
             ("logs_injection_enabled", ("false", False, "true", True)),
@@ -152,10 +150,8 @@ class Test_Consistent_Configs:
     def test_library_settings(self, library_env, test_agent, test_library):
         with test_library.dd_start_span("test"):
             pass
-        event = test_agent.wait_for_telemetry_event("app-started", wait_loops=400)
-        configuration = event["payload"]["configuration"]
-        configuration_by_name = {item["name"]: item for item in configuration}
 
+        configuration_by_name = test_agent.wait_for_telemetry_configurations()
         # # Check that the tags name match the expected value
         assert configuration_by_name.get("DD_ENV", {}).get("value") == "dev"
         assert configuration_by_name.get("DD_SERVICE", {}).get("value") == "service_test"
@@ -189,10 +185,8 @@ def test_library_settings(self, library_env, test_agent, test_library):
     def test_library_settings_2(self, library_env, test_agent, test_library):
         with test_library.dd_start_span("test"):
             pass
-        event = test_agent.wait_for_telemetry_event("app-started", wait_loops=400)
-        configuration = event["payload"]["configuration"]
-        configuration_by_name = {item["name"]: item for item in configuration}
 
+        configuration_by_name = test_agent.wait_for_telemetry_configurations()
         assert configuration_by_name.get("DD_TRACE_LOG_DIRECTORY", {}).get("value") == "/some/temporary/directory"
         assert configuration_by_name.get("DD_TRACE_HTTP_CLIENT_ERROR_STATUSES", {}).get("value") == "200-250"
         assert configuration_by_name.get("DD_TRACE_HTTP_SERVER_ERROR_STATUSES", {}).get("value") == "250-200"
@@ -230,10 +224,8 @@ class Test_Environment:
     def test_library_settings(self, library_env, test_agent, test_library):
         with test_library.dd_start_span("test"):
             pass
-        event = test_agent.wait_for_telemetry_event("app-started", wait_loops=400)
-        configuration = event["payload"]["configuration"]
 
-        configuration_by_name = {item["name"]: item for item in configuration}
+        configuration_by_name = test_agent.wait_for_telemetry_configurations()
         for apm_telemetry_name, environment_value in [
             ("trace_sample_rate", ("0.3", 0.3)),
             ("logs_injection_enabled", ("true", True)),
@@ -496,9 +488,8 @@ def test_stable_configuration_origin(
             )
             test_library.container_restart()
             test_library.dd_start_span("test")
-        event = test_agent.wait_for_telemetry_event("app-started", wait_loops=400)
-        configuration = {c["name"]: c for c in event["payload"]["configuration"]}
 
+        configuration = test_agent.wait_for_telemetry_configurations()
         for cfg_name, origin in expected_origin.items():
             apm_telemetry_name = _mapped_telemetry_name(context, cfg_name)
             telemetry_item = configuration[apm_telemetry_name]