gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-09 05:36:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

test(scanner): require heal retry stats evidence (#7514)

Browse Source 
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
houseme
2026-09-08 22:52:34 +08:00
committed by GitHub
parent 128080ceb9
commit 33a8469e27
4 changed files with 108 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/scanner_abba.py
+15
View File
@@ -26,6 +26,7 @@ METRICS = (
"heal_mainline_throttle_delayed",
"heal_lock_wait_p99_ms", "heal_attempts", "heal_attempt_failures",
"heal_retry_attempts",
"heal_start_p95_ms", "heal_duplicate_task_count", "heal_lock_hold_p95_ms",
)
REPEATABILITY_LIMIT = Decimal("0.05")
P2_WORK_MULTIPLE_LIMIT = Decimal("1.2")
@@ -382,6 +383,9 @@ def validate_result(result, request, expected):
"foreground pressure high samples exceed samples")
require(metrics["heal_attempt_failures"] <= metrics["heal_attempts"], "heal failures exceed attempts")
require(metrics["heal_retry_attempts"] <= metrics["heal_attempts"], "heal retries exceed attempts")
require(metrics["heal_duplicate_task_count"] == 0, "duplicate heal task admission")
require(metrics["heal_start_p95_ms"] > 0, "zero heal start p95")
require(metrics["heal_lock_hold_p95_ms"] > 0, "zero heal lock hold p95")
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")
@@ -550,6 +554,17 @@ def evaluate(cells):
"candidate_attempt_cost_per_healed_object": (
None if not candidate_attempt_costs else float(max(candidate_attempt_costs))
),
},
"w09": {
"heal_start_p95_ms": [
cell["result"]["metrics"]["heal_start_p95_ms"] for cell in group
],
"heal_duplicate_task_count": [
cell["result"]["metrics"]["heal_duplicate_task_count"] for cell in group
],
"heal_lock_hold_p95_ms": [
cell["result"]["metrics"]["heal_lock_hold_p95_ms"] for cell in group
],
}})
return ("fail" if failed else "inconclusive" if inconclusive else "pass"), comparisons
scripts/summarize_scanner_heal_perf.py
+43
View File
@@ -125,6 +125,32 @@ def require_measured_comparison_evidence(comparison: dict[str, Any], index: int)
)
require(candidate_attempt_cost is None or candidate_attempt_cost >= 0,
f"comparison {index} candidate attempt cost below minimum")
w09 = comparison.get("w09")
require(isinstance(w09, dict), f"comparison {index} missing W09 evidence")
start_p95 = require_metric_series(
w09.get("heal_start_p95_ms"),
f"comparison {index} heal_start_p95_ms",
Decimal("0"),
)
require(all(item is not None and item > 0 for item in start_p95),
f"comparison {index} heal_start_p95_ms must be measured")
duplicate_tasks = require_metric_series(
w09.get("heal_duplicate_task_count"),
f"comparison {index} heal_duplicate_task_count",
Decimal("0"),
Decimal("0"),
)
require(all(item is not None and item == 0 for item in duplicate_tasks),
f"comparison {index} heal_duplicate_task_count must be measured")
lock_hold = require_metric_series(
w09.get("heal_lock_hold_p95_ms"),
f"comparison {index} heal_lock_hold_p95_ms",
Decimal("0"),
)
require(all(item is not None and item > 0 for item in lock_hold),
f"comparison {index} heal_lock_hold_p95_ms must be measured")
require(len(start_p95) == len(duplicate_tasks) == len(lock_hold),
f"comparison {index} W09 evidence length mismatch")
def require_complete_abba_matrix(manifest: dict[str, Any], report: dict[str, Any], comparisons: list[dict[str, Any]]) -> None:
@@ -177,6 +203,9 @@ def summarize_abba(abba_dir: Path) -> dict[str, Any]:
throughput_losses: list[Decimal] = []
p1_rows = []
p2_values: list[Decimal | None] = []
start_p95_values: list[Decimal | None] = []
duplicate_task_values: list[Decimal | None] = []
lock_hold_values: list[Decimal | None] = []
for index, comparison in enumerate(comparisons):
require(isinstance(comparison, dict), f"comparison {index} must be an object")
state = comparison.get("status")
@@ -201,6 +230,14 @@ def summarize_abba(abba_dir: Path) -> dict[str, Any]:
p2 = comparison.get("p2_post_stop_work_multiples")
if isinstance(p2, list):
p2_values.extend(maybe_number(value, "p2_post_stop_work_multiple") for value in p2)
w09 = comparison.get("w09")
if isinstance(w09, dict):
for value in w09.get("heal_start_p95_ms", []):
start_p95_values.append(maybe_number(value, "heal_start_p95_ms"))
for value in w09.get("heal_duplicate_task_count", []):
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"))
measured = report.get("evidence") == "measured"
passed = report_state in PASS_STATES and performance_state in PASS_STATES and measured
@@ -239,6 +276,9 @@ def summarize_abba(abba_dir: Path) -> dict[str, Any]:
"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)),
"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,
"provenance": {
"abba_dir": str(abba_dir.resolve()),
@@ -339,6 +379,9 @@ def markdown(summary: dict[str, Any]) -> str:
f"- worst_p99_regression: {pct(p99)}",
f"- worst_throughput_loss: {pct(throughput)}",
f"- p2_worst_post_stop_work_multiple: {ratio(p2)}",
f"- w09_worst_heal_start_p95_ms: {abba['w09_worst_heal_start_p95_ms'] if abba['w09_worst_heal_start_p95_ms'] is not None else 'pending'}",
f"- w09_duplicate_task_count: {abba['w09_duplicate_task_count'] if abba['w09_duplicate_task_count'] is not None else 'pending'}",
f"- w09_worst_lock_hold_p95_ms: {abba['w09_worst_lock_hold_p95_ms'] if abba['w09_worst_lock_hold_p95_ms'] is not None else 'pending'}",
]
if abba.get("completed_cells") is not None:
lines.append(f"- completed_cells: {abba['completed_cells']}")
scripts/test_scanner_abba.py
+17 -3
View File
@@ -56,7 +56,8 @@ def fake_adapter():
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"])
healed_objects=request["expected_healed_objects"],
heal_duplicate_task_count=0)
result["convergence"] = {"writes_stopped": True, "last_mutation_observed": True,
"first_complete_publication": True, "last_mutation_time": 1,
"last_mutation_observed_time": 2,
@@ -106,6 +107,10 @@ def fake_adapter():
result["metrics"]["foreground_pressure_high_samples"] = result["metrics"]["foreground_pressure_samples"] + 1
elif fault == "attempt-accounting":
result["metrics"]["heal_attempt_failures"] = result["metrics"]["heal_attempts"] + 1
elif fault == "duplicate-heal-task":
result["metrics"]["heal_duplicate_task_count"] = 1
elif fault == "missing-start-p95":
result["metrics"]["heal_start_p95_ms"] = 0
elif fault == "pacing-benefit" and request["scenario"] == "running-heal" \
and request["comparison"] == "build" and request["leg"].startswith("B"):
result["metrics"].update(p99_ms=9, heal_mainline_throttle_delayed=5)
@@ -365,12 +370,20 @@ class ScannerAbbaTest(unittest.TestCase):
expected_attempt_cost = [None, 1.0, 1.0, None] if comparison["comparison"] == "background" else [1.0, 1.0, 1.0, 1.0]
self.assertEqual(w10_w11["attempt_cost_per_healed_object"], expected_attempt_cost)
self.assertEqual(w10_w11["candidate_attempt_cost_per_healed_object"], 1.0)
self.assertEqual(
comparison["w09"],
{
"heal_start_p95_ms": [10, 10, 10, 10],
"heal_duplicate_task_count": [0, 0, 0, 0],
"heal_lock_hold_p95_ms": [10, 10, 10, 10],
},
)
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",
"zero-pressure-samples", "pressure-sample-order", "attempt-accounting",
"missing-pacing-metric"):
"missing-pacing-metric", "duplicate-heal-task", "missing-start-p95"):
with self.subTest(fault=fault), tempfile.TemporaryDirectory() as directory:
self.root = Path(directory)
with self.assertRaises((ValueError, OSError, subprocess.SubprocessError)):
@@ -616,7 +629,8 @@ class ScannerAbbaTest(unittest.TestCase):
}
metrics = dict.fromkeys(harness.METRICS, 10)
metrics.update(p99_ms=10, throughput_ops=100, errors=0, requests=100,
walk_objects=100, cold_walk_objects=20, healed_objects=10)
walk_objects=100, cold_walk_objects=20, healed_objects=10,
heal_duplicate_task_count=0)
result = {
"evidence": request["evidence"],
"fixed": request["fixed"],
scripts/test_summarize_scanner_heal_perf.py
+33
View File
@@ -99,6 +99,11 @@ class ScannerHealPerfSummaryTest(unittest.TestCase):
"attempt_cost_per_healed_object": [None, 1.2, 1.3, None],
"candidate_attempt_cost_per_healed_object": 1.3,
},
"w09": {
"heal_start_p95_ms": [42.0, 40.0, 41.0, 43.0],
"heal_duplicate_task_count": [0, 0, 0, 0],
"heal_lock_hold_p95_ms": [7.0, 6.0, 6.5, 7.5],
},
}
self.report = {
"status": "pass",
@@ -150,6 +155,8 @@ class ScannerHealPerfSummaryTest(unittest.TestCase):
result = summary.build_summary(args)
self.assertEqual(result["verdict"], "PASS")
self.assertEqual(result["abba"]["provenance"]["manifest_sha256"], sha(self.abba / "manifest.json"))
self.assertEqual(result["abba"]["w09_duplicate_task_count"], 0.0)
self.assertEqual(result["abba"]["w09_worst_heal_start_p95_ms"], 43.0)
self.assertEqual(result["cache_cost"]["max_save_body_amplification"], 2.0)
def test_synthetic_report_fails_as_performance_conclusion(self):
@@ -267,6 +274,32 @@ class ScannerHealPerfSummaryTest(unittest.TestCase):
with self.assertRaisesRegex(ValueError, "W10/W11|performance evidence|length mismatch|above maximum"):
summary.build_summary(args)
def test_passing_abba_report_requires_w09_evidence(self):
cases = {
"missing": lambda row: row.pop("w09"),
"start": lambda row: row["w09"].pop("heal_start_p95_ms"),
"zero start": lambda row: row["w09"].update(heal_start_p95_ms=[0, 40.0, 41.0, 43.0]),
"duplicates": lambda row: row["w09"].update(heal_duplicate_task_count=[0, 1, 0, 0]),
"unknown duplicates": lambda row: row["w09"].update(heal_duplicate_task_count=[None, 0, 0, 0]),
"lock": lambda row: row["w09"].pop("heal_lock_hold_p95_ms"),
"zero lock": lambda row: row["w09"].update(heal_lock_hold_p95_ms=[0, 6.0, 6.5, 7.5]),
"length": lambda row: row["w09"].update(heal_lock_hold_p95_ms=[1]),
}
for name, mutate in cases.items():
with self.subTest(fault=name):
self.setUp()
mutate(self.report["comparisons"][0])
self.write_inputs()
args = type("Args", (), {
"abba_dir": self.abba,
"cache_cost_log": None,
"require_cache_cost": False,
"json_out": None,
"markdown_out": None,
})
with self.assertRaisesRegex(ValueError, "W09|performance evidence|above maximum|length mismatch|must be measured"):
summary.build_summary(args)
def test_passing_measured_report_requires_release_evidence_manifest(self):
del self.manifest["release_evidence"]
self.write_inputs()