diff --git a/.config/make/tests.mak b/.config/make/tests.mak index 726ca04cf..7d29ad6fc 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..70fb05c21 --- /dev/null +++ b/scripts/scanner_abba.py @@ -0,0 +1,384 @@ +#!/usr/bin/env python3 +"""Run isolated scanner/heal ABBA cells through a deployment-specific adapter.""" + +import argparse +from decimal import Decimal +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", +) +REPEATABILITY_LIMIT = Decimal("0.05") +P2_WORK_MULTIPLE_LIMIT = Decimal("1.2") + + +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 decimal_number(value, name, minimum=0): + if isinstance(value, Decimal): + require(value.is_finite() and value >= Decimal(str(minimum)), f"invalid {name}") + return value + number(value, name, minimum) + return Decimal(str(value)) + + +def ratio(numerator, denominator, name): + denominator = decimal_number(denominator, f"{name} denominator") + require(denominator > 0, f"invalid {name} denominator") + return decimal_number(numerator, name) / denominator + + +def relative_change(current, baseline, name): + return ratio(current, baseline, name) - Decimal("1") + + +def repeatability_change(first, second, name): + first = decimal_number(first, name) + second = decimal_number(second, name) + if first == 0 and second == 0: + return Decimal("0") + if first == 0 or second == 0: + return Decimal("Infinity") + return abs(second / first - Decimal("1")) + + +def report_number(value): + return None if value.is_infinite() else float(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(relative_change(a2[k], a1[k], k)) for k in ("p99_ms", "throughput_ops")) + repeat_drift = max(abs(relative_change(b2[k], b1[k], k)) for k in ("p99_ms", "throughput_ops")) + noise = max(drift, repeat_drift) > REPEATABILITY_LIMIT + a = {key: (decimal_number(a1[key], key) + decimal_number(a2[key], key)) / Decimal("2") for key in METRICS} + b = {key: (decimal_number(b1[key], key) + decimal_number(b2[key], key)) / Decimal("2") for key in METRICS} + p99 = relative_change(b["p99_ms"], a["p99_ms"], "p99_ms") + throughput = relative_change(b["throughput_ops"], a["throughput_ops"], "throughput_ops") + thresholds = {"p99_regression": Decimal("0.10") if control else Decimal("0.05"), + "throughput_loss": Decimal("0.05") if control else Decimal("0.03")} + passed = p99 <= thresholds["p99_regression"] and throughput >= -thresholds["throughput_loss"] + p1 = None + work_drift = None + if not control: + if group[0]["scenario"] == "cold-hot": + require(a["cold_walk_objects"] > 0, "cold-hot baseline has no cold walk samples") + work_drift = max(repeatability_change(a1[key], a2[key], key) for key in ("walk_objects", "cold_walk_objects")) + work_drift = max(work_drift, *(repeatability_change(b1[key], b2[key], key) for key in ("walk_objects", "cold_walk_objects"))) + noise |= work_drift > REPEATABILITY_LIMIT + required = ratio(a["cold_walk_objects"], a["walk_objects"], "cold walk baseline") * Decimal("0.80") + reduction = Decimal("1") - ratio(b["walk_objects"], a["walk_objects"], "walk reduction") + p1 = {"required_reduction": float(required), "observed_reduction": float(reduction), + "repeatability_drift": report_number(work_drift)} + if group[0]["scenario"] == "cold-hot": + # Compare counts before division can round repeating decimal ratios. + passed &= a["walk_objects"] - b["walk_objects"] >= a["cold_walk_objects"] * Decimal("0.80") + 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(ratio(value, 1, "p2 work multiple") <= P2_WORK_MULTIPLE_LIMIT 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": report_number(drift), "b2_b1_drift": report_number(repeat_drift), + "p99_regression": float(p99), "throughput_change": float(throughput), + "thresholds": {key: float(value) for key, value in thresholds.items()}, + "p1": p1, "p2_max_work_multiple": float(P2_WORK_MULTIPLE_LIMIT), + "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") + metrics = list((output / "metrics").glob("admin-metrics.*.ndjson")) + endpoints = [endpoint for endpoint in connection["metrics_endpoints"].split(",") if endpoint] + require(metrics and len(metrics) == len(endpoints) * len(samples), "missing distributed metrics samples") + for sample in metrics: + # The collector requests n=1, so each file contains one final JSON record. + status = read_json(sample) + require(status.get("errors") == [], "distributed metrics errors") + require(status.get("final") is True, "incomplete distributed metrics") + hosts = status.get("by_host") + require(isinstance(hosts, dict) and hosts, "missing by-host metrics") + for host in hosts.values(): + require(isinstance(host, dict) and isinstance(host.get("scanner"), dict) and host["scanner"], + "missing per-host scanner metrics") + 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..79935f924 --- /dev/null +++ b/scripts/test_scanner_abba.py @@ -0,0 +1,264 @@ +#!/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 Mock, 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 == "exact-thresholds" and request["leg"].startswith("B"): + result["metrics"].update(p99_ms=10.5, throughput_ops=97) + elif fault == "just-over-threshold" and request["comparison"] == "build" and request["leg"].startswith("B"): + result["metrics"]["p99_ms"] = 10.500001 + elif fault == "p1-regression" and not baseline: + result["metrics"]["walk_objects"] = 30 + elif fault in ("p1-exact-fraction", "p1-over-fraction"): + result["metrics"].update(walk_objects=9 if baseline else 5 + (fault == "p1-over-fraction"), + cold_walk_objects=5 if baseline else 0) + elif fault == "unstable-p1-control" and request["comparison"] == "build": + if request["leg"] == "A1": + result["metrics"].update(walk_objects=1000, cold_walk_objects=1000) + elif request["leg"] == "A2": + result["metrics"].update(walk_objects=10, cold_walk_objects=10) + elif request["leg"].startswith("B"): + result["metrics"].update(walk_objects=100, cold_walk_objects=0) + 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_exact_threshold_boundaries_pass(self): + with patch.object(harness, "SCENARIOS", ("cold-hot",)): + self.assertEqual(self.run_harness("exact-thresholds"), 0) + + def test_just_over_threshold_fails(self): + with patch.object(harness, "SCENARIOS", ("cold-hot",)): + self.assertEqual(self.run_harness("just-over-threshold"), 1) + + def test_p1_fractional_boundary(self): + for fault, expected in (("p1-exact-fraction", 0), ("p1-over-fraction", 1)): + 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), expected) + + def test_live_collector_rejects_missing_or_failed_node_metrics(self): + telemetry = self.root / "telemetry" + for name in ("status", "heal", "metrics"): + (telemetry / name).mkdir(parents=True) + (telemetry / "scanner-summary.csv").write_text("timestamp\n") + valid = {"errors": [], "final": True, "by_host": {"node-b:9000": {"scanner": {"objects": 10}}}} + for index in range(16): + harness.write_json(telemetry / f"status/scanner-status.{index}.json", {"metrics": {"objects": 10}}) + for node in ("node-a", "node-b"): + harness.write_json(telemetry / f"heal/background-heal-status.{node}.{index}.json", + {"healOperations": {"queueLength": 0}}) + harness.write_json(telemetry / f"metrics/admin-metrics.{node}.{index}.ndjson", + {**valid, "by_host": {f"{node}:9000": {"scanner": {"objects": 10}}}}) + sample = telemetry / "metrics/admin-metrics.node-b.15.ndjson" + prepared = {"collector": {"alias": "test", "endpoint": "http://node-a:9000", + "metrics_endpoints": "http://node-a:9000,http://node-b:9000"}} + cases = ( + ("valid", valid, None), + ("missing", None, "missing distributed metrics samples"), + ("empty", "", "Expecting value"), + ("http-error", {"Code": "AccessDenied"}, "distributed metrics errors"), + ("partial-error", {**valid, "errors": ["node unavailable"]}, "distributed metrics errors"), + ("unfinished", {**valid, "final": False}, "incomplete distributed metrics"), + ("missing-host", {**valid, "by_host": {}}, "missing by-host metrics"), + ("missing-scanner", {**valid, "by_host": {"node-a:9000": {}}}, "missing per-host scanner metrics"), + ("collector-exit", valid, "scanner collector failed"), + ) + for name, payload, error in cases: + with self.subTest(fault=name): + if payload is None: + sample.unlink() + elif isinstance(payload, str): + sample.write_text(payload) + else: + harness.write_json(sample, payload) + process = Mock(pid=123, wait=Mock(return_value=1 if name == "collector-exit" else 0)) + with patch.object(harness.subprocess, "Popen", return_value=process), \ + patch.object(harness, "invoke", return_value={"sample_count": 10}), \ + patch.object(harness.time, "monotonic", side_effect=(0, 900)), \ + patch.object(harness.os, "killpg"): + if error: + with self.assertRaisesRegex(ValueError, error): + harness.collect_live(prepared, {"duration_seconds": 900}, self.root / "request.json", self.adapter) + else: + self.assertEqual(harness.collect_live(prepared, {"duration_seconds": 900}, + self.root / "request.json", self.adapter), {"sample_count": 10}) + + def test_unstable_p1_work_control_is_inconclusive(self): + with patch.object(harness, "SCENARIOS", ("cold-hot",)): + self.assertEqual(self.run_harness("unstable-p1-control"), 3) + comparison = harness.read_json(self.root / "out/report.json")["comparisons"][0] + self.assertEqual(comparison["status"], "inconclusive") + self.assertGreater(comparison["p1"]["repeatability_drift"], 0.05) + + 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"