feat: experimental setup for paper submission

asap-tools/data-sources/paced-feeder/src/table.rs

@@ -57,6 +57,15 @@ struct GaugeValues {
     dist: ParetoDist,
 }
 
+/// A recording to replay instead of generated values: `steps` values per
+/// series, series-major in [`series_labels`] order.
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+struct ReplayValues {
+    steps: usize,
+    values: Vec<f64>,
+}
+
 #[derive(Deserialize)]
 #[serde(deny_unknown_fields)]
 struct ParetoDist {
@@ -161,6 +170,31 @@ fn metric(name: &str, spec: &Value, seed: u64) -> Result<(Metric, u32)> {
         );
     }
     let series = series_labels(&spec.labels)?;
+    if let Some(replay) = spec.values.get("replay") {
+        ensure!(spec.kind == "gauge", "`replay` values apply to gauges only");
+        let replay: ReplayValues =
+            serde_json::from_value(replay.clone()).context("gauge `values.replay`")?;
+        ensure!(replay.steps > 0, "`replay.steps` must be positive");
+        ensure!(
+            replay.values.len() == series.len() * replay.steps,
+            "`replay.values` has {} values, not {} series x {} steps",
+            replay.values.len(),
+            series.len(),
+            replay.steps
+        );
+        let metric = Metric {
+            name: name.to_string(),
+            labels: spec.labels,
+            scales: vec![1.0; series.len()],
+            series,
+            kind: Kind::Replay {
+                steps: replay.steps,
+                values: Arc::new(replay.values),
+            },
+            seed: splitmix64(seed ^ fnv1a(name)),
+        };
+        return Ok((metric, spec.samples_per_sec_per_series));
+    }
     let (skew, kind) = if spec.kind == "gauge" {
         let values: GaugeValues = serde_json::from_value(spec.values).context("gauge `values`")?;
         ensure!(
@@ -415,6 +449,24 @@ mod tests {
         assert_eq!(series[11], ["1", "2", "2.1"]);
     }
 
+    #[test]
+    fn replay_values_loop_per_series() {
+        let spec = json!({
+            "type": "gauge",
+            "samples_per_sec_per_series": 10,
+            "labels": [{"name": "job", "cardinality": 2}],
+            "values": {"replay": {"steps": 3, "values": [1.0, 2.0, 3.0, 10.0, 20.0, 30.0]}},
+        });
+        let (m, rate) = metric("data", &spec, 7).unwrap();
+        assert_eq!(rate, 10);
+        let seen: Vec<f64> = (0..4).map(|n| m.value(1, n)).collect();
+        assert_eq!(seen, [10.0, 20.0, 30.0, 10.0]);
+        // One value per series and step, no more and no fewer.
+        let mut short = spec.clone();
+        short["values"]["replay"]["values"] = json!([1.0, 2.0]);
+        assert!(metric("data", &short, 7).is_err());
+    }
+
     #[test]
     fn rejects_a_child_before_its_parent() {
         let labels = vec![label("label_0", 3, Some("job")), label("job", 2, None)];

asap-tools/data-sources/paced-feeder/src/workload.rs

@@ -59,6 +59,9 @@ pub enum Kind {
     /// `(n + 1) · scale` at sample index `n`: starts at the first sample, grows
     /// by the series' scale per sample and never resets.
     Counter,
+    /// Recorded values, `steps` per series, series-major: sample index `n`
+    /// takes step `n mod steps`, so the recording loops.
+    Replay { steps: usize, values: Arc<Vec<f64>> },
 }
 
 #[derive(Debug, Clone)]
@@ -213,6 +216,9 @@ impl Metric {
                 scale * pareto_scale * u.powf(-1.0 / pareto_shape)
             }
             Kind::Counter => (sample_index + 1) as f64 * scale,
+            Kind::Replay { steps, ref values } => {
+                values[series * steps + (sample_index % steps as u64) as usize]
+            }
         }
     }
 }

asap-tools/experiments/datasets/alibaba/make_replay.py

@@ -0,0 +1,110 @@
+#!/usr/bin/env python3
+"""A replay table for paced-feeder from Alibaba v2022 MSMetrics: pod CPU
+utilization in the random workloads' label shape (4 region x 5 service x 10
+job x 50 label_0), so their plans apply unchanged.
+
+Each job is one microservice and each label_0 one of its pods; region and
+service only group microservices. Label values follow the feeder's table
+convention: a child's value is its parent's, a `.`, and its index (`job="3.1.7"`).
+Microservices are drawn by `--seed` among those with at least 50 pods that
+report at every step of the input; pods are their first 50 by name.
+
+Writes `<out>.csv` (`offset_ms,region,service,job,label_0,value`, sorted by
+offset, offsets from the first step) and `<out>_mapping.csv` (labels to
+msname/msinstanceid). With `--table`, also `<out>_table.json`: that
+workload table with its `data` metric's values replaced by the replay, which
+paced-feeder loops (one trace step per sample).
+
+Usage: python3 make_replay.py MSMetricsUpdate_0.csv MSMetricsUpdate_1.csv \
+    --out replay [--table eval_table.json]
+"""
+
+import argparse
+import csv
+import json
+import random
+from collections import defaultdict
+
+REGIONS, SERVICES, JOBS, PODS = 4, 5, 10, 50
+
+
+def rows(paths):
+    for path in paths:
+        with open(path, newline="") as f:
+            for r in csv.DictReader(f):
+                yield r
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("csv", nargs="+")
+    parser.add_argument("--out", required=True)
+    parser.add_argument("--seed", type=int, default=862)
+    parser.add_argument("--table", help="workload table whose `data` gets the replay")
+    args = parser.parse_args()
+
+    steps, seen = set(), defaultdict(set)  # pod -> steps
+    pods = defaultdict(set)  # msname -> pods
+    for r in rows(args.csv):
+        steps.add(r["timestamp"])
+        seen[r["msinstanceid"]].add(r["timestamp"])
+        pods[r["msname"]].add(r["msinstanceid"])
+    full = {p for p, s in seen.items() if len(s) == len(steps)}
+    eligible = sorted(m for m, ps in pods.items() if len(ps & full) >= PODS)
+    jobs = REGIONS * SERVICES * JOBS
+    print(f"{len(steps)} steps; {len(eligible)} microservices eligible")
+    chosen = random.Random(args.seed).sample(eligible, jobs)
+
+    labels = {}  # pod -> (region, service, job, label_0)
+    with open(f"{args.out}_mapping.csv", "w", newline="") as f:
+        w = csv.writer(f)
+        w.writerow(["region", "service", "job", "label_0", "msname", "msinstanceid"])
+        for i, msname in enumerate(chosen):
+            region = str(i // (SERVICES * JOBS))
+            service = f"{region}.{i // JOBS % SERVICES}"
+            job = f"{service}.{i % JOBS}"
+            ordered = sorted(
+                pods[msname] & full, key=lambda p: int(p.rsplit("_", 1)[1])
+            )
+            for k, pod in enumerate(ordered[:PODS]):
+                labels[pod] = (region, service, job, f"{job}.{k}")
+                w.writerow([*labels[pod], msname, pod])
+
+    start = min(int(s) for s in steps)
+    out = sorted(
+        (int(r["timestamp"]) - start, *labels[r["msinstanceid"]], r["cpu_utilization"])
+        for r in rows(args.csv)
+        if r["msinstanceid"] in labels
+    )
+    with open(f"{args.out}.csv", "w", newline="") as f:
+        w = csv.writer(f)
+        w.writerow(["offset_ms", "region", "service", "job", "label_0", "value"])
+        w.writerows(out)
+    print(f"{len(out)} rows, {len(labels)} series")
+    if args.table:
+        write_table(args.table, f"{args.out}_table.json", out, len(steps))
+
+
+def write_table(table_path, out_path, rows, steps):
+    """`table_path` with `data`'s values replaced by `rows`, series-major in
+    paced-feeder's series order (mixed radix, label_0 fastest)."""
+    by_series = {}
+    for offset, *_, label_0, value in rows:  # sorted by offset
+        by_series.setdefault(label_0, []).append(round(float(value), 6))
+    values = []
+    for r in range(REGIONS):
+        for s in range(SERVICES):
+            for j in range(JOBS):
+                for k in range(PODS):
+                    values += by_series[f"{r}.{s}.{j}.{k}"]
+    assert len(values) == len(by_series) * steps
+    table = json.load(open(table_path))
+    data = table["workloads"][0]["e2e"]["metrics"]["data"]
+    data["values"] = {"replay": {"steps": steps, "values": values}}
+    with open(out_path, "w") as f:
+        json.dump(table, f)
+    print(out_path)
+
+
+if __name__ == "__main__":
+    main()

asap-tools/experiments/datasets/fake_metric/random_workload_artifacts.py

@@ -133,7 +133,24 @@ def workload_set(workload):
     return f"seed-{workload['seed']}" + (f"-{variant}" if variant else "")
 
 
-def pilot_configs(workload, pilot_dir):
+def as_dashboards(groups, dashboards, key):
+    """`groups` repeated once per dashboard, each copy its own client thread
+    and output group; `key` ("id" or "name") stays unique per copy."""
+    if dashboards == 1:
+        return groups
+    copies = []
+    for d in range(1, dashboards + 1):
+        for group in groups:
+            group = copy.deepcopy(group)
+            if key == "id":
+                group["id"] = len(copies) + 1
+            else:
+                group[key] = f"{group[key]}_d{d}"
+            copies.append(group)
+    return copies
+
+
+def pilot_configs(workload, pilot_dir, dashboards=1):
     """The five-system pilot's run configs, {path under EXPERIMENTS_DIR:
     content}: ASAPQuery (the saved plan in `pilot_dir`) and Prometheus in
     one experiment_run_e2e.py config, VictoriaMetrics in another, and raw
@@ -144,15 +161,19 @@ def pilot_configs(workload, pilot_dir):
     freshness delay, then evaluates each query at multiples of its interval
     for PILOT_DURATION_MS: the same windows relative to its data start.
     Raw ClickHouse runs the manifest's clickhouse_sql, and ClickHouse+MV
-    the same RQEs' Query over exact views (sketch_bench_workloads)."""
+    the same RQEs' Query over exact views (sketch_bench_workloads).
+
+    With `dashboards` K > 1, every query group runs K times concurrently,
+    as K dashboards over the same data, from `<workload>-dashK/`."""
     queries = [q for q in workload["queries"] if q["support"] == PILOT_SUPPORT]
     lookback_secs = max(q["lookback_ms"] for q in queries) // 1000
     starting_delay = lookback_secs + QUERY_FRESHNESS_DELAY_SECS
     duration = PILOT_DURATION_MS // 1000
     feed_secs = starting_delay + FEED_BOUND_QUERY_PHASES * duration
     seed = workload["data_facts"]["feeder_seed"]
     table = pilot_dir / "eval_table.json"
-    out_dir = Path(CONFIG_DIR) / workload_set(workload) / workload["id"]
+    variant = f"-dash{dashboards}" if dashboards > 1 else ""
+    out_dir = Path(CONFIG_DIR) / workload_set(workload) / (workload["id"] + variant)
 
     e2e, _ = build_config(
         str(table),
@@ -163,6 +184,10 @@ def pilot_configs(workload, pilot_dir):
         feeder_duration_secs=feed_secs,
         prebuilt_output_dir=str(pilot_dir),
     )
+    e2e_params = e2e["experiment_params"]
+    e2e_params["query_groups"] = as_dashboards(
+        e2e_params["query_groups"], dashboards, "id"
+    )
     # One monitoring tool per run, so VictoriaMetrics is its own baseline run.
     vm = copy.deepcopy(e2e)
     del vm["controller"]
@@ -232,7 +257,7 @@ def pilot_configs(workload, pilot_dir):
                 "paced_feeder": feeder,
                 "clickhouse_image_tag": CLICKHOUSE_IMAGE_TAG,
             },
-            "query_groups": groups,
+            "query_groups": as_dashboards(groups, dashboards, "name"),
             "data_ingestion_interval_ms": workload["data_facts"]["scrape_interval_ms"],
         }
         files[name] = "# @package experiment_params\n" + yaml.safe_dump(
@@ -277,7 +302,16 @@ def main():
         help="the workload's archived pilot plan (eval_table.json and planner "
         f"configs): also write the pilot's run configs under {CONFIG_DIR}",
     )
+    parser.add_argument(
+        "--dashboards",
+        type=int,
+        default=1,
+        help="with --pilot-dir: issue every query group this many times "
+        "concurrently, into <workload>-dashK/",
+    )
     args = parser.parse_args()
+    if args.dashboards < 1:
+        parser.error("--dashboards must be at least 1")
     workload = json.loads(args.manifest.read_text())
     rendered = render(workload)
     args.out.mkdir(parents=True, exist_ok=True)
@@ -305,7 +339,9 @@ def main():
     )
     print(table)
     if args.pilot_dir:
-        for path, content in pilot_configs(workload, args.pilot_dir).items():
+        for path, content in pilot_configs(
+            workload, args.pilot_dir, args.dashboards
+        ).items():
             path = Path(EXPERIMENTS_DIR) / path
             path.parent.mkdir(parents=True, exist_ok=True)
             path.write_text(content)

asap-tools/experiments/datasets/fake_metric/random_workloads.py

@@ -69,6 +69,8 @@
     "base": (1, 50),
     # Lookbacks x3: 1m -> 3m, ..., 20m -> 60m; quantiles 100s -> 300s.
     "long": (3, 50),
+    # x2 series: 20,000 at 10/s, 200,000 samples/s.
+    "x2": (1, 100),
     # x5 series: 50,000 at 10/s, 500,000 samples/s.
     "x5": (1, 250),
 }