eval(#508c): DEBS/multisketch accuracy harness + benchmarks + docs

.gitignore

@@ -52,3 +52,12 @@ otel-app/otel-app
 deploy/scripts/__pycache__/
 *.pyc
 deploy/eval-results/
+
+# Compiled eval-driver binaries (built by the f2 eval scripts into their cmd dirs)
+asap-precompute-go/monitor/grpcclient/f2driver
+asap-precompute-go/monitor/grpcclient/e2edriver
+asap-precompute-go/monitor/grpcclient/cmd/*/f2driver
+asap-precompute-go/monitor/grpcclient/cmd/*/e2edriver
+
+# Local dataset working copies (not part of the repo)
+datasets_eval/debs/data/

datasets_eval/debs/scripts/debs_backend_accuracy.sh

@@ -0,0 +1,121 @@
+#!/usr/bin/env bash
+# Real-dataset QUERY ACCURACY through the REAL ASAPQuery-backend, on DEBS 2022.
+# Maps DEBS trading events → OTLP for the four sketch families, replays them
+# through a fused asap_edge agent into data_plane, then QUERIES the warm backend
+# and scores each answer against the EXACT ground truth computed offline from the
+# same DEBS rows:
+#   quantile (DDSketch+KLL) · topk (CountSketch) · distinct (HLL) · freq (CountMin)
+#
+# Uses the multisketch single-host stack.sh (agent + data_plane + control_plane),
+# the same real backend the cluster runs. Usage: debs_backend_accuracy.sh [n_events]
+set -uo pipefail
+ROOT=/mydata/ASAPCollector
+MS="${ROOT}/datasets_eval/multisketch"
+DEBS_CSV="${DEBS_CSV:-${ROOT}/datasets_eval/debs/data/debs2022-gc-trading-day-08-11-21.csv}"
+# N events. 500k DEBS events ≈ 1.55M OTLP points, which SEND in ~23s — well
+# within one boundary-aligned 60s window (measured). The wall-clock-anchor
+# collapses every row into ONE window; boundary alignment (below) gives the send
+# a full 60s of headroom so nothing seals mid-replay. A smaller slice (e.g. 60k)
+# is the flat intro of the trace — no heavy hitters, half the prices zero — so
+# keep the rich 500k slice; it fits.
+N="${1:-500000}"
+JSONL=/tmp/debs_otlp.jsonl; GT=/tmp/debs_gt.json
+BASE="http://127.0.0.1:9091"
+trap '[ -n "${KEEP_STACK:-}" ] || "${MS}/stack.sh" down >/dev/null 2>&1 || true' EXIT
+
+echo "== 1. map DEBS ($N events) → OTLP + exact GT =="
+python3 "${ROOT}/datasets_eval/debs/scripts/debs_otlp_map.py" "$DEBS_CSV" "$N" "$JSONL" "$GT"
+
+echo "== 2. bring up the real all-families stack (agent + data_plane + control_plane) =="
+AGENT_CFG="${AGENT_CFG:-${MS}/agent-allfamilies-coldoff.yaml}"
+echo "   agent config: ${AGENT_CFG}"
+"${MS}/stack.sh" up "${MS}/workloads/all-families.yaml" "${AGENT_CFG}" >/tmp/debs_stack.log 2>&1
+sleep 8
+
+echo "== 3. replay DEBS OTLP → agent :4317 (wall-clock-anchored) =="
+# Align to a fresh 60s wall-clock boundary so the collapsed instant sits at the
+# START of a window — the whole send then has a full 60s of headroom before the
+# window seals (otherwise a mid-window anchor can straddle the seal and drop the
+# tail of the replay).
+python3 -c 'import time; t=time.time(); s=60-(t%60); s=s if s>5 else s+60; print(f"   aligning to next 60s boundary in {s:.1f}s"); time.sleep(s)'
+# --anchor-span-s 50: spread the points across 50s (distinct ns each) inside the
+# one window, so the count-type sketches (topk_cs/freq_cms carry value==1.0 per
+# event) keep per-key multiplicity instead of deduping every event of a key onto
+# one identical (series, ts, value) sample.
+python3 "${ROOT}/datasets_eval/google_cluster/run.py" replay \
+  --jsonl "$JSONL" --endpoint 127.0.0.1:4317 --pace-factor 0 --wall-clock-anchor \
+  --anchor-span-s 45 \
+  >/tmp/debs_replay.log 2>&1
+echo "   waiting for the warm window to seal…"; sleep 75
+
+echo "== 4. query the backend + score vs exact GT =="
+python3 - "$BASE" "$GT" <<'PY'
+import sys, json, time, urllib.request, urllib.parse
+BASE, GT = sys.argv[1], sys.argv[2]
+gt = json.load(open(GT))
+def q(promql, t=None):
+    p = {"query": promql}
+    if t is not None: p["time"] = t
+    u = f"{BASE}/api/v1/query?" + urllib.parse.urlencode(p)
+    try:
+        body = json.loads(urllib.request.urlopen(u, timeout=60).read().decode())
+        if isinstance(body, dict):
+            return body.get("data", {}).get("result", []) or []
+    except Exception:
+        pass
+    return []
+def val(res):
+    try:
+        return float(res[0]["value"][1]) if res else None
+    except Exception:
+        return None
+# Instant sketch reads only resolve against the window whose seal is 'current' —
+# after it rolls the same query goes empty. Sweep time= backward to the offset
+# that returns the MOST series, i.e. the sealed replay window.
+def best_time(probe):
+    now = int(time.time()); best_t, best_n = now, -1
+    for off in range(0, 200, 8):
+        n = len(q(probe, t=now-off))
+        if n > best_n: best_n, best_t = n, now-off
+    return best_t, best_n
+# Probe with the instant topk aggregate (the bare metric never returns series).
+bt, bn = best_time('topk(50, google_cluster_2019_cpu_rate_topk_cs)')
+print(f"# sealed-window probe: t=now-{int(time.time())-bt}s has {bn} topk_cs series")
+# landed-event sanity: total inserts the CountSketch actually saw this window.
+tot = val(q('sum(count_over_time(google_cluster_2019_cpu_rate_topk_cs[300s]))', t=bt))
+print(f"# CountSketch landed events this window: {tot}  (GT events sent: {gt['events']})")
+
+print(f"\n{'query':<26} {'backend':>14} {'exact GT':>14} {'accuracy':>22}")
+print("-"*80)
+
+# distinct (HLL) — cardinality of distinct symbols
+r = q('count(google_cluster_2019_cpu_rate_card_hll)', t=bt)
+est = val(r); tru = gt["distinct"]
+acc = f"rel_err {abs(est-tru)/tru:.2%}" if est else f"({r})"
+print(f"{'distinct (HLL)':<26} {str(est):>14} {tru:>14} {acc:>22}")
+
+# frequency of one key — from the CountSketch heap (point-frequency sketch).
+sym = gt["freq_query"]["symbol"]; tru = gt["freq_query"]["true_count"]
+est = val(q(f'google_cluster_2019_cpu_rate_topk_cs{{item="{sym}"}}', t=bt))
+acc = f"rel_err {abs(est-tru)/tru:.2%}" if est else f"(not in heap)"
+print(f"{'freq '+sym[:16]+' (CS)':<26} {str(est):>14} {tru:>14} {acc:>22}")
+
+# topk (CountSketch) — recall@10 + dump the actual returned items/counts.
+r = q('topk(10, google_cluster_2019_cpu_rate_topk_cs)', t=bt)
+got_items = [(s["metric"].get("item"), s["value"][1]) for s in r]
+got = {i for i, _ in got_items if i}
+truk = {d["symbol"] for d in gt["topk10"]}
+rec = len(got & truk) / len(truk) if truk else 0
+print(f"{'topk@10 (CountSketch)':<26} {str(len(got & truk))+'/'+str(len(truk)):>14} {str(len(truk)):>14} {'recall '+f'{rec:.0%}':>22}")
+print(f"   backend top: {got_items[:5]}")
+print(f"   GT      top: {[(d['symbol'], d['count']) for d in gt['topk10'][:5]]}")
+
+# quantile (DDSketch, KLL) — dump p50/p90/p99 to see if it's a scale/tail issue.
+for fam, m in [("DDSketch", "q_ddsketch"), ("KLL", "q_kll")]:
+    for ql, key in [(0.50, "p50"), (0.90, "p90"), (0.99, "p99")]:
+        est = val(q(f'quantile_over_time({ql}, google_cluster_2019_cpu_rate_{m}[300s])', t=bt))
+        tru = gt["quantile"][key]
+        acc = f"rel_err {abs(est-tru)/tru:.2%}" if est and tru else f"(empty)"
+        print(f"{fam+' '+key:<26} {str(round(est,2) if est else est):>14} {str(tru):>14} {acc:>22}")
+PY
+echo "== done =="

datasets_eval/debs/scripts/debs_otlp_map.py

@@ -0,0 +1,93 @@
+#!/usr/bin/env python3
+"""Map real DEBS 2022 trading events → an OTLP-replay JSONL that exercises the
+asap agent's four sketch families, and compute the EXACT ground truth for each
+query so the backend answer can be scored:
+
+  metric (agent family)              query           GT
+  ---------------------------------  --------------  --------------------------
+  top_endpoint_qps    (CountSketch)  topk symbols    exact top-10 by trade count
+  endpoint_request_freq (CountMin)   freq of a key   exact count of one symbol
+  unique_users_per_min (HLL)         distinct count  exact distinct symbols
+  http_requests_total_latency_ms(KLL) quantile       exact p50/p90/p99 of Ask price
+
+JSONL line: {"timestamp_ms","series_id","metric_name","value","attributes":{}}
+Usage: debs_otlp_map.py <csv> <n_events> <out.jsonl> <gt.json>
+"""
+import sys, json, time
+from collections import Counter
+
+csv, n, out_jsonl, gt_path = sys.argv[1], int(sys.argv[2]), sys.argv[3], sys.argv[4]
+
+counts = Counter()      # symbol -> trade count (topk / frequency GT)
+prices = []             # Ask prices (quantile GT)
+distinct = set()
+now_ms = int(time.time() * 1000)
+
+read = 0
+with open(csv, errors="replace") as f, open(out_jsonl, "w") as w:
+    for line in f:
+        if line[0] == "#" or line.startswith("ID,"):
+            continue
+        p = line.split(",")
+        sym = p[0]
+        if not sym:
+            continue
+        ts = now_ms + read  # monotone, wall-clock-anchored downstream
+        # Metric names + labels match agent-allfamilies-coldoff.yaml so the
+        # EXISTING all-families stack serves them (topk item_label=host,
+        # cms/hll item_label=service).
+        # topk (CountSketch) + frequency (CountMin): one point per event, symbol.
+        w.write(json.dumps({"timestamp_ms": ts, "series_id": f"tk:{sym}",
+                            "metric": "google_cluster_2019_cpu_rate_topk_cs", "value": 1.0,
+                            "attributes": {"host": sym}}) + "\n")
+        w.write(json.dumps({"timestamp_ms": ts, "series_id": f"fq:{sym}",
+                            "metric": "google_cluster_2019_cpu_rate_freq_cms", "value": 1.0,
+                            "attributes": {"service": sym}}) + "\n")
+        # distinct (HLL): distinct label = symbol.
+        w.write(json.dumps({"timestamp_ms": ts, "series_id": f"hll:{sym}",
+                            "metric": "google_cluster_2019_cpu_rate_card_hll", "value": 1.0,
+                            "attributes": {"service": sym}}) + "\n")
+        counts[sym] += 1
+        distinct.add(sym)
+        # quantile (DDSketch + KLL): Ask price (col 5) when present.
+        # ALL prices fold into ONE global sketch (constant item) so the backend
+        # quantile query returns a single global p99 comparable to the global GT.
+        # (Per-symbol DDSketch would need per-symbol GT; a single global quantile
+        # is the clean headline number the user asked for.)
+        if len(p) > 4 and p[4]:
+            try:
+                v = float(p[4])
+                for m in ("google_cluster_2019_cpu_rate_q_ddsketch",
+                          "google_cluster_2019_cpu_rate_q_kll"):
+                    w.write(json.dumps({"timestamp_ms": ts, "series_id": "px:ALL",
+                                        "metric": m, "value": v,
+                                        "attributes": {"host": "ALL"}}) + "\n")
+                prices.append(v)
+            except ValueError:
+                pass
+        read += 1
+        if read >= n:
+            break
+
+prices.sort()
+def pq(q):
+    return prices[min(len(prices) - 1, int(q * len(prices)))] if prices else None
+top10 = counts.most_common(10)
+# Point-frequency query is answered from the CountSketch heap (topk_cs), which
+# retains the top heap_size=100 keys with their estimated counts — so pick a key
+# comfortably INSIDE the heap (rank ~15) rather than at its edge. (The CountMin
+# keyed frequency path is a warm-only safe-miss by design — Phase 2b unfinished —
+# so in the cold-OFF stack a CMS point query returns no result; the CountSketch
+# is itself a point-frequency sketch and answers the same question.)
+freq_key, freq_true = counts.most_common(20)[-1] if len(counts) > 20 else top10[0]
+gt = {
+    "events": read,
+    "topk10": [{"symbol": s, "count": c} for s, c in top10],
+    "freq_query": {"symbol": freq_key, "true_count": freq_true},
+    "distinct": len(distinct),
+    "quantile": {"p50": pq(0.50), "p90": pq(0.90), "p99": pq(0.99), "n": len(prices)},
+}
+json.dump(gt, open(gt_path, "w"), indent=1)
+print(f"# DEBS OTLP map: events={read} distinct={len(distinct)} prices={len(prices)}", file=sys.stderr)
+print(f"# top symbol: {top10[0]}  freq_query symbol: {freq_key}(true={freq_true})", file=sys.stderr)
+print(f"# price quantiles p50={pq(0.5)} p90={pq(0.9)} p99={pq(0.99)}", file=sys.stderr)

datasets_eval/google_cluster/e2e/INTEGRATED_SWEEP_RESULTS.md

@@ -0,0 +1,30 @@
+# Integrated ε-sweep — end-to-end metrics under coordinated sampling
+
+`epsilon_e2e_sweep.py` extends `pareto_sweep.py` to emit **6 metric classes in one
+run**, swept over the admission p (= the ε-floor `p=1/(1+ε²·rate)` the autonomous
+coordinator sets for the corresponding ε): query **accuracy**, query **latency**
+(p50/p99), data **freshness** (sample-generation → backend-queryable), and
+per-process **CPU / RSS / wire bandwidth**.
+
+## Phase-1 (single-node, google-cluster-2019 cpu_rate, pooled, GT p99=0.0494)
+
+| arm | ε | p | accuracy | lat p99 | freshness | bk RSS | pr RSS | cpu cores | wire kbps |
+|---|---|---|---|---|---|---|---|---|---|
+| raw    | 0     | 1.00 | 1.000  | 2.5 ms | —     | 47 MB | 71 MB | 0.115 | 14.7 |
+| dd     | 0     | 1.00 | 0.867* | 2.6 ms | 19 ms | 15 MB | 60 MB | 0.100 | 0.4 |
+| dd     | 0.005 | 0.50 | 0.867  | 2.6 ms | 19 ms | 15 MB | 60 MB | 0.099 | 0.4 |
+| dd     | 0.008 | 0.25 | 0.850  | 2.6 ms | 19 ms | 15 MB | 58 MB | 0.091 | 0.4 |
+| dd     | 0.014 | 0.10 | 0.850  | 2.6 ms | 19 ms | 15 MB | 55 MB | 0.081 | 0.3 |
+| dd     | 0.020 | 0.05 | 0.850  | 2.6 ms | 20 ms | 16 MB | 51 MB | 0.093 | 0.3 |
+
+**Takeaways:** sampling to p=0.05 keeps accuracy flat (high-N pooled is
+sampling-robust), drops edge CPU 0.10→0.08, keeps wire ~40× below raw, backend RSS
+bounded ~15 MB. Warm latency ~2.6 ms (vs cold-tier ~51 ms, Fig 7 / #501).
+
+**Caveats (single-node):** (1) `*`accuracy ~13% low — the regenerated `[30s]`
+window reads a sub-pool; the *calibrated* accuracy-vs-p is in
+`docs/evaluation-plan-figures.md` Fig 3. (2) freshness ~19 ms is a single-node
+artifact — the **compressed** replay (31 d → ~20 s) seals windows mid-run, so
+emit→queryable collapses to ~0. Real freshness + per-component (edge/dp/cp) +
+cold-tier routing are the **Phase-2 cluster run** (wall-clock-paced,
+`measure_freshness.sh` + the probe; per-container `snapshot_resources.sh`).