From f3daa66fc459a6d21f23413584a6b099b35bc76e Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 13 Oct 2026 01:46:20 +0000 Subject: [PATCH 1/2] feat(asap-tools): generate records metrics in the paced feeder `--workload-table` now also sends a table's `records` metrics (the multi-grouping `http` and `flows`). Each record is a stateless hash of the seed, the metric and its index: labels are Zipf(`skew`) draws (a child given its parent, value `i` of rank `i`), keys are Zipf ranks, a burst sends `share` of the records to its label value from its own keys, and Pareto values are scaled by their `scaled_by` label's factor (aqpbm-datagen's log-uniform [1, 10]) and by the anomaly factor on the anomaly label's first value. A tick starting at t spreads its records over 10 ms slots in (t, t + 1 s], slot s at t + (s + 1) * 10 ms, and is sent concurrently like gauge/counter ticks, so one feeder can send both kinds. Records are generated inside the encoding tasks. Remote write sends one sample per value field, named `.`; ClickHouse gets one row per record in a table named after the metric. `--records-per-sec` scales a run down, and `--verify-facts` also compares the shares of each label's heaviest values and of the burst with sketch-bench's `label_shares()`. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01AnJ7LocWxguh3q4QWkCYae --- .../data-sources/paced-feeder/README.md | 119 ++- .../data-sources/paced-feeder/src/encode.rs | 434 ++++++++- .../data-sources/paced-feeder/src/main.rs | 214 +++-- .../data-sources/paced-feeder/src/records.rs | 880 ++++++++++++++++++ .../data-sources/paced-feeder/src/table.rs | 216 ++++- .../data-sources/paced-feeder/src/workload.rs | 77 +- 6 files changed, 1779 insertions(+), 161 deletions(-) create mode 100644 asap-tools/data-sources/paced-feeder/src/records.rs diff --git a/asap-tools/data-sources/paced-feeder/README.md b/asap-tools/data-sources/paced-feeder/README.md index c7362364..fea7a119 100644 --- a/asap-tools/data-sources/paced-feeder/README.md +++ b/asap-tools/data-sources/paced-feeder/README.md @@ -15,7 +15,8 @@ the same stream without a dataset file on disk. 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, +`(--seed, metric, series, sample index)` (a record only on +`(--seed, metric, record index)`), never on wall-clock time or the sink, and each series emits `--samples-per-sec` samples per second, evenly spaced. ### `--groups` workload @@ -63,13 +64,69 @@ metric in it, or only `--metric NAME`. For example: 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 +#### `records` metrics + +The multi-grouping templates' metrics (`http`, `flows`, or `data_3_http` with +several metrics) are streams of records, each with a value of every label and +of every value field: + +```json +"http": {"type": "records", "records_per_sec": 2000000, + "labels": [{"name": "region", "cardinality": 4, "skew": 0.5}, + {"name": "service", "cardinality": 25, "skew": 1.1}, + {"name": "endpoint", "cardinality": 25, "child_of": "service", "skew": 1.1}, + {"name": "status", "cardinality": 4, "skew": 2.0}], + "values": {"latency": {"pareto": 2.0, "scaled_by": "service", + "anomaly": {"label": "endpoint", "factor": 10}}, + "user_id": {"zipf": 0.8, "keys": 1000000}}}, +"flows": {"type": "records", "records_per_sec": 4000000, + "labels": [{"name": "dst_subnet", "cardinality": 1000, "skew": 1.1}, …], + "values": {"src_ip": {"zipf": 0.0, "keys": 1000000, + "burst": {"label": "dst_subnet", "value": 999, + "share": 0.05, "keys": 10000}}}} +``` + +Every draw of record `n` is a hash of the metric's seed (from `--seed` and its +name), `n` and the draw, so records match sketch-bench's `label_shares()` +(`scripts/export_autosketch_eval_table.py`), which the optimizer assumed: + +- **Labels** are drawn independently, a child given its parent: value `i` (a + child's index `i` under its parent) has weight `(i+1)^-skew`, so `0` is the + heaviest. Values are named as for gauges (`service="3"`, + `endpoint="3.17"`). +- **Keys** (`zipf` θ over `keys`): the key of Zipf rank `r` is `r`, so key 0 is + the most frequent. +- **Burst**: with probability `share` a record is a burst record: its `label` + is `value` (a child index for a child label) and its key is uniform over + `keys` burst keys numbered after the background keys (`1000000` to + `1009999` for `flows`). Other records draw both normally, so `dst_subnet`'s + shares are Zipf times `1 - share`, plus `share` at `999`. +- **Pareto** (`latency`): `1000 · U^(-1/a)` (`U` in (0, 1]), times its service's + fixed factor with `scaled_by`, times `factor` on the anomaly label's first + value (`endpoint="0.0"`, the heaviest endpoint of the heaviest service). A + service's factor is aqpbm-datagen's `scale_factor` (as in sketch-bench's + `configs/datagen/hydra_http_latency.yaml`, scale 1000 and `scale_range: + [1.0, 10.0]`): `10^u` with `u` in [0, 1) a SplitMix64 hash of the value's + FNV-1a xor `--seed`, so it is the same in every metric of a table. Scaling + keeps the tail index `a`. +- **Time.** A tick starting at `t` spreads its `records_per_sec` records + evenly over 100 slots of 10 ms; slot `s` has every record at + `t + (s + 1) · 10 ms`, so the tick lies in `(t, t + 1 s]`. +- `--records-per-sec` overrides every records metric's rate, e.g. to scale a + run down. + +`--verify-facts` checks the generated data 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. +number among the generated series (for records metrics, among all label-value +combinations a record can have), and exits non-zero if any differ. For records +metrics it also draws records `0..1e6` and compares each label's three +heaviest values' shares (a child's index pooled over parents, i.e. its share +given the parent), the burst value's share and the share of burst records with +`label_shares()`; a share fails if it is off by more than 5 standard errors, +`5·sqrt(p(1-p)/n)`. ```sh $ paced_feeder --workload-table synthetic-templatesclassic-shared1.json --verify-facts @@ -80,6 +137,24 @@ data: 10000 series job,label_0 10000 10000 data_total: 10000 series … + +$ paced_feeder --workload-table synthetic-templatesmultigroup-shared1.json --verify-facts +flows: 4000000 records/s, 3000000 label combinations + labels expected generated + dst_subnet 1000 1000 + … + shares of 1000000 records (tolerance 5 standard errors) + label value expected generated tolerance + dst_subnet 0 0.17047 0.17069 0.00188 + … + dst_subnet 999 0.05009 0.05037 0.00109 + … + burst records 0.05000 0.05029 0.00109 + burst keys: 9938 distinct of 10000 +http: 2000000 records/s, 10000 label combinations + … + endpoint *.0 (given service) 0.29841 0.29855 0.00229 + … ``` Event time starts at the next whole wall-clock second (or `--start-ms`), so @@ -112,6 +187,14 @@ 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`). +A record of a records metric becomes one sample per value field, named +`.` (`http.latency`, `http.user_id`), with the record's labels +and timestamp. ASAPQuery's SQL mode routes a sample to the aggregations whose +metric is `table.value_column` (or whose table is the sample's name), so each +field feeds only its own aggregations. Many records of a slot share a series +and timestamp, so Prometheus, which keeps one sample per series and +timestamp, is not a target for records metrics. + With `--groups`, the ClickHouse table (`--table`) must have these columns (any engine and ordering): @@ -131,6 +214,19 @@ CREATE TABLE data_total (ts DateTime64(3), job String, label_0 String, value Flo ENGINE = MergeTree ORDER BY (job, label_0, ts); ``` +A records metric goes to the table named after it, with columns `ts`, the +labels in the description's order, then the value fields in name order (keys +are integers below 2^53, exact in `Float64`): + +```sql +CREATE TABLE http (ts DateTime64(3), region String, service String, + endpoint String, status String, latency Float64, user_id Float64) +ENGINE = MergeTree ORDER BY (region, service, endpoint, status, ts); +CREATE TABLE flows (ts DateTime64(3), dst_subnet String, dst_port String, + proto String, src_ip Float64) +ENGINE = MergeTree ORDER BY (dst_subnet, dst_port, proto, ts); +``` + `CLICKHOUSE_USER` / `CLICKHOUSE_PASSWORD` (or the matching flags) set credentials. @@ -150,6 +246,11 @@ 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 ClickHouse, where every INSERT creates a part. +For records metrics, each request generates its own records as it encodes +them, so generation runs on all cores. For remote write, `--batch-rows` +counts samples, a record being one per value field (`http`: 25,000 records +per request). Rows in the stats are records. + `--stats-out` writes, per tick, the rows, bytes, start lag and elapsed time, plus totals: - `achieved_rows_per_sec`; @@ -168,3 +269,13 @@ Measured on a CloudLab c6320 (feeder pinned to 2 cores, 10 s at 1e6 rows/s): |---|---|---| | ClickHouse 26.10 | 0 | 0.15 cores | | ASAPQuery precompute engine | 0 | 0.34 cores | + +Records metrics at full rate against a local HTTP stub that discards the +bodies (shared, loaded 56-core machine, default flags, mean tick time over +10 s, no overruns): + +| Metric | Remote write | ClickHouse | +|---|---|---| +| `http` (2e6 records/s, 4e6 samples/s) | 171 ms, 2.5 cores | 104 ms, 1.1 cores | +| `flows` (4e6 records/s) | 195 ms, 2.8 cores | 117 ms, 1.5 cores | +| both (6e6 records/s) | 340 ms, 4.9 cores | 194 ms, 2.5 cores | diff --git a/asap-tools/data-sources/paced-feeder/src/encode.rs b/asap-tools/data-sources/paced-feeder/src/encode.rs index bb6249b2..9cda3334 100644 --- a/asap-tools/data-sources/paced-feeder/src/encode.rs +++ b/asap-tools/data-sources/paced-feeder/src/encode.rs @@ -1,11 +1,30 @@ -//! 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. +//! Wire encoders for the two sinks. Each series' labels (or each label value +//! of a records metric) are encoded once up front, so the per-sample cost at +//! high rates is just the values and timestamp. -use crate::workload::{Metric, Row}; +use crate::records::{RecordMetric, Records}; +use crate::workload::{Metric, Row, Rows}; use prost::encoding::{encode_key, encode_varint, encoded_len_varint, WireType}; +use std::ops::Range; +use std::sync::Arc; pub trait Encoder: Send + Sync { - fn encode(&self, rows: &[Row]) -> Vec; + /// Encodes `rows[range]`, which are of the kind the encoder was built for. + fn encode(&self, rows: &Rows, range: Range) -> Vec; +} + +fn samples(rows: &Rows) -> &[Row] { + match rows { + Rows::Samples(rows) => rows, + Rows::Records(_) => panic!("a series encoder was given records"), + } +} + +fn records(rows: &Rows) -> &Records { + match rows { + Rows::Records(records) => records, + Rows::Samples(_) => panic!("a records encoder was given samples"), + } } /// Prometheus remote write: a snappy-compressed `WriteRequest` with one @@ -33,10 +52,7 @@ impl RemoteWriteEncoder { labels.sort_unstable(); let mut buf = Vec::new(); 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); - encode_bytes_field(1, &label, &mut buf); + encode_label(name, value, &mut buf); } buf }) @@ -46,26 +62,125 @@ impl RemoteWriteEncoder { } impl Encoder for RemoteWriteEncoder { - fn encode(&self, rows: &[Row]) -> Vec { + fn encode(&self, rows: &Rows, range: Range) -> Vec { let mut request = Vec::new(); let mut series = Vec::new(); - let mut sample = Vec::new(); - for row in rows { - sample.clear(); - encode_key(1, WireType::SixtyFourBit, &mut sample); - sample.extend_from_slice(&row.value.to_le_bytes()); - encode_key(2, WireType::Varint, &mut sample); - encode_varint(row.timestamp_ms as u64, &mut sample); - + for row in &samples(rows)[range] { series.clear(); series.extend_from_slice(&self.series_labels[row.series]); - encode_bytes_field(2, &sample, &mut series); - + encode_sample(row.value, row.timestamp_ms, &mut series); encode_bytes_field(1, &series, &mut request); } - snap::raw::Encoder::new() - .compress_vec(&request) - .expect("snappy compression of an in-memory buffer cannot fail") + compress(&request) + } +} + +/// Generates records `range` of `rows` (indices relative to their first) and +/// calls `f` with each one's timestamp, label value indices and values. +fn for_each_record( + metric: &RecordMetric, + rows: &Rows, + range: Range, + mut f: impl FnMut(i64, &[u32], &[f64]), +) { + let records = records(rows); + let mut labels = vec![0u32; metric.labels.len()]; + let mut values = vec![0.0; metric.fields.len()]; + let first = records.indices.start; + for index in first + range.start as u64..first + range.end as u64 { + metric.record(index, &mut labels, &mut values); + f( + metric.timestamp_ms(index, records.start_ms), + &labels, + &values, + ); + } +} + +/// Remote write of a records metric: each record is one sample per value +/// field, named `.`, with the record's labels and timestamp. +/// ASAPQuery's SQL mode routes a sample to the aggregations of +/// `table.value_column`, so each field feeds only its own aggregations. +pub struct RecordsRemoteWriteEncoder { + metric: Arc, + /// The labels in name order, `None` for `__name__`. + order: Vec>, + /// The encoded `Label` field of each value of each label. + label_values: Vec>>, + /// The encoded `__name__` `Label` field of each value field. + names: Vec>, +} + +/// The `__name__` of value field `field` of records metric `metric`. +pub fn record_sample_name(metric: &str, field: &str) -> String { + format!("{metric}.{field}") +} + +impl RecordsRemoteWriteEncoder { + pub fn new(metric: Arc) -> Self { + let mut sorted: Vec<(&str, Option)> = metric + .labels + .iter() + .enumerate() + .map(|(i, l)| (l.name.as_str(), Some(i))) + .collect(); + sorted.push(("__name__", None)); + sorted.sort_unstable(); + let label_values = metric + .labels + .iter() + .zip(&metric.label_values) + .map(|(label, values)| { + values + .iter() + .map(|value| { + let mut buf = Vec::new(); + encode_label(&label.name, value, &mut buf); + buf + }) + .collect() + }) + .collect(); + let names = metric + .fields + .iter() + .map(|field| { + let mut buf = Vec::new(); + encode_label( + "__name__", + &record_sample_name(&metric.name, field), + &mut buf, + ); + buf + }) + .collect(); + Self { + order: sorted.into_iter().map(|(_, i)| i).collect(), + label_values, + names, + metric, + } + } +} + +impl Encoder for RecordsRemoteWriteEncoder { + fn encode(&self, rows: &Rows, range: Range) -> Vec { + let mut request = Vec::new(); + let mut series = Vec::new(); + for_each_record(&self.metric, rows, range, |timestamp_ms, labels, values| { + for (field, &value) in values.iter().enumerate() { + series.clear(); + for label in &self.order { + series.extend_from_slice(match label { + None => &self.names[field], + Some(l) => &self.label_values[*l][labels[*l] as usize], + }); + } + encode_sample(value, timestamp_ms, &mut series); + encode_bytes_field(1, &series, &mut request); + } + }); + compress(&request) } } @@ -94,8 +209,7 @@ impl RowBinaryEncoder { .map(|values| { let mut buf = Vec::new(); for s in values { - encode_varint(s.len() as u64, &mut buf); - buf.extend_from_slice(s.as_bytes()); + encode_string(s, &mut buf); } buf }) @@ -105,7 +219,8 @@ impl RowBinaryEncoder { } impl Encoder for RowBinaryEncoder { - fn encode(&self, rows: &[Row]) -> Vec { + fn encode(&self, rows: &Rows, range: Range) -> Vec { + let rows = &samples(rows)[range]; let mut buf = Vec::with_capacity(rows.len() * 32); for row in rows { buf.extend_from_slice(&row.timestamp_ms.to_le_bytes()); @@ -116,6 +231,96 @@ impl Encoder for RowBinaryEncoder { } } +/// ClickHouse `RowBinary` of a records metric for columns `ts DateTime64(3)`, +/// each label as `String` in description order, then each value field as +/// `Float64` in name order. +pub struct RecordsRowBinaryEncoder { + metric: Arc, + /// Each value of each label as a varint length plus bytes. + label_values: Vec>>, +} + +/// The column list `RecordsRowBinaryEncoder` writes for `metric`, in order. +pub fn record_columns(metric: &RecordMetric) -> String { + let labels = metric.labels.iter().map(|l| l.name.as_str()); + std::iter::once("ts") + .chain(labels) + .chain(metric.fields.iter().map(String::as_str)) + .collect::>() + .join(", ") +} + +impl RecordsRowBinaryEncoder { + pub fn new(metric: Arc) -> Self { + let label_values = metric + .label_values + .iter() + .map(|values| { + values + .iter() + .map(|value| { + let mut buf = Vec::new(); + encode_string(value, &mut buf); + buf + }) + .collect() + }) + .collect(); + Self { + metric, + label_values, + } + } +} + +impl Encoder for RecordsRowBinaryEncoder { + fn encode(&self, rows: &Rows, range: Range) -> Vec { + let mut buf = Vec::with_capacity(range.len() * 64); + for_each_record(&self.metric, rows, range, |timestamp_ms, labels, values| { + buf.extend_from_slice(×tamp_ms.to_le_bytes()); + for (label_values, &v) in self.label_values.iter().zip(labels) { + buf.extend_from_slice(&label_values[v as usize]); + } + for value in values { + buf.extend_from_slice(&value.to_le_bytes()); + } + }); + buf + } +} + +/// A `Label` message as `TimeSeries` field 1. +fn encode_label(name: &str, value: &str, buf: &mut Vec) { + let mut label = Vec::new(); + encode_bytes_field(1, name.as_bytes(), &mut label); + encode_bytes_field(2, value.as_bytes(), &mut label); + encode_bytes_field(1, &label, buf); +} + +/// A `Sample` message as `TimeSeries` field 2. +fn encode_sample(value: f64, timestamp_ms: i64, series: &mut Vec) { + // One-byte keys of fields 1 (fixed 64-bit) and 2 (varint). + let len = 1 + 8 + 1 + encoded_len_varint(timestamp_ms as u64); + encode_key(2, WireType::LengthDelimited, series); + encode_varint(len as u64, series); + encode_key(1, WireType::SixtyFourBit, series); + series.extend_from_slice(&value.to_le_bytes()); + encode_key(2, WireType::Varint, series); + encode_varint(timestamp_ms as u64, series); +} + +/// A ClickHouse `String`: varint length plus bytes. +fn encode_string(s: &str, buf: &mut Vec) { + encode_varint(s.len() as u64, buf); + buf.extend_from_slice(s.as_bytes()); +} + +fn compress(request: &[u8]) -> Vec { + snap::raw::Encoder::new() + .compress_vec(request) + .expect("snappy compression of an in-memory buffer cannot fail") +} + fn encode_bytes_field(tag: u32, bytes: &[u8], buf: &mut Vec) { encode_key(tag, WireType::LengthDelimited, buf); buf.reserve(encoded_len_varint(bytes.len() as u64) + bytes.len()); @@ -169,6 +374,7 @@ mod tests { name: name.to_string(), cardinality, child_of: child_of.map(str::to_string), + skew: None, }; let labels = vec![ label("zone", 2, None), @@ -188,9 +394,14 @@ mod tests { Workload { samples_per_sec: 1, metrics: vec![metric], + records: Vec::new(), } } + fn encode_all(encoder: &dyn Encoder, rows: &[Row]) -> Vec { + encoder.encode(&Rows::Samples(rows.to_vec()), 0..rows.len()) + } + fn decode_remote_write(body: &[u8]) -> WriteRequest { let raw = snap::raw::Decoder::new().decompress_vec(body).unwrap(); WriteRequest::decode(raw.as_slice()).unwrap() @@ -199,9 +410,11 @@ mod tests { #[test] fn remote_write_decodes_to_one_series_per_row() { 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)); + let rows = w.series_rows(0, 2, 0, 1_700_000_000_000); + let request = decode_remote_write(&encode_all( + &RemoteWriteEncoder::new(&w.metrics[0], "data"), + &rows, + )); assert_eq!(request.timeseries.len(), rows.len()); for (ts, row) in request.timeseries.iter().zip(&rows) { let values = &w.metrics[0].series[row.series]; @@ -232,9 +445,11 @@ mod tests { 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)); + let rows = w.series_rows(0, 0, 0, 1_000); + let request = decode_remote_write(&encode_all( + &RemoteWriteEncoder::new(metric, &metric.name), + &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 { @@ -289,8 +504,8 @@ mod tests { 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 rows = w.series_rows(0, 0, 0, 42_000); + let decoded = decode_row_binary(&encode_all(&RowBinaryEncoder::new(metric), &rows), 2); let expected: Vec<_> = rows .iter() .map(|r| (r.timestamp_ms, metric.series[r.series].clone(), r.value)) @@ -306,8 +521,8 @@ mod tests { 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); + let rows = w.series_rows(0, 3, 0, 0); + let decoded = decode_row_binary(&encode_all(&RowBinaryEncoder::new(metric), &rows), 4); assert_eq!(decoded.len(), 12); assert_eq!( decoded[4], @@ -318,4 +533,153 @@ mod tests { ) ); } + + fn http() -> Arc { + let spec = crate::records::tests::http_spec(); + Arc::new(RecordMetric::from_table("http", &spec, Some(1_000), 42).unwrap()) + } + + /// Record `index` of `metric` as (timestamp, label values, values). + fn record(metric: &RecordMetric, index: u64, start_ms: i64) -> (i64, Vec, Vec) { + let mut labels = vec![0; metric.labels.len()]; + let mut values = vec![0.0; metric.fields.len()]; + metric.record(index, &mut labels, &mut values); + let names = labels + .iter() + .zip(&metric.label_values) + .map(|(&v, values)| values[v as usize].clone()) + .collect(); + (metric.timestamp_ms(index, start_ms), names, values) + } + + #[test] + fn remote_write_sends_a_sample_per_record_field() { + let metric = http(); + // Slot 3 of tick 1: records 1030..1040 at 1_000 + 1_000 + 40 ms. + let rows = Rows::Records(Records { + start_ms: 1_000, + indices: 1_030..1_040, + }); + let request = decode_remote_write( + &RecordsRemoteWriteEncoder::new(Arc::clone(&metric)).encode(&rows, 2..5), + ); + assert_eq!(request.timeseries.len(), 6); + for (i, pair) in request.timeseries.chunks(2).enumerate() { + let (timestamp, labels, values) = record(&metric, 1_032 + i as u64, 1_000); + assert_eq!(timestamp, 2_040); + // Value fields in name order: latency, then user_id. + for (ts, (field, value)) in pair + .iter() + .zip([("latency", values[0]), ("user_id", values[1])]) + { + let got: Vec<(&str, &str)> = ts + .labels + .iter() + .map(|l| (l.name.as_str(), l.value.as_str())) + .collect(); + let name = format!("http.{field}"); + assert_eq!( + got, + vec![ + ("__name__", name.as_str()), + ("endpoint", labels[2].as_str()), + ("region", labels[0].as_str()), + ("service", labels[1].as_str()), + ("status", labels[3].as_str()), + ] + ); + assert_eq!(ts.samples, vec![Sample { value, timestamp }]); + } + } + } + + /// Decodes RowBinary rows of `ts`, `num_labels` strings and `num_values` floats. + fn decode_record_rows( + buf: &[u8], + num_labels: usize, + num_values: usize, + ) -> Vec<(i64, Vec, Vec)> { + let mut cursor = buf; + let mut take = |n: usize| -> Vec { + let (head, rest) = cursor.split_at(n); + cursor = rest; + head.to_vec() + }; + let mut rows = Vec::new(); + let mut remaining = buf.len(); + while remaining > 0 { + let ts = i64::from_le_bytes(take(8).try_into().unwrap()); + let labels: Vec = (0..num_labels) + .map(|_| { + let len = take(1)[0] as usize; + String::from_utf8(take(len)).unwrap() + }) + .collect(); + let values: Vec = (0..num_values) + .map(|_| f64::from_le_bytes(take(8).try_into().unwrap())) + .collect(); + remaining -= 8 + 8 * num_values + labels.iter().map(|l| 1 + l.len()).sum::(); + rows.push((ts, labels, values)); + } + rows + } + + #[test] + fn row_binary_writes_a_row_per_record() { + let metric = http(); + assert_eq!( + record_columns(&metric), + "ts, region, service, endpoint, status, latency, user_id" + ); + let rows = Rows::Records(metric.tick(0, 5_000)); + assert_eq!(rows.len(), 1_000); + let body = RecordsRowBinaryEncoder::new(Arc::clone(&metric)).encode(&rows, 0..1_000); + let decoded = decode_record_rows(&body, 4, 2); + let expected: Vec<_> = (0..1_000).map(|i| record(&metric, i, 5_000)).collect(); + assert_eq!(decoded, expected); + assert_eq!((decoded[0].0, decoded[999].0), (5_010, 6_000)); + } + + #[test] + fn both_sinks_send_the_same_records() { + let metric = http(); + let rows = Rows::Records(metric.tick(2, 0)); + let range = 0..rows.len(); + let request = decode_remote_write( + &RecordsRemoteWriteEncoder::new(Arc::clone(&metric)).encode(&rows, range.clone()), + ); + let decoded = decode_record_rows( + &RecordsRowBinaryEncoder::new(Arc::clone(&metric)).encode(&rows, range), + 4, + 2, + ); + assert_eq!(request.timeseries.len(), 2 * decoded.len()); + for (pair, (ts, labels, values)) in request.timeseries.chunks(2).zip(&decoded) { + for (series, value) in pair.iter().zip(values) { + assert_eq!( + series.samples, + vec![Sample { + value: *value, + timestamp: *ts + }] + ); + let mut sent: Vec<&str> = series.labels[1..] + .iter() + .map(|l| l.value.as_str()) + .collect(); + let mut want: Vec<&str> = labels.iter().map(String::as_str).collect(); + sent.sort_unstable(); + want.sort_unstable(); + assert_eq!(sent, want); + } + } + } + + #[test] + #[should_panic(expected = "a series encoder was given records")] + fn series_encoders_reject_records() { + let w = grouped(); + let rows = Rows::Records(http().tick(0, 0)); + RowBinaryEncoder::new(&w.metrics[0]).encode(&rows, 0..1); + } } diff --git a/asap-tools/data-sources/paced-feeder/src/main.rs b/asap-tools/data-sources/paced-feeder/src/main.rs index f7e1efff..c466cf15 100644 --- a/asap-tools/data-sources/paced-feeder/src/main.rs +++ b/asap-tools/data-sources/paced-feeder/src/main.rs @@ -7,17 +7,21 @@ //! is written to `--stats-out` so a run can be checked for keeping up. mod encode; +mod records; mod table; mod workload; use anyhow::{bail, Context, Result}; use clap::{Parser, ValueEnum}; -use encode::{row_binary_columns, Encoder, RemoteWriteEncoder, RowBinaryEncoder}; +use encode::{ + record_columns, row_binary_columns, Encoder, RecordsRemoteWriteEncoder, + RecordsRowBinaryEncoder, RemoteWriteEncoder, RowBinaryEncoder, +}; 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::{Rows, Workload}; #[derive(Debug, Clone, Copy, ValueEnum, Serialize)] #[serde(rename_all = "kebab-case")] @@ -65,6 +69,10 @@ struct Args { /// the table's `samples_per_sec_per_series` with `--workload-table`. #[arg(long)] samples_per_sec: Option, + /// Records per second of each `records` metric of `--workload-table`, + /// overriding the table's `records_per_sec`, e.g. to scale a run down. + #[arg(long, requires = "workload_table")] + records_per_sec: Option, #[arg(long, default_value_t = 1.5)] pareto_shape: f64, #[arg(long, default_value_t = 1.0)] @@ -72,8 +80,9 @@ struct Args { #[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. + /// generated number of distinct label-value tuples, and for records + /// metrics the expected and generated shares of each label's heaviest + /// values, then exit without sending; non-zero if any check fails. #[arg(long)] verify_facts: bool, @@ -87,7 +96,8 @@ struct Args { /// 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. + /// and 100,000 for ClickHouse, where each INSERT creates a part. For + /// remote write it counts samples: a record is one per value field. #[arg(long)] batch_rows: Option, /// Concurrent requests within one tick. @@ -119,7 +129,8 @@ struct RunStats { pareto_scale: Option, workload_table: Option, metrics: Vec, - samples_per_sec: u32, + /// Unset without gauge/counter metrics. + samples_per_sec: Option, seed: u64, start_ms: i64, duration_secs: u64, @@ -137,34 +148,57 @@ struct RunStats { ticks: Vec, } -/// Where one metric's rows go: the request to send them with and their encoder. +/// Where one metric's rows go: the request to send them with, their encoder +/// and the rows per request. struct Target { request: reqwest::RequestBuilder, encoder: Box, + batch_rows: usize, } struct Sink { client: reqwest::Client, - /// One per metric of the workload, in order. + /// One per metric of the workload, in [`Workload::names`] order. targets: Vec, } impl Sink { - fn new(args: &Args, sink: SinkKind, url: &str, workload: &Workload) -> Result { + fn new( + args: &Args, + sink: SinkKind, + url: &str, + workload: &Workload, + batch_rows: usize, + ) -> Result { let client = reqwest::Client::builder() .pool_max_idle_per_host(args.concurrency) .build()?; - let targets = workload + let remote_write = || { + client + .post(url) + .header("Content-Encoding", "snappy") + .header("Content-Type", "application/x-protobuf") + .header("X-Prometheus-Remote-Write-Version", "0.1.0") + }; + let clickhouse = |table: &str, columns: String| { + let query = format!("INSERT INTO {table} ({columns}) FORMAT RowBinary"); + 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); + } + request + }; + let mut targets: Vec = 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"), + request: remote_write(), encoder: Box::new(RemoteWriteEncoder::new(metric, &metric.name)), + batch_rows, }, SinkKind::Clickhouse => { let table = if args.workload_table.is_some() { @@ -172,30 +206,38 @@ impl Sink { } 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, + request: clickhouse(table, row_binary_columns(metric)), encoder: Box::new(RowBinaryEncoder::new(metric)), + batch_rows, } } }) .collect(); + targets.extend(workload.records.iter().map(|metric| match sink { + // Remote write's batch counts samples, one per value field. + SinkKind::RemoteWrite => Target { + request: remote_write(), + encoder: Box::new(RecordsRemoteWriteEncoder::new(Arc::clone(metric))), + batch_rows: (batch_rows / metric.fields.len()).max(1), + }, + SinkKind::Clickhouse => Target { + request: clickhouse(&metric.name, record_columns(metric)), + encoder: Box::new(RecordsRowBinaryEncoder::new(Arc::clone(metric))), + batch_rows, + }, + })); Ok(Self { client, targets }) } - async fn send(&self, metric: usize, rows: &[Row]) -> Result { + async fn send( + &self, + metric: usize, + rows: &Rows, + range: std::ops::Range, + ) -> Result { let target = &self.targets[metric]; - let body = target.encoder.encode(rows); + let body = target.encoder.encode(rows, range); let bytes = body.len(); let request = target .request @@ -239,6 +281,7 @@ fn build_workload(args: &Args) -> Result { path, args.metric.as_deref(), args.samples_per_sec, + args.records_per_sec, args.seed, ); } @@ -264,22 +307,67 @@ fn build_workload(args: &Args) -> Result { } /// Prints the facts check of every metric; true if every fact holds. -fn verify_facts(workload: &Workload) -> bool { +fn verify_facts(workload: &Workload) -> Result { 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 &= print_label_facts(&metric.labels, &metric.series); + } + for metric in &workload.records { + // Every combination of label values a record can have. + let domain = table::series_labels(&metric.labels)?; + println!( + "{}: {} records/s, {} label combinations", + metric.name, + metric.records_per_sec, + domain.len() + ); + ok &= print_label_facts(&metric.labels, &domain); + let report = metric.share_checks(records::SHARE_SAMPLE_RECORDS); + println!( + " shares of {} records (tolerance {} standard errors)", + report.records, + records::SHARE_TOLERANCE_SIGMAS + ); + println!( + " {:<12} {:<28} {:>10} {:>10} {:>10}", + "label", "value", "expected", "generated", "tolerance" + ); + for check in &report.checks { + let holds = check.holds(); ok &= holds; println!( - " {:<40} {:>12} {:>12}{}", - fact.labels.join(","), - fact.expected, - fact.actual, + " {:<12} {:<28} {:>10.5} {:>10.5} {:>10.5}{}", + check.label, + check.value, + check.expected, + check.actual, + check.tolerance, if holds { "" } else { " MISMATCH" } ); } + if let Some((seen, keys)) = report.burst_keys { + println!(" burst keys: {seen} distinct of {keys}"); + } + } + Ok(ok) +} + +/// Prints the distinct label-value tuples of every label subset of `series`; +/// true if each matches the description. +fn print_label_facts(labels: &[workload::LabelSpec], series: &[Vec]) -> bool { + let mut ok = true; + println!(" {:<40} {:>12} {:>12}", "labels", "expected", "generated"); + for fact in table::facts(labels, 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 } @@ -288,17 +376,19 @@ fn verify_facts(workload: &Workload) -> bool { async fn main() -> Result<()> { let args = Args::parse(); 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 workload.samples_per_sec > 1_000 { - bail!("samples per second must be at most 1000"); + if !workload.metrics.is_empty() { + 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 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"); + if !verify_facts(&workload)? { + bail!("the generated data does not match the description"); } return Ok(()); } @@ -325,16 +415,23 @@ async fn main() -> Result<()> { bail!("--batch-rows and --concurrency must be positive"); } - let sink = Arc::new(Sink::new(&args, sink_kind, url, &workload)?); + let sink = Arc::new(Sink::new(&args, sink_kind, url, &workload, batch_rows)?); 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(); + let metric_names = workload.names(); + let series = if workload.metrics.is_empty() { + String::new() + } else { + format!( + " ({} series x {}/s)", + workload.num_series(), + workload.samples_per_sec + ) + }; eprintln!( - "paced_feeder: {sink_kind:?} -> {url}, {} ({} series) x {}/s = {} rows/s for {duration_secs}s, start_ms={start_ms}", + "paced_feeder: {sink_kind:?} -> {url}, {}{series} = {} rows/s for {duration_secs}s, start_ms={start_ms}", metric_names.join(", "), - workload.num_series(), - workload.samples_per_sec, workload.rows_per_tick(), ); @@ -351,22 +448,19 @@ async fn main() -> Result<()> { 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 + // tick lies in one window (see `tick_rows`) and is under one second // behind the watermark, within the allowed lateness. let mut tasks = tokio::task::JoinSet::new(); - 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 metric in 0..metric_names.len() { + let metric_rows = Arc::new(workload.tick_rows(metric, tick, start_ms)); + let batch_rows = sink.targets[metric].batch_rows; 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; + let sent = sink.send(metric, &metric_rows, begin..end).await; sent.map(|bytes| (end - begin, bytes)) }); } @@ -424,7 +518,7 @@ async fn main() -> Result<()> { pareto_scale: grouped.then_some(args.pareto_scale), workload_table: args.workload_table.clone(), metrics: metric_names, - samples_per_sec: workload.samples_per_sec, + samples_per_sec: (!workload.metrics.is_empty()).then_some(workload.samples_per_sec), seed: args.seed, start_ms, duration_secs, diff --git a/asap-tools/data-sources/paced-feeder/src/records.rs b/asap-tools/data-sources/paced-feeder/src/records.rs new file mode 100644 index 00000000..86148c10 --- /dev/null +++ b/asap-tools/data-sources/paced-feeder/src/records.rs @@ -0,0 +1,880 @@ +//! `records` metrics of a workload table: a stream of records, each with a +//! value of every label and of every value field, e.g. sketch-bench's `http` +//! (`region`, `service`, `endpoint`, `status`; `latency`, `user_id`). +//! +//! Record `n` depends only on `(metric seed, n)`: every draw is a stateless +//! hash of the seed, the record index and the draw's stream (label `i` is +//! stream `i`, value field `j` is stream `labels + j`, the burst coin is +//! stream `labels + fields`). The data matches sketch-bench's +//! `label_shares()`: +//! +//! - **Labels.** Each label is drawn independently (a child given its parent) +//! from Zipf(`skew`) over its `cardinality` values (a child's: under its +//! parent), the value of index `i` having weight `(i+1)^-skew`, so `0` is the +//! heaviest. Values are named as for gauges: `"3"`, or a child's `"3.17"`. +//! - **Keys** (`{"zipf": θ, "keys": K}`): the key of Zipf(θ) rank `r` over `K` +//! keys is `r`. +//! - **Burst** (`"burst": {"label", "value", "share", "keys"}` on a keys field): +//! with probability `share` a record is a burst record: its `label` is +//! `value` (a child index for a child label) and its key is uniform over the +//! `keys` burst keys `K, K+1, …`, disjoint from the background keys. Other +//! records draw both as above, so `label`'s shares are Zipf mixed with the +//! burst: `(1 - share)·p_i`, plus `share` at `value`. +//! - **Pareto** (`{"pareto": a}`): `PARETO_SCALE · U^(-1/a)`, times, with +//! `"scaled_by": L`, the fixed factor of the record's value of `L` +//! (log-uniform in `SCALE_RANGE`, from a hash of the value and the feeder's +//! seed), times, with `"anomaly": {"label": L, "factor": f}`, `f` when the +//! record's value of `L` is its first one (`0`, or `0.0` for a child: the +//! heaviest). Scaling keeps the tail index `a`. + +use crate::table::{fnv1a, is_identifier, label_parents}; +use crate::workload::{splitmix64, unit_uniform, LabelSpec}; +use anyhow::{bail, ensure, Context, Result}; +use serde::Deserialize; +use serde_json::Value; +use std::collections::{BTreeMap, HashSet}; +use std::ops::Range; + +/// Records are timestamped to 10 ms slots and spread evenly across them. +pub const SLOTS_PER_SEC: u32 = 100; +const SLOT_MS: i64 = 1_000 / SLOTS_PER_SEC as i64; +/// Scale of `pareto` values, as in sketch-bench's +/// `configs/datagen/hydra_http_latency.yaml` (`scale: 1000.0`). +pub const PARETO_SCALE: f64 = 1_000.0; +/// `[lo, hi]` of the `scaled_by` factors, as in the same spec's `scale_range`. +pub const SCALE_RANGE: [f64; 2] = [1.0, 10.0]; +/// Keys are sent as `Float64`, which holds integers exactly up to 2^53. +const MAX_KEY: u64 = 1 << 53; +/// The facts check counts shares over this many records. +pub const SHARE_SAMPLE_RECORDS: u64 = 1_000_000; +/// A share's check fails if it is off by more than this many standard errors +/// of a share estimated from the sample, `sqrt(p(1-p)/n)`. +pub const SHARE_TOLERANCE_SIGMAS: f64 = 5.0; +/// The facts check shows each label's heaviest values. +const TOP_SHARES: usize = 3; + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct RecordsSpec { + #[serde(rename = "type")] + _kind: String, + labels: Vec, + records_per_sec: u64, + values: BTreeMap, +} + +/// Either `zipf` and `keys` (with an optional `burst`), or `pareto` (with +/// optional `scaled_by` and `anomaly`). +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct FieldSpec { + zipf: Option, + keys: Option, + burst: Option, + pareto: Option, + scaled_by: Option, + anomaly: Option, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct BurstSpec { + label: String, + value: u64, + share: f64, + keys: u64, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct AnomalySpec { + label: String, + factor: f64, +} + +/// Zipf over `n` values, value `i` with weight `(i+1)^-theta`, drawn by +/// inverse transform: the first value whose CDF reaches `u`. A guide table +/// narrows the binary search to the values whose CDF crosses `u`'s bucket, so +/// a draw over 1e6 keys touches a few cache lines instead of 20. +#[derive(Debug, Clone, PartialEq)] +pub struct Zipf { + cdf: Vec, + /// `guide[j]`: the first value whose CDF reaches `j / n`, for `j` in `0..=n`. + guide: Vec, +} + +/// Zipf(`theta`) shares of `n` values, the value of index `i` with weight +/// `(i+1)^-theta`. +pub fn zipf_shares(n: u64, theta: f64) -> Vec { + let weights: Vec = (0..n).map(|i| ((i + 1) as f64).powf(-theta)).collect(); + let total: f64 = weights.iter().sum(); + weights.into_iter().map(|w| w / total).collect() +} + +impl Zipf { + pub fn new(n: u64, theta: f64) -> Self { + let mut cdf: Vec = zipf_shares(n, theta) + .into_iter() + .scan(0.0, |sum, p| { + *sum += p; + Some(*sum) + }) + .collect(); + // Rounding can leave the sum just under 1; any `u` must find a value. + *cdf.last_mut().expect("a Zipf has values") = 1.0; + let buckets = cdf.len(); + let mut guide = Vec::with_capacity(buckets + 1); + let mut i = 0; + for j in 0..=buckets { + let bound = j as f64 / buckets as f64; + while cdf[i] < bound { + i += 1; + } + guide.push(i as u32); + } + Self { cdf, guide } + } + + /// The value index for `u` in (0, 1]. + pub fn draw(&self, u: f64) -> u64 { + // `u` is in bucket `j`: (j/n, (j+1)/n], so its value lies in + // `guide[j]..=guide[j+1]`; one bucket more on each side absorbs + // rounding in `u·n`. + let buckets = self.cdf.len(); + let j = ((u * buckets as f64).ceil() as usize).clamp(1, buckets) - 1; + let lo = self.guide[j.saturating_sub(1)] as usize; + let hi = self.guide[(j + 2).min(buckets)] as usize; + (lo + self.cdf[lo..=hi].partition_point(|&c| c < u)) as u64 + } +} + +#[derive(Debug, Clone, PartialEq)] +pub struct Burst { + /// The label and value (a child index for a child label) of burst records. + pub label: usize, + pub value: u64, + pub share: f64, + /// The keys field and its number of burst keys, numbered after its keys. + pub field: usize, + pub keys: u64, +} + +#[derive(Debug, Clone, PartialEq)] +enum Field { + Keys { + zipf: Zipf, + keys: u64, + }, + Pareto { + shape: f64, + /// The `scaled_by` label and each of its values' factor. + scaled_by: Option<(usize, Vec)>, + /// The anomaly label and factor; the anomalous value is the label's first. + anomaly: Option<(usize, f64)>, + }, +} + +#[derive(Debug, Clone, PartialEq)] +pub struct RecordMetric { + pub name: String, + pub labels: Vec, + /// Every value of each label, a child's across all parents: value `v` of a + /// record is `label_values[label][v]`. + pub label_values: Vec>, + parents: Vec>, + label_zipfs: Vec, + /// Value field names, in name order. + pub fields: Vec, + field_kinds: Vec, + pub burst: Option, + pub records_per_sec: u64, + seed: u64, +} + +/// Records `indices` of a metric, with event time from `start_ms`. Records +/// are generated as they are encoded, so each request's are generated by its +/// own task. +#[derive(Debug, Clone, PartialEq)] +pub struct Records { + pub start_ms: i64, + pub indices: Range, +} + +impl Records { + pub fn len(&self) -> usize { + (self.indices.end - self.indices.start) as usize + } +} + +impl RecordMetric { + /// A `records` metric from its table description. `records_per_sec` + /// overrides the table's rate; `seed` is the feeder's. + pub fn from_table( + name: &str, + spec: &Value, + records_per_sec: Option, + seed: u64, + ) -> Result { + let spec: RecordsSpec = serde_json::from_value(spec.clone())?; + let labels = spec.labels; + let parents = label_parents(&labels)?; + let label_index = |name: &str| -> Result { + labels + .iter() + .position(|l| l.name == name) + .with_context(|| format!("no label `{name}`")) + }; + let mut label_zipfs = Vec::new(); + for label in &labels { + let skew = label + .skew + .with_context(|| format!("label `{}` needs a `skew`", label.name))?; + ensure!(skew >= 0.0, "label `{}`: skew must be ≥ 0", label.name); + ensure!( + label.cardinality <= u64::from(u32::MAX), + "label `{}`: too many values", + label.name + ); + label_zipfs.push(Zipf::new(label.cardinality, skew)); + } + let label_values = label_values(&labels, &parents)?; + + ensure!(!spec.values.is_empty(), "a records metric needs values"); + let mut burst = None; + let mut fields = Vec::new(); + let mut field_kinds = Vec::new(); + for (field, value) in spec.values { + ensure!( + is_identifier(&field) && field != "ts" && label_index(&field).is_err(), + "value `{field}` must match [A-Za-z_][A-Za-z0-9_]* and not be `ts` or a label" + ); + let kind = match value { + FieldSpec { + zipf: Some(theta), + keys: Some(keys), + burst: field_burst, + pareto: None, + scaled_by: None, + anomaly: None, + } => { + ensure!(theta >= 0.0, "value `{field}`: zipf must be ≥ 0"); + ensure!( + keys > 0 && keys <= MAX_KEY, + "value `{field}`: keys must be in [1, 2^53]" + ); + if let Some(b) = field_burst { + ensure!(burst.is_none(), "at most one value may have a burst"); + let label = label_index(&b.label) + .with_context(|| format!("value `{field}`: burst"))?; + ensure!( + b.value < labels[label].cardinality, + "value `{field}`: the burst value is not a value of `{}`", + b.label + ); + ensure!( + b.share > 0.0 && b.share < 1.0, + "value `{field}`: the burst share must be in (0, 1)" + ); + ensure!( + b.keys > 0 && keys.saturating_add(b.keys) <= MAX_KEY, + "value `{field}`: burst keys must be positive, and all keys ≤ 2^53" + ); + burst = Some(Burst { + label, + value: b.value, + share: b.share, + field: fields.len(), + keys: b.keys, + }); + } + Field::Keys { + zipf: Zipf::new(keys, theta), + keys, + } + } + FieldSpec { + zipf: None, + keys: None, + burst: None, + pareto: Some(shape), + scaled_by, + anomaly, + } => { + ensure!(shape > 0.0, "value `{field}`: pareto must be positive"); + let scaled_by = match scaled_by { + None => None, + Some(by) => { + let label = label_index(&by) + .with_context(|| format!("value `{field}`: scaled_by"))?; + let factors = label_values[label] + .iter() + .map(|v| scale_factor(v, seed)) + .collect(); + Some((label, factors)) + } + }; + let anomaly = match anomaly { + None => None, + Some(a) => { + ensure!( + a.factor > 0.0, + "value `{field}`: the anomaly factor must be positive" + ); + let label = label_index(&a.label) + .with_context(|| format!("value `{field}`: anomaly"))?; + Some((label, a.factor)) + } + }; + Field::Pareto { + shape, + scaled_by, + anomaly, + } + } + _ => bail!( + "value `{field}` must be either `zipf` and `keys` (and `burst`), \ + or `pareto` (and `scaled_by`, `anomaly`)" + ), + }; + fields.push(field); + field_kinds.push(kind); + } + let records_per_sec = records_per_sec.unwrap_or(spec.records_per_sec); + ensure!(records_per_sec > 0, "records per second must be positive"); + Ok(Self { + name: name.to_string(), + labels, + label_values, + parents, + label_zipfs, + fields, + field_kinds, + burst, + records_per_sec, + seed: splitmix64(seed ^ fnv1a(name)), + }) + } + + /// Record `index`: writes each label's value index to `labels` and each + /// field's value to `values`. + pub fn record(&self, index: u64, labels: &mut [u32], values: &mut [f64]) { + let num_labels = self.labels.len() as u64; + let u = |stream: u64| unit_uniform(self.seed, stream, index); + let burst = self.burst.as_ref().filter(|b| { + let stream = num_labels + self.fields.len() as u64; + u(stream) <= b.share + }); + for (i, zipf) in self.label_zipfs.iter().enumerate() { + let child = match burst { + Some(b) if b.label == i => b.value, + _ => zipf.draw(u(i as u64)), + }; + labels[i] = match self.parents[i] { + None => child, + Some(p) => u64::from(labels[p]) * self.labels[i].cardinality + child, + } as u32; + } + for (j, kind) in self.field_kinds.iter().enumerate() { + let u = u(num_labels + j as u64); + values[j] = match kind { + Field::Keys { zipf, keys } => match burst { + // `u` in (0, 1] maps to 0..burst keys. + Some(b) if b.field == j => { + (keys + (u * b.keys as f64).ceil() as u64 - 1) as f64 + } + _ => zipf.draw(u) as f64, + }, + Field::Pareto { + shape, + scaled_by, + anomaly, + } => { + let mut value = PARETO_SCALE * u.powf(-1.0 / shape); + if let Some((label, factors)) = scaled_by { + value *= factors[labels[*label] as usize]; + } + if let Some((label, factor)) = anomaly { + if labels[*label] == 0 { + value *= factor; + } + } + value + } + }; + } + } + + /// The records of one-second tick `tick`, numbered from + /// `tick · records_per_sec`, with event time starting at `start_ms`. + pub fn tick(&self, tick: u64, start_ms: i64) -> Records { + let first = tick * self.records_per_sec; + Records { + start_ms, + indices: first..first + self.records_per_sec, + } + } + + /// The event time of record `index`: the end of its 10 ms slot, so a + /// tick starting at `t` has its records in `(t, t + 1 s]`. A second's + /// records are spread evenly over its slots, slot `s` holding offsets + /// `[s · r / 100, (s + 1) · r / 100)`. + pub fn timestamp_ms(&self, index: u64, start_ms: i64) -> i64 { + let (tick, offset) = (index / self.records_per_sec, index % self.records_per_sec); + let slot = (u64::from(SLOTS_PER_SEC) * (offset + 1) - 1) / self.records_per_sec; + start_ms + tick as i64 * 1_000 + (slot as i64 + 1) * SLOT_MS + } + + /// Each value's share of the records (a child's: given its parent), as + /// sketch-bench's `label_shares()`: Zipf(`skew`), mixed with the burst. + pub fn label_shares(&self, label: usize) -> Vec { + let spec = &self.labels[label]; + let mut shares = zipf_shares(spec.cardinality, spec.skew.unwrap_or_default()); + if let Some(b) = self.burst.as_ref().filter(|b| b.label == label) { + for p in &mut shares { + *p *= 1.0 - b.share; + } + shares[b.value as usize] += b.share; + } + shares + } + + /// The share checks of the facts check over records `0..n`: each label's + /// heaviest values (a child's index pooled over parents, so its share given + /// the parent), the burst value and the share of burst records. + pub fn share_checks(&self, n: u64) -> ShareReport { + let (num_labels, num_fields) = (self.labels.len(), self.fields.len()); + let mut counts: Vec> = self + .labels + .iter() + .map(|l| vec![0; l.cardinality as usize]) + .collect(); + let (mut burst_records, mut burst_keys) = (0u64, HashSet::new()); + let (mut labels, mut values) = (vec![0u32; num_labels], vec![0.0; num_fields]); + for index in 0..n { + self.record(index, &mut labels, &mut values); + for (i, &v) in labels.iter().enumerate() { + counts[i][(u64::from(v) % self.labels[i].cardinality) as usize] += 1; + } + if let Some(b) = &self.burst { + let Field::Keys { keys, .. } = self.field_kinds[b.field] else { + unreachable!("a burst is on a keys field"); + }; + let key = values[b.field] as u64; + if key >= keys { + burst_records += 1; + burst_keys.insert(key); + } + } + } + let check = |label: String, value: String, expected: f64, count: u64| ShareCheck { + label, + value, + expected, + actual: count as f64 / n as f64, + tolerance: SHARE_TOLERANCE_SIGMAS * (expected * (1.0 - expected) / n as f64).sqrt(), + }; + let mut checks = Vec::new(); + for (i, label) in self.labels.iter().enumerate() { + let shares = self.label_shares(i); + let mut shown: Vec = (0..shares.len()).collect(); + shown.sort_by(|&a, &b| shares[b].total_cmp(&shares[a]).then(a.cmp(&b))); + shown.truncate(TOP_SHARES); + if let Some(b) = self.burst.as_ref().filter(|b| b.label == i) { + if !shown.contains(&(b.value as usize)) { + shown.push(b.value as usize); + } + } + for v in shown { + let value = match &label.child_of { + None => v.to_string(), + Some(parent) => format!("*.{v} (given {parent})"), + }; + checks.push(check(label.name.clone(), value, shares[v], counts[i][v])); + } + } + let burst = self.burst.as_ref().map(|b| { + checks.push(check( + "burst".to_string(), + "records".to_string(), + b.share, + burst_records, + )); + (burst_keys.len() as u64, b.keys) + }); + ShareReport { + records: n, + checks, + burst_keys: burst, + } + } +} + +#[derive(Debug, Clone, PartialEq)] +pub struct ShareCheck { + pub label: String, + pub value: String, + pub expected: f64, + pub actual: f64, + pub tolerance: f64, +} + +impl ShareCheck { + pub fn holds(&self) -> bool { + (self.actual - self.expected).abs() <= self.tolerance + } +} + +#[derive(Debug, Clone, PartialEq)] +pub struct ShareReport { + pub records: u64, + pub checks: Vec, + /// Distinct burst keys among the records, and the burst's keys. + pub burst_keys: Option<(u64, u64)>, +} + +/// Every value of each label: a root's `0`, `1`, …; a child's, for each parent +/// value in order, `parent.0`, `parent.1`, … +fn label_values(labels: &[LabelSpec], parents: &[Option]) -> Result>> { + let mut values: Vec> = Vec::with_capacity(labels.len()); + for (label, parent) in labels.iter().zip(parents) { + let children = 0..label.cardinality; + let label_values: Vec = match parent { + None => children.map(|i| i.to_string()).collect(), + Some(p) => values[*p] + .iter() + .flat_map(|parent| children.clone().map(move |i| format!("{parent}.{i}"))) + .collect(), + }; + ensure!( + label_values.len() <= u32::MAX as usize, + "label `{}` has too many values", + label.name + ); + values.push(label_values); + } + Ok(values) +} + +/// The fixed `scaled_by` factor of label value `value`, as aqpbm-datagen's +/// `scale_factor`: a SplitMix64 hash of the value's FNV-1a and the seed, as +/// `u` in [0, 1), mapped to `lo·(hi/lo)^u`. +pub fn scale_factor(value: &str, seed: u64) -> f64 { + let [lo, hi] = SCALE_RANGE; + let u = (splitmix64(fnv1a(value) ^ seed) >> 11) as f64 / (1u64 << 53) as f64; + lo * (hi / lo).powf(u) +} + +#[cfg(test)] +pub mod tests { + use super::*; + use serde_json::json; + + /// sketch-bench's `http` schema, as an e2e table describes it. + pub fn http_spec() -> Value { + json!({"type": "records", "records_per_sec": 2_000_000, + "labels": [{"name": "region", "cardinality": 4, "skew": 0.5}, + {"name": "service", "cardinality": 25, "skew": 1.1}, + {"name": "endpoint", "cardinality": 25, "child_of": "service", "skew": 1.1}, + {"name": "status", "cardinality": 4, "skew": 2.0}], + "values": {"latency": {"pareto": 2.0, "scaled_by": "service", + "anomaly": {"label": "endpoint", "factor": 10}}, + "user_id": {"zipf": 0.8, "keys": 1_000_000}}}) + } + + /// sketch-bench's `flows` schema, as an e2e table describes it. + pub fn flows_spec() -> Value { + json!({"type": "records", "records_per_sec": 4_000_000, + "labels": [{"name": "dst_subnet", "cardinality": 1000, "skew": 1.1}, + {"name": "dst_port", "cardinality": 1000, "skew": 1.2}, + {"name": "proto", "cardinality": 3, "skew": 0.0}], + "values": {"src_ip": {"zipf": 0.0, "keys": 1_000_000, + "burst": {"label": "dst_subnet", "value": 999, + "share": 0.05, "keys": 10_000}}}}) + } + + fn metric(name: &str, spec: &Value) -> RecordMetric { + RecordMetric::from_table(name, spec, None, 42).unwrap() + } + + /// Records `0..n` as (label value indices, values). + fn records(m: &RecordMetric, n: u64) -> Vec<(Vec, Vec)> { + (0..n) + .map(|index| { + let mut labels = vec![0; m.labels.len()]; + let mut values = vec![0.0; m.fields.len()]; + m.record(index, &mut labels, &mut values); + (labels, values) + }) + .collect() + } + + fn within(actual: f64, expected: f64, n: usize) -> bool { + let sigma = (expected * (1.0 - expected) / n as f64).sqrt(); + (actual - expected).abs() <= SHARE_TOLERANCE_SIGMAS * sigma + } + + #[test] + fn zipf_guide_table_draws_as_a_plain_cdf_search() { + for (n, theta) in [(1, 0.0), (4, 2.0), (25, 1.1), (1_000, 0.0), (100_000, 0.8)] { + let zipf = Zipf::new(n, theta); + let plain = |u: f64| zipf.cdf.partition_point(|&c| c < u) as u64; + // Hashed draws, every bucket boundary and the ends of (0, 1]. + let us = (0..20_000) + .map(|i| unit_uniform(7, 0, i)) + .chain((1..=n.min(1_000)).map(|j| j as f64 / n as f64)) + .chain([f64::MIN_POSITIVE, 1e-12, 1.0 - 1e-16, 1.0]); + for u in us { + assert_eq!(zipf.draw(u), plain(u), "n={n} theta={theta} u={u}"); + } + } + } + + #[test] + fn label_shares_match_sketch_bench() { + // `label_shares()` of scripts/export_autosketch_eval_table.py. + let http = metric("http", &http_spec()); + let round = |p: f64| (p * 1e5).round() / 1e5; + let top3 = |m: &RecordMetric, l| -> Vec { + m.label_shares(l)[..3].iter().map(|&p| round(p)).collect() + }; + assert_eq!(top3(&http, 0), [0.35914, 0.25395, 0.20735]); + assert_eq!(top3(&http, 1), [0.29841, 0.13922, 0.08912]); + assert_eq!(top3(&http, 2), [0.29841, 0.13922, 0.08912]); + assert_eq!(top3(&http, 3), [0.70244, 0.17561, 0.07805]); + let flows = metric("flows", &flows_spec()); + assert_eq!(top3(&flows, 0), [0.17047, 0.07953, 0.05091]); + assert_eq!(round(flows.label_shares(0)[999]), 0.05009); + assert!((flows.label_shares(0).iter().sum::() - 1.0).abs() < 1e-9); + } + + #[test] + fn labels_follow_their_shares_and_children_their_parent() { + let http = metric("http", &http_spec()); + let n = 200_000; + let records = records(&http, n); + for (l, label) in http.labels.iter().enumerate() { + let shares = http.label_shares(l); + for (v, &expected) in shares.iter().enumerate().take(5) { + let count = records + .iter() + .filter(|(labels, _)| labels[l] as u64 % label.cardinality == v as u64) + .count(); + let actual = count as f64 / n as f64; + assert!( + within(actual, expected, n as usize), + "{}={v}: {actual} vs {expected}", + label.name + ); + } + } + // A child value names its parent, and given a parent the children + // follow their own Zipf shares. + let (service, endpoint) = (1, 2); + for (labels, _) in &records { + let name = &http.label_values[endpoint][labels[endpoint] as usize]; + let parent = &http.label_values[service][labels[service] as usize]; + assert!( + name.starts_with(&format!("{parent}.")), + "{name} under {parent}" + ); + } + let under = |s: u32| -> Vec { + records + .iter() + .filter(|(labels, _)| labels[service] == s) + .map(|(labels, _)| labels[endpoint] % 25) + .collect() + }; + let shares = http.label_shares(endpoint); + for s in [0, 3] { + let children = under(s); + for (i, &expected) in shares.iter().enumerate().take(3) { + let actual = children.iter().filter(|&&c| c == i as u32).count() as f64 + / children.len() as f64; + assert!( + within(actual, expected, children.len()), + "service {s} endpoint {i}" + ); + } + } + } + + #[test] + fn burst_records_hit_the_burst_value_from_their_own_keys() { + let flows = metric("flows", &flows_spec()); + let n = 400_000; + let records = records(&flows, n); + let (keys, burst_keys) = (1_000_000.0, 10_000.0); + let burst: Vec<_> = records.iter().filter(|(_, v)| v[0] >= keys).collect(); + assert!( + within(burst.len() as f64 / n as f64, 0.05, n as usize), + "{}", + burst.len() + ); + assert!(burst.iter().all(|(labels, _)| labels[0] == 999)); + assert!(burst + .iter() + .all(|(_, v)| v[0] < keys + burst_keys && v[0].fract() == 0.0)); + // About 20,000 draws over 10,000 keys see 1 - e^-2 of them. + let distinct: HashSet = burst.iter().map(|(_, v)| v[0] as u64).collect(); + assert!( + (8_400..=8_900).contains(&distinct.len()), + "{}", + distinct.len() + ); + // The background is uniform over its keys and every subnet. + let background: Vec<_> = records.iter().filter(|(_, v)| v[0] < keys).collect(); + let mean = background.iter().map(|(_, v)| v[0]).sum::() / background.len() as f64; + assert!((mean / keys - 0.5).abs() < 0.01, "{mean}"); + let subnet_999 = records.iter().filter(|(l, _)| l[0] == 999).count() as f64 / n as f64; + assert!( + within(subnet_999, flows.label_shares(0)[999], n as usize), + "{subnet_999}" + ); + } + + #[test] + fn latency_is_scaled_by_service_and_the_anomaly_endpoint() { + let http = metric("http", &http_spec()); + let (service, endpoint) = (1, 2); + let factors: Vec = http.label_values[service] + .iter() + .map(|v| scale_factor(v, 42)) + .collect(); + assert!(factors.iter().all(|f| (1.0..=10.0).contains(f))); + assert!(factors.windows(2).any(|w| w[0] != w[1])); + assert_ne!(scale_factor("3", 42), scale_factor("3", 43)); + // Every latency is exactly the Pareto draw of its stream, times its + // service's factor, times 10 on the first endpoint ("0.0"). + let latency_stream = http.labels.len() as u64; + for (index, (labels, values)) in records(&http, 20_000).iter().enumerate() { + let u = unit_uniform(http.seed, latency_stream, index as u64); + let anomaly = if http.label_values[endpoint][labels[endpoint] as usize] == "0.0" { + 10.0 + } else { + 1.0 + }; + let expected = + PARETO_SCALE * u.powf(-0.5) * factors[labels[service] as usize] * anomaly; + assert_eq!(values[0], expected); + } + } + + #[test] + fn keys_follow_zipf_ranks() { + let http = metric("http", &http_spec()); + let n = 200_000; + let user_ids: Vec = records(&http, n).iter().map(|(_, v)| v[1]).collect(); + assert!(user_ids.iter().all(|&k| k < 1e6 && k.fract() == 0.0)); + let shares = zipf_shares(1_000_000, 0.8); + for key in [0.0, 1.0, 9.0] { + let actual = user_ids.iter().filter(|&&k| k == key).count() as f64 / n as f64; + assert!( + within(actual, shares[key as usize], n as usize), + "key {key}: {actual}" + ); + } + } + + #[test] + fn a_tick_spreads_its_records_over_slots_ending_in_the_second() { + // 250 records/s: slots of 2 or 3 records. + let m = RecordMetric::from_table("http", &http_spec(), Some(250), 42).unwrap(); + let tick = m.tick(2, 10_000); + assert_eq!(tick.indices, 500..750); + let mut per_timestamp: BTreeMap = BTreeMap::new(); + for index in tick.indices { + *per_timestamp + .entry(m.timestamp_ms(index, 10_000)) + .or_default() += 1; + } + // Slot s of tick 2 is at 12_000 + 10 · (s + 1): within (12 s, 13 s]. + assert!(per_timestamp + .keys() + .copied() + .eq((1..=100).map(|s| 12_000 + 10 * s))); + assert!(per_timestamp.values().all(|n| (2..=3).contains(n))); + // The first record of the next tick starts the next second's slots. + assert_eq!(m.timestamp_ms(750, 10_000), 13_010); + } + + #[test] + fn records_depend_only_on_seed_name_and_index() { + let a = metric("http", &http_spec()); + assert_eq!( + records(&a, 100), + records(&metric("http", &http_spec()), 100) + ); + // Another metric of the same shape draws other records. + assert_ne!( + records(&a, 100), + records(&metric("data_1_http", &http_spec()), 100) + ); + let reseeded = RecordMetric::from_table("http", &http_spec(), None, 43).unwrap(); + assert_ne!(records(&a, 100), records(&reseeded, 100)); + } + + #[test] + fn share_checks_pass_and_catch_a_wrong_skew() { + for spec in [http_spec(), flows_spec()] { + let report = metric("m", &spec).share_checks(200_000); + assert!(report.checks.iter().all(ShareCheck::holds), "{report:?}"); + } + let report = metric("flows", &flows_spec()).share_checks(200_000); + let burst = report.checks.iter().find(|c| c.label == "burst").unwrap(); + assert_eq!(burst.expected, 0.05); + assert!(report + .checks + .iter() + .any(|c| c.label == "dst_subnet" && c.value == "999")); + // About 10,000 burst draws over 10,000 keys see 1 - 1/e of them. + let (seen, keys) = report.burst_keys.unwrap(); + assert!((6_000..=6_600).contains(&seen) && keys == 10_000, "{seen}"); + + // Records drawn with a flatter service skew than described. + let mut wrong = metric("http", &http_spec()); + wrong.label_zipfs[1] = Zipf::new(25, 0.5); + let report = wrong.share_checks(200_000); + let service_0 = report + .checks + .iter() + .find(|c| c.label == "service" && c.value == "0") + .unwrap(); + assert!(!service_0.holds(), "{service_0:?}"); + } + + #[test] + fn rejects_malformed_records_specs() { + let err = |spec: Value| -> String { + format!( + "{:#}", + RecordMetric::from_table("m", &spec, None, 1).unwrap_err() + ) + }; + let mut spec = http_spec(); + spec["labels"][0].as_object_mut().unwrap().remove("skew"); + assert!(err(spec).contains("needs a `skew`")); + + let mut spec = http_spec(); + spec["values"]["latency"]["scaled_by"] = json!("nope"); + assert!(err(spec).contains("no label `nope`")); + + let mut spec = http_spec(); + spec["values"]["latency"]["zipf"] = json!(1.0); + assert!(err(spec).contains("must be either")); + + let mut spec = flows_spec(); + spec["values"]["src_ip"]["burst"]["value"] = json!(1000); + assert!(err(spec).contains("not a value of `dst_subnet`")); + + let mut spec = flows_spec(); + spec["values"]["dst_port"] = json!({"zipf": 0.0, "keys": 5}); + assert!(err(spec).contains("not be `ts` or a label")); + + let mut spec = flows_spec(); + spec["values"]["src_ip"]["keys"] = json!(1u64 << 53); + assert!(err(spec).contains("all keys ≤ 2^53")); + + assert!(RecordMetric::from_table("m", &http_spec(), Some(0), 1).is_err()); + } +} diff --git a/asap-tools/data-sources/paced-feeder/src/table.rs b/asap-tools/data-sources/paced-feeder/src/table.rs index e6f73329..8454f8b5 100644 --- a/asap-tools/data-sources/paced-feeder/src/table.rs +++ b/asap-tools/data-sources/paced-feeder/src/table.rs @@ -6,13 +6,16 @@ //! 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. +//! has the product of all cardinalities as series. `records` metrics are +//! generated by [`crate::records`]. +use crate::records::RecordMetric; 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}; +use std::sync::Arc; #[derive(Deserialize)] struct Table { @@ -67,11 +70,13 @@ struct CounterValues { } /// 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. +/// order, or only `only`. `samples_per_sec` overrides every gauge and +/// counter's rate, `records_per_sec` every records metric's. pub fn load( path: &str, only: Option<&str>, samples_per_sec: Option, + records_per_sec: Option, seed: u64, ) -> Result { let text = std::fs::read_to_string(path).with_context(|| format!("reading {path}"))?; @@ -81,23 +86,52 @@ pub fn load( bail!("{path}: `workloads` is empty"); }; let mut metrics = Vec::new(); + let mut records = 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}`"))?; + let context = || format!("metric `{name}`"); + ensure!( + is_identifier(name), + "metric `{name}`: the name must match [A-Za-z_][A-Za-z0-9_]* (it is also the ClickHouse table)" + ); + if spec.get("type").and_then(Value::as_str) == Some("records") { + let metric = RecordMetric::from_table(name, spec, records_per_sec, seed) + .with_context(context)?; + records.push(Arc::new(metric)); + continue; + } + let (metric, rate) = metric(name, spec, seed).with_context(context)?; metrics.push(metric); rates.push(rate); } if let Some(only) = only { - ensure!(!metrics.is_empty(), "{path}: no metric `{only}`"); + ensure!( + !metrics.is_empty() || !records.is_empty(), + "{path}: no metric `{only}`" + ); } ensure!( - !metrics.is_empty(), + !metrics.is_empty() || !records.is_empty(), "{path}: workloads[0].e2e.metrics is empty" ); + ensure!( + !metrics.is_empty() || samples_per_sec.is_none(), + "--samples-per-sec applies to gauges and counters; use --records-per-sec for records" + ); + ensure!( + !records.is_empty() || records_per_sec.is_none(), + "--records-per-sec applies to records metrics only" + ); + if metrics.is_empty() { + return Ok(Workload { + samples_per_sec: 0, + metrics, + records, + }); + } let samples_per_sec = match samples_per_sec { Some(rate) => rate, // One feeder sends every metric at one rate. @@ -110,19 +144,22 @@ pub fn load( Ok(Workload { samples_per_sec, metrics, + records, }) } 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"); + bail!("type `{kind}` is not supported; the feeder generates `gauge`, `counter` and `records` metrics"); } let spec: MetricSpec = serde_json::from_value(spec.clone())?; + if let Some(label) = spec.labels.iter().find(|l| l.skew.is_some()) { + bail!( + "label `{}`: `skew` applies to records metrics only; series skew is the value's `zipf`", + label.name + ); + } 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`")?; @@ -164,8 +201,41 @@ fn metric(name: &str, spec: &Value, seed: u64) -> Result<(Metric, u32)> { /// 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>> { + let parents = label_parents(labels)?; + 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) +} + +/// Validates `labels` and returns the index of each label's parent, which +/// must come earlier. +pub fn label_parents(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; @@ -194,34 +264,7 @@ pub fn series_labels(labels: &[LabelSpec]) -> Result>> { ), }); } - 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) + Ok(parents) } /// Each series' value scale for Zipf(`theta`) skew over the series: the @@ -306,7 +349,7 @@ fn closed_cardinality(labels: &[LabelSpec], subset: &[usize]) -> u64 { .fold(1u64, u64::saturating_mul) } -fn is_identifier(name: &str) -> bool { +pub fn is_identifier(name: &str) -> bool { let mut chars = name.chars(); chars .next() @@ -315,7 +358,7 @@ fn is_identifier(name: &str) -> bool { } /// 64-bit FNV-1a, to derive a per-name seed. -fn fnv1a(s: &str) -> u64 { +pub fn fnv1a(s: &str) -> u64 { s.bytes().fold(0xcbf2_9ce4_8422_2325, |h, b| { (h ^ u64::from(b)).wrapping_mul(0x0000_0100_0000_01b3) }) @@ -331,6 +374,7 @@ mod tests { name: name.to_string(), cardinality, child_of: child_of.map(str::to_string), + skew: None, } } @@ -458,7 +502,7 @@ mod tests { #[test] fn loads_gauge_and_counter_from_a_table() { let path = write_table(classic_metrics()); - let w = load(&path, None, None, 1).unwrap(); + let w = load(&path, None, 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]); @@ -478,22 +522,92 @@ mod tests { assert_eq!(data.scales, total.scales); assert_ne!(data.seed, total.seed); - let only = load(&path, Some("data_total"), Some(2), 1).unwrap(); + let only = load(&path, Some("data_total"), Some(2), None, 1).unwrap(); assert_eq!(only.metrics.len(), 1); assert_eq!(only.samples_per_sec, 2); - assert!(load(&path, Some("nope"), None, 1).is_err()); + assert!(load(&path, Some("nope"), None, 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}"); - + fn rejects_mismatched_keys() { 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()); + let err = format!( + "{:#}", + load(&path, Some("data"), None, None, 1).unwrap_err() + ); assert!(err.contains("must equal the number of series"), "{err}"); } + + fn all_metrics() -> Value { + let mut metrics = classic_metrics(); + metrics["http"] = crate::records::tests::http_spec(); + metrics["flows"] = crate::records::tests::flows_spec(); + metrics + } + + #[test] + fn loads_records_metrics() { + let mut metrics = all_metrics(); + metrics + .as_object_mut() + .unwrap() + .retain(|_, m| m["type"] == "records"); + let path = write_table(metrics); + let w = load(&path, None, None, None, 1).unwrap(); + assert!(w.metrics.is_empty()); + assert_eq!(w.names(), ["flows", "http"]); + assert_eq!(w.rows_per_tick(), 6_000_000); + let scaled = load(&path, Some("http"), None, Some(500), 1).unwrap(); + assert_eq!(scaled.names(), ["http"]); + assert_eq!(scaled.records[0].records_per_sec, 500); + assert_eq!(scaled.records[0].fields, ["latency", "user_id"]); + } + + #[test] + fn one_feeder_sends_both_kinds_of_metric() { + let path = write_table(all_metrics()); + let both = load(&path, None, Some(2), Some(500), 1).unwrap(); + assert_eq!((both.metrics.len(), both.records.len()), (2, 2)); + assert_eq!(both.samples_per_sec, 2); + assert!(both.records.iter().all(|r| r.records_per_sec == 500)); + // `--metric` picks one of either kind. + assert_eq!( + load(&path, Some("http"), None, None, 1) + .unwrap() + .records + .len(), + 1 + ); + assert_eq!( + load(&path, Some("data"), None, None, 1) + .unwrap() + .metrics + .len(), + 1 + ); + // Each rate override applies to its own kind only. + assert!(load(&path, Some("http"), Some(10), None, 1).is_err()); + assert!(load(&path, Some("data"), None, Some(10), 1).is_err()); + } + + #[test] + fn rejects_unknown_types_and_skew_on_series_labels() { + let path = write_table(json!({"x": {"type": "histogram", "labels": []}})); + let err = format!("{:#}", load(&path, None, None, None, 1).unwrap_err()); + assert!(err.contains("type `histogram` is not supported"), "{err}"); + + let mut metrics = classic_metrics(); + metrics["data"]["labels"][0]["skew"] = json!(1.0); + let path = write_table(metrics); + let err = format!( + "{:#}", + load(&path, Some("data"), None, None, 1).unwrap_err() + ); + assert!( + err.contains("`skew` applies to records metrics only"), + "{err}" + ); + } } diff --git a/asap-tools/data-sources/paced-feeder/src/workload.rs b/asap-tools/data-sources/paced-feeder/src/workload.rs index de8f5d31..20567c85 100644 --- a/asap-tools/data-sources/paced-feeder/src/workload.rs +++ b/asap-tools/data-sources/paced-feeder/src/workload.rs @@ -1,11 +1,14 @@ -//! Seeded synthetic workload: one or more metrics, each a set of labelled -//! series emitting `samples_per_sec` evenly spaced samples per second. +//! Seeded synthetic workload: either one or more metrics, each a set of +//! labelled series emitting `samples_per_sec` evenly spaced samples per +//! second, or one or more `records` metrics (see [`crate::records`]). //! //! 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. +use crate::records::{RecordMetric, Records}; use serde::Deserialize; +use std::sync::Arc; /// One generated sample of one metric. `series` indexes into [`Metric::series`]. #[derive(Debug, Clone, Copy, PartialEq)] @@ -15,8 +18,25 @@ pub struct Row { pub value: f64, } +/// One tick's rows of one metric: samples of series, or records. +#[derive(Debug, Clone, PartialEq)] +pub enum Rows { + Samples(Vec), + Records(Records), +} + +impl Rows { + pub fn len(&self) -> usize { + match self { + Rows::Samples(rows) => rows.len(), + Rows::Records(records) => records.len(), + } + } +} + /// A label with `cardinality` values, or, with `child_of`, that many values -/// under each value of the parent label. +/// under each value of the parent label. `skew` (records metrics only) is the +/// Zipf exponent its values are drawn with. #[derive(Debug, Clone, PartialEq, Deserialize)] #[serde(deny_unknown_fields)] pub struct LabelSpec { @@ -24,6 +44,8 @@ pub struct LabelSpec { pub cardinality: u64, #[serde(default)] pub child_of: Option, + #[serde(default)] + pub skew: Option, } #[derive(Debug, Clone, PartialEq)] @@ -51,10 +73,14 @@ pub struct Metric { pub seed: u64, } +/// Gauge and counter `metrics`, or `records` metrics, never both: they have +/// different sub-second slots. #[derive(Debug, Clone)] pub struct Workload { + /// Samples per second per series of `metrics`; unused for `records`. pub samples_per_sec: u32, pub metrics: Vec, + pub records: Vec>, } impl Workload { @@ -82,6 +108,7 @@ impl Workload { name: name.to_string(), cardinality: cardinality as u64, child_of: None, + skew: None, }; let metric = Metric { name: name.to_string(), @@ -100,6 +127,7 @@ impl Workload { Self { samples_per_sec, metrics: vec![metric], + records: Vec::new(), } } @@ -107,15 +135,42 @@ impl Workload { self.metrics.iter().map(|m| m.series.len()).sum() } + /// Series samples or records per second. pub fn rows_per_tick(&self) -> usize { self.num_series() * self.samples_per_sec as usize + + self + .records + .iter() + .map(|r| r.records_per_sec as usize) + .sum::() + } + + /// The metric names, gauges and counters first, then records. + pub fn names(&self) -> Vec { + let series = self.metrics.iter().map(|m| m.name.clone()); + series + .chain(self.records.iter().map(|r| r.name.clone())) + .collect() + } + + /// Rows of the `target`-th metric of [`Self::names`] for one-second tick + /// `tick`, with event time in `(t, t + 1 s]`, where `t` is + /// `start_ms + tick · 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, target: usize, tick: u64, start_ms: i64) -> Rows { + match target.checked_sub(self.metrics.len()) { + None => Rows::Samples( + (0..self.samples_per_sec) + .flat_map(|slot| self.series_rows(target, tick, slot, start_ms)) + .collect(), + ), + Some(r) => Rows::Records(self.records[r].tick(tick, start_ms)), + } } - /// 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 { + /// Samples of `metric` for sub-second slot `slot` of tick `tick`, one per + /// series. Slot `j` of `n` is at `t + (j + 1) · 1 s / n`. + pub fn series_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; @@ -151,7 +206,7 @@ impl Metric { /// 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 { +pub fn unit_uniform(seed: u64, series: u64, sample_index: u64) -> f64 { let x = splitmix64(seed ^ splitmix64(series ^ splitmix64(sample_index))); // Top 53 bits, shifted into (0, 1] so the Pareto transform never divides by zero. ((x >> 11) + 1) as f64 / (1u64 << 53) as f64 @@ -176,7 +231,7 @@ mod tests { 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(0, 5, slot, 10_000); + let rows = w.series_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)); @@ -187,7 +242,7 @@ mod tests { fn values_depend_only_on_seed_series_and_sample_index() { let w = workload(); let values = |w: &Workload, start_ms| -> Vec { - w.slot_rows(0, 4, 1, start_ms) + w.series_rows(0, 4, 1, start_ms) .iter() .map(|r| r.value) .collect() From 0222064e313e20e0c6def734a3812a9dddfc774d Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 13 Oct 2026 02:04:11 +0000 Subject: [PATCH 2/2] fix(asap-tools): cap Zipf sizes and check records' label coverage cheaply A Zipf keeps a dense CDF and a u32 guide entry per value, so a records label's cardinality and a field's keys are capped at 1e8 (1.2 GB), with an error naming the cap; burst keys are drawn uniformly and keep only the 2^53 bound. `--verify-facts` no longer enumerates every label combination of a records metric: per label subset it prints the possible tuples and how many the sampled 1e6 records hold, failing only if they hold more. The share checks are unchanged. The `Workload` doc now says one feeder sends both kinds of metric. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01AnJ7LocWxguh3q4QWkCYae --- .../data-sources/paced-feeder/README.md | 14 ++-- .../data-sources/paced-feeder/src/main.rs | 27 +++--- .../data-sources/paced-feeder/src/records.rs | 84 +++++++++++++++++-- .../data-sources/paced-feeder/src/table.rs | 2 +- .../data-sources/paced-feeder/src/workload.rs | 3 +- 5 files changed, 104 insertions(+), 26 deletions(-) diff --git a/asap-tools/data-sources/paced-feeder/README.md b/asap-tools/data-sources/paced-feeder/README.md index fea7a119..3340763f 100644 --- a/asap-tools/data-sources/paced-feeder/README.md +++ b/asap-tools/data-sources/paced-feeder/README.md @@ -120,9 +120,10 @@ name), `n` and the draw, so records match sketch-bench's `label_shares()` 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 (for records metrics, among all label-value -combinations a record can have), and exits non-zero if any differ. For records -metrics it also draws records `0..1e6` and compares each label's three +number among the generated series, and exits non-zero if any differ. For +records metrics it draws records `0..1e6`, prints how many of the possible +tuples of each subset they hold (skew leaves rare ones unseen; more than +possible fails), and compares each label's three heaviest values' shares (a child's index pooled over parents, i.e. its share given the parent), the burst value's share and the share of burst records with `label_shares()`; a share fails if it is off by more than 5 standard errors, @@ -139,8 +140,9 @@ data_total: 10000 series … $ paced_feeder --workload-table synthetic-templatesmultigroup-shared1.json --verify-facts -flows: 4000000 records/s, 3000000 label combinations - labels expected generated +flows: 4000000 records/s + label-value tuples among 1000000 records + labels possible seen dst_subnet 1000 1000 … shares of 1000000 records (tolerance 5 standard errors) @@ -151,7 +153,7 @@ flows: 4000000 records/s, 3000000 label combinations … burst records 0.05000 0.05029 0.00109 burst keys: 9938 distinct of 10000 -http: 2000000 records/s, 10000 label combinations +http: 2000000 records/s … endpoint *.0 (given service) 0.29841 0.29855 0.00229 … diff --git a/asap-tools/data-sources/paced-feeder/src/main.rs b/asap-tools/data-sources/paced-feeder/src/main.rs index c466cf15..32413d05 100644 --- a/asap-tools/data-sources/paced-feeder/src/main.rs +++ b/asap-tools/data-sources/paced-feeder/src/main.rs @@ -314,16 +314,23 @@ fn verify_facts(workload: &Workload) -> Result { ok &= print_label_facts(&metric.labels, &metric.series); } for metric in &workload.records { - // Every combination of label values a record can have. - let domain = table::series_labels(&metric.labels)?; - println!( - "{}: {} records/s, {} label combinations", - metric.name, - metric.records_per_sec, - domain.len() - ); - ok &= print_label_facts(&metric.labels, &domain); - let report = metric.share_checks(records::SHARE_SAMPLE_RECORDS); + let n = records::SHARE_SAMPLE_RECORDS; + println!("{}: {} records/s", metric.name, metric.records_per_sec); + println!(" label-value tuples among {n} records"); + println!(" {:<40} {:>12} {:>12}", "labels", "possible", "seen"); + for fact in metric.label_coverage(n) { + // Every tuple a record has is possible; skew leaves some unseen. + let holds = fact.actual <= fact.expected; + ok &= holds; + println!( + " {:<40} {:>12} {:>12}{}", + fact.labels.join(","), + fact.expected, + fact.actual, + if holds { "" } else { " MISMATCH" } + ); + } + let report = metric.share_checks(n); println!( " shares of {} records (tolerance {} standard errors)", report.records, diff --git a/asap-tools/data-sources/paced-feeder/src/records.rs b/asap-tools/data-sources/paced-feeder/src/records.rs index 86148c10..7075d17b 100644 --- a/asap-tools/data-sources/paced-feeder/src/records.rs +++ b/asap-tools/data-sources/paced-feeder/src/records.rs @@ -27,7 +27,7 @@ //! record's value of `L` is its first one (`0`, or `0.0` for a child: the //! heaviest). Scaling keeps the tail index `a`. -use crate::table::{fnv1a, is_identifier, label_parents}; +use crate::table::{closed_cardinality, fnv1a, is_identifier, label_parents, Fact}; use crate::workload::{splitmix64, unit_uniform, LabelSpec}; use anyhow::{bail, ensure, Context, Result}; use serde::Deserialize; @@ -45,6 +45,9 @@ pub const PARETO_SCALE: f64 = 1_000.0; pub const SCALE_RANGE: [f64; 2] = [1.0, 10.0]; /// Keys are sent as `Float64`, which holds integers exactly up to 2^53. const MAX_KEY: u64 = 1 << 53; +/// The most values a Zipf draws from (a label's values or a field's keys): it +/// keeps a CDF entry and a `u32` guide entry per value, 1.2 GB at this cap. +pub const MAX_ZIPF_VALUES: u64 = 100_000_000; /// The facts check counts shares over this many records. pub const SHARE_SAMPLE_RECORDS: u64 = 1_000_000; /// A share's check fails if it is off by more than this many standard errors @@ -113,6 +116,7 @@ pub fn zipf_shares(n: u64, theta: f64) -> Vec { impl Zipf { pub fn new(n: u64, theta: f64) -> Self { + assert!(n <= MAX_ZIPF_VALUES, "a Zipf over {n} values"); let mut cdf: Vec = zipf_shares(n, theta) .into_iter() .scan(0.0, |sum, p| { @@ -231,9 +235,10 @@ impl RecordMetric { .with_context(|| format!("label `{}` needs a `skew`", label.name))?; ensure!(skew >= 0.0, "label `{}`: skew must be ≥ 0", label.name); ensure!( - label.cardinality <= u64::from(u32::MAX), - "label `{}`: too many values", - label.name + label.cardinality <= MAX_ZIPF_VALUES, + "label `{}`: cardinality {} is over the {MAX_ZIPF_VALUES} values a Zipf draws from", + label.name, + label.cardinality ); label_zipfs.push(Zipf::new(label.cardinality, skew)); } @@ -259,8 +264,8 @@ impl RecordMetric { } => { ensure!(theta >= 0.0, "value `{field}`: zipf must be ≥ 0"); ensure!( - keys > 0 && keys <= MAX_KEY, - "value `{field}`: keys must be in [1, 2^53]" + keys > 0 && keys <= MAX_ZIPF_VALUES, + "value `{field}`: keys must be in [1, {MAX_ZIPF_VALUES}], the values a Zipf draws from" ); if let Some(b) = field_burst { ensure!(burst.is_none(), "at most one value may have a burst"); @@ -438,6 +443,42 @@ impl RecordMetric { shares } + /// For every non-empty subset of the labels, the number of label-value + /// tuples a record can have (as for series: closed under `child_of`) and + /// the distinct tuples among records `0..n`. A child's value index names + /// its parent, so the tuples need not hold the ancestors. + pub fn label_coverage(&self, n: u64) -> Vec { + let num_labels = self.labels.len(); + assert!( + num_labels < 32, + "too many labels to enumerate their subsets" + ); + let (mut labels, mut values) = (vec![0u32; num_labels], vec![0.0; self.fields.len()]); + let mut drawn = Vec::with_capacity(n as usize * num_labels); + for index in 0..n { + self.record(index, &mut labels, &mut values); + drawn.extend_from_slice(&labels); + } + (1u32..1 << num_labels) + .map(|mask| { + let subset: Vec = (0..num_labels).filter(|i| mask & (1 << i) != 0).collect(); + let actual = drawn + .chunks(num_labels) + .map(|record| subset.iter().map(|&i| record[i]).collect::>()) + .collect::>() + .len() as u64; + Fact { + labels: subset + .iter() + .map(|&i| self.labels[i].name.clone()) + .collect(), + expected: closed_cardinality(&self.labels, &subset), + actual, + } + }) + .collect() + } + /// The share checks of the facts check over records `0..n`: each label's /// heaviest values (a child's index pooled over parents, so its share given /// the parent), the burst value and the share of burst records. @@ -814,6 +855,27 @@ pub mod tests { assert_ne!(records(&a, 100), records(&reseeded, 100)); } + #[test] + fn label_coverage_counts_distinct_tuples_within_the_domain() { + let coverage = metric("http", &http_spec()).label_coverage(200_000); + assert_eq!(coverage.len(), 15); + let by_labels = |names: &str| { + let fact = coverage + .iter() + .find(|f| f.labels.join(",") == names) + .unwrap(); + (fact.expected, fact.actual) + }; + // Small domains are covered; an endpoint names its service. + assert_eq!(by_labels("region"), (4, 4)); + assert_eq!(by_labels("endpoint"), (625, 625)); + assert_eq!(by_labels("service,endpoint"), (625, 625)); + // Heavy skew leaves some of the 10,000 full tuples unseen. + let (expected, actual) = by_labels("region,service,endpoint,status"); + assert!(expected == 10_000 && actual < expected, "{actual}"); + assert!(coverage.iter().all(|f| f.actual <= f.expected)); + } + #[test] fn share_checks_pass_and_catch_a_wrong_skew() { for spec in [http_spec(), flows_spec()] { @@ -871,8 +933,16 @@ pub mod tests { spec["values"]["dst_port"] = json!({"zipf": 0.0, "keys": 5}); assert!(err(spec).contains("not be `ts` or a label")); + // A Zipf's tables are capped; burst keys are drawn uniformly, so only + // their numbering is. + let mut spec = flows_spec(); + spec["values"]["src_ip"]["keys"] = json!(MAX_ZIPF_VALUES + 1); + assert!(err(spec).contains("keys must be in [1, 100000000]")); + let mut spec = http_spec(); + spec["labels"][0]["cardinality"] = json!(MAX_ZIPF_VALUES + 1); + assert!(err(spec).contains("cardinality 100000001 is over")); let mut spec = flows_spec(); - spec["values"]["src_ip"]["keys"] = json!(1u64 << 53); + spec["values"]["src_ip"]["burst"]["keys"] = json!(1u64 << 53); assert!(err(spec).contains("all keys ≤ 2^53")); assert!(RecordMetric::from_table("m", &http_spec(), Some(0), 1).is_err()); diff --git a/asap-tools/data-sources/paced-feeder/src/table.rs b/asap-tools/data-sources/paced-feeder/src/table.rs index 8454f8b5..97cd6546 100644 --- a/asap-tools/data-sources/paced-feeder/src/table.rs +++ b/asap-tools/data-sources/paced-feeder/src/table.rs @@ -333,7 +333,7 @@ pub fn facts(labels: &[LabelSpec], series: &[Vec]) -> Vec { /// 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 { +pub 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() { diff --git a/asap-tools/data-sources/paced-feeder/src/workload.rs b/asap-tools/data-sources/paced-feeder/src/workload.rs index 20567c85..4649d86e 100644 --- a/asap-tools/data-sources/paced-feeder/src/workload.rs +++ b/asap-tools/data-sources/paced-feeder/src/workload.rs @@ -73,8 +73,7 @@ pub struct Metric { pub seed: u64, } -/// Gauge and counter `metrics`, or `records` metrics, never both: they have -/// different sub-second slots. +/// Gauge and counter `metrics` and `records` metrics, sent by one feeder. #[derive(Debug, Clone)] pub struct Workload { /// Samples per second per series of `metrics`; unused for `records`.