diff --git a/asap-tools/data-sources/paced-feeder/README.md b/asap-tools/data-sources/paced-feeder/README.md index 560d9445..ab41bd70 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,70 @@ 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, 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, +`5·sqrt(p(1-p)/n)`. ```sh $ paced_feeder --workload-table synthetic-templatesclassic-shared1.json --verify-facts @@ -80,6 +138,25 @@ 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 + label-value tuples among 1000000 records + labels possible seen + 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 + … + 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 +189,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 +216,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. @@ -153,6 +251,14 @@ 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). Records requests are cut anywhere, so a series' records may +arrive out of order within a tick; that is safe because no records query +reads a series in arrival order (as `increase` does) and Prometheus is not a +target for them. 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`; @@ -171,3 +277,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 fa6101f8..e7cec82e 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.tick_rows(0, 2, 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, 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.tick_rows(0, 0, 1_000); - let request = - decode_remote_write(&RemoteWriteEncoder::new(metric, &metric.name).encode(&rows)); + let rows = w.series_rows(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.tick_rows(0, 0, 42_000); - let decoded = decode_row_binary(&RowBinaryEncoder::new(metric).encode(&rows), 2); + let rows = w.series_rows(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.tick_rows(0, 3, 0); - let decoded = decode_row_binary(&RowBinaryEncoder::new(metric).encode(&rows), 4); + let rows = w.series_rows(0, 3, 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 1522d8b4..0e888038 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::{series_batches, Row, Workload}; +use workload::{series_batches, 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, @@ -85,10 +94,11 @@ struct Args { #[arg(long)] start_ms: Option, - /// Rows per HTTP request, rounded down to whole series (at least one). - /// Defaults to 50,000 for remote write, which keeps a compressed request - /// under axum's 2 MB body limit in ASAPQuery's ingest, and 100,000 for - /// ClickHouse, where each INSERT creates a part. + /// Rows per HTTP request, for gauges and counters rounded down to whole + /// series (at least one). Defaults to 50,000 for remote write, which keeps + /// a compressed request under axum's 2 MB body limit in ASAPQuery's ingest, + /// and 100,000 for ClickHouse, where each INSERT creates a part. For + /// remote write it counts samples: a record is one per value field. #[arg(long)] batch_rows: Option, /// Concurrent requests within one tick. @@ -120,7 +130,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, @@ -138,34 +149,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() { @@ -173,30 +207,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 @@ -240,6 +282,7 @@ fn build_workload(args: &Args) -> Result { path, args.metric.as_deref(), args.samples_per_sec, + args.records_per_sec, args.seed, ); } @@ -265,13 +308,20 @@ 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 { + 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}{}", @@ -281,6 +331,51 @@ fn verify_facts(workload: &Workload) -> bool { if holds { "" } else { " MISMATCH" } ); } + let report = metric.share_checks(n); + 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!( + " {:<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 } @@ -289,17 +384,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(()); } @@ -326,16 +423,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(), ); @@ -350,22 +454,28 @@ async fn main() -> Result<()> { let started = Instant::now(); let (mut rows, mut bytes) = (0, 0); - // A tick's requests go out concurrently, each holding whole series - // (see `series_batches`): a series' samples arrive in order, but a - // group's samples across series may not. That is safe for ASAPQuery: - // the tick lies in one window (see `tick_rows`) and is under one - // second behind the watermark, within the allowed lateness. - let per_series = workload.samples_per_sec as usize; + // A tick's requests go out concurrently. Gauge and counter requests + // hold whole series (see `series_batches`): a series' samples arrive + // in order, but a group's samples across series may not. That is safe + // for ASAPQuery: the tick lies in one window (see `tick_rows`) and is + // under one second behind the watermark, within the allowed lateness. + // Records are cut anywhere: no records query reads a series in + // arrival order (as `increase` does), and they never go to Prometheus. let mut tasks = tokio::task::JoinSet::new(); - for metric in 0..workload.metrics.len() { + 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; + let per_series = match *metric_rows { + Rows::Samples(_) => workload.samples_per_sec as usize, + Rows::Records(_) => 1, + }; for range in series_batches(metric_rows.len(), per_series, batch_rows) { 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 rows = range.len(); - let sent = sink.send(metric, &metric_rows[range]).await; + let sent = sink.send(metric, &metric_rows, range).await; sent.map(|bytes| (rows, bytes)) }); } @@ -423,7 +533,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..7075d17b --- /dev/null +++ b/asap-tools/data-sources/paced-feeder/src/records.rs @@ -0,0 +1,950 @@ +//! `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::{closed_cardinality, fnv1a, is_identifier, label_parents, Fact}; +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 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 +/// 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 { + 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| { + *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 <= 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)); + } + 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_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"); + 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 + } + + /// 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. + 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 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()] { + 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")); + + // 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"]["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 e6f73329..97cd6546 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 @@ -290,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() { @@ -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 0d858588..81acc6d6 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,13 @@ pub struct Metric { pub seed: u64, } +/// 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`. pub samples_per_sec: u32, pub metrics: Vec, + pub records: Vec>, } impl Workload { @@ -82,6 +107,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 +126,7 @@ impl Workload { Self { samples_per_sec, metrics: vec![metric], + records: Vec::new(), } } @@ -107,17 +134,39 @@ 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(self.series_rows(target, tick, start_ms)), + Some(r) => Rows::Records(self.records[r].tick(tick, start_ms)), + } } - /// The samples of `metric` for one-second tick `tick`, series-major: each - /// series' `samples_per_sec` samples in timestamp order, then the next - /// series'. Slot `j` of `n` is at `t + (j + 1) · 1 s / n`, where `t` is the - /// tick's start, so the tick lies in `(t, t + 1 s]`: the same left-open - /// interval as ASAPQuery's one-second windows, so a tick never spans two - /// of them. - pub fn tick_rows(&self, metric: usize, tick: u64, start_ms: i64) -> Vec { + /// The samples of `metric` for tick `tick`, series-major: each series' + /// `samples_per_sec` samples in timestamp order, then the next series'. + /// Slot `j` of `n` is at `t + (j + 1) · 1 s / n`. + pub fn series_rows(&self, metric: usize, tick: u64, start_ms: i64) -> Vec { let per_sec = u64::from(self.samples_per_sec); let tick_ms = start_ms + tick as i64 * 1_000; let metric = &self.metrics[metric]; @@ -170,7 +219,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 @@ -194,7 +243,7 @@ mod tests { #[test] fn a_tick_is_series_major_in_timestamp_order() { let w = workload(); - let rows = w.tick_rows(0, 5, 10_000); + let rows = w.series_rows(0, 5, 10_000); assert_eq!(rows.len(), 12); for (series, samples) in rows.chunks(2).enumerate() { assert!(samples.iter().all(|r| r.series == series)); @@ -206,7 +255,7 @@ mod tests { #[test] fn batches_keep_each_series_in_one_request_in_order() { let w = Workload::grouped("data", 7, 3, 2, 3, 1.5, 1.0); - let rows = w.tick_rows(0, 2, 0); + let rows = w.series_rows(0, 2, 0); // 7 rows round down to 2 series (6 rows); 1 row still takes a series. for (batch_rows, sizes) in [(7, vec![6, 6, 6]), (1, vec![3; 6]), (100, vec![18])] { let batches: Vec<_> = series_batches(rows.len(), 3, batch_rows).collect(); @@ -232,7 +281,7 @@ mod tests { fn values_depend_only_on_seed_series_and_sample_index() { let w = workload(); let values = |w: &Workload, start_ms| -> Vec { - w.tick_rows(0, 4, start_ms) + w.series_rows(0, 4, start_ms) .iter() .map(|r| r.value) .collect()