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

test(scanner): measure heal pacing and cache cost

Browse Source 
Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
houseme
2026-09-08 01:26:43 +08:00
parent 7c85c72fd1
commit 0a40f85802
3 changed files with 153 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/scanner_abba.py
+70
View File
@@ -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
scripts/test_scanner_abba.py
+65 -1
View File
@@ -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)