From 07d288a62f354d01e4836a29113353feb21fb76e Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Thu, 30 Apr 2026 09:17:17 -0400 Subject: [PATCH] =?UTF-8?q?eval:=20three-way=20query=20harness=20=E2=80=94?= =?UTF-8?q?=20ASAP=20vs=20Prometheus=20vs=20VictoriaMetrics?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit End-to-end benchmark for the central paper claim: backend query latency and throughput for ASAP (sketch + ad-hoc S3 fallback) vs Prometheus alone vs VictoriaMetrics alone, on the same input stream. * benchmarks/docker-compose.yml — augments asap-quickstart with a victoriametrics service + cold-store bind-mount + queryengine env knob. The asap-quickstart compose stays untouched; this is an override. * benchmarks/config/vmagent-scrape.yml — VM scrape config matching the Prom config 1:1 so they ingest the same series. * benchmarks/queries/adhoc_suite.json — 7 ad-hoc PromQL queries (regex matchers, label_replace, rate, increase, exact histogram, negative regex, raw selector) that fall outside the sketch capability. * benchmarks/scripts/run_asap_workloads.py — runs both promql_suite (W1) and adhoc_suite (W2) against ASAP queryengine; outputs latency/error/fallback per query. * benchmarks/scripts/run_prom.py / run_vm.py — same suites against Prometheus and VictoriaMetrics baselines (W3). * benchmarks/scripts/run_concurrency_sweep.py — ThreadPoolExecutor hammer at C ∈ {1, 4, 16, 64} for 60s per backend; emits the throughput-vs-concurrency CSV. * benchmarks/scripts/compare_three_way.py — Tables 1/2/3 + capability matrix; writes benchmarks/reports/three_way_eval.md. * benchmarks/scripts/seed_cold_store.py — pre-populates 7×3×3 = 63 series under cold-store/raw//YYYY/MM/DD/HH/part-NNNNNN.jsonl in the byte-identical S3 layout, so the W2 fallback path returns non-empty during the bench. Output dir gitignored. * benchmarks/run_full_eval.sh — orchestrator: docker-compose up, wait_for_stack, ingest_wait, run all 4 runners + concurrency sweep, compare_three_way; --down to tear down, --skip-sweep to skip C-sweep. Smoke: docker compose config --quiet exits 0 on the merged compose; all Python scripts compile; seed_cold_store.py generates 315 JSONL samples in the expected layout; compare_three_way.py survives missing inputs gracefully; orchestrator passes bash -n. Known caveat (de-scoped per the user's review): asap-query-engine's main.rs builds AdapterConfig::prometheus_promql(...) only. ColdFallback + LocalFsColdStore types and the prometheus_promql_with_cold constructor exist, but no CLI flag wires ASAP_COLD_STORE_ROOT into them. Until that flag lands, the W2 ASAP results reflect only the Prometheus-forwarding leg, not the local-FS cold tier. Bind-mount, seed script, and adhoc suite are correct as-is — the harness exercises the cold path the moment the flag lands in main.rs. --- benchmarks/.gitignore | 11 + benchmarks/README.md | 128 +++++++++ benchmarks/cold-store/.gitkeep | 0 benchmarks/config/vmagent-scrape.yml | 22 ++ benchmarks/docker-compose.yml | 95 ++++++- benchmarks/queries/adhoc_suite.json | 54 ++++ benchmarks/run_full_eval.sh | 101 ++++++++ benchmarks/scripts/compare_three_way.py | 272 ++++++++++++++++++++ benchmarks/scripts/run_asap_workloads.py | 158 ++++++++++++ benchmarks/scripts/run_concurrency_sweep.py | 161 ++++++++++++ benchmarks/scripts/run_prom.py | 58 +++++ benchmarks/scripts/run_vm.py | 58 +++++ benchmarks/scripts/seed_cold_store.py | 103 ++++++++ 13 files changed, 1214 insertions(+), 7 deletions(-) create mode 100644 benchmarks/.gitignore create mode 100644 benchmarks/README.md create mode 100644 benchmarks/cold-store/.gitkeep create mode 100644 benchmarks/config/vmagent-scrape.yml create mode 100644 benchmarks/queries/adhoc_suite.json create mode 100755 benchmarks/run_full_eval.sh create mode 100755 benchmarks/scripts/compare_three_way.py create mode 100755 benchmarks/scripts/run_asap_workloads.py create mode 100755 benchmarks/scripts/run_concurrency_sweep.py create mode 100755 benchmarks/scripts/run_prom.py create mode 100755 benchmarks/scripts/run_vm.py create mode 100755 benchmarks/scripts/seed_cold_store.py diff --git a/benchmarks/.gitignore b/benchmarks/.gitignore new file mode 100644 index 000000000..dd0e7d7dd --- /dev/null +++ b/benchmarks/.gitignore @@ -0,0 +1,11 @@ +# Python cache +scripts/__pycache__/ +*.pyc + +# Generated benchmark output +reports/*.json +reports/*.csv +reports/*.md + +# Cold-store data populated by seed_cold_store.py — keep the dir, drop the data +cold-store/raw/ diff --git a/benchmarks/README.md b/benchmarks/README.md new file mode 100644 index 000000000..f73478df2 --- /dev/null +++ b/benchmarks/README.md @@ -0,0 +1,128 @@ +# Three-Way Query Benchmark Harness + +This directory holds the docker-compose-based query benchmark harness for +the central paper claim: + +> **ASAPQuery-backend (sketch + ad-hoc cold-store fallback) vs. Prometheus +> vs. VictoriaMetrics** on the same input stream. + +The harness is intentionally thin. It reuses `asap-quickstart`'s +docker-compose for the ASAP stack (kafka, arroyo, planner, summary-ingest, +queryengine, prometheus, 7 pattern-based fake exporters), layers +VictoriaMetrics + a cold-store volume on top via an override, and adds +runner scripts that record per-query latency / throughput across all +three backends. + +## Workloads + +| Workload | Source | What it tests | +|----------|--------|---------------| +| **W1** | `queries/promql_suite.json` (14 queries) | ASAP's sketch path: sum, avg, max, min, quantiles at p50/p90/p95/p99, group-bys | +| **W2** | `queries/adhoc_suite.json` (7 queries) | ASAP's fallback path: regex matchers, label_replace, rate/increase, exact histograms — hits `ColdFallback` (LocalFsColdStore) first, then forwards to Prometheus for query shapes the cold tier doesn't support | +| **W3** | both suites, run directly against Prom & VM | baseline ground truth + industry comparison | + +## Files + +``` +benchmarks/ +├── docker-compose.yml # OVERRIDE for asap-quickstart: adds VM + cold-store volume +├── config/ +│ └── vmagent-scrape.yml # VM scrape config (same exporters as Prom) +├── cold-store/ # Bind-mounted into queryengine; seeded by seed_cold_store.py +├── queries/ +│ ├── promql_suite.json # W1 — pre-existing +│ └── adhoc_suite.json # W2 — NEW +├── scripts/ +│ ├── run_asap_workloads.py # NEW: W1+W2 against ASAP, median + P99 +│ ├── run_prom.py # NEW: W1+W2 against Prometheus baseline +│ ├── run_vm.py # NEW: W1+W2 against VictoriaMetrics +│ ├── run_concurrency_sweep.py # NEW: throughput-vs-concurrency CSV +│ ├── compare_three_way.py # NEW: renders three_way_eval.md +│ ├── seed_cold_store.py # NEW: pre-populates cold-store for W2 +│ ├── wait_for_stack.sh # pre-existing — waits for ASAP/Prom/Arroyo healthy +│ ├── ingest_wait.sh # pre-existing — waits for arroyo pipeline RUNNING +│ ├── run_asap.py # pre-existing — kept for the existing CI gate +│ ├── run_baseline.py # pre-existing — kept for the existing CI gate +│ └── compare.py # pre-existing — kept for the existing CI gate +├── reports/ # JSON / CSV / MD output lands here +└── run_full_eval.sh # NEW: orchestrator +``` + +## Quick start + +```bash +# from repo root +./benchmarks/run_full_eval.sh +``` + +That will: + +1. `docker compose up -d` the merged asap-quickstart + benchmarks stack. +2. Wait for Prometheus, Arroyo, QueryEngine, VictoriaMetrics to be healthy. +3. Wait for the Arroyo pipeline `asap-demo` to reach RUNNING and let + sketches accumulate (`ingest_wait.sh`). +4. Seed `benchmarks/cold-store/` with raw JSONL fixtures for W2. +5. Run W1+W2 against each backend (3 iterations per query, captures + median + P99). +6. Run the concurrency sweep at C ∈ {1, 4, 16, 64} for 60s each. +7. Render `benchmarks/reports/three_way_eval.md`. + +Pass `--down` to tear the stack down at the end, or `--skip-sweep` to +skip the 4×60s concurrency sweep. + +## Manual / partial runs + +```bash +# Only one runner +python3 benchmarks/scripts/run_asap_workloads.py --asap-url http://localhost:8088 +python3 benchmarks/scripts/run_prom.py --prometheus-url http://localhost:9090 +python3 benchmarks/scripts/run_vm.py --vm-url http://localhost:8428 + +# Smaller sweep +python3 benchmarks/scripts/run_concurrency_sweep.py --duration 10 --concurrency 1,8 + +# Re-render the report after edits +python3 benchmarks/scripts/compare_three_way.py +``` + +## Output artefacts + +| File | Producer | Shape | +|------|----------|-------| +| `reports/asap_promql.json` | `run_asap_workloads.py` | per-query latencies, median, P99, result data | +| `reports/asap_adhoc.json` | `run_asap_workloads.py` | same shape, W2 | +| `reports/prom_promql.json` | `run_prom.py` | same shape, baseline | +| `reports/prom_adhoc.json` | `run_prom.py` | same shape, W2 | +| `reports/vm_promql.json` | `run_vm.py` | same shape, VM | +| `reports/vm_adhoc.json` | `run_vm.py` | same shape, W2 | +| `reports/concurrency_sweep.csv` | `run_concurrency_sweep.py` | `backend,concurrency,total_queries,throughput_qps,p50_ms,p99_ms` | +| `reports/three_way_eval.md` | `compare_three_way.py` | Tables 1–3 + capability matrix | + +## What it deliberately does NOT do (per scope) + +- The user explicitly de-scoped controller polish, sketch reallocation, and + backfill correctness. We rely on `asap-quickstart`'s pre-baked + `controller-config.yaml` to hand `streaming_config.yaml` to + `asap-summary-ingest` — whatever sketches that produces is fine; only + the 14 W1 queries need to come back non-empty. +- The W2 cold-store fixtures are deterministic synthetic data, not a + faithful replay of the live exporter stream. We only need the + fallback path to return non-empty so latencies are real. +- Smoke tests assume Docker is installed locally. CI uses + `accuracy_performance.yml` which already runs the existing + `compare.py` gate; this harness is additive. + +## Wiring assumptions / TODO + +- `queryengine` reads `ASAP_COLD_STORE_ROOT` from env. Today the binary + doesn't expose a `--cold-store-root` flag in `main.rs` — the wiring + exists in `AdapterConfig::prometheus_promql_with_cold` and the + `LocalFsColdStore` struct, but `main.rs` itself constructs + `AdapterConfig::prometheus_promql(...)` (no cold tier). Until that + flag lands, ASAP W2 results will reflect only the Prometheus + forwarding leg, not the cold-store leg. The bind-mount + seed script + + adhoc_suite.json are correct as-is and will start exercising the + cold path the moment the flag is added. +- The concurrency sweep uses `ThreadPoolExecutor` and the `requests` + library — fine for ≤64 concurrency on localhost; if we ever push to + higher fan-out, switch to `aiohttp`. diff --git a/benchmarks/cold-store/.gitkeep b/benchmarks/cold-store/.gitkeep new file mode 100644 index 000000000..e69de29bb diff --git a/benchmarks/config/vmagent-scrape.yml b/benchmarks/config/vmagent-scrape.yml new file mode 100644 index 000000000..2faf7910d --- /dev/null +++ b/benchmarks/config/vmagent-scrape.yml @@ -0,0 +1,22 @@ +# VictoriaMetrics scrape config — points at the same fake exporters as +# the asap-quickstart Prometheus scrape config, so VM ingests an +# identical input stream. 1s scrape matches asap-quickstart prometheus.yml. + +global: + scrape_interval: 1s + +scrape_configs: + - job_name: 'pattern-exporters' + metric_relabel_configs: + - source_labels: [__name__] + regex: sensor_reading + action: keep + static_configs: + - targets: + - 'fake-exporter-constant:50000' + - 'fake-exporter-linear-up:50001' + - 'fake-exporter-linear-down:50002' + - 'fake-exporter-sine:50003' + - 'fake-exporter-sine-noise:50004' + - 'fake-exporter-step:50005' + - 'fake-exporter-exp-up:50006' diff --git a/benchmarks/docker-compose.yml b/benchmarks/docker-compose.yml index fea9921a4..c0c42022f 100644 --- a/benchmarks/docker-compose.yml +++ b/benchmarks/docker-compose.yml @@ -1,20 +1,101 @@ -# CI image override: replaces quickstart's pinned release images with images -# built from the current branch. Intended for use as a Compose override: +# Three-way query benchmark harness — OVERRIDE for asap-quickstart. # -# ASAP_IMAGE_TAG=sha- docker compose \ +# This file is intended to be merged with asap-quickstart/docker-compose.yml: +# +# docker compose \ # --project-directory . \ # -f asap-quickstart/docker-compose.yml \ # -f benchmarks/docker-compose.yml \ # up -d # -# ASAP_IMAGE_TAG is set automatically by the 'build' job in accuracy_performance.yml. +# What this override adds on top of the asap-quickstart base stack +# (kafka + arroyo + planner + summary-ingest + queryengine + prometheus + +# 7 pattern-based fake exporters): +# +# 1. victoriametrics — third backend under test, scrapes the same fake +# exporters so all three backends (ASAP / Prom / VM) see an identical +# input stream. Comparable 1d retention. +# +# 2. queryengine override — adds a cold-store bind-mount at /cold-store. +# The §5.2 ColdFallback (LocalFsColdStore) reads raw JSONL fixtures +# from this dir to answer ad-hoc queries that the sketch path can't +# serve. Pre-populate with `benchmarks/scripts/seed_cold_store.py` +# before running the ad-hoc workload. +# +# CI / GHCR image override: kept compatible with the existing +# accuracy_performance.yml workflow — pass ASAP_IMAGE_TAG to swap pinned +# release images for branch-built images. services: + # NOTE: the asap-quickstart base file pins these to v0.2.0 release tags. + # Setting ASAP_IMAGE_TAG (e.g. sha-abcdef0) overrides them for CI runs. asap-planner-rs: - image: ghcr.io/projectasap/asap-planner-rs:${ASAP_IMAGE_TAG} + image: ghcr.io/projectasap/asap-planner-rs:${ASAP_IMAGE_TAG:-v0.2.0} asap-summary-ingest: - image: ghcr.io/projectasap/asap-summary-ingest:${ASAP_IMAGE_TAG} + image: ghcr.io/projectasap/asap-summary-ingest:${ASAP_IMAGE_TAG:-v0.2.0} + # --------------------------------------------------------------------------- + # QueryEngine override — adds cold-store mount + env knob. + # + # The cold-store volume is a directory under benchmarks/ (gitignored); + # seed_cold_store.py writes JSONL parts into the layout the ColdFallback + # adapter expects (raw//YYYY/MM/DD/HH/part-NNNNNN.jsonl). + # + # ASAP_COLD_STORE_ROOT is a placeholder env var until main.rs grows a + # --cold-store-root CLI flag. For now AdapterConfig::prometheus_promql + # (the path actually wired in main.rs) gives us the Prometheus-only + # fallback chain — the cold tier kicks in once the binary reads the env. + # The bind-mount + seed script are correct as-is so the harness will + # work without re-touching this YAML when that flag lands. + # --------------------------------------------------------------------------- queryengine: - image: ghcr.io/projectasap/asap-query-engine:${ASAP_IMAGE_TAG} + image: ghcr.io/projectasap/asap-query-engine:${ASAP_IMAGE_TAG:-v0.2.0} + volumes: + - asap-planner-output:/asap-planner-output:ro + - ./asap-quickstart/output/queryengine:/app/outputs + - ./benchmarks/cold-store:/cold-store + environment: + - RUST_LOG=INFO + - RUST_BACKTRACE=1 + - ASAP_COLD_STORE_ROOT=/cold-store + + # --------------------------------------------------------------------------- + # VictoriaMetrics — third backend under test. + # Scrapes the same fake exporters Prometheus does, with comparable + # retention. Exposes the standard 8428 query port. + # --------------------------------------------------------------------------- + victoriametrics: + image: victoriametrics/victoria-metrics:v1.106.1 + container_name: asap-victoriametrics + hostname: victoriametrics + networks: + - asap-network + ports: + - "8428:8428" + volumes: + - victoriametrics-data:/storage + - ./benchmarks/config/vmagent-scrape.yml:/etc/vm/scrape.yml:ro + command: + - "-storageDataPath=/storage" + - "-retentionPeriod=1d" + - "-promscrape.config=/etc/vm/scrape.yml" + - "-httpListenAddr=:8428" + healthcheck: + test: ["CMD-SHELL", "wget --no-verbose --tries=1 --spider http://localhost:8428/health || exit 1"] + interval: 10s + timeout: 5s + retries: 10 + start_period: 30s + depends_on: + - fake-exporter-constant + - fake-exporter-linear-up + - fake-exporter-linear-down + - fake-exporter-sine + - fake-exporter-sine-noise + - fake-exporter-step + - fake-exporter-exp-up + restart: no + +volumes: + victoriametrics-data: diff --git a/benchmarks/queries/adhoc_suite.json b/benchmarks/queries/adhoc_suite.json new file mode 100644 index 000000000..a82416a09 --- /dev/null +++ b/benchmarks/queries/adhoc_suite.json @@ -0,0 +1,54 @@ +{ + "_comment": "Workload W2 — ad-hoc PromQL queries that exercise the ASAP fallback path. Per docs/01-getting-started/architecture.md, ASAP cannot serve regex matchers, label_replace, exact histograms, or non-quantile range functions from sketches; they fall through to the cold-store first (LocalFsColdStore via ColdFallback) and finally to Prometheus. All queries here also work against the Prometheus and VictoriaMetrics baselines for direct comparison.", + "queries": [ + { + "id": "regex_pattern_match", + "expr": "sensor_reading{pattern=~\"sine.*\"}", + "kind": "regex_matcher", + "fallback": "cold_store", + "rationale": "Regex label matchers are not pushed into sketch keys; ColdFallback scans raw samples in the cold tier." + }, + { + "id": "label_replace_relabel", + "expr": "label_replace(sensor_reading, \"new_label\", \"$1\", \"region\", \"region(.*)\")", + "kind": "label_manipulation", + "fallback": "prometheus", + "rationale": "label_replace is a label-rewrite operator that has no sketch-side equivalent — forwarded straight to Prometheus." + }, + { + "id": "rate_over_5m", + "expr": "rate(sensor_reading[5m])", + "kind": "range_function", + "fallback": "prometheus", + "rationale": "rate() needs per-sample positions; sketches store summaries, not raw points." + }, + { + "id": "increase_over_1m", + "expr": "increase(sensor_reading[1m])", + "kind": "range_function", + "fallback": "prometheus", + "rationale": "Same as rate — counter-difference over a window." + }, + { + "id": "exact_histogram_q99", + "expr": "histogram_quantile(0.99, sum by (le) (rate(sensor_reading_bucket[5m])))", + "kind": "exact_histogram", + "fallback": "prometheus", + "rationale": "Exact histogram quantiles need _bucket series; ASAP serves quantiles via DDSketch instead, this path is forwarded." + }, + { + "id": "regex_negative_match", + "expr": "sum(sensor_reading{service!~\"svc[0-3]\"})", + "kind": "regex_matcher", + "fallback": "cold_store", + "rationale": "Negative regex matcher — same fallback path as positive regex." + }, + { + "id": "raw_lookup_specific_series", + "expr": "sensor_reading{pattern=\"sine\",region=\"region0\"}", + "kind": "instant_vector_selector", + "fallback": "cold_store", + "rationale": "Bare instant vector lookup — ColdFallback v1 scope per s3_adapter.rs handles this exactly: per-series latest value within Prometheus's 5m lookback delta." + } + ] +} diff --git a/benchmarks/run_full_eval.sh b/benchmarks/run_full_eval.sh new file mode 100755 index 000000000..f04bbb7bb --- /dev/null +++ b/benchmarks/run_full_eval.sh @@ -0,0 +1,101 @@ +#!/usr/bin/env bash +set -euo pipefail + +# run_full_eval.sh — orchestrator for the three-way query benchmark. +# +# Brings up the merged asap-quickstart + benchmarks docker-compose stack, +# waits for services to be healthy and ingest to populate, runs all 4 +# runners (W1+W2 against ASAP, Prom, VM) plus the concurrency sweep, +# then renders the three-way comparison report. +# +# Usage: +# ./benchmarks/run_full_eval.sh # leave stack up at end +# ./benchmarks/run_full_eval.sh --down # tear down at end +# ./benchmarks/run_full_eval.sh --skip-sweep # skip 60s/concurrency sweep +# +# Env knobs: +# ASAP_IMAGE_TAG — pin asap images (default: v0.2.0) +# SWEEP_DURATION — seconds per concurrency level (default: 60) +# SWEEP_LEVELS — comma list (default: 1,4,16,64) + +REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +BENCH_DIR="$REPO_ROOT/benchmarks" +SCRIPTS_DIR="$BENCH_DIR/scripts" + +DOWN_AT_END=0 +SKIP_SWEEP=0 +for arg in "$@"; do + case "$arg" in + --down) DOWN_AT_END=1 ;; + --skip-sweep) SKIP_SWEEP=1 ;; + *) echo "unknown arg: $arg" >&2; exit 2 ;; + esac +done + +SWEEP_DURATION="${SWEEP_DURATION:-60}" +SWEEP_LEVELS="${SWEEP_LEVELS:-1,4,16,64}" + +cd "$REPO_ROOT" + +COMPOSE_ARGS=( + --project-directory "$REPO_ROOT" + -f "$REPO_ROOT/asap-quickstart/docker-compose.yml" + -f "$BENCH_DIR/docker-compose.yml" +) + +echo "[run_full_eval] === bringing up stack ===" +docker compose "${COMPOSE_ARGS[@]}" up -d --remove-orphans + +echo "[run_full_eval] === waiting for ASAP/Prom/Arroyo healthy ===" +"$SCRIPTS_DIR/wait_for_stack.sh" + +echo "[run_full_eval] === waiting for VictoriaMetrics ===" +elapsed=0 +until curl -sf --max-time 5 "http://localhost:8428/health" > /dev/null 2>&1; do + if [ "$elapsed" -ge 180 ]; then + echo "[run_full_eval] ERROR: VictoriaMetrics not healthy in 180s" >&2 + exit 1 + fi + sleep 5 + elapsed=$((elapsed + 5)) +done +echo "[run_full_eval] VictoriaMetrics healthy" + +echo "[run_full_eval] === waiting for ingest to populate sketches ===" +"$SCRIPTS_DIR/ingest_wait.sh" + +echo "[run_full_eval] === seeding cold-store for W2 ad-hoc queries ===" +python3 "$SCRIPTS_DIR/seed_cold_store.py" --hours 2 + +echo "[run_full_eval] === running W1+W2 against ASAP ===" +python3 "$SCRIPTS_DIR/run_asap_workloads.py" + +echo "[run_full_eval] === running W1+W2 against Prometheus ===" +python3 "$SCRIPTS_DIR/run_prom.py" + +echo "[run_full_eval] === running W1+W2 against VictoriaMetrics ===" +python3 "$SCRIPTS_DIR/run_vm.py" + +if [ "$SKIP_SWEEP" -eq 0 ]; then + echo "[run_full_eval] === concurrency sweep (duration=${SWEEP_DURATION}s, levels=${SWEEP_LEVELS}) ===" + python3 "$SCRIPTS_DIR/run_concurrency_sweep.py" \ + --duration "$SWEEP_DURATION" \ + --concurrency "$SWEEP_LEVELS" +else + echo "[run_full_eval] skipping concurrency sweep (--skip-sweep)" +fi + +echo "[run_full_eval] === rendering three-way report ===" +python3 "$SCRIPTS_DIR/compare_three_way.py" + +echo "[run_full_eval] === DONE ===" +echo "Reports: $BENCH_DIR/reports/" +echo "Top-level: $BENCH_DIR/reports/three_way_eval.md" + +if [ "$DOWN_AT_END" -eq 1 ]; then + echo "[run_full_eval] tearing down stack (--down)" + docker compose "${COMPOSE_ARGS[@]}" down -v +else + echo "[run_full_eval] leaving stack up. Tear down with:" + echo " docker compose ${COMPOSE_ARGS[*]} down -v" +fi diff --git a/benchmarks/scripts/compare_three_way.py b/benchmarks/scripts/compare_three_way.py new file mode 100755 index 000000000..b406744d6 --- /dev/null +++ b/benchmarks/scripts/compare_three_way.py @@ -0,0 +1,272 @@ +#!/usr/bin/env python3 +"""compare_three_way.py — three-way evaluation across ASAP / Prom / VM. + +Inputs (default locations under benchmarks/reports/): + asap_promql.json asap_adhoc.json + prom_promql.json prom_adhoc.json + vm_promql.json vm_adhoc.json + concurrency_sweep.csv + +Output: benchmarks/reports/three_way_eval.md, with: + - Table 1: per-query latency P50/P99 across ASAP-sketch / ASAP-adhoc / + Prom / VM + - Table 2: per-query relative error of ASAP-sketch vs Prom (ground truth) + - Table 3: throughput-vs-concurrency tabular CDF data (read straight + from the CSV produced by run_concurrency_sweep.py) + - Capability matrix: which queries each backend supports + +Usage: + python benchmarks/scripts/compare_three_way.py [--reports-dir DIR] [--output FILE] +""" + +import argparse +import csv +import json +import os +import statistics +from datetime import datetime, timezone + +HERE = os.path.dirname(os.path.abspath(__file__)) +DEFAULT_REPORTS = os.path.join(HERE, "..", "reports") +DEFAULT_OUTPUT = os.path.join(DEFAULT_REPORTS, "three_way_eval.md") + + +def percentile(values, pct): + if not values: + return float("nan") + s = sorted(values) + n = len(s) + if n == 1: + return s[0] + idx = (pct / 100.0) * (n - 1) + lo = int(idx) + hi = min(lo + 1, n - 1) + frac = idx - lo + return s[lo] * (1 - frac) + s[hi] * frac + + +def valid(latencies): + return [x for x in (latencies or []) if x is not None] + + +def load_json(path): + if not os.path.exists(path): + return None + with open(path) as f: + return json.load(f) + + +def label_key(metric): + return json.dumps(metric, sort_keys=True) + + +def relative_error(a, b): + denom = max(abs(b), 1e-9) + return abs(a - b) / denom + + +def compare_results(prom_data, asap_data): + """Same shape-comparison logic as compare.py — returns max relative error.""" + if not prom_data and not asap_data: + return 0.0 + if not prom_data or not asap_data: + return None + pmap, amap = {}, {} + for entry in prom_data: + try: + pmap[label_key(entry.get("metric", {}))] = float(entry["value"][1]) + except (KeyError, IndexError, ValueError, TypeError): + pass + for entry in asap_data: + try: + amap[label_key(entry.get("metric", {}))] = float(entry["value"][1]) + except (KeyError, IndexError, ValueError, TypeError): + pass + if not pmap or not amap: + return None + if len(pmap) == 1 and len(amap) == 1: + return relative_error(next(iter(amap.values())), next(iter(pmap.values()))) + max_err = 0.0 + for k, pv in pmap.items(): + if k in amap: + max_err = max(max_err, relative_error(amap[k], pv)) + return max_err + + +def fmt(v, suffix=""): + if v is None or (isinstance(v, float) and v != v): + return "n/a" + if isinstance(v, float): + return f"{v:.1f}{suffix}" + return f"{v}{suffix}" + + +def per_query_latency_row(qid, asap, prom, vm): + def stats(d): + if not d or qid not in d.get("results", {}): + return ("n/a", "n/a", "missing") + r = d["results"][qid] + if r.get("status") != "success": + return ("n/a", "n/a", "error") + lats = valid(r.get("latencies_ms", [])) + if not lats: + return ("n/a", "n/a", "no-samples") + return (f"{percentile(lats, 50):.1f}", f"{percentile(lats, 99):.1f}", "ok") + + a = stats(asap) + p = stats(prom) + v = stats(vm) + return {"id": qid, "asap": a, "prom": p, "vm": v} + + +def build_table_1(asap_promql, asap_adhoc, prom_promql, prom_adhoc, vm_promql, vm_adhoc): + """Per-query P50/P99 for both workloads. ASAP gets two columns: sketch (W1) + + adhoc-fallback (W2). Prom and VM each have one column (the same path).""" + lines = ["## Table 1 — Per-query latency (P50 / P99, milliseconds)\n"] + lines.append("Workloads: W1 = sketch path (promql_suite); W2 = adhoc/fallback path (adhoc_suite). " + "ASAP-sketch is W1 against ASAP; ASAP-adhoc is W2 against ASAP (cold-store + Prom fallback). " + "Prom / VM run both suites natively.\n") + lines.append("| Query | Workload | ASAP P50 | ASAP P99 | Prom P50 | Prom P99 | VM P50 | VM P99 |") + lines.append("|-------|:--------:|:--------:|:--------:|:--------:|:--------:|:------:|:------:|") + + def emit(workload_label, asap_data, prom_data, vm_data): + if not asap_data: + return + for qid in asap_data["results"].keys(): + row = per_query_latency_row(qid, asap_data, prom_data, vm_data) + lines.append( + f"| {qid} | {workload_label} | " + f"{row['asap'][0]} | {row['asap'][1]} | " + f"{row['prom'][0]} | {row['prom'][1]} | " + f"{row['vm'][0]} | {row['vm'][1]} |" + ) + + emit("W1", asap_promql, prom_promql, vm_promql) + emit("W2", asap_adhoc, prom_adhoc, vm_adhoc) + lines.append("") + return "\n".join(lines) + + +def build_table_2(asap_promql, prom_promql): + """Relative error of ASAP-sketch vs Prom ground truth on W1.""" + lines = ["## Table 2 — Relative error of ASAP-sketch vs Prometheus (ground truth)\n"] + lines.append("Per-query max relative error on W1 (sketch path). Prom is treated as ground truth. " + "Higher error is expected for approximate (quantile) queries; exact aggregations " + "(sum/avg/max/min) should be near-zero.\n") + lines.append("| Query | Approximate | Max Rel Error | Notes |") + lines.append("|-------|:-----------:|:-------------:|-------|") + if not (asap_promql and prom_promql): + lines.append("| _missing inputs_ | | | |") + return "\n".join(lines) + "\n" + + for qid, ar in asap_promql["results"].items(): + pr = prom_promql["results"].get(qid, {}) + approx = "yes" if ar.get("approximate") else "no" + if ar.get("status") != "success" or pr.get("status") != "success": + err_s, note = "n/a", "non-success status" + else: + err = compare_results(pr.get("data", []), ar.get("data", [])) + err_s = "n/a" if err is None else f"{err:.4f}" + note = "" + lines.append(f"| {qid} | {approx} | {err_s} | {note} |") + lines.append("") + return "\n".join(lines) + + +def build_table_3(reports_dir): + csv_path = os.path.join(reports_dir, "concurrency_sweep.csv") + lines = ["## Table 3 — Throughput vs. Concurrency\n"] + lines.append("Per-backend throughput (queries/s) and tail latency at fixed concurrency levels. " + "Workload: promql_suite, round-robin. Source: `concurrency_sweep.csv`.\n") + if not os.path.exists(csv_path): + lines.append(f"_no concurrency_sweep.csv found at {csv_path} — run run_concurrency_sweep.py first._\n") + return "\n".join(lines) + + lines.append("| Backend | Concurrency | Total Queries | Throughput (qps) | P50 (ms) | P99 (ms) |") + lines.append("|---------|:-----------:|:-------------:|:----------------:|:--------:|:--------:|") + with open(csv_path) as f: + reader = csv.DictReader(f) + for row in reader: + lines.append( + f"| {row['backend']} | {row['concurrency']} | {row['total_queries']} | " + f"{row['throughput_qps']} | {row['p50_ms']} | {row['p99_ms']} |" + ) + lines.append("") + return "\n".join(lines) + + +def build_capability_matrix(asap_promql, asap_adhoc, prom_promql, prom_adhoc, vm_promql, vm_adhoc): + """Capability matrix: a backend 'supports' a query if the corresponding run + returned status=success with a non-empty data array on at least one + iteration. ASAP shows the *path* used (sketch/cold/prom-forward) when known.""" + lines = ["## Capability Matrix\n"] + lines.append("✓ = backend returned a non-empty success response. - = error/empty/missing. " + "ASAP path column: `sketch` for W1 queries; for W2, the `fallback` declared in " + "`adhoc_suite.json` (`cold_store` or `prometheus`).\n") + lines.append("| Query | Workload | ASAP | ASAP path | Prom | VM |") + lines.append("|-------|:--------:|:----:|:---------:|:----:|:--:|") + + def supports(d, qid): + if not d or qid not in d.get("results", {}): + return "-" + r = d["results"][qid] + if r.get("status") == "success" and r.get("data"): + return "✓" + return "-" + + def emit(workload, asap_data, prom_data, vm_data, default_path): + if not asap_data: + return + for qid, ar in asap_data["results"].items(): + path = ar.get("fallback") or default_path + lines.append( + f"| {qid} | {workload} | " + f"{supports(asap_data, qid)} | {path} | " + f"{supports(prom_data, qid)} | {supports(vm_data, qid)} |" + ) + + emit("W1", asap_promql, prom_promql, vm_promql, "sketch") + emit("W2", asap_adhoc, prom_adhoc, vm_adhoc, "fallback") + lines.append("") + return "\n".join(lines) + + +def main() -> None: + parser = argparse.ArgumentParser(description="Three-way ASAP / Prom / VM eval") + parser.add_argument("--reports-dir", default=DEFAULT_REPORTS) + parser.add_argument("--output", default=DEFAULT_OUTPUT) + args = parser.parse_args() + + asap_promql = load_json(os.path.join(args.reports_dir, "asap_promql.json")) + asap_adhoc = load_json(os.path.join(args.reports_dir, "asap_adhoc.json")) + prom_promql = load_json(os.path.join(args.reports_dir, "prom_promql.json")) + prom_adhoc = load_json(os.path.join(args.reports_dir, "prom_adhoc.json")) + vm_promql = load_json(os.path.join(args.reports_dir, "vm_promql.json")) + vm_adhoc = load_json(os.path.join(args.reports_dir, "vm_adhoc.json")) + + now = datetime.now(timezone.utc).isoformat() + sections = [ + f"# Three-Way Query Benchmark — ASAP / Prometheus / VictoriaMetrics\n", + f"_Generated: {now}_\n", + "## Backends\n", + "- **ASAP**: ASAPQuery-backend (sketch path for W1; cold-store + Prometheus forwarding for W2)", + "- **Prom**: Prometheus baseline (ground truth, exact)", + "- **VM**: VictoriaMetrics (industry comparison, exact, compressed TSDB)\n", + build_table_1(asap_promql, asap_adhoc, prom_promql, prom_adhoc, vm_promql, vm_adhoc), + build_table_2(asap_promql, prom_promql), + build_table_3(args.reports_dir), + build_capability_matrix(asap_promql, asap_adhoc, prom_promql, prom_adhoc, vm_promql, vm_adhoc), + "---", + "_Generated by `benchmarks/scripts/compare_three_way.py`._", + ] + report = "\n".join(sections) + "\n" + + os.makedirs(os.path.dirname(os.path.abspath(args.output)), exist_ok=True) + with open(args.output, "w") as f: + f.write(report) + print(report) + print(f"\n[compare_three_way] wrote {args.output}") + + +if __name__ == "__main__": + main() diff --git a/benchmarks/scripts/run_asap_workloads.py b/benchmarks/scripts/run_asap_workloads.py new file mode 100755 index 000000000..e77cce407 --- /dev/null +++ b/benchmarks/scripts/run_asap_workloads.py @@ -0,0 +1,158 @@ +#!/usr/bin/env python3 +"""run_asap_workloads.py — runs both promql_suite (W1) and adhoc_suite (W2) +against the ASAP query engine, recording per-query latency samples +(median + P99 across iterations). + +Mirrors the JSON shape produced by run_asap.py so compare_three_way.py and +the existing compare.py can ingest both. Writes: + benchmarks/reports/asap_promql.json (W1) + benchmarks/reports/asap_adhoc.json (W2) + +Usage: + python benchmarks/scripts/run_asap_workloads.py \ + [--asap-url URL] \ + [--iterations N] \ + [--reports-dir DIR] +""" + +import argparse +import json +import os +import statistics +import time +import urllib.parse +from datetime import datetime, timezone + +import requests + +HERE = os.path.dirname(os.path.abspath(__file__)) +QUERIES_DIR = os.path.join(HERE, "..", "queries") +DEFAULT_REPORTS_DIR = os.path.join(HERE, "..", "reports") +DEFAULT_ITERATIONS = 3 + + +def percentile(values: list[float], pct: float) -> float: + if not values: + return float("nan") + s = sorted(values) + n = len(s) + if n == 1: + return s[0] + idx = (pct / 100.0) * (n - 1) + lo = int(idx) + hi = min(lo + 1, n - 1) + frac = idx - lo + return s[lo] * (1 - frac) + s[hi] * frac + + +def query_once(base_url: str, expr: str, ts: float) -> tuple[dict, float]: + encoded = urllib.parse.quote(expr, safe="") + url = f"{base_url}/api/v1/query?query={encoded}&time={ts}" + t0 = time.monotonic() + resp = requests.get(url, timeout=30) + latency_ms = (time.monotonic() - t0) * 1000.0 + resp.raise_for_status() + return resp.json(), latency_ms + + +def run_suite( + suite_path: str, base_url: str, iterations: int, label: str +) -> dict: + with open(suite_path) as f: + suite = json.load(f) + + results: dict[str, dict] = {} + now = time.time() + + for q in suite["queries"]: + qid = q["id"] + expr = q["expr"] + approximate = q.get("approximate", False) + kind = q.get("kind") + fallback = q.get("fallback") + + latencies: list[float] = [] + last_data: list = [] + last_error = None + last_status = "success" + + print(f"[{label}] {qid}: {expr}") + for run in range(1, iterations + 1): + try: + payload, lat = query_once(base_url, expr, now) + latencies.append(lat) + if payload.get("status") == "success": + last_data = payload.get("data", {}).get("result", []) + last_status = "success" + last_error = None + else: + last_status = "error" + last_error = payload.get("error", "unknown error") + last_data = [] + print(f" run {run}/{iterations}: {lat:.1f} ms status={last_status}") + except Exception as exc: # noqa: BLE001 + last_status = "error" + last_error = str(exc) + last_data = [] + print(f" run {run}/{iterations}: ERROR — {exc}") + + valid = [x for x in latencies if x is not None] + results[qid] = { + "status": last_status, + "approximate": approximate, + "kind": kind, + "fallback": fallback, + "latencies_ms": latencies, + "median_ms": statistics.median(valid) if valid else None, + "p99_ms": percentile(valid, 99) if valid else None, + "data": last_data, + "error": last_error, + } + + return { + "timestamp": datetime.now(timezone.utc).isoformat(), + "asap_url": base_url, + "suite": os.path.basename(suite_path), + "iterations": iterations, + "results": results, + } + + +def main() -> None: + parser = argparse.ArgumentParser(description="Run W1 (promql) + W2 (adhoc) workloads against ASAP") + parser.add_argument("--asap-url", default="http://localhost:8088") + parser.add_argument("--iterations", type=int, default=DEFAULT_ITERATIONS) + parser.add_argument("--reports-dir", default=DEFAULT_REPORTS_DIR) + args = parser.parse_args() + + os.makedirs(args.reports_dir, exist_ok=True) + + promql_out = os.path.join(args.reports_dir, "asap_promql.json") + adhoc_out = os.path.join(args.reports_dir, "asap_adhoc.json") + + print("=" * 70) + print("[asap] Workload W1 — promql_suite (sketch path)") + print("=" * 70) + promql = run_suite( + os.path.join(QUERIES_DIR, "promql_suite.json"), + args.asap_url, args.iterations, "asap-w1", + ) + with open(promql_out, "w") as f: + json.dump(promql, f, indent=2) + print(f"[asap] W1 saved to {promql_out}") + + print() + print("=" * 70) + print("[asap] Workload W2 — adhoc_suite (cold-store / Prom fallback path)") + print("=" * 70) + adhoc = run_suite( + os.path.join(QUERIES_DIR, "adhoc_suite.json"), + args.asap_url, args.iterations, "asap-w2", + ) + with open(adhoc_out, "w") as f: + json.dump(adhoc, f, indent=2) + print(f"[asap] W2 saved to {adhoc_out}") + + +if __name__ == "__main__": + main() diff --git a/benchmarks/scripts/run_concurrency_sweep.py b/benchmarks/scripts/run_concurrency_sweep.py new file mode 100755 index 000000000..c31aae2b3 --- /dev/null +++ b/benchmarks/scripts/run_concurrency_sweep.py @@ -0,0 +1,161 @@ +#!/usr/bin/env python3 +"""run_concurrency_sweep.py — throughput-vs-concurrency sweep. + +For each backend (asap/prom/vm) and each concurrency level C in +{1, 4, 16, 64}, hammer the endpoint with C parallel worker threads, +each issuing the promql_suite in a tight loop for `--duration` seconds. +Records total queries served / wallclock = throughput, plus P50/P99 +per-query latency. + +Output: + benchmarks/reports/concurrency_sweep.csv +columns: backend, concurrency, total_queries, throughput_qps, p50_ms, p99_ms + +Usage: + python benchmarks/scripts/run_concurrency_sweep.py \ + [--asap-url URL] \ + [--prom-url URL] \ + [--vm-url URL] \ + [--duration 60] \ + [--concurrency 1,4,16,64] \ + [--backends asap,prom,vm] \ + [--output FILE] +""" + +import argparse +import csv +import json +import os +import threading +import time +import urllib.parse +from collections import deque +from concurrent.futures import ThreadPoolExecutor, as_completed + +import requests + +HERE = os.path.dirname(os.path.abspath(__file__)) +QUERIES_DIR = os.path.join(HERE, "..", "queries") +DEFAULT_OUTPUT = os.path.join(HERE, "..", "reports", "concurrency_sweep.csv") +DEFAULT_DURATION = 60 +DEFAULT_CONCURRENCIES = [1, 4, 16, 64] + + +def percentile(values, pct): + if not values: + return float("nan") + s = sorted(values) + n = len(s) + if n == 1: + return s[0] + idx = (pct / 100.0) * (n - 1) + lo = int(idx) + hi = min(lo + 1, n - 1) + frac = idx - lo + return s[lo] * (1 - frac) + s[hi] * frac + + +def hammer(url_base: str, queries: list[str], deadline: float, latencies: deque, lock: threading.Lock) -> int: + """One worker thread: round-robin through queries until deadline.""" + session = requests.Session() + count = 0 + i = 0 + n = len(queries) + while time.monotonic() < deadline: + expr = queries[i % n] + i += 1 + encoded = urllib.parse.quote(expr, safe="") + url = f"{url_base}/api/v1/query?query={encoded}" + t0 = time.monotonic() + try: + r = session.get(url, timeout=30) + lat = (time.monotonic() - t0) * 1000.0 + if r.status_code == 200: + with lock: + latencies.append(lat) + count += 1 + except Exception: # noqa: BLE001 + pass + return count + + +def sweep_one(backend: str, url: str, queries: list[str], concurrencies: list[int], duration: int) -> list[dict]: + rows = [] + for c in concurrencies: + print(f"[sweep] backend={backend} concurrency={c} duration={duration}s") + latencies: deque[float] = deque() + lock = threading.Lock() + deadline = time.monotonic() + duration + wallclock_start = time.monotonic() + with ThreadPoolExecutor(max_workers=c) as pool: + futures = [pool.submit(hammer, url, queries, deadline, latencies, lock) for _ in range(c)] + total = sum(f.result() for f in as_completed(futures)) + wallclock = time.monotonic() - wallclock_start + lat_list = list(latencies) + qps = total / wallclock if wallclock > 0 else 0.0 + p50 = percentile(lat_list, 50) + p99 = percentile(lat_list, 99) + print(f" -> total={total} qps={qps:.1f} p50={p50:.1f}ms p99={p99:.1f}ms") + rows.append({ + "backend": backend, + "concurrency": c, + "total_queries": total, + "throughput_qps": round(qps, 2), + "p50_ms": round(p50, 2) if p50 == p50 else "", + "p99_ms": round(p99, 2) if p99 == p99 else "", + }) + return rows + + +def main() -> None: + parser = argparse.ArgumentParser(description="Throughput-vs-concurrency sweep across all 3 backends") + parser.add_argument("--asap-url", default="http://localhost:8088") + parser.add_argument("--prom-url", default="http://localhost:9090") + parser.add_argument("--vm-url", default="http://localhost:8428") + parser.add_argument("--duration", type=int, default=DEFAULT_DURATION) + parser.add_argument( + "--concurrency", + default=",".join(str(c) for c in DEFAULT_CONCURRENCIES), + help="Comma-separated concurrency levels", + ) + parser.add_argument( + "--backends", + default="asap,prom,vm", + help="Comma-separated backends to sweep", + ) + parser.add_argument("--output", default=DEFAULT_OUTPUT) + args = parser.parse_args() + + concurrencies = [int(x) for x in args.concurrency.split(",") if x.strip()] + backends = [x.strip() for x in args.backends.split(",") if x.strip()] + + with open(os.path.join(QUERIES_DIR, "promql_suite.json")) as f: + suite = json.load(f) + queries = [q["expr"] for q in suite["queries"]] + + backend_urls = { + "asap": args.asap_url, + "prom": args.prom_url, + "vm": args.vm_url, + } + + all_rows = [] + for backend in backends: + if backend not in backend_urls: + print(f"[sweep] skipping unknown backend: {backend}") + continue + all_rows.extend(sweep_one(backend, backend_urls[backend], queries, concurrencies, args.duration)) + + os.makedirs(os.path.dirname(os.path.abspath(args.output)), exist_ok=True) + with open(args.output, "w", newline="") as f: + writer = csv.DictWriter( + f, + fieldnames=["backend", "concurrency", "total_queries", "throughput_qps", "p50_ms", "p99_ms"], + ) + writer.writeheader() + writer.writerows(all_rows) + print(f"\n[sweep] wrote {len(all_rows)} rows to {args.output}") + + +if __name__ == "__main__": + main() diff --git a/benchmarks/scripts/run_prom.py b/benchmarks/scripts/run_prom.py new file mode 100755 index 000000000..86efe8633 --- /dev/null +++ b/benchmarks/scripts/run_prom.py @@ -0,0 +1,58 @@ +#!/usr/bin/env python3 +"""run_prom.py — runs both W1 (promql_suite.json) and W2 (adhoc_suite.json) +against a Prometheus baseline. Output JSON shape matches run_asap_workloads.py +so compare_three_way.py can ingest it. + +Writes: + benchmarks/reports/prom_promql.json + benchmarks/reports/prom_adhoc.json + +Usage: + python benchmarks/scripts/run_prom.py \ + [--prometheus-url URL] \ + [--iterations N] \ + [--reports-dir DIR] +""" + +import argparse +import os + +from run_asap_workloads import ( # noqa: E402 -- intentional sibling-import + DEFAULT_ITERATIONS, + DEFAULT_REPORTS_DIR, + QUERIES_DIR, + run_suite, +) +import json + + +def main() -> None: + parser = argparse.ArgumentParser(description="Run W1+W2 against Prometheus baseline") + parser.add_argument("--prometheus-url", default="http://localhost:9090") + parser.add_argument("--iterations", type=int, default=DEFAULT_ITERATIONS) + parser.add_argument("--reports-dir", default=DEFAULT_REPORTS_DIR) + args = parser.parse_args() + + os.makedirs(args.reports_dir, exist_ok=True) + + for suite_name, out_name, label in [ + ("promql_suite.json", "prom_promql.json", "prom-w1"), + ("adhoc_suite.json", "prom_adhoc.json", "prom-w3"), + ]: + print("=" * 70) + print(f"[prom] {label} — {suite_name}") + print("=" * 70) + result = run_suite( + os.path.join(QUERIES_DIR, suite_name), + args.prometheus_url, args.iterations, label, + ) + # Tag the URL field for downstream tools that grep on it. + result["prometheus_url"] = result.pop("asap_url") + out_path = os.path.join(args.reports_dir, out_name) + with open(out_path, "w") as f: + json.dump(result, f, indent=2) + print(f"[prom] saved to {out_path}") + + +if __name__ == "__main__": + main() diff --git a/benchmarks/scripts/run_vm.py b/benchmarks/scripts/run_vm.py new file mode 100755 index 000000000..eebbae6a3 --- /dev/null +++ b/benchmarks/scripts/run_vm.py @@ -0,0 +1,58 @@ +#!/usr/bin/env python3 +"""run_vm.py — runs both W1 (promql_suite.json) and W2 (adhoc_suite.json) +against VictoriaMetrics. VM exposes a Prometheus-compatible /api/v1/query +endpoint on port 8428, so the same query_once helper from run_asap_workloads +works unmodified. + +Writes: + benchmarks/reports/vm_promql.json + benchmarks/reports/vm_adhoc.json + +Usage: + python benchmarks/scripts/run_vm.py \ + [--vm-url URL] \ + [--iterations N] \ + [--reports-dir DIR] +""" + +import argparse +import json +import os + +from run_asap_workloads import ( # noqa: E402 + DEFAULT_ITERATIONS, + DEFAULT_REPORTS_DIR, + QUERIES_DIR, + run_suite, +) + + +def main() -> None: + parser = argparse.ArgumentParser(description="Run W1+W2 against VictoriaMetrics") + parser.add_argument("--vm-url", default="http://localhost:8428") + parser.add_argument("--iterations", type=int, default=DEFAULT_ITERATIONS) + parser.add_argument("--reports-dir", default=DEFAULT_REPORTS_DIR) + args = parser.parse_args() + + os.makedirs(args.reports_dir, exist_ok=True) + + for suite_name, out_name, label in [ + ("promql_suite.json", "vm_promql.json", "vm-w1"), + ("adhoc_suite.json", "vm_adhoc.json", "vm-w3"), + ]: + print("=" * 70) + print(f"[vm] {label} — {suite_name}") + print("=" * 70) + result = run_suite( + os.path.join(QUERIES_DIR, suite_name), + args.vm_url, args.iterations, label, + ) + result["vm_url"] = result.pop("asap_url") + out_path = os.path.join(args.reports_dir, out_name) + with open(out_path, "w") as f: + json.dump(result, f, indent=2) + print(f"[vm] saved to {out_path}") + + +if __name__ == "__main__": + main() diff --git a/benchmarks/scripts/seed_cold_store.py b/benchmarks/scripts/seed_cold_store.py new file mode 100755 index 000000000..b448ea82e --- /dev/null +++ b/benchmarks/scripts/seed_cold_store.py @@ -0,0 +1,103 @@ +#!/usr/bin/env python3 +"""seed_cold_store.py — pre-populate benchmarks/cold-store/ with raw JSONL +samples in the layout the ASAP ColdFallback adapter expects. + +Layout (per asap-query-engine/src/drivers/query/fallback/cold_store/format.rs): + raw//YYYY/MM/DD/HH/part-NNNNNN.jsonl + +Each line is one RawSample: + {"ts_ms": , "labels": {"k":"v",...}, "value": } + +We seed enough data for the W2 ad-hoc queries to come back non-empty: +the queries hit `sensor_reading` with various label filters; we generate +N series across {region, service, host, pattern} that match what the +fake-exporters produce in the live path. + +Usage: + python benchmarks/scripts/seed_cold_store.py [--root DIR] [--hours 2] [--series 30] +""" + +import argparse +import json +import os +import time +from datetime import datetime, timezone + +HERE = os.path.dirname(os.path.abspath(__file__)) +DEFAULT_ROOT = os.path.join(HERE, "..", "cold-store") +PATTERNS = ["constant", "linear-up", "linear-down", "sine", "sine-noise", "step", "exp-up"] +REGIONS = ["region0", "region1", "region2"] +SERVICES = ["svc0", "svc1", "svc2", "svc3", "svc4"] +HOSTS = ["host0", "host1"] + + +def hour_dir(root: str, metric: str, ts_ms: int) -> str: + dt = datetime.fromtimestamp(ts_ms / 1000.0, tz=timezone.utc) + return os.path.join( + root, "raw", metric, + f"{dt.year:04d}", f"{dt.month:02d}", f"{dt.day:02d}", f"{dt.hour:02d}", + ) + + +def write_part(path: str, samples: list[dict], part_idx: int) -> None: + os.makedirs(path, exist_ok=True) + fname = os.path.join(path, f"part-{part_idx:06d}.jsonl") + with open(fname, "w") as f: + for s in samples: + f.write(json.dumps(s, sort_keys=True) + "\n") + + +def main() -> None: + parser = argparse.ArgumentParser(description="Seed local-FS cold-store with raw JSONL") + parser.add_argument("--root", default=DEFAULT_ROOT) + parser.add_argument("--metric", default="sensor_reading") + parser.add_argument("--hours", type=int, default=2, help="how many hour-buckets back from now to seed") + parser.add_argument("--samples-per-series-per-hour", type=int, default=60, + help="one sample per minute by default") + args = parser.parse_args() + + now_ms = int(time.time() * 1000) + hour_ms = 3600 * 1000 + + # Build the series cross-product. Bound it so we don't blow up the FS. + series = [] + for pattern in PATTERNS: + for region in REGIONS: + for service in SERVICES[:3]: # 3 services per region keeps fixture small + for host in HOSTS[:1]: + series.append({ + "pattern": pattern, "region": region, + "service": service, "host": host, + }) + + print(f"[seed] generating {len(series)} series x {args.hours} hours x " + f"{args.samples_per_series_per_hour} samples") + + total_lines = 0 + for h in range(args.hours): + bucket_start_ms = now_ms - (args.hours - h) * hour_ms + # Group samples by hour-bucket dir; within a bucket, write one part. + buckets: dict[str, list[dict]] = {} + step_ms = hour_ms // args.samples_per_series_per_hour + for i in range(args.samples_per_series_per_hour): + ts_ms = bucket_start_ms + i * step_ms + for idx, labels in enumerate(series): + # deterministic-but-varied value + v = float((idx * 7 + i * 3) % 1000) + sample = { + "ts_ms": ts_ms, + "labels": dict(sorted(labels.items())), + "value": v, + } + key = hour_dir(args.root, args.metric, ts_ms) + buckets.setdefault(key, []).append(sample) + + for path, samples in buckets.items(): + write_part(path, samples, part_idx=h) + total_lines += len(samples) + + print(f"[seed] wrote {total_lines} JSONL samples under {args.root}") + + +if __name__ == "__main__": + main()