mirror of
https://github.com/rustfs/rustfs.git
synced 2026-09-07 20:46:11 +00:00
Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| e82083de02 | |||
| 056f41a15e | |||
| 4dddcec1ba |
@@ -516,7 +516,6 @@ async fn submit_mrf_heal_request(manager: &HealManager, intent: &MrfIntent) -> c
|
||||
|
||||
struct MrfRuntime {
|
||||
queue: MrfQueue,
|
||||
retained_replay_intents: Vec<MrfIntent>,
|
||||
config: MrfConsumerConfig,
|
||||
new_since_flush: usize,
|
||||
/// True while the in-memory pending set has changed since the last
|
||||
@@ -536,7 +535,7 @@ impl MrfRuntime {
|
||||
fn snapshot(&self) -> (Vec<u8>, Vec<u8>) {
|
||||
let mut authoritative = Vec::new();
|
||||
let mut legacy = Vec::new();
|
||||
for intent in self.retained_replay_intents.iter().chain(self.queue.intents()) {
|
||||
for intent in self.queue.intents() {
|
||||
let scoped_identity =
|
||||
!matches!(intent.kind, rustfs_common::mrf_channel::MrfKind::MetadataCorruption) && intent.scope.is_some();
|
||||
if !encode_intent(intent, &mut authoritative) {
|
||||
@@ -674,11 +673,10 @@ pub async fn replay_journal_once(manager: &Arc<HealManager>) -> usize {
|
||||
struct ReplayOutcome {
|
||||
replayed: usize,
|
||||
journal_on_disk: bool,
|
||||
retained_replay_intents: Vec<MrfIntent>,
|
||||
}
|
||||
|
||||
fn replay_must_retain_journal(rearm_incomplete: bool, pending_depth: usize, retained_replay_depth: usize) -> bool {
|
||||
rearm_incomplete || pending_depth > 0 || retained_replay_depth > 0
|
||||
fn replay_must_retain_journal(rearm_incomplete: bool, pending_depth: usize) -> bool {
|
||||
rearm_incomplete || pending_depth > 0
|
||||
}
|
||||
|
||||
/// Shared replay core: read + decode + re-arm, then drain what fits. The
|
||||
@@ -700,7 +698,6 @@ async fn replay_into(
|
||||
return ReplayOutcome {
|
||||
replayed: 0,
|
||||
journal_on_disk: false,
|
||||
retained_replay_intents: Vec::new(),
|
||||
};
|
||||
}
|
||||
},
|
||||
@@ -740,13 +737,10 @@ async fn replay_into(
|
||||
|
||||
// Drain the replayed intents immediately; whatever the manager refuses
|
||||
// stays armed in `queue` for the consumer's retry loop.
|
||||
let mut retained_replay_intents = Vec::new();
|
||||
if backoff_until.is_none() {
|
||||
while let Some(mut intent) = queue.pop_front() {
|
||||
match submit_mrf_heal_request(manager, &intent).await {
|
||||
Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => {
|
||||
retained_replay_intents.push(intent);
|
||||
}
|
||||
Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => {}
|
||||
Ok(HealAdmissionResult::Full) | Ok(HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull)) => {
|
||||
intent.attempts = intent.attempts.saturating_add(1);
|
||||
if intent.attempts < MRF_MAX_ATTEMPTS {
|
||||
@@ -775,7 +769,7 @@ async fn replay_into(
|
||||
}
|
||||
}
|
||||
}
|
||||
let journal_on_disk = if replay_must_retain_journal(rearm_incomplete, queue.depth(), retained_replay_intents.len()) {
|
||||
let journal_on_disk = if replay_must_retain_journal(rearm_incomplete, queue.depth()) {
|
||||
true
|
||||
} else {
|
||||
!delete_journals().await
|
||||
@@ -783,7 +777,6 @@ async fn replay_into(
|
||||
ReplayOutcome {
|
||||
replayed,
|
||||
journal_on_disk,
|
||||
retained_replay_intents,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -793,7 +786,6 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
|
||||
let config = MrfConsumerConfig::default();
|
||||
let mut runtime = MrfRuntime {
|
||||
queue: MrfQueue::new(config.queue_capacity, config.journal_max_bytes),
|
||||
retained_replay_intents: Vec::new(),
|
||||
config: config.clone(),
|
||||
new_since_flush: 0,
|
||||
dirty: false,
|
||||
@@ -805,7 +797,6 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
|
||||
// on disk whenever any replayed intent still needs a successor snapshot.
|
||||
let replay = replay_into(&manager, &mut runtime.queue, &mut runtime.backoff_until).await;
|
||||
runtime.journal_on_disk = replay.journal_on_disk;
|
||||
runtime.retained_replay_intents = replay.retained_replay_intents;
|
||||
// Anything still pending (e.g. the manager was full and backoff armed)
|
||||
// must be re-persisted by the next flush before replay can delete the
|
||||
// startup anchor.
|
||||
@@ -823,7 +814,7 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
|
||||
// provably current AND idle (a dirty or pending state
|
||||
// gets one last persist attempt, matching the shutdown
|
||||
// retry the unconditional flush used to provide).
|
||||
if runtime.dirty || runtime.queue.depth() > 0 || !runtime.retained_replay_intents.is_empty() {
|
||||
if runtime.dirty || runtime.queue.depth() > 0 {
|
||||
runtime.flush().await;
|
||||
}
|
||||
tracing::info!(
|
||||
@@ -852,7 +843,6 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
|
||||
match tick_action(
|
||||
runtime.dirty,
|
||||
runtime.queue.depth(),
|
||||
runtime.retained_replay_intents.len(),
|
||||
runtime.journal_on_disk,
|
||||
) {
|
||||
TickAction::Flush => {
|
||||
@@ -867,8 +857,8 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
|
||||
runtime.dispatch(manager.as_ref()).await;
|
||||
}
|
||||
TickAction::DeleteJournal => {
|
||||
// Only remove a stale journal after every replayed
|
||||
// intent has a durable successor proof.
|
||||
// All replayed intents have either been accepted,
|
||||
// merged, or replaced by a pending successor snapshot.
|
||||
if delete_journals().await {
|
||||
runtime.journal_on_disk = false;
|
||||
gauge!("rustfs_heal_mrf_journal_bytes").set(0.0);
|
||||
@@ -897,13 +887,11 @@ enum TickAction {
|
||||
Idle,
|
||||
}
|
||||
|
||||
fn tick_action(dirty: bool, depth: usize, retained_replay_depth: usize, journal_on_disk: bool) -> TickAction {
|
||||
fn tick_action(dirty: bool, depth: usize, journal_on_disk: bool) -> TickAction {
|
||||
if dirty {
|
||||
TickAction::Flush
|
||||
} else if depth > 0 {
|
||||
TickAction::Retry
|
||||
} else if retained_replay_depth > 0 {
|
||||
TickAction::Idle
|
||||
} else if journal_on_disk {
|
||||
TickAction::DeleteJournal
|
||||
} else {
|
||||
@@ -936,73 +924,34 @@ mod tests {
|
||||
|
||||
// Dirty dominates: a changed pending set flushes even when idle
|
||||
// otherwise.
|
||||
assert!(matches!(tick_action(true, 0, 0, false), Flush));
|
||||
assert!(matches!(tick_action(true, 3, 0, true), Flush));
|
||||
assert!(matches!(tick_action(true, 0, false), Flush));
|
||||
assert!(matches!(tick_action(true, 3, true), Flush));
|
||||
|
||||
// Clean backlog: no rewrite, but keep draining so an expired
|
||||
// admission backoff retries on time.
|
||||
assert!(matches!(tick_action(false, 1, 0, false), Retry));
|
||||
assert!(matches!(tick_action(false, 2, 0, true), Retry));
|
||||
|
||||
// Replayed records accepted by the manager are still restart anchors
|
||||
// until a durable successor proof can tombstone them.
|
||||
assert!(matches!(tick_action(false, 0, 1, true), Idle));
|
||||
assert!(matches!(tick_action(false, 1, false), Retry));
|
||||
assert!(matches!(tick_action(false, 2, true), Retry));
|
||||
|
||||
// Quiescent with a stale journal file on disk: remove it.
|
||||
assert!(matches!(tick_action(false, 0, 0, true), DeleteJournal));
|
||||
assert!(matches!(tick_action(false, 0, true), DeleteJournal));
|
||||
|
||||
// Fully quiescent: nothing to do.
|
||||
assert!(matches!(tick_action(false, 0, 0, false), Idle));
|
||||
assert!(matches!(tick_action(false, 0, false), Idle));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn replay_cleanup_retains_journal_for_unarmed_or_refused_records() {
|
||||
assert!(
|
||||
replay_must_retain_journal(true, 0, 0),
|
||||
replay_must_retain_journal(true, 0),
|
||||
"a rejected replay record still needs its disk anchor"
|
||||
);
|
||||
assert!(
|
||||
replay_must_retain_journal(false, 1, 0),
|
||||
replay_must_retain_journal(false, 1),
|
||||
"a Full admission retry must keep the startup journal until the next snapshot"
|
||||
);
|
||||
assert!(
|
||||
replay_must_retain_journal(false, 0, 1),
|
||||
"an accepted replay record still needs a durable successor before cleanup"
|
||||
);
|
||||
assert!(
|
||||
!replay_must_retain_journal(false, 0, 0),
|
||||
"only a fully consumed replay snapshot with no retained anchors may be deleted"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn retained_replay_anchor_remains_in_successor_snapshot() {
|
||||
let retained = intent("accepted-replay", "object", 0);
|
||||
let mut runtime = MrfRuntime {
|
||||
queue: MrfQueue::new(8, 8192),
|
||||
retained_replay_intents: vec![retained.clone()],
|
||||
config: MrfConsumerConfig::default(),
|
||||
new_since_flush: 0,
|
||||
dirty: false,
|
||||
journal_on_disk: true,
|
||||
backoff_until: None,
|
||||
};
|
||||
assert_eq!(
|
||||
runtime.queue.try_push_typed(intent("new-pending", "object", 0)),
|
||||
MrfQueuePushResult::Enqueued
|
||||
);
|
||||
|
||||
let (authoritative, legacy) = runtime.snapshot();
|
||||
let (decoded, truncated) = decode_journal(&authoritative);
|
||||
let (legacy_decoded, legacy_truncated) = decode_journal(&legacy);
|
||||
|
||||
assert_eq!(truncated, 0);
|
||||
assert_eq!(legacy_truncated, 0);
|
||||
assert_eq!(decoded.len(), 2);
|
||||
assert_eq!(legacy_decoded.len(), 2);
|
||||
assert!(
|
||||
decoded.iter().any(|intent| intent.bucket == retained.bucket),
|
||||
"accepted replay anchor must remain crash-replayable"
|
||||
!replay_must_retain_journal(false, 0),
|
||||
"only a fully consumed replay snapshot may be deleted"
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
@@ -171,6 +171,22 @@ performance acceptance gate. Run the fake-adapter self-tests with:
|
||||
scripts/test_scanner_validation_harness.sh
|
||||
```
|
||||
|
||||
For CI summaries, PR evidence tables, and operator handoff, collapse the raw
|
||||
matrix into a quiet one-line verdict plus durable JSON/Markdown artifacts:
|
||||
|
||||
```bash
|
||||
scripts/summarize_scanner_heal_perf.py \
|
||||
--abba-dir /path/to/new-artifacts \
|
||||
--cache-cost-log /path/to/cache-cost-profile.log \
|
||||
--json-out /path/to/new-artifacts/perf-summary.json \
|
||||
--markdown-out /path/to/new-artifacts/perf-summary.md
|
||||
```
|
||||
|
||||
The command prints only `PASS scanner_heal_perf ...` for measured passing ABBA
|
||||
evidence, otherwise `FAIL scanner_heal_perf ...`. The JSON and Markdown outputs
|
||||
carry the key p99/throughput/P1/P2/cache-cost fields and artifact provenance
|
||||
hashes; raw per-cell logs remain in the original artifact tree for audit.
|
||||
|
||||
They cover the complete 120-cell schedule, data isolation, missing builds and
|
||||
oracles, zero samples/requests, swallowed request errors, offered-load drift,
|
||||
incomplete repairs, missing metrics, noise, and P1/P2/p99 regressions. A real
|
||||
|
||||
+3
-3
@@ -496,9 +496,9 @@ impl From<StorageError> for ApiError {
|
||||
|
||||
let message = if matches!(&err, StorageError::QuotaExceeded { .. }) {
|
||||
err.to_string()
|
||||
} else if matches!(&err, StorageError::MaxVersionsExceeded) {
|
||||
ApiError::error_code_to_message(&code)
|
||||
} else if code == S3ErrorCode::InternalError && matches!(&err, StorageError::Io(_)) {
|
||||
} else if matches!(&err, StorageError::MaxVersionsExceeded)
|
||||
|| (code == S3ErrorCode::InternalError && matches!(&err, StorageError::Io(_)))
|
||||
{
|
||||
ApiError::error_code_to_message(&code)
|
||||
} else if code == S3ErrorCode::InternalError {
|
||||
err.to_string()
|
||||
|
||||
Executable
+311
@@ -0,0 +1,311 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Summarize Scanner/Heal ABBA and cache-cost profile artifacts quietly."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
from collections import Counter
|
||||
from decimal import Decimal
|
||||
import hashlib
|
||||
import json
|
||||
from pathlib import Path
|
||||
import sys
|
||||
from typing import Any
|
||||
|
||||
MAX_JSON_BYTES = 1024 * 1024
|
||||
CACHE_COST_PREFIX = "CACHE_COST "
|
||||
PASS_STATES = {"pass"}
|
||||
FAIL_STATES = {"fail", "failed"}
|
||||
|
||||
|
||||
def require(condition: bool, message: str) -> None:
|
||||
if not condition:
|
||||
raise ValueError(message)
|
||||
|
||||
|
||||
def read_json(path: Path) -> dict[str, Any]:
|
||||
require(path.is_file(), f"missing JSON artifact: {path}")
|
||||
require(path.stat().st_size <= MAX_JSON_BYTES, f"oversized JSON artifact: {path}")
|
||||
with path.open(encoding="utf-8") as stream:
|
||||
value = json.load(stream)
|
||||
require(isinstance(value, dict), f"expected JSON object: {path}")
|
||||
return value
|
||||
|
||||
|
||||
def write_json(path: Path, value: dict[str, Any]) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(json.dumps(value, indent=2, sort_keys=True, allow_nan=False) + "\n", encoding="utf-8")
|
||||
|
||||
|
||||
def digest(path: Path) -> str:
|
||||
with path.open("rb") as stream:
|
||||
if hasattr(hashlib, "file_digest"):
|
||||
return hashlib.file_digest(stream, "sha256").hexdigest()
|
||||
hasher = hashlib.sha256()
|
||||
while chunk := stream.read(1024 * 1024):
|
||||
hasher.update(chunk)
|
||||
return hasher.hexdigest()
|
||||
|
||||
|
||||
def number(value: Any, name: str) -> Decimal:
|
||||
require(type(value) in (float, int), f"invalid numeric field: {name}")
|
||||
return Decimal(str(value))
|
||||
|
||||
|
||||
def maybe_number(value: Any, name: str) -> Decimal | None:
|
||||
if value is None:
|
||||
return None
|
||||
return number(value, name)
|
||||
|
||||
|
||||
def pct(value: Decimal | None) -> str:
|
||||
if value is None:
|
||||
return "pending"
|
||||
return f"{float(value * Decimal('100')):.2f}%"
|
||||
|
||||
|
||||
def ratio(value: Decimal | None) -> str:
|
||||
if value is None:
|
||||
return "pending"
|
||||
return f"{float(value):.3f}x"
|
||||
|
||||
|
||||
def max_decimal(values: list[Decimal | None]) -> Decimal | None:
|
||||
present = [value for value in values if value is not None]
|
||||
if not present:
|
||||
return None
|
||||
return max(present)
|
||||
|
||||
|
||||
def summarize_abba(abba_dir: Path) -> dict[str, Any]:
|
||||
manifest_path = abba_dir / "manifest.json"
|
||||
report_path = abba_dir / "report.json"
|
||||
manifest = read_json(manifest_path)
|
||||
report = read_json(report_path)
|
||||
comparisons = report.get("comparisons")
|
||||
require(isinstance(comparisons, list), "report.comparisons must be a list")
|
||||
|
||||
counts = Counter()
|
||||
p99_regressions: list[Decimal] = []
|
||||
throughput_losses: list[Decimal] = []
|
||||
p1_rows = []
|
||||
p2_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")
|
||||
require(isinstance(state, str) and state, f"comparison {index} missing status")
|
||||
counts[state] += 1
|
||||
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))
|
||||
p1 = comparison.get("p1")
|
||||
if isinstance(p1, dict):
|
||||
p1_rows.append({
|
||||
"scenario": comparison.get("scenario"),
|
||||
"comparison": comparison.get("comparison"),
|
||||
"round": comparison.get("round"),
|
||||
"required_reduction": float(number(p1.get("required_reduction"), "p1.required_reduction")),
|
||||
"observed_reduction": float(number(p1.get("observed_reduction"), "p1.observed_reduction")),
|
||||
"repeatability_drift": (
|
||||
None if p1.get("repeatability_drift") is None
|
||||
else float(number(p1.get("repeatability_drift"), "p1.repeatability_drift"))
|
||||
),
|
||||
})
|
||||
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)
|
||||
|
||||
report_state = report.get("status")
|
||||
performance_state = report.get("performance")
|
||||
require(isinstance(report_state, str) and report_state, "report.status missing")
|
||||
require(isinstance(performance_state, str) and performance_state, "report.performance missing")
|
||||
measured = report.get("evidence") == "measured"
|
||||
passed = report_state in PASS_STATES and performance_state in PASS_STATES and measured
|
||||
gate_state = "pass" if passed else "fail"
|
||||
if report_state == "synthetic_validated":
|
||||
reason = "synthetic evidence validates the harness only; measured performance remains pending"
|
||||
elif report_state not in PASS_STATES:
|
||||
reason = f"ABBA report status is {report_state}"
|
||||
elif performance_state not in PASS_STATES:
|
||||
reason = f"performance status is {performance_state}"
|
||||
elif not measured:
|
||||
reason = "measured evidence is required for a performance conclusion"
|
||||
else:
|
||||
reason = "measured ABBA report passed"
|
||||
|
||||
fixed = manifest.get("fixed", {})
|
||||
require(isinstance(fixed, dict), "manifest.fixed must be an object")
|
||||
return {
|
||||
"gate_state": gate_state,
|
||||
"reason": reason,
|
||||
"status": report_state,
|
||||
"performance": performance_state,
|
||||
"evidence": report.get("evidence"),
|
||||
"cells": report.get("cells", 0),
|
||||
"comparisons_total": len(comparisons),
|
||||
"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)),
|
||||
"p2_worst_post_stop_work_multiple": None if max_decimal(p2_values) is None else float(max_decimal(p2_values)),
|
||||
"p1_reductions": p1_rows,
|
||||
"provenance": {
|
||||
"abba_dir": str(abba_dir.resolve()),
|
||||
"manifest_sha256": digest(manifest_path),
|
||||
"report_sha256": digest(report_path),
|
||||
"baseline_revision": manifest.get("baseline", {}).get("revision"),
|
||||
"baseline_sha256": manifest.get("baseline", {}).get("sha256"),
|
||||
"candidate_revision": manifest.get("candidate", {}).get("revision"),
|
||||
"candidate_sha256": manifest.get("candidate", {}).get("sha256"),
|
||||
"adapter_sha256": manifest.get("adapter_sha256"),
|
||||
"collector_sha256": manifest.get("collector_sha256"),
|
||||
"config_sha256": fixed.get("config_sha256"),
|
||||
"dataset_sha256": fixed.get("dataset_sha256"),
|
||||
"release_flags": fixed.get("release_flags"),
|
||||
"durability": fixed.get("durability"),
|
||||
"topology": fixed.get("topology"),
|
||||
"offered_load_ops": fixed.get("offered_load_ops"),
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def cache_cost_records(path: Path) -> list[dict[str, Any]]:
|
||||
require(path.is_file(), f"missing cache-cost log: {path}")
|
||||
records = []
|
||||
with path.open(encoding="utf-8", errors="replace") as stream:
|
||||
for line_no, line in enumerate(stream, start=1):
|
||||
if CACHE_COST_PREFIX not in line:
|
||||
continue
|
||||
payload = line.split(CACHE_COST_PREFIX, 1)[1].strip()
|
||||
value = json.loads(payload)
|
||||
require(isinstance(value, dict), f"cache-cost line {line_no} is not a JSON object")
|
||||
require(value.get("schema") == 1, f"cache-cost line {line_no} has unsupported schema")
|
||||
records.append(value)
|
||||
require(records, f"no {CACHE_COST_PREFIX.strip()} records found in {path}")
|
||||
return records
|
||||
|
||||
|
||||
def summarize_cache_cost(path: Path) -> dict[str, Any]:
|
||||
records = cache_cost_records(path)
|
||||
scenarios = Counter()
|
||||
max_wire = Decimal("0")
|
||||
max_save_amp = Decimal("0")
|
||||
max_clone_ns = Decimal("0")
|
||||
max_encode_ns = Decimal("0")
|
||||
max_save_ns = Decimal("0")
|
||||
build_sources = set()
|
||||
for index, record in enumerate(records):
|
||||
scenarios[str(record.get("scenario"))] += 1
|
||||
wire = number(record.get("cache_wire_bytes"), f"cache_cost {index} cache_wire_bytes")
|
||||
save_body = number(record.get("save_body_bytes_per_sample"), f"cache_cost {index} save_body_bytes_per_sample")
|
||||
require(wire > 0, f"cache_cost {index} has zero wire bytes")
|
||||
max_wire = max(max_wire, wire)
|
||||
max_save_amp = max(max_save_amp, save_body / wire)
|
||||
for field, target in (("clone", "max_clone_ns"), ("encode", "max_encode_ns"), ("save_inclusive", "max_save_ns")):
|
||||
quantiles = record.get(field)
|
||||
require(isinstance(quantiles, dict), f"cache_cost {index} missing {field} quantiles")
|
||||
value = number(quantiles.get("max_ns"), f"cache_cost {index} {field}.max_ns")
|
||||
if target == "max_clone_ns":
|
||||
max_clone_ns = max(max_clone_ns, value)
|
||||
elif target == "max_encode_ns":
|
||||
max_encode_ns = max(max_encode_ns, value)
|
||||
else:
|
||||
max_save_ns = max(max_save_ns, value)
|
||||
build = record.get("build", {})
|
||||
require(isinstance(build, dict), f"cache_cost {index} build must be an object")
|
||||
build_sources.add((build.get("source_revision"), build.get("source_tree"), build.get("test_opt_level_override")))
|
||||
return {
|
||||
"records": len(records),
|
||||
"scenario_counts": dict(sorted(scenarios.items())),
|
||||
"max_cache_wire_bytes": int(max_wire),
|
||||
"max_save_body_amplification": float(max_save_amp),
|
||||
"max_clone_ns": int(max_clone_ns),
|
||||
"max_encode_ns": int(max_encode_ns),
|
||||
"max_save_inclusive_ns": int(max_save_ns),
|
||||
"build_sources": [
|
||||
{"source_revision": revision, "source_tree": tree, "test_opt_level_override": opt}
|
||||
for revision, tree, opt in sorted(build_sources, key=lambda item: tuple("" if part is None else str(part) for part in item))
|
||||
],
|
||||
"provenance": {
|
||||
"cache_cost_log": str(path.resolve()),
|
||||
"cache_cost_log_sha256": digest(path),
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
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"]))
|
||||
p99 = None if abba["worst_p99_regression"] is None else Decimal(str(abba["worst_p99_regression"]))
|
||||
throughput = None if abba["worst_throughput_loss"] is None else Decimal(str(abba["worst_throughput_loss"]))
|
||||
lines = [
|
||||
f"# Scanner/Heal Performance Summary",
|
||||
"",
|
||||
f"- verdict: {summary['verdict']}",
|
||||
f"- reason: {summary['reason']}",
|
||||
f"- abba: status={abba['status']} performance={abba['performance']} evidence={abba['evidence']} cells={abba['cells']} comparisons={abba['comparisons_total']}",
|
||||
f"- worst_p99_regression: {pct(p99)}",
|
||||
f"- worst_throughput_loss: {pct(throughput)}",
|
||||
f"- p2_worst_post_stop_work_multiple: {ratio(p2)}",
|
||||
]
|
||||
if summary.get("cache_cost") is not None:
|
||||
cache = summary["cache_cost"]
|
||||
lines.extend([
|
||||
f"- cache_cost_records: {cache['records']}",
|
||||
f"- max_cache_wire_bytes: {cache['max_cache_wire_bytes']}",
|
||||
f"- max_save_body_amplification: {cache['max_save_body_amplification']:.3f}x",
|
||||
f"- max_clone_ns: {cache['max_clone_ns']}",
|
||||
f"- max_encode_ns: {cache['max_encode_ns']}",
|
||||
f"- max_save_inclusive_ns: {cache['max_save_inclusive_ns']}",
|
||||
])
|
||||
lines.extend([
|
||||
"",
|
||||
"## Provenance",
|
||||
"",
|
||||
])
|
||||
for key, value in abba["provenance"].items():
|
||||
lines.append(f"- {key}: {value}")
|
||||
if summary.get("cache_cost") is not None:
|
||||
for key, value in summary["cache_cost"]["provenance"].items():
|
||||
lines.append(f"- {key}: {value}")
|
||||
return "\n".join(lines) + "\n"
|
||||
|
||||
|
||||
def build_summary(args: argparse.Namespace) -> dict[str, Any]:
|
||||
abba = summarize_abba(args.abba_dir)
|
||||
cache = summarize_cache_cost(args.cache_cost_log) if args.cache_cost_log else None
|
||||
if args.require_cache_cost and cache is None:
|
||||
raise ValueError("cache-cost profile log is required")
|
||||
verdict = "PASS" if abba["gate_state"] == "pass" else "FAIL"
|
||||
reason = abba["reason"]
|
||||
return {"schema": 1, "verdict": verdict, "reason": reason, "abba": abba, "cache_cost": cache}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--abba-dir", type=Path, required=True, help="Directory containing manifest.json and report.json")
|
||||
parser.add_argument("--cache-cost-log", type=Path, help="Rust test output containing CACHE_COST JSON lines")
|
||||
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")
|
||||
args = parser.parse_args()
|
||||
try:
|
||||
summary = build_summary(args)
|
||||
if args.json_out:
|
||||
write_json(args.json_out, summary)
|
||||
if args.markdown_out:
|
||||
args.markdown_out.parent.mkdir(parents=True, exist_ok=True)
|
||||
args.markdown_out.write_text(markdown(summary), encoding="utf-8")
|
||||
abba = summary["abba"]
|
||||
print(
|
||||
f"{summary['verdict']} scanner_heal_perf "
|
||||
f"status={abba['status']} performance={abba['performance']} evidence={abba['evidence']} "
|
||||
f"comparisons={abba['comparisons_total']} reason={summary['reason']}"
|
||||
)
|
||||
return 0 if summary["verdict"] == "PASS" else 1
|
||||
except (ValueError, OSError, json.JSONDecodeError) as error:
|
||||
print(f"FAIL scanner_heal_perf error={error}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Executable
+171
@@ -0,0 +1,171 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import hashlib
|
||||
import io
|
||||
import json
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||
import summarize_scanner_heal_perf as summary
|
||||
|
||||
|
||||
def sha(path: Path) -> str:
|
||||
with path.open("rb") as stream:
|
||||
hasher = hashlib.sha256()
|
||||
while chunk := stream.read(1024 * 1024):
|
||||
hasher.update(chunk)
|
||||
return hasher.hexdigest()
|
||||
|
||||
|
||||
class ScannerHealPerfSummaryTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.temp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.temp.cleanup)
|
||||
self.root = Path(self.temp.name)
|
||||
self.abba = self.root / "abba"
|
||||
self.abba.mkdir()
|
||||
self.manifest = {
|
||||
"fixed": {
|
||||
"config_sha256": "1" * 64,
|
||||
"dataset_sha256": "2" * 64,
|
||||
"release_flags": "--profile production",
|
||||
"durability": "drive-sync=on",
|
||||
"topology": "EC8+4",
|
||||
"offered_load_ops": 100,
|
||||
},
|
||||
"baseline": {"revision": "a" * 40, "sha256": "3" * 64},
|
||||
"candidate": {"revision": "b" * 40, "sha256": "4" * 64},
|
||||
"adapter_sha256": "5" * 64,
|
||||
"collector_sha256": "6" * 64,
|
||||
}
|
||||
self.comparison = {
|
||||
"scenario": "cold-hot",
|
||||
"comparison": "build",
|
||||
"round": 1,
|
||||
"status": "pass",
|
||||
"p99_regression": 0.02,
|
||||
"throughput_change": -0.01,
|
||||
"p1": {"required_reduction": 0.8, "observed_reduction": 0.82, "repeatability_drift": 0.01},
|
||||
"p2_post_stop_work_multiples": [None, 1.1, 1.0, None],
|
||||
}
|
||||
self.report = {
|
||||
"status": "pass",
|
||||
"performance": "pass",
|
||||
"evidence": "measured",
|
||||
"cells": 120,
|
||||
"comparisons": [self.comparison],
|
||||
}
|
||||
self.write_inputs()
|
||||
|
||||
def write_inputs(self):
|
||||
(self.abba / "manifest.json").write_text(json.dumps(self.manifest), encoding="utf-8")
|
||||
(self.abba / "report.json").write_text(json.dumps(self.report), encoding="utf-8")
|
||||
|
||||
def test_measured_pass_writes_quiet_artifacts(self):
|
||||
cache_log = self.root / "cache.log"
|
||||
cache_log.write_text(
|
||||
"compiler noise\nCACHE_COST "
|
||||
+ json.dumps({
|
||||
"schema": 1,
|
||||
"scenario": "small_dirty",
|
||||
"cache_wire_bytes": 100,
|
||||
"save_body_bytes_per_sample": 200,
|
||||
"clone": {"max_ns": 10},
|
||||
"encode": {"max_ns": 20},
|
||||
"save_inclusive": {"max_ns": 30},
|
||||
"build": {"source_revision": "abc", "source_tree": "clean", "test_opt_level_override": "0"},
|
||||
})
|
||||
+ "\nmore noise\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
args = type("Args", (), {
|
||||
"abba_dir": self.abba,
|
||||
"cache_cost_log": cache_log,
|
||||
"require_cache_cost": False,
|
||||
"json_out": None,
|
||||
"markdown_out": None,
|
||||
})
|
||||
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["cache_cost"]["max_save_body_amplification"], 2.0)
|
||||
|
||||
def test_synthetic_report_fails_as_performance_conclusion(self):
|
||||
self.report.update(status="synthetic_validated", performance="pending", evidence="synthetic")
|
||||
self.write_inputs()
|
||||
args = type("Args", (), {
|
||||
"abba_dir": self.abba,
|
||||
"cache_cost_log": None,
|
||||
"require_cache_cost": False,
|
||||
"json_out": None,
|
||||
"markdown_out": None,
|
||||
})
|
||||
result = summary.build_summary(args)
|
||||
self.assertEqual(result["verdict"], "FAIL")
|
||||
self.assertIn("synthetic evidence", result["reason"])
|
||||
|
||||
def test_requires_cache_profile_when_requested(self):
|
||||
args = type("Args", (), {
|
||||
"abba_dir": self.abba,
|
||||
"cache_cost_log": None,
|
||||
"require_cache_cost": True,
|
||||
"json_out": None,
|
||||
"markdown_out": None,
|
||||
})
|
||||
with self.assertRaisesRegex(ValueError, "cache-cost profile log is required"):
|
||||
summary.build_summary(args)
|
||||
|
||||
def test_cli_prints_one_line_and_exits_nonzero_for_pending_performance(self):
|
||||
self.report.update(status="inconclusive", performance="inconclusive")
|
||||
self.write_inputs()
|
||||
stdout = io.StringIO()
|
||||
stderr = io.StringIO()
|
||||
argv = [
|
||||
"summarize_scanner_heal_perf.py",
|
||||
"--abba-dir",
|
||||
str(self.abba),
|
||||
"--json-out",
|
||||
str(self.root / "summary.json"),
|
||||
"--markdown-out",
|
||||
str(self.root / "summary.md"),
|
||||
]
|
||||
with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
|
||||
with mock.patch.object(sys, "argv", argv):
|
||||
code = summary.main()
|
||||
self.assertEqual(code, 1)
|
||||
self.assertEqual(stdout.getvalue().count("\n"), 1)
|
||||
self.assertTrue(stdout.getvalue().startswith("FAIL scanner_heal_perf "))
|
||||
self.assertEqual(stderr.getvalue(), "")
|
||||
self.assertEqual(summary.read_json(self.root / "summary.json")["verdict"], "FAIL")
|
||||
self.assertIn("worst_p99_regression", (self.root / "summary.md").read_text(encoding="utf-8"))
|
||||
|
||||
def test_invalid_cache_cost_lines_fail_closed(self):
|
||||
cache_log = self.root / "cache.log"
|
||||
cache_log.write_text("CACHE_COST {\"schema\": 2}\n", encoding="utf-8")
|
||||
with self.assertRaisesRegex(ValueError, "unsupported schema"):
|
||||
summary.summarize_cache_cost(cache_log)
|
||||
|
||||
def test_script_entrypoint_is_quiet(self):
|
||||
script = Path(__file__).with_name("summarize_scanner_heal_perf.py")
|
||||
process = subprocess.run(
|
||||
[sys.executable, str(script), "--abba-dir", str(self.abba)],
|
||||
check=False,
|
||||
text=True,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
)
|
||||
self.assertEqual(process.returncode, 0, process.stderr)
|
||||
self.assertEqual(process.stdout.count("\n"), 1)
|
||||
self.assertEqual(process.stderr, "")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user