vectordotdev · thomasqueirozb · Jul 8, 2026 · Jul 8, 2026 · Jul 8, 2026 · Jul 9, 2026
@@ -1,5 +1,3 @@
-use std::time::Duration;
-
 use serde_json::Value;
 use tracing::info;
 use vector::test_util::trace_init;
@@ -8,7 +6,6 @@ use super::*;
 
 const LOGS_ENDPOINT: &str = "/api/v2/logs";
 const MAX_RETRIES: usize = 10;
-const WAIT_INTERVAL: Duration = Duration::from_secs(1);
 
 fn expected_log_events() -> usize {
     std::env::var("EXPECTED_LOG_EVENTS")
@@ -76,22 +73,18 @@ async fn validate() {
     // Retry until we have log payloads or hit max retries.
     // This is to ensure events flow through to fakeintake before asking for them.
     info!("getting log payloads from agent-only pipeline");
-    let mut agent_payloads = Vec::new();
-    for _ in 0..MAX_RETRIES {
-        agent_payloads = get_fakeintake_payloads::<FakeIntakeResponseJson>(
-            &fake_intake_agent_address(),
-            LOGS_ENDPOINT,
-        )
-        .await
-        .payloads;
-
-        if !agent_payloads.is_empty() {
-            break;
-        }
-
-        info!("No valid payloads yet, retrying...");
-        tokio::time::sleep(WAIT_INTERVAL).await;
-    }
+    let agent_address = fake_intake_agent_address();
+    let mut agent_payloads = poll_until(
+        MAX_RETRIES,
+        WAIT_INTERVAL,
+        || async {
+            get_fakeintake_payloads::<FakeIntakeResponseJson>(&agent_address, LOGS_ENDPOINT)
+                .await
+                .payloads
+        },
+        |payloads: &Vec<_>| !payloads.is_empty(),
+    )
+    .await;
 
     // If we still don't have valid payloads after retries, fail the test
     assert!(

@@ -15,6 +15,16 @@ mod sketches;
 
 use super::*;
 
+// Metrics may take a moment to flow through to fakeintake, so pipeline
+// fetches are retried until data shows up (or we give up).
+//
+// The dogstatsd emitter deliberately waits 10s after startup before sending
+// any metrics (tests/e2e/datadog-metrics/dogstatsd_client/client.py), and the
+// Agent's default aggregator flush interval (~15s) adds further delay on top
+// of that before metrics reach fakeintake. 45s covers that ~25s worst case
+// with margin for network/processing overhead.
+const MAX_RETRIES: usize = 45;
+
 async fn decompress_payload(payload: &[u8]) -> std::io::Result<Vec<u8>> {
     if is_zstd(payload) {
         let mut decompressor = ZstdDecoder::new(payload);
@@ -120,10 +130,6 @@ where
 async fn validate() {
     trace_init();
 
-    // Even with configuring docker service dependencies, we need a small buffer of time
-    // to ensure events flow through to fakeintake before asking for them
-    std::thread::sleep(std::time::Duration::from_secs(2));
-
     series::validate().await;
 
     sketches::validate().await;

@@ -83,17 +83,11 @@ fn generate_series_intake(payloads: &[MetricPayload]) -> SeriesIntake {
     intake
 }
 
-// runs assertions that each set of payloads should be true to regardless
-// of the pipeline
-fn common_series_assertions(series: &SeriesIntake) {
-    // we should have received some metrics from the emitter
-    assert!(!series.is_empty());
-    info!("metric series received: {}", series.len());
-
-    // specifically we should have received each of these
+// checks which of the expected metric types are present in the given series
+// NOTE: no count expected due to the in-app type being Rate
+// (https://docs.datadoghq.com/metrics/types/?tab=count#submission-types-and-datadog-in-app-types)
+fn found_metric_types(series: &SeriesIntake) -> [(bool, &'static str); 4] {
     let mut found = [
-        // NOTE: no count expected due to the in-app type being Rate
-        // (https://docs.datadoghq.com/metrics/types/?tab=count#submission-types-and-datadog-in-app-types)
         (false, "rate"),
         (false, "gauge"),
         (false, "set"),
@@ -105,13 +99,34 @@ fn common_series_assertions(series: &SeriesIntake) {
                 .metric_name
                 .starts_with(&format!("foo_metric.{}", found.1))
             {
-                info!("received {}", found.1);
                 found.0 = true;
             }
         });
     });
 
     found
+}
+
+// whether the series intake looks complete, i.e. every expected metric type
+// has arrived. Used to decide whether it's worth continuing to poll fakeintake.
+//
+// This is tied to the dogstatsd emitter (tests/e2e/datadog-metrics/dogstatsd_client/client.py),
+// which always emits exactly these four metric types once per run. fakeintake has no
+// "wait until this test's data is complete" API of its own, so we have to infer readiness
+// from what the emitter is known to send.
+fn series_intake_is_complete(series: &SeriesIntake) -> bool {
+    !series.is_empty() && found_metric_types(series).iter().all(|(found, _)| *found)
+}
+
+// runs assertions that each set of payloads should be true to regardless
+// of the pipeline
+fn common_series_assertions(series: &SeriesIntake) {
+    // we should have received some metrics from the emitter
+    assert!(!series.is_empty());
+    info!("metric series received: {}", series.len());
+
+    // specifically we should have received each of these
+    found_metric_types(series)
         .iter()
         .for_each(|(found, mtype)| assert!(found, "Didn't receive metric type {}", *mtype));
 }
@@ -180,16 +195,30 @@ fn unpack_v1_series(in_payloads: &[FakeIntakePayloadJson]) -> Vec<DatadogSeriesM
 
 async fn get_v1_series_from_pipeline(address: String) -> SeriesIntake {
     info!("getting v1 series payloads");
-    let payloads =
-        get_fakeintake_payloads::<FakeIntakeResponseJson>(&address, SERIES_ENDPOINT_V1).await;
 
-    info!("unpacking payloads");
-    let payloads = unpack_v1_series(&payloads.payloads);
-    info!("converting payloads");
-    let payloads = convert_v1_payloads_v2(&payloads);
-
-    info!("generating series intake");
-    let intake = generate_series_intake(&payloads);
+    // Retry until we have series data or hit max retries. This is to ensure
+    // events flow through to fakeintake before asking for them.
+    let intake = poll_until(
+        MAX_RETRIES,
+        WAIT_INTERVAL,
+        || async {
+            let payloads = get_fakeintake_payloads::<FakeIntakeResponseJson>(
+                &address,
+                SERIES_ENDPOINT_V1,
+            )
+            .await;
+
+            info!("unpacking payloads");
+            let payloads = unpack_v1_series(&payloads.payloads);
+            info!("converting payloads");
+            let payloads = convert_v1_payloads_v2(&payloads);
+
+            info!("generating series intake");
+            generate_series_intake(&payloads)
+        },
+        |intake: &SeriesIntake| series_intake_is_complete(intake),
+    )
+    .await;
 
     common_series_assertions(&intake);
 
@@ -200,16 +229,30 @@ async fn get_v1_series_from_pipeline(address: String) -> SeriesIntake {
 
 async fn get_v2_series_from_pipeline(address: String) -> SeriesIntake {
     info!("getting v2 series payloads");
-    let payloads =
-        get_fakeintake_payloads::<FakeIntakeResponseRaw>(&address, SERIES_ENDPOINT_V2).await;
-
-    info!("unpacking payloads");
-    let payloads = unpack_proto_payloads::<MetricPayload>(&payloads)
-        .await
-        .expect("Failed to unpack v2 series payloads");
 
-    info!("generating series intake");
-    let intake = generate_series_intake(&payloads);
+    // Retry until we have series data or hit max retries. This is to ensure
+    // events flow through to fakeintake before asking for them.
+    let intake = poll_until(
+        MAX_RETRIES,
+        WAIT_INTERVAL,
+        || async {
+            let payloads = get_fakeintake_payloads::<FakeIntakeResponseRaw>(
+                &address,
+                SERIES_ENDPOINT_V2,
+            )
+            .await;
+
+            info!("unpacking payloads");
+            let payloads = unpack_proto_payloads::<MetricPayload>(&payloads)
+                .await
+                .expect("Failed to unpack v2 series payloads");
+
+            info!("generating series intake");
+            generate_series_intake(&payloads)
+        },
+        |intake: &SeriesIntake| series_intake_is_complete(intake),
+    )
+    .await;
 
     common_series_assertions(&intake);
 

@@ -74,35 +74,60 @@ fn generate_sketch_intake(mut payloads: Vec<SketchPayload>) -> SketchIntake {
     intake
 }
 
+fn has_distribution(sketches: &SketchIntake) -> bool {
+    sketches
+        .keys()
+        .any(|ctx| ctx.metric_name.starts_with("foo_metric.distribution"))
+}
+
+// whether the sketch intake looks complete, i.e. the expected distribution
+// metric has arrived. Used to decide whether it's worth continuing to poll fakeintake.
+//
+// This is tied to the dogstatsd emitter (tests/e2e/datadog-metrics/dogstatsd_client/client.py),
+// which always emits a distribution metric once per run. fakeintake has no "wait until this
+// test's data is complete" API of its own, so we have to infer readiness from what the emitter
+// is known to send.
+fn sketch_intake_is_complete(sketches: &SketchIntake) -> bool {
+    !sketches.is_empty() && has_distribution(sketches)
+}
+
 // runs assertions that each set of payloads should be true to regardless
 // of the pipeline
 fn common_sketch_assertions(sketches: &SketchIntake) {
     // we should have received some metrics from the emitter
     assert!(!sketches.is_empty());
     info!("metric sketch received: {}", sketches.len());
 
-    let mut found = false;
-    sketches.keys().for_each(|ctx| {
-        if ctx.metric_name.starts_with("foo_metric.distribution") {
-            found = true;
-        }
-    });
-
-    assert!(found, "Didn't receive metric type distribution");
+    assert!(
+        has_distribution(sketches),
+        "Didn't receive metric type distribution"
+    );
 }
 
 async fn get_sketches_from_pipeline(address: String) -> SketchIntake {
     info!("getting sketch payloads");
-    let payloads =
-        get_fakeintake_payloads::<FakeIntakeResponseRaw>(&address, SKETCHES_ENDPOINT).await;
-
-    info!("unpacking payloads");
-    let payloads = unpack_proto_payloads(&payloads)
-        .await
-        .expect("Failed to unpack sketch payloads");
 
-    info!("generating sketch intake");
-    let sketches = generate_sketch_intake(payloads);
+    // Retry until we have sketch data or hit max retries. This is to ensure
+    // events flow through to fakeintake before asking for them.
+    let sketches = poll_until(
+        MAX_RETRIES,
+        WAIT_INTERVAL,
+        || async {
+            let payloads =
+                get_fakeintake_payloads::<FakeIntakeResponseRaw>(&address, SKETCHES_ENDPOINT)
+                    .await;
+
+            info!("unpacking payloads");
+            let payloads: Vec<SketchPayload> = unpack_proto_payloads(&payloads)
+                .await
+                .expect("Failed to unpack sketch payloads");
+
+            info!("generating sketch intake");
+            generate_sketch_intake(payloads)
+        },
+        |sketches: &SketchIntake| sketch_intake_is_complete(sketches),
+    )
+    .await;
 
     common_sketch_assertions(&sketches);
 

@@ -1,10 +1,51 @@
 pub mod logs;
 pub mod metrics;
 
+use std::future::Future;
+use std::time::Duration;
+
 use reqwest::{Client, Method};
 use serde::{Deserialize, de::DeserializeOwned};
 use serde_json::Value;
-use tracing::trace;
+use tracing::{trace, warn};
+
+// Fakeintake may not be ready to accept connections yet, particularly right
+// after the compose services start, so transient request failures are retried.
+const MAX_FETCH_ATTEMPTS: usize = 10;
+const FETCH_RETRY_INTERVAL: Duration = Duration::from_secs(1);
+
+// Shared wait between polling attempts for tests that need to retry until
+// expected data has arrived at fakeintake (data itself may take a moment to
+// flow through the pipeline, on top of any transient connection issues
+// already handled by `get_fakeintake_payloads`). How many attempts are needed
+// varies per test, so `max_retries` is left to the caller.
+const WAIT_INTERVAL: Duration = Duration::from_secs(1);
+
+// Calls `fetch` up to `max_retries` times, sleeping `wait` between attempts,
+// until `is_complete` reports the fetched value is ready to use.
+pub(super) async fn poll_until<T, F, Fut, P>(
+    max_retries: usize,
+    wait: Duration,
+    mut fetch: F,
+    mut is_complete: P,
+) -> T
+where
+    F: FnMut() -> Fut,
+    Fut: Future<Output = T>,
+    P: FnMut(&T) -> bool,
+{
+    assert!(max_retries > 0, "max_retries must be greater than zero");
+
+    let mut value = fetch().await;
+    let mut attempt = 1;
+    while attempt < max_retries && !is_complete(&value) {
+        tokio::time::sleep(wait).await;
+        value = fetch().await;
+        attempt += 1;
+    }
+
+    value
+}
 
 fn fake_intake_vector_address() -> String {
     std::env::var("FAKE_INTAKE_VECTOR_ENDPOINT")
@@ -60,19 +101,29 @@ where
     R: FakeIntakeResponseT + DeserializeOwned,
 {
     let url = &R::build_url(base, endpoint);
-    let response = Client::new()
-        .request(Method::GET, url)
-        .send()
-        .await
-        .unwrap_or_else(|_| panic!("Sending GET request to {url} failed"));
-
-    trace!(
-        "Fakeintake response headers for {endpoint}: {:?}",
-        response.headers()
-    );
-
-    response
-        .json::<R>()
-        .await
-        .expect("Parsing fakeintake payloads failed")
+
+    let mut last_error = String::new();
+    for attempt in 1..=MAX_FETCH_ATTEMPTS {
+        match Client::new().request(Method::GET, url).send().await {
+            Ok(response) => {
+                trace!(
+                    "Fakeintake response headers for {endpoint}: {:?}",
+                    response.headers()
+                );
+
+                match response.json::<R>().await {
+                    Ok(parsed) => return parsed,
+                    Err(e) => last_error = format!("Parsing fakeintake payloads failed: {e}"),
+                }
+            }
+            Err(e) => last_error = format!("Sending GET request to {url} failed: {e}"),
+        }
+
+        if attempt < MAX_FETCH_ATTEMPTS {
+            warn!("{last_error}, retrying...");
+            tokio::time::sleep(FETCH_RETRY_INTERVAL).await;
+        }
+    }
+
+    panic!("{last_error} after {MAX_FETCH_ATTEMPTS} attempts");
 }