ci: add a perf-movements reader for the benchmark pipelines

.agents/skills/README.md

@@ -27,6 +27,7 @@ explicitly with `$<skill-name>` in Codex and `/<skill-name>` in Claude Code.
 | Skill | When to use | What it does |
 |---|---|---|
 | **mz-benchmark** | Adding or debugging benchmark scenarios | Three measurement frameworks: Feature Benchmark (micro-benchmarks), Scalability Test (throughput under concurrency), Parallel Benchmark (sustained latency) |
+| **mz-perf-movements** | Asking what performance moved in CI | Ranks regressions and improvements across the nightly, release-qualification and spec-sheet pipelines, reading the self-comparing steps' own tables and reconstructing a baseline for the spec sheet |
 | **mz-profile** | Something is slow or using too much memory | CPU profiling with samply, memory profiling with heaptrack, binary size analysis |
 | **mz-query-tracing** | Understanding where time goes in SQL execution | Distributed tracing with OpenTelemetry/Tempo to get a latency breakdown of any SQL statement |
 

.agents/skills/mz-perf-movements/SKILL.md

@@ -0,0 +1,164 @@
+---
+name: mz-perf-movements
+description: >
+  Summarize performance movements in the nightly, release-qualification and
+  spec-sheet pipelines. Trigger: "what moved in nightly", "did perf regress",
+  "summarize the benchmark results", "any performance regressions this week",
+  "did my optimization show up in CI", "compare spec sheet builds", or a pasted
+  nightly / release-qualification / spec-sheet Buildkite URL asked about in
+  performance rather than failure terms. For a red build use mz-debug-ci; for
+  writing or changing a benchmark use mz-benchmark.
+argument-hint: "[pipeline] [--since N builds] [--only regressions|improvements]"
+---
+
+Report which benchmark numbers moved, in which direction, and by how much.
+
+This skill answers "what changed in performance", not "why is the build red".
+A failing benchmark job is often a gate trip that this skill will show as a
+large movement, but diagnosing a crashed or timed-out job belongs to
+mz-debug-ci.
+
+## The one thing to understand first
+
+The three pipelines produce two different kinds of evidence, and conflating
+them produces wrong answers.
+
+**Self-comparing steps** run both sides themselves. The feature benchmark,
+parallel benchmark and scalability benchmark each build the merge base or a
+named ancestor, run it alongside the change, and print a comparison table.
+For these, the movement is already computed in CI against a same-hardware,
+same-run baseline. Trust it.
+
+**Absolute-value steps** measure one build in isolation. The cluster spec
+sheet uploads raw per-repetition CSV rows and nothing else. A movement only
+exists once you compare against earlier builds yourself, which means the
+baseline is cross-run and carries hardware and scheduling noise that the
+self-comparing steps do not have. Treat a spec-sheet movement as weaker
+evidence than a feature-benchmark movement of the same size.
+
+## Running it
+
+```bash
+bin/pyactivate -m materialize.buildkite_insights.perf_movements.perf_movements
+```
+
+Useful flags:
+
+| Flag | Effect |
+|---|---|
+| `--pipeline nightly` | Repeatable. Defaults to all three pipelines. |
+| `--branch main` | Branch to read. Defaults to `main`. |
+| `--threshold 5.0` | Minimum movement in percent to report. |
+| `--only regressions` | Also `improvements`, or `all` (default). |
+| `--max-builds 6` | Builds considered when reconstructing a spec-sheet baseline. |
+| `--min-history 3` | Earlier observations a reconstructed baseline needs. |
+| `--format json` | Machine-readable rows instead of the table. |
+| `--urls` | List each contributing job URL after the table. |
+| `--fetch always` | Bypass the local cache. Default `auto` reuses data up to 96 hours old. |
+
+Output is one ranked table, deteriorations first, then improvements, each side
+ordered by magnitude. A row marked `WORSE!` is one where the step itself
+published a regression verdict; those are always reported regardless of
+`--threshold`.
+
+### Credentials
+
+Fetching from Buildkite needs `BUILDKITE_TOKEN` (or `BUILDKITE_CI_API_KEY`)
+with read access to builds and artifacts. Without one the tool exits 1 saying
+so, before any request. An expired or under-scoped token instead surfaces as the
+HTTP status of the first failing request.
+
+If no token is configured but the Buildkite MCP server is available, fetch the
+input through the MCP and feed it to the same parsers, so the numbers and the
+ranking stay identical. The MCP saves a job log as a JSON document
+(`{"entries": [{"c": ...}]}`) even though it announces plain text; pass that file
+as-is, the tool recognizes both that shape and raw text.
+
+```bash
+# a self-comparing step's job log
+bin/pyactivate -m materialize.buildkite_insights.perf_movements.perf_movements \
+  --from-log /path/to/job.log --step-key feature-benchmark
+
+# spec sheet, newest build against earlier ones
+bin/pyactivate -m materialize.buildkite_insights.perf_movements.perf_movements \
+  --from-csv latest.cluster.csv \
+  --baseline-csv build45.cluster.csv --baseline-csv build44.cluster.csv \
+  --min-history 2
+```
+
+## What each pipeline contributes
+
+| Pipeline | Step keys | Kind | Metrics |
+|---|---|---|---|
+| nightly | `feature-benchmark` | self-comparing | `wallclock`, `memory_mz`, `memory_clusterd` |
+| nightly | `parallel-benchmark` | self-comparing | `qps`, `queries`, `avg`, `min`, `max`, `std`, `p50` through `p99_999999` |
+| nightly | `scalability-benchmark-dml-dql`, `-ddl`, `-connection` | self-comparing | `tps` per concurrency |
+| release-qualification | `feature-benchmark-scale-plus-one` | self-comparing | as above, at `--scale=+1` |
+| release-qualification | `long-parallel-benchmark` | self-comparing | as above |
+| spec-sheet | `cluster-spec-sheet-cluster`, `-source-ingestion`, `-staging` | absolute | `time_ms`, `size_bytes`, `qps` |
+
+`feature-benchmark` runs as 12 parallel shards and `parallel-benchmark` as 5;
+every shard is read, since each covers different scenarios.
+
+## Reading the numbers
+
+**Direction is not the sign.** A parallel-benchmark `qps` rising is good and an
+`avg` rising is bad; both print as a positive change. The tool resolves this
+per metric, but a human reading a raw job log must not.
+
+**Ungated statistics have no verdict.** The parallel benchmark prints `max`,
+`min`, `p99` and the deeper percentiles with empty `THRESHOLD` and `REGRESSION?`
+cells. They are real measurements and are reported, but CI never fails on them,
+and `max` in particular is a single worst sample.
+
+**`slope` is not reported.** The parallel benchmark's `slope` is a
+millisecond-per-second drift, so its ratio against a baseline near zero carries
+no meaning. `queries` is reported and counts as higher-is-better, being the
+number of queries the load phase completed.
+
+**The step's threshold is a build gate, not a noise floor.** Feature-benchmark
+thresholds run 10% to 50%. A 6% wallclock movement passes CI and is still worth
+knowing about, which is why `--threshold` is the caller's floor and the step's
+threshold is shown as context only.
+
+**Reruns overwrite.** The feature benchmark reruns scenarios that regressed, so
+a scenario appears in several tables in one log. The last table it appears in
+is the runner's settled result; that is the one reported.
+
+**A single spec-sheet build is not a trend.** With `--min-history 3` a key needs
+three earlier observations before it is reported at all, and the baseline is the
+median of those builds, so one outlier build cannot manufacture a movement.
+
+## Steps this skill does not cover
+
+`limits`, `limits-instance-size`, `bounded-memory`, `bounded-memory-search`,
+`orchestratord-rolling-upgrade-downtime` and `cargo-bench` publish neither a
+per-metric comparison nor per-repetition CSV rows, so no movement can be derived
+from Buildkite alone. The spec sheet's `.cluster_object_limits.csv` is also
+excluded: its headline number is a maximum healthy object count derived from the
+`healthy` and `failure_mode` columns, which is a different aggregation than the
+median used for the other streams. Say so when asked about these rather than
+substituting a number from elsewhere.
+
+Richer history lives in the `test_analytics` database
+(`raw.test_analytics`, tables `feature_benchmark_result`,
+`cluster_spec_sheet_result` and siblings, schemas in
+`misc/python/materialize/test_analytics/setup/tables/`). Reading it needs a
+Materialize app password for that region, which is a separate credential from
+any Buildkite or CI dashboard token.
+
+## Verifying a change to the parsers
+
+The parsers depend on print statements in
+`misc/python/materialize/feature_benchmark/report.py`,
+`test/parallel-benchmark/mzcompose.py` and
+`misc/python/materialize/scalability/result/scalability_change.py`. When one of
+those changes, the parser goes quiet rather than loud, so the fixtures are
+copied verbatim from a real build:
+
+```bash
+bin/pytest misc/python/materialize/buildkite_insights/perf_movements/perf_movements_test.py
+```
+
+An empty report from a build you know produced benchmark output means a printer
+changed. Re-capture the fixture from the live log before adjusting a regex.

misc/python/materialize/buildkite_insights/perf_movements/collect.py

@@ -0,0 +1,198 @@
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+"""Fetches build data from Buildkite and turns it into movements.
+
+Every fetch goes through the buildkite_insights cache, so repeated runs over the same
+window cost nothing beyond the newest build.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import Any
+
+from materialize.buildkite_insights.cache import artifacts_cache, builds_cache
+from materialize.buildkite_insights.cache.cache_constants import FetchMode
+from materialize.buildkite_insights.cache.logs_cache import get_or_download_log
+from materialize.buildkite_insights.perf_movements import spec_sheet
+from materialize.buildkite_insights.perf_movements.log_parsers import (
+    parse_feature_benchmark_log,
+    parse_parallel_benchmark_log,
+    parse_scalability_log,
+)
+from materialize.buildkite_insights.perf_movements.movement import JobRef, Movement
+
+LogParser = Callable[[str, JobRef], list[Movement]]
+
+LOG_PARSER_BY_STEP_KEY: dict[str, LogParser] = {
+    "feature-benchmark": parse_feature_benchmark_log,
+    "feature-benchmark-scale-plus-one": parse_feature_benchmark_log,
+    "parallel-benchmark": parse_parallel_benchmark_log,
+    "long-parallel-benchmark": parse_parallel_benchmark_log,
+    "scalability-benchmark-dml-dql": parse_scalability_log,
+    "scalability-benchmark-ddl": parse_scalability_log,
+    "scalability-benchmark-connection": parse_scalability_log,
+}
+
+SPEC_SHEET_STEP_KEYS = frozenset(
+    {
+        "cluster-spec-sheet-cluster",
+        "cluster-spec-sheet-source-ingestion",
+        "cluster-spec-sheet-staging",
+    }
+)
+
+# Steps whose output carries no per-metric comparison and no per-repetition CSV:
+# `limits`, `limits-instance-size`, `bounded-memory`, `bounded-memory-search`,
+# `orchestratord-rolling-upgrade-downtime` and `cargo-bench`. They are not collected.
+PIPELINES = ("nightly", "release-qualification", "spec-sheet")
+
+
+def _jobs_with_step_keys(build: Any, step_keys: frozenset[str]) -> list[dict[str, Any]]:
+    return [
+        job
+        for job in build.get("jobs", [])
+        if job.get("step_key") in step_keys and job.get("id")
+    ]
+
+
+def _job_ref(pipeline_slug: str, build: Any, job: dict[str, Any]) -> JobRef:
+    return JobRef(
+        pipeline_slug=pipeline_slug,
+        build_number=build["number"],
+        job_id=job["id"],
+        step_key=job["step_key"],
+        commit_hash=build.get("commit"),
+    )
+
+
+def fetch_builds(
+    pipeline_slug: str,
+    branch: str,
+    fetch_mode: FetchMode,
+    max_builds: int,
+) -> list[Any]:
+    """Newest builds first, restricted to `branch`."""
+    builds = builds_cache.get_or_query_builds(
+        pipeline_slug=pipeline_slug,
+        fetch_mode=fetch_mode,
+        max_fetches=1,
+        branch=branch,
+        build_states=None,
+    )
+    return builds[:max_builds]
+
+
+def collect_log_movements(
+    pipeline_slug: str,
+    build: Any,
+    fetch_mode: FetchMode,
+) -> list[Movement]:
+    """Movements from every self-comparing step of a single build."""
+    movements = []
+
+    for job in _jobs_with_step_keys(build, frozenset(LOG_PARSER_BY_STEP_KEY)):
+        parser = LOG_PARSER_BY_STEP_KEY[job["step_key"]]
+        log = get_or_download_log(
+            pipeline_slug=pipeline_slug,
+            fetch_mode=fetch_mode,
+            build_number=build["number"],
+            job_id=job["id"],
+        )
+        movements.extend(parser(log, _job_ref(pipeline_slug, build, job)))
+
+    return movements
+
+
+def _collect_spec_sheet_medians(
+    pipeline_slug: str,
+    build: Any,
+    fetch_mode: FetchMode,
+) -> dict[tuple[spec_sheet.SampleKey, str], float]:
+    per_file = []
+
+    for job in _jobs_with_step_keys(build, SPEC_SHEET_STEP_KEYS):
+        artifacts = artifacts_cache.get_or_query_job_artifact_list(
+            pipeline_slug=pipeline_slug,
+            fetch_mode=fetch_mode,
+            build_number=build["number"],
+            job_id=job["id"],
+        )
+        for artifact in artifacts:
+            filename = artifact.get("filename", "")
+            if not filename.endswith(spec_sheet.RESULT_FILE_SUFFIXES):
+                continue
+            content = artifacts_cache.get_or_download_artifact(
+                pipeline_slug=pipeline_slug,
+                fetch_mode=fetch_mode,
+                build_number=build["number"],
+                job_id=job["id"],
+                artifact_id=artifact["id"],
+                is_zst_compressed=False,
+            )
+            per_file.append(
+                spec_sheet.median_per_key(spec_sheet.parse_results_csv(content))
+            )
+
+    return spec_sheet.merge_medians(per_file)
+
+
+def collect_spec_sheet_movements(
+    pipeline_slug: str,
+    builds: list[Any],
+    fetch_mode: FetchMode,
+    min_history: int,
+) -> list[Movement]:
+    """Compare the newest spec sheet build against the builds behind it.
+
+    `builds` must be newest first. The reported job reference points at the first spec
+    sheet job of the newest build, since the movement is a property of that whole build
+    rather than of any single shard.
+    """
+    if not builds:
+        return []
+
+    latest_build = builds[0]
+    latest_jobs = _jobs_with_step_keys(latest_build, SPEC_SHEET_STEP_KEYS)
+    if not latest_jobs:
+        return []
+
+    latest = _collect_spec_sheet_medians(pipeline_slug, latest_build, fetch_mode)
+    history = [
+        _collect_spec_sheet_medians(pipeline_slug, build, fetch_mode)
+        for build in builds[1:]
+    ]
+
+    return spec_sheet.compare_to_window(
+        latest=latest,
+        history=history,
+        job=_job_ref(pipeline_slug, latest_build, latest_jobs[0]),
+        min_history=min_history,
+    )
+
+
+def collect_pipeline_movements(
+    pipeline_slug: str,
+    branch: str,
+    fetch_mode: FetchMode,
+    max_builds: int,
+    min_history: int,
+) -> list[Movement]:
+    """All movements for one pipeline's newest build on `branch`."""
+    builds = fetch_builds(pipeline_slug, branch, fetch_mode, max_builds)
+    if not builds:
+        return []
+
+    if pipeline_slug == "spec-sheet":
+        return collect_spec_sheet_movements(
+            pipeline_slug, builds, fetch_mode, min_history
+        )
+
+    return collect_log_movements(pipeline_slug, builds[0], fetch_mode)