diff --git a/changelog.d/prometheus_scrape_scrape_delay.feature.md b/changelog.d/prometheus_scrape_scrape_delay.feature.md new file mode 100644 index 0000000000000..7f5f76c643f3b --- /dev/null +++ b/changelog.d/prometheus_scrape_scrape_delay.feature.md @@ -0,0 +1,5 @@ +The `prometheus_scrape` source has a new `scrape_delay` option that controls when scrapes happen relative to the configured `scrape_interval_secs`. Sources sharing an interval otherwise all scrape at the same instant, so an instance running many of them raises load in a single spike each interval. + +`none`, the default, keeps the existing behavior unchanged. A delay in seconds, such as `30s`, holds the first scrape back by that much and then keeps the same fixed cadence, so sources can be staggered by hand. `auto` instead picks a position inside each interval, derived from a hash of the host name, the component ID and the scrape number, which reduces persistent alignment with other periodic work. See the option docs for the trade-offs `auto` makes. + +authors: taloric diff --git a/src/sources/http_client/client.rs b/src/sources/http_client/client.rs index 21812b0f8e5ea..f0a1fac1d019c 100644 --- a/src/sources/http_client/client.rs +++ b/src/sources/http_client/client.rs @@ -374,6 +374,8 @@ impl SourceConfig for HttpClientConfig { let inputs = GenericHttpClientInputs { urls, interval: self.interval, + initial_delay: Duration::ZERO, + jitter_seed: None, timeout: self.timeout, headers: self.headers.clone(), content_type, diff --git a/src/sources/prometheus/scrape.rs b/src/sources/prometheus/scrape.rs index 8f5245229620b..69ec5adc467bc 100644 --- a/src/sources/prometheus/scrape.rs +++ b/src/sources/prometheus/scrape.rs @@ -4,7 +4,7 @@ use bytes::Bytes; use futures_util::FutureExt; use http::{Uri, response::Parts}; use serde_with::serde_as; -use snafu::ResultExt; +use snafu::{ResultExt, Snafu}; use vector_lib::{config::LogNamespace, configurable::configurable_component, event::Event}; use super::parser; @@ -34,6 +34,90 @@ static NOT_FOUND_NO_PATH: &str = "No path is set on the endpoint and we got a 40 did you mean to use /metrics?\ This behavior changed in version 0.11."; +/// Errors returned when a `scrape_delay` value cannot be parsed. +#[derive(Clone, Debug, Eq, PartialEq, Snafu)] +pub(crate) enum ScrapeDelayParseError { + #[snafu(display("A delay in seconds must be a valid integer"))] + SecondsParse, + #[snafu(display("The delay is too large to schedule"))] + DelayTooLarge, + // last case evaluated must explain all valid formats accepted + #[snafu(display("Must be \"none\", \"auto\", or a delay in seconds such as \"30s\""))] + UnableToParse, +} + +/// When scrapes happen, relative to the configured scrape interval. +#[configurable_component] +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +// No `#[serde(untagged)]`: `try_from`/`into` already make this (de)serialize as a plain string, so +// the enum representation is never used, and `Configurable` rejects untagged enums that have more +// than one unit variant. +#[serde(try_from = "String", into = "String")] +#[configurable(metadata(docs::examples = "none"))] +#[configurable(metadata(docs::examples = "auto"))] +#[configurable(metadata(docs::examples = "30s"))] +pub(crate) enum ScrapeDelay { + /// Scrape as soon as the source starts, then once every `scrape_interval_secs` exactly. + #[default] + None, + + /// Scrape once per interval, choosing a position inside each interval independently. + /// + /// Under normal polling, the source starts one scrape round in each interval, but it does not + /// remain at one fixed phase. Intervals missed while the schedule is not polled are skipped + /// rather than replayed. This reduces persistent alignment with other periodic work. Two + /// consecutive scrape starts can be anywhere from nearly zero to nearly two intervals apart, + /// which can increase short-lived overlap and load compared with a fixed cadence. The scheduler + /// does not enforce a minimum gap between starts or place an upper bound on in-flight scrapes. + /// + /// The positions come from a hash of the host name, the component ID and the scrape number + /// rather than from a random number, so the sequence is reproducible relative to source start. + /// Different seeds normally produce different sequences, but individual scrapes can still + /// coincide, and instances with the same host name and component ID use the same sequence. + Auto, + + /// Scrape after exactly this many seconds, written as `30s`. + /// + /// This is the manual counterpart to `auto`: give each source a different value to stagger them + /// by hand. + Fixed(Duration), +} + +impl TryFrom for ScrapeDelay { + type Error = ScrapeDelayParseError; + + fn try_from(input: String) -> std::result::Result { + match input.as_str() { + "none" => Ok(Self::None), + "auto" => Ok(Self::Auto), + s => match s.strip_suffix('s') { + Some(secs) => { + let delay = secs + .parse::() + .map(Duration::from_secs) + .map_err(|_| Self::Error::SecondsParse)?; + + std::time::Instant::now() + .checked_add(delay) + .map(|_| Self::Fixed(delay)) + .ok_or(Self::Error::DelayTooLarge) + } + None => Err(Self::Error::UnableToParse), + }, + } + } +} + +impl From for String { + fn from(delay: ScrapeDelay) -> String { + match delay { + ScrapeDelay::None => "none".to_owned(), + ScrapeDelay::Auto => "auto".to_owned(), + ScrapeDelay::Fixed(delay) => format!("{}s", delay.as_secs()), + } + } +} + /// Configuration for the `prometheus_scrape` source. #[serde_as] #[configurable_component(source( @@ -56,6 +140,31 @@ pub struct PrometheusScrapeConfig { #[configurable(metadata(docs::human_name = "Scrape Interval"))] interval: Duration, + /// When scrapes happen, relative to the configured scrape interval. + /// + /// `none` scrapes as soon as the source starts and then every `scrape_interval_secs` exactly, + /// which means the source raises load at one unvarying period, and sources that share an + /// interval all scrape at the same instant. + /// + /// A delay in seconds, such as `30s`, holds the first scrape back by exactly that much and then + /// keeps the same fixed cadence. Give each source a different value to stagger them by hand. + /// + /// `auto` chooses a position inside each interval independently. Under normal polling, the + /// source starts one scrape round in each interval, but it does not remain at one fixed phase; + /// intervals missed while the schedule is not polled are skipped rather than replayed. This + /// reduces persistent alignment with other periodic work and with sources sharing the same + /// interval. Two consecutive scrape starts can be anywhere from nearly zero to nearly two + /// intervals apart, which can increase short-lived overlap and load compared with a fixed + /// cadence. The scheduler does not enforce a minimum gap between starts or place an upper bound + /// on in-flight scrapes. The positions come from a hash of the host name, the component ID and + /// the scrape number rather than from a random number, so the sequence is reproducible relative + /// to source start. Hash-derived positions do not guarantee distinct slots: individual scrapes + /// can still coincide, and instances with the same host name and component ID use the same + /// sequence. + #[serde(default)] + #[configurable(metadata(docs::human_name = "Scrape Delay"))] + scrape_delay: ScrapeDelay, + /// The timeout for each scrape request. #[serde(default = "default_timeout")] #[serde_as(as = "serde_with:: DurationSecondsWithFrac")] @@ -99,6 +208,23 @@ pub struct PrometheusScrapeConfig { auth: Option, } +impl PrometheusScrapeConfig { + /// The delay applied before the first scrape. + /// + /// Only the first scrape is moved; every scrape after it is driven by the configured interval, + /// so this shifts the phase of the source without ever changing the spacing between samples. + /// + /// `auto` sits with `none` here on purpose. It has no separate initial offset: its first scrape + /// is simply the first of the jittered ones, placed inside the first interval by the schedule + /// itself. + const fn first_scrape_delay(&self) -> Duration { + match self.scrape_delay { + ScrapeDelay::None | ScrapeDelay::Auto => Duration::ZERO, + ScrapeDelay::Fixed(delay) => delay, + } + } +} + fn query_example() -> serde_json::Value { serde_json::json! ({ "match[]": [ @@ -113,6 +239,7 @@ impl GenerateConfig for PrometheusScrapeConfig { serde_json::to_value(Self { endpoints: vec!["http://localhost:9090/metrics".to_string()], interval: default_interval(), + scrape_delay: ScrapeDelay::None, timeout: default_timeout(), instance_tag: Some("instance".to_string()), endpoint_tag: Some("endpoint".to_string()), @@ -145,9 +272,27 @@ impl SourceConfig for PrometheusScrapeConfig { warn_if_interval_too_low(self.timeout, self.interval); + // Picks where inside each interval this source scrapes. The component ID gives sources in + // this Vector instance different deterministic sequences, and the host name does the same + // for instances with different host names. These inputs reduce persistent alignment but do + // not guarantee distinct positions. If the host name cannot be read, only the component ID + // differentiates the sequences. + let delay_seed = format!( + "{}\0{}", + crate::get_hostname().unwrap_or_default(), + cx.key.id() + ); + + let first_scrape_delay = self.first_scrape_delay(); + let inputs = GenericHttpClientInputs { urls, interval: self.interval, + initial_delay: first_scrape_delay, + jitter_seed: match self.scrape_delay { + ScrapeDelay::Auto => Some(delay_seed), + ScrapeDelay::None | ScrapeDelay::Fixed(_) => None, + }, timeout: self.timeout, headers: HashMap::new(), content_type: "text/plain".to_string(), @@ -320,12 +465,12 @@ mod test { service::{make_service_fn, service_fn}, }; use similar_asserts::assert_eq; - use tokio::time::{Duration, sleep}; + use tokio::time::{Duration, sleep, timeout}; use warp::Filter; use super::*; use crate::{ - Error, config, + Error, SourceSender, config, http::{ParameterValue, QueryParameterValue}, sinks::prometheus::exporter::PrometheusExporterConfig, test_util::{ @@ -340,6 +485,137 @@ mod test { crate::test_util::test_generate_config::(); } + #[test] + fn scrape_delay_defaults_to_none() { + let config: PrometheusScrapeConfig = toml::from_str( + r#" + endpoints = ["http://localhost:9090/metrics"] + "#, + ) + .unwrap(); + + assert_eq!(config.scrape_delay, ScrapeDelay::None); + } + + #[test] + fn scrape_delay_parses_every_form() { + let parse = |delay: &str| { + let config: PrometheusScrapeConfig = toml::from_str(&format!( + r#" + endpoints = ["http://localhost:9090/metrics"] + scrape_delay = "{delay}" + "# + )) + .unwrap(); + + config.scrape_delay + }; + + assert_eq!(parse("none"), ScrapeDelay::None); + assert_eq!(parse("auto"), ScrapeDelay::Auto); + assert_eq!(parse("30s"), ScrapeDelay::Fixed(Duration::from_secs(30))); + assert_eq!(parse("0s"), ScrapeDelay::Fixed(Duration::ZERO)); + } + + #[test] + fn scrape_delay_rejects_invalid_values() { + for delay in ["30", "auto+30s", "thirty seconds", "", "s"] { + let parsed = ScrapeDelay::try_from(delay.to_string()); + + assert!( + parsed.is_err(), + "expected `{delay}` to be rejected, got {parsed:?}" + ); + } + + // Well-formed, but too far out to turn into a schedule. + assert_eq!( + ScrapeDelay::try_from(format!("{}s", u64::MAX)), + Err(ScrapeDelayParseError::DelayTooLarge) + ); + } + + #[test] + fn scrape_delay_round_trips_through_a_string() { + for delay in ["none", "auto", "30s"] { + let parsed = ScrapeDelay::try_from(delay.to_string()).unwrap(); + + assert_eq!(String::from(parsed), delay); + } + } + + #[test] + fn only_a_fixed_delay_holds_back_the_first_scrape() { + let first_scrape_delay = |delay| { + let config: PrometheusScrapeConfig = toml::from_str(&format!( + r#" + endpoints = ["http://localhost:9090/metrics"] + scrape_interval_secs = 60 + scrape_delay = "{delay}" + "# + )) + .unwrap(); + + config.first_scrape_delay() + }; + + // A fixed delay is taken literally; the interval has no say in it. + assert_eq!(first_scrape_delay("30s"), Duration::from_secs(30)); + + // `auto` has no initial offset of its own to add. Its first scrape is the first of the + // jittered ones, which the schedule already places somewhere inside the first interval; + // adding a delay on top would only push every scrape out of its interval. + assert_eq!(first_scrape_delay("auto"), Duration::ZERO); + + assert_eq!(first_scrape_delay("none"), Duration::ZERO); + } + + #[tokio::test] + async fn scrape_delay_defers_the_first_request() { + let (_guard, in_addr) = next_addr(); + let (request_tx, mut request_rx) = tokio::sync::mpsc::unbounded_channel(); + let dummy_endpoint = warp::path!("metrics").map(move || { + request_tx.send(()).unwrap(); + "test_metric 1\n" + }); + + tokio::spawn(warp::serve(dummy_endpoint).run(in_addr)); + wait_for_tcp(in_addr).await; + + let source_config = PrometheusScrapeConfig { + endpoints: vec![format!("http://{}/metrics", in_addr)], + interval: Duration::from_secs(10), + scrape_delay: ScrapeDelay::Fixed(Duration::from_secs(2)), + timeout: default_timeout(), + instance_tag: None, + endpoint_tag: None, + honor_labels: false, + query: HashMap::new(), + auth: None, + tls: None, + }; + let (out, _events) = SourceSender::new_test(); + let source = source_config + .build(SourceContext::new_test(out, None)) + .await + .unwrap(); + let source_task = tokio::spawn(source); + + assert!( + timeout(Duration::from_millis(500), request_rx.recv()) + .await + .is_err() + ); + assert_eq!( + timeout(Duration::from_secs(3), request_rx.recv()) + .await + .unwrap(), + Some(()) + ); + + source_task.abort(); + } + #[tokio::test] async fn test_prometheus_sets_headers() { let (_guard, in_addr) = next_addr(); @@ -356,6 +632,7 @@ mod test { let config = PrometheusScrapeConfig { endpoints: vec![format!("http://{}/metrics", in_addr)], interval: Duration::from_secs(1), + scrape_delay: ScrapeDelay::None, timeout: default_timeout(), instance_tag: Some("instance".to_string()), endpoint_tag: Some("endpoint".to_string()), @@ -390,6 +667,7 @@ mod test { let config = PrometheusScrapeConfig { endpoints: vec![format!("http://{}/metrics", in_addr)], interval: Duration::from_secs(1), + scrape_delay: ScrapeDelay::None, timeout: default_timeout(), instance_tag: Some("instance".to_string()), endpoint_tag: Some("endpoint".to_string()), @@ -442,6 +720,7 @@ mod test { let config = PrometheusScrapeConfig { endpoints: vec![format!("http://{}/metrics", in_addr)], interval: Duration::from_secs(1), + scrape_delay: ScrapeDelay::None, timeout: default_timeout(), instance_tag: Some("instance".to_string()), endpoint_tag: Some("endpoint".to_string()), @@ -508,6 +787,7 @@ mod test { let config = PrometheusScrapeConfig { endpoints: vec![format!("http://{}/metrics", in_addr)], interval: Duration::from_secs(1), + scrape_delay: ScrapeDelay::None, timeout: default_timeout(), instance_tag: Some("instance".to_string()), endpoint_tag: Some("endpoint".to_string()), @@ -563,6 +843,7 @@ mod test { let config = PrometheusScrapeConfig { endpoints: vec![format!("http://{}/metrics?key1=val1", in_addr)], interval: Duration::from_secs(1), + scrape_delay: ScrapeDelay::None, timeout: default_timeout(), instance_tag: Some("instance".to_string()), endpoint_tag: Some("endpoint".to_string()), @@ -683,6 +964,7 @@ mod test { honor_labels: false, query: HashMap::new(), interval: Duration::from_secs(1), + scrape_delay: ScrapeDelay::None, timeout: default_timeout(), tls: None, auth: None, @@ -771,6 +1053,7 @@ mod integration_tests { let config = PrometheusScrapeConfig { endpoints: vec!["http://prometheus:9090/metrics".into()], interval: Duration::from_secs(1), + scrape_delay: ScrapeDelay::None, timeout: Duration::from_secs(1), instance_tag: Some("instance".to_string()), endpoint_tag: Some("endpoint".to_string()), diff --git a/src/sources/util/http_client.rs b/src/sources/util/http_client.rs index 4c302ef049c4d..519eb302e9520 100644 --- a/src/sources/util/http_client.rs +++ b/src/sources/util/http_client.rs @@ -40,6 +40,17 @@ pub(crate) struct GenericHttpClientInputs { pub urls: Vec, /// Interval between calls. pub interval: Duration, + /// Delay before the first call. + pub initial_delay: Duration, + /// When set, each call happens at a position inside its own interval rather than on a fixed + /// cadence, and intervals that go by while the stream is not polled are skipped rather than + /// replayed. Different seeds normally produce different deterministic sequences, reducing + /// persistent alignment between components without guaranteeing that individual calls never + /// coincide. + /// + /// `None` keeps `tokio::time::interval` unchanged: a fixed cadence whose missed ticks fire + /// back to back once polling resumes. + pub jitter_seed: Option, /// Timeout for the HTTP request. pub timeout: Duration, /// Map of Header+Value to apply to HTTP request. @@ -62,6 +73,109 @@ pub(crate) const fn default_timeout() -> Duration { Duration::from_secs(5) } +/// Picks the stream that decides when each call happens. +/// +/// Only a caller that asked for jitter gets [`schedule`]. Everything else keeps +/// `tokio::time::interval` verbatim, including its `MissedTickBehavior::Burst` catch-up after a +/// stall, so sources that do not opt in behave exactly as they did before jitter existed. +/// +/// The jitter window is one whole interval, which makes the jittered calls land across the interval +/// rather than near one preferred position. This reduces persistent alignment with periodic work, +/// but does not guarantee that individual calls from different components never coincide. +fn ticks( + start: tokio::time::Instant, + interval: Duration, + jitter_seed: Option, +) -> futures_util::stream::BoxStream<'static, ()> { + match jitter_seed { + Some(seed) => schedule(start, interval, interval, seed).boxed(), + None => IntervalStream::new(tokio::time::interval_at(start, interval)) + .map(|_| ()) + .boxed(), + } +} + +/// Builds the jittered stream that decides when each call happens. +/// +/// Call `n` fires at `start + n * interval`, pushed forward by a jitter of up to `jitter_window`. +/// Every call is anchored to that ideal grid rather than sleeping for `interval + jitter` after the +/// previous one, so the jitter cannot accumulate: with a `jitter_window` of one interval, call `n` +/// always lands in `[start + n * interval, start + (n + 1) * interval)`. Under normal polling this +/// schedules one call in each interval, while the gap between two consecutive calls can fall +/// anywhere in `(0, 2 * interval)`. +/// +/// Grid points that go by without the stream being polled are dropped rather than fired late one +/// after another, so a stall never turns into a burst of catch-up calls. This differs from +/// `tokio::time::interval`, which defaults to `MissedTickBehavior::Burst` and does replay them; +/// callers that need the `tokio` semantics must not use this schedule. +/// +/// A zero `jitter_window` reproduces a plain fixed-cadence interval, still without the catch-up +/// burst. +fn schedule( + start: tokio::time::Instant, + interval: Duration, + jitter_window: Duration, + seed: String, +) -> impl futures_util::Stream + Send { + assert!(!interval.is_zero(), "`interval` must be non-zero."); + + stream::unfold((start, 0u64), move |state| { + // Time can pass between two polls without any call happening: the previous call can run + // long, the runtime can be busy, or the pipeline downstream can be applying backpressure. + // Drop the grid points that went by rather than firing one call for each of them, which + // would turn a stall into exactly the burst this schedule exists to avoid. + let (grid, tick) = skip_missed(state, interval); + let deadline = grid + window_offset(&format!("{seed}\0{tick}"), jitter_window); + let next = (grid + interval, tick.wrapping_add(1)); + + async move { + tokio::time::sleep_until(deadline).await; + Some(((), next)) + } + }) +} + +/// Moves a grid point forward past the calls that were missed while nothing polled the schedule. +/// +/// At most one grid point is left behind the current instant, so recovering from a stall costs a +/// single call that fires straight away instead of one call per interval that went by. +fn skip_missed( + (mut grid, mut tick): (tokio::time::Instant, u64), + interval: Duration, +) -> (tokio::time::Instant, u64) { + if interval.is_zero() { + return (grid, tick); + } + + let now = tokio::time::Instant::now(); + while now.saturating_duration_since(grid) >= interval { + grid += interval; + tick = tick.wrapping_add(1); + } + + (grid, tick) +} + +/// Maps `seed` onto a position in the range `[0, window)`. +/// +/// The position comes from a hash instead of a random number, which spreads different seeds +/// approximately uniformly across the window while keeping the result reproducible: the same seed +/// always lands in the same place, so a schedule built on this can be replayed and tested. +/// +/// Callers are expected to seed this with the host name, the component ID and the call number. The +/// component ID gives components in one Vector instance different sequences, the host name does the +/// same for instances with different host names, and the call number keeps a component from +/// settling onto one position. Hash-derived positions can still coincide, and instances with the +/// same host name and component ID follow the same sequence. +fn window_offset(seed: &str, window: Duration) -> Duration { + let window_ms = u64::try_from(window.as_millis()).unwrap_or(u64::MAX); + if window_ms == 0 { + return Duration::ZERO; + } + + Duration::from_millis(seahash::hash(seed.as_bytes()) % window_ms) +} + /// Builds the context, allowing the source-specific implementation to leverage data from the /// config and the current HTTP request. pub(crate) trait HttpClientBuilder { @@ -160,7 +274,8 @@ pub(crate) async fn call< // proxy and tls settings. let client = HttpClient::new(inputs.tls.clone(), &inputs.proxy).expect("Building HTTP client failed"); - let mut stream = IntervalStream::new(tokio::time::interval(inputs.interval)) + let start = tokio::time::Instant::now() + inputs.initial_delay; + let mut stream = ticks(start, inputs.interval, inputs.jitter_seed) .take_until(inputs.shutdown) .map(move |_| stream::iter(inputs.urls.clone())) .flatten() @@ -307,3 +422,196 @@ pub(crate) async fn call< } } } + +#[cfg(test)] +mod tests { + use std::collections::HashSet; + + use futures_util::StreamExt; + use tokio::time::{Duration, Instant}; + + use super::{schedule, ticks, window_offset}; + + /// Collects when each of the first `count` calls fires, relative to `Instant::now()`. + async fn tick_offsets( + interval: Duration, + jitter_window: Duration, + count: usize, + ) -> Vec { + let started_at = Instant::now(); + let mut ticks = Box::pin(schedule( + started_at, + interval, + jitter_window, + "some-seed".to_owned(), + )); + let mut offsets = Vec::with_capacity(count); + + for _ in 0..count { + ticks.next().await.unwrap(); + offsets.push(Instant::now() - started_at); + } + + offsets + } + + /// Counts how many calls fire without any time passing, i.e. the catch-up burst. + async fn immediate_ticks(stream: &mut futures_util::stream::BoxStream<'static, ()>) -> usize { + let mut immediate = 0; + loop { + let before = Instant::now(); + stream.next().await.unwrap(); + if Instant::now() != before { + return immediate; + } + immediate += 1; + } + } + + #[tokio::test(start_paused = true)] + async fn ticks_without_a_seed_keep_the_tokio_catch_up_burst() { + let interval = Duration::from_secs(10); + let mut stream = ticks(Instant::now(), interval, None); + + stream.next().await.unwrap(); + + // Nothing polls for five intervals, as happens under downstream backpressure. + tokio::time::advance(interval * 5 + Duration::from_secs(1)).await; + + // `tokio::time::interval` defaults to `MissedTickBehavior::Burst`, so every missed tick is + // replayed back to back. Sources that did not opt into jitter must keep seeing exactly + // that: this is the behavior that shipped before the jitter option existed. + assert_eq!(immediate_ticks(&mut stream).await, 5); + } + + #[tokio::test(start_paused = true)] + async fn ticks_with_a_seed_skip_missed_calls() { + let interval = Duration::from_secs(10); + let mut stream = ticks(Instant::now(), interval, Some("some-seed".to_owned())); + + stream.next().await.unwrap(); + tokio::time::advance(interval * 5 + Duration::from_secs(1)).await; + + // The jittered schedule drops the grid points that went by, so recovering costs the single + // call that is due now rather than one call per interval missed. + assert_eq!(immediate_ticks(&mut stream).await, 1); + } + + #[tokio::test(start_paused = true)] + async fn schedule_without_jitter_keeps_a_fixed_cadence() { + let offsets = tick_offsets(Duration::from_secs(10), Duration::ZERO, 4).await; + + assert_eq!( + offsets, + vec![ + Duration::ZERO, + Duration::from_secs(10), + Duration::from_secs(20), + Duration::from_secs(30), + ] + ); + } + + #[test] + #[should_panic(expected = "`interval` must be non-zero.")] + fn schedule_rejects_a_zero_interval() { + let _schedule = schedule( + Instant::now(), + Duration::ZERO, + Duration::ZERO, + String::new(), + ); + } + + #[tokio::test(start_paused = true)] + async fn schedule_starts_from_the_given_instant() { + let started_at = Instant::now(); + let mut ticks = Box::pin(schedule( + started_at + Duration::from_secs(3), + Duration::from_secs(10), + Duration::ZERO, + "some-seed".to_owned(), + )); + + ticks.next().await.unwrap(); + assert_eq!(Instant::now() - started_at, Duration::from_secs(3)); + + ticks.next().await.unwrap(); + assert_eq!(Instant::now() - started_at, Duration::from_secs(13)); + } + + #[tokio::test(start_paused = true)] + async fn schedule_varies_the_gap_between_calls() { + let interval = Duration::from_secs(10); + let offsets = tick_offsets(interval, interval, 12).await; + + let gaps = offsets + .windows(2) + .map(|pair| pair[1] - pair[0]) + .collect::>(); + + // The point of the jitter: consecutive calls must not settle onto one fixed gap. + assert!( + gaps.iter().collect::>().len() > 1, + "expected the gaps between calls to vary, got {gaps:?}" + ); + + // Anchoring to the grid bounds the gap at twice the interval, which is what stops the + // jitter from turning into unbounded drift. + for gap in &gaps { + assert!( + *gap < interval * 2, + "expected every gap to stay under {:?}, got {gap:?}", + interval * 2 + ); + } + } + + #[tokio::test(start_paused = true)] + async fn schedule_does_not_let_jitter_accumulate() { + let interval = Duration::from_secs(10); + let window = interval; + let offsets = tick_offsets(interval, window, 50).await; + + // Each call is anchored to `n * interval` rather than to the previous call, so the schedule + // cannot drift: with uninterrupted polling, after 50 calls every one of them is still + // inside its own interval, giving one call per interval and an average period of + // `interval`. + for (n, offset) in offsets.iter().enumerate() { + let grid = interval * n as u32; + + assert!( + *offset >= grid && *offset < grid + window, + "call {n} should have fired in [{grid:?}, {:?}), got {offset:?}", + grid + window + ); + } + } + + #[test] + fn window_offset_is_zero_without_a_window() { + assert_eq!(window_offset("some-seed", Duration::ZERO), Duration::ZERO); + } + + #[test] + fn window_offset_covers_the_window_without_leaving_it() { + let window = Duration::from_secs(60); + + // Twelve 5-second buckets. Staying inside the window is what keeps a call in its own + // interval; covering every bucket is what stops the sources on a host from bunching up in + // one part of the interval, which is the whole point of picking a position at all. + let mut occupied_buckets = [false; 12]; + + for i in 0..1_000 { + let offset = window_offset(&format!("seed-{i}"), window); + + assert!(offset < window, "seed-{i} landed outside the window"); + occupied_buckets[(offset.as_secs_f64() / 5.0) as usize] = true; + } + + assert!( + occupied_buckets.into_iter().all(|occupied| occupied), + "expected offsets to cover the full window, got occupied buckets {occupied_buckets:?}" + ); + } +} diff --git a/website/cue/reference/components/sources/generated/prometheus_scrape.cue b/website/cue/reference/components/sources/generated/prometheus_scrape.cue index e6685e12af0e5..6e0bdcb8b0437 100644 --- a/website/cue/reference/components/sources/generated/prometheus_scrape.cue +++ b/website/cue/reference/components/sources/generated/prometheus_scrape.cue @@ -260,6 +260,36 @@ generated: components: sources: prometheus_scrape: configuration: { } } } + scrape_delay: { + description: """ + When scrapes happen, relative to the configured scrape interval. + + `none` scrapes as soon as the source starts and then every `scrape_interval_secs` exactly, + which means the source raises load at one unvarying period, and sources that share an + interval all scrape at the same instant. + + A delay in seconds, such as `30s`, holds the first scrape back by exactly that much and then + keeps the same fixed cadence. Give each source a different value to stagger them by hand. + + `auto` chooses a position inside each interval independently. Under normal polling, the + source starts one scrape round in each interval, but it does not remain at one fixed phase; + intervals missed while the schedule is not polled are skipped rather than replayed. This + reduces persistent alignment with other periodic work and with sources sharing the same + interval. Two consecutive scrape starts can be anywhere from nearly zero to nearly two + intervals apart, which can increase short-lived overlap and load compared with a fixed + cadence. The scheduler does not enforce a minimum gap between starts or place an upper bound + on in-flight scrapes. The positions come from a hash of the host name, the component ID and + the scrape number rather than from a random number, so the sequence is reproducible relative + to source start. Hash-derived positions do not guarantee distinct slots: individual scrapes + can still coincide, and instances with the same host name and component ID use the same + sequence. + """ + required: false + type: string: { + default: "none" + examples: ["none", "auto", "30s"] + } + } scrape_interval_secs: { description: """ The interval between scrapes. Requests are run concurrently so if a scrape takes longer