diff --git a/.github/workflows/asap-sketch-benchmark.yml b/.github/workflows/asap-sketch-benchmark.yml index 33cf9f1..f3b0ec1 100644 --- a/.github/workflows/asap-sketch-benchmark.yml +++ b/.github/workflows/asap-sketch-benchmark.yml @@ -2,6 +2,10 @@ name: ASAP sketch benchmark on: workflow_dispatch: + inputs: + regression_margin: + description: "Fail when median signals/s drops more than this fraction" + default: "0.10" pull_request: paths: - ".github/workflows/asap-sketch-benchmark.yml" @@ -12,6 +16,7 @@ on: permissions: contents: read + actions: read # download the previous nightly's artifact concurrency: group: asap-sketch-nightly @@ -19,7 +24,9 @@ concurrency: jobs: benchmark: - runs-on: ubuntu-24.04 + # A registered self-hosted runner (repository variable BENCH_RUNNER, its + # label) gives exclusive cores; otherwise fall back to a shared runner. + runs-on: ${{ vars.BENCH_RUNNER || 'ubuntu-24.04' }} timeout-minutes: 240 steps: - uses: actions/checkout@v4 @@ -33,7 +40,7 @@ jobs: - name: Test streaming window correctness run: cargo test --manifest-path asap-precompute-rs/Cargo.toml --release --features stream-benchmark --bin asap-stream-bench - name: Run persistent OTLP network benchmark - run: ./benchmarks/run-nightly.sh --allow-shared-infrastructure + run: ./benchmarks/run-nightly.sh ${{ !vars.BENCH_RUNNER && '--allow-shared-infrastructure' || '' }} - name: Verify streaming measurements run: | python3 - <<'PYCODE' @@ -43,9 +50,9 @@ jobs: runs = json.loads((root / "runs.json").read_text()) summary = json.loads((root / "summary.json").read_text()) capacity = json.loads((root / "capacity.json").read_text()) - assert len(runs) == 189 - assert len(summary) == 63 - assert len(json.loads((root / "resources.json").read_text())) == 441 + assert len(runs) == 315 + assert len(summary) == 105 + assert len(json.loads((root / "resources.json").read_text())) == 735 assert len(capacity) == 3 assert all(row["repetitions"] == 3 for row in summary) assert all("sustainable" in row for row in summary) @@ -69,6 +76,25 @@ jobs: assert (root / "throughput.svg").is_file() assert (root / "resources.svg").is_file() PYCODE + - name: Compare with previous nightly + if: github.event_name != 'pull_request' + env: + GH_TOKEN: ${{ github.token }} + MARGIN: ${{ inputs.regression_margin || '0.10' }} + run: | + previous=$(gh run list --repo "$GITHUB_REPOSITORY" --workflow asap-sketch-benchmark.yml \ + --branch main --status success --limit 1 --json databaseId --jq '.[0].databaseId // empty') + if [ -z "$previous" ]; then echo "No previous successful nightly; skipping."; exit 0; fi + gh run download "$previous" --repo "$GITHUB_REPOSITORY" --name "asap-sketch-benchmark-$previous" --dir "$RUNNER_TEMP/previous" + status=0 + python3 benchmarks/compare-summaries.py "$RUNNER_TEMP/previous/streaming/summary.csv" \ + benchmark-results/streaming/summary.csv --margin "$MARGIN" > comparison.md || status=$? + { echo "### Throughput vs nightly run $previous"; echo; cat comparison.md; } >> "$GITHUB_STEP_SUMMARY" + cat comparison.md + if [ "$status" -ne 0 ]; then + echo "::error title=Benchmark regression::median signals/s dropped more than $MARGIN vs run $previous (see job summary)" + fi + exit "$status" - name: Upload review artifacts if: always() uses: actions/upload-artifact@v4 diff --git a/.github/workflows/benchmark-on-demand.yml b/.github/workflows/benchmark-on-demand.yml new file mode 100644 index 0000000..bb40703 --- /dev/null +++ b/.github/workflows/benchmark-on-demand.yml @@ -0,0 +1,83 @@ +name: Benchmark on demand + +# Comment `/benchmark` on a pull request (write access required) to run a +# reduced streaming sweep on main and on the PR head and post the comparison. + +on: + issue_comment: + types: [created] + +permissions: + contents: read + +concurrency: + group: asap-sketch-nightly # share the benchmark runner with the nightly + cancel-in-progress: false + +jobs: + benchmark: + if: >- + github.event.issue.pull_request && + startsWith(github.event.comment.body, '/benchmark') && + contains(fromJSON('["OWNER","MEMBER","COLLABORATOR"]'), github.event.comment.author_association) + runs-on: ${{ vars.BENCH_RUNNER || 'ubuntu-24.04' }} + timeout-minutes: 120 + outputs: + comparison: ${{ steps.compare.outputs.comparison }} + env: + SWEEP: >- + --traffic-rates 25000 100000 --window-points 65536 --repetitions 1 + ${{ !vars.BENCH_RUNNER && '--allow-shared-infrastructure' || '' }} + steps: + - uses: actions/checkout@v4 + with: + ref: main + persist-credentials: false + - uses: actions/setup-python@v5 + with: + python-version: "3.13" + - uses: dtolnay/rust-toolchain@stable + - uses: Swatinem/rust-cache@v2.7.8 + with: + workspaces: asap-precompute-rs + - name: Sweep main + run: | + ./benchmarks/run-nightly.sh $SWEEP + cp benchmark-results/streaming/summary.csv "$RUNNER_TEMP/base.csv" + cp benchmarks/compare-summaries.py "$RUNNER_TEMP/" + rm -rf benchmark-results + # PR code runs only in this job, which holds a read-only token. + - uses: actions/checkout@v4 + with: + ref: refs/pull/${{ github.event.issue.number }}/head + persist-credentials: false + clean: false # keep the cargo target directory + - name: Sweep PR head + run: ./benchmarks/run-nightly.sh $SWEEP + - name: Compare + id: compare + run: | + { + echo "### \`/benchmark\`: PR head (\`$(git rev-parse --short HEAD)\`) vs main" + echo + python3 "$RUNNER_TEMP/compare-summaries.py" "$RUNNER_TEMP/base.csv" \ + benchmark-results/streaming/summary.csv --margin 0.10 || true + echo + echo "Reduced sweep, 1 repetition. Run: $GITHUB_SERVER_URL/$GITHUB_REPOSITORY/actions/runs/$GITHUB_RUN_ID" + } > comparison.md + cat comparison.md >> "$GITHUB_STEP_SUMMARY" + { echo "comparison<> "$GITHUB_OUTPUT" + + comment: + needs: benchmark + runs-on: ubuntu-24.04 + permissions: + pull-requests: write + steps: + - name: Post comparison + env: + GH_TOKEN: ${{ github.token }} + BODY: ${{ needs.benchmark.outputs.comparison }} + run: >- + gh api "repos/$GITHUB_REPOSITORY/issues/${{ github.event.issue.number }}/comments" + -f body="$BODY" diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..7b056ae --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,84 @@ +name: CI + +# Per-merge tier: lint, unit/integration tests, and an end-to-end acceptance +# smoke on real collector containers. Capacity is measured by the nightly +# (asap-sketch-benchmark.yml) and on-demand (benchmark-on-demand.yml) tiers. + +on: + pull_request: + push: + branches: [main] + +permissions: + contents: read + +concurrency: + group: ci-${{ github.ref }} + cancel-in-progress: true + +jobs: + checks: + runs-on: ubuntu-24.04 + defaults: + run: + working-directory: asap-precompute-rs + steps: + - uses: actions/checkout@v4 + # Same toolchain setup as the nightly benchmark; ubuntu-24.04 already + # ships the cmake/clang the otel-arrow dependencies need. + - uses: dtolnay/rust-toolchain@stable + with: + components: rustfmt, clippy + - uses: Swatinem/rust-cache@v2.7.8 + with: + workspaces: asap-precompute-rs + - run: cargo fmt --check + - run: cargo clippy --all-targets --features otap-engine,stream-benchmark -- -D warnings + - run: cargo test --features otap-engine,stream-benchmark + + acceptance: + needs: checks + runs-on: ubuntu-24.04 + timeout-minutes: 60 + # Correctness smoke, not a capacity measurement: every stage shares the + # runner's 4 vCPUs instead of getting its own core. + env: + DOCKER_COMMAND: docker + CPU_GENERATOR_A: "0-3" + CPU_GENERATOR_B: "0-3" + CPU_BRANCH_A: "0-3" + CPU_BRANCH_B: "0-3" + CPU_MERGE: "0-3" + CPU_DETECTOR: "0-3" + CPU_EXACT_GENERATOR_A: "0-3" + CPU_EXACT_GENERATOR_B: "0-3" + CPU_EXACT_BRANCH_A: "0-3" + CPU_EXACT_BRANCH_B: "0-3" + CPU_EXACT_MERGE: "0-3" + CPU_EXACT_DETECTOR: "0-3" + CPU_RAW_GENERATOR_A: "0-3" + CPU_RAW_GENERATOR_B: "0-3" + CPU_RAW_BRANCH_A: "0-3" + CPU_RAW_BRANCH_B: "0-3" + CPU_RAW_MERGE: "0-3" + CPU_RAW_SINK: "0-3" + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.13" + - uses: dtolnay/rust-toolchain@stable + - uses: Swatinem/rust-cache@v2.7.8 + with: + workspaces: asap-precompute-rs + - name: Build collector image + run: deploy/telemetry-functions/run.sh --build-only + - name: Run acceptance use cases + run: python3 deploy/telemetry-functions/acceptance.py --report acceptance.json + - name: Upload acceptance report + if: always() + uses: actions/upload-artifact@v4 + with: + name: acceptance-${{ github.run_id }} + path: acceptance.json + if-no-files-found: warn diff --git a/asap-precompute-rs/src/otap/answer_nodes.rs b/asap-precompute-rs/src/otap/answer_nodes.rs index 5056c6b..2000ad5 100644 --- a/asap-precompute-rs/src/otap/answer_nodes.rs +++ b/asap-precompute-rs/src/otap/answer_nodes.rs @@ -187,7 +187,10 @@ fn events_pdata(events: &[Event]) -> Result { .observed_time_unix_nano(e.window_end_ms * 1_000_000) .severity_number(SeverityNumber::Info) .severity_text("INFO") - .body(AnyValue::new_string(e.body.clone())) + // The method is part of the line: Loki drops a second entry with + // the same stream, timestamp and line, and sketch and exact events + // can otherwise read the same. + .body(AnyValue::new_string(format!("[{}] {}", e.method, e.body))) .attributes(attributes) .finish() }) diff --git a/asap-precompute-rs/src/otap/processor.rs b/asap-precompute-rs/src/otap/processor.rs index 0adaf84..e9dd3b2 100644 --- a/asap-precompute-rs/src/otap/processor.rs +++ b/asap-precompute-rs/src/otap/processor.rs @@ -216,6 +216,9 @@ const fn default_transmit_sketch() -> bool { true } +/// How long after a tumbling window's end the idle timer waits before closing it. +const IDLE_CLOSE_GRACE: Duration = Duration::from_secs(5); + /// Tumbling windows close within a second of their end even when idle; /// sliding windows emit once per slide. fn timer_period(config: &PluginConfig) -> Duration { @@ -505,7 +508,15 @@ impl local::Processor for AsapSketchesProcessor { Ok(()) } Message::Control(NodeControlMsg::TimerTick { .. }) => { + // Data crossing a boundary closes a tumbling window on time; the + // timer only closes windows of idle streams, so it waits a grace + // for points still queued upstream (e.g. on a contended host). let now_ms = asap_wall_clock_ms(); + let now_ms = if self.plugin_config.window_slide.is_zero() { + now_ms.saturating_sub(IDLE_CLOSE_GRACE.as_millis() as u64) + } else { + now_ms + }; let envs = self.precompute.tick(now_ms); self.emit_envelopes(&envs, effect_handler).await } diff --git a/benchmarks/README.md b/benchmarks/README.md index 3059375..5f73b5a 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -11,7 +11,27 @@ profiling. ./benchmarks/run-nightly.sh ``` -## Viewing nightly results today +## CI tiers + +| Tier | Workflow | Trigger | What it checks | +|---|---|---|---| +| Per-merge smoke | `ci.yml` | every PR and push to `main` | `cargo fmt`, `clippy -D warnings`, `cargo test`, then `deploy/telemetry-functions/acceptance.py` on real collector containers. All stages share the runner's 4 vCPUs, so this is a correctness smoke, not a capacity measurement. | +| Nightly sweep | `asap-sketch-benchmark.yml` | 02:30 UTC daily, manual | Full `run-nightly.sh` sweep, then `compare-summaries.py` against the previous successful `main` run. The step fails (with an annotation and the table in the job summary) when any row's median signals/s drops more than `regression_margin` (default 10%). | +| On-demand comparison | `benchmark-on-demand.yml` | PR comment starting with `/benchmark` from an owner, member or collaborator | Reduced sweep (25k and 100k offered/s/source, 65,536-point windows, 1 repetition) on `main` and on the PR head; the comparison table is posted as a PR comment. | + +The benchmark tiers run on the runner named by the repository variable +`BENCH_RUNNER` (a self-hosted runner label, which needs Docker, `gh` and +enough cores for exclusive placement) and fall back to `ubuntu-24.04` with +`--allow-shared-infrastructure` when it is unset. Shared runners are noisy, so +treat a regression flagged there as a prompt to re-run, not a verdict. + +Compare any two runs locally: + +```sh +benchmarks/compare-summaries.py BASE/streaming/summary.csv NEW/streaming/summary.csv --margin 0.10 +``` + +## Viewing nightly results There is not yet a published ASAP dashboard. Results are available as GitHub Actions artifacts: @@ -22,9 +42,8 @@ Actions artifacts: 4. Read `streaming/summary.json`, `streaming/capacity.json`, and the `throughput.svg` / `resources.svg` charts. -Artifacts expire after 30 days. GitHub-hosted runners are shared -infrastructure, so small differences between runs should not be treated as -regressions without repeated measurements. +Artifacts expire after 30 days, so the regression check compares against the +previous successful run rather than a long history. ## Future dashboard publication @@ -46,11 +65,10 @@ Each dashboard point should link to its commit and Actions run and record the benchmark schema version, OTAP revision, runner type, scenario parameters, and observation duration so incompatible runs are not plotted as one series. -The first dashboard should remain informational and require manual review. -Automated regression thresholds should be introduced only after enough -scheduled runs establish normal GitHub-hosted-runner variance. If stable -cross-project numbers are required later, the same summary format can be fed by -an isolated runner or an upstream OTAP shared-server contribution. +The dashboard should remain informational; the nightly regression check +above is the only automated threshold. If stable cross-project numbers are +required later, the same summary format can be fed by an upstream OTAP +shared-server contribution. ## Upstream references diff --git a/benchmarks/compare-summaries.py b/benchmarks/compare-summaries.py new file mode 100755 index 0000000..382f5bf --- /dev/null +++ b/benchmarks/compare-summaries.py @@ -0,0 +1,45 @@ +#!/usr/bin/env python3 +"""Compare two run-streaming.py summary.csv files. + +Prints a Markdown table of median_signals_per_second for the rows present in +both (same scenario, offered rate, window and egress cap) and exits 1 when any +row dropped by more than --margin (a fraction; 0.10 = 10%). + + benchmarks/compare-summaries.py BASE.csv NEW.csv --margin 0.10 +""" +import argparse +import csv +import sys + +KEY = ("scenario", "target_signals_per_second_per_source", "window_points_per_source", "branch_egress_mbit_per_second") + + +def load(path): + with open(path, newline="") as f: + return {tuple(row.get(k, "") for k in KEY): float(row["median_signals_per_second"]) for row in csv.DictReader(f)} + + +def main(): + parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + parser.add_argument("base") + parser.add_argument("new") + parser.add_argument("--margin", type=float, default=0.10) + args = parser.parse_args() + base, new = load(args.base), load(args.new) + print("| scenario | offered/s/source | window | base signals/s | new signals/s | change |") + print("|---|---:|---:|---:|---:|---:|") + regressed = 0 + for key in (k for k in base if k in new): + scenario, offered, window, _ = key + change = new[key] / base[key] - 1 if base[key] else 0.0 + flag = "" + if change < -args.margin: + regressed += 1 + flag = " **regression**" + print(f"| {scenario} | {int(float(offered)):,} | {int(float(window)):,} | {base[key]:,.0f} | {new[key]:,.0f} | {change:+.1%}{flag} |") + print(f"\n{regressed} of {len(base.keys() & new.keys())} rows dropped more than {args.margin:.0%}.") + sys.exit(1 if regressed else 0) + + +if __name__ == "__main__": + main() diff --git a/deploy/telemetry-functions/acceptance.py b/deploy/telemetry-functions/acceptance.py index 2ae6ac4..db61e40 100755 --- a/deploy/telemetry-functions/acceptance.py +++ b/deploy/telemetry-functions/acceptance.py @@ -9,7 +9,7 @@ fails. Time is scaled down 6x so a use case takes a few minutes: 10-second windows, a -2-second merge grace, changes after 1/6 of their configured time and growth 6x +5-second merge grace, changes after 1/6 of their configured time and growth 6x faster per minute. Rules counted in windows (baselines, thresholds) are as deployed. @@ -50,7 +50,8 @@ def scaled_copy(directory, into): text = src.read_text() text = text.replace("window_size: 60s", f"window_size: {WINDOW_S}s") text = text.replace("window: 60s", f"window: {WINDOW_S}s") - text = text.replace("{grace: 5s}", "{grace: 2s}") + # The merge grace is not scaled: shared CI runners need the full 5 s + # for every source's sketch to arrive. text = re.sub(r"after_seconds: (\d+)", lambda m: f"after_seconds: {int(m[1]) // SCALE}", text) text = re.sub(r"growth_per_minute: ([0-9.]+)", lambda m: f"growth_per_minute: {float(m[1]) * SCALE:g}", text) @@ -126,9 +127,9 @@ def common_checks(stack, flt): yield check("full rate reaches the end (no backpressure)", full and all(abs(v - expected) <= 0.01 * expected for v in full), f"{len(full)} windows, points/window {sorted(set(int(v) for v in full))} vs {expected}") - yield check("sketch and exact windows hold the same points", - shared and all(w["sketch"][t] == w["exact"][t] for t in shared), - f"{len(shared)} windows compared") + diff = [(t, int(w["sketch"][t]), int(w["exact"][t])) for t in shared if w["sketch"][t] != w["exact"][t]] + yield check("sketch and exact windows hold the same points", shared and not diff, + f"{len(shared)} windows compared" + (f"; differ (window end, sketch, exact): {diff}" if diff else "")) # The raw pipeline writes every point to its own store; whether it keeps # up is the measurement, so its rate is reported, not required. raw = [int(w["raw"][t]) for t in sorted(w["raw"])[1:-1]] @@ -204,8 +205,11 @@ def topk_checks(stack, t): recall = len(top["sketch"] & top["exact"]) / max(1, len(top["exact"])) yield check("sketch top-10 recall vs exact >= 0.9", recall >= 0.9, f"recall {recall:.2f}") events = stack.events('| event_kind="novelty" | method="sketch"') + flagged = stack.vector('max_over_time(asap_topk_new{key="/new/hot-route"}[10m])') yield check("novelty event in Loki", any("/new/hot-route" in e for e in events), - next((e for e in events if "/new/hot-route" in e), "none")) + next((e for e in events if "/new/hot-route" in e), + f"none; {len(events)} sketch novelty events, " + f"asap_topk_new gauge from {sorted(m['method'] for m, _ in flagged)}")) CASES = {