From a1944f0d70fed7b8ae772d1ec5038b4202a97458 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 13 Oct 2026 01:39:54 +0000 Subject: [PATCH 1/5] 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 a2196250b526ebe24fe3b52f69a24db6c5501fcc Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 13 Oct 2026 01:10:52 +0000 Subject: [PATCH 2/5] feat(asap-tools): generate the paced feeder's data from a workload table `--workload-table FILE` reads `workloads[0].e2e.metrics` of a sketch-bench workload table and sends its gauge and counter metrics (or only `--metric`): labels with `child_of` fan out under their parent, the others form a cross product; series get Zipf scales from a seeded rank shuffle; gauges are scaled Pareto draws and counters grow by their scale per sample. `--verify-facts` checks, for every label subset, the distinct label-value tuples of the generated series against the description. The `--groups` workload sends the same bytes as before. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01AnJ7LocWxguh3q4QWkCYae --- .../data-sources/paced-feeder/README.md | 95 +++- .../data-sources/paced-feeder/src/encode.rs | 204 +++++-- .../data-sources/paced-feeder/src/main.rs | 290 ++++++---- .../data-sources/paced-feeder/src/table.rs | 499 ++++++++++++++++++ .../data-sources/paced-feeder/src/workload.rs | 221 +++++--- 5 files changed, 1087 insertions(+), 222 deletions(-) create mode 100644 asap-tools/data-sources/paced-feeder/src/table.rs diff --git a/asap-tools/data-sources/paced-feeder/README.md b/asap-tools/data-sources/paced-feeder/README.md index 0ff1d954..c7362364 100644 --- a/asap-tools/data-sources/paced-feeder/README.md +++ b/asap-tools/data-sources/paced-feeder/README.md @@ -8,17 +8,79 @@ one second of event time per wall-clock second, to either: It drives the data-plane vs ClickHouse evaluation (`asap-tools/experiments/DATA_PLANE_CLICKHOUSE_EVAL_PLAN.md`). Two feeders -started with the same workload flags send identical data, so each system gets +started with the same workload flags (and table) send identical data, so each system gets the same stream without a dataset file on disk. ## Workload +The feeder generates either the fixed `--groups` workload or the metrics of a +workload table (`--workload-table`). Either way, a value depends only on +`(--seed, metric, series, sample index)`, never on wall-clock time or the sink, +and each series emits `--samples-per-sec` samples per second, evenly spaced. + +### `--groups` workload + - `--groups C` values of `label_0` (`g0`, `g1`, …) and `--series-per-group s` - values of `instance` (`i0`, `i1`, …) per group: `C·s` series. -- Each series emits `--samples-per-sec` samples per second, evenly spaced. + values of `instance` (`i0`, `i1`, …) per group: `C·s` series of one gauge, + named by `--metric` (default `data`). - Values are Pareto(`--pareto-shape`, `--pareto-scale`), all ≥ the scale. -- A value depends only on `(--seed, series, sample index)`, never on wall-clock - time or the sink. +- `--samples-per-sec` defaults to 1. + +### `--workload-table` workload + +`--workload-table FILE` reads `workloads[0].e2e.metrics` of a sketch-bench +workload table (`scripts/export_autosketch_eval_table.py`) and sends every +metric in it, or only `--metric NAME`. For example: + +```json +"data": {"type": "gauge", + "labels": [{"name": "job", "cardinality": 10}, + {"name": "label_0", "cardinality": 1000, "child_of": "job"}], + "samples_per_sec_per_series": 200, + "values": {"value": {"zipf": 1.1, "keys": 10000}, "dist": {"pareto": 2.0}}}, +"data_total": {"type": "counter", "labels": […], "samples_per_sec_per_series": 200, + "values": {"increment": {"zipf": 1.1, "keys": 10000}}} +``` + +- **Labels.** A label has `cardinality` values, or, with `child_of`, that many + under each value of its parent, which must be listed earlier. A root label's + values are `0`, `1`, …; a child's are its parent's value, a `.`, and the + child index (`job="3"`, `label_0="3.17"`), so a child value names its parent + and is unique across parents. Labels without `child_of` form a cross + product. The metric has one series per combination: the product of all + cardinalities (1e4 above). +- **Key skew.** `zipf` θ over `keys` keys, where a key is a series (`keys` + must equal the number of series). The series of Zipf rank `r` (0 the + heaviest) has scale `N·(r+1)^-θ / Σ_j (j+1)^-θ`, so scales average 1. Ranks + are a shuffle seeded by `--seed` alone, so the heavy series spread across + parents (not all under `job="0"`) and a series is equally heavy in every + metric of the table. +- **Gauge** (`value` and `dist`): sample `n` of series `i` is + `scale_i · U^(-1/a)` with `U` a hashed uniform in (0, 1]: Pareto(a, 1) + scaled by the series' weight, which keeps the tail index `a`. +- **Counter** (`increment`): sample `n` of series `i` is `(n+1) · scale_i`, + i.e. a constant increment `scale_i` per sample, starting at the first + sample and never resetting, so `rate()` is `scale_i · samples/s`. +- `--samples-per-sec` overrides `samples_per_sec_per_series`, e.g. to scale a + run down. Without it, the sent metrics must share one rate. +- Other metric types (the multi-grouping `records` metrics) are rejected. + +`--verify-facts` checks the generated series against the description without +sending anything: for every non-empty subset of a metric's labels it prints +the expected number of distinct label-value tuples (the product of the +cardinalities of the subset plus every label's `child_of` ancestors) and the +number among the generated series, and exits non-zero if any differ. + +```sh +$ paced_feeder --workload-table synthetic-templatesclassic-shared1.json --verify-facts +data: 10000 series + labels expected generated + job 10 10 + label_0 10000 10000 + job,label_0 10000 10000 +data_total: 10000 series + … +``` Event time starts at the next whole wall-clock second (or `--start-ms`), so `NOW()`-relative queries see the data as it arrives. @@ -39,15 +101,36 @@ target/release/paced_feeder --sink clickhouse \ --url http://localhost:8123 --table data \ --groups 1000 --series-per-group 100 --duration-secs 900 \ --stats-out clickhouse-feed.json + +# A workload table's metrics (data and data_total) to ASAPQuery +target/release/paced_feeder --sink remote-write \ + --url http://localhost:9090/api/v1/write \ + --workload-table synthetic-templatesclassic-shared1.json --duration-secs 900 ``` -The ClickHouse table must have these columns (any engine and ordering): +Remote write sends each metric's name as `__name__` and its labels sorted by +name, as Prometheus requires; it also works against Prometheus' +remote-write receiver (`--web.enable-remote-write-receiver`). + +With `--groups`, the ClickHouse table (`--table`) must have these columns (any +engine and ordering): ```sql CREATE TABLE data (ts DateTime64(3), label_0 String, instance String, value Float64) ENGINE = MergeTree ORDER BY (label_0, instance, ts); ``` +With `--workload-table`, each metric goes to the table named after it, with +columns `ts`, then the metric's labels in the description's order, then +`value`. For the example above: + +```sql +CREATE TABLE data (ts DateTime64(3), job String, label_0 String, value Float64) +ENGINE = MergeTree ORDER BY (job, label_0, ts); +CREATE TABLE data_total (ts DateTime64(3), job String, label_0 String, value Float64) +ENGINE = MergeTree ORDER BY (job, label_0, ts); +``` + `CLICKHOUSE_USER` / `CLICKHOUSE_PASSWORD` (or the matching flags) set credentials. diff --git a/asap-tools/data-sources/paced-feeder/src/encode.rs b/asap-tools/data-sources/paced-feeder/src/encode.rs index 32a4caba..bb6249b2 100644 --- a/asap-tools/data-sources/paced-feeder/src/encode.rs +++ b/asap-tools/data-sources/paced-feeder/src/encode.rs @@ -1,7 +1,7 @@ //! Wire encoders for the two sinks. Each series' labels are encoded once up //! front, so the per-sample cost at high rates is just the value and timestamp. -use crate::workload::{Row, Workload}; +use crate::workload::{Metric, Row}; use prost::encoding::{encode_key, encode_varint, encoded_len_varint, WireType}; pub trait Encoder: Send + Sync { @@ -12,21 +12,27 @@ pub trait Encoder: Send + Sync { /// `TimeSeries` (labels + one sample) per row. pub struct RemoteWriteEncoder { /// Encoded `Label` fields (`TimeSeries` field 1) for each series, sorted by - /// label name as Prometheus requires: `__name__`, `instance`, `label_0`. + /// label name as Prometheus requires, `__name__` included. series_labels: Vec>, } impl RemoteWriteEncoder { - pub fn new(workload: &Workload, metric: &str) -> Self { - let series_labels = (0..workload.num_series()) - .map(|series| { - let (label_0, instance) = workload.series_labels(series); + /// Series of `metric`, sent with `__name__` = `name`. + pub fn new(metric: &Metric, name: &str) -> Self { + let series_labels = metric + .series + .iter() + .map(|values| { + let mut labels: Vec<(&str, &str)> = metric + .labels + .iter() + .map(|l| l.name.as_str()) + .zip(values.iter().map(String::as_str)) + .collect(); + labels.push(("__name__", name)); + labels.sort_unstable(); let mut buf = Vec::new(); - for (name, value) in [ - ("__name__", metric), - ("instance", instance.as_str()), - ("label_0", label_0.as_str()), - ] { + for (name, value) in labels { let mut label = Vec::new(); encode_bytes_field(1, name.as_bytes(), &mut label); encode_bytes_field(2, value.as_bytes(), &mut label); @@ -63,23 +69,31 @@ impl Encoder for RemoteWriteEncoder { } } -/// ClickHouse `RowBinary` for columns `(ts DateTime64(3), label_0 String, -/// instance String, value Float64)`. +/// ClickHouse `RowBinary` for columns `ts DateTime64(3)`, then each label of +/// the metric as `String` in its order, then `value Float64`. pub struct RowBinaryEncoder { - /// `label_0` then `instance`, each as a varint length plus bytes. + /// Each series' label values, each as a varint length plus bytes. series_labels: Vec>, } -/// The column list `RowBinaryEncoder` writes, in order. -pub const ROW_BINARY_COLUMNS: &str = "ts, label_0, instance, value"; +/// The column list `RowBinaryEncoder` writes for `metric`, in order. +pub fn row_binary_columns(metric: &Metric) -> String { + let labels = metric.labels.iter().map(|l| l.name.as_str()); + std::iter::once("ts") + .chain(labels) + .chain(std::iter::once("value")) + .collect::>() + .join(", ") +} impl RowBinaryEncoder { - pub fn new(workload: &Workload) -> Self { - let series_labels = (0..workload.num_series()) - .map(|series| { - let (label_0, instance) = workload.series_labels(series); + pub fn new(metric: &Metric) -> Self { + let series_labels = metric + .series + .iter() + .map(|values| { let mut buf = Vec::new(); - for s in [label_0, instance] { + for s in values { encode_varint(s.len() as u64, &mut buf); buf.extend_from_slice(s.as_bytes()); } @@ -112,6 +126,8 @@ fn encode_bytes_field(tag: u32, bytes: &[u8], buf: &mut Vec) { #[cfg(test)] mod tests { use super::*; + use crate::table::series_labels; + use crate::workload::{Kind, LabelSpec, Metric, Workload}; use prost::Message; // Mirrors the remote write types in @@ -143,28 +159,52 @@ mod tests { timestamp: i64, } - fn workload() -> Workload { + fn grouped() -> Workload { + Workload::grouped("data", 1, 2, 3, 1, 1.5, 1.0) + } + + /// Labels out of name order, an uppercase one sorting before `__name__`. + fn multi_label() -> Workload { + let label = |name: &str, cardinality, child_of: Option<&str>| LabelSpec { + name: name.to_string(), + cardinality, + child_of: child_of.map(str::to_string), + }; + let labels = vec![ + label("zone", 2, None), + label("job", 2, None), + label("label_0", 3, Some("job")), + label("Az", 1, None), + ]; + let series = series_labels(&labels).unwrap(); + let metric = Metric { + name: "data_total".to_string(), + scales: vec![1.0; series.len()], + labels, + series, + kind: Kind::Counter, + seed: 0, + }; Workload { - seed: 1, - groups: 2, - series_per_group: 3, samples_per_sec: 1, - pareto_shape: 1.5, - pareto_scale: 1.0, + metrics: vec![metric], } } + fn decode_remote_write(body: &[u8]) -> WriteRequest { + let raw = snap::raw::Decoder::new().decompress_vec(body).unwrap(); + WriteRequest::decode(raw.as_slice()).unwrap() + } + #[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 body = RemoteWriteEncoder::new(&w, "data").encode(&rows); - - let raw = snap::raw::Decoder::new().decompress_vec(&body).unwrap(); - let request = WriteRequest::decode(raw.as_slice()).unwrap(); + let w = grouped(); + let rows = w.slot_rows(0, 2, 0, 1_700_000_000_000); + let request = + decode_remote_write(&RemoteWriteEncoder::new(&w.metrics[0], "data").encode(&rows)); assert_eq!(request.timeseries.len(), rows.len()); for (ts, row) in request.timeseries.iter().zip(&rows) { - let (label_0, instance) = w.series_labels(row.series); + let values = &w.metrics[0].series[row.series]; let labels: Vec<(&str, &str)> = ts .labels .iter() @@ -174,8 +214,8 @@ mod tests { labels, vec![ ("__name__", "data"), - ("instance", instance.as_str()), - ("label_0", label_0.as_str()), + ("instance", values[1].as_str()), + ("label_0", values[0].as_str()), ] ); assert_eq!( @@ -189,29 +229,93 @@ mod tests { } #[test] - fn row_binary_round_trips_every_column() { - let w = workload(); - let rows = w.slot_rows(0, 0, 42_000); - let buf = RowBinaryEncoder::new(&w).encode(&rows); + fn remote_write_sorts_labels_by_name() { + let w = multi_label(); + let metric = &w.metrics[0]; + let rows = w.slot_rows(0, 0, 0, 1_000); + let request = + decode_remote_write(&RemoteWriteEncoder::new(metric, &metric.name).encode(&rows)); + assert_eq!(request.timeseries.len(), 12); + for (ts, row) in request.timeseries.iter().zip(&rows) { + let [zone, job, label_0, az] = metric.series[row.series].as_slice() else { + panic!("four labels"); + }; + let labels: Vec<(&str, &str)> = ts + .labels + .iter() + .map(|l| (l.name.as_str(), l.value.as_str())) + .collect(); + assert_eq!( + labels, + vec![ + ("Az", az.as_str()), + ("__name__", "data_total"), + ("job", job.as_str()), + ("label_0", label_0.as_str()), + ("zone", zone.as_str()), + ] + ); + } + } - let mut cursor = buf.as_slice(); + /// Decodes RowBinary rows of `ts`, `num_labels` strings and `value`. + fn decode_row_binary(buf: &[u8], num_labels: usize) -> Vec<(i64, Vec, f64)> { + let mut cursor = buf; let mut take = |n: usize| -> Vec { let (head, rest) = cursor.split_at(n); cursor = rest; head.to_vec() }; - for row in &rows { + let mut rows = Vec::new(); + let mut remaining = buf.len(); + while remaining > 0 { let ts = i64::from_le_bytes(take(8).try_into().unwrap()); // Labels are short, so their varint length is one byte. - let label_0_len = take(1)[0] as usize; - let label_0 = String::from_utf8(take(label_0_len)).unwrap(); - let instance_len = take(1)[0] as usize; - let instance = String::from_utf8(take(instance_len)).unwrap(); + let labels: Vec = (0..num_labels) + .map(|_| { + let len = take(1)[0] as usize; + String::from_utf8(take(len)).unwrap() + }) + .collect(); let value = f64::from_le_bytes(take(8).try_into().unwrap()); - assert_eq!(ts, row.timestamp_ms); - assert_eq!((label_0, instance), w.series_labels(row.series)); - assert_eq!(value, row.value); + remaining -= 16 + labels.iter().map(|l| 1 + l.len()).sum::(); + rows.push((ts, labels, value)); } - assert!(cursor.is_empty()); + rows + } + + #[test] + fn row_binary_round_trips_every_column() { + let w = grouped(); + let metric = &w.metrics[0]; + assert_eq!(row_binary_columns(metric), "ts, label_0, instance, value"); + let rows = w.slot_rows(0, 0, 0, 42_000); + let decoded = decode_row_binary(&RowBinaryEncoder::new(metric).encode(&rows), 2); + let expected: Vec<_> = rows + .iter() + .map(|r| (r.timestamp_ms, metric.series[r.series].clone(), r.value)) + .collect(); + assert_eq!(decoded, expected); + } + + #[test] + fn row_binary_writes_labels_in_description_order() { + let w = multi_label(); + let metric = &w.metrics[0]; + assert_eq!( + row_binary_columns(metric), + "ts, zone, job, label_0, Az, value" + ); + let rows = w.slot_rows(0, 3, 0, 0); + let decoded = decode_row_binary(&RowBinaryEncoder::new(metric).encode(&rows), 4); + assert_eq!(decoded.len(), 12); + assert_eq!( + decoded[4], + ( + 4_000, + vec!["0".into(), "1".into(), "1.1".into(), "0".into()], + 4.0 + ) + ); } } diff --git a/asap-tools/data-sources/paced-feeder/src/main.rs b/asap-tools/data-sources/paced-feeder/src/main.rs index 8d1ae688..f7e1efff 100644 --- a/asap-tools/data-sources/paced-feeder/src/main.rs +++ b/asap-tools/data-sources/paced-feeder/src/main.rs @@ -7,11 +7,12 @@ //! is written to `--stats-out` so a run can be checked for keeping up. mod encode; +mod table; mod workload; use anyhow::{bail, Context, Result}; use clap::{Parser, ValueEnum}; -use encode::{Encoder, RemoteWriteEncoder, RowBinaryEncoder, ROW_BINARY_COLUMNS}; +use encode::{row_binary_columns, Encoder, RemoteWriteEncoder, RowBinaryEncoder}; use serde::Serialize; use std::sync::Arc; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; @@ -30,42 +31,55 @@ enum SinkKind { #[derive(Debug, Parser)] #[command(about = "Send a seeded synthetic workload in real time to ASAPQuery or ClickHouse")] struct Args { - #[arg(long, value_enum)] - sink: SinkKind, + #[arg(long, value_enum, required_unless_present = "verify_facts")] + sink: Option, /// Remote write endpoint (`http://host:port/api/v1/write`) or ClickHouse /// HTTP base URL (`http://host:8123`). + #[arg(long, required_unless_present = "verify_facts")] + url: Option, + /// With `--groups`: the metric name sent as `__name__` (remote write + /// only; default `data`). With `--workload-table`: send only this metric. #[arg(long)] - url: String, - /// Metric name sent as `__name__` (remote write only). - #[arg(long, default_value = "data")] - metric: String, - /// Target table, optionally `database.table` (ClickHouse only), made of - /// `[A-Za-z0-9_]` characters. Must have columns `ts DateTime64(3), label_0 String, instance String, value Float64`. - #[arg(long, default_value = "data")] + metric: Option, + /// Target table, optionally `database.table` (ClickHouse with `--groups` + /// only), made of `[A-Za-z0-9_]` characters. Must have columns `ts DateTime64(3), label_0 String, instance String, value Float64`. + #[arg(long, default_value = "data", conflicts_with = "workload_table")] table: String, #[arg(long, env = "CLICKHOUSE_USER")] clickhouse_user: Option, #[arg(long, env = "CLICKHOUSE_PASSWORD")] clickhouse_password: Option, + /// Generate the metrics of a sketch-bench workload table's + /// `workloads[0].e2e.metrics` instead of the `--groups` workload. Each + /// metric's ClickHouse table is named after it. + #[arg(long, conflicts_with_all = ["groups", "series_per_group", "pareto_shape", "pareto_scale"])] + workload_table: Option, /// Number of `label_0` values (C). - #[arg(long)] - groups: usize, + #[arg(long, required_unless_present = "workload_table")] + groups: Option, /// Number of `instance` values per group (s). + #[arg(long, required_unless_present = "workload_table")] + series_per_group: Option, + /// Samples per second per series. Defaults to 1 with `--groups`, and to + /// the table's `samples_per_sec_per_series` with `--workload-table`. #[arg(long)] - series_per_group: usize, - #[arg(long, default_value_t = 1)] - samples_per_sec: u32, + samples_per_sec: Option, #[arg(long, default_value_t = 1.5)] pareto_shape: f64, #[arg(long, default_value_t = 1.0)] pareto_scale: f64, #[arg(long, default_value_t = 42)] seed: u64, + /// Print, for every subset of each metric's labels, the expected and the + /// generated number of distinct label-value tuples, then exit without + /// sending; non-zero if any differ. + #[arg(long)] + verify_facts: bool, /// Seconds of event time to send. - #[arg(long)] - duration_secs: u64, + #[arg(long, required_unless_present = "verify_facts")] + duration_secs: Option, /// Event time of the first tick, in ms since the epoch. Defaults to the /// next whole wall-clock second, so `NOW()`-relative queries see live data. #[arg(long)] @@ -98,11 +112,14 @@ struct TickStats { #[derive(Debug, Serialize)] struct RunStats { sink: SinkKind, - groups: usize, - series_per_group: usize, + /// Unset with `--workload-table`. + groups: Option, + series_per_group: Option, + pareto_shape: Option, + pareto_scale: Option, + workload_table: Option, + metrics: Vec, samples_per_sec: u32, - pareto_shape: f64, - pareto_scale: f64, seed: u64, start_ms: i64, duration_secs: u64, @@ -120,50 +137,67 @@ struct RunStats { ticks: Vec, } -struct Sink { - client: reqwest::Client, +/// Where one metric's rows go: the request to send them with and their encoder. +struct Target { request: reqwest::RequestBuilder, encoder: Box, } +struct Sink { + client: reqwest::Client, + /// One per metric of the workload, in order. + targets: Vec, +} + impl Sink { - fn new(args: &Args, workload: &Workload) -> Result { + fn new(args: &Args, sink: SinkKind, url: &str, workload: &Workload) -> Result { let client = reqwest::Client::builder() .pool_max_idle_per_host(args.concurrency) .build()?; - let (request, encoder): (_, Box) = match args.sink { - SinkKind::RemoteWrite => ( - client - .post(&args.url) - .header("Content-Encoding", "snappy") - .header("Content-Type", "application/x-protobuf") - .header("X-Prometheus-Remote-Write-Version", "0.1.0"), - Box::new(RemoteWriteEncoder::new(workload, &args.metric)), - ), - SinkKind::Clickhouse => { - let query = format!( - "INSERT INTO {} ({ROW_BINARY_COLUMNS}) FORMAT RowBinary", - args.table - ); - let mut request = client.post(&args.url).query(&[("query", query)]); - if let Some(user) = &args.clickhouse_user { - request = request.header("X-ClickHouse-User", user); - } - if let Some(password) = &args.clickhouse_password { - request = request.header("X-ClickHouse-Key", password); + let targets = workload + .metrics + .iter() + .map(|metric| match sink { + SinkKind::RemoteWrite => Target { + request: client + .post(url) + .header("Content-Encoding", "snappy") + .header("Content-Type", "application/x-protobuf") + .header("X-Prometheus-Remote-Write-Version", "0.1.0"), + encoder: Box::new(RemoteWriteEncoder::new(metric, &metric.name)), + }, + SinkKind::Clickhouse => { + let table = if args.workload_table.is_some() { + &metric.name + } else { + &args.table + }; + let query = format!( + "INSERT INTO {table} ({}) FORMAT RowBinary", + row_binary_columns(metric) + ); + let mut request = client.post(url).query(&[("query", query)]); + if let Some(user) = &args.clickhouse_user { + request = request.header("X-ClickHouse-User", user); + } + if let Some(password) = &args.clickhouse_password { + request = request.header("X-ClickHouse-Key", password); + } + Target { + request, + encoder: Box::new(RowBinaryEncoder::new(metric)), + } } - (request, Box::new(RowBinaryEncoder::new(workload))) - } - }; - Ok(Self { - client, - request, - encoder, - }) + }) + .collect(); + Ok(Self { client, targets }) } - async fn send(&self, body: Vec) -> Result<()> { - let request = self + async fn send(&self, metric: usize, rows: &[Row]) -> Result { + let target = &self.targets[metric]; + let body = target.encoder.encode(rows); + let bytes = body.len(); + let request = target .request .try_clone() .context("request builder must be clonable")? @@ -175,7 +209,7 @@ impl Sink { let text = response.text().await.unwrap_or_default(); bail!("HTTP {status}: {}", text.trim()); } - Ok(()) + Ok(bytes) } } @@ -199,17 +233,81 @@ async fn wait_for_wall_clock(epoch_ms: i64) -> Instant { Instant::now() } +fn build_workload(args: &Args) -> Result { + if let Some(path) = &args.workload_table { + return table::load( + path, + args.metric.as_deref(), + args.samples_per_sec, + args.seed, + ); + } + let (groups, series_per_group) = ( + args.groups.unwrap_or_default(), + args.series_per_group.unwrap_or_default(), + ); + if groups == 0 || series_per_group == 0 { + bail!("--groups and --series-per-group must be positive"); + } + if args.pareto_shape <= 0.0 || args.pareto_scale <= 0.0 { + bail!("--pareto-shape and --pareto-scale must be positive"); + } + Ok(Workload::grouped( + args.metric.as_deref().unwrap_or("data"), + args.seed, + groups, + series_per_group, + args.samples_per_sec.unwrap_or(1), + args.pareto_shape, + args.pareto_scale, + )) +} + +/// Prints the facts check of every metric; true if every fact holds. +fn verify_facts(workload: &Workload) -> bool { + let mut ok = true; + for metric in &workload.metrics { + println!("{}: {} series", metric.name, metric.series.len()); + println!(" {:<40} {:>12} {:>12}", "labels", "expected", "generated"); + for fact in table::facts(&metric.labels, &metric.series) { + let holds = fact.expected == fact.actual; + ok &= holds; + println!( + " {:<40} {:>12} {:>12}{}", + fact.labels.join(","), + fact.expected, + fact.actual, + if holds { "" } else { " MISMATCH" } + ); + } + } + ok +} + #[tokio::main] async fn main() -> Result<()> { let args = Args::parse(); - if args.groups == 0 || args.series_per_group == 0 || args.samples_per_sec == 0 { - bail!("--groups, --series-per-group and --samples-per-sec must be positive"); + let workload = build_workload(&args)?; + if workload.samples_per_sec == 0 { + bail!("samples per second must be positive"); } // Timestamps are in ms, so more samples per second would repeat a timestamp // within a series, which remote write drops and ClickHouse keeps. - if args.samples_per_sec > 1_000 { - bail!("--samples-per-sec must be at most 1000"); + if workload.samples_per_sec > 1_000 { + bail!("samples per second must be at most 1000"); + } + if args.verify_facts { + if !verify_facts(&workload) { + bail!("the generated series do not match the description"); + } + return Ok(()); } + let (Some(sink_kind), Some(url), Some(duration_secs)) = + (args.sink, args.url.as_deref(), args.duration_secs) + else { + bail!("--sink, --url and --duration-secs are required"); + }; + // The table name is interpolated into the INSERT query. if args.table.is_empty() || !args @@ -219,10 +317,7 @@ async fn main() -> Result<()> { { bail!("--table must be made of [A-Za-z0-9_.] characters"); } - if args.pareto_shape <= 0.0 || args.pareto_scale <= 0.0 { - bail!("--pareto-shape and --pareto-scale must be positive"); - } - let batch_rows = args.batch_rows.unwrap_or(match args.sink { + let batch_rows = args.batch_rows.unwrap_or(match sink_kind { SinkKind::RemoteWrite => 50_000, SinkKind::Clickhouse => 100_000, }); @@ -230,34 +325,25 @@ async fn main() -> Result<()> { bail!("--batch-rows and --concurrency must be positive"); } - let workload = Workload { - seed: args.seed, - groups: args.groups, - series_per_group: args.series_per_group, - samples_per_sec: args.samples_per_sec, - pareto_shape: args.pareto_shape, - pareto_scale: args.pareto_scale, - }; - let sink = Arc::new(Sink::new(&args, &workload)?); + let sink = Arc::new(Sink::new(&args, sink_kind, url, &workload)?); let permits = Arc::new(Semaphore::new(args.concurrency)); let start_ms = args.start_ms.unwrap_or_else(next_whole_second_ms); + let metric_names: Vec = workload.metrics.iter().map(|m| m.name.clone()).collect(); eprintln!( - "paced_feeder: {:?} -> {}, {} series x {}/s = {} rows/s for {}s, start_ms={start_ms}", - args.sink, - args.url, + "paced_feeder: {sink_kind:?} -> {url}, {} ({} series) x {}/s = {} rows/s for {duration_secs}s, start_ms={start_ms}", + metric_names.join(", "), workload.num_series(), - args.samples_per_sec, + workload.samples_per_sec, workload.rows_per_tick(), - args.duration_secs ); let base = wait_for_wall_clock(start_ms).await; - let mut ticks = Vec::with_capacity(args.duration_secs as usize); + let mut ticks = Vec::with_capacity(duration_secs as usize); let mut errors: Vec = Vec::new(); let mut failed_requests = 0; - for tick in 0..args.duration_secs { + for tick in 0..duration_secs { let scheduled = base + Duration::from_secs(tick); tokio::time::sleep_until(scheduled.into()).await; let started = Instant::now(); @@ -267,22 +353,23 @@ async fn main() -> Result<()> { // 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)) - }); + for metric in 0..workload.metrics.len() { + let metric_rows: Arc> = Arc::new( + (0..workload.samples_per_sec) + .flat_map(|slot| workload.slot_rows(metric, tick, slot, start_ms)) + .collect(), + ); + for begin in (0..metric_rows.len()).step_by(batch_rows) { + let end = (begin + batch_rows).min(metric_rows.len()); + let (metric_rows, sink) = (Arc::clone(&metric_rows), Arc::clone(&sink)); + let permit = Arc::clone(&permits).acquire_owned().await?; + tasks.spawn(async move { + let _permit = permit; + let sent = sink.send(metric, &metric_rows[begin..end]).await; + sent.map(|bytes| (end - begin, bytes)) + }); + } } while let Some(joined) = tasks.join_next().await { @@ -326,18 +413,21 @@ async fn main() -> Result<()> { // A run that keeps up still spans the full duration, even though the last // tick's work ends before its second does. - let wall_secs = base.elapsed().as_secs_f64().max(args.duration_secs as f64); + let wall_secs = base.elapsed().as_secs_f64().max(duration_secs as f64); let rows_sent: usize = ticks.iter().map(|t| t.rows).sum(); + let grouped = args.workload_table.is_none(); let run = RunStats { - sink: args.sink, + sink: sink_kind, groups: args.groups, series_per_group: args.series_per_group, - samples_per_sec: args.samples_per_sec, - pareto_shape: args.pareto_shape, - pareto_scale: args.pareto_scale, + pareto_shape: grouped.then_some(args.pareto_shape), + pareto_scale: grouped.then_some(args.pareto_scale), + workload_table: args.workload_table.clone(), + metrics: metric_names, + samples_per_sec: workload.samples_per_sec, seed: args.seed, start_ms, - duration_secs: args.duration_secs, + duration_secs, target_rows_per_sec: workload.rows_per_tick(), rows_sent, bytes_sent: ticks.iter().map(|t| t.bytes).sum(), diff --git a/asap-tools/data-sources/paced-feeder/src/table.rs b/asap-tools/data-sources/paced-feeder/src/table.rs new file mode 100644 index 00000000..e6f73329 --- /dev/null +++ b/asap-tools/data-sources/paced-feeder/src/table.rs @@ -0,0 +1,499 @@ +//! Metrics from a sketch-bench workload table's `workloads[0].e2e.metrics`, +//! and the facts check that the generated series match their description. +//! +//! A label has `cardinality` values, or, with `child_of`, that many under each +//! value of its parent. A root label's values are `0`, `1`, …; a child's are +//! its parent's value, a `.`, and the child index (`job="3"`, +//! `label_0="3.17"`), so a child value names its parent and is unique across +//! parents. Labels without `child_of` form a cross product, and the metric +//! has the product of all cardinalities as series. + +use crate::workload::{splitmix64, Kind, LabelSpec, Metric, Workload}; +use anyhow::{bail, ensure, Context, Result}; +use serde::Deserialize; +use serde_json::Value; +use std::collections::{BTreeMap, HashSet}; + +#[derive(Deserialize)] +struct Table { + workloads: Vec, +} + +#[derive(Deserialize)] +struct TableWorkload { + e2e: E2e, +} + +#[derive(Deserialize)] +struct E2e { + metrics: BTreeMap, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct MetricSpec { + #[serde(rename = "type")] + kind: String, + labels: Vec, + samples_per_sec_per_series: u32, + values: Value, +} + +/// Zipf(`zipf`) skew over `keys` keys; a key is a series. +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct KeySkew { + zipf: f64, + keys: u64, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct GaugeValues { + value: KeySkew, + dist: ParetoDist, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct ParetoDist { + pareto: f64, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct CounterValues { + increment: KeySkew, +} + +/// The metrics of `workloads[0].e2e.metrics` in the table at `path`, in name +/// order, or only `only`. `samples_per_sec` overrides every metric's rate. +pub fn load( + path: &str, + only: Option<&str>, + samples_per_sec: Option, + seed: u64, +) -> Result { + let text = std::fs::read_to_string(path).with_context(|| format!("reading {path}"))?; + let table: Table = serde_json::from_str(&text) + .with_context(|| format!("{path}: no workloads[0].e2e.metrics"))?; + let Some(first) = table.workloads.into_iter().next() else { + bail!("{path}: `workloads` is empty"); + }; + let mut metrics = Vec::new(); + let mut rates = Vec::new(); + for (name, spec) in &first.e2e.metrics { + if only.is_some_and(|only| only != name) { + continue; + } + let (metric, rate) = + metric(name, spec, seed).with_context(|| format!("metric `{name}`"))?; + metrics.push(metric); + rates.push(rate); + } + if let Some(only) = only { + ensure!(!metrics.is_empty(), "{path}: no metric `{only}`"); + } + ensure!( + !metrics.is_empty(), + "{path}: workloads[0].e2e.metrics is empty" + ); + let samples_per_sec = match samples_per_sec { + Some(rate) => rate, + // One feeder sends every metric at one rate. + None if rates.iter().all(|&r| r == rates[0]) => rates[0], + None => bail!( + "metrics have different samples_per_sec_per_series ({rates:?}); \ + send them with separate feeders (--metric) or set --samples-per-sec" + ), + }; + Ok(Workload { + samples_per_sec, + metrics, + }) +} + +fn metric(name: &str, spec: &Value, seed: u64) -> Result<(Metric, u32)> { + ensure!( + is_identifier(name), + "the name must match [A-Za-z_][A-Za-z0-9_]* (it is also the ClickHouse table)" + ); + let kind = spec.get("type").and_then(Value::as_str).unwrap_or_default(); + if kind != "gauge" && kind != "counter" { + bail!("type `{kind}` is not supported; the feeder generates `gauge` and `counter` metrics"); + } + let spec: MetricSpec = serde_json::from_value(spec.clone())?; + let series = series_labels(&spec.labels)?; + let (skew, kind) = if spec.kind == "gauge" { + let values: GaugeValues = serde_json::from_value(spec.values).context("gauge `values`")?; + ensure!( + values.dist.pareto > 0.0, + "the Pareto shape must be positive" + ); + let kind = Kind::Gauge { + pareto_shape: values.dist.pareto, + pareto_scale: 1.0, + }; + (values.value, kind) + } else { + let values: CounterValues = + serde_json::from_value(spec.values).context("counter `values`")?; + (values.increment, Kind::Counter) + }; + ensure!( + skew.keys == series.len() as u64, + "Zipf keys ({}) must equal the number of series ({}): each series is a key", + skew.keys, + series.len() + ); + ensure!(skew.zipf >= 0.0, "the Zipf skew must be non-negative"); + // The ranks depend on the seed alone, so a key is equally heavy in every + // metric of the table. + let scales = zipf_scales(series.len(), skew.zipf, seed); + let metric = Metric { + name: name.to_string(), + labels: spec.labels, + series, + scales, + kind, + seed: splitmix64(seed ^ fnv1a(name)), + }; + Ok((metric, spec.samples_per_sec_per_series)) +} + +/// Each series' label values, in `labels` order. Series are numbered in mixed +/// radix with the last label varying fastest. +pub fn series_labels(labels: &[LabelSpec]) -> Result>> { + ensure!(!labels.is_empty(), "a metric needs at least one label"); + // Index of each label's parent, which must come earlier. + let mut parents = Vec::with_capacity(labels.len()); + for (i, label) in labels.iter().enumerate() { + let name = &label.name; + ensure!( + is_identifier(name) && !name.starts_with("__"), + "label `{name}` must match [A-Za-z_][A-Za-z0-9_]* and not start with `__`" + ); + ensure!( + name != "ts" && name != "value", + "label `{name}` clashes with a ClickHouse column" + ); + ensure!( + labels[..i].iter().all(|l| &l.name != name), + "label `{name}` is repeated" + ); + ensure!(label.cardinality > 0, "label `{name}` has cardinality 0"); + parents.push(match &label.child_of { + None => None, + Some(parent) => Some( + labels[..i] + .iter() + .position(|l| &l.name == parent) + .with_context(|| { + format!("label `{name}`: child_of `{parent}` must name an earlier label") + })?, + ), + }); + } + let num_series = labels + .iter() + .try_fold(1usize, |n, l| { + n.checked_mul(usize::try_from(l.cardinality).ok()?) + }) + .context("too many series")?; + + let mut series = Vec::with_capacity(num_series); + let mut index = vec![0u64; labels.len()]; + for _ in 0..num_series { + let mut values: Vec = Vec::with_capacity(labels.len()); + for (i, parent) in parents.iter().enumerate() { + values.push(match parent { + None => index[i].to_string(), + Some(p) => format!("{}.{}", values[*p], index[i]), + }); + } + series.push(values); + // Advance the mixed-radix index, last label fastest. + for i in (0..labels.len()).rev() { + index[i] += 1; + if index[i] < labels[i].cardinality { + break; + } + index[i] = 0; + } + } + Ok(series) +} + +/// Each series' value scale for Zipf(`theta`) skew over the series: the +/// series of rank `r` (0 is the heaviest) gets `(r + 1)^-theta`, normalized so +/// the scales average 1. Ranks are a seeded shuffle of the series, so the +/// heavy series spread across parent labels instead of filling the first one. +pub fn zipf_scales(num_series: usize, theta: f64, seed: u64) -> Vec { + let weights: Vec = (0..num_series) + .map(|r| ((r + 1) as f64).powf(-theta)) + .collect(); + let norm = num_series as f64 / weights.iter().sum::(); + zipf_ranks(num_series, seed) + .into_iter() + .map(|r| weights[r] * norm) + .collect() +} + +/// A seeded permutation: the Zipf rank of each series (Fisher-Yates). +pub fn zipf_ranks(num_series: usize, seed: u64) -> Vec { + let mut ranks: Vec = (0..num_series).collect(); + let mut state = splitmix64(seed ^ fnv1a("zipf ranks")); + for i in (1..num_series).rev() { + state = splitmix64(state); + ranks.swap(i, (state % (i as u64 + 1)) as usize); + } + ranks +} + +/// One row of the facts check: the distinct value tuples of a label subset. +#[derive(Debug, Clone, PartialEq)] +pub struct Fact { + pub labels: Vec, + /// The product of the fan-outs of `labels` closed under `child_of`. + pub expected: u64, + /// Distinct tuples among the generated series. + pub actual: u64, +} + +/// For every non-empty subset of `labels`, the expected and the generated +/// number of distinct label-value tuples. +pub fn facts(labels: &[LabelSpec], series: &[Vec]) -> Vec { + let n = labels.len(); + assert!(n < 32, "too many labels to enumerate their subsets"); + (1u32..1 << n) + .map(|mask| { + let subset: Vec = (0..n).filter(|i| mask & (1 << i) != 0).collect(); + let actual = series + .iter() + .map(|values| { + subset + .iter() + .map(|&i| values[i].as_str()) + .collect::>() + }) + .collect::>() + .len() as u64; + Fact { + labels: subset.iter().map(|&i| labels[i].name.clone()).collect(), + expected: closed_cardinality(labels, &subset), + actual, + } + }) + .collect() +} + +/// The product of the cardinalities of `subset` and every ancestor of its +/// labels: a child value names its parent, so grouping by a child groups by +/// its ancestors too. +fn closed_cardinality(labels: &[LabelSpec], subset: &[usize]) -> u64 { + let mut closed = HashSet::new(); + let mut todo: Vec = subset.to_vec(); + while let Some(i) = todo.pop() { + if closed.insert(i) { + if let Some(parent) = &labels[i].child_of { + todo.extend(labels.iter().position(|l| &l.name == parent)); + } + } + } + closed + .iter() + .map(|&i| labels[i].cardinality) + .fold(1u64, u64::saturating_mul) +} + +fn is_identifier(name: &str) -> bool { + let mut chars = name.chars(); + chars + .next() + .is_some_and(|c| c.is_ascii_alphabetic() || c == '_') + && chars.all(|c| c.is_ascii_alphanumeric() || c == '_') +} + +/// 64-bit FNV-1a, to derive a per-name seed. +fn fnv1a(s: &str) -> u64 { + s.bytes().fold(0xcbf2_9ce4_8422_2325, |h, b| { + (h ^ u64::from(b)).wrapping_mul(0x0000_0100_0000_01b3) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn label(name: &str, cardinality: u64, child_of: Option<&str>) -> LabelSpec { + LabelSpec { + name: name.to_string(), + cardinality, + child_of: child_of.map(str::to_string), + } + } + + fn classic_labels() -> Vec { + vec![label("job", 10, None), label("label_0", 100, Some("job"))] + } + + #[test] + fn child_labels_fan_out_under_their_parent() { + let series = series_labels(&classic_labels()).unwrap(); + assert_eq!(series.len(), 1_000); + assert_eq!(series[0], ["0", "0.0"]); + assert_eq!(series[1], ["0", "0.1"]); + assert_eq!(series[100], ["1", "1.0"]); + assert_eq!(series[999], ["9", "9.99"]); + // A child value names its parent, so it is unique across parents. + let children: HashSet<&str> = series.iter().map(|s| s[1].as_str()).collect(); + assert_eq!(children.len(), 1_000); + assert!(series + .iter() + .all(|s| s[1].starts_with(&format!("{}.", s[0])))); + } + + #[test] + fn root_labels_form_a_cross_product() { + let labels = vec![ + label("region", 2, None), + label("service", 3, None), + label("endpoint", 2, Some("service")), + ]; + let series = series_labels(&labels).unwrap(); + assert_eq!(series.len(), 12); + let pairs: HashSet<(&str, &str)> = series + .iter() + .map(|s| (s[0].as_str(), s[1].as_str())) + .collect(); + assert_eq!(pairs.len(), 6); + assert_eq!(series[11], ["1", "2", "2.1"]); + } + + #[test] + fn rejects_a_child_before_its_parent() { + let labels = vec![label("label_0", 3, Some("job")), label("job", 2, None)]; + assert!(series_labels(&labels).is_err()); + } + + #[test] + fn facts_hold_on_generated_series() { + let labels = vec![ + label("region", 4, None), + label("service", 5, None), + label("endpoint", 3, Some("service")), + label("status", 2, None), + ]; + let series = series_labels(&labels).unwrap(); + let facts = facts(&labels, &series); + assert_eq!(facts.len(), 15); + assert!(facts.iter().all(|f| f.expected == f.actual), "{facts:?}"); + let by = |names: &[&str]| facts.iter().find(|f| f.labels == names).unwrap().expected; + // `endpoint` alone groups by its service too. + assert_eq!(by(&["endpoint"]), 15); + assert_eq!(by(&["region", "endpoint"]), 60); + assert_eq!(by(&["region", "service", "endpoint", "status"]), 120); + } + + #[test] + fn facts_catch_a_wrong_expectation() { + let labels = classic_labels(); + let series = series_labels(&labels).unwrap(); + // Claiming label_0 is a cross product rather than a fan-out expects + // 100 label_0 values, but the generated series have 1000. + let wrong = vec![label("job", 10, None), label("label_0", 100, None)]; + let facts = facts(&wrong, &series); + let label_0 = facts.iter().find(|f| f.labels == ["label_0"]).unwrap(); + assert_eq!((label_0.expected, label_0.actual), (100, 1_000)); + } + + #[test] + fn zipf_scales_average_one_and_spread_the_heavy_series() { + let scales = zipf_scales(1_000, 1.1, 42); + let mean = scales.iter().sum::() / 1_000.0; + assert!((mean - 1.0).abs() < 1e-9, "mean {mean}"); + + let ranks = zipf_ranks(1_000, 42); + let mut sorted = ranks.clone(); + sorted.sort_unstable(); + assert!(sorted.into_iter().eq(0..1_000)); + // Scales decrease with rank. + let by_rank = |r: usize| scales[ranks.iter().position(|&x| x == r).unwrap()]; + assert!((1..1_000).all(|r| by_rank(r) < by_rank(r - 1))); + // The 10 heaviest series fall in several of the 10 jobs (100 series + // each), not all in the first. + let jobs: HashSet = (0..1_000) + .filter(|&s| ranks[s] < 10) + .map(|s| s / 100) + .collect(); + assert!(jobs.len() >= 5, "{jobs:?}"); + // The ranks depend on the seed. + assert_ne!(ranks, zipf_ranks(1_000, 43)); + } + + fn write_table(metrics: Value) -> String { + let path = std::env::temp_dir().join(format!( + "paced-feeder-table-{}-{}.json", + std::process::id(), + fnv1a(&metrics.to_string()) + )); + let table = json!({"workloads": [{"dataset": "x", "e2e": {"metrics": metrics}}]}); + std::fs::write(&path, table.to_string()).unwrap(); + path.to_string_lossy().into_owned() + } + + fn classic_metrics() -> Value { + let labels = json!([{"name": "job", "cardinality": 2}, + {"name": "label_0", "cardinality": 3, "child_of": "job"}]); + let keys = json!({"zipf": 1.1, "keys": 6}); + json!({ + "data": {"type": "gauge", "labels": labels, "samples_per_sec_per_series": 4, + "values": {"value": keys, "dist": {"pareto": 2.0}}}, + "data_total": {"type": "counter", "labels": labels, "samples_per_sec_per_series": 4, + "values": {"increment": keys}}, + }) + } + + #[test] + fn loads_gauge_and_counter_from_a_table() { + let path = write_table(classic_metrics()); + let w = load(&path, None, None, 1).unwrap(); + assert_eq!(w.samples_per_sec, 4); + assert_eq!(w.num_series(), 12); + let (data, total) = (&w.metrics[0], &w.metrics[1]); + assert_eq!( + (data.name.as_str(), total.name.as_str()), + ("data", "data_total") + ); + assert_eq!( + data.kind, + Kind::Gauge { + pareto_shape: 2.0, + pareto_scale: 1.0 + } + ); + assert_eq!(total.kind, Kind::Counter); + // The same series is heavy in both, but the gauges' draws differ by metric. + assert_eq!(data.scales, total.scales); + assert_ne!(data.seed, total.seed); + + let only = load(&path, Some("data_total"), Some(2), 1).unwrap(); + assert_eq!(only.metrics.len(), 1); + assert_eq!(only.samples_per_sec, 2); + assert!(load(&path, Some("nope"), None, 1).is_err()); + } + + #[test] + fn rejects_records_and_mismatched_keys() { + let path = write_table(json!({"http": {"type": "records", "labels": [], "rate": 1}})); + let err = format!("{:#}", load(&path, None, None, 1).unwrap_err()); + assert!(err.contains("type `records` is not supported"), "{err}"); + + let mut metrics = classic_metrics(); + metrics["data"]["values"]["value"]["keys"] = json!(7); + let path = write_table(metrics); + let err = format!("{:#}", load(&path, Some("data"), None, 1).unwrap_err()); + assert!(err.contains("must equal the number of series"), "{err}"); + } +} diff --git a/asap-tools/data-sources/paced-feeder/src/workload.rs b/asap-tools/data-sources/paced-feeder/src/workload.rs index 3b93561e..de8f5d31 100644 --- a/asap-tools/data-sources/paced-feeder/src/workload.rs +++ b/asap-tools/data-sources/paced-feeder/src/workload.rs @@ -1,11 +1,13 @@ -//! Seeded synthetic workload: `groups × series_per_group` series, each emitting -//! `samples_per_sec` Pareto-distributed values per second. +//! Seeded synthetic workload: one or more metrics, each a set of labelled +//! series emitting `samples_per_sec` evenly spaced samples per second. //! -//! A sample's value depends only on `(seed, series, sample index)`, never on -//! wall-clock time or on which sink sends it, so two feeders started with the -//! same parameters produce identical data. +//! A sample's value depends only on `(metric seed, series, sample index)`, +//! never on wall-clock time or on which sink sends it, so two feeders started +//! with the same parameters produce identical data. -/// One generated sample. `series` indexes into [`Workload::series_labels`]. +use serde::Deserialize; + +/// One generated sample of one metric. `series` indexes into [`Metric::series`]. #[derive(Debug, Clone, Copy, PartialEq)] pub struct Row { pub series: usize, @@ -13,56 +15,137 @@ pub struct Row { pub value: f64, } +/// A label with `cardinality` values, or, with `child_of`, that many values +/// under each value of the parent label. +#[derive(Debug, Clone, PartialEq, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct LabelSpec { + pub name: String, + pub cardinality: u64, + #[serde(default)] + pub child_of: Option, +} + +#[derive(Debug, Clone, PartialEq)] +pub enum Kind { + /// `scale · pareto_scale · U^(-1/pareto_shape)`: Pareto(shape, scale) times + /// the series' scale. + Gauge { + pareto_shape: f64, + pareto_scale: f64, + }, + /// `(n + 1) · scale` at sample index `n`: starts at the first sample, grows + /// by the series' scale per sample and never resets. + Counter, +} + #[derive(Debug, Clone)] -pub struct Workload { +pub struct Metric { + pub name: String, + pub labels: Vec, + /// Each series' label values, in `labels` order. + pub series: Vec>, + /// Each series' value multiplier. + pub scales: Vec, + pub kind: Kind, pub seed: u64, - pub groups: usize, - pub series_per_group: usize, +} + +#[derive(Debug, Clone)] +pub struct Workload { pub samples_per_sec: u32, - pub pareto_shape: f64, - pub pareto_scale: f64, + pub metrics: Vec, } impl Workload { + /// The fixed `--groups C × --series-per-group s` workload: one gauge with + /// series `g * s + i` labelled `label_0="g{g}", instance="i{i}"`. + pub fn grouped( + name: &str, + seed: u64, + groups: usize, + series_per_group: usize, + samples_per_sec: u32, + pareto_shape: f64, + pareto_scale: f64, + ) -> Self { + let series: Vec> = (0..groups * series_per_group) + .map(|s| { + vec![ + format!("g{}", s / series_per_group), + format!("i{}", s % series_per_group), + ] + }) + .collect(); + let scales = vec![1.0; series.len()]; + let label = |name: &str, cardinality: usize| LabelSpec { + name: name.to_string(), + cardinality: cardinality as u64, + child_of: None, + }; + let metric = Metric { + name: name.to_string(), + labels: vec![ + label("label_0", groups), + label("instance", series_per_group), + ], + series, + scales, + kind: Kind::Gauge { + pareto_shape, + pareto_scale, + }, + seed, + }; + Self { + samples_per_sec, + metrics: vec![metric], + } + } + pub fn num_series(&self) -> usize { - self.groups * self.series_per_group + self.metrics.iter().map(|m| m.series.len()).sum() } pub fn rows_per_tick(&self) -> usize { self.num_series() * self.samples_per_sec as usize } - /// `(label_0, instance)` for a series: series `g * series_per_group + i` - /// is `("g{g}", "i{i}")`. - pub fn series_labels(&self, series: usize) -> (String, String) { - let group = series / self.series_per_group; - let instance = series % self.series_per_group; - (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 { + /// Samples of `metric` 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, metric: usize, 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) * 1_000 / per_sec) as i64; let sample_index = tick * per_sec + slot; - (0..self.num_series()) + let metric = &self.metrics[metric]; + (0..metric.series.len()) .map(|series| Row { series, timestamp_ms, - value: self.value(series, sample_index), + value: metric.value(series, sample_index), }) .collect() } +} - /// Pareto(shape, scale) by inverse transform: `scale · U^(-1/shape)` with - /// `U` uniform in (0, 1], so every value is ≥ `scale` and positive. +impl Metric { pub fn value(&self, series: usize, sample_index: u64) -> f64 { - let u = unit_uniform(self.seed, series as u64, sample_index); - self.pareto_scale * u.powf(-1.0 / self.pareto_shape) + let scale = self.scales[series]; + match self.kind { + // Pareto by inverse transform with `U` uniform in (0, 1], so every + // value is ≥ `scale · pareto_scale` and positive. + Kind::Gauge { + pareto_shape, + pareto_scale, + } => { + let u = unit_uniform(self.seed, series as u64, sample_index); + scale * pareto_scale * u.powf(-1.0 / pareto_shape) + } + Kind::Counter => (sample_index + 1) as f64 * scale, + } } } @@ -74,7 +157,7 @@ fn unit_uniform(seed: u64, series: u64, sample_index: u64) -> f64 { ((x >> 11) + 1) as f64 / (1u64 << 53) as f64 } -fn splitmix64(mut z: u64) -> u64 { +pub fn splitmix64(mut z: u64) -> u64 { z = z.wrapping_add(0x9E37_79B9_7F4A_7C15); z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9); z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB); @@ -86,21 +169,14 @@ mod tests { use super::*; fn workload() -> Workload { - Workload { - seed: 7, - groups: 3, - series_per_group: 2, - samples_per_sec: 2, - pareto_shape: 1.5, - pareto_scale: 1.0, - } + Workload::grouped("data", 7, 3, 2, 2, 1.5, 1.0) } #[test] fn each_slot_has_every_series_at_its_timestamp() { let w = workload(); for (slot, timestamp_ms) in [(0, 15_500), (1, 16_000)] { - let rows = w.slot_rows(5, slot, 10_000); + let rows = w.slot_rows(0, 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)); @@ -110,43 +186,42 @@ mod tests { #[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); - // A different seed changes them. - let other = Workload { seed: 8, ..w }; - assert_ne!( - a, - other - .slot_rows(4, 1, 0) + let values = |w: &Workload, start_ms| -> Vec { + w.slot_rows(0, 4, 1, start_ms) .iter() .map(|r| r.value) - .collect::>() - ); + .collect() + }; + // Same slot, different start time: same values. + assert_eq!(values(&w, 0), values(&w, 1_000_000)); + // A different seed changes them. + let other = Workload::grouped("data", 8, 3, 2, 2, 1.5, 1.0); + assert_ne!(values(&w, 0), values(&other, 0)); + } + + #[test] + fn grouped_values_keep_the_seeded_pareto_draw() { + // The grouped workload's values are the plain Pareto draw of + // `(seed, series, sample index)`, as before metrics had scales. + let w = workload(); + let u = unit_uniform(7, 3, 9); + assert_eq!(w.metrics[0].value(3, 9), u.powf(-1.0 / 1.5)); } #[test] fn labels_split_series_into_groups_and_instances() { let w = workload(); - assert_eq!(w.series_labels(0), ("g0".to_string(), "i0".to_string())); - assert_eq!(w.series_labels(3), ("g1".to_string(), "i1".to_string())); - assert_eq!(w.series_labels(5), ("g2".to_string(), "i1".to_string())); + let series = &w.metrics[0].series; + assert_eq!(series[0], ["g0", "i0"]); + assert_eq!(series[3], ["g1", "i1"]); + assert_eq!(series[5], ["g2", "i1"]); } #[test] fn values_follow_pareto_distribution() { - let w = Workload { - groups: 1, - series_per_group: 1, - ..workload() - }; + let w = Workload::grouped("data", 7, 1, 1, 2, 1.5, 1.0); let n = 200_000u64; - let mut values: Vec = (0..n).map(|i| w.value(0, i)).collect(); + let mut values: Vec = (0..n).map(|i| w.metrics[0].value(0, i)).collect(); assert!(values.iter().all(|&v| v >= 1.0)); values.sort_by(f64::total_cmp); // Pareto(a, xm) quantile: xm · (1 - q)^(-1/a). @@ -159,4 +234,18 @@ mod tests { ); } } + + #[test] + fn counters_increase_strictly_per_series() { + let metric = Metric { + kind: Kind::Counter, + scales: vec![0.5, 2.0], + ..workload().metrics[0].clone() + }; + for series in 0..2 { + let values: Vec = (0..100).map(|n| metric.value(series, n)).collect(); + assert_eq!(values[0], metric.scales[series]); + assert!(values.windows(2).all(|w| w[1] > w[0])); + } + } } From 581e5d9cca47cb3fd9eb4a5a65b88fe2914d5adf Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 13 Oct 2026 02:02:57 +0000 Subject: [PATCH 3/5] test(asap-tools): describe the grouped workload's values without history Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01AnJ7LocWxguh3q4QWkCYae --- asap-tools/data-sources/paced-feeder/src/workload.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/asap-tools/data-sources/paced-feeder/src/workload.rs b/asap-tools/data-sources/paced-feeder/src/workload.rs index de8f5d31..d82d02f4 100644 --- a/asap-tools/data-sources/paced-feeder/src/workload.rs +++ b/asap-tools/data-sources/paced-feeder/src/workload.rs @@ -200,9 +200,9 @@ mod tests { } #[test] - fn grouped_values_keep_the_seeded_pareto_draw() { - // The grouped workload's values are the plain Pareto draw of - // `(seed, series, sample index)`, as before metrics had scales. + fn grouped_values_are_the_seeded_pareto_draw() { + // Every series of the grouped workload has scale 1, so its values are + // the plain Pareto draw of `(seed, series, sample index)`. let w = workload(); let u = unit_uniform(7, 3, 9); assert_eq!(w.metrics[0].value(3, 9), u.powf(-1.0 / 1.5)); From 660a6db21b22fc3a15f2c17e7a854514c684a04a Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 13 Oct 2026 03:09:47 +0000 Subject: [PATCH 4/5] 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 5/5] 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()); + } +}