Skip to content

feat(asap-tools): add paced feeder for the data-plane eval - #793

Merged
milindsrivastava1997 merged 2 commits into
mainfrom
feat/paced-feeder
Oct 12, 2026
Merged

milindsrivastava1997 merged 2 commits into
mainfrom
feat/paced-feeder

Conversation

@zzylol

@zzylol zzylol commented Oct 5, 2026 •

Copy link
Copy Markdown
Contributor

Summary

Adds asap-tools/data-sources/paced-feeder, the data feeder for the data-plane vs ClickHouse evaluation (#789). It generates the synthetic workload on the fly and sends it in real time, one second of event time per wall-clock second:

  • ASAPQuery: Prometheus remote write to /api/v1/write (snappy-compressed protobuf).
  • ClickHouse: INSERT … FORMAT RowBinary into (ts DateTime64(3), label_0 String, instance String, value Float64).

The data has C groups (label_0) × s series per group (instance), --samples-per-sec samples per series, and Pareto(shape, scale) values. Each value is a stateless hash of (seed, series, sample index), so two feeders with the same flags send identical data and no dataset file is needed. At 1e6 series × 900 s, a file would be about 9e8 rows (~50 GB) per workload.

Pacing. Each tick sends its --samples-per-sec sub-second slots in turn. A slot's rows are split into batches sent concurrently, and the next slot waits for all of them, so a series' samples never arrive out of order. --samples-per-sec is capped at 1000 so millisecond timestamps never repeat. --stats-out writes per-tick rows, bytes, start lag and elapsed time, plus achieved_rows_per_sec, overrun_ticks and errors; rows and bytes count only successful requests. Any failed request makes the feeder exit non-zero, and a tick in which every request fails (e.g. a wrong --url, table or credentials) stops the run early, still writing the stats. --table must be made of [A-Za-z0-9_.] characters, since it is interpolated into the INSERT.

Batch sizes. The default batch size depends on the sink. Remote write uses 50,000 rows, which compresses to about 0.76 MB per request, under axum's 2 MB default body limit in ASAPQuery's ingest. ClickHouse uses 100,000 rows: at 50,000 rows, 1e6 rows/s fell behind (ticks took about 1.15 s, since each INSERT creates a part).

It is a standalone crate (empty [workspace], own Cargo.lock), like otlp_exporter and fake_exporter.

Testing

  • Unit tests (6):
    • every series gets a sample in each sub-second slot of a tick;
    • values depend only on seed, series and sample index;
    • series map to the right group and instance labels;
    • values follow the Pareto distribution (empirical vs theoretical p50/p90/p99 within 5%);
    • remote write decodes, with structs mirroring prometheus_remote_write.rs, to the right labels (sorted) and samples;
    • RowBinary round-trips every column.
  • Review fixes: a stub HTTP server checking per-series arrival order received 100,000 rows at 4 samples/s with 700-row batches, none out of order, matching the stats; a dead --url stopped after tick 0 with stats written; a bad --table and --samples-per-sec 2000 are rejected.
  • Live runs on a CloudLab c6320, against ClickHouse 26.10 (single binary) and the precompute_engine binary from main:
Run Result
ClickHouse, 1e5 rows/s, 10 s 1,000,000 rows; 100,000 distinct series; 10 distinct timestamps; all values ≥ 1; no overruns
ClickHouse, 1e6 rows/s, 10 s, feeder pinned to 2 cores 1e6 rows/s, 0 overruns, feeder 0.15 cores
ASAPQuery remote write, 1e6 rows/s, 10 s, feeder pinned to 2 cores 1e6 rows/s, 0 overruns, 0 failed requests, feeder 0.34 cores. The engine logged the parsed series data{instance="i0",label_0="g0"} routed to the configured aggregation

cargo clippy --all-targets -- -D warnings and cargo fmt --check are clean.

🤖 Generated with Claude Code

Generate a seeded synthetic workload (C groups x s series, Pareto values)
on the fly and send it in real time to ASAPQuery over Prometheus remote
write or to ClickHouse with RowBinary INSERTs. Values depend only on the
seed, series and sample index, so two feeders send identical data. Per-tick
timing is written as JSON so a run can be checked for keeping up.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

let rows = Arc::new(workload.tick_rows(tick, start_ms));
let mut tasks = tokio::task::JoinSet::new();
for begin in (0..rows.len()).step_by(batch_rows) {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Per-series ordering isn't guaranteed within a tick when --samples-per-sec > 1. tick_rows orders by sub-second slot k, then series, so when rows_per_tick > batch_rows the k=0 and k=1 rows for a series land in different batches, which are sent concurrently via the JoinSet. A k=1 batch (ts+500ms) can be ingested before the k=0 batch, and the precompute engine's per-group watermark may drop the earlier samples as late — ASAPQuery then sees less data than ClickHouse. Ordering only holds across ticks. Options: order rows by series first, or send each sub-second slot's batches before spawning the next slot's.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in 9ca2f3d: each tick now sends its sub-second slots in turn, and a slot's batches all return before the next slot's are sent (Workload::slot_rows replaces tick_rows). Checked against a stub server at 4 samples/s with small batches: 100,000 rows, none out of order per series.

let tick_start_ms = start_ms + tick as i64 * 1_000;
let mut rows = Vec::with_capacity(self.rows_per_tick());
for k in 0..per_sec {
let timestamp_ms = tick_start_ms + (k * 1_000 / per_sec) as i64;

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

With samples_per_sec > 1000, k * 1_000 / per_sec collapses consecutive k onto the same millisecond, so a series gets duplicate timestamps with different values. Prometheus-style ingest rejects/overwrites them while ClickHouse keeps both, so the two systems diverge. Suggest validating samples_per_sec <= 1000 alongside the existing checks in main().

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in 9ca2f3d: main() now rejects --samples-per-sec > 1000.


let stats = TickStats {
tick,
rows: rows.len(),

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

rows: rows.len() counts every row in the tick regardless of whether its batches succeeded, while bytes only accumulates on Ok. If some requests fail, rows_sent / achieved_rows_per_sec overstate the delivered rate and disagree with bytes_sent. Count rows per successful batch instead.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in 9ca2f3d: each task returns its batch's row count with its bytes, and the tick sums both only on success.

Ok(n) => bytes += n,
Err(e) => {
failed_requests += 1;
if errors.len() < 10 {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Request failures are only collected, and the run bails at the very end. A wrong --url, missing table, or bad credentials makes every request fail, but the feeder still burns the full --duration-secs (900s in the README examples) before reporting. Consider aborting if the first tick fails entirely, or after N consecutive failed ticks.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in 9ca2f3d: a tick in which every request fails stops the run (after recording that tick), so a wrong --url/table/credentials fails after tick 0. The stats are still written and the exit is non-zero.


let mut bytes = 0;
while let Some(joined) = tasks.join_next().await {
match joined? {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

joined? propagates a JoinError (panicked encode/send task) straight out of main, skipping the --stats-out write, so per-tick stats for the ticks that did complete are lost. Treat a JoinError like a request error (record it and continue/break), so stats are still written.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in 9ca2f3d: a JoinError is now recorded as a failed request, so the loop continues and --stats-out is still written.

SinkKind::Clickhouse => {
let query = format!(
"INSERT INTO {} ({ROW_BINARY_COLUMNS}) FORMAT RowBinary",
args.table

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

--table is interpolated into the INSERT SQL unquoted/unescaped. Names with special characters (e.g. db.my-table) produce malformed SQL, and a crafted value can append arbitrary SQL. Quote identifiers with backticks (escaping embedded backticks), or validate the name against [A-Za-z0-9_.].

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in 9ca2f3d: --table is validated against [A-Za-z0-9_.] (non-empty) before it is interpolated.

@zzylol
zzylol marked this pull request as ready for review October 12, 2026 17:09
Send each sub-second slot's batches before the next slot's so a series'
samples arrive in timestamp order, cap --samples-per-sec at 1000, count
rows only for successful requests, stop after a tick where every request
fails, record task panics as failed requests so stats are still written,
and validate --table before interpolating it into the INSERT.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01AnJ7LocWxguh3q4QWkCYae
@milindsrivastava1997
milindsrivastava1997 merged commit e89e8e8 into main Oct 12, 2026
12 checks passed
@milindsrivastava1997
milindsrivastava1997 deleted the feat/paced-feeder branch October 12, 2026 17:24
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants