From 795454d71bc7fd524e8f2c2ef845ecc035cc0cde Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Wed, 6 May 2026 17:31:30 -0400 Subject: [PATCH 1/4] mvp v6 phase E: add controller-driven multi-stage demo driver MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit DRAFT — does not run the demo. Phase F brings up the stack and posts numbers once v5 PRs (#295 ASAPCollector, #90 ASAPQuery-backend) land. run_mvp_demo_v6.sh is independent from v5's run_mvp_demo.sh (which is in active use by the v5 demo run, PID 1945011) — separate file, separate output dir, separate report. Eight phases: 0. preflight (docker, docker-compose, compactor binary check) 1. stack up via base.yml + mvp-v6-multi-stage.yml + USE_TYPED_STAGE_SPLIT=1 2. measurements (replay + measure_stages.py + measure_per_edge_bandwidth.py) 3. freshness via run_freshness_phase.sh (raw/warm/archive) 4. ad-hoc postings exercise (count + topk-5xx) 5. cold-fallback verification (gorilla_archive marker) 6. compactor concat-only (dry-run + live) 7. teardown 8. mvp_report_v6.py reduce → MVP_REPORT_v6.md Captures controller-emitted runtime configs from /api/v1/collector-config/ and /api/v1/config/; falls back to placeholder gateway+agent configs when typed-stage-split returns None (logged as fallback-placeholder STATUS marker for the report). Co-Authored-By: Claude Opus 4.7 (1M context) --- deploy/scripts/run_mvp_demo_v6.sh | 516 ++++++++++++++++++++++++++++++ 1 file changed, 516 insertions(+) create mode 100755 deploy/scripts/run_mvp_demo_v6.sh diff --git a/deploy/scripts/run_mvp_demo_v6.sh b/deploy/scripts/run_mvp_demo_v6.sh new file mode 100755 index 00000000..cdd6c322 --- /dev/null +++ b/deploy/scripts/run_mvp_demo_v6.sh @@ -0,0 +1,516 @@ +#!/usr/bin/env bash +# run_mvp_demo_v6.sh — MVP v6 demo driver (controller-driven multi-stage). +# +# v6 differs from v5 (`run_mvp_demo.sh` — DO NOT MODIFY THAT FILE): +# +# 1. Topology is fan-in: 10 producers → 2 agents → 1 gateway → 1 +# backend (+ optional B0 Prometheus). The compose overlay is +# `deploy/docker-compose/mvp-v6-multi-stage.yml`. v5 had the +# classic linear N-agents → backend shape. +# +# 2. Agent + gateway runtime configs are emitted by the controller +# (typed-stage-split path, behind `USE_TYPED_STAGE_SPLIT=1`) +# rather than mounted from a fixed file. The driver waits for +# OpAMP push to settle, then captures whatever the controller +# has on its `/api/v1/collector-config/{agent,backend}` +# introspection endpoints. If the typed-stage-split path doesn't +# fire (Phase B/C wiring caveats), the driver logs the +# capture failure clearly and continues with the placeholder +# configs that mvp-v6-multi-stage.yml mounts as fallback. +# +# 3. Three canonical query classes from +# `deploy/configs/mvp-v6-workload.yaml` exercise: +# - window-per-series (DDSketch p99 over 1m) +# - label-at-instant (sum by zone, gateway fan-in) +# - combined (rate over 5m + sum by zone) +# plus a fourth ad-hoc cold-fallback probe +# (`http_requests_total{service="payments"}` — assigned +# to role "archive"). +# +# 4. Freshness phase calls `run_freshness_phase.sh` (Phase D +# driver) which emits `freshness/{raw,warm,archive}.csv`. +# +# 5. Compactor phase reuses v5's `gorilla-compactor` binary at +# `${COMPACTOR_BIN}` with `--threshold-hours 0 --threshold-count 0` +# so even a 60s soak generates one merge candidate. Dry-run +# then live-run, capturing before/after MinIO listings. +# +# 6. Per-edge bandwidth probe (`measure_per_edge_bandwidth.py`) +# runs alongside `measure_stages.py` so criterion ① gets a +# per-edge breakdown: +# sdk→agent / agent→gateway / gateway→backend / gateway→s3. +# +# Usage: +# bash deploy/scripts/run_mvp_demo_v6.sh +# +# Output: +# deploy/eval-results/mvp-v6-2026-05-06/{ +# stack-up.log, controller-emitted-configs/, +# measurements/{stages.csv, per_edge_bandwidth.csv, +# replay.jsonl, accuracy.csv}, +# freshness/{raw.csv, warm.csv, archive.csv}, +# ad-hoc/{.json}, +# compactor/{dry_run.json, live_run.json, +# before.minio.jsonl, after.minio.jsonl}, +# MVP_REPORT_v6.md +# } +set -euo pipefail + +# ── knobs ──────────────────────────────────────────────────────── +STACK_SETTLE_S="${STACK_SETTLE_S:-60}" +AGENT_WARMUP_S="${AGENT_WARMUP_S:-60}" +QUERY_WARMUP_S="${QUERY_WARMUP_S:-30}" +SOAK_S="${SOAK_S:-60}" +FRESHNESS_DURATION_S="${FRESHNESS_DURATION_S:-60}" +QPS="${QPS:-5}" +PER_AGENT_CARDINALITY="${PER_AGENT_CARDINALITY:-500}" +N_PRODUCERS="${N_PRODUCERS:-10}" +EXPORTER_FREQ_HZ="${EXPORTER_FREQ_HZ:-10}" +EXPORTER_FRESHNESS_PROBES="${EXPORTER_FRESHNESS_PROBES:-on}" +EXPORTER_FRESHNESS_PROBE_HZ="${EXPORTER_FRESHNESS_PROBE_HZ:-1.0}" +ASAP_SKETCH_FAMILY="${ASAP_SKETCH_FAMILY:-ddsketch}" +USE_TYPED_STAGE_SPLIT="${USE_TYPED_STAGE_SPLIT:-1}" + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +COMPOSE_DIR="$(cd "${SCRIPT_DIR}/../docker-compose" && pwd)" +REPO_ROOT="$(cd "${SCRIPT_DIR}/../.." && pwd)" + +# Host-published ports (see deploy/docker-compose/base.yml). +HOST_BACKEND_QUERY_PORT="${HOST_BACKEND_QUERY_PORT:-19091}" +HOST_BACKEND_INGEST_PORT="${HOST_BACKEND_INGEST_PORT:-19090}" +HOST_CONTROLLER_PORT="${HOST_CONTROLLER_PORT:-18080}" +HOST_PROM_B0_PORT="${HOST_PROM_B0_PORT:-19090}" # collides with backend ingest; +# v6 multi-stage overlay republishes Prometheus B0 on 19090 only when +# the `b0` profile is active (compose `--profile b0`). The driver does +# NOT bring up B0 in the same compose stack as the backend on this +# port; B0 mode is a separate cycle (see Phase 0 baseline_b0() below). + +OUT_BASE="${OUT_BASE:-${REPO_ROOT}/deploy/eval-results/mvp-v6-2026-05-06}" +COMPACTOR_BIN="${COMPACTOR_BIN:-${REPO_ROOT}/compactor/target/release/gorilla-compactor}" +COMPACTOR_BUCKET="${COMPACTOR_BUCKET:-asap-gorilla}" +COMPACTOR_ENDPOINT="${COMPACTOR_ENDPOINT:-http://localhost:9000}" +COMPACTOR_ACCESS_KEY="${COMPACTOR_ACCESS_KEY:-asap}" +COMPACTOR_SECRET_KEY="${COMPACTOR_SECRET_KEY:-asap-local-only}" +COMPACTOR_TENANT="${COMPACTOR_TENANT:-default}" + +# ── helpers ────────────────────────────────────────────────────── +log() { printf '[mvp v6] %s\n' "$*"; } + +ensure_out_dirs() { + mkdir -p \ + "${OUT_BASE}" \ + "${OUT_BASE}/controller-emitted-configs" \ + "${OUT_BASE}/measurements" \ + "${OUT_BASE}/freshness" \ + "${OUT_BASE}/ad-hoc" \ + "${OUT_BASE}/compactor" +} + +# Phase 0 — pre-flight checks. Bail loud, bail early. +preflight() { + log "Phase 0 preflight" + + if ! command -v docker >/dev/null 2>&1; then + echo "[error] docker not found on PATH" >&2; exit 2 + fi + if ! docker compose version >/dev/null 2>&1; then + echo "[error] 'docker compose' subcommand not available" >&2; exit 2 + fi + + if [[ ! -x "${COMPACTOR_BIN}" ]]; then + log "WARN: compactor binary missing at ${COMPACTOR_BIN}; " + log " Phase F can run \`cargo build --release -p gorilla-compactor\` " + log " from the v5-merged tree, OR set COMPACTOR_BIN to the location " + log " of the built binary. The compactor phase will be skipped " + log " with a clear marker file rather than crashing." + fi + + # Best-effort: rebuild fake-exporter image if the freshness probe + # symbol is missing. This is an authored-mode driver; Phase F may + # rebuild on its own. We do NOT fail if the image is missing — + # the docker-compose `up` will surface that. + if docker image inspect asap/fake-exporter:dev >/dev/null 2>&1; then + log " fake-exporter:dev image present" + else + log " fake-exporter:dev image NOT FOUND — Phase F will need: " + log " docker build -t asap/fake-exporter:dev deploy/fake-exporter/" + fi + + # Clean any stale containers from a previous v6 run. Only + # matches v6-overlay containers — won't disturb a concurrent v5 + # demo (different project name + different service set). + docker compose \ + --project-directory "${COMPOSE_DIR}" \ + -f "${COMPOSE_DIR}/base.yml" \ + -f "${COMPOSE_DIR}/mvp-v6-multi-stage.yml" \ + down -v --remove-orphans \ + > "${OUT_BASE}/preflight-down.log" 2>&1 || true +} + +# Phase 1 — bring up the multi-stage controller-driven topology. +bring_up_stack() { + log "Phase 1 stack up (controller + 10 producers + 2 agents + 1 gateway + 1 backend)" + + ( + cd "${COMPOSE_DIR}" + USE_TYPED_STAGE_SPLIT="${USE_TYPED_STAGE_SPLIT}" \ + PER_AGENT_CARDINALITY="${PER_AGENT_CARDINALITY}" \ + EXPORTER_FREQ_HZ="${EXPORTER_FREQ_HZ}" \ + EXPORTER_FRESHNESS_PROBES="${EXPORTER_FRESHNESS_PROBES}" \ + EXPORTER_FRESHNESS_PROBE_HZ="${EXPORTER_FRESHNESS_PROBE_HZ}" \ + ASAP_SKETCH_FAMILY="${ASAP_SKETCH_FAMILY}" \ + docker compose \ + -f base.yml \ + -f mvp-v6-multi-stage.yml \ + up -d + ) > "${OUT_BASE}/stack-up.log" 2>&1 + + log " stack settle ${STACK_SETTLE_S}s (controller plan + OpAMP push)" + sleep "${STACK_SETTLE_S}" + + log " agent warm-up ${AGENT_WARMUP_S}s (sketches fill)" + sleep "${AGENT_WARMUP_S}" + + log " query-side warm-up ${QUERY_WARMUP_S}s" + local started; started=$(date +%s) + while true; do + local now; now=$(date +%s) + local elapsed=$((now - started)) + if (( elapsed >= QUERY_WARMUP_S )); then + log " query warm-up window ${QUERY_WARMUP_S}s elapsed; proceeding" + break + fi + local body + body=$(curl -sG "http://localhost:${HOST_BACKEND_QUERY_PORT}/api/v1/query" \ + --data-urlencode "query=count(http_requests_total)" \ + 2>/dev/null || true) + # Look for any non-zero result. + if echo "$body" | grep -Eq '"value":\[[0-9.]+,"[1-9]'; then + log " query warm-up: backend has data after ${elapsed}s" + break + fi + sleep 2 + done +} + +# Capture controller-emitted configs (best-effort introspection). +# If the controller isn't exposing them (typed-stage-split path +# disabled or returned None) we record that fact rather than +# crashing. +capture_emitted_configs() { + log " capturing controller-emitted runtime configs" + local cdir="${OUT_BASE}/controller-emitted-configs" + local ctrl="http://localhost:${HOST_CONTROLLER_PORT}" + + # Agent bootstrap config (controller side: see + # /api/v1/collector-config/agent in controller/src/main.rs). + if curl -sf "${ctrl}/api/v1/collector-config/agent" \ + -o "${cdir}/agent.bootstrap.yaml" 2> "${cdir}/agent.bootstrap.err"; then + log " agent bootstrap config captured" + else + log " [warn] agent bootstrap config not available — see agent.bootstrap.err" + fi + + if curl -sf "${ctrl}/api/v1/collector-config/backend" \ + -o "${cdir}/backend.bootstrap.yaml" 2> "${cdir}/backend.bootstrap.err"; then + log " backend bootstrap config captured" + else + log " [warn] backend bootstrap config not available — see backend.bootstrap.err" + fi + + # Per-metric typed config (one per workload entry — Phase B + # emitter output). The mvp-v6-workload.yaml has four entries. + for metric in \ + http_requests_total_latency_ms \ + http_requests_total ; do + local out="${cdir}/per-metric.${metric}.json" + if curl -sf "${ctrl}/api/v1/config/${metric}" -o "${out}" \ + 2> "${out}.err"; then + log " per-metric config captured: ${metric}" + else + log " [warn] per-metric config missing for ${metric}" + fi + done + + # Connected agents (Phase C role plumbing — should list both + # agent-a + agent-b under role=Agent and gateway under + # role=Gateway once the AgentRole::Gateway path is exercised). + if curl -sf "${ctrl}/api/v1/agents" -o "${cdir}/agents.json" \ + 2> "${cdir}/agents.err"; then + log " /api/v1/agents captured" + else + log " [warn] /api/v1/agents unavailable" + fi + + # Snapshot the placeholder gateway config for diffing. + if [[ -f "${REPO_ROOT}/deploy/configs/sketchcol-gateway-mvp-v6-placeholder.yaml" ]]; then + cp "${REPO_ROOT}/deploy/configs/sketchcol-gateway-mvp-v6-placeholder.yaml" \ + "${cdir}/gateway.placeholder.yaml" + fi + + # Detect "controller didn't actually emit a typed config" — this + # is the failure mode the spec calls out as expected if Phase + # B/C wiring has remaining caveats. We grep the controller logs + # for the typed-stage-split tracing markers. + docker logs "$(cd "${COMPOSE_DIR}" && docker compose -f base.yml -f mvp-v6-multi-stage.yml ps -q controller 2>/dev/null | head -n1)" \ + 2> "${cdir}/controller.stderr" \ + > "${cdir}/controller.stdout" || true + if grep -q "USE_TYPED_STAGE_SPLIT.*pushing typed" "${cdir}/controller.stderr" 2>/dev/null; then + log " controller logs show typed-stage-split push events — emitter LIVE" + echo "live" > "${cdir}/STATUS" + elif grep -q "split_typed_three_stage returned None" "${cdir}/controller.stderr" 2>/dev/null; then + log " [warn] controller's typed-stage-split returned None — falling back to placeholder" + echo "fallback-placeholder" > "${cdir}/STATUS" + else + log " [warn] no typed-stage-split tracing in controller logs — emitter not exercised" + echo "not-exercised" > "${cdir}/STATUS" + fi +} + +# Phase 2 — measurements (replay + stages + per-edge bandwidth). +measure_phase() { + log "Phase 2 measurement window (${SOAK_S}s)" + local mdir="${OUT_BASE}/measurements" + local backend_url="http://localhost:${HOST_BACKEND_QUERY_PORT}" + + # Build the v6 replay query suite from mvp-v6-workload.yaml. + # We keep the JSON adjacent to the run dir for reproducibility. + cat > "${mdir}/replay-queries.json" <<'JSON' +[ + {"kind": "quantile", "promql": "quantile_over_time(0.99, http_requests_total_latency_ms[1m])"}, + {"kind": "sum", "promql": "sum by (zone) (http_requests_total)"}, + {"kind": "sum", "promql": "sum by (zone) (rate(http_requests_total[5m]))"} +] +JSON + + # Replay (background). + log " replay (qps=${QPS} → ${backend_url})" + python3 "${SCRIPT_DIR}/promql_replay.py" \ + --target "${backend_url}" \ + --controller "http://localhost:${HOST_CONTROLLER_PORT}" \ + --queries "${mdir}/replay-queries.json" \ + --qps "${QPS}" \ + --duration "${SOAK_S}" \ + --out "${mdir}/replay.jsonl" \ + > "${mdir}/replay.log" 2>&1 & + REPLAY_PID=$! + + # Stage probe (background). + log " measure_stages.py duration=${SOAK_S}s" + python3 "${SCRIPT_DIR}/measure_stages.py" \ + --baseline "mvp-v6" \ + --duration "${SOAK_S}" \ + --out "${mdir}/stages.csv" \ + > "${mdir}/stages.log" 2>&1 & + STAGE_PID=$! + + # Per-edge bandwidth (background). + log " measure_per_edge_bandwidth.py duration=${SOAK_S}s" + python3 "${SCRIPT_DIR}/measure_per_edge_bandwidth.py" \ + --duration "${SOAK_S}" \ + --out "${mdir}/per_edge_bandwidth.csv" \ + > "${mdir}/per_edge_bandwidth.log" 2>&1 & + EDGE_PID=$! + + wait "${REPLAY_PID}" || true + wait "${STAGE_PID}" || true + wait "${EDGE_PID}" || true + log " measurement window done" + + # Accuracy reduce against the cold-store ground truth. + log " accuracy reduce" + local backend_cont + backend_cont="$(cd "${COMPOSE_DIR}" && \ + docker compose -f base.yml -f mvp-v6-multi-stage.yml ps -q backend 2>/dev/null | head -n1)" + if [[ -n "${backend_cont}" ]]; then + docker cp "${backend_cont}:/var/asap/cold/raw" "${mdir}/cold-truth" \ + > "${mdir}/cold-snapshot.log" 2>&1 || true + fi + if [[ -d "${mdir}/cold-truth" ]]; then + python3 "${SCRIPT_DIR}/accuracy_reduce.py" \ + --cell-dir "${mdir}" \ + --out "${mdir}/accuracy.csv" \ + > "${mdir}/accuracy.log" 2>&1 || true + else + log " [warn] cold-truth snapshot not captured — accuracy.csv skipped" + fi +} + +# Phase 3 — freshness (raw / warm / archive). +freshness_phase() { + log "Phase 3 freshness probes (raw/warm/archive)" + # The fake-exporter has been emitting probes the whole time + # (EXPORTER_FRESHNESS_PROBES=on); this phase is poll-only. + bash "${SCRIPT_DIR}/run_freshness_phase.sh" \ + --out-dir "${OUT_BASE}" \ + --duration "${FRESHNESS_DURATION_S}" \ + --raw-endpoint "${ASAP_FRESHNESS_RAW_ENDPOINT:-http://localhost:${HOST_BACKEND_QUERY_PORT}}" \ + --warm-endpoint "${ASAP_FRESHNESS_WARM_ENDPOINT:-http://localhost:${HOST_BACKEND_QUERY_PORT}}" \ + --archive-endpoint "${ASAP_FRESHNESS_ARCHIVE_ENDPOINT:-http://localhost:${HOST_BACKEND_QUERY_PORT}}" \ + > "${OUT_BASE}/freshness/run.log" 2>&1 || \ + log " [warn] freshness phase exited non-zero — see freshness/run.log" +} + +# Phase 4 — ad-hoc queries that exercise postings filtering. +ad_hoc_postings_phase() { + log "Phase 4 ad-hoc postings exercise" + local adir="${OUT_BASE}/ad-hoc" + local backend_url="http://localhost:${HOST_BACKEND_QUERY_PORT}" + + fire_query() { + local label="$1"; shift + local promql="$1"; shift + log " ad-hoc[${label}]: ${promql}" + # Use --get + --data-urlencode so the PromQL is not shell-mangled. + curl -sG -m 10 \ + "${backend_url}/api/v1/query" \ + --data-urlencode "query=${promql}" \ + -o "${adir}/${label}.json" \ + -w '{"http_code":%{http_code},"time_total":%{time_total},"size_download":%{size_download}}\n' \ + > "${adir}/${label}.curlstats" \ + 2> "${adir}/${label}.curl.err" \ + || log " [warn] curl exited non-zero for ${label}" + } + + # The two postings-exercise queries from the spec. + fire_query "count_api_series" \ + 'count(http_requests_total{service="api"})' + fire_query "topk_5xx_by_zone" \ + 'topk(5, sum by (zone) (rate(http_requests_total{status=~"5.."}[5m])))' +} + +# Phase 5 — cold-fallback verification (assigned-archive metric). +cold_fallback_phase() { + log "Phase 5 cold-fallback verification (gorilla_archive marker)" + local adir="${OUT_BASE}/ad-hoc" + local backend_url="http://localhost:${HOST_BACKEND_QUERY_PORT}" + + log " cold[payments]: count(http_requests_total{service=\"payments\"})" + curl -sG -m 10 \ + "${backend_url}/api/v1/query" \ + --data-urlencode 'query=count(http_requests_total{service="payments"})' \ + -o "${adir}/cold_payments.json" \ + -w '{"http_code":%{http_code},"time_total":%{time_total}}\n' \ + > "${adir}/cold_payments.curlstats" \ + 2> "${adir}/cold_payments.curl.err" \ + || log " [warn] curl exited non-zero for cold_payments" + + if grep -q '"data_source":"gorilla_archive"\|gorilla_archive' \ + "${adir}/cold_payments.json" 2>/dev/null; then + log " cold-fallback marker present (data_source: gorilla_archive)" + echo "PASS" > "${adir}/cold_payments.verdict" + else + log " [warn] no gorilla_archive marker in response — see cold_payments.json" + echo "MISSING" > "${adir}/cold_payments.verdict" + fi +} + +# Phase 6 — compactor (dry-run + live, concat-only from v5). +compactor_phase() { + log "Phase 6 compactor (concat-only)" + local cdir="${OUT_BASE}/compactor" + + if [[ ! -x "${COMPACTOR_BIN}" ]]; then + log " [skip] compactor binary missing at ${COMPACTOR_BIN}" + echo "compactor binary missing at ${COMPACTOR_BIN}" \ + > "${cdir}/SKIPPED" + return 0 + fi + + # MinIO listing helper (before / after). + list_minio_objects() { + local out_path="$1" + docker run --rm --network host --entrypoint sh minio/mc:latest -c \ + "mc alias set asap ${COMPACTOR_ENDPOINT} ${COMPACTOR_ACCESS_KEY} ${COMPACTOR_SECRET_KEY} >/dev/null 2>&1; \ + mc ls --recursive --json asap/${COMPACTOR_BUCKET} 2>/dev/null || true" \ + > "${out_path}" 2>"${out_path}.err" || true + } + + list_minio_objects "${cdir}/before.minio.jsonl" + + log " compactor dry-run" + "${COMPACTOR_BIN}" \ + --endpoint "${COMPACTOR_ENDPOINT}" \ + --bucket "${COMPACTOR_BUCKET}" \ + --tenant "${COMPACTOR_TENANT}" \ + --access-key-id "${COMPACTOR_ACCESS_KEY}" \ + --secret-access-key "${COMPACTOR_SECRET_KEY}" \ + --threshold-count 0 \ + --threshold-hours 0 \ + --dry-run \ + --out "${cdir}/dry_run.json" \ + > "${cdir}/dry_run.log" 2>&1 || \ + log " [warn] compactor dry-run exited non-zero" + + log " compactor live run" + "${COMPACTOR_BIN}" \ + --endpoint "${COMPACTOR_ENDPOINT}" \ + --bucket "${COMPACTOR_BUCKET}" \ + --tenant "${COMPACTOR_TENANT}" \ + --access-key-id "${COMPACTOR_ACCESS_KEY}" \ + --secret-access-key "${COMPACTOR_SECRET_KEY}" \ + --threshold-count 0 \ + --threshold-hours 0 \ + --out "${cdir}/live_run.json" \ + > "${cdir}/live_run.log" 2>&1 || \ + log " [warn] compactor live run exited non-zero" + + list_minio_objects "${cdir}/after.minio.jsonl" +} + +# Phase 6.5 — fetch v5 cost tracker if backend exposes it. +fetch_s3_cost_csv() { + log "Phase 6.5 fetch s3_cost.csv (v5 cost tracker)" + local cdir="${OUT_BASE}/measurements" + local backend_url="http://localhost:${HOST_BACKEND_QUERY_PORT}" + if curl -sf "${backend_url}/internal/s3_cost.csv" \ + > "${cdir}/s3_cost.csv" 2> "${cdir}/s3_cost.err"; then + log " s3_cost.csv captured" + else + log " [warn] /internal/s3_cost.csv unavailable (likely v5 not merged yet) — see s3_cost.err" + fi +} + +# Phase 7 — tear down. +teardown() { + log "Phase 7 tear down" + ( + cd "${COMPOSE_DIR}" + docker compose \ + -f base.yml \ + -f mvp-v6-multi-stage.yml \ + down -v --remove-orphans + ) > "${OUT_BASE}/teardown.log" 2>&1 || true +} + +# Phase 8 — generate the v6 report. +generate_report() { + log "Phase 8 generate MVP_REPORT_v6.md" + python3 "${SCRIPT_DIR}/mvp_report_v6.py" \ + --results-dir "${OUT_BASE}" \ + --num-producers "${N_PRODUCERS}" \ + --per-agent-cardinality "${PER_AGENT_CARDINALITY}" \ + --out "${OUT_BASE}/MVP_REPORT_v6.md" \ + > "${OUT_BASE}/report.log" 2>&1 || \ + log " [warn] mvp_report_v6.py exited non-zero — see report.log" +} + +# ── main ───────────────────────────────────────────────────────── +main() { + ensure_out_dirs + preflight + bring_up_stack + capture_emitted_configs + measure_phase + freshness_phase + ad_hoc_postings_phase + cold_fallback_phase + compactor_phase + fetch_s3_cost_csv + teardown + generate_report + + log "MVP demo v6 complete. Report: ${OUT_BASE}/MVP_REPORT_v6.md" +} + +main "$@" From 33bcc2d75d90bfc940dd1af48fdae8ac18192d97 Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Wed, 6 May 2026 17:31:38 -0400 Subject: [PATCH 2/4] mvp v6 phase E: add per-edge bandwidth measurement script MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extends measure_stages.py's docker stats sampling with per-edge labelling for the v6 fan-in topology: edge_sdk_to_agent (10 producers TX) edge_agent_to_gateway (2 agents TX) edge_gateway_to_backend (backend RX — distinguishes the gateway's split egress) edge_gateway_to_s3 (minio RX — same rationale) Output schema: edge,sample_ts_ms,window_s,bytes_total,bytes_per_s. One row per (edge, sample) for a per-edge time-series; mvp_report_v6.py reduces to per-edge mean B/s for criterion ① bandwidth. Stdlib only; same `docker stats --no-stream` mechanism as measure_stages.py so two probes can run in parallel without docker-daemon contention. Co-Authored-By: Claude Opus 4.7 (1M context) --- deploy/scripts/measure_per_edge_bandwidth.py | 332 +++++++++++++++++++ 1 file changed, 332 insertions(+) create mode 100755 deploy/scripts/measure_per_edge_bandwidth.py diff --git a/deploy/scripts/measure_per_edge_bandwidth.py b/deploy/scripts/measure_per_edge_bandwidth.py new file mode 100755 index 00000000..14b11ffb --- /dev/null +++ b/deploy/scripts/measure_per_edge_bandwidth.py @@ -0,0 +1,332 @@ +#!/usr/bin/env python3 +"""measure_per_edge_bandwidth.py — MVP v6 per-edge bandwidth probe. + +Extends `measure_stages.py`'s `docker stats` net-rx/tx capture with +per-edge labelling for the v6 multi-stage topology: + + sdk → agent (×10 producers → 2 agents) + agent → gateway (2 agents → 1 gateway) + gateway → backend (1 gateway → 1 backend) + gateway → s3 (Gorilla archive PUTs / MinIO traffic) + +Why this lives in its own script (vs. a flag on measure_stages.py): + + measure_stages.py emits ONE row per (baseline, stage, container) + summarising the whole window. The v6 report wants per-edge + bytes/sec time-series so the headline can show "per-edge + bandwidth", not just per-stage. Two outputs, two scripts. + +Per-edge bytes/sec is computed as the sum of relevant containers' +NetIO TX (or MinIO's RX for the gateway-→s3 edge) deltas across each +1Hz sample, divided by the wall-clock interval. + +Hard fact: docker stats reports CUMULATIVE rx + tx since container +start, so a delta over the window measures wire-bytes during the +window (modulo TCP retransmits etc., which we accept as noise). + +Output CSV columns: + + edge,sample_ts_ms,window_s,bytes_total,bytes_per_s + +`edge` is one of: + edge_sdk_to_agent + edge_agent_to_gateway + edge_gateway_to_backend + edge_gateway_to_s3 + +One row per (edge, sample) — i.e. ~60 rows per edge for a 60s run +at 1Hz. Reduce to per-edge averages downstream (mvp_report_v6.py +handles this). + +Stdlib only. Calls `docker stats --no-stream` once per sample — same +mechanism measure_stages.py uses, so two probes can run in parallel +without a docker rate-limit issue (the daemon is happy to serve +multiple `docker stats --no-stream` calls per second). + +Usage: + + python3 measure_per_edge_bandwidth.py \\ + --duration 60 \\ + --period 1.0 \\ + --out per_edge_bandwidth.csv +""" +from __future__ import annotations + +import argparse +import csv +import re +import subprocess +import sys +import time +from typing import Iterable + + +# Edge container-set definitions for the v6 topology. The values +# are role tags consumed by `_role_for_container` below. +# +# Trade-off: keeping the edge → container mapping in this file +# rather than parsing a YAML keeps the script stdlib-only. The +# mapping mirrors mvp-v6-multi-stage.yml; if that overlay's +# service names change, this script must follow. + +# Container role classifier. We match on the bare service name +# stripped of compose's `--` suffix. For +# producer-{a,b}-N we collapse to "producer". +def _bare_service(name: str) -> str: + bare = name + bare = re.sub(r"-\d+$", "", bare) # strip replica index + bare = re.sub(r"^docker-compose-", "", bare) # strip project prefix + bare = re.sub(r"^docker_compose_", "", bare) + bare = re.sub(r"^deploy-docker-compose-", "", bare) + return bare + + +def _role_for_container(name: str) -> str: + """One of {producer, agent, gateway, backend, minio, other}.""" + bare = _bare_service(name) + if bare.startswith("producer-"): + return "producer" + if bare.startswith("agent-"): + return "agent" + if bare == "gateway": + return "gateway" + if bare == "backend": + return "backend" + if bare == "minio": + return "minio" + if bare.startswith("fake-exporter") or bare.startswith("fake_exporter"): + # base.yml's fake-exporter; under v6 overlay it's a stub + # alpine that does nothing, so its bytes-on-wire is ~0. Tag + # it as producer so a misconfiguration (overlay not active) + # still attributes to the right edge — better than silently + # under-counting. + return "producer" + return "other" + + +# ── docker stats helpers (subset of measure_stages.py) ──────────── + + +def _parse_size_to_bytes(s: str) -> float: + s = s.strip() + if not s: + return float("nan") + # docker stats can report e.g. "1.23GB", "456MB", "12.3kB", "789B". + for suffix, factor in ( + ("GB", 1e9), ("MB", 1e6), ("kB", 1e3), ("B", 1.0), + ): + if s.endswith(suffix): + try: + return float(s[: -len(suffix)]) * factor + except ValueError: + return float("nan") + try: + return float(s) + except ValueError: + return float("nan") + + +def docker_stats_snapshot() -> dict[str, tuple[float, float]]: + """Returns {container_name: (cumulative_rx_bytes, cumulative_tx_bytes)}.""" + try: + out = subprocess.check_output( + [ + "docker", "stats", "--no-stream", + "--format", "{{.Name}}|{{.NetIO}}", + ], + text=True, + timeout=10, + ) + except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e: + print(f"# docker stats failed: {e}", file=sys.stderr) + return {} + out_map: dict[str, tuple[float, float]] = {} + for line in out.strip().splitlines(): + try: + name, netio = line.split("|", 1) + except ValueError: + continue + if "/" not in netio: + continue + rx_str, tx_str = (p.strip() for p in netio.split("/", 1)) + rx = _parse_size_to_bytes(rx_str) + tx = _parse_size_to_bytes(tx_str) + if rx == rx and tx == tx: # NaN guard + out_map[name] = (rx, tx) + return out_map + + +# ── per-edge classifier ─────────────────────────────────────────── + +# Per-edge attribution rules. Each edge is a `(metric, role)` +# tuple — we use that container's TX (or RX) bytes as the edge's +# wire bytes during the window. +# +# Why we attribute to one side of the wire only: +# - Producers have one outbound stream → agent. Sum their TX. +# - Agents fan out to gateway. Sum their TX (this overcounts +# by including the agent's own /metrics scrape from +# Prometheus, but that's negligible vs. OTLP traffic in the +# v6 topology). +# - Gateway fans out to backend AND s3. We can't distinguish +# the two from a single-NIC TX counter — so: +# edge_gateway_to_backend = backend.RX +# edge_gateway_to_s3 = minio.RX +# This gets us per-destination accounting at the cost of +# trusting the receiver's RX counter. +# +# The fan-in vs. fan-out asymmetry is intentional: we want the +# tightest counter per edge. For the producer→agent edge we trust +# producer.TX (a single source per row). For the gateway split we +# trust the destination's RX so the two sub-edges are +# distinguishable. + +EDGES: list[tuple[str, str]] = [ + # (edge_label, classifier_token) + ("edge_sdk_to_agent", "PRODUCER_TX"), + ("edge_agent_to_gateway", "AGENT_TX"), + ("edge_gateway_to_backend", "BACKEND_RX"), + ("edge_gateway_to_s3", "MINIO_RX"), +] + + +def _bytes_for_edge( + snap: dict[str, tuple[float, float]], edge_token: str +) -> float: + """Sum the relevant cumulative counter across all containers + matching this edge's classifier token.""" + total = 0.0 + found = False + for name, (rx, tx) in snap.items(): + role = _role_for_container(name) + if edge_token == "PRODUCER_TX" and role == "producer": + total += tx + found = True + elif edge_token == "AGENT_TX" and role == "agent": + total += tx + found = True + elif edge_token == "BACKEND_RX" and role == "backend": + total += rx + found = True + elif edge_token == "MINIO_RX" and role == "minio": + total += rx + found = True + return total if found else float("nan") + + +# ── sampling loop ───────────────────────────────────────────────── + + +def sample_edges( + duration_s: float, period_s: float = 1.0 +) -> list[tuple[str, int, float, float, float]]: + """Sample per-edge cumulative bytes at `period_s` Hz for + `duration_s`. Returns a list of (edge, sample_ts_ms, window_s, + bytes_total, bytes_per_s) rows. + + The first sample is the BASELINE — its bytes_total/bytes_per_s + are NaN (we have no prior sample to diff against). Subsequent + samples diff against the immediately previous snapshot. The + final row per edge is also a "tail" diff that covers + [t_n-1 → t_n], same as every row in between. + """ + n_samples = max(2, int(duration_s / period_s) + 1) + rows: list[tuple[str, int, float, float, float]] = [] + prev_snap: dict[str, tuple[float, float]] | None = None + prev_t: float | None = None + + for i in range(n_samples): + t = time.monotonic() + snap = docker_stats_snapshot() + wall_ms = int(time.time() * 1000) + if prev_snap is None: + # Baseline row per edge. + for label, _ in EDGES: + rows.append((label, wall_ms, 0.0, float("nan"), float("nan"))) + else: + window = max(t - (prev_t or t), 1e-6) + for label, token in EDGES: + cur = _bytes_for_edge(snap, token) + prev = _bytes_for_edge(prev_snap, token) + if cur == cur and prev == prev: # both non-NaN + bytes_delta = max(0.0, cur - prev) + bytes_per_s = bytes_delta / window + else: + bytes_delta = float("nan") + bytes_per_s = float("nan") + rows.append((label, wall_ms, window, bytes_delta, bytes_per_s)) + prev_snap = snap + prev_t = t + if i < n_samples - 1: + time.sleep(max(0.0, period_s - (time.monotonic() - t))) + return rows + + +# ── main ────────────────────────────────────────────────────────── + + +def main(argv: Iterable[str] | None = None) -> int: + p = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + p.add_argument( + "--duration", + type=float, + default=60.0, + help="Total sampling window in seconds (default 60).", + ) + p.add_argument( + "--period", + type=float, + default=1.0, + help="Sample period in seconds (default 1.0).", + ) + p.add_argument( + "--out", + required=True, + help="Output CSV path. Columns: edge,sample_ts_ms,window_s," + "bytes_total,bytes_per_s.", + ) + args = p.parse_args(list(argv) if argv is not None else None) + + print( + f"# per-edge bandwidth probe — duration={args.duration}s " + f"period={args.period}s out={args.out}", + file=sys.stderr, + ) + + rows = sample_edges(args.duration, args.period) + if not rows: + print("# no docker stats samples — aborting", file=sys.stderr) + return 1 + + with open(args.out, "w", newline="") as f: + w = csv.writer(f) + w.writerow(("edge", "sample_ts_ms", "window_s", "bytes_total", "bytes_per_s")) + for r in rows: + label, ts, window, bt, bps = r + w.writerow( + ( + label, + ts, + f"{window:.3f}", + f"{bt:.1f}" if bt == bt else "", + f"{bps:.3f}" if bps == bps else "", + ) + ) + print(f"# wrote {args.out} ({len(rows)} rows)") + + # Per-edge summary: count of valid samples + mean bytes/s. + by_edge: dict[str, list[float]] = {} + for label, _, _, _, bps in rows: + if bps == bps: # not NaN + by_edge.setdefault(label, []).append(bps) + for label in sorted(by_edge): + vals = by_edge[label] + mean = sum(vals) / len(vals) if vals else float("nan") + print( + f"# edge={label} samples={len(vals)} mean_bytes_per_s={mean:.1f}" + ) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) From bb3ed27f21396ab120a0909cc3972cb4166a8145 Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Wed, 6 May 2026 17:31:48 -0400 Subject: [PATCH 3/4] mvp v6 phase E: add MVP_REPORT_v6.md generator MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Independent from v5's mvp_report.py (which is in active use, PID 1945011). Eight sections: §1 Stage-separated resource table (per-stage TOTAL across containers) §2 Six-criterion verdict table (bandwidth/latency/resource/accuracy/ cold-fallback/freshness) §3 Per-query-class breakdown (window/label/combined latency + accuracy) §4 Postings filtering effect (gated on v5 merge — renders v5-merge-pending markers if response fields absent) §5 Compaction effect (before/after MinIO object count + bytes) §6 S3-ops cost (gated on v5 cost tracker endpoint) §7 Honest caveats (no replan, no hot reconfig, 10K not 1M, single host) §8 Controller-emitted runtime configs status (live / fallback-placeholder / not-exercised) Pure stdlib, idempotent. Reads the v6 results dir layout written by run_mvp_demo_v6.sh. Co-Authored-By: Claude Opus 4.7 (1M context) --- deploy/scripts/mvp_report_v6.py | 943 ++++++++++++++++++++++++++++++++ 1 file changed, 943 insertions(+) create mode 100755 deploy/scripts/mvp_report_v6.py diff --git a/deploy/scripts/mvp_report_v6.py b/deploy/scripts/mvp_report_v6.py new file mode 100755 index 00000000..12ef641b --- /dev/null +++ b/deploy/scripts/mvp_report_v6.py @@ -0,0 +1,943 @@ +#!/usr/bin/env python3 +"""mvp_report_v6.py — MVP v6 report generator. + +Independent from `mvp_report.py` (v5). v6 differs in: + + * **Single run, multi-stage topology.** v5 cycled four baselines + side-by-side; v6 captures one controller-driven cell. The B0 + baseline number for criterion ① is read out of the same run's + Prometheus B0 snapshot when the b0 profile was active, otherwise + the cell shows "—" and the criterion gets verdict UNKNOWN. + + * **Six criteria with per-class breakdown.** §2's verdict rows + cover bandwidth / latency / combined-resource / accuracy / + cold-fallback / freshness. §3 breaks each query class out + separately so the reader sees which sketch+stage the controller + chose for window / label / combined. v5 had four criteria. + + * **Stage table is per-stage TOTAL only** — no v5-style backend + {ingest,query} double-row trick. v6 wants the simpler + representation: agent / gateway / backend-ingest / + backend-storage / backend-query, one row each. + + * **Per-edge bandwidth from `per_edge_bandwidth.csv`.** §1's + bandwidth column reads the per-edge probe's mean bytes/s rather + than docker stats' all-container-rolled-up netio. + + * **Postings + compaction + S3-cost sections gated on v5 merge.** + If v5 hasn't merged the postings_filtered_series_count fields or + the s3_cost.csv endpoint, those sections render with a + "v5-merge-pending" marker rather than missing data. + +Pure stdlib. Idempotent — re-running over the same CSVs reproduces +the same MD. + +Input layout (v6): + + / + controller-emitted-configs/{STATUS,agent.bootstrap.yaml,...} + measurements/{stages.csv, per_edge_bandwidth.csv, + replay.jsonl, accuracy.csv, s3_cost.csv} + freshness/{raw.csv, warm.csv, archive.csv} + ad-hoc/{count_api_series.json, topk_5xx_by_zone.json, + cold_payments.json, cold_payments.verdict, ...} + compactor/{dry_run.json, live_run.json, + before.minio.jsonl, after.minio.jsonl, SKIPPED?} + +Output: + + /MVP_REPORT_v6.md +""" +from __future__ import annotations + +import argparse +import csv +import json +import os +import statistics +import sys +from typing import Any + + +STAGE_ORDER = [ + "agent", + "gateway", + "backend-ingest", + "backend-storage", + "backend-query", +] + +QUERY_CLASSES = [ + # (label, kind tag in replay-queries.json, promql snippet match) + ("window-per-series", "quantile", "quantile_over_time"), + ("label-at-instant", "sum", "sum by (zone) (http_requests_total)"), + ("combined-window-label", "sum", "sum by (zone) (rate(http_requests_total"), +] + +EDGE_ORDER = [ + "edge_sdk_to_agent", + "edge_agent_to_gateway", + "edge_gateway_to_backend", + "edge_gateway_to_s3", +] + + +# ── tiny helpers ───────────────────────────────────────────────── + + +def _is_nan(x: float) -> bool: + return x != x # noqa: PLR0124 + + +def _percentile(xs: list[float], p: float) -> float: + if not xs: + return float("nan") + s = sorted(xs) + idx = max(0, min(len(s) - 1, int(round(p * (len(s) - 1))))) + return s[idx] + + +def _read_csv(path: str) -> list[dict[str, str]]: + if not os.path.exists(path): + return [] + with open(path, "r") as f: + return list(csv.DictReader(f)) + + +def _read_json(path: str) -> dict | None: + if not os.path.exists(path): + return None + try: + with open(path, "r") as f: + return json.load(f) + except (json.JSONDecodeError, OSError): + return None + + +def _read_text(path: str) -> str: + if not os.path.exists(path): + return "" + try: + with open(path, "r") as f: + return f.read() + except OSError: + return "" + + +def _read_jsonl(path: str) -> list[dict]: + out: list[dict] = [] + if not os.path.exists(path): + return out + with open(path, "r") as f: + for line in f: + line = line.strip() + if not line: + continue + try: + out.append(json.loads(line)) + except json.JSONDecodeError: + continue + return out + + +# ── §1 stage-separated resource table ──────────────────────────── + + +def _aggregate_stages(rows: list[dict]) -> dict[str, dict[str, float]]: + """Sum per-stage totals across all containers in that stage. + + v6's table is single-baseline; we sum to one row per stage. + """ + out: dict[str, dict[str, float]] = {} + for r in rows: + stage = r.get("stage", "") + if stage not in STAGE_ORDER: + continue + agg = out.setdefault(stage, { + "cpu_cores": 0.0, "rss_mib": 0.0, + "net_in_kibps": 0.0, "net_out_kibps": 0.0, + "disk_mib": 0.0, + }) + for k in ("cpu_cores", "rss_mib", "net_in_kibps", "net_out_kibps", "disk_mib"): + try: + v = float(r.get(k, "") or "nan") + except ValueError: + v = float("nan") + if not _is_nan(v): + agg[k] += v + return out + + +def render_section_1_stage_table(stages_rows: list[dict]) -> list[str]: + md: list[str] = [] + md.append("## §1 Stage-separated resource table") + md.append("") + md.append( + "Per-stage TOTAL across all containers in that stage. The " + "v6 multi-stage topology has 10 producer / 2 agent / 1 " + "gateway / 1 backend / 1 storage (MinIO) containers; the " + "rows below reduce all relevant containers per stage to a " + "single number. CPU is mean cores over the 60s window, RSS " + "is mean MiB, net is window-rate KiB/s (from `docker stats` " + "NetIO deltas), disk is end-of-window MiB." + ) + md.append("") + + aggs = _aggregate_stages(stages_rows) + if not aggs: + md.append("_stages.csv missing — no rows to render_") + md.append("") + return md + + md.append("| Stage | CPU (cores) | RSS (MiB) | Net in (KiB/s) | Net out (KiB/s) | Disk (MiB) |") + md.append("|---|---|---|---|---|---|") + for stage in STAGE_ORDER: + d = aggs.get(stage, {}) + + def fmt(key: str, fmt_str: str) -> str: + v = d.get(key, float("nan")) + return fmt_str.format(v) if not _is_nan(v) else "—" + + md.append( + "| {stage} | {cpu} | {rss} | {nin} | {nout} | {disk} |".format( + stage=stage, + cpu=fmt("cpu_cores", "{:.3f}"), + rss=fmt("rss_mib", "{:.1f}"), + nin=fmt("net_in_kibps", "{:.1f}"), + nout=fmt("net_out_kibps", "{:.1f}"), + disk=fmt("disk_mib", "{:.1f}"), + ) + ) + md.append("") + return md + + +# ── §2 per-criterion verdict (6 criteria) ──────────────────────── + + +def _per_edge_mean_bytes_per_s(edge_rows: list[dict]) -> dict[str, float]: + """Mean bytes/s per edge label.""" + by_edge: dict[str, list[float]] = {} + for r in edge_rows: + try: + bps = float(r.get("bytes_per_s", "") or "nan") + except ValueError: + continue + if _is_nan(bps): + continue + by_edge.setdefault(r.get("edge", "?"), []).append(bps) + return {e: (sum(v) / len(v) if v else float("nan")) for e, v in by_edge.items()} + + +def criterion_bandwidth(edge_rows: list[dict]) -> tuple[str, str, dict]: + means = _per_edge_mean_bytes_per_s(edge_rows) + sdk_to_agent = means.get("edge_sdk_to_agent", float("nan")) + agent_to_gw = means.get("edge_agent_to_gateway", float("nan")) + gw_to_backend = means.get("edge_gateway_to_backend", float("nan")) + gw_to_s3 = means.get("edge_gateway_to_s3", float("nan")) + if _is_nan(sdk_to_agent) and _is_nan(agent_to_gw): + return "UNKNOWN", "per_edge_bandwidth.csv missing or empty", {} + + # Per-edge B/s, one decimal place. We don't compute a B0 + # comparison here (single-cell run) — that's the §1 stage table + # job. Verdict is PASS if the egress edge (agent→gateway) is + # less than the ingress edge (sdk→agent), which is the + # ASAP value-prop marker (sketches reduce volume on egress). + if not _is_nan(sdk_to_agent) and not _is_nan(agent_to_gw): + verdict = "PASS" if agent_to_gw < sdk_to_agent else "FAIL" + else: + verdict = "UNKNOWN" + + line = ( + f"sdk→agent={sdk_to_agent:.1f} B/s, " + f"agent→gateway={agent_to_gw:.1f} B/s, " + f"gateway→backend={gw_to_backend:.1f} B/s, " + f"gateway→s3={gw_to_s3:.1f} B/s" + ) + return verdict, line, means + + +def _classify_query_class(replay_row: dict) -> str | None: + """Return one of the QUERY_CLASSES labels for a replay JSONL + row, or None if it doesn't match.""" + promql = replay_row.get("query", "") + for label, _kind, snippet in QUERY_CLASSES: + if snippet in promql: + return label + return None + + +def _per_class_latency_p50_p99(replay_rows: list[dict]) -> dict[str, dict[str, float]]: + by_class: dict[str, list[float]] = {} + for r in replay_rows: + cls = _classify_query_class(r) + if cls is None: + continue + try: + d = float(r.get("duration_ms")) + except (TypeError, ValueError): + continue + if _is_nan(d): + continue + by_class.setdefault(cls, []).append(d) + out: dict[str, dict[str, float]] = {} + for cls, vals in by_class.items(): + out[cls] = { + "p50": _percentile(vals, 0.5), + "p99": _percentile(vals, 0.99), + "n": float(len(vals)), + } + return out + + +def criterion_latency(replay_rows: list[dict]) -> tuple[str, str, dict]: + if not replay_rows: + return "UNKNOWN", "replay.jsonl missing or empty", {} + by_class = _per_class_latency_p50_p99(replay_rows) + if not by_class: + return "UNKNOWN", "no replay rows matched a v6 query class", {} + parts = [] + worst_p99 = 0.0 + for cls, _, _ in QUERY_CLASSES: + d = by_class.get(cls) + if d is None: + parts.append(f"{cls}: n=0") + continue + parts.append( + f"{cls}: p50={d['p50']:.1f}ms p99={d['p99']:.1f}ms n={int(d['n'])}" + ) + if d["p99"] > worst_p99: + worst_p99 = d["p99"] + # Spec doesn't pin a numeric SLA, but a 5s p99 is a reasonable + # "something's broken" gate — the ASAP backend should be much + # faster than that on a 60s soak. + verdict = "PASS" if worst_p99 < 5_000.0 and worst_p99 > 0 else ( + "UNKNOWN" if worst_p99 == 0 else "FAIL" + ) + return verdict, "; ".join(parts), by_class + + +def criterion_combined_resource(stages_rows: list[dict]) -> tuple[str, str, dict]: + """Total CPU cores + total RSS MiB summed across stages. + + Single-cell run: no B0 comparison number to take a percentage + against. Verdict is UNKNOWN unless the cell included a B0 + Prometheus snapshot (recorded in stages.csv as a row with + stage=backend-storage-b0). + """ + aggs = _aggregate_stages(stages_rows) + if not aggs: + return "UNKNOWN", "stages.csv missing or empty", {} + total_cpu = sum(d.get("cpu_cores", 0.0) for d in aggs.values()) + total_rss = sum(d.get("rss_mib", 0.0) for d in aggs.values()) + line = ( + f"total cpu_cores={total_cpu:.3f}, total rss_mib={total_rss:.1f} " + f"(stages: {', '.join(sorted(aggs.keys()))})" + ) + # No B0 comparison in single-cell mode — verdict is "captured". + return "CAPTURED", line, {"cpu": total_cpu, "rss": total_rss} + + +def criterion_accuracy(accuracy_rows: list[dict]) -> tuple[str, str, dict]: + by_kind: dict[str, list[float]] = {} + for row in accuracy_rows: + k = row.get("kind", "") + try: + err = float(row.get("error", "") or "nan") + except ValueError: + err = float("nan") + if not _is_nan(err): + by_kind.setdefault(k, []).append(err) + if not by_kind: + return "UNKNOWN", "accuracy.csv empty (warm tier may not have flushed)", {} + parts: list[str] = [] + medians: dict[str, float] = {} + for k in sorted(by_kind): + med = statistics.median(by_kind[k]) + medians[k] = med + parts.append(f"{k}: median rel-err={med:.4f} (n={len(by_kind[k])})") + worst = max(medians.values()) if medians else float("inf") + verdict = "PASS" if worst <= 0.05 else "FAIL" + return verdict, "; ".join(parts), medians + + +def criterion_cold_fallback(adhoc_dir: str) -> tuple[str, str, dict]: + response_path = os.path.join(adhoc_dir, "cold_payments.json") + verdict_path = os.path.join(adhoc_dir, "cold_payments.verdict") + curl_path = os.path.join(adhoc_dir, "cold_payments.curlstats") + body = _read_json(response_path) + raw_text = _read_text(response_path) + verdict_marker = _read_text(verdict_path).strip() + curlstats = _read_text(curl_path).strip() + + if body is None and not raw_text: + return "UNKNOWN", "cold_payments.json missing", {} + + has_marker = "gorilla_archive" in raw_text + if has_marker: + verdict = "PASS" + line = ( + "`data_source: gorilla_archive` present in response — " + "GorillaQueryEngine served the ad-hoc query." + ) + elif verdict_marker == "PASS": + # Driver thought it was OK but the marker grep missed — + # surface as PARTIAL. + verdict = "PARTIAL" + line = "driver wrote PASS but no `gorilla_archive` marker found in body" + else: + verdict = "FAIL" + line = "no `gorilla_archive` marker in cold_payments.json" + if curlstats: + line += f" · curl: {curlstats}" + return verdict, line, {"has_marker": has_marker, "marker": verdict_marker} + + +def criterion_freshness(fresh_dir: str) -> tuple[str, str, dict]: + summary: dict[str, dict[str, float]] = {} + for path in ("raw", "warm", "archive"): + rows = _read_csv(os.path.join(fresh_dir, f"{path}.csv")) + deltas: list[float] = [] + for r in rows: + try: + d = float(r.get("delta_ms", "") or "nan") + except ValueError: + continue + if not _is_nan(d): + deltas.append(d) + if deltas: + summary[path] = { + "p50": _percentile(deltas, 0.5), + "p99": _percentile(deltas, 0.99), + "n": float(len(deltas)), + } + else: + summary[path] = {"p50": float("nan"), "p99": float("nan"), "n": 0.0} + + if all(_is_nan(s["p50"]) for s in summary.values()): + return "UNKNOWN", "no freshness samples on any path", summary + + parts = [] + for path in ("raw", "warm", "archive"): + s = summary[path] + if _is_nan(s["p50"]): + parts.append(f"{path}: no observations") + else: + parts.append( + f"{path}: p50={s['p50']:.0f}ms p99={s['p99']:.0f}ms n={int(s['n'])}" + ) + warm_p50 = summary.get("warm", {}).get("p50", float("nan")) + archive_p50 = summary.get("archive", {}).get("p50", float("nan")) + # Same gates as v4: warm ≤ 30s, archive ≤ 90s, in milliseconds. + if _is_nan(warm_p50): + verdict = "UNKNOWN" + elif warm_p50 <= 30_000.0 and ( + _is_nan(archive_p50) or archive_p50 <= 90_000.0 + ): + verdict = "PASS" + else: + verdict = "FAIL" + return verdict, "; ".join(parts), summary + + +# ── §3 per-query-class breakdown ──────────────────────────────── + + +def render_section_3_per_class( + replay_rows: list[dict], + accuracy_rows: list[dict], + emitted_status: str, +) -> list[str]: + md: list[str] = [] + md.append("## §3 Per-query-class breakdown") + md.append("") + md.append( + "Three canonical query classes from " + "`deploy/configs/mvp-v6-workload.yaml`. Sketch + stage " + "assignments come from the controller-emitted configs (see " + "§9 below); latency from `replay.jsonl`; accuracy from " + "`accuracy.csv`." + ) + md.append("") + md.append( + "| Class | Sketch / stage (controller plan) | p50 (ms) | p99 (ms) | " + "median rel-err | n |" + ) + md.append("|---|---|---|---|---|---|") + + by_class = _per_class_latency_p50_p99(replay_rows) + # Accuracy by kind (we don't have a class-level join in + # accuracy.csv; we map kind→class through the + # QUERY_CLASSES table). + acc_by_kind: dict[str, list[float]] = {} + for r in accuracy_rows: + k = r.get("kind", "") + try: + err = float(r.get("error", "") or "nan") + except ValueError: + err = float("nan") + if not _is_nan(err): + acc_by_kind.setdefault(k, []).append(err) + + # Pre-canned plan annotation per class. The actual planner + # output is captured in the controller-emitted configs; this + # column shows the EXPECTED plan from + # mvp-v6-workload.yaml::assign_to_role. + plan_annotation = { + "window-per-series": "DDSketch / agent", + "label-at-instant": "identity / gateway (sum-by-zone fan-in)", + "combined-window-label": "rate@agent + sum-by-zone@gateway", + } + + for cls, kind, _ in QUERY_CLASSES: + d = by_class.get(cls, {}) + p50 = d.get("p50", float("nan")) + p99 = d.get("p99", float("nan")) + n = int(d.get("n", 0.0)) + errs = acc_by_kind.get(kind, []) + med_err = statistics.median(errs) if errs else float("nan") + + def fmt(v: float, fmt_str: str) -> str: + return fmt_str.format(v) if not _is_nan(v) else "—" + + plan_note = plan_annotation.get(cls, "—") + if emitted_status != "live": + plan_note += " *(plan from workload spec; emitter " + emitted_status + ")*" + + md.append( + "| {cls} | {plan} | {p50} | {p99} | {err} | {n} |".format( + cls=cls, + plan=plan_note, + p50=fmt(p50, "{:.1f}"), + p99=fmt(p99, "{:.1f}"), + err=fmt(med_err, "{:.4f}"), + n=n, + ) + ) + md.append("") + return md + + +# ── §4 postings filtering effect ───────────────────────────────── + + +def _extract_int_from_response(body: dict | None, key: str) -> int | None: + """Look for the v5 postings field in the response infos / data.""" + if body is None: + return None + data = body.get("data") or {} + infos = data.get("infos") or body.get("infos") or [] + if isinstance(infos, list): + for s in infos: + s = str(s) + i = s.find(key) + if i >= 0: + tail = s[i + len(key):].strip(": ").strip() + # Take the first integer-looking token. + token = "" + for ch in tail: + if ch.isdigit(): + token += ch + else: + break + if token: + try: + return int(token) + except ValueError: + return None + return None + + +def render_section_4_postings(adhoc_dir: str) -> list[str]: + md: list[str] = [] + md.append("## §4 Postings filtering effect") + md.append("") + + queries = [ + ("count_api_series", 'count(http_requests_total{service="api"})'), + ("topk_5xx_by_zone", 'topk(5, sum by (zone) (rate(http_requests_total{status=~"5.."}[5m])))'), + ] + md.append( + "Two ad-hoc queries with label predicates. Per the v5 " + "postings index: " + "`postings_filtered_series_count` is the count of series " + "that survived the predicate after sidecar lookup; the " + "would-have-scanned column is the same metric WITHOUT the " + "predicate (a coarse upper bound)." + ) + md.append("") + md.append("| Query | Series matched | Would have scanned | Status |") + md.append("|---|---|---|---|") + any_data = False + for label, promql in queries: + body = _read_json(os.path.join(adhoc_dir, f"{label}.json")) + matched = _extract_int_from_response(body, "postings_filtered_series_count") + would = _extract_int_from_response(body, "series_scanned_total") + if matched is None and would is None: + md.append( + f"| `{promql}` | — | — | v5-merge-pending (no postings field in response) |" + ) + else: + any_data = True + md.append( + f"| `{promql}` | {matched if matched is not None else '—'} | " + f"{would if would is not None else '—'} | OK |" + ) + md.append("") + if not any_data: + md.append( + "_v5 postings field not present on responses; rerun once " + "`#295 ASAPCollector` and `#90 ASAPQuery-backend` land._" + ) + md.append("") + return md + + +# ── §5 compaction effect ───────────────────────────────────────── + + +def _count_minio_objects(jsonl_path: str) -> tuple[int, int]: + """Return (object_count, total_bytes). Each line is one + `mc ls --recursive --json` record.""" + if not os.path.exists(jsonl_path): + return (0, 0) + count = 0 + total_bytes = 0 + with open(jsonl_path, "r") as f: + for line in f: + line = line.strip() + if not line: + continue + try: + rec = json.loads(line) + except json.JSONDecodeError: + continue + # `mc ls --json` rows have status=success, type=file, size=. + if rec.get("type") in ("file", "FILE"): + count += 1 + try: + total_bytes += int(rec.get("size", 0)) + except (TypeError, ValueError): + pass + return (count, total_bytes) + + +def render_section_5_compaction(compactor_dir: str) -> list[str]: + md: list[str] = [] + md.append("## §5 Compaction effect") + md.append("") + skipped = os.path.exists(os.path.join(compactor_dir, "SKIPPED")) + if skipped: + reason = _read_text(os.path.join(compactor_dir, "SKIPPED")).strip() + md.append(f"_Compactor SKIPPED: {reason}_") + md.append("") + return md + + before = _count_minio_objects(os.path.join(compactor_dir, "before.minio.jsonl")) + after = _count_minio_objects(os.path.join(compactor_dir, "after.minio.jsonl")) + md.append( + "Concat-only compactor (v5) byte-concatenates 6+ adjacent " + "blocks ≥6h old into one merged object. **No decode / " + "re-encode** — each source chunk remains an atomic Gorilla " + "chunk inside the merged file. The new manifest records " + "each chunk's `byte_offset` + `byte_length` so the backend " + "can issue `Range:` partial reads." + ) + md.append("") + md.append("| Stage | Object count | Total bytes |") + md.append("|---|---|---|") + md.append(f"| before | {before[0]} | {before[1]} |") + md.append(f"| after | {after[0]} | {after[1]} |") + md.append("") + + dry = _read_json(os.path.join(compactor_dir, "dry_run.json")) + if isinstance(dry, dict): + eligible = dry.get("eligible_blocks") or dry.get("blocks") or [] + if isinstance(eligible, list): + md.append(f"Dry-run plan: {len(eligible)} merge candidate(s).") + md.append("") + return md + + +# ── §6 cost (S3 ops) ───────────────────────────────────────────── + + +def render_section_6_cost(measurements_dir: str) -> list[str]: + md: list[str] = [] + md.append("## §6 S3-ops cost (measured)") + md.append("") + rows = _read_csv(os.path.join(measurements_dir, "s3_cost.csv")) + if not rows: + md.append( + "_v5 cost-tracker not present — `/internal/s3_cost.csv` " + "endpoint unavailable. Will populate once " + "`#90 ASAPQuery-backend` lands._" + ) + md.append("") + return md + md.append( + "Counts of PUT / GET / HEAD / DELETE issued against MinIO " + "during the cell's measurement window (the v5 backend's " + "internal cost tracker)." + ) + md.append("") + if rows: + # Render straight: take the first row, keys are the column names. + keys = sorted(rows[0].keys()) + md.append("| " + " | ".join(keys) + " |") + md.append("|" + "---|" * len(keys)) + for r in rows: + md.append("| " + " | ".join(str(r.get(k, "")) for k in keys) + " |") + md.append("") + return md + + +# ── §7 honest caveats ───────────────────────────────────────────── + + +def render_section_7_caveats() -> list[str]: + md: list[str] = [] + md.append("## §7 Honest caveats (non-goals)") + md.append("") + md.append( + "* **No dynamic replan.** The controller plans once at " + "startup off `mvp-v6-workload.yaml`. v6 does not exercise " + "the in-flight replan path — that's a follow-up." + ) + md.append( + "* **No OpAMP hot reconfig under churn.** OpAMP push happens " + "once per stage at boot; we don't kill an agent and verify " + "the controller re-pushes. v5's `ReplannerOpampGateway` " + "covers some of this; v6's typed-stage-split path doesn't." + ) + md.append( + "* **10K series, not 1M.** Per-agent cardinality 500, 10 " + "producers → 5K aggregate at the gateway. The 1M target " + "needs the cardinality-redesign work plus a multi-host " + "topology — out of scope for v6." + ) + md.append( + "* **Single host.** All containers share kernel scheduler " + "+ loopback NIC. Wire-bytes per-edge counters are still " + "meaningful (TX/RX is per-container) but absolute latencies " + "are loopback-flattered." + ) + md.append( + "* **B0 Prometheus reference is opt-in.** The v6 driver " + "does NOT bring up B0 in the same compose stack as the " + "ASAP backend (port collision on 19090). To get an A-vs-B " + "comparison row, run a separate B0 cycle and join the " + "stages.csv files manually." + ) + md.append( + "* **Postings + cost tracker gated on v5 merge.** Sections " + "§4 and §6 render with a `v5-merge-pending` marker until " + "PRs #295 (collector) and #90 (backend) land." + ) + md.append("") + return md + + +# ── §8 emitted-config status ───────────────────────────────────── + + +def _read_emitted_status(cdir: str) -> str: + status_path = os.path.join(cdir, "STATUS") + s = _read_text(status_path).strip() + return s or "unknown" + + +def render_section_8_emitted(cdir: str) -> list[str]: + md: list[str] = [] + md.append("## §8 Controller-emitted runtime configs") + md.append("") + status = _read_emitted_status(cdir) + md.append(f"Emitter status: **{status}**") + md.append("") + if status == "live": + md.append( + "Controller's typed-stage-split path produced per-stage " + "configs and pushed them via OpAMP / BackendClient. The " + "captured artifacts live under " + "`controller-emitted-configs/`." + ) + elif status == "fallback-placeholder": + md.append( + "Controller's typed-stage-split path returned None for " + "the v6 workload — the gateway and agents are running " + "the placeholder configs mounted by the compose overlay. " + "This is the v6 spec's fallback mode; criterion verdicts " + "in §2 are still meaningful." + ) + elif status == "not-exercised": + md.append( + "No typed-stage-split tracing in the controller logs — " + "either `USE_TYPED_STAGE_SPLIT` was unset (default 0) " + "or the workload didn't bind to a SketchExpr. Phase F " + "should set USE_TYPED_STAGE_SPLIT=1 in the compose env." + ) + else: + md.append( + "Emitter status unknown — controller-emitted-configs/STATUS " + "missing or empty. The driver may not have completed " + "Phase 1's capture step." + ) + md.append("") + # List captured artifacts. + if os.path.isdir(cdir): + files = sorted( + f for f in os.listdir(cdir) + if not f.startswith(".") and f != "STATUS" + ) + if files: + md.append("Captured artifacts:") + md.append("") + for f in files: + size = os.path.getsize(os.path.join(cdir, f)) + md.append(f"- `{f}` ({size} B)") + md.append("") + return md + + +# ── markdown assembly ──────────────────────────────────────────── + + +def render_markdown_v6( + results_dir: str, + num_producers: int, + per_agent_cardinality: int, +) -> str: + if not os.path.isdir(results_dir): + return f"# MVP report v6 — results dir missing ({results_dir})\n" + + measurements_dir = os.path.join(results_dir, "measurements") + fresh_dir = os.path.join(results_dir, "freshness") + adhoc_dir = os.path.join(results_dir, "ad-hoc") + compactor_dir = os.path.join(results_dir, "compactor") + emitted_dir = os.path.join(results_dir, "controller-emitted-configs") + + stages_rows = _read_csv(os.path.join(measurements_dir, "stages.csv")) + edge_rows = _read_csv(os.path.join(measurements_dir, "per_edge_bandwidth.csv")) + accuracy_rows = _read_csv(os.path.join(measurements_dir, "accuracy.csv")) + replay_rows = _read_jsonl(os.path.join(measurements_dir, "replay.jsonl")) + emitted_status = _read_emitted_status(emitted_dir) + + md: list[str] = [] + md.append("# ASAPCollector MVP demo — issue #46 (v6, controller-driven multi-stage)") + md.append("") + md.append( + "Single-cell controller-driven run: 10 producers → 2 agents " + "→ 1 gateway → 1 backend (+ MinIO archive). Controller plans " + "from `deploy/configs/mvp-v6-workload.yaml`; per-stage " + "configs are emitted via the typed-stage-split path " + "(`USE_TYPED_STAGE_SPLIT=1`). Six criteria + per-class " + "latency + per-edge bandwidth in this report." + ) + md.append("") + md.append("## Workload shape") + md.append("") + md.append("| Knob | v6 value |") + md.append("|------|---------|") + md.append(f"| Per-agent cardinality | **{per_agent_cardinality}** |") + md.append(f"| Number of producers | **{num_producers}** (×5 → agent-a, ×5 → agent-b) |") + md.append(f"| Aggregate series at gateway | **{num_producers * per_agent_cardinality}** |") + md.append(f"| Sketch family (default) | DDSketch (overridden per-metric by controller) |") + md.append(f"| Stack settle + warm-up + soak | 60s + 60s + 60s |") + md.append(f"| Replay shapes | window/label/combined @ 5 QPS for 60s |") + md.append("") + + md.extend(render_section_1_stage_table(stages_rows)) + + # §2 verdict table. + bw_v, bw_line, _ = criterion_bandwidth(edge_rows) + lat_v, lat_line, _ = criterion_latency(replay_rows) + res_v, res_line, _ = criterion_combined_resource(stages_rows) + acc_v, acc_line, _ = criterion_accuracy(accuracy_rows) + cold_v, cold_line, _ = criterion_cold_fallback(adhoc_dir) + fresh_v, fresh_line, _ = criterion_freshness(fresh_dir) + + md.append("## §2 Per-criterion verdict (6 criteria)") + md.append("") + md.append("| # | Criterion | Verdict | Detail |") + md.append("|---|---|---|---|") + md.append(f"| 1 | Bandwidth (per-edge B/s) | **{bw_v}** | {bw_line} |") + md.append(f"| 2 | Query latency (p50/p99 per class) | **{lat_v}** | {lat_line} |") + md.append(f"| 3 | Combined resource (sum of stages) | **{res_v}** | {res_line} |") + md.append(f"| 4 | Accuracy (rel-err per class) | **{acc_v}** | {acc_line} |") + md.append(f"| 5 | Cold-fallback (gorilla_archive marker) | **{cold_v}** | {cold_line} |") + md.append(f"| 6 | Freshness (p50/p99 per path) | **{fresh_v}** | {fresh_line} |") + md.append("") + + md.extend(render_section_3_per_class(replay_rows, accuracy_rows, emitted_status)) + md.extend(render_section_4_postings(adhoc_dir)) + md.extend(render_section_5_compaction(compactor_dir)) + md.extend(render_section_6_cost(measurements_dir)) + md.extend(render_section_7_caveats()) + md.extend(render_section_8_emitted(emitted_dir)) + + # Per-edge bandwidth appendix. + md.append("## Appendix A — per-edge bandwidth") + md.append("") + md.append("| Edge | Mean B/s | Samples |") + md.append("|---|---|---|") + means = _per_edge_mean_bytes_per_s(edge_rows) + counts: dict[str, int] = {} + for r in edge_rows: + try: + float(r.get("bytes_per_s", "") or "nan") + except ValueError: + continue + counts[r.get("edge", "?")] = counts.get(r.get("edge", "?"), 0) + 1 + for e in EDGE_ORDER: + m = means.get(e, float("nan")) + n = counts.get(e, 0) + m_str = f"{m:.1f}" if not _is_nan(m) else "—" + md.append(f"| {e} | {m_str} | {n} |") + md.append("") + + return "\n".join(md) + "\n" + + +# ── main ───────────────────────────────────────────────────────── + + +def main(argv: list[str] | None = None) -> int: + ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + ap.add_argument( + "--results-dir", + required=True, + help="v6 run output directory (the one passed as OUT_BASE to run_mvp_demo_v6.sh).", + ) + ap.add_argument( + "--num-producers", + type=int, + default=10, + help="Number of fake-exporter producers (default 10).", + ) + ap.add_argument( + "--per-agent-cardinality", + type=int, + default=500, + help="Per-producer cardinality (default 500).", + ) + ap.add_argument( + "--out", + required=True, + help="Output MD path. Caller usually points this at /MVP_REPORT_v6.md.", + ) + args = ap.parse_args(argv) + + md = render_markdown_v6( + args.results_dir, + num_producers=args.num_producers, + per_agent_cardinality=args.per_agent_cardinality, + ) + + with open(args.out, "w") as f: + f.write(md) + print(f"wrote {args.out} ({len(md)} chars)") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) From 8cb48235f0c0f39e5864e33d8efdf35c872cccd2 Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Wed, 6 May 2026 17:31:57 -0400 Subject: [PATCH 4/4] mvp v6 phase E: add unit tests for mvp_report_v6.py MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Hermetic — no compose stack, no docker. Synthesises a minimal v6 results directory layout (with mock CSVs / JSON / JSONL fixtures clearly marked synthetic) and asserts: * happy-path: all 8 sections present, all 6 criteria rows in §2, all 5 stages in §1, all 3 query classes in §3 * idempotency: re-running over the same CSVs produces byte-identical MD * sparse fixture (only stages.csv): renderer still produces complete MD with UNKNOWN verdicts, v5-merge-pending markers in §4 / §6 * fallback emitter status renders correctly in §8 * missing --results-dir produces self-explanatory MD rather than crashing * helper unit tests (per-edge mean, query-class classifier, minio-listing counter, postings field extractor) 9 tests, all pass. Co-Authored-By: Claude Opus 4.7 (1M context) --- deploy/scripts/tests/test_mvp_report_v6.py | 433 +++++++++++++++++++++ 1 file changed, 433 insertions(+) create mode 100644 deploy/scripts/tests/test_mvp_report_v6.py diff --git a/deploy/scripts/tests/test_mvp_report_v6.py b/deploy/scripts/tests/test_mvp_report_v6.py new file mode 100644 index 00000000..f13e6ec1 --- /dev/null +++ b/deploy/scripts/tests/test_mvp_report_v6.py @@ -0,0 +1,433 @@ +"""Unit tests for mvp_report_v6.py. + +Hermetic — no compose stack, no docker. We synthesise a minimal v6 +results directory layout and assert the generator produces the +expected MD shape (per-section presence, per-criterion verdict +formatting, idempotency). + +Synthesis policy: every numeric value in these fixtures is clearly +SYNTHETIC (small round numbers, "MOCK" markers in comments). Phase F +captures real numbers — these tests just exercise the renderer. +""" +from __future__ import annotations + +import csv +import importlib.util +import json +import os +import sys +from pathlib import Path + +import pytest + +# Load mvp_report_v6.py without requiring an installed package. +HERE = Path(__file__).resolve().parent +SCRIPT = HERE.parent / "mvp_report_v6.py" +spec = importlib.util.spec_from_file_location("mvp_report_v6", SCRIPT) +assert spec is not None and spec.loader is not None +mvp_report_v6 = importlib.util.module_from_spec(spec) +sys.modules["mvp_report_v6"] = mvp_report_v6 +spec.loader.exec_module(mvp_report_v6) + + +# ── fixture builder ────────────────────────────────────────────── + + +def _write_csv(path: Path, header: list[str], rows: list[list]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("w", newline="") as f: + w = csv.writer(f) + w.writerow(header) + for r in rows: + w.writerow(r) + + +def _write_jsonl(path: Path, rows: list[dict]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("w") as f: + for r in rows: + f.write(json.dumps(r) + "\n") + + +def _build_results_dir(root: Path, *, kind: str = "happy") -> Path: + """Create a v6 results dir with synthetic CSVs. + + `kind` selects which fixture variant: + - "happy" — all sections populated, all criteria PASS + - "sparse" — only stages.csv present; everything else missing + - "fallback"— emitted-config STATUS = fallback-placeholder + """ + root.mkdir(parents=True, exist_ok=True) + measurements = root / "measurements" + fresh = root / "freshness" + adhoc = root / "ad-hoc" + compactor = root / "compactor" + emitted = root / "controller-emitted-configs" + + for d in (measurements, fresh, adhoc, compactor, emitted): + d.mkdir(parents=True, exist_ok=True) + + if kind == "sparse": + # Only stages.csv. Verdicts that need other CSVs go UNKNOWN. + # MOCK numbers — single agent + gateway row, that's it. + _write_csv( + measurements / "stages.csv", + ["baseline", "stage", "container", + "cpu_cores", "rss_mib", + "net_in_kibps", "net_out_kibps", "disk_mib"], + [ + ["mvp-v6", "agent", "agent-a", "0.10", "100.0", "10.0", "5.0", ""], + ["mvp-v6", "agent", "agent-b", "0.10", "100.0", "10.0", "5.0", ""], + ["mvp-v6", "gateway", "gateway", "0.20", "200.0", "20.0", "15.0", ""], + ], + ) + return root + + # ── happy + fallback share the same data; only STATUS differs. + + # MOCK stages: one row per stage, round numbers. + _write_csv( + measurements / "stages.csv", + ["baseline", "stage", "container", + "cpu_cores", "rss_mib", + "net_in_kibps", "net_out_kibps", "disk_mib"], + [ + # 2 agents + ["mvp-v6", "agent", "agent-a", "0.10", "100.0", "10.0", "5.0", "0"], + ["mvp-v6", "agent", "agent-b", "0.10", "100.0", "10.0", "5.0", "0"], + ["mvp-v6", "gateway", "gateway", "0.20", "200.0", "20.0", "15.0", "0"], + ["mvp-v6", "backend-ingest", "backend", "0.30", "300.0", "30.0", "0.0", "0"], + ["mvp-v6", "backend-query", "backend", "0.10", "300.0", "0.0", "5.0", "0"], + ["mvp-v6", "backend-storage", "minio", "0.05", "50.0", "5.0", "5.0", "100.0"], + ], + ) + + # MOCK per-edge bandwidth: 3 samples per edge. + # Note: producer→agent should be larger than agent→gateway so + # the bandwidth verdict is PASS. Numbers below are bytes/sec + # (intentionally tiny — these are synthetic). + edge_rows: list[list] = [] + base_ts = 1_700_000_000_000 + edges = [ + ("edge_sdk_to_agent", 1000.0), + ("edge_agent_to_gateway", 500.0), + ("edge_gateway_to_backend", 400.0), + ("edge_gateway_to_s3", 100.0), + ] + for edge_label, bps in edges: + for i in range(3): + edge_rows.append([ + edge_label, base_ts + i * 1000, "1.000", + f"{bps:.1f}", f"{bps:.3f}", + ]) + _write_csv( + measurements / "per_edge_bandwidth.csv", + ["edge", "sample_ts_ms", "window_s", "bytes_total", "bytes_per_s"], + edge_rows, + ) + + # MOCK accuracy: one row per (kind, query) — all under 5%. + _write_csv( + measurements / "accuracy.csv", + ["cell", "kind", "query", "t", "duration_ms", "plan_id", + "truth", "answer", "error", "recall", "n_truth_samples"], + [ + ["mvp-v6", "quantile", "quantile_over_time(0.99, http_requests_total_latency_ms[1m])", + "0", "5.0", "p1", "100", "101", "0.01", "", "1000"], + ["mvp-v6", "sum", "sum by (zone) (http_requests_total)", + "0", "3.0", "p1", "5000", "5005", "0.001", "", "1000"], + ["mvp-v6", "sum", "sum by (zone) (rate(http_requests_total[5m]))", + "0", "4.0", "p1", "10", "10.05", "0.005", "", "1000"], + ], + ) + + # MOCK replay JSONL: 5 attempts per query class. + replay_rows = [] + queries = [ + ("quantile", "quantile_over_time(0.99, http_requests_total_latency_ms[1m])", 12.0), + ("sum", "sum by (zone) (http_requests_total)", 8.0), + ("sum", "sum by (zone) (rate(http_requests_total[5m]))", 9.0), + ] + for q_kind, q, base_lat in queries: + for i in range(5): + replay_rows.append({ + "ts": "2026-05-06T00:00:00Z", + "query": q, + "kind": q_kind, + "duration_ms": base_lat + i, + "status": "success", + "http_code": 200, + "result_type": "vector", + "result": [], + "plan_id": "p1", + "fallback_used": None, + }) + _write_jsonl(measurements / "replay.jsonl", replay_rows) + + # MOCK freshness: 3 paths × 5 samples. + for path_label, p50_anchor in (("raw", 200), ("warm", 1500), ("archive", 8000)): + rows = [] + for i in range(5): + sample_ts = base_ts + i * 100 + observed = sample_ts - p50_anchor + i # delta_ms ≈ p50_anchor + rows.append([ + path_label, sample_ts, observed, sample_ts - observed, + ]) + _write_csv( + fresh / f"{path_label}.csv", + ["path", "sample_ts_ms", "observed_ts_ms", "delta_ms"], + rows, + ) + + # MOCK ad-hoc: cold fallback marker present. + cold_resp = { + "status": "success", + "data": { + "resultType": "vector", + "result": [{"metric": {}, "value": [0, "42"]}], + "infos": ["data_source: gorilla_archive", "chunks_scanned: 3"], + }, + } + (adhoc / "cold_payments.json").write_text(json.dumps(cold_resp)) + (adhoc / "cold_payments.verdict").write_text("PASS\n") + (adhoc / "cold_payments.curlstats").write_text( + '{"http_code":200,"time_total":0.012}' + ) + + # MOCK postings exercise responses without v5 fields (most + # realistic state pre-v5-merge). + bare_resp = { + "status": "success", + "data": {"resultType": "vector", "result": [{"metric": {}, "value": [0, "1"]}]}, + } + (adhoc / "count_api_series.json").write_text(json.dumps(bare_resp)) + (adhoc / "topk_5xx_by_zone.json").write_text(json.dumps(bare_resp)) + + # MOCK compactor: minio listing before and after, +1 object then + # -5 objects (simulating concat of 6→1). + before_lines = [] + for i in range(6): + before_lines.append(json.dumps({ + "type": "file", + "key": f"raw/part-{i:04d}.gor", + "size": 1_000_000, + })) + after_lines = [ + json.dumps({ + "type": "file", + "key": "raw/merged-0000.gor", + "size": 6_000_000, + }), + ] + (compactor / "before.minio.jsonl").write_text("\n".join(before_lines)) + (compactor / "after.minio.jsonl").write_text("\n".join(after_lines)) + (compactor / "dry_run.json").write_text(json.dumps({ + "eligible_blocks": [{"key": f"raw/part-{i:04d}.gor"} for i in range(6)], + "dry_run": True, + })) + (compactor / "live_run.json").write_text(json.dumps({ + "merged_blocks": 1, + "source_count": 6, + })) + + # MOCK emitted-config status. + if kind == "fallback": + (emitted / "STATUS").write_text("fallback-placeholder\n") + else: + (emitted / "STATUS").write_text("live\n") + (emitted / "agent.bootstrap.yaml").write_text( + "# MOCK bootstrap agent config\nreceivers: {}\n" + ) + + return root + + +# ── tests ──────────────────────────────────────────────────────── + + +def test_renders_all_sections_for_happy_fixture(tmp_path): + results = _build_results_dir(tmp_path, kind="happy") + out = tmp_path / "MVP_REPORT_v6.md" + + rc = mvp_report_v6.main([ + "--results-dir", str(results), + "--num-producers", "10", + "--per-agent-cardinality", "500", + "--out", str(out), + ]) + assert rc == 0 + md = out.read_text() + + # Section headers — all 8 must be present. + for header in ( + "## §1 Stage-separated resource table", + "## §2 Per-criterion verdict (6 criteria)", + "## §3 Per-query-class breakdown", + "## §4 Postings filtering effect", + "## §5 Compaction effect", + "## §6 S3-ops cost (measured)", + "## §7 Honest caveats (non-goals)", + "## §8 Controller-emitted runtime configs", + ): + assert header in md, f"missing section header {header!r}" + + # All 6 criteria rows present in §2. + for marker in ( + "Bandwidth (per-edge B/s)", + "Query latency (p50/p99 per class)", + "Combined resource (sum of stages)", + "Accuracy (rel-err per class)", + "Cold-fallback (gorilla_archive marker)", + "Freshness (p50/p99 per path)", + ): + assert marker in md, f"missing criterion {marker!r}" + + # Verdict tokens — happy fixture should show PASS for at least + # bandwidth (egress < ingress), accuracy (<5%), cold-fallback + # (marker present), and freshness (warm p50 ~1.5s ≤ 30s). + assert "PASS" in md + + # §1 stage rows. + for stage in ("agent", "gateway", "backend-ingest", + "backend-storage", "backend-query"): + assert f"| {stage} |" in md + + # §3 per-class rows. + for cls in ("window-per-series", "label-at-instant", "combined-window-label"): + assert cls in md + + # §8 status. + assert "Emitter status: **live**" in md + + +def test_idempotent_rerun_produces_identical_md(tmp_path): + """Running the renderer twice over the same CSVs must give + byte-identical MD output.""" + results = _build_results_dir(tmp_path, kind="happy") + out1 = tmp_path / "first.md" + out2 = tmp_path / "second.md" + + for out in (out1, out2): + rc = mvp_report_v6.main([ + "--results-dir", str(results), + "--num-producers", "10", + "--per-agent-cardinality", "500", + "--out", str(out), + ]) + assert rc == 0 + + assert out1.read_text() == out2.read_text() + + +def test_sparse_fixture_renders_with_unknown_verdicts(tmp_path): + """When most CSVs are missing, the renderer must still produce + a complete MD — verdicts go UNKNOWN rather than raising.""" + results = _build_results_dir(tmp_path, kind="sparse") + out = tmp_path / "MVP_REPORT_v6.md" + + rc = mvp_report_v6.main([ + "--results-dir", str(results), + "--num-producers", "10", + "--per-agent-cardinality", "500", + "--out", str(out), + ]) + assert rc == 0 + md = out.read_text() + + # Should still have all 8 section headers. + assert "## §1 Stage-separated resource table" in md + assert "## §2 Per-criterion verdict (6 criteria)" in md + assert "## §7 Honest caveats (non-goals)" in md + + # Most criteria UNKNOWN due to missing CSVs. + assert "UNKNOWN" in md + # v5-merge-pending markers present in §4 / §6. + assert "v5-merge-pending" in md or "v5 cost-tracker not present" in md + + +def test_fallback_status_renders_correctly(tmp_path): + results = _build_results_dir(tmp_path, kind="fallback") + out = tmp_path / "MVP_REPORT_v6.md" + + rc = mvp_report_v6.main([ + "--results-dir", str(results), + "--num-producers", "10", + "--per-agent-cardinality", "500", + "--out", str(out), + ]) + assert rc == 0 + md = out.read_text() + + assert "Emitter status: **fallback-placeholder**" in md + # The §3 plan annotations should mention the fallback state. + assert "emitter fallback-placeholder" in md + + +def test_missing_results_dir_returns_error_md(tmp_path): + """If --results-dir doesn't exist, the renderer emits a + self-explanatory MD rather than crashing.""" + out = tmp_path / "out.md" + rc = mvp_report_v6.main([ + "--results-dir", str(tmp_path / "does-not-exist"), + "--num-producers", "10", + "--per-agent-cardinality", "500", + "--out", str(out), + ]) + assert rc == 0 + md = out.read_text() + assert "results dir missing" in md + + +# ── unit tests for individual helpers ──────────────────────────── + + +def test_per_edge_mean_bytes_per_s(): + rows = [ + {"edge": "edge_sdk_to_agent", "bytes_per_s": "100"}, + {"edge": "edge_sdk_to_agent", "bytes_per_s": "200"}, + {"edge": "edge_agent_to_gateway", "bytes_per_s": "50"}, + # malformed row (NaN-like) is dropped: + {"edge": "edge_agent_to_gateway", "bytes_per_s": ""}, + ] + means = mvp_report_v6._per_edge_mean_bytes_per_s(rows) + assert means["edge_sdk_to_agent"] == 150.0 + assert means["edge_agent_to_gateway"] == 50.0 + + +def test_classify_query_class(): + assert mvp_report_v6._classify_query_class( + {"query": "quantile_over_time(0.99, http_requests_total_latency_ms[1m])"} + ) == "window-per-series" + assert mvp_report_v6._classify_query_class( + {"query": "sum by (zone) (http_requests_total)"} + ) == "label-at-instant" + assert mvp_report_v6._classify_query_class( + {"query": "sum by (zone) (rate(http_requests_total[5m]))"} + ) == "combined-window-label" + # Unknown query — None. + assert mvp_report_v6._classify_query_class({"query": "vector(1)"}) is None + + +def test_count_minio_objects(tmp_path): + p = tmp_path / "before.minio.jsonl" + p.write_text("\n".join([ + json.dumps({"type": "file", "size": 100}), + json.dumps({"type": "file", "size": 200}), + # Non-file rows skipped. + json.dumps({"type": "directory"}), + # Malformed line skipped. + "not-json", + ])) + count, total = mvp_report_v6._count_minio_objects(str(p)) + assert count == 2 + assert total == 300 + + +def test_extract_int_from_response_handles_missing_field(): + body = { + "status": "success", + "data": {"infos": ["chunks_scanned: 5", "data_source: warm"]}, + } + assert mvp_report_v6._extract_int_from_response(body, "chunks_scanned") == 5 + assert mvp_report_v6._extract_int_from_response(body, "postings_filtered_series_count") is None + assert mvp_report_v6._extract_int_from_response(None, "anything") is None