From 8eeee93abfbdbd04d1bbfe4432ac6995ba07db88 Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 5 Sep 2026 12:53:47 +0800 Subject: [PATCH] test(scanner): add bounded ABBA validation harness Refs rustfs/backlog#2266 and rustfs/backlog#2240. Co-Authored-By: heihutu Co-Authored-By: zhi22915 --- .config/make/tests.mak | 1 + docs/operations/scanner-benchmark-runbook.md | 136 ++++++++ scripts/README.md | 2 + scripts/run_scanner_validation_harness.sh | 7 + scripts/scanner_abba.py | 329 +++++++++++++++++++ scripts/test_scanner_abba.py | 181 ++++++++++ scripts/test_scanner_validation_harness.sh | 2 + 7 files changed, 658 insertions(+) create mode 100644 scripts/scanner_abba.py create mode 100755 scripts/test_scanner_abba.py diff --git a/.config/make/tests.mak b/.config/make/tests.mak index 3ee3337fa..1ea7fcdd9 100644 --- a/.config/make/tests.mak +++ b/.config/make/tests.mak @@ -31,6 +31,7 @@ script-tests: ## Run shell script tests ./scripts/test_object_batch_bench_enhanced.sh ./scripts/test_hotpath_warp_ab_gate.sh ./scripts/test_hotpath_warp_abba.sh + ./scripts/test_scanner_validation_harness.sh ./scripts/test_exact_1mib_handoff_abba.sh ./scripts/test_pinned_paired_abba_bench.sh ./scripts/test_manual_transition_runbooks.sh diff --git a/docs/operations/scanner-benchmark-runbook.md b/docs/operations/scanner-benchmark-runbook.md index 83e9315a2..cedfc4966 100644 --- a/docs/operations/scanner-benchmark-runbook.md +++ b/docs/operations/scanner-benchmark-runbook.md @@ -34,6 +34,142 @@ The `scanner` and `heal` subsystems are served by `GetConfigKVHandler` (`rustfs/ ## Test Matrix +### Formal Scanner/Heal ABBA + +The `--abba` mode runs five independent scenario cells: `cold-hot`, `fresh-hot`, +`multi-hot-new`, `running-heal`, and `mrf-replay`. Each scenario runs at least +three A1/B1/B2/A2 groups for both baseline/candidate with background work on, +and candidate-only background off/on. A measured leg lasts at least 900 +seconds; the minimum matrix contains 120 legs (30 hours before setup/oracles). +The existing `performance-ab.yml` supplies the pattern for immutable build +provenance and failure propagation, but its short Warp workload is not this +scanner gate. No scheduled workflow starts this matrix automatically. + +```bash +scripts/run_scanner_validation_harness.sh --abba \ + --manifest scanner-abba.json --adapter /path/to/isolated-deployment-adapter \ + --out-dir /path/to/new-artifacts --data-root /path/to/new-test-data +``` + +Both roots must be new and non-overlapping. Every leg receives a unique data +directory. The runner checks disk capacity before each leg, never removes data, +and stops the adapter after success or failure. Retain raw artifacts and inspect +task ownership before removing any test data. The operator must reserve the +target machines and map the assigned directory to separate data paths on every +node; the runner cannot prove remote isolation from local path names. + +The manifest has the following JSON contract (all fields are required): + +| Field | Value | +|---|---| +| `schema`, `evidence` | `1`, and `measured` or `synthetic`. | +| `rounds`, `duration_seconds`, `min_free_bytes` | 3..10 groups, 900..86400 seconds for measured runs, and the independently estimated free-space reservation in bytes. Synthetic runs may use 1 second. | +| `baseline`, `candidate` | Each contains executable `binary`, full 40-character `revision`, and verified `sha256`. The runner rehashes binaries before every leg. | +| `fixed` | `config_sha256`, `dataset_sha256`, `release_flags`, `durability`, `disk_type`, `cache_state`, `load_command`, `resource_isolation`, `topology` (`EC8+4`), and positive `offered_load_ops`. Hashes use 64 lowercase hexadecimal characters. | +| `oracles` | A map with all five scenario names. Each value contains positive integer `objects`, `versions`, `bytes`, and `sha256` of the independently prepared canonical object/version/content manifest. | +| `expected_healed_objects` | A map with all five scenario names and independently seeded repair counts. Running-heal and MRF-replay require a positive count. | + +Record exact build flags and effective durability settings, not just defaults. +Use deterministic workload seeds so every isolated leg has the same expected +object/version/content result. Fix the foreground arrival rate (offered load), +cache preparation procedure, configuration, and hardware across every leg. +Do not include credentials in the manifest, adapter output, or saved commands; +the collector reads `RUSTFS_ACCESS_KEY` and `RUSTFS_SECRET_KEY` from its environment. + +#### Deployment Adapter Contract + +The runner invokes an executable as `adapter ACTION request.json response.json` +with no shell evaluation. Actions are separate processes: `prepare`, `measure`, +`oracle`, and `stop`. Every action must return zero and write a JSON object of +at most 1 MiB. Logs are kept separately and require an operator-managed disk +quota. Missing output, timeout, nonzero exit, unknown/missing metrics, zero +samples, and request errors fail the run. Adapters must terminate their own +children on failure and `stop` must be idempotent even after partial preparation. + +The request contains the fixed manifest fields, selected build, scenario, round, +leg, comparison (`build` or `background`), background mode (`on` or `off`), +duration, unique `data_dir`, expected object oracle, and expected repair count. +Adapter responsibilities: + +1. `prepare` deploys the selected binary into an authorized isolated topology, + checks actual binary/config/durability, initializes deterministic scenario + data and the requested cache state, and returns `{"ready": true}`. For measured + runs it also returns `collector` with exactly `alias`, `endpoint`, and + comma-separated `metrics_endpoints`; the runner starts the existing scanner + collector at 60-second cadence while `measure` runs. +2. `measure` maintains the fixed offered load for the entire requested duration. + `cold-hot` retains cold buckets while mutating a hot bucket; `fresh-hot` + creates a bucket after scanner startup; `multi-hot-new` combines several hot + buckets with a newly created bucket; `running-heal` applies foreground load + during active repair; `mrf-replay` replays independently seeded durable repair + work. Capture same-window status for bucket-freshness issue #7108. Actual + fault injection and dataset generation belong to the reviewed adapter. +3. `oracle` independently enumerates all objects and versions, reads and checks + their complete bytes, and verifies repairs. Return `complete: true`, integer + `errors: 0`, and `actual` matching the manifest's expected oracle. Never copy + expected values into a measured oracle or infer completion from empty queues. +4. `stop` stops task-owned workload/server processes and returns `stopped: true`. + Preserve data and artifacts for diagnosis. An adapter may restore previous + settings but must not delete arbitrary paths or stop unrelated deployments. + +The `measure` response echoes the observed `evidence`, `fixed`, `build`, +`data_dir`, and `background`, plus `sample_count` (1..3600), `elapsed_seconds`, +and `metrics`. All metrics must be finite nonnegative numbers: `p99_ms`, +`throughput_ops`, `rss_bytes`, `cpu_seconds`, `iops`, `rpc_count`, +`cache_clone_bytes`, `encode_bytes`, `save_bytes`, `oldest_age_seconds`, +`walk_objects`, `cold_walk_objects`, `healed_objects`, `errors`, and `requests`. +Requests, throughput, and p99 must be positive; errors must be zero. Repair +counts must match the manifest when background work is on. Keep underlying +request samples, counter reset checks, profiler captures, and per-node telemetry +in the cell artifact directory; aggregate values alone do not establish their +measurement provenance. Missing production instrumentation is a pending gate, +not permission to report a fabricated zero. + +For P2, `measure.convergence` contains booleans `writes_stopped`, +`last_mutation_observed`, `first_complete_publication`; numeric +`last_mutation_time`, `last_mutation_observed_time`, `writes_stopped_time`, `window_start`, `window_end`, +`budget_available_seconds`, `walk_objects`, and `full_walk_objects`. Times use +one monotonic clock. The window starts after writes stop and the final mutation +is observed, and ends at the first complete publication. The reference is an +independent full walk of the same static namespace. Record available budget +seconds to interpret elapsed time. During continuing writes, omit this proof +and report useful-work ratio and justified invalidation/re-scan work separately; +the runner reports P2 pending and does not impose a fixed cumulative walk bound. + +The nightly heal workflow clones **`rustfs/auto-testing`** separately and invokes +`auto-testing/rustfs_heal_test.sh`; that script is not a local `scripts/test` +entry point. If an adapter uses it, record and verify the external checkout's +owner and full commit before use. The current workflow clones the default branch, +so its contents must not be attributed to a RustFS source SHA. + +#### Evidence Gates + +`report.json` records each group's verdict and the raw responses remain in their +cell directories. Candidate/build p99 regression must be at most 5% and +throughput loss at most 3%; candidate background on/off limits are 10% and 5%. +P1 requires cold-hot walk reduction of at least the baseline cold-walk share +times 80%, rather than a fixed 80% reduction for every workload. P2 requires +candidate post-stop work at most 1.2 times the independent full-walk reference. +Missing candidate convergence proof yields `inconclusive`. A2/A1 or B2/B1 p99 +or throughput drift above 5% also yields `inconclusive`, with exit code 3. +Correctness errors and non-noisy performance regressions exit 1. Every group +must pass; a favorable median cannot hide a failing group. + +Synthetic success is explicitly `synthetic_validated`, with `performance: +pending`. It validates orchestration and gate logic only. It proves no runtime, +distributed, crash, mixed-version, or performance behavior and cannot close the +performance acceptance gate. Run the fake-adapter self-tests with: + +```bash +scripts/test_scanner_validation_harness.sh +``` + +They cover the complete 120-cell schedule, data isolation, missing builds and +oracles, zero samples/requests, swallowed request errors, offered-load drift, +incomplete repairs, missing metrics, noise, and P1/P2/p99 regressions. A real +deployment adapter and actual ABBA artifacts remain required before any measured +performance or release claim. + Collect at least two runs on the same RustFS commit and the same workload. Keep hardware, commit, object count, object size, bucket count, scanner-enabled state, and foreground workload constant between runs. | Run | Purpose | Example scanner settings | diff --git a/scripts/README.md b/scripts/README.md index da2bd119c..d76e7ba33 100644 --- a/scripts/README.md +++ b/scripts/README.md @@ -54,6 +54,8 @@ their issue closes. | `probe.sh` | dev-tool | Probe-style e2e run | `make probe-e2e` | | `run_scanner_validation_harness.sh` | dev-tool | Scanner validation harness | `docs/operations/scanner-benchmark-runbook.md` | | `test_scanner_validation_harness.sh` | dev-tool | Self-test for the scanner validation harness | — | +| `scanner_abba.py` | dev-tool | Scanner/heal ABBA orchestration and evidence gates via `run_scanner_validation_harness.sh --abba` | `docs/operations/scanner-benchmark-runbook.md` | +| `test_scanner_abba.py` | dev-tool | Synthetic ABBA adapter and failure-path tests | `test_scanner_validation_harness.sh` | | `test_build_rustfs_options.sh` | dev-tool | Shell test for rustfs build-option wiring | `make test` (script-tests) | | `test_entrypoint_credentials.sh` | dev-tool | Container entrypoint credential-handling test | `make test` (script-tests) | | `test_helm_chart_version.sh` | dev-tool | Test for `helm_chart_version.sh` | — | diff --git a/scripts/run_scanner_validation_harness.sh b/scripts/run_scanner_validation_harness.sh index 12c962203..5822cd757 100755 --- a/scripts/run_scanner_validation_harness.sh +++ b/scripts/run_scanner_validation_harness.sh @@ -1,6 +1,12 @@ #!/usr/bin/env bash set -euo pipefail +if [[ "${1:-}" == "--abba" ]]; then + shift + SCRIPT_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) + exec "$SCRIPT_DIR/python_bin.sh" "$SCRIPT_DIR/scanner_abba.py" "$@" +fi + ALIAS="" ENDPOINT="" ACCESS_KEY="${RUSTFS_ACCESS_KEY:-}" @@ -23,6 +29,7 @@ TELEMETRY_PIDS=() usage() { cat <<'USAGE' Usage: + scripts/run_scanner_validation_harness.sh --abba --help scripts/run_scanner_validation_harness.sh --alias \ --endpoint [options] diff --git a/scripts/scanner_abba.py b/scripts/scanner_abba.py new file mode 100644 index 000000000..9c461dc7b --- /dev/null +++ b/scripts/scanner_abba.py @@ -0,0 +1,329 @@ +#!/usr/bin/env python3 +"""Run isolated scanner/heal ABBA cells through a deployment-specific adapter.""" + +import argparse +import hashlib +import json +import math +import os +from pathlib import Path +import shutil +import signal +import subprocess +import sys +import time + +SCENARIOS = ("cold-hot", "fresh-hot", "multi-hot-new", "running-heal", "mrf-replay") +LEGS = ("A1", "B1", "B2", "A2") +MAX_JSON_BYTES = 1024 * 1024 +METRICS = ( + "p99_ms", "throughput_ops", "rss_bytes", "cpu_seconds", "iops", "rpc_count", + "cache_clone_bytes", "encode_bytes", "save_bytes", "oldest_age_seconds", + "walk_objects", "cold_walk_objects", "healed_objects", "errors", "requests", +) + + +def require(condition, message): + if not condition: + raise ValueError(message) + + +def number(value, name, minimum=0): + require(type(value) in (float, int) and math.isfinite(value) and value >= minimum, + f"invalid {name}") + return value + + +def digest(path): + with Path(path).open("rb") as stream: + return hashlib.file_digest(stream, "sha256").hexdigest() + + +def read_json(path): + require(path.stat().st_size <= MAX_JSON_BYTES, f"oversized JSON: {path.name}") + with path.open() as stream: + value = json.load(stream) + require(isinstance(value, dict), f"expected JSON object: {path.name}") + return value + + +def write_json(path, value): + data = json.dumps(value, indent=2, allow_nan=False) + "\n" + require(len(data.encode()) <= MAX_JSON_BYTES, "oversized result") + path.write_text(data) + + +def sha(value): + return isinstance(value, str) and len(value) == 64 and all(c in "0123456789abcdef" for c in value) + + +def validate_manifest(manifest): + require(manifest.get("schema") == 1, "unsupported manifest schema") + require(manifest.get("evidence") in ("synthetic", "measured"), "missing evidence type") + fixed = manifest["fixed"] + for key in ("config_sha256", "dataset_sha256"): + require(sha(fixed.get(key)), f"invalid fixed.{key}") + for key in ("release_flags", "durability", "disk_type", "cache_state", "load_command", "resource_isolation"): + require(isinstance(fixed.get(key), str) and fixed[key].strip(), f"missing fixed.{key}") + require(fixed.get("topology") == "EC8+4", "formal matrix requires EC8+4") + number(fixed.get("offered_load_ops"), "offered load", 1) + require(type(manifest.get("rounds")) is int and 3 <= manifest["rounds"] <= 10, + "rounds must be 3..10") + minimum = 900 if manifest["evidence"] == "measured" else 1 + require(type(manifest.get("duration_seconds")) is int and + minimum <= manifest["duration_seconds"] <= 86400, "invalid duration_seconds") + number(manifest.get("min_free_bytes"), "min_free_bytes", 1) + for phase in ("baseline", "candidate"): + build = manifest[phase] + path = Path(build["binary"]).resolve(strict=True) + require(path.is_file() and os.access(path, os.X_OK), f"missing executable {phase} build") + require(sha(build.get("sha256")) and digest(path) == build["sha256"], f"{phase} binary hash mismatch") + require(isinstance(build.get("revision"), str) and len(build["revision"]) == 40 and + all(c in "0123456789abcdef" for c in build["revision"]), f"invalid {phase} revision") + build["binary"] = str(path) + for scenario in SCENARIOS: + expected = manifest["oracles"][scenario] + for key in ("objects", "versions", "bytes"): + require(type(expected.get(key)) is int and expected[key] > 0, f"missing {scenario} oracle {key}") + require(sha(expected.get("sha256")), f"missing {scenario} content/version digest") + number(manifest["expected_healed_objects"].get(scenario), f"{scenario} expected repairs") + if scenario in ("running-heal", "mrf-replay"): + require(manifest["expected_healed_objects"][scenario] > 0, f"{scenario} requires repairs") + + +def invoke(adapter, action, request, timeout): + """The adapter writes bounded JSON separately; stderr/stdout remain raw evidence.""" + output = request.parent / f"{action}.json" + with (request.parent / f"{action}.log").open("wb") as log: + process = subprocess.Popen([str(adapter), action, str(request), str(output)], + stdout=log, stderr=subprocess.STDOUT, start_new_session=True) + try: + returncode = process.wait(timeout=timeout) + if returncode: + raise subprocess.CalledProcessError(returncode, [str(adapter), action]) + finally: + if process.poll() != 0: + try: + os.killpg(process.pid, signal.SIGTERM) + except ProcessLookupError: + pass + try: + process.wait(timeout=10) + except subprocess.TimeoutExpired: + os.killpg(process.pid, signal.SIGKILL) + process.wait() + return read_json(output) + + +def validate_result(result, request, expected): + require(result.get("evidence") == request["evidence"], "adapter evidence type mismatch") + require(result.get("fixed") == request["fixed"], "offered load/config/cache/durability drift") + require(result.get("build") == request["build"], "deployed build provenance mismatch") + require(result.get("data_dir") == request["data_dir"], "adapter data isolation mismatch") + require(result.get("background") == request["background"], "background mode mismatch") + require(type(result.get("sample_count")) is int and 1 <= result["sample_count"] <= 3600, + "sample_count must be 1..3600") + number(result.get("elapsed_seconds"), "elapsed_seconds", request["duration_seconds"]) + metrics = result["metrics"] + for key in METRICS: + number(metrics.get(key), key) + for key in ("requests", "p99_ms", "throughput_ops"): + require(metrics[key] > 0, f"zero {key}") + require(metrics["errors"] == 0, "workload request errors") + require(metrics["cold_walk_objects"] <= metrics["walk_objects"], "cold walk exceeds total walk") + require(result.get("oracle") == expected, "object/version/byte oracle mismatch") + if request["background"] == "on": + require(metrics["walk_objects"] > 0, "zero background walk") + require(metrics["healed_objects"] == request["expected_healed_objects"], "incomplete repair oracle") + if request["scenario"] in ("running-heal", "mrf-replay"): + require(metrics["healed_objects"] > 0, "zero completed repairs") + return result + + +def convergence(result): + window = result.get("convergence") + if not window or window.get("writes_stopped") is not True or window.get("last_mutation_observed") is not True or window.get("first_complete_publication") is not True: + return None + for key in ("last_mutation_time", "last_mutation_observed_time", "writes_stopped_time", "window_start", "window_end", "walk_objects", "full_walk_objects", "budget_available_seconds"): + number(window.get(key), f"convergence.{key}") + require(window["last_mutation_time"] <= window["writes_stopped_time"] <= window["window_start"] < window["window_end"], + "invalid post-mutation convergence window") + require(window["last_mutation_time"] <= window["last_mutation_observed_time"] <= window["window_start"], + "convergence started before last mutation was observed") + require(window["full_walk_objects"] > 0, "zero full walk reference") + require(0 < window["budget_available_seconds"] <= window["window_end"] - window["window_start"], + "invalid convergence budget window") + return window["walk_objects"] / window["full_walk_objects"] + + +def evaluate(cells): + comparisons = [] + inconclusive = False + failed = False + for offset in range(0, len(cells), 4): + group = cells[offset:offset + 4] + require([cell["leg"] for cell in group] == list(LEGS), "incomplete ABBA group") + a1, b1, b2, a2 = (cell["result"]["metrics"] for cell in group) + control = group[0]["comparison"] == "background" + drift = max(abs(a2[k] / a1[k] - 1) for k in ("p99_ms", "throughput_ops")) + repeat_drift = max(abs(b2[k] / b1[k] - 1) for k in ("p99_ms", "throughput_ops")) + noise = max(drift, repeat_drift) > 0.05 + a = {key: (a1[key] + a2[key]) / 2 for key in METRICS} + b = {key: (b1[key] + b2[key]) / 2 for key in METRICS} + p99 = b["p99_ms"] / a["p99_ms"] - 1 + throughput = b["throughput_ops"] / a["throughput_ops"] - 1 + thresholds = {"p99_regression": 0.10 if control else 0.05, + "throughput_loss": 0.05 if control else 0.03} + passed = p99 <= thresholds["p99_regression"] and throughput >= -thresholds["throughput_loss"] + p1 = None + if not control: + if group[0]["scenario"] == "cold-hot": + require(a["cold_walk_objects"] > 0, "cold-hot baseline has no cold walk samples") + required = a["cold_walk_objects"] / a["walk_objects"] * 0.80 + reduction = 1 - b["walk_objects"] / a["walk_objects"] + p1 = {"required_reduction": required, "observed_reduction": reduction} + if group[0]["scenario"] == "cold-hot": + passed &= reduction >= required + p2 = [convergence(cell["result"]) if cell["background"] == "on" else None for cell in group] + candidate_p2 = [value for cell, value in zip(group, p2) if cell["leg"].startswith("B")] + p2_pending = any(value is None for value in candidate_p2) + passed &= all(value <= 1.2 for value in candidate_p2 if value is not None) + inconclusive |= noise or p2_pending + if not noise and not passed: + failed = True + comparisons.append({"scenario": group[0]["scenario"], "comparison": group[0]["comparison"], + "round": group[0]["round"], "status": "inconclusive" if noise else ("fail" if not passed else "inconclusive" if p2_pending else "pass"), + "a2_a1_drift": drift, "b2_b1_drift": repeat_drift, + "p99_regression": p99, "throughput_change": throughput, + "thresholds": thresholds, "p1": p1, "p2_max_work_multiple": 1.2, + "p2_post_stop_work_multiples": p2}) + return ("fail" if failed else "inconclusive" if inconclusive else "pass"), comparisons + + +def collect_live(prepared, request, request_path, adapter): + collector = Path(__file__).with_name("run_scanner_validation_harness.sh") + # Only allow connection fields here; the runner owns cadence and output paths. + connection = prepared["collector"] + require(set(connection) == {"alias", "endpoint", "metrics_endpoints"}, "invalid collector connection") + require(all(isinstance(value, str) and value for value in connection.values()), "missing collector endpoint") + output = request_path.parent / "telemetry" + args = ["bash", str(collector), "--alias", connection["alias"], "--endpoint", connection["endpoint"], + "--metrics-endpoints", connection["metrics_endpoints"], "--deployment", "distributed", + "--samples", str(request["duration_seconds"] // 60 + 1), "--interval-secs", "60", + "--out-dir", str(output)] + with (request_path.parent / "collector.log").open("wb") as log: + process = subprocess.Popen(args, stdout=log, stderr=subprocess.STDOUT, start_new_session=True) + try: + started = time.monotonic() + result = invoke(adapter, "measure", request_path, request["duration_seconds"] + 300) + require(time.monotonic() - started >= request["duration_seconds"], "measurement ended before required window") + require(process.wait(timeout=120) == 0, "scanner collector failed") + require(output.joinpath("scanner-summary.csv").stat().st_size > 0, "missing collector samples") + samples = list((output / "status").glob("scanner-status.*.json")) + require(len(samples) == request["duration_seconds"] // 60 + 1, "missing scanner samples") + for sample in samples: + status = read_json(sample) + require(isinstance(status.get("metrics"), dict) and status["metrics"], "invalid scanner status response") + heals = list((output / "heal").glob("background-heal-status.*.json")) + require(bool(heals), "missing heal samples") + for sample in heals: + status = read_json(sample) + require(isinstance(status.get("healOperations"), dict) and status["healOperations"], "invalid heal status response") + return result + finally: + # Stop telemetry children as well when measurement fails or times out. + try: + os.killpg(process.pid, signal.SIGTERM) + except ProcessLookupError: + pass + try: + process.wait(timeout=10) + except subprocess.TimeoutExpired: + os.killpg(process.pid, signal.SIGKILL) + process.wait() + + +def run(manifest, adapter, output, data_root): + validate_manifest(manifest) + require(adapter.is_file() and os.access(adapter, os.X_OK), "missing executable adapter") + require(not output.exists() and not data_root.exists(), "output/data root must be new; existing data is preserved") + require(output != data_root and output not in data_root.parents and data_root not in output.parents, + "output and data roots must not overlap") + output.mkdir(parents=True) + data_root.mkdir(parents=True) + require(shutil.disk_usage(data_root).free >= manifest["min_free_bytes"], "insufficient free disk space") + manifest["adapter_sha256"] = digest(adapter) + manifest["collector_sha256"] = digest(Path(__file__).with_name("run_scanner_validation_harness.sh")) + write_json(output / "manifest.json", manifest) + cells = [] + write_json(output / "report.json", {"status": "incomplete", "performance": "pending"}) + try: + for scenario in SCENARIOS: + for comparison in ("build", "background"): + for round_id in range(1, manifest["rounds"] + 1): + for leg in LEGS: + phase = "baseline" if comparison == "build" and leg.startswith("A") else "candidate" + background = "off" if comparison == "background" and leg.startswith("A") else "on" + name = f"{scenario}-{comparison}-{round_id}-{leg}" + cell_dir = output / name + cell_dir.mkdir() + data_dir = data_root / name + data_dir.mkdir() + request = {"schema": 1, "scenario": scenario, "comparison": comparison, "round": round_id, + "leg": leg, "background": background, "build": manifest[phase], + "evidence": manifest["evidence"], "fixed": manifest["fixed"], + "duration_seconds": manifest["duration_seconds"], "data_dir": str(data_dir), + "expected_healed_objects": manifest["expected_healed_objects"][scenario], + "expected_oracle": manifest["oracles"][scenario]} + require(digest(Path(request["build"]["binary"])) == request["build"]["sha256"], "binary changed during run") + require(digest(adapter) == manifest["adapter_sha256"], "adapter changed during run") + require(shutil.disk_usage(data_root).free >= manifest["min_free_bytes"], "insufficient free disk space") + request_path = cell_dir / "request.json" + write_json(request_path, request) + print(name, flush=True) + try: + prepared = invoke(adapter, "prepare", request_path, 300) + require(prepared.get("ready") is True, "deployment not ready") + if manifest["evidence"] == "measured": + result = collect_live(prepared, request, request_path, adapter) + else: + result = invoke(adapter, "measure", request_path, 300) + # An independent operation must enumerate all object versions and bytes. + oracle = invoke(adapter, "oracle", request_path, 300) + require(oracle.get("complete") is True and oracle.get("errors") == 0, "correctness oracle failed") + require(type(oracle.get("errors")) is int, "invalid oracle error count") + result["oracle"] = oracle["actual"] + validate_result(result, request, request["expected_oracle"]) + cells.append({**request, "result": result}) + finally: + stopped = invoke(adapter, "stop", request_path, 300) + require(stopped.get("stopped") is True, "adapter failed to stop deployment") + status, comparisons = evaluate(cells) + synthetic = manifest["evidence"] == "synthetic" + report = {"status": "synthetic_validated" if synthetic and status == "pass" else status, + "evidence": manifest["evidence"], "performance": "pending" if synthetic else status, + "cells": len(cells), "comparisons": comparisons} + write_json(output / "report.json", report) + return 0 if status == "pass" else 3 if status == "inconclusive" else 1 + except (ValueError, KeyError, OSError, subprocess.SubprocessError) as error: + write_json(output / "report.json", {"status": "failed", "performance": "pending", + "completed_cells": len(cells), "error": str(error)}) + raise + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--manifest", type=Path, required=True) + parser.add_argument("--adapter", type=Path, required=True) + parser.add_argument("--out-dir", type=Path, required=True) + parser.add_argument("--data-root", type=Path, required=True) + args = parser.parse_args() + try: + return run(read_json(args.manifest), args.adapter.resolve(), args.out_dir.resolve(), args.data_root.resolve()) + except (ValueError, KeyError, OSError, subprocess.SubprocessError) as error: + print(f"ERROR: {error}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/test_scanner_abba.py b/scripts/test_scanner_abba.py new file mode 100755 index 000000000..30fc00839 --- /dev/null +++ b/scripts/test_scanner_abba.py @@ -0,0 +1,181 @@ +#!/usr/bin/env python3 +"""Synthetic adapter and failure-propagation tests; never start a RustFS server.""" + +import contextlib +import copy +import io +import json +import os +from pathlib import Path +import subprocess +import sys +import tempfile +import unittest +from unittest.mock import patch + +import scanner_abba as harness + + +def fake_adapter(): + action, request_path, output_path = sys.argv[1:] + request = harness.read_json(Path(request_path)) + fault = os.environ.get("SCANNER_ABBA_TEST_FAULT", "") + if action == "prepare": + result = {"ready": True} + elif action == "stop": + result = {"stopped": True} + elif action == "oracle": + if fault == "oracle-exit": + return 42 + if fault == "missing-oracle": + return 0 + result = {"complete": True, "errors": 0, "actual": request["expected_oracle"]} + if fault == "oracle-mismatch": + result["actual"]["bytes"] += 1 + else: + if fault == "measure-exit": + return 42 + result = {key: request[key] for key in ("evidence", "fixed", "build", "data_dir", "background")} + result.update({"sample_count": 10, "elapsed_seconds": request["duration_seconds"], + "metrics": dict.fromkeys(harness.METRICS, 10)}) + baseline = request["comparison"] == "build" and request["leg"].startswith("A") + result["metrics"].update(p99_ms=10, throughput_ops=100, errors=0, requests=100, + walk_objects=100 if baseline else 20, cold_walk_objects=100 if baseline else 0, + healed_objects=request["expected_healed_objects"]) + result["convergence"] = {"writes_stopped": True, "last_mutation_observed": True, + "first_complete_publication": True, "last_mutation_time": 1, + "last_mutation_observed_time": 2, + "writes_stopped_time": 2, "window_start": 2, "window_end": 3, + "budget_available_seconds": 1, "walk_objects": 110, "full_walk_objects": 100} + if fault == "zero-samples": + result["sample_count"] = 0 + elif fault == "request-errors": + result["metrics"]["errors"] = 1 + elif fault == "load-drift": + result["fixed"]["offered_load_ops"] += 1 + elif fault == "noise" and request["leg"] == "A2": + result["metrics"]["p99_ms"] = 20 + elif fault == "zero-requests": + result["metrics"]["requests"] = 0 + elif fault == "no-publication": + result["convergence"]["first_complete_publication"] = False + elif fault == "p2-regression": + result["convergence"]["walk_objects"] = 121 + elif fault == "latency-regression" and request["leg"].startswith("B"): + result["metrics"]["p99_ms"] = 12 + elif fault == "p1-regression" and not baseline: + result["metrics"]["walk_objects"] = 30 + elif fault == "missing-metric": + del result["metrics"]["save_bytes"] + elif fault == "incomplete-repair": + result["metrics"]["healed_objects"] = 0 + harness.write_json(Path(output_path), result) + return 0 + + +class ScannerAbbaTest(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.root = Path(self.temp.name) + self.binary = Path(sys.executable).resolve() + self.adapter = Path(__file__).resolve() + self.manifest = { + "schema": 1, "evidence": "synthetic", "rounds": 3, "duration_seconds": 1, "min_free_bytes": 1, + "fixed": {"config_sha256": "1" * 64, "dataset_sha256": "2" * 64, + "release_flags": "--release", "durability": "drive-sync=on", + "disk_type": "synthetic", "cache_state": "cold", "load_command": "fake", + "topology": "EC8+4", "offered_load_ops": 100, "resource_isolation": "synthetic"}, + "oracles": {s: {"objects": 10, "versions": 20, "bytes": 30, "sha256": "3" * 64} for s in harness.SCENARIOS}, + "expected_healed_objects": {s: 10 for s in harness.SCENARIOS}, + } + build = {"binary": str(self.binary), "sha256": harness.digest(self.binary), "revision": "a" * 40} + self.manifest.update(baseline=build.copy(), candidate=build.copy()) + + def run_harness(self, fault=""): + with patch.dict(os.environ, {"SCANNER_ABBA_TEST_FAULT": fault}), contextlib.redirect_stdout(io.StringIO()): + return harness.run(copy.deepcopy(self.manifest), self.adapter, self.root / "out", self.root / "data") + + def test_complete_synthetic_matrix_is_not_performance_evidence(self): + self.assertEqual(self.run_harness(), 0) + report = harness.read_json(self.root / "out/report.json") + self.assertEqual((report["status"], report["performance"], report["cells"]), ("synthetic_validated", "pending", 120)) + requests = [harness.read_json(path) for path in (self.root / "out").glob("*/request.json")] + self.assertEqual(len({r["data_dir"] for r in requests}), 120) + for scenario in harness.SCENARIOS: + for comparison in ("build", "background"): + for round_id in (1, 2, 3): + legs = [r for r in requests if (r["scenario"], r["comparison"], r["round"]) == (scenario, comparison, round_id)] + self.assertEqual({r["leg"] for r in legs}, set(harness.LEGS)) + self.assertTrue(all(c["p2_max_work_multiple"] == 1.2 for c in report["comparisons"])) + + def test_fail_closed_adapter_and_data_errors(self): + for fault in ("measure-exit", "oracle-exit", "missing-oracle", "oracle-mismatch", "zero-samples", + "zero-requests", "request-errors", "load-drift", "missing-metric", "incomplete-repair"): + with self.subTest(fault=fault), tempfile.TemporaryDirectory() as directory: + self.root = Path(directory) + with self.assertRaises((ValueError, OSError, subprocess.SubprocessError)): + self.run_harness(fault) + report = harness.read_json(self.root / "out/report.json") + self.assertEqual(report["status"], "failed") + self.assertTrue(list((self.root / "out").glob("*/stop.json"))) + + def test_noise_is_inconclusive_and_nonzero(self): + with patch.object(harness, "SCENARIOS", ("cold-hot",)): + self.assertEqual(self.run_harness("noise"), 3) + self.assertEqual(harness.read_json(self.root / "out/report.json")["status"], "inconclusive") + + def test_missing_first_publication_is_inconclusive(self): + with patch.object(harness, "SCENARIOS", ("cold-hot",)): + self.assertEqual(self.run_harness("no-publication"), 3) + + def test_performance_regressions_fail(self): + for fault in ("p1-regression", "p2-regression", "latency-regression"): + with self.subTest(fault=fault), tempfile.TemporaryDirectory() as directory: + self.root = Path(directory) + with patch.object(harness, "SCENARIOS", ("cold-hot",)): + self.assertEqual(self.run_harness(fault), 1) + + def test_manifest_rejects_missing_build_or_oracle(self): + for section, key in (("baseline", "binary"), ("oracles", "cold-hot")): + manifest = copy.deepcopy(self.manifest) + del manifest[section][key] + with self.subTest(section=section), self.assertRaises((ValueError, KeyError)): + harness.validate_manifest(manifest) + + def test_short_measured_window_and_fewer_rounds_rejected(self): + self.manifest["evidence"] = "measured" + with self.assertRaisesRegex(ValueError, "duration_seconds"): + harness.validate_manifest(self.manifest) + self.manifest["duration_seconds"] = 900 + self.manifest["rounds"] = 2 + with self.assertRaisesRegex(ValueError, "rounds"): + harness.validate_manifest(self.manifest) + + def test_existing_data_preserved(self): + (self.root / "data").mkdir() + marker = self.root / "data/keep" + marker.write_text("existing") + with self.assertRaisesRegex(ValueError, "preserved"): + self.run_harness() + self.assertEqual(marker.read_text(), "existing") + + def test_invalid_or_live_write_window_does_not_claim_p2(self): + self.assertIsNone(harness.convergence({"convergence": {"writes_stopped": False}})) + with self.assertRaises(ValueError): + harness.convergence({"convergence": {"writes_stopped": True, "last_mutation_observed": True, + "first_complete_publication": True}}) + + def test_nan_and_oversized_samples_rejected(self): + with self.assertRaises(ValueError): + harness.number(float("nan"), "latency") + path = self.root / "oversized.json" + path.write_bytes(b" " * (harness.MAX_JSON_BYTES + 1)) + with self.assertRaisesRegex(ValueError, "oversized"): + harness.read_json(path) + + +if __name__ == "__main__": + if len(sys.argv) == 4 and sys.argv[1] in ("prepare", "measure", "oracle", "stop"): + sys.exit(fake_adapter()) + unittest.main() diff --git a/scripts/test_scanner_validation_harness.sh b/scripts/test_scanner_validation_harness.sh index 9107926b1..d360097b8 100755 --- a/scripts/test_scanner_validation_harness.sh +++ b/scripts/test_scanner_validation_harness.sh @@ -312,3 +312,5 @@ if PATH="$BIN_DIR:$PATH" "$SCRIPT" --secret-key rustfsadmin >"$secret_arg_log" 2 fi grep -q -- 'unknown arg: --secret-key' "$secret_arg_log" + +"$ROOT_DIR/scripts/python_bin.sh" "$ROOT_DIR/scripts/test_scanner_abba.py"