diff --git a/asap-tools/data-sources/paced-feeder/README.md b/asap-tools/data-sources/paced-feeder/README.md index 55a0ed82..73238787 100644 --- a/asap-tools/data-sources/paced-feeder/README.md +++ b/asap-tools/data-sources/paced-feeder/README.md @@ -53,12 +53,18 @@ credentials. ## Pacing and stats -Each tick's rows are split into `--batch-rows` requests and sent with up to -`--concurrency` in flight. For remote write, a tick sends its -`--samples-per-sec` sub-second slots in turn, and the next slot starts only -after every request of the current one returns, so each aggregation group's -samples reach ASAPQuery in timestamp order. ClickHouse sorts on insert, so it -gets the whole tick at once. +Each tick's rows are split into requests of about `--batch-rows` rows and +sent with up to `--concurrency` in flight. A tick's rows are series-major and +requests are cut only between series, so each series' samples of a tick go +in one request, in timestamp order; the next tick starts only after every +request of the current one returns. Each series' samples therefore arrive in +order (counters never seem to reset, and Prometheus needs no out-of-order +window), while samples of different series of a group may arrive out of +order within a tick. A tick starting at `t` holds samples spaced evenly over +`(t, t + 1 s]`, the same left-open interval as ASAPQuery's one-second +windows, so a tick never spans two windows, and that reordering is safe for +ASAPQuery as long as its allowed lateness is at least 1 s (the default is +5 s). `--samples-per-sec` is at most 1000, so timestamps (in ms) never repeat. The default batch size is 50,000 rows for remote write, which keeps a compressed request under axum's 2 MB body limit in ASAPQuery's ingest, and 100,000 for diff --git a/asap-tools/data-sources/paced-feeder/src/encode.rs b/asap-tools/data-sources/paced-feeder/src/encode.rs index 32a4caba..880c2202 100644 --- a/asap-tools/data-sources/paced-feeder/src/encode.rs +++ b/asap-tools/data-sources/paced-feeder/src/encode.rs @@ -157,7 +157,7 @@ mod tests { #[test] fn remote_write_decodes_to_one_series_per_row() { let w = workload(); - let rows = w.slot_rows(2, 0, 1_700_000_000_000); + let rows = w.tick_rows(2, 1_700_000_000_000); let body = RemoteWriteEncoder::new(&w, "data").encode(&rows); let raw = snap::raw::Decoder::new().decompress_vec(&body).unwrap(); @@ -191,7 +191,7 @@ mod tests { #[test] fn row_binary_round_trips_every_column() { let w = workload(); - let rows = w.slot_rows(0, 0, 42_000); + let rows = w.tick_rows(0, 42_000); let buf = RowBinaryEncoder::new(&w).encode(&rows); let mut cursor = buf.as_slice(); diff --git a/asap-tools/data-sources/paced-feeder/src/main.rs b/asap-tools/data-sources/paced-feeder/src/main.rs index 42db84de..498067ac 100644 --- a/asap-tools/data-sources/paced-feeder/src/main.rs +++ b/asap-tools/data-sources/paced-feeder/src/main.rs @@ -16,7 +16,7 @@ use serde::Serialize; use std::sync::Arc; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use tokio::sync::Semaphore; -use workload::{Row, Workload}; +use workload::{series_batches, Workload}; #[derive(Debug, Clone, Copy, ValueEnum, Serialize)] #[serde(rename_all = "kebab-case")] @@ -71,9 +71,10 @@ struct Args { #[arg(long)] start_ms: Option, - /// Rows per HTTP request. Defaults to 50,000 for remote write, which keeps - /// a compressed request under axum's 2 MB body limit in ASAPQuery's ingest, - /// and 100,000 for ClickHouse, where each INSERT creates a part. + /// Rows per HTTP request, rounded down to whole series (at least one). + /// Defaults to 50,000 for remote write, which keeps a compressed request + /// under axum's 2 MB body limit in ASAPQuery's ingest, and 100,000 for + /// ClickHouse, where each INSERT creates a part. #[arg(long)] batch_rows: Option, /// Concurrent requests within one tick. @@ -179,6 +180,15 @@ impl Sink { } } +/// A tick's samples lie in one 1s window only if ticks start on whole seconds; +/// a tick straddling a boundary can close a window before its other series arrive. +fn validate_start_ms(start_ms: i64) -> Result<()> { + if start_ms % 1_000 != 0 { + bail!("--start-ms must be a whole second (multiple of 1000), got {start_ms}"); + } + Ok(()) +} + fn next_whole_second_ms() -> i64 { let now_ms = SystemTime::now() .duration_since(UNIX_EPOCH) @@ -229,6 +239,9 @@ async fn main() -> Result<()> { if batch_rows == 0 || args.concurrency == 0 { bail!("--batch-rows and --concurrency must be positive"); } + if let Some(start_ms) = args.start_ms { + validate_start_ms(start_ms)?; + } let workload = Workload { seed: args.seed, @@ -263,43 +276,38 @@ async fn main() -> Result<()> { let started = Instant::now(); let (mut rows, mut bytes) = (0, 0); - // ASAPQuery can drop samples behind a group's watermark, so remote write - // sends one slot at a time: a slot's requests all return before the - // next slot's are sent. ClickHouse sorts on insert, so it sends the - // whole tick at once. - let slots = (0..args.samples_per_sec).map(|slot| workload.slot_rows(tick, slot, start_ms)); - let waves: Vec> = match args.sink { - SinkKind::RemoteWrite => slots.collect(), - SinkKind::Clickhouse => vec![slots.flatten().collect()], - }; - for wave in waves { - let wave_rows = Arc::new(wave); - let mut tasks = tokio::task::JoinSet::new(); - for begin in (0..wave_rows.len()).step_by(batch_rows) { - let end = (begin + batch_rows).min(wave_rows.len()); - let (wave_rows, sink) = (Arc::clone(&wave_rows), Arc::clone(&sink)); - let permit = Arc::clone(&permits).acquire_owned().await?; - tasks.spawn(async move { - let _permit = permit; - let body = sink.encoder.encode(&wave_rows[begin..end]); - let bytes = body.len(); - sink.send(body).await.map(|()| (end - begin, bytes)) - }); - } + // A tick's requests go out concurrently, each holding whole series + // (see `series_batches`): a series' samples arrive in order, but a + // group's samples across series may not. That is safe for ASAPQuery: + // the tick lies in one window (see `tick_rows`) and is under one + // second behind the watermark, within the allowed lateness. + let tick_rows = Arc::new(workload.tick_rows(tick, start_ms)); + let per_series = args.samples_per_sec as usize; + let mut tasks = tokio::task::JoinSet::new(); + for range in series_batches(tick_rows.len(), per_series, batch_rows) { + let (tick_rows, sink) = (Arc::clone(&tick_rows), Arc::clone(&sink)); + let permit = Arc::clone(&permits).acquire_owned().await?; + tasks.spawn(async move { + let _permit = permit; + let rows = range.len(); + let body = sink.encoder.encode(&tick_rows[range]); + let bytes = body.len(); + sink.send(body).await.map(|()| (rows, bytes)) + }); + } - while let Some(joined) = tasks.join_next().await { - // A panicked task counts as a failed request, so the stats of - // the ticks before it are still written. - match joined.map_err(anyhow::Error::from).and_then(|sent| sent) { - Ok((n_rows, n_bytes)) => { - rows += n_rows; - bytes += n_bytes; - } - Err(e) => { - failed_requests += 1; - if errors.len() < 10 { - errors.push(format!("tick {tick}: {e:#}")); - } + while let Some(joined) = tasks.join_next().await { + // A panicked task counts as a failed request, so the stats of + // the ticks before it are still written. + match joined.map_err(anyhow::Error::from).and_then(|sent| sent) { + Ok((n_rows, n_bytes)) => { + rows += n_rows; + bytes += n_bytes; + } + Err(e) => { + failed_requests += 1; + if errors.len() < 10 { + errors.push(format!("tick {tick}: {e:#}")); } } } @@ -374,3 +382,14 @@ async fn main() -> Result<()> { } Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn start_ms_must_be_whole_second() { + assert!(validate_start_ms(1_700_000_000_000).is_ok()); + assert!(validate_start_ms(1_700_000_000_500).is_err()); + } +} diff --git a/asap-tools/data-sources/paced-feeder/src/workload.rs b/asap-tools/data-sources/paced-feeder/src/workload.rs index 5e68ea90..22e650b0 100644 --- a/asap-tools/data-sources/paced-feeder/src/workload.rs +++ b/asap-tools/data-sources/paced-feeder/src/workload.rs @@ -40,19 +40,21 @@ impl Workload { (format!("g{group}"), format!("i{instance}")) } - /// Samples for sub-second slot `slot` of one-second tick `tick`, one per - /// series, with event time starting at `start_ms`. Slots within the second - /// are spaced evenly. - pub fn slot_rows(&self, tick: u64, slot: u32, start_ms: i64) -> Vec { + /// The samples of one-second tick `tick`, series-major: each series' + /// `samples_per_sec` samples in timestamp order, then the next series'. + /// Slot `j` of `n` is at `t + (j + 1) ยท 1 s / n`, where `t` is the tick's + /// start, so the tick lies in `(t, t + 1 s]`: the same left-open interval as + /// ASAPQuery's one-second windows, so a tick never spans two of them. + pub fn tick_rows(&self, tick: u64, start_ms: i64) -> Vec { let per_sec = u64::from(self.samples_per_sec); - let slot = u64::from(slot); - let timestamp_ms = start_ms + tick as i64 * 1_000 + (slot * 1_000 / per_sec) as i64; - let sample_index = tick * per_sec + slot; + let tick_ms = start_ms + tick as i64 * 1_000; (0..self.num_series()) - .map(|series| Row { - series, - timestamp_ms, - value: self.value(series, sample_index), + .flat_map(|series| { + (0..per_sec).map(move |slot| Row { + series, + timestamp_ms: tick_ms + ((slot + 1) * 1_000 / per_sec) as i64, + value: self.value(series, tick * per_sec + slot), + }) }) .collect() } @@ -65,6 +67,23 @@ impl Workload { } } +/// Row ranges of the requests for `rows` series-major rows with +/// `per_series` rows per series: about `batch_rows` each, cut only between +/// series (at least one series per request), so every series' samples of a +/// tick reach the sink in one request, in timestamp order. Concurrent requests +/// then reorder samples only across series, never within one, which keeps +/// counters monotonic for consumers that read them in arrival order. +pub fn series_batches( + rows: usize, + per_series: usize, + batch_rows: usize, +) -> impl Iterator> { + let batch = (batch_rows / per_series).max(1) * per_series; + (0..rows) + .step_by(batch) + .map(move |begin| begin..(begin + batch).min(rows)) +} + /// Counter-based uniform draw in (0, 1]: a stateless hash of the inputs, so /// any sample can be regenerated independently of the ones before it. fn unit_uniform(seed: u64, series: u64, sample_index: u64) -> f64 { @@ -96,37 +115,59 @@ mod tests { } #[test] - fn each_slot_has_every_series_at_its_timestamp() { + fn a_tick_is_series_major_in_timestamp_order() { let w = workload(); - for (slot, timestamp_ms) in [(0, 15_000), (1, 15_500)] { - let rows = w.slot_rows(5, slot, 10_000); - assert_eq!(rows.len(), 6); - assert!(rows.iter().all(|r| r.timestamp_ms == timestamp_ms)); - assert!(rows.iter().map(|r| r.series).eq(0..6)); + let rows = w.tick_rows(5, 10_000); + assert_eq!(rows.len(), 12); + for (series, samples) in rows.chunks(2).enumerate() { + assert!(samples.iter().all(|r| r.series == series)); + let timestamps: Vec = samples.iter().map(|r| r.timestamp_ms).collect(); + assert_eq!(timestamps, [15_500, 16_000]); + } + } + + #[test] + fn batches_keep_each_series_in_one_request_in_order() { + let w = Workload { + samples_per_sec: 3, + ..workload() + }; + let rows = w.tick_rows(2, 0); + // 7 rows round down to 2 series (6 rows); 1 row still takes a series. + for (batch_rows, sizes) in [(7, vec![6, 6, 6]), (1, vec![3; 6]), (100, vec![18])] { + let batches: Vec<_> = series_batches(rows.len(), 3, batch_rows).collect(); + assert_eq!(batches.iter().map(|b| b.len()).collect::>(), sizes); + for series in 0..w.num_series() { + let holding: Vec<_> = batches + .iter() + .filter(|b| rows[(*b).clone()].iter().any(|r| r.series == series)) + .collect(); + assert_eq!(holding.len(), 1, "series {series}"); + let timestamps: Vec = rows[holding[0].clone()] + .iter() + .filter(|r| r.series == series) + .map(|r| r.timestamp_ms) + .collect(); + assert_eq!(timestamps.len(), 3); + assert!(timestamps.windows(2).all(|t| t[0] < t[1])); + } } } #[test] fn values_depend_only_on_seed_series_and_sample_index() { let w = workload(); - // Same slot, different start time: same values. - let a: Vec = w.slot_rows(4, 1, 0).iter().map(|r| r.value).collect(); - let b: Vec = w - .slot_rows(4, 1, 1_000_000) - .iter() - .map(|r| r.value) - .collect(); - assert_eq!(a, b); + let values = |w: &Workload, start_ms| -> Vec { + w.tick_rows(4, start_ms).iter().map(|r| r.value).collect() + }; + // Same tick, different start time: same values. + assert_eq!(values(&w, 0), values(&w, 1_000_000)); // A different seed changes them. - let other = Workload { seed: 8, ..w }; - assert_ne!( - a, - other - .slot_rows(4, 1, 0) - .iter() - .map(|r| r.value) - .collect::>() - ); + let other = Workload { + seed: 8, + ..w.clone() + }; + assert_ne!(values(&w, 0), values(&other, 0)); } #[test]