From 0a40f85802dec7b4c23b20e4b327a88aeb7da636 Mon Sep 17 00:00:00 2001 From: houseme Date: Tue, 8 Sep 2026 01:26:43 +0800 Subject: [PATCH] test(scanner): measure heal pacing and cache cost Co-Authored-By: heihutu Co-Authored-By: zhi22915 --- docs/operations/scanner-benchmark-runbook.md | 18 +++++ scripts/scanner_abba.py | 70 ++++++++++++++++++++ scripts/test_scanner_abba.py | 66 +++++++++++++++++- 3 files changed, 153 insertions(+), 1 deletion(-) diff --git a/docs/operations/scanner-benchmark-runbook.md b/docs/operations/scanner-benchmark-runbook.md index 04f06aee4..5b0350417 100644 --- a/docs/operations/scanner-benchmark-runbook.md +++ b/docs/operations/scanner-benchmark-runbook.md @@ -125,6 +125,11 @@ 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`. +The adapter also reports the measurement-window delta of +`rustfs_heal_mainline_throttle_total{source="admin",result="delayed"}` as +`heal_mainline_throttle_delayed`; a cumulative process-lifetime value is not a +valid input. +The clone, encode, and save byte fields are also deltas from the same window. 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 @@ -132,6 +137,19 @@ 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. +Each comparison records a `w22` section with clone, encode, and save bytes per +walked object, clone/encode and save/encode byte ratios, and candidate changes. +These are traffic amplification indicators, not allocation attribution or an +fsync profile. The `running-heal` build comparison also records a `w10` +section. `status=observed` requires sampled high foreground pressure, at least +one admin pacing delay in the same window, and an improvement in either +foreground p99 or throughput. `no_measured_benefit` means pacing ran but neither +foreground metric improved; `pending` means the run did not prove that pacing +engaged; `inconclusive` means ABBA repeatability failed. Baseline and candidate +delay counts are both retained so an operator can reject unrelated or +process-lifetime counter contamination. Correct repair oracles and the existing +regression limits still apply in every case. + 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`, diff --git a/scripts/scanner_abba.py b/scripts/scanner_abba.py index 9e6c33dbb..a76643b3a 100644 --- a/scripts/scanner_abba.py +++ b/scripts/scanner_abba.py @@ -23,6 +23,7 @@ METRICS = ( "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", + "heal_mainline_throttle_delayed", "heal_lock_wait_p99_ms", "heal_attempts", "heal_attempt_failures", "heal_retry_attempts", ) @@ -59,6 +60,12 @@ def relative_change(current, baseline, name): return ratio(current, baseline, name) - Decimal("1") +def relative_change_or_none(current, baseline, name): + if decimal_number(baseline, f"{name} baseline") == 0: + return None + return relative_change(current, baseline, name) + + def repeatability_change(first, second, name): first = decimal_number(first, name) second = decimal_number(second, name) @@ -283,6 +290,62 @@ def pressure_high_ratio(metrics): metrics["foreground_pressure_samples"], "foreground pressure high samples") +def scanner_cache_cost(metrics): + walked = decimal_number(metrics["walk_objects"], "walk_objects") + encoded = decimal_number(metrics["encode_bytes"], "encode_bytes") + return { + "clone_bytes_per_walk_object": None if walked == 0 else float(ratio(metrics["cache_clone_bytes"], walked, "clone bytes per walk object")), + "encode_bytes_per_walk_object": None if walked == 0 else float(ratio(encoded, walked, "encode bytes per walk object")), + "save_bytes_per_walk_object": None if walked == 0 else float(ratio(metrics["save_bytes"], walked, "save bytes per walk object")), + "clone_to_encode_byte_ratio": None if encoded == 0 else float(ratio(metrics["cache_clone_bytes"], encoded, "clone to encode bytes")), + "save_to_encode_byte_amplification": None if encoded == 0 else float(ratio(metrics["save_bytes"], encoded, "save to encode bytes")), + } + + +def scanner_cache_cost_change(candidate, baseline): + changes = {} + for key in ("cache_clone_bytes", "encode_bytes", "save_bytes"): + change = relative_change_or_none(candidate[key], baseline[key], key) + changes[f"{key}_change"] = None if change is None else float(change) + return changes + + +def running_heal_pacing(group, baseline, candidate, p99, throughput, noisy): + if group[0]["scenario"] != "running-heal" or group[0]["comparison"] != "build": + return None + baseline_seconds = sum(decimal_number(cell["result"]["elapsed_seconds"], "elapsed_seconds") for cell in (group[0], group[3])) / 2 + candidate_seconds = sum(decimal_number(cell["result"]["elapsed_seconds"], "elapsed_seconds") for cell in (group[1], group[2])) / 2 + baseline_rate = ratio(baseline["heal_attempts"], baseline_seconds, "baseline heal attempt rate") + candidate_rate = ratio(candidate["heal_attempts"], candidate_seconds, "candidate heal attempt rate") + rate_change = relative_change_or_none(candidate_rate, baseline_rate, "heal attempt rate") + candidate_high_ratio = pressure_high_ratio(candidate) + baseline_delayed = decimal_number(baseline["heal_mainline_throttle_delayed"], "baseline pacing delays") + delayed = decimal_number(candidate["heal_mainline_throttle_delayed"], "candidate pacing delays") + pacing_observed = candidate_high_ratio > 0 and delayed > 0 + foreground_improved = p99 < 0 or throughput > 0 + status = ( + "inconclusive" + if noisy + else "observed" + if pacing_observed and foreground_improved + else "no_measured_benefit" + if pacing_observed + else "pending" + ) + return { + "status": status, + "pacing_observed": pacing_observed, + "candidate_pressure_high_ratio": float(candidate_high_ratio), + "baseline_delay_events": float(baseline_delayed), + "candidate_delay_events": float(delayed), + "baseline_heal_attempts_per_second": float(baseline_rate), + "candidate_heal_attempts_per_second": float(candidate_rate), + "heal_attempt_rate_change": None if rate_change is None else float(rate_change), + "foreground_p99_change": float(p99), + "foreground_throughput_change": float(throughput), + } + + 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: @@ -342,6 +405,7 @@ def evaluate(cells): candidate_attempt_costs = [ value for cell, value in zip(group, attempt_costs) if cell["leg"].startswith("B") and value is not None ] + w10 = running_heal_pacing(group, a, b, p99, throughput, noise) inconclusive |= noise or p2_pending if not noise and not passed: failed = True @@ -352,6 +416,12 @@ def evaluate(cells): "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_report, + "w22": { + "baseline": scanner_cache_cost(a), + "candidate": scanner_cache_cost(b), + "candidate_vs_baseline": scanner_cache_cost_change(b, a), + }, + "w10": w10, "w10_w11": { "foreground_pressure_high_sample_ratios": [ float(pressure_high_ratio(cell["result"]["metrics"])) for cell in group diff --git a/scripts/test_scanner_abba.py b/scripts/test_scanner_abba.py index c982589b7..427311bbb 100755 --- a/scripts/test_scanner_abba.py +++ b/scripts/test_scanner_abba.py @@ -94,6 +94,8 @@ def fake_adapter(): result["metrics"].update(walk_objects=100, cold_walk_objects=0) elif fault == "missing-metric": del result["metrics"]["save_bytes"] + elif fault == "missing-pacing-metric": + del result["metrics"]["heal_mainline_throttle_delayed"] elif fault == "incomplete-repair": result["metrics"]["healed_objects"] = 0 elif fault == "zero-pressure-samples": @@ -102,6 +104,12 @@ 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 == "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) + elif fault == "pacing-pending" and request["scenario"] == "running-heal" \ + and request["comparison"] == "build" and request["leg"].startswith("B"): + result["metrics"]["heal_mainline_throttle_delayed"] = 0 harness.write_json(Path(output_path), result) return 0 @@ -278,6 +286,31 @@ class ScannerAbbaTest(unittest.TestCase): 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"])) for comparison in report["comparisons"]: + w22 = comparison["w22"] + self.assertEqual(w22["baseline"]["save_to_encode_byte_amplification"], 1.0) + self.assertEqual(w22["candidate"]["clone_to_encode_byte_ratio"], 1.0) + self.assertEqual( + w22["candidate_vs_baseline"], + {"cache_clone_bytes_change": 0.0, "encode_bytes_change": 0.0, "save_bytes_change": 0.0}, + ) + if comparison["scenario"] == "running-heal" and comparison["comparison"] == "build": + self.assertEqual( + comparison["w10"], + { + "status": "no_measured_benefit", + "pacing_observed": True, + "candidate_pressure_high_ratio": 1.0, + "baseline_delay_events": 10.0, + "candidate_delay_events": 10.0, + "baseline_heal_attempts_per_second": 10.0, + "candidate_heal_attempts_per_second": 10.0, + "heal_attempt_rate_change": 0.0, + "foreground_p99_change": 0.0, + "foreground_throughput_change": 0.0, + }, + ) + else: + self.assertIsNone(comparison["w10"]) w10_w11 = comparison["w10_w11"] self.assertEqual(w10_w11["foreground_pressure_high_sample_ratios"], [1.0, 1.0, 1.0, 1.0]) self.assertEqual(w10_w11["heal_lock_wait_p99_ms"], [10, 10, 10, 10]) @@ -288,7 +321,8 @@ class ScannerAbbaTest(unittest.TestCase): 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"): + "zero-pressure-samples", "pressure-sample-order", "attempt-accounting", + "missing-pacing-metric"): with self.subTest(fault=fault), tempfile.TemporaryDirectory() as directory: self.root = Path(directory) with self.assertRaises((ValueError, OSError, subprocess.SubprocessError)): @@ -302,6 +336,36 @@ class ScannerAbbaTest(unittest.TestCase): self.assertEqual(self.run_harness("noise"), 3) self.assertEqual(harness.read_json(self.root / "out/report.json")["status"], "inconclusive") + def test_noisy_running_heal_does_not_claim_pacing_benefit(self): + with patch.object(harness, "SCENARIOS", ("running-heal",)): + self.assertEqual(self.run_harness("noise"), 3) + comparisons = harness.read_json(self.root / "out/report.json")["comparisons"] + build = next(comparison for comparison in comparisons if comparison["comparison"] == "build") + self.assertEqual(build["w10"]["status"], "inconclusive") + + def test_idle_cache_window_reports_unavailable_ratios(self): + metrics = dict.fromkeys(harness.METRICS, 0) + self.assertEqual( + harness.scanner_cache_cost(metrics), + { + "clone_bytes_per_walk_object": None, + "encode_bytes_per_walk_object": None, + "save_bytes_per_walk_object": None, + "clone_to_encode_byte_ratio": None, + "save_to_encode_byte_amplification": None, + }, + ) + + def test_running_heal_pacing_status_requires_engagement_and_benefit(self): + for fault, expected in (("pacing-benefit", "observed"), ("pacing-pending", "pending")): + with self.subTest(fault=fault), tempfile.TemporaryDirectory() as directory: + self.root = Path(directory) + with patch.object(harness, "SCENARIOS", ("running-heal",)): + self.assertEqual(self.run_harness(fault), 0) + comparisons = harness.read_json(self.root / "out/report.json")["comparisons"] + build = next(comparison for comparison in comparisons if comparison["comparison"] == "build") + self.assertEqual(build["w10"]["status"], expected) + def test_missing_first_publication_is_inconclusive(self): with patch.object(harness, "SCENARIOS", ("cold-hot",)): self.assertEqual(self.run_harness("no-publication"), 3)