Fix Prefect flaky test (DataDog#23219)

* Add Waitfor for tasks

* Add timeout

* Increase window between runs

* Reduce timeout

Author: lucia-sb

prefect/tests/conftest.py

@@ -7,8 +7,9 @@
 from typing import Callable
 
 import pytest
+import requests
 
-from datadog_checks.dev.conditions import CheckDockerLogs, CheckEndpoints
+from datadog_checks.dev.conditions import CheckDockerLogs, CheckEndpoints, WaitFor
 from datadog_checks.dev.docker import docker_run, get_docker_hostname
 from datadog_checks.dev.utils import find_free_port
 from datadog_checks.prefect import PrefectCheck
@@ -20,6 +21,23 @@
 }
 
 
+def _check_task_runs_available(prefect_url: str):
+    """Verify that the Prefect API has task runs and task-run events queryable."""
+    resp = requests.post(f"{prefect_url}/task_runs/filter", json={}, timeout=1)
+    resp.raise_for_status()
+    task_runs = resp.json()
+    assert task_runs, "No task runs available in Prefect API yet"
+
+    resp = requests.post(
+        f"{prefect_url}/events/filter",
+        json={"filter": {"event": {"prefix": ["prefect.task-run"]}}},
+        timeout=1,
+    )
+    resp.raise_for_status()
+    events = resp.json().get("events", [])
+    assert events, "No task-run events available in Prefect API yet"
+
+
 @pytest.fixture(scope='session')
 def dd_environment(instance: Callable[[str], dict[str, str | dict[str, list[str]] | None | bool | int]]):
     port = find_free_port(get_docker_hostname())
@@ -35,6 +53,7 @@ def dd_environment(instance: Callable[[str], dict[str, str | dict[str, list[str]
             COMPOSE_FILE_E2E, patterns=["Finished all tasks"], service="prefect-worker", attempts=120, wait=1
         ),
         CheckDockerLogs(COMPOSE_FILE_E2E, patterns=["Retried"], service="prefect-worker", attempts=120, wait=1),
+        WaitFor(lambda: _check_task_runs_available(prefect_url), attempts=60, wait=2),
     ]
 
     with docker_run(

prefect/tests/test_e2e.py

@@ -83,7 +83,7 @@
 
 @pytest.mark.e2e
 def test_e2e_metrics(dd_agent_check):
-    aggregator = dd_agent_check()
+    aggregator = dd_agent_check(check_times=2, pause=20000)
 
     cross_check_metrics = (
         'flow_runs.retry_gaps_duration',