From 70b60aba9e067de7a72984970604822605975aa3 Mon Sep 17 00:00:00 2001 From: Zeying Zhu <50204836+zzylol@users.noreply.github.com> Date: Mon, 12 Oct 2026 19:29:44 +0000 Subject: [PATCH] Add end-to-end acceptance and cost/accuracy experiments acceptance.py runs each use case (latency, cardinality, drift, top-k) on the real collector containers with time scaled 6x and checks sketch answers against the exact ground truth through Prometheus and Loki: every stage up, full rate delivered, identical points per window, bounds containing the exact answer, classes, drift and novelty detected with no false positives, events in Loki. Exits non-zero on any failure. experiments.py measures branch CPU with and without each function over offered rates (inline cost) and error vs sketch size (KLL k, Count-Min width). Answer paths report asap.answer.delay, seconds from a window's end to its answers. --- asap-precompute-rs/src/otap/answer_nodes.rs | 30 ++- deploy/telemetry-functions/README.md | 80 ++++-- deploy/telemetry-functions/acceptance.py | 244 ++++++++++++++++++ deploy/telemetry-functions/compose.yaml | 18 +- deploy/telemetry-functions/experiments.py | 199 ++++++++++++++ .../telemetry-functions/report_experiments.py | 129 +++++++++ deploy/telemetry-functions/run.sh | 2 + 7 files changed, 676 insertions(+), 26 deletions(-) create mode 100755 deploy/telemetry-functions/acceptance.py create mode 100755 deploy/telemetry-functions/experiments.py create mode 100755 deploy/telemetry-functions/report_experiments.py diff --git a/asap-precompute-rs/src/otap/answer_nodes.rs b/asap-precompute-rs/src/otap/answer_nodes.rs index 48eba55..9386787 100644 --- a/asap-precompute-rs/src/otap/answer_nodes.rs +++ b/asap-precompute-rs/src/otap/answer_nodes.rs @@ -147,6 +147,33 @@ fn now_ms() -> u64 { .unwrap_or(0) } +/// Adds `asap.answer.delay` (seconds from a window's end to its answers) once +/// per window and method: the end-to-end latency of each answer path. +fn with_delay(envs: &[SketchEnvelope], now_ms: u64) -> Vec { + let mut out = envs.to_vec(); + let mut seen = std::collections::BTreeSet::new(); + for e in envs { + let method = e + .labels + .iter() + .find(|kv| kv.key == "method") + .map(|kv| kv.value.clone()); + if let Some(method) = method { + if seen.insert((e.window_end_ms, method.clone())) { + out.push(SketchEnvelope { + labels: vec![KeyValue::new("method", method)], + metric_name: "asap.answer.delay".into(), + window_start_ms: e.window_start_ms, + window_end_ms: e.window_end_ms, + value: now_ms.saturating_sub(e.window_end_ms) as f64 / 1000.0, + ..Default::default() + }); + } + } + } + out +} + /// Plain OTLP gauges with every label on the data point, so backends such /// as Prometheus keep them as series labels. fn gauges_pdata(envs: &[SketchEnvelope]) -> Result { @@ -234,7 +261,8 @@ async fn emit( effects: &mut local::EffectHandler, ) -> Result<(), Error> { if !envs.is_empty() { - match gauges_pdata(envs) { + let envs = with_delay(envs, now_ms()); + match gauges_pdata(&envs) { Ok(pdata) => effects.send_message_with_source_node(pdata).await?, Err(e) => effects.info(&format!("answers: encode failed: {e}")).await, } diff --git a/deploy/telemetry-functions/README.md b/deploy/telemetry-functions/README.md index 1af874a..771da44 100644 --- a/deploy/telemetry-functions/README.md +++ b/deploy/telemetry-functions/README.md @@ -3,26 +3,41 @@ Every stage is the official OTAP dataflow engine (`asap-collector`: the upstream controller with the ASAP nodes linked in), one container per stage, each pinned to its own CPU, connected by OTLP/HTTP over the compose network. +This is the deployment behind [RFC 0001](../../docs/rfcs/0001-sketch-telemetry-functions.md). ```text -generator-a -> branch-a --sketch--> merge -> detector --> Prometheus -> Grafana -generator-b -> branch-b --sketch--/ - \--raw copy (fanout)--> exact ------/ +generator-a -> branch-a --sketch--> merge -> detector --metrics--> Prometheus -> Grafana +generator-b -> branch-b --sketch--/ \--events---> Loki ------/ + \--raw copy (fanout)--> exact (same metrics and events, method="exact") ``` -| Stage | Pipeline | What it does | -| --- | --- | --- | -| generator-a/b | `pipelines/generator-*.yaml` | `scenario_generator`: seeded latency points over 20 routes (temporary stand-in for the official traffic generator) | -| branch-a/b | `pipelines/branch.yaml` | `fanout` copies the input: `asap_sketches` builds one KLL per route per 1-minute window; the raw copy goes to `exact` | -| merge | `pipelines/merge.yaml` | `window_merge`: merges both sources' sketches per window, 5 s after the window ends | -| detector | `pipelines/detector.yaml` | `sketch_detector`: p50/p99 per route with lower/upper bounds | -| exact | `pipelines/exact.yaml` | `exact_answers`: exact p50/p99 per route from every raw point (ground truth and exact baseline) | +| Stage | What it runs | +| --- | --- | +| generator-a/b | `scenario_generator`: seeded traffic for the use case (temporary stand-in for the official traffic generator) | +| branch-a/b | the official `fanout` copies each message: `asap_sketches` builds one sketch per series per 1-minute window; the raw copy goes to `exact` | +| merge | `window_merge`: merges both sources' sketches per window, 5 s after the window ends | +| detector | `sketch_detector`: answers with bounds, baseline comparisons, events | +| exact | `exact_answers`: the same answers from every raw point: ground truth and exact baseline | -Prometheus receives both answer streams over OTLP (`asap_quantile`, -`asap_points`, labelled `method="sketch"` or `"exact"`) and scrapes each -collector's engine telemetry (`/api/v1/metrics?format=prometheus`): CPU time -per pipeline (`cpu_time_total`), resident memory (`memory_rss`) and payload -bytes per receiver, labelled with the stage. +## Use cases + +`USE_CASE` selects the stage configs in the directory of the same name. + +| `USE_CASE` | Sketch | Scenario | Dashboard | +| --- | --- | --- | --- | +| `latency` | KLL per route | latency over 20 routes | `/d/asap-latency`: p50/p99 with bound vs exact | +| `cardinality` | HLL per attribute key (`per_attribute: ["*"]`) | `user.id` unique, `k8s.pod.name` growing 5%/min, `http.route` and `http.method` fixed | `/d/asap-cardinality`: distinct values with bound, class per key (stable, growing, identifier-like), events | +| `drift` | KLL per route | `http.route-3`'s median triples at minute 10 | `/d/asap-drift`: KS distance from the last 5 windows, drift events | +| `topk` | Count-Min + top-10 heap per attribute key | 200 Zipf routes; a new route takes 15% of traffic at minute 10 | `/d/asap-topk`: top-10 with bound vs exact, new heavy hitters | + +Answers reach Prometheus over OTLP as `asap_quantile`, `asap_distinct`, +`asap_topk`, `asap_points`, `asap_cardinality_state`, `asap_drift_ks`, +`asap_drift`, `asap_topk_new` and `asap_answer_delay`, labelled +`method="sketch"` or `"exact"`. Events (class changes, drift start/end, new +heavy hitters) reach Loki as OTLP logs from `service.name=asap-detector`. +Prometheus also scrapes each engine's telemetry +(`/api/v1/metrics?format=prometheus`): CPU time per pipeline, resident memory +and payload bytes per receiver, labelled with the stage. ## Run @@ -30,9 +45,40 @@ Linux, Docker with Compose, 7 free CPUs (0-6 by default; `CPU_BRANCH_A=10` etc. move a stage). From the repository root: ```sh -deploy/telemetry-functions/run.sh # DOCKER_COMMAND="sudo docker" if needed +USE_CASE=cardinality deploy/telemetry-functions/run.sh # DOCKER_COMMAND="sudo docker" if needed ``` -Open . The first answers appear about +Open `http://localhost:3000/d/asap-`. The first answers appear about 65 seconds after the first full minute. Stop with `docker compose -f deploy/telemetry-functions/compose.yaml down`. + +## End-to-end acceptance + +```sh +deploy/telemetry-functions/run.sh --build-only +deploy/telemetry-functions/acceptance.py --report acceptance.json +``` + +Runs all four use cases on the containers with time scaled down 6x (10-second +windows, changes at second 100) and checks the sketch answers against the +exact ground truth through Prometheus and Loki. It exits non-zero if any check +fails. The checks: + +| Use case | Checks | +| --- | --- | +| all | every stage up; points per window equal the offered rate (no backpressure); sketch and exact windows hold the same points | +| latency | exact p50/p99 inside the sketch's bound (routes with at least 2,000 points) | +| cardinality | sketch and exact give every key the same class; the generated classes are found; distinct-count error at most 2%; class events in Loki | +| drift | the changed route drifts in both methods; no other route does; drift event in Loki | +| topk | the new route enters both top-10s; sketch recall@10 at least 0.9; novelty event in Loki | + +CI runs it on every PR (`.github/workflows/ci.yml`) with the stages sharing +the runner's CPUs: a correctness check, not a capacity measurement. + +## Experiments + +`experiments.py inline-cost` measures branch CPU with and without each +function over offered rates; `experiments.py accuracy-size` measures error +against exact as KLL `k` and Count-Min width change. Results and the +merge/serialization benchmark (`asap-merge-bench`) are in +[docs/results](../../docs/results/). diff --git a/deploy/telemetry-functions/acceptance.py b/deploy/telemetry-functions/acceptance.py new file mode 100755 index 0000000..a54c859 --- /dev/null +++ b/deploy/telemetry-functions/acceptance.py @@ -0,0 +1,244 @@ +#!/usr/bin/env python3 +"""End-to-end acceptance of the telemetry functions on real collector containers. + +For each use case: start every stage (generators, branches, merge, detector, +exact baseline, Prometheus, Loki, Grafana) with `docker compose`, wait for the +scenario to play out, then check the sketch answers against the exact ground +truth through Prometheus and Loki. Prints one line per check and exits non-zero +if any check fails. + +Time is scaled down 6x so a use case takes a few minutes: 10-second windows, a +2-second merge grace, changes after 1/6 of their configured time and growth 6x +faster per minute. Rules counted in windows (baselines, thresholds) are as +deployed. + + deploy/telemetry-functions/acceptance.py --docker-command "sudo docker" + deploy/telemetry-functions/acceptance.py --use-cases drift topk + +Requires the asap-collector:local image (run.sh builds it, or pass --build). +""" +import argparse +import json +import os +import re +import shlex +import subprocess +import sys +import tempfile +import time +import urllib.parse +import urllib.request +from pathlib import Path + +HERE = Path(__file__).resolve().parent +REPO = HERE.parents[1] +SCALE = 6 +RATE = 20000 # signals/s per source, as in the generator configs +WINDOW_S = 60 // SCALE +USE_CASES = ("latency", "cardinality", "drift", "topk") +STAGES = ("generator-a", "generator-b", "branch-a", "branch-b", "merge", "detector", "exact") + + +def scaled_copy(use_case, into): + """The use case's stage configs with time scaled down SCALE times.""" + for src in (HERE / use_case).glob("*.yaml"): + text = src.read_text() + text = text.replace("window_size: 60s", f"window_size: {WINDOW_S}s") + text = text.replace("window: 60s", f"window: {WINDOW_S}s") + text = text.replace("{grace: 5s}", "{grace: 2s}") + text = re.sub(r"after_seconds: (\d+)", lambda m: f"after_seconds: {int(m[1]) // SCALE}", text) + text = re.sub(r"growth_per_minute: ([0-9.]+)", + lambda m: f"growth_per_minute: {float(m[1]) * SCALE:g}", text) + (into / src.name).write_text(text) + + +class Stack: + def __init__(self, docker, pipelines, prometheus, loki): + self.docker, self.pipelines = docker, pipelines + self.prometheus, self.loki = prometheus, loki + + def compose(self, *args): + env = dict(os.environ, PIPELINES=str(self.pipelines)) + cmd = self.docker + ["compose", "-f", str(HERE / "compose.yaml"), *args] + return subprocess.run(cmd, env=env, capture_output=True, text=True, check=True) + + def get(self, base, path, params): + url = base + path + "?" + urllib.parse.urlencode(params) + with urllib.request.urlopen(url, timeout=10) as response: + return json.load(response)["data"]["result"] + + def vector(self, expr): + return [(r["metric"], float(r["value"][1])) for r in self.get(self.prometheus, "/api/v1/query", {"query": expr})] + + def matrix(self, expr, minutes=5): + rows = self.get(self.prometheus, "/api/v1/query", {"query": f"{expr}[{minutes}m]"}) + return [(r["metric"], {int(float(t)): float(v) for t, v in r["values"]}) for r in rows] + + def events(self, selector): + rows = self.get(self.loki, "/loki/api/v1/query_range", + {"query": '{service_name="asap-detector"} ' + selector, "limit": 500, + "start": str(int((time.time() - 1800) * 1e9))}) + return [v[1] for s in rows for v in s["values"]] + + +def check(name, ok, detail): + return {"check": name, "ok": bool(ok), "detail": detail} + + +def windows(stack, flt): + """Per-window total points (sketch, exact) for one series family.""" + out = {} + for method in ("sketch", "exact"): + totals = {} + for _, samples in stack.matrix(f'asap_points{{method="{method}",{flt}}}'): + for t, v in samples.items(): + totals[t] = totals.get(t, 0) + v + out[method] = totals + return out + + +def latest_window(stack, flt): + """End of the latest window answered by both sketch and exact.""" + w = windows(stack, flt) + shared = sorted(set(w["sketch"]) & set(w["exact"])) + return shared[-1] if shared else int(time.time()) + + +def common_checks(stack, flt): + up = stack.vector('up{job="collectors"}') + yield check("every stage is up", len(up) == len(STAGES) and all(v == 1 for _, v in up), + f"{sum(v == 1 for _, v in up)}/{len(STAGES)} collectors up") + w = windows(stack, flt) + shared = sorted(set(w["sketch"]) & set(w["exact"]))[1:-1] # skip partial edges + expected = 2 * RATE * WINDOW_S + full = [w["exact"][t] for t in shared] + yield check("full rate reaches the end (no backpressure)", + full and all(abs(v - expected) <= 0.01 * expected for v in full), + f"{len(full)} windows, points/window {sorted(set(int(v) for v in full))} vs {expected}") + yield check("sketch and exact windows hold the same points", + shared and all(w["sketch"][t] == w["exact"][t] for t in shared), + f"{len(shared)} windows compared") + + +def latency_checks(stack, t): + exact = {(m["http_route"], m["quantile"]): v for m, v in stack.vector(f'asap_quantile{{method="exact"}} @ {t}')} + points = {m["http_route"]: v for m, v in stack.vector(f'asap_points{{method="exact"}} @ {t}')} + bounds = {(m["http_route"], m["quantile"], m["bound"]): v + for m, v in stack.vector(f'asap_quantile{{method="sketch"}} @ {t}')} + inside = total = 0 + for (route, q), truth in exact.items(): + if points.get(route, 0) < 2000 or (route, q, "lower") not in bounds: + continue + total += 1 + inside += bounds[(route, q, "lower")] <= truth <= bounds[(route, q, "upper")] + yield check("exact p50/p99 inside the sketch bound (routes >= 2,000 points)", + total and inside >= 0.95 * total, f"{inside}/{total} inside") + p99 = [(abs(bounds[(r, "0.99", "estimate")] - v) / v) for (r, q), v in exact.items() + if q == "0.99" and points.get(r, 0) >= 20000 and (r, q, "estimate") in bounds] + yield check("p99 relative error for routes >= 20,000 points", p99, + f"max {max(p99):.2%} over {len(p99)} routes" if p99 else "no routes") + + +def cardinality_checks(stack, t): + state = {(m["attribute_key"], m["method"]): m["state"] + for m, v in stack.vector(f"asap_cardinality_state @ {t} == 1")} + keys = sorted({k for k, _ in state}) + yield check("sketch and exact agree on every key's class", + keys and all(state.get((k, "sketch")) == state.get((k, "exact")) for k in keys), + ", ".join(f"{k}={state.get((k, 'sketch'))}/{state.get((k, 'exact'))}" for k in keys)) + want = {"user.id": "identifier-like", "http.route": "stable", "http.method": "stable"} + yield check("identifier and stable keys classified as generated", + all(state.get((k, m)) == s for k, s in want.items() for m in ("sketch", "exact")), str(want)) + grew = stack.vector('max_over_time(asap_cardinality_state{state="growing",attribute_key="k8s.pod.name"}[10m])') + yield check("growing key (k8s.pod.name) classified growing by both", + {m["method"] for m, v in grew if v == 1} == {"sketch", "exact"}, + f"growing seen by {sorted(m['method'] for m, v in grew if v == 1)}") + distinct = {(m["attribute_key"], m["method"]): v + for m, v in stack.vector(f'asap_distinct{{bound=~"estimate|"}} @ {t}')} + errs = {k: abs(distinct[(k, "sketch")] - distinct[(k, "exact")]) / distinct[(k, "exact")] + for k in keys if (k, "sketch") in distinct and (k, "exact") in distinct} + yield check("distinct-count error <= 2% for every key", errs and max(errs.values()) <= 0.02, + ", ".join(f"{k} {e:.2%}" for k, e in errs.items())) + events = stack.events('| event_kind="cardinality" | method="sketch"') + yield check("class events in Loki", any("identifier-like" in e for e in events), + f"{len(events)} sketch events") + + +def drift_checks(stack, t): + seen = stack.vector("max_over_time(asap_drift[10m])") + hit = {(m["http_route"], m["method"]) for m, v in seen if v == 1} + yield check("route-3 drift detected by sketch and exact", + {("http.route-3", "sketch"), ("http.route-3", "exact")} <= hit, str(sorted(hit))) + yield check("no other route flagged (no false positives)", + all(r == "http.route-3" for r, _ in hit), f"{len(seen)} series watched") + events = stack.events('| event_kind="drift" | method="sketch"') + yield check("drift event in Loki", any("http.route-3" in e and "drifted" in e for e in events), + events[0] if events else "none") + + +def topk_checks(stack, t): + new = stack.vector('max_over_time(asap_topk{key="/new/hot-route",bound=~"estimate|"}[10m])') + yield check("new route in both top-10s", {m["method"] for m, _ in new} == {"sketch", "exact"}, + str(sorted(m["method"] for m, _ in new))) + top = {m: {r["key"] for r, _ in stack.vector(f'asap_topk{{method="{m}",bound=~"estimate|"}} @ {t}')} + for m in ("sketch", "exact")} + recall = len(top["sketch"] & top["exact"]) / max(1, len(top["exact"])) + yield check("sketch top-10 recall vs exact >= 0.9", recall >= 0.9, f"recall {recall:.2f}") + events = stack.events('| event_kind="novelty" | method="sketch"') + yield check("novelty event in Loki", any("/new/hot-route" in e for e in events), + next((e for e in events if "/new/hot-route" in e), "none")) + + +CASES = { + # use case: (series filter for point counts, checks, seconds to wait) + "latency": ('attribute_key=""', latency_checks, 5 * WINDOW_S + 40), + "cardinality": ('attribute_key="http.route"', cardinality_checks, 600 // SCALE + 6 * WINDOW_S + 40), + "drift": ('attribute_key=""', drift_checks, 600 // SCALE + 3 * WINDOW_S + 40), + "topk": ('attribute_key="http.route"', topk_checks, 600 // SCALE + 3 * WINDOW_S + 40), +} + + +def run_case(args, use_case): + with tempfile.TemporaryDirectory(prefix=f"asap-{use_case}-") as tmp: + os.chmod(tmp, 0o755) + scaled_copy(use_case, Path(tmp)) + for f in Path(tmp).iterdir(): + f.chmod(0o644) + stack = Stack(shlex.split(args.docker_command), tmp, args.prometheus, args.loki) + stack.compose("down", "--remove-orphans") + stack.compose("up", "-d") + try: + flt, checks, wait = CASES[use_case] + print(f"== {use_case}: running for {wait}s", flush=True) + time.sleep(wait) + t = latest_window(stack, flt) + results = list(common_checks(stack, flt)) + list(checks(stack, t)) + finally: + stack.compose("down", "--remove-orphans") + for r in results: + print(f" [{'PASS' if r['ok'] else 'FAIL'}] {r['check']}: {r['detail']}", flush=True) + return [dict(r, use_case=use_case) for r in results] + + +def main(): + parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + parser.add_argument("--use-cases", nargs="+", choices=USE_CASES, default=list(USE_CASES)) + parser.add_argument("--docker-command", default="docker") + parser.add_argument("--prometheus", default="http://127.0.0.1:9090") + parser.add_argument("--loki", default="http://127.0.0.1:3100") + parser.add_argument("--build", action="store_true", help="build the asap-collector image first") + parser.add_argument("--report", type=Path, help="write all results as JSON") + args = parser.parse_args() + if args.build: + env = dict(os.environ, DOCKER_COMMAND=args.docker_command) + subprocess.run([str(HERE / "run.sh"), "--build-only"], env=env, check=True) + results = [r for case in args.use_cases for r in run_case(args, case)] + if args.report: + args.report.write_text(json.dumps(results, indent=2) + "\n") + failed = [r for r in results if not r["ok"]] + print(f"{len(results) - len(failed)}/{len(results)} checks passed") + sys.exit(1 if failed else 0) + + +if __name__ == "__main__": + main() diff --git a/deploy/telemetry-functions/compose.yaml b/deploy/telemetry-functions/compose.yaml index 33aa284..5be17fb 100644 --- a/deploy/telemetry-functions/compose.yaml +++ b/deploy/telemetry-functions/compose.yaml @@ -1,7 +1,8 @@ # Every pipeline stage is the official OTAP engine (asap-collector) in its # own container, pinned to one CPU, talking OTLP/HTTP over the compose # network. Prometheus, Grafana and Loki are the observers. -# USE_CASE picks the pipelines directory (latency, cardinality, ...); +# PIPELINES is the stage-config directory (default ./latency; run.sh sets it +# from USE_CASE); # CPU_* variables move a stage to another core. x-collector: &collector image: asap-collector:local @@ -11,43 +12,43 @@ services: generator-a: <<: *collector command: ["--config", "/pipelines/generator-a.yaml"] - volumes: ["./${USE_CASE:-latency}:/pipelines:ro"] + volumes: ["${PIPELINES:-./latency}:/pipelines:ro"] cpuset: "${CPU_GENERATOR_A:-0}" depends_on: [branch-a] generator-b: <<: *collector command: ["--config", "/pipelines/generator-b.yaml"] - volumes: ["./${USE_CASE:-latency}:/pipelines:ro"] + volumes: ["${PIPELINES:-./latency}:/pipelines:ro"] cpuset: "${CPU_GENERATOR_B:-1}" depends_on: [branch-b] branch-a: <<: *collector command: ["--config", "/pipelines/branch.yaml"] - volumes: ["./${USE_CASE:-latency}:/pipelines:ro"] + volumes: ["${PIPELINES:-./latency}:/pipelines:ro"] cpuset: "${CPU_BRANCH_A:-2}" depends_on: [merge, exact] branch-b: <<: *collector command: ["--config", "/pipelines/branch.yaml"] - volumes: ["./${USE_CASE:-latency}:/pipelines:ro"] + volumes: ["${PIPELINES:-./latency}:/pipelines:ro"] cpuset: "${CPU_BRANCH_B:-3}" depends_on: [merge, exact] merge: <<: *collector command: ["--config", "/pipelines/merge.yaml"] - volumes: ["./${USE_CASE:-latency}:/pipelines:ro"] + volumes: ["${PIPELINES:-./latency}:/pipelines:ro"] cpuset: "${CPU_MERGE:-4}" depends_on: [detector] detector: <<: *collector command: ["--config", "/pipelines/detector.yaml"] - volumes: ["./${USE_CASE:-latency}:/pipelines:ro"] + volumes: ["${PIPELINES:-./latency}:/pipelines:ro"] cpuset: "${CPU_DETECTOR:-5}" depends_on: [prometheus] exact: <<: *collector command: ["--config", "/pipelines/exact.yaml"] - volumes: ["./${USE_CASE:-latency}:/pipelines:ro"] + volumes: ["${PIPELINES:-./latency}:/pipelines:ro"] cpuset: "${CPU_EXACT:-6}" depends_on: [prometheus] @@ -70,3 +71,4 @@ services: loki: image: grafana/loki:3.5.0 command: ["-config.file=/etc/loki/local-config.yaml"] + ports: ["127.0.0.1:${LOKI_PORT:-3100}:3100"] diff --git a/deploy/telemetry-functions/experiments.py b/deploy/telemetry-functions/experiments.py new file mode 100755 index 0000000..3c2f59b --- /dev/null +++ b/deploy/telemetry-functions/experiments.py @@ -0,0 +1,199 @@ +#!/usr/bin/env python3 +"""Cost and accuracy experiments on the real collector containers. + +inline-cost Branch CPU with and without each function, over offered rates. + "without" forwards raw points only (the pipeline before the + function); "with" adds the fanout and the sketch. Also records + delivered points/s, branch memory and each answer path's delay. +accuracy-size Error against exact versus sketch size: KLL k (latency p99 + error) and Count-Min width (top-10 recall and count error), + with the bytes merge receives per window. + +Both reuse acceptance.py's time-scaled stack (10-second windows) and write a +CSV per experiment to --output. + + deploy/telemetry-functions/experiments.py inline-cost --docker-command "sudo docker" +""" +import argparse +import csv +import re +import shlex +import statistics +import tempfile +import time +from pathlib import Path + +import acceptance as acc + +BRANCH_WITHOUT = """# Stage 1 without the function: forward raw points to the exact baseline only. +version: otel_dataflow/v1 +engine: + http_admin: + bind_address: 0.0.0.0:8085 +policies: + resources: + core_allocation: + type: core_count + count: 1 +groups: + default: + pipelines: + main: + nodes: + receiver: + type: receiver:otlp + config: + protocols: + http: {listening_addr: "0.0.0.0:4318", wait_for_result: false} + to_exact: + type: exporter:otlp_http + config: + endpoint: http://exact:4318 + client_pool_size: 1 + max_in_flight: 1 + http: {timeout: 30s, compression: none} + connections: + - {from: receiver, to: to_exact} +""" + + +def prepare(use_case, into, rate, branch_variant="with", edits=()): + acc.scaled_copy(use_case, into) + for f in into.glob("generator-*.yaml"): + f.write_text(re.sub(r"signals_per_second: \d+", f"signals_per_second: {rate}", f.read_text())) + if branch_variant == "without": + (into / "branch.yaml").write_text(BRANCH_WITHOUT) + if branch_variant in ("with", "without") and not edits: + # The exact stage only counts points per route (cheap), so it never + # throttles the branch being measured. + exact = into / "exact.yaml" + exact.write_text(re.sub( + r"(?s)(type: urn:asap:processor:exact_answers\n) config:.*?\n( to_prometheus:)", + r"\1 config: {window: 10s, per_attribute: [http.route]}\n\2", + exact.read_text())) + for name, old, new in edits: + path = into / name + text = path.read_text() + assert old in text, (name, old) + path.write_text(text.replace(old, new)) + into.chmod(0o755) + for f in into.iterdir(): + f.chmod(0o644) + + +def measure(stack, seconds): + """Averages over the last `seconds`.""" + rng = f"{seconds}s" + one = lambda expr: (stack.vector(expr) or [({}, float("nan"))])[0][1] + by_stage = lambda expr: {m.get("stage"): v for m, v in stack.vector(expr)} + cpu = by_stage(f'sum by (stage) (rate(cpu_time_total{{otel_scope_pipeline_id="main"}}[{rng}]))') + rss = by_stage("max by (stage) (memory_rss)") + # Total points over the range divided by the windows it covers. + sel = 'asap_points{method="exact",attribute_key="http.route"}' + windows_seen = one(f"max(count_over_time({sel}[{rng}]))") + delivered = one(f"sum(sum_over_time({sel}[{rng}]))") / (windows_seen * acc.WINDOW_S) + delay = {m["method"]: v for m, v in stack.vector(f"avg by (method) (avg_over_time(asap_answer_delay[{rng}]))")} + return { + "branch_cpu_cores": statistics.mean([cpu.get("branch-a", 0), cpu.get("branch-b", 0)]), + "branch_rss_mib": max(rss.get("branch-a", 0), rss.get("branch-b", 0)) / 2**20, + "exact_cpu_cores": cpu.get("exact", 0), + "delivered_points_per_s": delivered, + "sketch_answer_delay_s": delay.get("sketch", float("nan")), + "exact_answer_delay_s": delay.get("exact", float("nan")), + } + + +def run(args, use_case, rate, variant="with", edits=(), extra=None): + with tempfile.TemporaryDirectory(prefix="asap-exp-") as tmp: + prepare(use_case, Path(tmp), rate, variant, edits) + stack = acc.Stack(shlex.split(args.docker_command), tmp, args.prometheus, args.loki) + stack.compose("down", "--remove-orphans") + stack.compose("up", "-d") + try: + time.sleep(args.warmup + args.measure) + row = {"use_case": use_case, "variant": variant, "offered_per_source": rate} + row.update(measure(stack, args.measure)) + if extra: + row.update(extra(stack)) + finally: + stack.compose("down", "--remove-orphans") + print(row, flush=True) + return row + + +def merge_bytes(stack, seconds=60): + v = stack.vector(f'sum(rate(payload_size_bytes_total{{stage="merge",otel_scope_node_type="receiver"}}[{seconds}s]))') + return {"merge_bytes_per_window": v[0][1] * acc.WINDOW_S if v else float("nan")} + + +def latency_error(stack): + t = acc.latest_window(stack, 'attribute_key=""') + exact = {m["http_route"]: v for m, v in stack.vector(f'asap_quantile{{method="exact",quantile="0.99"}} @ {t}')} + sketch = {m["http_route"]: v for m, v in stack.vector(f'asap_quantile{{method="sketch",quantile="0.99",bound="estimate"}} @ {t}')} + points = {m["http_route"]: v for m, v in stack.vector(f'asap_points{{method="exact"}} @ {t}')} + errs = [abs(sketch[r] - exact[r]) / exact[r] for r in exact if r in sketch and points.get(r, 0) >= 20000] + return {"p99_rel_error_median": statistics.median(errs) if errs else float("nan"), + "p99_rel_error_max": max(errs) if errs else float("nan"), **merge_bytes(stack)} + + +def topk_error(stack): + t = acc.latest_window(stack, 'attribute_key="http.route"') + get = lambda m: {r["key"]: v for r, v in stack.vector(f'asap_topk{{method="{m}",bound=~"estimate|"}} @ {t}')} + sketch, exact = get("sketch"), get("exact") + errs = [abs(sketch[k] - exact[k]) / exact[k] for k in exact if k in sketch] + return {"recall_at_10": len(set(sketch) & set(exact)) / max(1, len(exact)), + "count_rel_error_mean": statistics.mean(errs) if errs else float("nan"), **merge_bytes(stack)} + + +def inline_cost(args): + rows = [] + for use_case in args.use_cases: + for rate in args.rates: + for variant in ("without", "with"): + rows.append(run(args, use_case, rate, variant)) + return rows + + +def accuracy_size(args): + rows = [] + for k in ([] if args.only == "cms" else (50, 100, 200, 400)): + edit = ("branch.yaml", "sketch_params: {k: 200}", f"sketch_params: {{k: {k}}}") + rows.append(run(args, "latency", args.rates[0], edits=[edit], + extra=lambda s, k=k: {"param": f"kll_k={k}", **latency_error(s)})) + # 100,000 Zipf routes: enough keys for Count-Min collisions to matter. + keys = [(g, "cardinality: 200, zipf: 1.1", "cardinality: 100000, zipf: 1.1") + for g in ("generator-a.yaml", "generator-b.yaml")] + for cols in ([] if args.only == "kll" else (256, 512, 1024, 2048)): + edit = ("branch.yaml", "columns: 2048", f"columns: {cols}") + rows.append(run(args, "topk", args.rates[0], edits=[edit, *keys], + extra=lambda s, c=cols: {"param": f"cms_columns={c}", **topk_error(s)})) + return rows + + +def main(): + parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + parser.add_argument("experiment", choices=("inline-cost", "accuracy-size")) + parser.add_argument("--use-cases", nargs="+", default=["latency", "cardinality", "topk"]) + parser.add_argument("--rates", type=int, nargs="+", default=[20000, 50000, 100000, 200000], + help="offered points/s per source") + parser.add_argument("--warmup", type=int, default=40) + parser.add_argument("--measure", type=int, default=60) + parser.add_argument("--docker-command", default="docker") + parser.add_argument("--prometheus", default="http://127.0.0.1:9090") + parser.add_argument("--loki", default="http://127.0.0.1:3100") + parser.add_argument("--output", type=Path, default=Path("experiment-results")) + parser.add_argument("--only", choices=("kll", "cms"), help="accuracy-size: run one sweep") + args = parser.parse_args() + rows = {"inline-cost": inline_cost, "accuracy-size": accuracy_size}[args.experiment](args) + args.output.mkdir(parents=True, exist_ok=True) + path = args.output / f"{args.experiment}.csv" + with path.open("w", newline="") as f: + fields = list(dict.fromkeys(k for r in rows for k in r)) + writer = csv.DictWriter(f, fieldnames=fields, restval="") + writer.writeheader() + writer.writerows(rows) + print(f"wrote {path}") + + +if __name__ == "__main__": + main() diff --git a/deploy/telemetry-functions/report_experiments.py b/deploy/telemetry-functions/report_experiments.py new file mode 100755 index 0000000..5abc8d7 --- /dev/null +++ b/deploy/telemetry-functions/report_experiments.py @@ -0,0 +1,129 @@ +#!/usr/bin/env python3 +"""Tables and a figure from experiments.py output. + + report_experiments.py RESULTS_DIR OUT_DIR --machine "56-core x86 server ..." + +Reads from RESULTS_DIR: inline-cost.csv (exact stage only counting, so it +never throttles the branch), accuracy-size*.csv, and optionally +exact-baseline.csv (an inline-cost run whose exact stage computes full exact +answers). Writes experiments.md and inline-cost.png to OUT_DIR. Numbers are +absolute. +""" +import argparse +import csv +from pathlib import Path + +import matplotlib + +matplotlib.use("Agg") +import matplotlib.pyplot as plt # noqa: E402 + +NAMES = {"latency": "Latency quantiles (KLL)", "cardinality": "Attribute cardinality (HLL)", + "topk": "Heavy hitters (Count-Min + heap)"} + + +def rows(path): + with open(path) as f: + return list(csv.DictReader(f)) + + +def num(r, k): + try: + return float(r[k]) + except (KeyError, ValueError): + return float("nan") + + +def inline_cost(data, out): + lines = ["## Inline cost", "", + "Each branch receives one source. *Without* forwards raw points only; *with* adds the fanout " + "and the sketch. CPU is the branch pipeline's own CPU time (average of the two branches). " + "The exact stage only counts points here, so it never slows the branches; " + "*added per point* divides the extra branch CPU by the points each branch delivered.", ""] + fig, axes = plt.subplots(1, len(NAMES), figsize=(12, 3.6), dpi=150, sharey=True) + for ax, (case, name) in zip(axes, NAMES.items()): + rs = [r for r in data if r["use_case"] == case] + if not rs: + continue + rates = sorted({int(r["offered_per_source"]) for r in rs}) + get = lambda variant, k: [num(next(r for r in rs if r["variant"] == variant and int(r["offered_per_source"]) == x), k) for x in rates] + without, with_ = get("without", "branch_cpu_cores"), get("with", "branch_cpu_cores") + delivered = get("with", "delivered_points_per_s") + x = [r / 1000 for r in rates] + ax.plot(x, without, marker="o", color="#64748b", label="branch, without function") + ax.plot(x, with_, marker="o", color="#2563eb", label="branch, with function") + ax.set_title(name, fontsize=9) + ax.set_xlabel("offered points/s per source (thousands)") + ax.grid(alpha=0.3) + lines += [f"### {name}", "", + "| Offered per source | Delivered per branch | Branch CPU without | Branch CPU with | Added per point | Branch memory with | Sketch answer delay |", + "| ---: | ---: | ---: | ---: | ---: | ---: | ---: |"] + for rate, dv, a, b, m, d1 in zip(rates, delivered, without, with_, get("with", "branch_rss_mib"), get("with", "sketch_answer_delay_s")): + per_branch = dv / 2 + lines.append(f"| {rate:,} | {per_branch:,.0f} | {a:.3f} cores | {b:.3f} cores | {(b - a) / per_branch * 1e6:.1f} µs | {m:.0f} MiB | {d1:.1f} s |") + lines.append("") + axes[0].set_ylabel("CPU cores") + axes[0].legend(fontsize=7) + fig.tight_layout() + fig.savefig(out / "inline-cost.png") + lines += ["![Branch CPU with and without each function](inline-cost.png)", ""] + return lines + + +def exact_baseline(data): + lines = ["## Exact baseline cost", "", + "The alternative to inline sketches: ship every raw point to one place and compute exact answers. " + "Here the exact stage (one pinned CPU) holds each window's raw values or distinct values; when it " + "saturates, fanout back-pressure lowers the delivered rate. Delivered rates are approximate " + "(5-6 ten-second windows per measurement).", "", + "| Use case | Offered (both sources) | Exact stage CPU | Points delivered |", + "| --- | ---: | ---: | ---: |"] + for r in data: + if r["variant"] != "with": + continue + lines.append(f"| {NAMES.get(r['use_case'], r['use_case'])} | {2 * int(r['offered_per_source']):,}/s | " + f"{num(r, 'exact_cpu_cores'):.2f} cores | {num(r, 'delivered_points_per_s'):,.0f}/s |") + return lines + [""] + + +def accuracy_size(data): + lines = ["## Accuracy versus sketch size", "", + "One window per setting, at 20,000 points/s per source. KLL: the latency scenario (20 routes). " + "Count-Min + heap: 100,000 Zipf routes so collisions matter. Merge bytes per window is what the " + "merge stage receives from both branches.", "", + "| Setting | p99 error, median / max (routes >= 20,000 points) | Recall@10 | Mean count error, top 10 | Merge bytes per window |", + "| --- | ---: | ---: | ---: | ---: |"] + for r in data: + p99 = f"{num(r, 'p99_rel_error_median'):.2%} / {num(r, 'p99_rel_error_max'):.2%}" if r.get("p99_rel_error_median") else "" + recall = f"{num(r, 'recall_at_10'):.2f}" if r.get("recall_at_10") else "" + cerr = f"{num(r, 'count_rel_error_mean'):.2%}" if r.get("count_rel_error_mean") else "" + lines.append(f"| {r['param']} | {p99} | {recall} | {cerr} | {num(r, 'merge_bytes_per_window'):,.0f} B |") + return lines + [""] + + +def main(): + p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + p.add_argument("results", type=Path) + p.add_argument("out", type=Path) + p.add_argument("--machine", required=True) + a = p.parse_args() + a.out.mkdir(parents=True, exist_ok=True) + md = ["# Telemetry-function experiments", "", + f"{a.machine}. Every stage is an `asap-collector` container pinned to its own CPU " + "(deploy/telemetry-functions), with 10-second windows (`experiments.py`); " + "each point is one 60-second measurement after a 40-second warm-up.", ""] + if (a.results / "inline-cost.csv").exists(): + md += inline_cost(rows(a.results / "inline-cost.csv"), a.out) + if (a.results / "exact-baseline.csv").exists(): + md += exact_baseline(rows(a.results / "exact-baseline.csv")) + accuracy = [r for f in sorted(a.results.glob("accuracy-size*.csv")) for r in rows(f)] + if accuracy: + md += accuracy_size(accuracy) + (a.out / "experiments.md").write_text("\n".join(md)) + for f in a.results.glob("*.csv"): + (a.out / f.name).write_text(f.read_text()) + print(f"wrote {a.out / 'experiments.md'}") + + +if __name__ == "__main__": + main() diff --git a/deploy/telemetry-functions/run.sh b/deploy/telemetry-functions/run.sh index 25994d7..58a3671 100755 --- a/deploy/telemetry-functions/run.sh +++ b/deploy/telemetry-functions/run.sh @@ -14,6 +14,8 @@ trap 'rm -rf "$context"' EXIT cp "$target/release/asap-collector" "$context/" strip --strip-debug "$context/asap-collector" "${docker[@]}" build -q -f deploy/collector.Dockerfile -t asap-collector:local "$context" +[ "${1:-}" = "--build-only" ] && exit 0 export USE_CASE=${USE_CASE:-latency} +export PIPELINES=${PIPELINES:-./$USE_CASE} "${docker[@]}" compose -f deploy/telemetry-functions/compose.yaml up -d echo "Grafana: http://localhost:${GRAFANA_PORT:-3000}/d/asap-$USE_CASE"