From a1944f0d70fed7b8ae772d1ec5038b4202a97458 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 13 Oct 2026 01:39:54 +0000 Subject: [PATCH 1/3] fix(asap-tools): send paced feeder ticks concurrently, with samples in (t, t+1s] Remote write sent one sub-second slot at a time, so at most one request was in flight per slot. At 200 samples/s per series ASAPQuery's ingest took about 45 ms per 1e4-row request and the feeder reached only 0.2-0.27M rows/s. Each tick now goes out concurrently. Its samples sit in (t, t+1s], the same left-open interval as ASAPQuery's one-second windows, so the reordering within a tick never crosses a window boundary and stays within the allowed lateness. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01AnJ7LocWxguh3q4QWkCYae --- .../data-sources/paced-feeder/README.md | 13 ++-- .../data-sources/paced-feeder/src/main.rs | 69 +++++++++---------- .../data-sources/paced-feeder/src/workload.rs | 9 +-- 3 files changed, 46 insertions(+), 45 deletions(-) diff --git a/asap-tools/data-sources/paced-feeder/README.md b/asap-tools/data-sources/paced-feeder/README.md index 55a0ed82..0ff1d954 100644 --- a/asap-tools/data-sources/paced-feeder/README.md +++ b/asap-tools/data-sources/paced-feeder/README.md @@ -54,11 +54,14 @@ 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. +`--concurrency` in flight. The next tick starts only after every request of +the current one returns, so samples arrive out of order only 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. Out-of-order samples within it are then safe for +ASAPQuery as long as its allowed lateness is at least 1 s (the default is +5 s). Prometheus' remote-write receiver needs an out-of-order window of at +least 1 s (`storage.tsdb.out_of_order_time_window`). `--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/main.rs b/asap-tools/data-sources/paced-feeder/src/main.rs index 42db84de..8d1ae688 100644 --- a/asap-tools/data-sources/paced-feeder/src/main.rs +++ b/asap-tools/data-sources/paced-feeder/src/main.rs @@ -263,43 +263,40 @@ 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, so a group's samples within + // the tick may arrive out of order. That is safe for ASAPQuery: the + // tick lies in one window (see `slot_rows`) and is under one second + // behind the watermark, within the allowed lateness. + let tick_rows: Arc> = Arc::new( + (0..args.samples_per_sec) + .flat_map(|slot| workload.slot_rows(tick, slot, start_ms)) + .collect(), + ); + let mut tasks = tokio::task::JoinSet::new(); + for begin in (0..tick_rows.len()).step_by(batch_rows) { + let end = (begin + batch_rows).min(tick_rows.len()); + 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 body = sink.encoder.encode(&tick_rows[begin..end]); + let bytes = body.len(); + sink.send(body).await.map(|()| (end - begin, 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:#}")); } } } diff --git a/asap-tools/data-sources/paced-feeder/src/workload.rs b/asap-tools/data-sources/paced-feeder/src/workload.rs index 5e68ea90..3b93561e 100644 --- a/asap-tools/data-sources/paced-feeder/src/workload.rs +++ b/asap-tools/data-sources/paced-feeder/src/workload.rs @@ -41,12 +41,13 @@ impl Workload { } /// 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. + /// series. Slots are spaced evenly over `(t, t + 1 s]`, where `t` is the + /// tick's start: the same left-open interval as ASAPQuery's one-second + /// windows, so a tick never spans two of them. pub fn slot_rows(&self, tick: u64, slot: u32, 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 timestamp_ms = start_ms + tick as i64 * 1_000 + ((slot + 1) * 1_000 / per_sec) as i64; let sample_index = tick * per_sec + slot; (0..self.num_series()) .map(|series| Row { @@ -98,7 +99,7 @@ mod tests { #[test] fn each_slot_has_every_series_at_its_timestamp() { let w = workload(); - for (slot, timestamp_ms) in [(0, 15_000), (1, 15_500)] { + for (slot, timestamp_ms) in [(0, 15_500), (1, 16_000)] { let rows = w.slot_rows(5, slot, 10_000); assert_eq!(rows.len(), 6); assert!(rows.iter().all(|r| r.timestamp_ms == timestamp_ms)); From 660a6db21b22fc3a15f2c17e7a854514c684a04a Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 13 Oct 2026 03:09:47 +0000 Subject: [PATCH 2/3] fix(asap-tools): keep each series' samples of a tick in one ordered request Concurrent requests split a tick by time slot, so a series' later sample could arrive first and ASAPQuery's increase accumulator, which reads samples in arrival order, took the lower counter value for a reset. A tick's rows are now series-major and requests are cut only between series (the batch size rounded down to whole series, at least one), so each series' samples of a tick arrive in one request in timestamp order; requests stay concurrent. Prometheus therefore needs no out-of-order window. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01AnJ7LocWxguh3q4QWkCYae --- .../data-sources/paced-feeder/README.md | 19 +-- .../data-sources/paced-feeder/src/encode.rs | 4 +- .../data-sources/paced-feeder/src/main.rs | 33 +++--- .../data-sources/paced-feeder/src/workload.rs | 110 ++++++++++++------ 4 files changed, 104 insertions(+), 62 deletions(-) diff --git a/asap-tools/data-sources/paced-feeder/README.md b/asap-tools/data-sources/paced-feeder/README.md index 0ff1d954..73238787 100644 --- a/asap-tools/data-sources/paced-feeder/README.md +++ b/asap-tools/data-sources/paced-feeder/README.md @@ -53,15 +53,18 @@ credentials. ## Pacing and stats -Each tick's rows are split into `--batch-rows` requests and sent with up to -`--concurrency` in flight. The next tick starts only after every request of -the current one returns, so samples arrive out of order only 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. Out-of-order samples within it are then safe for +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). Prometheus' remote-write receiver needs an out-of-order window of at -least 1 s (`storage.tsdb.out_of_order_time_window`). +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 8d1ae688..184d85e7 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. @@ -263,25 +264,23 @@ async fn main() -> Result<()> { let started = Instant::now(); let (mut rows, mut bytes) = (0, 0); - // A tick's requests go out concurrently, so a group's samples within - // the tick may arrive out of order. That is safe for ASAPQuery: the - // tick lies in one window (see `slot_rows`) and is under one second - // behind the watermark, within the allowed lateness. - let tick_rows: Arc> = Arc::new( - (0..args.samples_per_sec) - .flat_map(|slot| workload.slot_rows(tick, slot, start_ms)) - .collect(), - ); + // 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 begin in (0..tick_rows.len()).step_by(batch_rows) { - let end = (begin + batch_rows).min(tick_rows.len()); + 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 body = sink.encoder.encode(&tick_rows[begin..end]); + let rows = range.len(); + let body = sink.encoder.encode(&tick_rows[range]); let bytes = body.len(); - sink.send(body).await.map(|()| (end - begin, bytes)) + sink.send(body).await.map(|()| (rows, bytes)) }); } diff --git a/asap-tools/data-sources/paced-feeder/src/workload.rs b/asap-tools/data-sources/paced-feeder/src/workload.rs index 3b93561e..22e650b0 100644 --- a/asap-tools/data-sources/paced-feeder/src/workload.rs +++ b/asap-tools/data-sources/paced-feeder/src/workload.rs @@ -40,20 +40,21 @@ impl Workload { (format!("g{group}"), format!("i{instance}")) } - /// Samples for sub-second slot `slot` of one-second tick `tick`, one per - /// series. Slots are spaced evenly over `(t, t + 1 s]`, where `t` is the - /// tick's start: the same left-open interval as ASAPQuery's one-second - /// windows, so a tick never spans two of them. - 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) * 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() } @@ -66,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 { @@ -97,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_500), (1, 16_000)] { - 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] From c5842b57f2f9446427b70bda3a21641a3afc5e60 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Tue, 13 Oct 2026 23:02:37 -0400 Subject: [PATCH 3/3] fix(asap-tools): reject non-whole-second --start-ms in paced feeder A tick straddling a 1s boundary can close a window before its other series' samples arrive. Co-Authored-By: Claude Opus 5.5 --- .../data-sources/paced-feeder/src/main.rs | 23 +++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/asap-tools/data-sources/paced-feeder/src/main.rs b/asap-tools/data-sources/paced-feeder/src/main.rs index 184d85e7..498067ac 100644 --- a/asap-tools/data-sources/paced-feeder/src/main.rs +++ b/asap-tools/data-sources/paced-feeder/src/main.rs @@ -180,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) @@ -230,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, @@ -370,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()); + } +}