Repository navigation
feat(asap-tools): add paced feeder for the data-plane eval - #793
Conversation
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) { |
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
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; |
There was a problem hiding this comment.
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().
There was a problem hiding this comment.
Fixed in 9ca2f3d: main() now rejects --samples-per-sec > 1000.
|
|
||
| let stats = TickStats { | ||
| tick, | ||
| rows: rows.len(), |
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
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 { |
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
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? { |
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
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 |
There was a problem hiding this comment.
--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_.].
There was a problem hiding this comment.
Fixed in 9ca2f3d: --table is validated against [A-Za-z0-9_.] (non-empty) before it is interpolated.
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
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:/api/v1/write(snappy-compressed protobuf).INSERT … FORMAT RowBinaryinto(ts DateTime64(3), label_0 String, instance String, value Float64).The data has
Cgroups (label_0) ×sseries per group (instance),--samples-per-secsamples 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-secsub-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-secis capped at 1000 so millisecond timestamps never repeat.--stats-outwrites per-tick rows, bytes, start lag and elapsed time, plusachieved_rows_per_sec,overrun_ticksand 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.--tablemust 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], ownCargo.lock), likeotlp_exporterandfake_exporter.Testing
prometheus_remote_write.rs, to the right labels (sorted) and samples;--urlstopped after tick 0 with stats written; a bad--tableand--samples-per-sec 2000are rejected.precompute_enginebinary frommain:data{instance="i0",label_0="g0"}routed to the configured aggregationcargo clippy --all-targets -- -D warningsandcargo fmt --checkare clean.🤖 Generated with Claude Code