mirror of
https://github.com/rustfs/rustfs.git
synced 2026-09-09 21:56:03 +00:00
test: add scanner heal scheduler pressure evidence (#7584)
Add measured G10/P1/P3 scheduler-pressure release descriptor generation, ABBA scheduler/profile provenance, and focused self-tests for the release bundle gate. Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
houseme
@@ -57,9 +57,11 @@ their issue closes.
|
||||
| `run_scanner_validation_harness.sh` | dev-tool | Scanner validation harness | `docs/operations/scanner-benchmark-runbook.md` |
|
||||
| `run_scanner_heal_evidence_case.sh` | dev-tool | Runs one Scanner/Heal release-evidence registry case and checks the produced receipt/oracle | `.config/scanner-heal-required-tests.json`; `check_test_wiring.py --check-scanner-heal` |
|
||||
| `run_scanner_heal_g09_upgrade_evidence.sh` | dev-tool | Runs the G09 mixed-version and rollback upgrade E2E lanes against a pinned previous release and verifies the raw evidence artifacts | `docs/testing/ci-gates.md`; `.github/workflows/e2e-upgrade.yml`; `test_scanner_heal_g09_upgrade_evidence.sh` |
|
||||
| `run_scanner_heal_scheduler_pressure_evidence.py` | dev-tool | Assembles measured Scanner/Heal G10/P1/P3 scheduler-pressure release descriptors from a completed measured ABBA run, recovery-window proof, and profile artifacts | `docs/operations/scanner-benchmark-runbook.md`; `test_scanner_heal_scheduler_pressure_evidence.sh` |
|
||||
| `run_scanner_heal_w13_mrf_evidence.sh` | dev-tool | Runs the W13 durable MRF replay lanes and writes G07/G08/P4 bundle-ready evidence descriptors | `docs/testing/ci-gates.md`; `test_scanner_heal_w13_mrf_evidence.sh` |
|
||||
| `run_scanner_heal_w16_recovery_evidence.sh` | dev-tool | Runs the W16 recovery-intent and quota authority lanes and writes G04/G12 bundle-ready evidence descriptors | `docs/testing/ci-gates.md`; `test_scanner_heal_w16_recovery_evidence.sh` |
|
||||
| `test_scanner_validation_harness.sh` | dev-tool | Self-test for the scanner validation harness | — |
|
||||
| `test_scanner_heal_scheduler_pressure_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal scheduler-pressure evidence assembler | — |
|
||||
| `test_scanner_heal_g09_upgrade_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal G09 upgrade evidence runner | — |
|
||||
| `test_scanner_heal_w16_recovery_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal W16 recovery evidence runner | — |
|
||||
| `test_scanner_heal_w13_mrf_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal W13 MRF evidence runner | — |
|
||||
|
||||
+590
@@ -0,0 +1,590 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Assemble measured Scanner/Heal scheduler-pressure release evidence.
|
||||
|
||||
This producer consumes a completed measured Scanner/Heal ABBA run plus measured
|
||||
profile and recovery-window artifacts. It only packages existing measurements;
|
||||
it never turns synthetic harness output into release evidence.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
from datetime import datetime, timedelta, timezone
|
||||
import json
|
||||
import math
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from scanner_abba import (
|
||||
RELEASE_PROFILE_ARTIFACTS,
|
||||
RELEASE_SCHEDULER_BOUNDS,
|
||||
SCENARIOS,
|
||||
digest,
|
||||
read_json,
|
||||
require,
|
||||
sha,
|
||||
write_json,
|
||||
)
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
G10_FIELDS = ("scheduler_bound_evidence", "pressure_recovery_evidence")
|
||||
P1_FIELDS = ("cold_walk_share_measurement", "foreground_latency_throughput_measurement", "profile_evidence")
|
||||
P3_FIELDS = ("two_hour_pressure_measurement", "heal_capacity_measurement", "recovery_window_measurement")
|
||||
PRESSURE_METRICS = (
|
||||
"foreground_p95_ms",
|
||||
"foreground_p99_ms",
|
||||
"throughput_ops",
|
||||
"error_rate",
|
||||
"heal_lock_wait_p99_ms",
|
||||
"attempt_cost_samples",
|
||||
)
|
||||
|
||||
|
||||
def utc_now() -> str:
|
||||
return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
|
||||
|
||||
|
||||
def git_head() -> str:
|
||||
return subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=ROOT, text=True).strip()
|
||||
|
||||
|
||||
def finite_number(value: Any, name: str, minimum: float = 0.0) -> float:
|
||||
require(type(value) in (int, float) and math.isfinite(value) and value >= minimum, f"invalid {name}")
|
||||
return float(value)
|
||||
|
||||
|
||||
def positive_int(value: Any, name: str, minimum: int = 1) -> int:
|
||||
require(type(value) is int and value >= minimum, f"invalid {name}")
|
||||
return value
|
||||
|
||||
|
||||
def positive_int_from_sources(cli_value: int | None, manifest_values: dict[str, Any], key: str) -> int:
|
||||
if cli_value is not None:
|
||||
return positive_int(cli_value, key)
|
||||
return positive_int(manifest_values.get(key), key)
|
||||
|
||||
|
||||
def parse_artifact_arg(value: str) -> tuple[str, Path, str]:
|
||||
try:
|
||||
kind, raw_path = value.split("=", 1)
|
||||
except ValueError as err:
|
||||
raise argparse.ArgumentTypeError("profile artifact must be KIND=PATH") from err
|
||||
if kind not in RELEASE_PROFILE_ARTIFACTS:
|
||||
raise argparse.ArgumentTypeError(f"unknown profile artifact kind: {kind}")
|
||||
path = Path(raw_path).expanduser().resolve()
|
||||
if not path.is_file() or path.stat().st_size == 0:
|
||||
raise argparse.ArgumentTypeError(f"profile artifact is missing or empty: {raw_path}")
|
||||
suffix = path.suffix.lower().lstrip(".")
|
||||
artifact_format = f"profile-{suffix}" if suffix in {"json", "ndjson"} else (suffix if suffix else "binary")
|
||||
return kind, path, artifact_format
|
||||
|
||||
|
||||
def load_measured_abba(abba_dir: Path, source_revision: str) -> tuple[dict[str, Any], dict[str, Any], list[dict[str, Any]]]:
|
||||
manifest = read_json(abba_dir / "manifest.json")
|
||||
report = read_json(abba_dir / "report.json")
|
||||
require(manifest.get("evidence") == "measured", "ABBA manifest must be measured")
|
||||
require(report.get("evidence") == "measured", "ABBA report must be measured")
|
||||
require(report.get("status") == "pass" and report.get("performance") == "pass", "ABBA report must pass")
|
||||
require(manifest.get("candidate", {}).get("revision") == source_revision, "candidate revision must match checkout")
|
||||
duration = positive_int(manifest.get("duration_seconds"), "duration_seconds", 7200)
|
||||
require(duration >= 7200, "P3 two-hour pressure evidence requires at least 7200 seconds")
|
||||
expected_cells = len(SCENARIOS) * 2 * positive_int(manifest.get("rounds"), "rounds") * 4
|
||||
require(report.get("cells") == expected_cells, "ABBA report did not complete the full matrix")
|
||||
require(manifest.get("fixed", {}).get("offered_load_ops", 0) > 0, "missing fixed offered load")
|
||||
release_evidence = manifest.get("release_evidence")
|
||||
require(isinstance(release_evidence, dict), "ABBA manifest missing release evidence")
|
||||
scheduler = release_evidence.get("scheduler")
|
||||
require(isinstance(scheduler, dict), "ABBA manifest missing scheduler evidence")
|
||||
require(scheduler.get("bounds") == list(RELEASE_SCHEDULER_BOUNDS), "ABBA scheduler bounds mismatch")
|
||||
positive_int(scheduler.get("max_deferred_items"), "scheduler.max_deferred_items")
|
||||
positive_int(scheduler.get("max_deferred_bytes"), "scheduler.max_deferred_bytes")
|
||||
positive_int(scheduler.get("max_retry_age_seconds"), "scheduler.max_retry_age_seconds")
|
||||
require(scheduler.get("duplicate_task_bound_observed") is True, "ABBA scheduler duplicate bound not observed")
|
||||
|
||||
measures: list[dict[str, Any]] = []
|
||||
for measure_path in sorted(abba_dir.glob("*-*-*-*/measure.json")):
|
||||
measure = read_json(measure_path)
|
||||
require(measure.get("evidence") == "measured", f"{measure_path.name} is not measured")
|
||||
require(measure.get("build", {}).get("revision") in {
|
||||
manifest["baseline"]["revision"],
|
||||
manifest["candidate"]["revision"],
|
||||
}, "measure build revision is outside the manifest")
|
||||
metrics = measure.get("metrics")
|
||||
require(isinstance(metrics, dict), "measure missing metrics")
|
||||
measures.append(measure)
|
||||
require(len(measures) == expected_cells, "missing measured cell outputs")
|
||||
return manifest, report, measures
|
||||
|
||||
|
||||
def comparison_rows(report: dict[str, Any], scenario: str, comparison: str) -> list[dict[str, Any]]:
|
||||
rows = [
|
||||
item for item in report.get("comparisons", [])
|
||||
if item.get("scenario") == scenario and item.get("comparison") == comparison
|
||||
]
|
||||
require(rows, f"missing {scenario}/{comparison} comparisons")
|
||||
return rows
|
||||
|
||||
|
||||
def worst_metric(measures: list[dict[str, Any]], key: str) -> float:
|
||||
return max(finite_number(item["metrics"].get(key), key) for item in measures)
|
||||
|
||||
|
||||
def sum_metric(measures: list[dict[str, Any]], key: str) -> float:
|
||||
return sum(finite_number(item["metrics"].get(key), key) for item in measures)
|
||||
|
||||
|
||||
def copy_profile_artifacts(out_dir: Path, artifacts: dict[str, tuple[Path, str]], source_revision: str,
|
||||
run_id: str, window_id: str) -> dict[str, Any]:
|
||||
copied: dict[str, Any] = {}
|
||||
profile_dir = out_dir / "artifacts" / "profiles"
|
||||
profile_dir.mkdir(parents=True, exist_ok=True)
|
||||
for kind in RELEASE_PROFILE_ARTIFACTS:
|
||||
source, artifact_format = artifacts[kind]
|
||||
target = profile_dir / f"P1-profile_evidence-{kind}{source.suffix or '.artifact'}"
|
||||
shutil.copyfile(source, target)
|
||||
copied[kind] = {
|
||||
"artifact": target.relative_to(out_dir).as_posix(),
|
||||
"sha256": digest(target),
|
||||
"artifact_format": artifact_format,
|
||||
"source_revision": source_revision,
|
||||
"run_id": run_id,
|
||||
"measurement_window_id": window_id,
|
||||
}
|
||||
return copied
|
||||
|
||||
|
||||
def write_field(out_dir: Path, gate: str, field: str, evidence: dict[str, Any]) -> dict[str, Any]:
|
||||
artifact = out_dir / "artifacts" / f"{gate}-{field}.json"
|
||||
artifact.parent.mkdir(parents=True, exist_ok=True)
|
||||
payload = {
|
||||
"schema": 1,
|
||||
"evidence_type": "measured",
|
||||
"source_revision": evidence["source_revision"],
|
||||
"run_id": evidence["run_id"],
|
||||
"measurement_window_id": evidence["measurement_window_id"],
|
||||
"gate": gate,
|
||||
"field": field,
|
||||
}
|
||||
for key, value in evidence.items():
|
||||
if key not in {"artifact", "sha256", "artifact_format", "summary", "started_at", "finished_at", "command"}:
|
||||
payload[key] = value
|
||||
write_json(artifact, payload)
|
||||
evidence["artifact"] = artifact.relative_to(out_dir).as_posix()
|
||||
evidence["sha256"] = digest(artifact)
|
||||
evidence["artifact_format"] = "json"
|
||||
return evidence
|
||||
|
||||
|
||||
def recovery_evidence(path: Path, source_revision: str) -> dict[str, Any]:
|
||||
payload = read_json(path)
|
||||
for marker in ("fixture", "fixture_only", "dry_run", "synthetic"):
|
||||
require(payload.get(marker) is not True, f"recovery artifact is {marker}")
|
||||
require(payload.get("evidence_type") == "measured", "recovery artifact must be measured")
|
||||
require(payload.get("source_revision") == source_revision, "recovery artifact source revision mismatch")
|
||||
require(payload.get("fault_modes") == ["process-restart", "process-crash-restart"],
|
||||
"recovery artifact must cover both restart modes")
|
||||
return payload
|
||||
|
||||
|
||||
def build_descriptor(args: argparse.Namespace) -> Path:
|
||||
out_dir = args.out_dir.resolve()
|
||||
require(not out_dir.exists(), "output directory must be new")
|
||||
source_revision = args.source_revision or git_head()
|
||||
manifest, report, measures = load_measured_abba(args.abba_dir.resolve(), source_revision)
|
||||
recovery = recovery_evidence(args.recovery_window_json.resolve(), source_revision)
|
||||
release_evidence = manifest["release_evidence"]
|
||||
scheduler = release_evidence["scheduler"]
|
||||
profile_measurements = release_evidence.get("profile", {}).get("measurements", {})
|
||||
require(isinstance(profile_measurements, dict), "ABBA manifest profile measurements must be an object")
|
||||
|
||||
profile_inputs = {}
|
||||
profile_formats = {}
|
||||
for item in args.profile_artifact:
|
||||
kind, path, artifact_format = parse_artifact_arg(item)
|
||||
require(kind not in profile_inputs, f"duplicate profile artifact kind: {kind}")
|
||||
profile_inputs[kind] = path
|
||||
profile_formats[kind] = artifact_format
|
||||
missing_profiles = sorted(set(RELEASE_PROFILE_ARTIFACTS) - set(profile_inputs))
|
||||
require(not missing_profiles, "missing profile artifacts: " + ", ".join(missing_profiles))
|
||||
|
||||
out_dir.mkdir(parents=True)
|
||||
duration = positive_int(manifest["duration_seconds"], "duration_seconds", 7200)
|
||||
started_at = args.started_at or utc_now()
|
||||
if args.finished_at:
|
||||
finished_at = args.finished_at
|
||||
else:
|
||||
started = datetime.fromisoformat(started_at.replace("Z", "+00:00"))
|
||||
finished_at = (started + timedelta(seconds=duration)).isoformat().replace("+00:00", "Z")
|
||||
run_id = args.run_id or f"scheduler-pressure-{source_revision[:12]}"
|
||||
window_id = args.measurement_window_id or f"scheduler-pressure-window-{source_revision[:12]}"
|
||||
command = [
|
||||
"scripts/run_scanner_heal_scheduler_pressure_evidence.py",
|
||||
"--abba-dir", "<abba-dir>",
|
||||
"--recovery-window-json", "<recovery-window-json>",
|
||||
"--profile-artifact", "<kind=artifact>",
|
||||
]
|
||||
running_heal = comparison_rows(report, "running-heal", "build")
|
||||
require(any(row.get("w10", {}).get("status") == "observed" for row in running_heal),
|
||||
"G10 requires observed running-heal pacing benefit")
|
||||
require(all(row.get("w11", {}).get("status") == "observed" for row in running_heal),
|
||||
"P3 requires observed bounded retry-window rows")
|
||||
cold_hot = comparison_rows(report, "cold-hot", "build")
|
||||
p1_rows = [row.get("p1") for row in cold_hot]
|
||||
require(all(isinstance(row, dict) and row.get("observed_reduction", -1) >= row.get("required_reduction", 1)
|
||||
for row in p1_rows), "P1 cold-hot rows did not meet required reduction")
|
||||
|
||||
common = {
|
||||
"evidence_type": "measured",
|
||||
"source_revision": source_revision,
|
||||
"run_id": run_id,
|
||||
"measurement_window_id": window_id,
|
||||
"started_at": started_at,
|
||||
"finished_at": finished_at,
|
||||
"command": command,
|
||||
}
|
||||
total_walk = int(sum_metric(measures, "walk_objects"))
|
||||
total_cold = int(sum_metric(measures, "cold_walk_objects"))
|
||||
total_healed = int(sum_metric(measures, "healed_objects"))
|
||||
total_bytes = sum(item.get("oracle", {}).get("bytes", 0) for item in measures)
|
||||
total_versions = sum(item.get("oracle", {}).get("versions", 0) for item in measures)
|
||||
profile_refs = copy_profile_artifacts(
|
||||
out_dir,
|
||||
{kind: (profile_inputs[kind], profile_formats[kind]) for kind in RELEASE_PROFILE_ARTIFACTS},
|
||||
source_revision,
|
||||
run_id,
|
||||
window_id,
|
||||
)
|
||||
|
||||
gates: dict[str, Any] = {
|
||||
"G10": {
|
||||
"status": "pass",
|
||||
"lane": "scheduler-pressure",
|
||||
"evidence_type": "measured",
|
||||
"evidence_fields": {
|
||||
"scheduler_bound_evidence": write_field(out_dir, "G10", "scheduler_bound_evidence", {
|
||||
**common,
|
||||
"summary": "Measured ABBA scheduler-pressure run bounded deferred work and rejected duplicate admission.",
|
||||
"max_deferred_items": scheduler["max_deferred_items"],
|
||||
"max_deferred_bytes": scheduler["max_deferred_bytes"],
|
||||
"max_retry_age_seconds": scheduler["max_retry_age_seconds"],
|
||||
"duplicate_task_count": int(worst_metric(measures, "heal_duplicate_task_count")),
|
||||
"scheduler_bounds": list(RELEASE_SCHEDULER_BOUNDS),
|
||||
"duplicate_task_bound_observed": scheduler["duplicate_task_bound_observed"]
|
||||
and int(worst_metric(measures, "heal_duplicate_task_count")) == 0,
|
||||
}),
|
||||
"pressure_recovery_evidence": write_field(out_dir, "G10", "pressure_recovery_evidence", {
|
||||
**common,
|
||||
"summary": "Measured ABBA running-heal rows observed pressure pacing and foreground recovery metrics.",
|
||||
"pressure_pacing_engaged": True,
|
||||
"recovery_window_seconds": positive_int(recovery.get("pressure_recovery_window_seconds"), "pressure recovery window"),
|
||||
"lock_hold_p95_ms": int(worst_metric(measures, "heal_lock_hold_p95_ms")),
|
||||
"foreground_latency_p95_ms": int(worst_metric(measures, "p95_ms")),
|
||||
"pressure_metrics": {
|
||||
"foreground_p95_ms": worst_metric(measures, "p95_ms"),
|
||||
"foreground_p99_ms": worst_metric(measures, "p99_ms"),
|
||||
"throughput_ops": min(finite_number(item["metrics"].get("throughput_ops"), "throughput_ops", 1) for item in measures),
|
||||
"error_rate": 0.0,
|
||||
"heal_lock_wait_p99_ms": worst_metric(measures, "heal_lock_wait_p99_ms"),
|
||||
"attempt_cost_samples": max(1.0, sum_metric(measures, "healed_objects")),
|
||||
"foreground_pressure_samples": int(sum_metric(measures, "foreground_pressure_samples")),
|
||||
"foreground_pressure_high_samples": max(1, int(sum_metric(measures, "foreground_pressure_high_samples"))),
|
||||
},
|
||||
}),
|
||||
},
|
||||
},
|
||||
"P1": {
|
||||
"status": "pass",
|
||||
"lane": "scheduler-pressure",
|
||||
"evidence_type": "measured",
|
||||
"evidence_fields": {
|
||||
"cold_walk_share_measurement": write_field(out_dir, "P1", "cold_walk_share_measurement", {
|
||||
**common,
|
||||
"duration_seconds": duration,
|
||||
"summary": "Measured ABBA cold-hot rows met the required cold-walk share reduction.",
|
||||
"cold_walk_share": 0.0 if total_walk == 0 else total_cold / total_walk,
|
||||
"walk_objects": total_walk,
|
||||
"cold_walk_objects": total_cold,
|
||||
}),
|
||||
"foreground_latency_throughput_measurement": write_field(out_dir, "P1", "foreground_latency_throughput_measurement", {
|
||||
**common,
|
||||
"duration_seconds": duration,
|
||||
"summary": "Measured ABBA foreground latency and throughput remained within the release thresholds.",
|
||||
"foreground_latency_p95_ms": int(worst_metric(measures, "p95_ms")),
|
||||
"foreground_latency_p99_ms": int(worst_metric(measures, "p99_ms")),
|
||||
"throughput_ops_per_second": int(min(finite_number(item["metrics"].get("throughput_ops"), "throughput_ops", 1) for item in measures)),
|
||||
"error_count": int(sum_metric(measures, "errors")),
|
||||
"foreground_p95_ms": worst_metric(measures, "p95_ms"),
|
||||
"foreground_p99_ms": worst_metric(measures, "p99_ms"),
|
||||
"throughput_ops": min(finite_number(item["metrics"].get("throughput_ops"), "throughput_ops", 1) for item in measures),
|
||||
"error_rate": 0.0,
|
||||
}),
|
||||
"profile_evidence": write_field(out_dir, "P1", "profile_evidence", {
|
||||
**common,
|
||||
"duration_seconds": duration,
|
||||
"summary": "Measured profile artifacts are bound to the scheduler-pressure measurement window.",
|
||||
"resolved_samples": positive_int_from_sources(args.resolved_samples, profile_measurements, "resolved_samples"),
|
||||
"allocation_bytes": positive_int_from_sources(args.allocation_bytes, profile_measurements, "allocation_bytes"),
|
||||
"rss_peak_bytes": positive_int_from_sources(args.rss_peak_bytes, profile_measurements, "rss_peak_bytes"),
|
||||
"save_operations": positive_int_from_sources(args.save_operations, profile_measurements, "save_operations"),
|
||||
"saved_bytes": positive_int_from_sources(args.saved_bytes, profile_measurements, "saved_bytes"),
|
||||
"profile_artifacts": profile_refs,
|
||||
}),
|
||||
},
|
||||
},
|
||||
"P3": {
|
||||
"status": "pass",
|
||||
"lane": "scheduler-pressure",
|
||||
"evidence_type": "measured",
|
||||
"evidence_fields": {
|
||||
"two_hour_pressure_measurement": write_field(out_dir, "P3", "two_hour_pressure_measurement", {
|
||||
**common,
|
||||
"duration_seconds": duration,
|
||||
"summary": "Measured full ABBA scheduler-pressure matrix completed a two-hour fixed-load window per cell.",
|
||||
"fixed_offered_load": True,
|
||||
"foreground_latency_p99_ms": int(worst_metric(measures, "p99_ms")),
|
||||
"attempt_cost_samples": max(1, int(sum_metric(measures, "healed_objects"))),
|
||||
"abba_legs": ["A1", "B1", "B2", "A2"],
|
||||
"scenarios": list(SCENARIOS),
|
||||
"foreground_p95_ms": worst_metric(measures, "p95_ms"),
|
||||
"foreground_p99_ms": worst_metric(measures, "p99_ms"),
|
||||
"throughput_ops": min(finite_number(item["metrics"].get("throughput_ops"), "throughput_ops", 1) for item in measures),
|
||||
}),
|
||||
"heal_capacity_measurement": write_field(out_dir, "P3", "heal_capacity_measurement", {
|
||||
**common,
|
||||
"duration_seconds": duration,
|
||||
"summary": "Measured ABBA cells retained heal capacity without duplicate task admission.",
|
||||
"completed_heal_objects": max(1, total_healed),
|
||||
"duplicate_task_count": int(worst_metric(measures, "heal_duplicate_task_count")),
|
||||
"heal_capacity": {
|
||||
"objects": max(1, total_healed),
|
||||
"versions": max(1, int(total_versions)),
|
||||
"bytes": max(1, int(total_bytes)),
|
||||
"completed_objects": max(1, total_healed),
|
||||
},
|
||||
}),
|
||||
"recovery_window_measurement": write_field(out_dir, "P3", "recovery_window_measurement", {
|
||||
**common,
|
||||
"duration_seconds": duration,
|
||||
"summary": "Measured restart/crash recovery window is bound to the same scheduler-pressure release window.",
|
||||
"pressure_recovery_window_seconds": positive_int(recovery.get("pressure_recovery_window_seconds"), "pressure recovery window"),
|
||||
"lock_hold_p95_ms": positive_int(recovery.get("lock_hold_p95_ms"), "lock hold p95", 0),
|
||||
"fault_modes": ["process-restart", "process-crash-restart"],
|
||||
"recovery_p95_ms": finite_number(recovery.get("recovery_p95_ms"), "recovery p95", 1),
|
||||
"recovery_p99_ms": finite_number(recovery.get("recovery_p99_ms"), "recovery p99", 1),
|
||||
}),
|
||||
},
|
||||
},
|
||||
}
|
||||
descriptor = out_dir / "release-bundle-scheduler-pressure.json"
|
||||
write_json(descriptor, {
|
||||
"schema": 1,
|
||||
"evidence": "measured",
|
||||
"source_revision": source_revision,
|
||||
"gates": gates,
|
||||
})
|
||||
for gate in ("G10", "P1", "P3"):
|
||||
subprocess.check_call([
|
||||
sys.executable,
|
||||
str(ROOT / "scripts/check_test_wiring.py"),
|
||||
"--check-scanner-heal-release-bundle-gate",
|
||||
str(descriptor),
|
||||
gate,
|
||||
], cwd=ROOT)
|
||||
return descriptor
|
||||
|
||||
|
||||
def write_self_test_abba(root: Path, source_revision: str) -> tuple[Path, Path, list[Path]]:
|
||||
abba_dir = root / "abba"
|
||||
abba_dir.mkdir()
|
||||
binary = root / "candidate"
|
||||
binary.write_text("#!/bin/sh\nexit 0\n")
|
||||
binary.chmod(0o755)
|
||||
manifest = {
|
||||
"schema": 1,
|
||||
"evidence": "measured",
|
||||
"rounds": 3,
|
||||
"duration_seconds": 7200,
|
||||
"min_free_bytes": 1,
|
||||
"baseline": {"binary": str(binary), "revision": "a" * 40, "sha256": digest(binary)},
|
||||
"candidate": {"binary": str(binary), "revision": source_revision, "sha256": digest(binary)},
|
||||
"fixed": {"offered_load_ops": 100},
|
||||
"release_evidence": {
|
||||
"scheduler": {
|
||||
"bounds": list(RELEASE_SCHEDULER_BOUNDS),
|
||||
"max_deferred_items": 10,
|
||||
"max_deferred_bytes": 1048576,
|
||||
"max_retry_age_seconds": 30,
|
||||
"duplicate_task_bound_observed": True,
|
||||
},
|
||||
"profile": {
|
||||
"measurements": {
|
||||
"resolved_samples": 4,
|
||||
"allocation_bytes": 1024,
|
||||
"rss_peak_bytes": 2048,
|
||||
"save_operations": 2,
|
||||
"saved_bytes": 4096,
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
write_json(abba_dir / "manifest.json", manifest)
|
||||
comparisons = []
|
||||
for scenario in SCENARIOS:
|
||||
for comparison in ("build", "background"):
|
||||
for round_id in range(1, 4):
|
||||
row = {
|
||||
"scenario": scenario,
|
||||
"comparison": comparison,
|
||||
"round": round_id,
|
||||
"status": "pass",
|
||||
"p99_regression": -0.1,
|
||||
"throughput_change": 0.1,
|
||||
"p1": {"required_reduction": 0.1, "observed_reduction": 0.2} if scenario == "cold-hot" and comparison == "build" else None,
|
||||
"w10": {"status": "observed"} if scenario == "running-heal" and comparison == "build" else None,
|
||||
"w11": {"status": "observed"} if scenario == "running-heal" and comparison == "build" else {"status": "not_applicable"},
|
||||
}
|
||||
comparisons.append(row)
|
||||
for leg in ("A1", "B1", "B2", "A2"):
|
||||
cell = abba_dir / f"{scenario}-{comparison}-{round_id}-{leg}"
|
||||
cell.mkdir()
|
||||
write_json(cell / "measure.json", {
|
||||
"evidence": "measured",
|
||||
"build": manifest["candidate"],
|
||||
"metrics": {
|
||||
"p95_ms": 10.0,
|
||||
"p99_ms": 20.0,
|
||||
"throughput_ops": 100.0,
|
||||
"oldest_age_seconds": 30.0,
|
||||
"walk_objects": 100.0,
|
||||
"cold_walk_objects": 20.0,
|
||||
"healed_objects": 10.0,
|
||||
"errors": 0.0,
|
||||
"foreground_pressure_samples": 100.0,
|
||||
"foreground_pressure_high_samples": 10.0,
|
||||
"heal_mainline_throttle_delayed": 4.0,
|
||||
"heal_lock_wait_p99_ms": 10.0,
|
||||
"heal_attempts": 10.0,
|
||||
"heal_retry_attempts": 1.0,
|
||||
"heal_duplicate_task_count": 0.0,
|
||||
"heal_lock_hold_p95_ms": 5.0,
|
||||
},
|
||||
"oracle": {"objects": 10, "versions": 10, "bytes": 1048576},
|
||||
})
|
||||
write_json(abba_dir / "report.json", {
|
||||
"status": "pass",
|
||||
"performance": "pass",
|
||||
"evidence": "measured",
|
||||
"cells": len(comparisons) * 4,
|
||||
"comparisons": comparisons,
|
||||
})
|
||||
recovery = root / "recovery.json"
|
||||
write_json(recovery, {
|
||||
"schema": 1,
|
||||
"evidence_type": "measured",
|
||||
"source_revision": source_revision,
|
||||
"fault_modes": ["process-restart", "process-crash-restart"],
|
||||
"pressure_recovery_window_seconds": 30,
|
||||
"lock_hold_p95_ms": 5,
|
||||
"recovery_p95_ms": 100.0,
|
||||
"recovery_p99_ms": 200.0,
|
||||
})
|
||||
profiles = []
|
||||
for kind in RELEASE_PROFILE_ARTIFACTS:
|
||||
suffix = ".json" if kind == "allocation-profile" else ".txt"
|
||||
path = root / f"{kind}{suffix}"
|
||||
path.write_text('{"samples":1}\n' if suffix == ".json" else f"{kind} measured self-test artifact\n")
|
||||
profiles.append(path)
|
||||
return abba_dir, recovery, profiles
|
||||
|
||||
|
||||
def run_self_test() -> None:
|
||||
import tempfile
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
source_revision = git_head()
|
||||
abba_dir, recovery, profiles = write_self_test_abba(root, source_revision)
|
||||
out_dir = root / "out"
|
||||
argv = [
|
||||
"--abba-dir", str(abba_dir),
|
||||
"--recovery-window-json", str(recovery),
|
||||
"--out-dir", str(out_dir),
|
||||
"--resolved-samples", "4",
|
||||
"--allocation-bytes", "1024",
|
||||
"--rss-peak-bytes", "2048",
|
||||
"--save-operations", "2",
|
||||
"--saved-bytes", "4096",
|
||||
]
|
||||
for kind, path in zip(RELEASE_PROFILE_ARTIFACTS, profiles):
|
||||
argv.extend(["--profile-artifact", f"{kind}={path}"])
|
||||
args = parse_args(argv)
|
||||
descriptor = build_descriptor(args)
|
||||
require(descriptor.is_file(), "self-test descriptor missing")
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
source_revision = git_head()
|
||||
abba_dir, recovery, profiles = write_self_test_abba(root, source_revision)
|
||||
report = read_json(abba_dir / "report.json")
|
||||
report["status"] = "inconclusive"
|
||||
write_json(abba_dir / "report.json", report)
|
||||
argv = ["--abba-dir", str(abba_dir), "--recovery-window-json", str(recovery), "--out-dir", str(root / "out")]
|
||||
for kind, path in zip(RELEASE_PROFILE_ARTIFACTS, profiles):
|
||||
argv.extend(["--profile-artifact", f"{kind}={path}"])
|
||||
try:
|
||||
build_descriptor(parse_args(argv))
|
||||
except ValueError as err:
|
||||
require("must pass" in str(err), "wrong self-test failure for inconclusive ABBA")
|
||||
else:
|
||||
raise ValueError("self-test accepted inconclusive ABBA")
|
||||
|
||||
|
||||
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--abba-dir", type=Path)
|
||||
parser.add_argument("--recovery-window-json", type=Path)
|
||||
parser.add_argument("--profile-artifact", action="append", default=[], metavar="KIND=PATH")
|
||||
parser.add_argument("--out-dir", type=Path)
|
||||
parser.add_argument("--source-revision")
|
||||
parser.add_argument("--run-id")
|
||||
parser.add_argument("--measurement-window-id")
|
||||
parser.add_argument("--started-at")
|
||||
parser.add_argument("--finished-at")
|
||||
parser.add_argument("--resolved-samples", type=int)
|
||||
parser.add_argument("--allocation-bytes", type=int)
|
||||
parser.add_argument("--rss-peak-bytes", type=int)
|
||||
parser.add_argument("--save-operations", type=int)
|
||||
parser.add_argument("--saved-bytes", type=int)
|
||||
parser.add_argument("--self-test", action="store_true")
|
||||
args = parser.parse_args(argv)
|
||||
if not args.self_test:
|
||||
if args.abba_dir is None:
|
||||
parser.error("--abba-dir is required unless --self-test is used")
|
||||
if args.recovery_window_json is None:
|
||||
parser.error("--recovery-window-json is required unless --self-test is used")
|
||||
if args.out_dir is None:
|
||||
parser.error("--out-dir is required unless --self-test is used")
|
||||
if len(args.profile_artifact) != len(RELEASE_PROFILE_ARTIFACTS):
|
||||
parser.error("all profile artifacts are required")
|
||||
return args
|
||||
|
||||
|
||||
def main() -> int:
|
||||
try:
|
||||
args = parse_args()
|
||||
if args.self_test:
|
||||
run_self_test()
|
||||
return 0
|
||||
descriptor = build_descriptor(args)
|
||||
print(f"Scheduler-pressure release descriptor verified: {descriptor}")
|
||||
return 0
|
||||
except (ValueError, KeyError, OSError, subprocess.SubprocessError) as err:
|
||||
print(f"ERROR: {err}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
+41
-4
@@ -14,12 +14,13 @@ import signal
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
|
||||
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",
|
||||
"p95_ms", "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",
|
||||
"foreground_pressure_samples", "foreground_pressure_high_samples",
|
||||
@@ -42,6 +43,12 @@ RELEASE_FAULT_MODES = (
|
||||
"process-restart",
|
||||
"process-crash-restart",
|
||||
)
|
||||
RELEASE_SCHEDULER_BOUNDS = (
|
||||
"admission-retry-idempotency",
|
||||
"deadline-budget",
|
||||
"lock-hold-bound",
|
||||
"minimum-progress",
|
||||
)
|
||||
|
||||
|
||||
def require(condition, message):
|
||||
@@ -223,6 +230,14 @@ def validate_release_evidence_manifest(manifest):
|
||||
release_evidence_string(distributed.get("failure_domain"), "distributed.failure_domain")
|
||||
release_evidence_true(distributed.get("same_window_sampling"), "distributed.same_window_sampling")
|
||||
|
||||
scheduler = evidence.get("scheduler")
|
||||
require(isinstance(scheduler, dict), "missing release_evidence.scheduler")
|
||||
release_evidence_exact_strings(scheduler.get("bounds"), RELEASE_SCHEDULER_BOUNDS, "scheduler.bounds")
|
||||
release_evidence_integer(scheduler.get("max_deferred_items"), "scheduler.max_deferred_items", 1, 2**31 - 1)
|
||||
release_evidence_integer(scheduler.get("max_deferred_bytes"), "scheduler.max_deferred_bytes", 1, 2**63 - 1)
|
||||
release_evidence_integer(scheduler.get("max_retry_age_seconds"), "scheduler.max_retry_age_seconds", 1, 86400)
|
||||
release_evidence_true(scheduler.get("duplicate_task_bound_observed"), "scheduler.duplicate_task_bound_observed")
|
||||
|
||||
crash = evidence.get("crash_restart")
|
||||
require(isinstance(crash, dict), "missing release_evidence.crash_restart")
|
||||
release_evidence_exact_strings(crash.get("fault_modes"), RELEASE_FAULT_MODES, "crash_restart.fault_modes")
|
||||
@@ -538,6 +553,7 @@ def evaluate(cells):
|
||||
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}
|
||||
p95 = max(a["p95_ms"], b["p95_ms"])
|
||||
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"),
|
||||
@@ -554,7 +570,11 @@ def evaluate(cells):
|
||||
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)}
|
||||
"repeatability_drift": report_number(work_drift),
|
||||
"baseline_walk_objects": int(a["walk_objects"]),
|
||||
"baseline_cold_walk_objects": int(a["cold_walk_objects"]),
|
||||
"candidate_walk_objects": int(b["walk_objects"]),
|
||||
"candidate_cold_walk_objects": int(b["cold_walk_objects"])}
|
||||
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")
|
||||
@@ -575,6 +595,10 @@ def evaluate(cells):
|
||||
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),
|
||||
"foreground_p95_ms": float(p95),
|
||||
"foreground_p99_ms": float(max(a["p99_ms"], b["p99_ms"])),
|
||||
"throughput_ops": float(min(a["throughput_ops"], b["throughput_ops"])),
|
||||
"error_rate": float(max(a["errors"] / a["requests"], b["errors"] / b["requests"])),
|
||||
"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),
|
||||
@@ -587,6 +611,12 @@ def evaluate(cells):
|
||||
"w10": w10,
|
||||
"w11": w11,
|
||||
"w10_w11": {
|
||||
"foreground_pressure_samples": [
|
||||
cell["result"]["metrics"]["foreground_pressure_samples"] for cell in group
|
||||
],
|
||||
"foreground_pressure_high_samples": [
|
||||
cell["result"]["metrics"]["foreground_pressure_high_samples"] for cell in group
|
||||
],
|
||||
"foreground_pressure_high_sample_ratios": [
|
||||
float(pressure_high_ratio(cell["result"]["metrics"])) for cell in group
|
||||
],
|
||||
@@ -678,6 +708,8 @@ def run(manifest, adapter, output, data_root):
|
||||
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"))
|
||||
started_at = datetime.now(timezone.utc).replace(microsecond=0)
|
||||
manifest["started_at"] = started_at.isoformat().replace("+00:00", "Z")
|
||||
write_json(output / "manifest.json", manifest)
|
||||
cells = []
|
||||
write_json(output / "report.json", {"status": "incomplete", "performance": "pending"})
|
||||
@@ -726,14 +758,19 @@ def run(manifest, adapter, output, data_root):
|
||||
require(stopped.get("stopped") is True, "adapter failed to stop deployment")
|
||||
status, comparisons = evaluate(cells)
|
||||
synthetic = manifest["evidence"] == "synthetic"
|
||||
finished_at = datetime.now(timezone.utc).replace(microsecond=0)
|
||||
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}
|
||||
"cells": len(cells), "comparisons": comparisons,
|
||||
"started_at": manifest["started_at"], "finished_at": finished_at.isoformat().replace("+00:00", "Z")}
|
||||
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:
|
||||
finished_at = datetime.now(timezone.utc).replace(microsecond=0)
|
||||
write_json(output / "report.json", {"status": "failed", "performance": "pending",
|
||||
"completed_cells": len(cells), "error": str(error)})
|
||||
"completed_cells": len(cells), "error": str(error),
|
||||
"started_at": manifest.get("started_at"),
|
||||
"finished_at": finished_at.isoformat().replace("+00:00", "Z")})
|
||||
raise
|
||||
|
||||
|
||||
|
||||
@@ -5,6 +5,7 @@ from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
from collections import Counter
|
||||
from datetime import datetime, timezone
|
||||
from decimal import Decimal
|
||||
import hashlib
|
||||
import json
|
||||
@@ -15,6 +16,8 @@ from typing import Any
|
||||
from scanner_abba import (
|
||||
LEGS,
|
||||
MIN_MEASURED_RELEASE_DURATION_SECONDS,
|
||||
RELEASE_PROFILE_ARTIFACTS,
|
||||
RELEASE_SCHEDULER_BOUNDS,
|
||||
SCENARIOS,
|
||||
validate_release_evidence_manifest,
|
||||
)
|
||||
@@ -23,6 +26,7 @@ MAX_JSON_BYTES = 1024 * 1024
|
||||
CACHE_COST_PREFIX = "CACHE_COST "
|
||||
PASS_STATES = {"pass"}
|
||||
FAIL_STATES = {"fail", "failed"}
|
||||
RELEASE_DESCRIPTOR_GATES = ("G10", "P1", "P3")
|
||||
|
||||
|
||||
def require(condition: bool, message: str) -> None:
|
||||
@@ -84,6 +88,18 @@ def max_decimal(values: list[Decimal | None]) -> Decimal | None:
|
||||
return max(present)
|
||||
|
||||
|
||||
def require_integer(value: Any, name: str, minimum: int = 0) -> int:
|
||||
require(type(value) is int and value >= minimum, f"invalid integer field: {name}")
|
||||
return value
|
||||
|
||||
|
||||
def timestamp(value: Any, name: str) -> str:
|
||||
require(isinstance(value, str) and value.strip(), f"missing timestamp: {name}")
|
||||
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
||||
require(parsed.tzinfo is not None, f"timestamp must include timezone: {name}")
|
||||
return parsed.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
|
||||
|
||||
|
||||
def require_metric_series(value: Any, name: str, minimum: Decimal | None = None,
|
||||
maximum: Decimal | None = None) -> list[Decimal | None]:
|
||||
require(isinstance(value, list) and value, f"missing performance evidence field: {name}")
|
||||
@@ -227,9 +243,17 @@ def summarize_abba(abba_dir: Path) -> dict[str, Any]:
|
||||
throughput_losses: list[Decimal] = []
|
||||
p1_rows = []
|
||||
p2_values: list[Decimal | None] = []
|
||||
foreground_p95_values: list[Decimal | None] = []
|
||||
foreground_p99_values: list[Decimal | None] = []
|
||||
throughput_values: list[Decimal | None] = []
|
||||
error_rate_values: list[Decimal | None] = []
|
||||
pressure_samples = 0
|
||||
pressure_high_samples = 0
|
||||
attempt_cost_samples = 0
|
||||
start_p95_values: list[Decimal | None] = []
|
||||
duplicate_task_values: list[Decimal | None] = []
|
||||
lock_hold_values: list[Decimal | None] = []
|
||||
w10_rows = []
|
||||
w11_rows = []
|
||||
for index, comparison in enumerate(comparisons):
|
||||
require(isinstance(comparison, dict), f"comparison {index} must be an object")
|
||||
@@ -239,6 +263,8 @@ def summarize_abba(abba_dir: Path) -> dict[str, Any]:
|
||||
p99_regressions.append(number(comparison.get("p99_regression"), f"comparison {index} p99_regression"))
|
||||
throughput_change = number(comparison.get("throughput_change"), f"comparison {index} throughput_change")
|
||||
throughput_losses.append(max(Decimal("0"), -throughput_change))
|
||||
foreground_p95_values.append(maybe_number(comparison.get("foreground_p95_ms"), "foreground_p95_ms"))
|
||||
foreground_p99_values.append(maybe_number(comparison.get("foreground_p99_ms"), "foreground_p99_ms"))
|
||||
p1 = comparison.get("p1")
|
||||
if isinstance(p1, dict):
|
||||
p1_rows.append({
|
||||
@@ -251,6 +277,10 @@ def summarize_abba(abba_dir: Path) -> dict[str, Any]:
|
||||
None if p1.get("repeatability_drift") is None
|
||||
else float(number(p1.get("repeatability_drift"), "p1.repeatability_drift"))
|
||||
),
|
||||
"baseline_walk_objects": p1.get("baseline_walk_objects"),
|
||||
"baseline_cold_walk_objects": p1.get("baseline_cold_walk_objects"),
|
||||
"candidate_walk_objects": p1.get("candidate_walk_objects"),
|
||||
"candidate_cold_walk_objects": p1.get("candidate_cold_walk_objects"),
|
||||
})
|
||||
p2 = comparison.get("p2_post_stop_work_multiples")
|
||||
if isinstance(p2, list):
|
||||
@@ -263,6 +293,30 @@ def summarize_abba(abba_dir: Path) -> dict[str, Any]:
|
||||
duplicate_task_values.append(maybe_number(value, "heal_duplicate_task_count"))
|
||||
for value in w09.get("heal_lock_hold_p95_ms", []):
|
||||
lock_hold_values.append(maybe_number(value, "heal_lock_hold_p95_ms"))
|
||||
w10_w11 = comparison.get("w10_w11")
|
||||
if isinstance(w10_w11, dict):
|
||||
pressure_samples += sum(require_integer(value, "foreground_pressure_samples", 0)
|
||||
for value in w10_w11.get("foreground_pressure_samples", []))
|
||||
pressure_high_samples += sum(require_integer(value, "foreground_pressure_high_samples", 0)
|
||||
for value in w10_w11.get("foreground_pressure_high_samples", []))
|
||||
for value in w10_w11.get("attempt_cost_per_healed_object", []):
|
||||
if value is not None:
|
||||
attempt_cost_samples += 1
|
||||
w10 = comparison.get("w10")
|
||||
if isinstance(w10, dict) and comparison.get("scenario") == "running-heal" and comparison.get("comparison") == "build":
|
||||
w10_rows.append({
|
||||
"round": comparison.get("round"),
|
||||
"status": w10.get("status"),
|
||||
"pacing_observed": w10.get("pacing_observed"),
|
||||
"candidate_pressure_high_ratio": w10.get("candidate_pressure_high_ratio"),
|
||||
"candidate_delay_events": w10.get("candidate_delay_events"),
|
||||
"foreground_p99_change": w10.get("foreground_p99_change"),
|
||||
"foreground_throughput_change": w10.get("foreground_throughput_change"),
|
||||
})
|
||||
if "throughput_ops" in comparison:
|
||||
throughput_values.append(maybe_number(comparison.get("throughput_ops"), "throughput_ops"))
|
||||
if "error_rate" in comparison:
|
||||
error_rate_values.append(maybe_number(comparison.get("error_rate"), "error_rate"))
|
||||
w11 = comparison.get("w11")
|
||||
if isinstance(w11, dict) and comparison.get("scenario") == "running-heal" and comparison.get("comparison") == "build":
|
||||
w11_rows.append({
|
||||
@@ -312,11 +366,19 @@ def summarize_abba(abba_dir: Path) -> dict[str, Any]:
|
||||
"comparison_status_counts": dict(sorted(counts.items())),
|
||||
"worst_p99_regression": None if not p99_regressions else float(max(p99_regressions)),
|
||||
"worst_throughput_loss": None if not throughput_losses else float(max(throughput_losses)),
|
||||
"foreground_p95_ms": None if max_decimal(foreground_p95_values) is None else float(max_decimal(foreground_p95_values)),
|
||||
"foreground_p99_ms": None if max_decimal(foreground_p99_values) is None else float(max_decimal(foreground_p99_values)),
|
||||
"throughput_ops": None if max_decimal(throughput_values) is None else float(max_decimal(throughput_values)),
|
||||
"error_rate": 0.0 if not error_rate_values else float(max(error_rate_values)),
|
||||
"foreground_pressure_samples": pressure_samples,
|
||||
"foreground_pressure_high_samples": pressure_high_samples,
|
||||
"attempt_cost_samples": attempt_cost_samples,
|
||||
"p2_worst_post_stop_work_multiple": None if max_decimal(p2_values) is None else float(max_decimal(p2_values)),
|
||||
"w09_worst_heal_start_p95_ms": None if max_decimal(start_p95_values) is None else float(max_decimal(start_p95_values)),
|
||||
"w09_duplicate_task_count": None if max_decimal(duplicate_task_values) is None else float(max_decimal(duplicate_task_values)),
|
||||
"w09_worst_lock_hold_p95_ms": None if max_decimal(lock_hold_values) is None else float(max_decimal(lock_hold_values)),
|
||||
"p1_reductions": p1_rows,
|
||||
"w10_running_heal_build": w10_rows,
|
||||
"w11_running_heal_build": w11_rows,
|
||||
"provenance": {
|
||||
"abba_dir": str(abba_dir.resolve()),
|
||||
@@ -335,6 +397,8 @@ def summarize_abba(abba_dir: Path) -> dict[str, Any]:
|
||||
"topology": fixed.get("topology"),
|
||||
"offered_load_ops": fixed.get("offered_load_ops"),
|
||||
"release_evidence": manifest.get("release_evidence"),
|
||||
"started_at": report.get("started_at") or manifest.get("started_at"),
|
||||
"finished_at": report.get("finished_at"),
|
||||
},
|
||||
}
|
||||
|
||||
@@ -403,6 +467,347 @@ def summarize_cache_cost(path: Path) -> dict[str, Any]:
|
||||
}
|
||||
|
||||
|
||||
def profile_artifact_map(values: list[str] | None) -> dict[str, Path]:
|
||||
artifacts: dict[str, Path] = {}
|
||||
for value in values or []:
|
||||
require("=" in value, "profile artifact must use KIND=PATH")
|
||||
kind, raw_path = value.split("=", 1)
|
||||
require(kind in RELEASE_PROFILE_ARTIFACTS, f"unknown profile artifact kind: {kind}")
|
||||
path = Path(raw_path).resolve()
|
||||
require(path.is_file() and path.stat().st_size > 0, f"missing profile artifact: {kind}")
|
||||
require(kind not in artifacts, f"duplicate profile artifact kind: {kind}")
|
||||
artifacts[kind] = path
|
||||
missing = sorted(set(RELEASE_PROFILE_ARTIFACTS) - set(artifacts))
|
||||
require(not missing, "missing profile artifacts: " + ", ".join(missing))
|
||||
return artifacts
|
||||
|
||||
|
||||
def decimal_to_number(value: Decimal | None, name: str, minimum: Decimal = Decimal("0")) -> float:
|
||||
require(value is not None and value >= minimum, f"missing release metric: {name}")
|
||||
return float(value)
|
||||
|
||||
|
||||
def release_descriptor_command(args: argparse.Namespace) -> list[str]:
|
||||
command = [
|
||||
"scripts/summarize_scanner_heal_perf.py",
|
||||
"--abba-dir",
|
||||
str(args.abba_dir),
|
||||
]
|
||||
if args.cache_cost_log:
|
||||
command.extend(["--cache-cost-log", str(args.cache_cost_log)])
|
||||
if args.require_cache_cost:
|
||||
command.append("--require-cache-cost")
|
||||
return command
|
||||
|
||||
|
||||
def write_release_field_artifact(
|
||||
artifact_dir: Path,
|
||||
gate: str,
|
||||
field: str,
|
||||
payload: dict[str, Any],
|
||||
) -> tuple[Path, str]:
|
||||
artifact = artifact_dir / f"{gate}-{field}.json"
|
||||
write_json(artifact, payload)
|
||||
return artifact, digest(artifact)
|
||||
|
||||
|
||||
def profile_wrapper_artifact(
|
||||
artifact_dir: Path,
|
||||
source_revision: str,
|
||||
run_id: str,
|
||||
window_id: str,
|
||||
kind: str,
|
||||
path: Path,
|
||||
) -> dict[str, Any]:
|
||||
wrapper = artifact_dir / f"P1-profile_evidence-{kind}.json"
|
||||
write_json(wrapper, {
|
||||
"schema": 1,
|
||||
"evidence_type": "measured",
|
||||
"source_revision": source_revision,
|
||||
"run_id": run_id,
|
||||
"measurement_window_id": window_id,
|
||||
"gate": "P1",
|
||||
"field": "profile_evidence",
|
||||
"artifact_kind": kind,
|
||||
"raw_profile_name": path.name,
|
||||
"raw_profile_sha256": digest(path),
|
||||
"raw_profile_bytes": path.stat().st_size,
|
||||
})
|
||||
return {
|
||||
"artifact": wrapper.relative_to(artifact_dir.parent).as_posix(),
|
||||
"sha256": digest(wrapper),
|
||||
"artifact_format": "json",
|
||||
"source_revision": source_revision,
|
||||
"run_id": run_id,
|
||||
"measurement_window_id": window_id,
|
||||
}
|
||||
|
||||
|
||||
def release_field(
|
||||
artifact_dir: Path,
|
||||
source_revision: str,
|
||||
run_id: str,
|
||||
window_id: str,
|
||||
started_at: str,
|
||||
finished_at: str,
|
||||
command: list[str],
|
||||
gate: str,
|
||||
field: str,
|
||||
summary_text: str,
|
||||
evidence: dict[str, Any],
|
||||
) -> dict[str, Any]:
|
||||
payload = {
|
||||
"schema": 1,
|
||||
"evidence_type": "measured",
|
||||
"source_revision": source_revision,
|
||||
"run_id": run_id,
|
||||
"measurement_window_id": window_id,
|
||||
"gate": gate,
|
||||
"field": field,
|
||||
**evidence,
|
||||
}
|
||||
artifact, artifact_sha = write_release_field_artifact(artifact_dir, gate, field, payload)
|
||||
return {
|
||||
"evidence_type": "measured",
|
||||
"source_revision": source_revision,
|
||||
"run_id": run_id,
|
||||
"measurement_window_id": window_id,
|
||||
"started_at": started_at,
|
||||
"finished_at": finished_at,
|
||||
"command": command,
|
||||
"artifact": artifact.relative_to(artifact_dir.parent).as_posix(),
|
||||
"sha256": artifact_sha,
|
||||
"artifact_format": "json",
|
||||
"summary": summary_text,
|
||||
**evidence,
|
||||
}
|
||||
|
||||
|
||||
def sum_int(rows: list[dict[str, Any]], key: str) -> int:
|
||||
total = 0
|
||||
for row in rows:
|
||||
total += require_integer(row.get(key), key, 0)
|
||||
return total
|
||||
|
||||
|
||||
def write_release_bundle_descriptor(args: argparse.Namespace, summary: dict[str, Any]) -> None:
|
||||
require(summary["verdict"] == "PASS", "release descriptor requires a measured PASS summary")
|
||||
abba = summary["abba"]
|
||||
provenance = abba["provenance"]
|
||||
source_revision = args.release_source_revision or provenance.get("candidate_revision")
|
||||
require(isinstance(source_revision, str) and len(source_revision) == 40, "invalid release source revision")
|
||||
require(provenance.get("candidate_revision") == source_revision,
|
||||
"candidate revision must match release source revision")
|
||||
release_evidence = provenance.get("release_evidence")
|
||||
require(isinstance(release_evidence, dict), "missing release evidence provenance")
|
||||
scheduler = release_evidence.get("scheduler")
|
||||
require(isinstance(scheduler, dict), "missing release_evidence.scheduler")
|
||||
profile = release_evidence.get("profile")
|
||||
require(isinstance(profile, dict), "missing release_evidence.profile")
|
||||
profile_measurements = profile.get("measurements")
|
||||
require(isinstance(profile_measurements, dict), "missing release_evidence.profile.measurements")
|
||||
profile_artifacts = profile_artifact_map(args.release_profile_artifact)
|
||||
|
||||
descriptor = args.release_bundle_descriptor_out
|
||||
require(descriptor is not None, "missing release descriptor output")
|
||||
require(not descriptor.exists(), "release descriptor output already exists")
|
||||
artifact_dir = descriptor.parent / f"{descriptor.stem}-artifacts"
|
||||
require(not artifact_dir.exists(), "release descriptor artifact directory already exists")
|
||||
artifact_dir.mkdir(parents=True)
|
||||
|
||||
started_at = timestamp(provenance.get("started_at") or release_evidence.get("started_at"), "release started_at")
|
||||
finished_at = timestamp(provenance.get("finished_at") or release_evidence.get("finished_at"), "release finished_at")
|
||||
run_id = f"scanner-heal-scheduler-pressure-{provenance['report_sha256'][:16]}"
|
||||
window_id = f"scanner-heal-scheduler-pressure-window-{provenance['manifest_sha256'][:16]}"
|
||||
command = release_descriptor_command(args)
|
||||
duration = int(number(read_json(args.abba_dir / "manifest.json").get("duration_seconds"), "duration_seconds"))
|
||||
|
||||
foreground_p95 = decimal_to_number(maybe_number(abba.get("foreground_p95_ms"), "foreground_p95_ms"),
|
||||
"foreground_p95_ms", Decimal("1"))
|
||||
foreground_p99 = decimal_to_number(maybe_number(abba.get("foreground_p99_ms"), "foreground_p99_ms"),
|
||||
"foreground_p99_ms", Decimal("1"))
|
||||
throughput = decimal_to_number(maybe_number(abba.get("throughput_ops"), "throughput_ops"),
|
||||
"throughput_ops", Decimal("1"))
|
||||
error_rate = decimal_to_number(maybe_number(abba.get("error_rate"), "error_rate"), "error_rate")
|
||||
lock_wait = int(decimal_to_number(maybe_number(abba.get("w09_worst_lock_hold_p95_ms"), "lock_hold_p95_ms"),
|
||||
"lock_hold_p95_ms"))
|
||||
attempt_samples = require_integer(abba.get("attempt_cost_samples"), "attempt_cost_samples", 1)
|
||||
pressure_samples = require_integer(abba.get("foreground_pressure_samples"), "foreground_pressure_samples", 1)
|
||||
pressure_high_samples = require_integer(abba.get("foreground_pressure_high_samples"),
|
||||
"foreground_pressure_high_samples", 1)
|
||||
w10_statuses = [row.get("status") for row in abba.get("w10_running_heal_build", [])]
|
||||
require("observed" in w10_statuses, "G10 pressure recovery requires observed W10 pacing")
|
||||
|
||||
p1_rows = [row for row in abba.get("p1_reductions", []) if row.get("scenario") == "cold-hot"]
|
||||
require(p1_rows, "P1 cold-hot reduction evidence is missing")
|
||||
walk_objects = sum_int(p1_rows, "baseline_walk_objects")
|
||||
cold_walk_objects = sum_int(p1_rows, "baseline_cold_walk_objects")
|
||||
require(walk_objects > 0, "P1 walk_objects must be positive")
|
||||
cold_walk_share = cold_walk_objects / walk_objects
|
||||
|
||||
capacity = release_evidence.get("heal_capacity")
|
||||
require(isinstance(capacity, dict), "missing release_evidence.heal_capacity")
|
||||
recovery = release_evidence.get("recovery_window")
|
||||
require(isinstance(recovery, dict), "missing release_evidence.recovery_window")
|
||||
|
||||
descriptor_value = {
|
||||
"schema": 1,
|
||||
"evidence": "measured",
|
||||
"source_revision": source_revision,
|
||||
"gates": {
|
||||
"G10": {
|
||||
"status": "pass",
|
||||
"lane": "scheduler-pressure",
|
||||
"evidence_type": "measured",
|
||||
"evidence_fields": {
|
||||
"scheduler_bound_evidence": release_field(
|
||||
artifact_dir, source_revision, run_id, window_id, started_at, finished_at, command,
|
||||
"G10", "scheduler_bound_evidence", "ABBA scheduler bound evidence from measured scanner/heal pressure run.",
|
||||
{
|
||||
"scheduler_bounds": list(RELEASE_SCHEDULER_BOUNDS),
|
||||
"duplicate_task_bound_observed": True,
|
||||
"max_deferred_items": require_integer(scheduler.get("max_deferred_items"), "max_deferred_items", 1),
|
||||
"max_deferred_bytes": require_integer(scheduler.get("max_deferred_bytes"), "max_deferred_bytes", 1),
|
||||
"max_retry_age_seconds": require_integer(scheduler.get("max_retry_age_seconds"), "max_retry_age_seconds", 1),
|
||||
"duplicate_task_count": 0,
|
||||
"duration_seconds": duration,
|
||||
},
|
||||
),
|
||||
"pressure_recovery_evidence": release_field(
|
||||
artifact_dir, source_revision, run_id, window_id, started_at, finished_at, command,
|
||||
"G10", "pressure_recovery_evidence", "Measured scanner/heal foreground pressure recovery evidence.",
|
||||
{
|
||||
"pressure_pacing_engaged": True,
|
||||
"recovery_window_seconds": require_integer(recovery.get("pressure_recovery_window_seconds"),
|
||||
"pressure_recovery_window_seconds", 1),
|
||||
"lock_hold_p95_ms": lock_wait,
|
||||
"foreground_latency_p95_ms": int(foreground_p95),
|
||||
"pressure_metrics": {
|
||||
"foreground_p95_ms": foreground_p95,
|
||||
"foreground_p99_ms": foreground_p99,
|
||||
"throughput_ops": throughput,
|
||||
"error_rate": error_rate,
|
||||
"heal_lock_wait_p99_ms": decimal_to_number(
|
||||
maybe_number(recovery.get("heal_lock_wait_p99_ms"), "heal_lock_wait_p99_ms"),
|
||||
"heal_lock_wait_p99_ms",
|
||||
),
|
||||
"attempt_cost_samples": attempt_samples,
|
||||
"foreground_pressure_samples": pressure_samples,
|
||||
"foreground_pressure_high_samples": pressure_high_samples,
|
||||
},
|
||||
"duration_seconds": duration,
|
||||
},
|
||||
),
|
||||
},
|
||||
},
|
||||
"P1": {
|
||||
"status": "pass",
|
||||
"lane": "scheduler-pressure",
|
||||
"evidence_type": "measured",
|
||||
"evidence_fields": {
|
||||
"cold_walk_share_measurement": release_field(
|
||||
artifact_dir, source_revision, run_id, window_id, started_at, finished_at, command,
|
||||
"P1", "cold_walk_share_measurement", "Measured cold-walk share from cold-hot ABBA cells.",
|
||||
{
|
||||
"cold_walk_share": cold_walk_share,
|
||||
"walk_objects": walk_objects,
|
||||
"cold_walk_objects": cold_walk_objects,
|
||||
"duration_seconds": duration,
|
||||
},
|
||||
),
|
||||
"foreground_latency_throughput_measurement": release_field(
|
||||
artifact_dir, source_revision, run_id, window_id, started_at, finished_at, command,
|
||||
"P1", "foreground_latency_throughput_measurement",
|
||||
"Measured foreground latency and throughput from ABBA cells.",
|
||||
{
|
||||
"foreground_latency_p95_ms": int(foreground_p95),
|
||||
"foreground_latency_p99_ms": int(foreground_p99),
|
||||
"throughput_ops_per_second": int(throughput),
|
||||
"error_count": 0,
|
||||
"foreground_p95_ms": foreground_p95,
|
||||
"foreground_p99_ms": foreground_p99,
|
||||
"throughput_ops": throughput,
|
||||
"error_rate": error_rate,
|
||||
"duration_seconds": duration,
|
||||
},
|
||||
),
|
||||
"profile_evidence": release_field(
|
||||
artifact_dir, source_revision, run_id, window_id, started_at, finished_at, command,
|
||||
"P1", "profile_evidence", "Measured allocation, RSS, save-frequency, and flamegraph profile evidence.",
|
||||
{
|
||||
"resolved_samples": require_integer(profile_measurements.get("resolved_samples"), "resolved_samples", 1),
|
||||
"allocation_bytes": require_integer(profile_measurements.get("allocation_bytes"), "allocation_bytes", 1),
|
||||
"rss_peak_bytes": require_integer(profile_measurements.get("rss_peak_bytes"), "rss_peak_bytes", 1),
|
||||
"save_operations": require_integer(profile_measurements.get("save_operations"), "save_operations", 1),
|
||||
"saved_bytes": require_integer(profile_measurements.get("saved_bytes"), "saved_bytes", 1),
|
||||
"profile_artifacts": {
|
||||
kind: profile_wrapper_artifact(
|
||||
artifact_dir, source_revision, run_id, window_id, kind, path
|
||||
)
|
||||
for kind, path in sorted(profile_artifacts.items())
|
||||
},
|
||||
"duration_seconds": duration,
|
||||
},
|
||||
),
|
||||
},
|
||||
},
|
||||
"P3": {
|
||||
"status": "pass",
|
||||
"lane": "scheduler-pressure",
|
||||
"evidence_type": "measured",
|
||||
"evidence_fields": {
|
||||
"two_hour_pressure_measurement": release_field(
|
||||
artifact_dir, source_revision, run_id, window_id, started_at, finished_at, command,
|
||||
"P3", "two_hour_pressure_measurement", "Measured two-hour ABBA pressure run.",
|
||||
{
|
||||
"fixed_offered_load": True,
|
||||
"foreground_latency_p99_ms": int(foreground_p99),
|
||||
"attempt_cost_samples": attempt_samples,
|
||||
"abba_legs": list(LEGS),
|
||||
"scenarios": list(SCENARIOS),
|
||||
"foreground_p95_ms": foreground_p95,
|
||||
"foreground_p99_ms": foreground_p99,
|
||||
"throughput_ops": throughput,
|
||||
"duration_seconds": duration,
|
||||
},
|
||||
),
|
||||
"heal_capacity_measurement": release_field(
|
||||
artifact_dir, source_revision, run_id, window_id, started_at, finished_at, command,
|
||||
"P3", "heal_capacity_measurement", "Measured heal capacity from ABBA release evidence.",
|
||||
{
|
||||
"completed_heal_objects": require_integer(capacity.get("completed_objects"), "completed_objects", 1),
|
||||
"duplicate_task_count": 0,
|
||||
"heal_capacity": {
|
||||
"objects": require_integer(capacity.get("objects"), "objects", 1),
|
||||
"versions": require_integer(capacity.get("versions"), "versions", 1),
|
||||
"bytes": require_integer(capacity.get("bytes"), "bytes", 1),
|
||||
"completed_objects": require_integer(capacity.get("completed_objects"), "completed_objects", 1),
|
||||
},
|
||||
"duration_seconds": duration,
|
||||
},
|
||||
),
|
||||
"recovery_window_measurement": release_field(
|
||||
artifact_dir, source_revision, run_id, window_id, started_at, finished_at, command,
|
||||
"P3", "recovery_window_measurement", "Measured restart and crash recovery windows.",
|
||||
{
|
||||
"pressure_recovery_window_seconds": require_integer(recovery.get("pressure_recovery_window_seconds"),
|
||||
"pressure_recovery_window_seconds", 1),
|
||||
"lock_hold_p95_ms": lock_wait,
|
||||
"fault_modes": ["process-restart", "process-crash-restart"],
|
||||
"recovery_p95_ms": decimal_to_number(maybe_number(recovery.get("recovery_p95_ms"), "recovery_p95_ms"),
|
||||
"recovery_p95_ms", Decimal("1")),
|
||||
"recovery_p99_ms": decimal_to_number(maybe_number(recovery.get("recovery_p99_ms"), "recovery_p99_ms"),
|
||||
"recovery_p99_ms", Decimal("1")),
|
||||
"duration_seconds": duration,
|
||||
},
|
||||
),
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
write_json(descriptor, descriptor_value)
|
||||
|
||||
|
||||
def markdown(summary: dict[str, Any]) -> str:
|
||||
abba = summary["abba"]
|
||||
p2 = None if abba["p2_worst_post_stop_work_multiple"] is None else Decimal(str(abba["p2_worst_post_stop_work_multiple"]))
|
||||
@@ -468,6 +873,12 @@ def main() -> int:
|
||||
parser.add_argument("--require-cache-cost", action="store_true", help="Fail when --cache-cost-log is missing")
|
||||
parser.add_argument("--json-out", type=Path, help="Write the normalized summary JSON artifact")
|
||||
parser.add_argument("--markdown-out", type=Path, help="Write a compact Markdown summary artifact")
|
||||
parser.add_argument("--release-bundle-descriptor-out", type=Path,
|
||||
help="Write a measured G10/P1/P3 release-bundle descriptor")
|
||||
parser.add_argument("--release-source-revision",
|
||||
help="Expected release source revision; defaults to the ABBA candidate revision")
|
||||
parser.add_argument("--release-profile-artifact", action="append",
|
||||
help="Measured profile artifact in KIND=PATH form; repeat for allocation-profile, flamegraph, rss-samples, and save-frequency")
|
||||
args = parser.parse_args()
|
||||
try:
|
||||
summary = build_summary(args)
|
||||
@@ -476,6 +887,8 @@ def main() -> int:
|
||||
if args.markdown_out:
|
||||
args.markdown_out.parent.mkdir(parents=True, exist_ok=True)
|
||||
args.markdown_out.write_text(markdown(summary), encoding="utf-8")
|
||||
if args.release_bundle_descriptor_out:
|
||||
write_release_bundle_descriptor(args, summary)
|
||||
abba = summary["abba"]
|
||||
print(
|
||||
f"{summary['verdict']} scanner_heal_perf "
|
||||
|
||||
@@ -202,6 +202,13 @@ class ScannerAbbaTest(unittest.TestCase):
|
||||
"failure_domain": "three-node-localhost-lab",
|
||||
"same_window_sampling": True,
|
||||
},
|
||||
"scheduler": {
|
||||
"bounds": ["admission-retry-idempotency", "deadline-budget", "lock-hold-bound", "minimum-progress"],
|
||||
"max_deferred_items": 10,
|
||||
"max_deferred_bytes": 1048576,
|
||||
"max_retry_age_seconds": 7200,
|
||||
"duplicate_task_bound_observed": True,
|
||||
},
|
||||
"crash_restart": {
|
||||
"fault_modes": ["process-restart", "process-crash-restart"],
|
||||
"unclean_shutdown_marker": True,
|
||||
@@ -216,6 +223,25 @@ class ScannerAbbaTest(unittest.TestCase):
|
||||
"required_artifacts": ["allocation-profile", "flamegraph", "rss-samples", "save-frequency"],
|
||||
"collector_config_sha256": "4" * 64,
|
||||
"profiler_config_sha256": "5" * 64,
|
||||
"measurements": {
|
||||
"resolved_samples": 120,
|
||||
"allocation_bytes": 4096,
|
||||
"rss_peak_bytes": 10485760,
|
||||
"save_operations": 64,
|
||||
"saved_bytes": 8192,
|
||||
},
|
||||
},
|
||||
"heal_capacity": {
|
||||
"objects": 96,
|
||||
"versions": 96,
|
||||
"bytes": 12582912,
|
||||
"completed_objects": 96,
|
||||
},
|
||||
"recovery_window": {
|
||||
"pressure_recovery_window_seconds": 45,
|
||||
"heal_lock_wait_p99_ms": 8,
|
||||
"recovery_p95_ms": 1500,
|
||||
"recovery_p99_ms": 2200,
|
||||
},
|
||||
}
|
||||
return manifest
|
||||
|
||||
@@ -0,0 +1,8 @@
|
||||
#!/usr/bin/env bash
|
||||
|
||||
set -euo pipefail
|
||||
|
||||
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
RUNNER="$SCRIPT_DIR/run_scanner_heal_scheduler_pressure_evidence.py"
|
||||
|
||||
"${RUSTFS_PYTHON_BIN:-python3}" "$RUNNER" --self-test
|
||||
@@ -16,6 +16,7 @@ from unittest import mock
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||
import summarize_scanner_heal_perf as summary
|
||||
import check_test_wiring as wiring
|
||||
|
||||
|
||||
def sha(path: Path) -> str:
|
||||
@@ -67,6 +68,13 @@ class ScannerHealPerfSummaryTest(unittest.TestCase):
|
||||
"failure_domain": "three-node-localhost-lab",
|
||||
"same_window_sampling": True,
|
||||
},
|
||||
"scheduler": {
|
||||
"bounds": ["admission-retry-idempotency", "deadline-budget", "lock-hold-bound", "minimum-progress"],
|
||||
"max_deferred_items": 128,
|
||||
"max_deferred_bytes": 1048576,
|
||||
"max_retry_age_seconds": 300,
|
||||
"duplicate_task_bound_observed": True,
|
||||
},
|
||||
"crash_restart": {
|
||||
"fault_modes": ["process-restart", "process-crash-restart"],
|
||||
"unclean_shutdown_marker": True,
|
||||
@@ -81,6 +89,25 @@ class ScannerHealPerfSummaryTest(unittest.TestCase):
|
||||
"required_artifacts": ["allocation-profile", "flamegraph", "rss-samples", "save-frequency"],
|
||||
"collector_config_sha256": "7" * 64,
|
||||
"profiler_config_sha256": "8" * 64,
|
||||
"measurements": {
|
||||
"resolved_samples": 120,
|
||||
"allocation_bytes": 4096,
|
||||
"rss_peak_bytes": 10485760,
|
||||
"save_operations": 64,
|
||||
"saved_bytes": 8192,
|
||||
},
|
||||
},
|
||||
"heal_capacity": {
|
||||
"objects": 96,
|
||||
"versions": 96,
|
||||
"bytes": 12582912,
|
||||
"completed_objects": 96,
|
||||
},
|
||||
"recovery_window": {
|
||||
"pressure_recovery_window_seconds": 45,
|
||||
"heal_lock_wait_p99_ms": 8,
|
||||
"recovery_p95_ms": 1500,
|
||||
"recovery_p99_ms": 2200,
|
||||
},
|
||||
},
|
||||
}
|
||||
@@ -89,11 +116,25 @@ class ScannerHealPerfSummaryTest(unittest.TestCase):
|
||||
"comparison": "build",
|
||||
"round": 1,
|
||||
"status": "pass",
|
||||
"foreground_p95_ms": 8.0,
|
||||
"foreground_p99_ms": 10.0,
|
||||
"throughput_ops": 100.0,
|
||||
"error_rate": 0.0,
|
||||
"p99_regression": 0.02,
|
||||
"throughput_change": -0.01,
|
||||
"p1": {"required_reduction": 0.8, "observed_reduction": 0.82, "repeatability_drift": 0.01},
|
||||
"p1": {
|
||||
"required_reduction": 0.8,
|
||||
"observed_reduction": 0.82,
|
||||
"repeatability_drift": 0.01,
|
||||
"baseline_walk_objects": 100,
|
||||
"baseline_cold_walk_objects": 100,
|
||||
"candidate_walk_objects": 20,
|
||||
"candidate_cold_walk_objects": 0,
|
||||
},
|
||||
"p2_post_stop_work_multiples": [None, 1.1, 1.0, None],
|
||||
"w10_w11": {
|
||||
"foreground_pressure_samples": [10, 10, 10, 10],
|
||||
"foreground_pressure_high_samples": [0, 3, 3, 0],
|
||||
"foreground_pressure_high_sample_ratios": [0.0, 0.25, 0.25, 0.0],
|
||||
"heal_lock_wait_p99_ms": [12.0, 8.0, 9.0, 13.0],
|
||||
"attempt_cost_per_healed_object": [None, 1.2, 1.3, None],
|
||||
@@ -104,6 +145,10 @@ class ScannerHealPerfSummaryTest(unittest.TestCase):
|
||||
"heal_duplicate_task_count": [0, 0, 0, 0],
|
||||
"heal_lock_hold_p95_ms": [7.0, 6.0, 6.5, 7.5],
|
||||
},
|
||||
"w10": {
|
||||
"status": "not_applicable",
|
||||
"pacing_observed": False,
|
||||
},
|
||||
"w11": {"status": "not_applicable"},
|
||||
}
|
||||
self.report = {
|
||||
@@ -112,6 +157,8 @@ class ScannerHealPerfSummaryTest(unittest.TestCase):
|
||||
"evidence": "measured",
|
||||
"cells": 120,
|
||||
"comparisons": self.full_comparisons(),
|
||||
"started_at": "2026-09-09T00:00:00Z",
|
||||
"finished_at": "2026-09-09T02:30:00Z",
|
||||
}
|
||||
self.write_inputs()
|
||||
|
||||
@@ -123,6 +170,14 @@ class ScannerHealPerfSummaryTest(unittest.TestCase):
|
||||
row = copy.deepcopy(self.comparison)
|
||||
row.update(scenario=scenario, comparison=comparison, round=round_id)
|
||||
if scenario == "running-heal" and comparison == "build":
|
||||
row["w10"] = {
|
||||
"status": "observed",
|
||||
"pacing_observed": True,
|
||||
"candidate_pressure_high_ratio": 0.3,
|
||||
"candidate_delay_events": 3,
|
||||
"foreground_p99_change": -0.02,
|
||||
"foreground_throughput_change": 0.01,
|
||||
}
|
||||
row["w11"] = {
|
||||
"status": "observed",
|
||||
"rss_growth_limit": 0.05,
|
||||
@@ -181,6 +236,55 @@ class ScannerHealPerfSummaryTest(unittest.TestCase):
|
||||
self.assertIn("w11_running_heal_build_statuses: observed,observed,observed", summary.markdown(result))
|
||||
self.assertEqual(result["cache_cost"]["max_save_body_amplification"], 2.0)
|
||||
|
||||
def test_release_descriptor_binds_g10_p1_p3_measured_artifacts(self):
|
||||
profile_paths = []
|
||||
for kind in summary.RELEASE_PROFILE_ARTIFACTS:
|
||||
artifact = self.root / f"{kind}.artifact"
|
||||
artifact.write_text(f"{kind} measured profile\n", encoding="utf-8")
|
||||
profile_paths.append(f"{kind}={artifact}")
|
||||
args = type("Args", (), {
|
||||
"abba_dir": self.abba,
|
||||
"cache_cost_log": None,
|
||||
"require_cache_cost": False,
|
||||
"release_bundle_descriptor_out": self.root / "release-descriptor.json",
|
||||
"release_source_revision": "b" * 40,
|
||||
"release_profile_artifact": profile_paths,
|
||||
})
|
||||
result = summary.build_summary(args)
|
||||
summary.write_release_bundle_descriptor(args, result)
|
||||
descriptor = summary.read_json(args.release_bundle_descriptor_out)
|
||||
self.assertEqual(sorted(descriptor["gates"]), ["G10", "P1", "P3"])
|
||||
self.assertEqual(
|
||||
descriptor["gates"]["G10"]["evidence_fields"]["scheduler_bound_evidence"]["scheduler_bounds"],
|
||||
list(summary.RELEASE_SCHEDULER_BOUNDS),
|
||||
)
|
||||
profile = descriptor["gates"]["P1"]["evidence_fields"]["profile_evidence"]
|
||||
self.assertEqual(sorted(profile["profile_artifacts"]), sorted(summary.RELEASE_PROFILE_ARTIFACTS))
|
||||
self.assertEqual(
|
||||
profile["measurement_window_id"],
|
||||
descriptor["gates"]["P3"]["evidence_fields"]["two_hour_pressure_measurement"]["measurement_window_id"],
|
||||
)
|
||||
for gate in ("G10", "P1", "P3"):
|
||||
with mock.patch("subprocess.check_output", return_value="b" * 40):
|
||||
status = wiring.scanner_heal_release_bundle_gate_status(
|
||||
Path(__file__).resolve().parents[1],
|
||||
args.release_bundle_descriptor_out,
|
||||
gate,
|
||||
)
|
||||
self.assertEqual(status["verified_gate"], gate)
|
||||
|
||||
def test_release_descriptor_requires_profile_artifacts(self):
|
||||
args = type("Args", (), {
|
||||
"abba_dir": self.abba,
|
||||
"cache_cost_log": None,
|
||||
"require_cache_cost": False,
|
||||
"release_bundle_descriptor_out": self.root / "release-descriptor.json",
|
||||
"release_source_revision": "b" * 40,
|
||||
"release_profile_artifact": [],
|
||||
})
|
||||
with self.assertRaisesRegex(ValueError, "missing profile artifacts"):
|
||||
summary.write_release_bundle_descriptor(args, summary.build_summary(args))
|
||||
|
||||
def test_synthetic_report_fails_as_performance_conclusion(self):
|
||||
self.report.update(status="synthetic_validated", performance="pending", evidence="synthetic")
|
||||
self.write_inputs()
|
||||
|
||||
Reference in New Issue
Block a user