Add CI tiers: per-PR smoke with acceptance, nightly regression, on-demand comparison

.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,14 +16,17 @@ on:
 
 permissions:
   contents: read
+  actions: read # download the previous nightly's artifact
 
 concurrency:
   group: asap-sketch-nightly
   cancel-in-progress: false
 
 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

.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<<EOF_COMPARISON"; cat comparison.md; echo "EOF_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"

.github/workflows/ci.yml

@@ -0,0 +1,73 @@
+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: "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

asap-precompute-rs/src/otap/answer_nodes.rs

@@ -232,7 +232,10 @@ fn events_pdata(events: &[Event]) -> Result<OtapPdata, String> {
                 .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()
         })

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<OtapPdata> 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
             }

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
 

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()