diff --git a/tests/e2e/datadog/logs/mod.rs b/tests/e2e/datadog/logs/mod.rs index 3221a6a6f3982..0b890d072eefc 100644 --- a/tests/e2e/datadog/logs/mod.rs +++ b/tests/e2e/datadog/logs/mod.rs @@ -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::( - &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::(&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!( diff --git a/tests/e2e/datadog/metrics/mod.rs b/tests/e2e/datadog/metrics/mod.rs index 68912a88d6890..fa0b64999a668 100644 --- a/tests/e2e/datadog/metrics/mod.rs +++ b/tests/e2e/datadog/metrics/mod.rs @@ -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> { 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; diff --git a/tests/e2e/datadog/metrics/series.rs b/tests/e2e/datadog/metrics/series.rs index 2c4e20a7ab269..c9ca2247cf0f9 100644 --- a/tests/e2e/datadog/metrics/series.rs +++ b/tests/e2e/datadog/metrics/series.rs @@ -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 SeriesIntake { info!("getting v1 series payloads"); - let payloads = - get_fakeintake_payloads::(&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::( + &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::(&address, SERIES_ENDPOINT_V2).await; - - info!("unpacking payloads"); - let payloads = unpack_proto_payloads::(&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::( + &address, + SERIES_ENDPOINT_V2, + ) + .await; + + info!("unpacking payloads"); + let payloads = unpack_proto_payloads::(&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); diff --git a/tests/e2e/datadog/metrics/sketches.rs b/tests/e2e/datadog/metrics/sketches.rs index 0a0012f825e6c..1c0bffefc32a3 100644 --- a/tests/e2e/datadog/metrics/sketches.rs +++ b/tests/e2e/datadog/metrics/sketches.rs @@ -74,6 +74,23 @@ fn generate_sketch_intake(mut payloads: Vec) -> 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) { @@ -81,28 +98,36 @@ fn common_sketch_assertions(sketches: &SketchIntake) { 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::(&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::(&address, SKETCHES_ENDPOINT) + .await; + + info!("unpacking payloads"); + let payloads: Vec = 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); diff --git a/tests/e2e/datadog/mod.rs b/tests/e2e/datadog/mod.rs index 42dd240a52688..3e4c9d52bcec9 100644 --- a/tests/e2e/datadog/mod.rs +++ b/tests/e2e/datadog/mod.rs @@ -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( + max_retries: usize, + wait: Duration, + mut fetch: F, + mut is_complete: P, +) -> T +where + F: FnMut() -> Fut, + Fut: Future, + 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::() - .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::().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"); }