diff --git a/deploy/telemetry-functions/README.md b/deploy/telemetry-functions/README.md index f573002..8c35d32 100644 --- a/deploy/telemetry-functions/README.md +++ b/deploy/telemetry-functions/README.md @@ -7,8 +7,8 @@ each pinned to its own CPU, connected by OTLP/HTTP over the compose network. Three pipelines run side by side over the same input and share no processor: ```text -sketch: generator-a/b -> branch-a/b (sketch per window) -> merge -> detector -> Prometheus -exact: exact-generator-a/b -> exact-branch-a/b (exact window) -> exact-merge -> exact-detector -> Prometheus +sketch: generator-a/b -> branch-a/b (sketch per window) -> merge -> detector -> Prometheus, Loki +exact: exact-generator-a/b -> exact-branch-a/b (exact window) -> exact-merge -> exact-detector -> Prometheus, Loki raw: raw-generator-a/b -> raw-branch-a/b (forward) -> raw-merge (forward) -> raw-sink -> loki-raw ``` @@ -16,23 +16,17 @@ The two generators of each pipeline (sources a and b) run the same configs with the same scenario start (`SCENARIO_START`), and every point is a function of scenario time, so all three pipelines receive identical points. -| Stage | Config | What it runs | -| --- | --- | --- | -| generator-a/b, exact-generator-a/b, raw-generator-a/b | `latency/generator-*.yaml` | `scenario_generator`: seeded latency points over 20 routes (temporary stand-in for the official traffic generator) | -| branch-a/b | `latency/branch.yaml` | `asap_sketches`: one KLL per route per 1-minute window | -| merge | `latency/merge.yaml` | `window_merge`: merges both sources' sketches per window, 5 s after the window ends | -| detector | `latency/detector.yaml` | `sketch_detector`: p50/p99 per route with lower/upper bounds (`method="sketch"`) | -| exact-branch-a/b | `latency/exact-branch.yaml` | `exact_branch`: every value per route per 1-minute window | -| exact-merge | `common/exact-merge.yaml` | `exact_merge`: adds both sources' exact windows, 5 s after the window ends | -| exact-detector | `latency/exact-detector.yaml` | `exact_detector`: exact p50/p99 per route (`method="exact"`): the ground truth | -| raw-branch-a/b, raw-merge | `common/raw-*.yaml` | forward raw points (OTLP receiver -> exporter) | -| raw-sink | `common/raw-sink.yaml` | `raw_sink`: writes every raw point as one log record to `loki-raw` and reports when `loki-raw` acknowledged each window's last point | - -Prometheus receives the answers 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. +| Stage | What it runs | +| --- | --- | +| generator-a/b, exact-generator-a/b, raw-generator-a/b | `scenario_generator`: seeded traffic for the use case (temporary stand-in for the official traffic generator) | +| branch-a/b | `asap_sketches`: one sketch per series per 1-minute window | +| merge | `window_merge`: merges both sources' sketches per window, 5 s after the window ends | +| detector | `sketch_detector`: answers with bounds, baseline comparisons, events (`method="sketch"`) | +| exact-branch-a/b | `exact_branch`: every value per quantile series and a value -> count map per attribute key, per 1-minute window | +| exact-merge | `exact_merge`: adds both sources' exact windows, 5 s after the window ends | +| exact-detector | `exact_detector`: the same answers and events from the exact windows (`method="exact"`): the ground truth | +| raw-branch-a/b, raw-merge | forward raw points (OTLP receiver -> exporter) | +| raw-sink | `raw_sink`: writes every raw point as one log record to `loki-raw` and reports when `loki-raw` acknowledged each window's last point | Every pipeline reports its window completion delay as `asap_answer_delay` (seconds from a window's end until its result is complete at the end of the @@ -42,6 +36,27 @@ reports each point's time from its event time to that acknowledgement (`asap_raw_transit`, p50 and p99) and points `loki-raw` rejected (`asap_raw_failed`). +## 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 (4 x 512) + top-10 heap per attribute key | 100,000 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"` (`asap_answer_delay`, `asap_points` and +`asap_raw_transit` also `"raw"`). 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 Linux, Docker with Compose, 18 free CPUs (0-17 by default; `CPU_BRANCH_A=20` @@ -49,9 +64,42 @@ etc. move a stage) and memory for `loki-raw` (capped at `LOKI_RAW_MEMORY`, default 16g). 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; every pipeline reports its window completion delay (the raw pipeline's points per window are reported, not required) | +| 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 pipelines` runs the three pipelines side by side over offered +rates and records each stage's CPU, memory, delivered rate and window +completion delay (the sketch branch's CPU minus the raw branch's is the cost of +the function); `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..2ae6ac4 --- /dev/null +++ b/deploy/telemetry-functions/acceptance.py @@ -0,0 +1,265 @@ +#!/usr/bin/env python3 +"""End-to-end acceptance of the telemetry functions on real collector containers. + +For each use case: start the three pipelines (sketch, exact, raw) plus +Prometheus, Loki and Grafana with `docker compose`, wait for the scenario to +play out, then check the sketch answers against the exact pipeline's ground +truth through Prometheus and Loki, and that every pipeline reports its window +completion delay. 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 = tuple(f"{p}{s}" for p in ("", "exact-", "raw-") for s in ("generator-a", "generator-b")) +STAGES += ("branch-a", "branch-b", "merge", "detector", "exact-branch-a", "exact-branch-b", + "exact-merge", "exact-detector", "raw-branch-a", "raw-branch-b", "raw-merge", "raw-sink") +METHODS = ("sketch", "exact", "raw") + + +def scaled_copy(directory, into): + """A stage-config directory with time scaled down SCALE times.""" + into.mkdir(exist_ok=True) + for src in (HERE / directory).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, common, prometheus, loki): + self.docker, self.pipelines, self.common = docker, pipelines, common + self.prometheus, self.loki = prometheus, loki + # The generators of all three pipelines count from the same second. + self.start = int(time.time()) + 10 + + def compose(self, *args): + # A CI runner has about 16 GB; the raw pipeline's store gets 4. + env = dict(os.environ, PIPELINES=str(self.pipelines), COMMON=str(self.common), + SCENARIO_START=str(self.start), LOKI_RAW_MEMORY="4g") + 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 per method; sketch and exact for one series + family, raw for the whole window.""" + out = {} + for method in METHODS: + totals = {} + sel = f'method="{method}"' + ("" if method == "raw" else f",{flt}") + for _, samples in stack.matrix(f"asap_points{{{sel}}}"): + 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") + # The raw pipeline writes every point to its own store; whether it keeps + # up is the measurement, so its rate is reported, not required. + raw = [int(w["raw"][t]) for t in sorted(w["raw"])[1:-1]] + delay = {m["method"]: v for m, v in stack.vector("avg by (method) (avg_over_time(asap_answer_delay[5m]))")} + yield check("every pipeline reports its window completion delay", + set(delay) >= set(METHODS), + ", ".join(f"{m} {delay[m]:.1f} s" for m in METHODS if m in delay) + + f"; raw points/window {sorted(set(raw))} vs {expected}") + + +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) + pipelines, common = Path(tmp) / use_case, Path(tmp) / "common" + scaled_copy(use_case, pipelines) + scaled_copy("common", common) + for f in [pipelines, common, *pipelines.iterdir(), *common.iterdir()]: + f.chmod(0o755 if f.is_dir() else 0o644) + stack = Stack(shlex.split(args.docker_command), pipelines, common, 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 9b279f8..033fc0c 100644 --- a/deploy/telemetry-functions/compose.yaml +++ b/deploy/telemetry-functions/compose.yaml @@ -8,9 +8,9 @@ # pinned to one CPU, talking OTLP/HTTP over the compose network. The generators # of a source share one config (`downstream` is their pipeline's first stage) # and SCENARIO_START, so all three pipelines get identical points. -# USE_CASE picks the stage-config directory (latency, cardinality, ...), and -# ./common holds the stages every use case shares; CPU_* variables move a stage -# to another core. Prometheus, Grafana and Loki are the observers. +# PIPELINES is the use case's stage-config directory (run.sh sets it from +# USE_CASE), COMMON the stages every use case shares; CPU_* variables move a +# stage to another core. Prometheus, Grafana and Loki are the observers. x-collector: &collector image: asap-collector:local restart: "no" @@ -24,117 +24,117 @@ services: generator-a: <<: *generator command: ["--config", "/pipelines/generator-a.yaml"] - volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] + volumes: ["${PIPELINES:-./latency}:/pipelines:ro", "${COMMON:-./common}:/common:ro"] cpuset: "${CPU_GENERATOR_A:-0}" links: ["branch-a:downstream"] depends_on: [branch-a] generator-b: <<: *generator command: ["--config", "/pipelines/generator-b.yaml"] - volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] + volumes: ["${PIPELINES:-./latency}:/pipelines:ro", "${COMMON:-./common}:/common:ro"] cpuset: "${CPU_GENERATOR_B:-1}" links: ["branch-b:downstream"] depends_on: [branch-b] branch-a: <<: *collector command: ["--config", "/pipelines/branch.yaml"] - volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] + volumes: ["${PIPELINES:-./latency}:/pipelines:ro", "${COMMON:-./common}:/common:ro"] cpuset: "${CPU_BRANCH_A:-2}" depends_on: [merge] branch-b: <<: *collector command: ["--config", "/pipelines/branch.yaml"] - volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] + volumes: ["${PIPELINES:-./latency}:/pipelines:ro", "${COMMON:-./common}:/common:ro"] cpuset: "${CPU_BRANCH_B:-3}" depends_on: [merge] merge: <<: *collector command: ["--config", "/pipelines/merge.yaml"] - volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] + volumes: ["${PIPELINES:-./latency}:/pipelines:ro", "${COMMON:-./common}:/common:ro"] cpuset: "${CPU_MERGE:-4}" depends_on: [detector] detector: <<: *collector command: ["--config", "/pipelines/detector.yaml"] - volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] + volumes: ["${PIPELINES:-./latency}:/pipelines:ro", "${COMMON:-./common}:/common:ro"] cpuset: "${CPU_DETECTOR:-5}" depends_on: [prometheus] # exact pipeline exact-generator-a: <<: *generator command: ["--config", "/pipelines/generator-a.yaml"] - volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] + volumes: ["${PIPELINES:-./latency}:/pipelines:ro", "${COMMON:-./common}:/common:ro"] cpuset: "${CPU_EXACT_GENERATOR_A:-6}" links: ["exact-branch-a:downstream"] depends_on: [exact-branch-a] exact-generator-b: <<: *generator command: ["--config", "/pipelines/generator-b.yaml"] - volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] + volumes: ["${PIPELINES:-./latency}:/pipelines:ro", "${COMMON:-./common}:/common:ro"] cpuset: "${CPU_EXACT_GENERATOR_B:-7}" links: ["exact-branch-b:downstream"] depends_on: [exact-branch-b] exact-branch-a: <<: *collector command: ["--config", "/pipelines/exact-branch.yaml"] - volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] + volumes: ["${PIPELINES:-./latency}:/pipelines:ro", "${COMMON:-./common}:/common:ro"] cpuset: "${CPU_EXACT_BRANCH_A:-8}" depends_on: [exact-merge] exact-branch-b: <<: *collector command: ["--config", "/pipelines/exact-branch.yaml"] - volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] + volumes: ["${PIPELINES:-./latency}:/pipelines:ro", "${COMMON:-./common}:/common:ro"] cpuset: "${CPU_EXACT_BRANCH_B:-9}" depends_on: [exact-merge] exact-merge: <<: *collector command: ["--config", "/common/exact-merge.yaml"] - volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] + volumes: ["${PIPELINES:-./latency}:/pipelines:ro", "${COMMON:-./common}:/common:ro"] cpuset: "${CPU_EXACT_MERGE:-10}" depends_on: [exact-detector] exact-detector: <<: *collector command: ["--config", "/pipelines/exact-detector.yaml"] - volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] + volumes: ["${PIPELINES:-./latency}:/pipelines:ro", "${COMMON:-./common}:/common:ro"] cpuset: "${CPU_EXACT_DETECTOR:-11}" depends_on: [prometheus] # raw pipeline raw-generator-a: <<: *generator command: ["--config", "/pipelines/generator-a.yaml"] - volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] + volumes: ["${PIPELINES:-./latency}:/pipelines:ro", "${COMMON:-./common}:/common:ro"] cpuset: "${CPU_RAW_GENERATOR_A:-12}" links: ["raw-branch-a:downstream"] depends_on: [raw-branch-a] raw-generator-b: <<: *generator command: ["--config", "/pipelines/generator-b.yaml"] - volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] + volumes: ["${PIPELINES:-./latency}:/pipelines:ro", "${COMMON:-./common}:/common:ro"] cpuset: "${CPU_RAW_GENERATOR_B:-13}" links: ["raw-branch-b:downstream"] depends_on: [raw-branch-b] raw-branch-a: <<: *collector command: ["--config", "/common/raw-branch.yaml"] - volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] + volumes: ["${PIPELINES:-./latency}:/pipelines:ro", "${COMMON:-./common}:/common:ro"] cpuset: "${CPU_RAW_BRANCH_A:-14}" depends_on: [raw-merge] raw-branch-b: <<: *collector command: ["--config", "/common/raw-branch.yaml"] - volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] + volumes: ["${PIPELINES:-./latency}:/pipelines:ro", "${COMMON:-./common}:/common:ro"] cpuset: "${CPU_RAW_BRANCH_B:-15}" depends_on: [raw-merge] raw-merge: <<: *collector command: ["--config", "/common/raw-merge.yaml"] - volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] + volumes: ["${PIPELINES:-./latency}:/pipelines:ro", "${COMMON:-./common}:/common:ro"] cpuset: "${CPU_RAW_MERGE:-16}" depends_on: [raw-sink] raw-sink: <<: *collector command: ["--config", "/common/raw-sink.yaml"] - volumes: ["./${USE_CASE:-latency}:/pipelines:ro", "./common:/common:ro"] + volumes: ["${PIPELINES:-./latency}:/pipelines:ro", "${COMMON:-./common}:/common:ro"] cpuset: "${CPU_RAW_SINK:-17}" depends_on: [prometheus, loki-raw] @@ -164,3 +164,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..c702eb9 --- /dev/null +++ b/deploy/telemetry-functions/experiments.py @@ -0,0 +1,168 @@ +#!/usr/bin/env python3 +"""Cost and accuracy experiments on the real collector containers. + +pipelines The three pipelines (sketch, exact, raw) side by side over + offered rates, fed identical points: CPU of each stage, memory, + points delivered per second and window completion delay. The raw + pipeline's branch only forwards points, so the sketch branch's + CPU minus the raw branch's is the cost of the function. +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 pipelines --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 + +# Stages of each pipeline: per-source branches, merge, end. +PIPELINES = { + "sketch": ("branch", "merge", "detector"), + "exact": ("exact-branch", "exact-merge", "exact-detector"), + "raw": ("raw-branch", "raw-merge", "raw-sink"), +} + + +def prepare(use_case, root, rate, edits=()): + pipelines, common = root / use_case, root / "common" + acc.scaled_copy(use_case, pipelines) + acc.scaled_copy("common", common) + for f in pipelines.glob("generator-*.yaml"): + f.write_text(re.sub(r"signals_per_second: \d+", f"signals_per_second: {rate}", f.read_text())) + for name, old, new in edits: + path = pipelines / name + text = path.read_text() + assert old in text, (name, old) + path.write_text(text.replace(old, new)) + for f in [pipelines, common, *pipelines.iterdir(), *common.iterdir()]: + f.chmod(0o755 if f.is_dir() else 0o644) + return pipelines, common + + +def measure(stack, use_case, seconds): + """Averages over the last `seconds`.""" + rng = f"{seconds}s" + by = lambda label, expr: {m.get(label): 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)") + delay = by("method", f"avg by (method) (avg_over_time(asap_answer_delay[{rng}]))") + flt = acc.CASES[use_case][0] + row = {} + for name, (branch, merge, end) in PIPELINES.items(): + stages = [f"{branch}-a", f"{branch}-b", merge, end] + # Total points over the range divided by the windows it covers. + sel = f'asap_points{{method="{name}"' + ("" if name == "raw" else f",{flt}") + "}" + seen = (stack.vector(f"max(count_over_time({sel}[{rng}]))") or [({}, float("nan"))])[0][1] + total = (stack.vector(f"sum(sum_over_time({sel}[{rng}]))") or [({}, float("nan"))])[0][1] + row.update({ + f"{name}_delivered_per_s": total / (seen * acc.WINDOW_S), + f"{name}_branch_cpu_cores": statistics.mean([cpu.get(stages[0], 0), cpu.get(stages[1], 0)]), + f"{name}_merge_cpu_cores": cpu.get(merge, 0), + f"{name}_end_cpu_cores": cpu.get(end, 0), + f"{name}_rss_mib": sum(rss.get(s, 0) for s in stages) / 2**20, + f"{name}_delay_s": delay.get(name, float("nan")), + }) + transit = by("quantile", f"avg by (quantile) (avg_over_time(asap_raw_transit[{rng}]))") + row["raw_transit_p99_s"] = transit.get("0.99", float("nan")) + return row + + +def run(args, use_case, rate, edits=(), extra=None): + with tempfile.TemporaryDirectory(prefix="asap-exp-") as tmp: + pipelines, common = prepare(use_case, Path(tmp), rate, edits) + stack = acc.Stack(shlex.split(args.docker_command), pipelines, common, 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, "offered_per_source": rate} + row.update(measure(stack, use_case, 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 pipelines(args): + return [run(args, use_case, rate) for use_case in args.use_cases for rate in args.rates] + + +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)})) + # The top-k use case's 100,000 Zipf routes are enough keys for Count-Min + # collisions to matter. + for cols in ([] if args.only == "kll" else (256, 512, 1024, 2048)): + edit = ("branch.yaml", "columns: 512", f"columns: {cols}") + rows.append(run(args, "topk", args.rates[0], edits=[edit], + 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=("pipelines", "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 = {"pipelines": pipelines, "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..904544c --- /dev/null +++ b/deploy/telemetry-functions/report_experiments.py @@ -0,0 +1,122 @@ +#!/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 pipelines.csv and accuracy-size*.csv from RESULTS_DIR and writes +experiments.md and pipelines.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") + + +PIPES = {"sketch": "#2563eb", "exact": "#ea580c", "raw": "#64748b"} + + +def pipelines(data, out): + lines = ["## Sketch, exact and raw pipelines", "", + "The three pipelines run side by side, fed identical points by their own generators, each stage " + "on its own CPU. *Delivered* is points per second that reach the end of the pipeline (both " + "sources). *Branch* is one source's first stage (the average of the two), *merge* the second " + "stage, *end* the detector or the raw sink. *Delay* is the window completion delay: seconds from " + "a window's end until the detector has emitted its answers (sketch, exact) or the raw store has " + "acknowledged its last point (raw). *Function cost* is the sketch branch's CPU minus the raw " + "branch's (which only forwards points), per point.", ""] + cases = [c for c in NAMES if any(r["use_case"] == c for r in data)] + fig, axes = plt.subplots(2, len(cases), figsize=(4 * len(cases), 6.4), dpi=150, squeeze=False) + for col, case in enumerate(cases): + rs = sorted((r for r in data if r["use_case"] == case), key=lambda r: int(r["offered_per_source"])) + x = [int(r["offered_per_source"]) / 1000 for r in rs] + for name, color in PIPES.items(): + axes[0][col].plot(x, [num(r, f"{name}_branch_cpu_cores") for r in rs], marker="o", color=color, label=name) + axes[1][col].plot(x, [num(r, f"{name}_delay_s") for r in rs], marker="o", color=color, label=name) + axes[0][col].set_title(NAMES[case], fontsize=9) + axes[1][col].set_xlabel("offered points/s per source (thousands)") + for ax in axes[:, col]: + ax.grid(alpha=0.3) + lines += [f"### {NAMES[case]}", "", + "| Offered per source | Pipeline | Delivered | Branch CPU | Merge CPU | End CPU | Memory (all stages) | Delay | Function cost |", + "| ---: | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: |"] + for r in rs: + rate = int(r["offered_per_source"]) + per_branch = num(r, "raw_delivered_per_s") / 2 + for name in PIPES: + cost = "" + if name == "sketch": + added = num(r, "sketch_branch_cpu_cores") - num(r, "raw_branch_cpu_cores") + cost = f"{added / per_branch * 1e6:.1f} µs/point" + lines.append( + f"| {rate:,} | {name} | {num(r, f'{name}_delivered_per_s'):,.0f}/s | " + f"{num(r, f'{name}_branch_cpu_cores'):.3f} cores | {num(r, f'{name}_merge_cpu_cores'):.3f} cores | " + f"{num(r, f'{name}_end_cpu_cores'):.3f} cores | {num(r, f'{name}_rss_mib'):.0f} MiB | " + f"{num(r, f'{name}_delay_s'):.1f} s | {cost} |") + lines.append("") + axes[0][0].set_ylabel("branch CPU cores") + axes[1][0].set_ylabel("window completion delay (s)") + axes[0][0].legend(fontsize=7) + fig.tight_layout() + fig.savefig(out / "pipelines.png") + lines += ["![Branch CPU and window completion delay of each pipeline](pipelines.png)", ""] + 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: the top-k scenario's 100,000 Zipf routes. 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 / "pipelines.csv").exists(): + md += pipelines(rows(a.results / "pipelines.csv"), a.out) + 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 59bd0d4..7135d10 100755 --- a/deploy/telemetry-functions/run.sh +++ b/deploy/telemetry-functions/run.sh @@ -14,7 +14,10 @@ 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} +export COMMON=${COMMON:-./common} # Every generator counts scenario time from this second, so the three # pipelines get identical points. export SCENARIO_START=${SCENARIO_START:-$(($(date +%s) + 10))}