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

test(scanner): bind raw profile artifacts (#7598)

Browse Source 
Require Scanner/Heal release profile wrappers to carry bundled raw profile artifacts plus the matching profile cost metrics.

Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
houseme
2026-09-10 00:49:01 +08:00
committed by GitHub
parent 42f00d8532
commit 45806bf295
5 changed files with 190 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/check_test_wiring.py
+108 -7
View File
@@ -1985,7 +1985,35 @@ def release_bundle_json_artifact_mirrored_fields(gate: str, field: str) -> tuple
return tuple(dict.fromkeys(fields))
def validate_release_bundle_json_artifact_payload(path: Path, source_revision: str, gate: str, field: str,
PROFILE_ARTIFACT_REQUIRED_METRICS = {
"allocation-profile": ("resolved_samples", "allocation_bytes"),
"flamegraph": ("resolved_samples",),
"rss-samples": ("resolved_samples", "rss_peak_bytes"),
"save-frequency": ("resolved_samples", "save_operations", "saved_bytes"),
}
def validate_profile_json_artifact_payload(bundle_path: Path, path: Path, payload: dict[str, object],
gate: str, field: str, artifact_kind: str) -> None:
prefix = f"{gate}.{field}.{artifact_kind}"
raw_profile = release_bundle_artifact_path(
bundle_path,
payload.get("raw_profile_artifact"),
gate,
f"{field}.{artifact_kind}.raw_profile",
)
require(sha(payload.get("raw_profile_sha256")) and digest(raw_profile) == payload["raw_profile_sha256"],
f"{prefix} raw profile hash mismatch")
raw_bytes = evidence_integer(payload.get("raw_profile_bytes"), f"{prefix}.raw_profile_bytes", 1, 2**63 - 1)
require(raw_bytes == raw_profile.stat().st_size, f"{prefix} raw profile bytes mismatch")
evidence_string(payload.get("raw_profile_format"), f"{prefix}.raw_profile_format",
r"[A-Za-z0-9][A-Za-z0-9._+:-]{1,63}")
require(raw_profile.resolve() != path.resolve(), f"{prefix} raw profile must be distinct from wrapper")
for metric in PROFILE_ARTIFACT_REQUIRED_METRICS[artifact_kind]:
evidence_integer(payload.get(metric), f"{prefix}.{metric}", 1, 2**63 - 1)
def validate_release_bundle_json_artifact_payload(bundle_path: Path, path: Path, source_revision: str, gate: str, field: str,
run_id: str, window_id: str,
artifact_kind: str | None = None) -> None:
payload = read_json(path)
@@ -2000,6 +2028,8 @@ def validate_release_bundle_json_artifact_payload(path: Path, source_revision: s
require(payload.get("field") == field, f"{prefix} JSON artifact field mismatch")
if artifact_kind is not None:
require(payload.get("artifact_kind") == artifact_kind, f"{prefix} JSON artifact kind mismatch")
if field == "profile_evidence":
validate_profile_json_artifact_payload(bundle_path, path, payload, gate, field, artifact_kind)
mrf_artifact_kind = SCANNER_HEAL_RELEASE_MRF_ARTIFACT_KINDS.get((gate, field))
if mrf_artifact_kind is not None:
require(payload.get("artifact_kind") == mrf_artifact_kind,
@@ -2536,7 +2566,7 @@ def validate_release_bundle_artifact(bundle_path: Path, source_revision: str, ga
artifact = release_bundle_artifact_path(bundle_path, evidence.get("artifact"), gate, field)
require(sha(evidence.get("sha256")) and digest(artifact) == evidence["sha256"], f"{gate}.{field} artifact hash mismatch")
if is_json_artifact_format(artifact_format):
validate_release_bundle_json_artifact_payload(artifact, source_revision, gate, field, run_id, window_id)
validate_release_bundle_json_artifact_payload(bundle_path, artifact, source_revision, gate, field, run_id, window_id)
summary = evidence.get("summary")
require(isinstance(summary, str) and summary.strip(), f"{gate}.{field} missing human summary")
if gate.startswith("P"):
@@ -2733,6 +2763,7 @@ def validate_release_bundle_artifact(bundle_path: Path, source_revision: str, ga
f"{gate}.{artifact_field} measurement window mismatch")
if is_json_artifact_format(artifact_format):
validate_release_bundle_json_artifact_payload(
bundle_path,
artifact_path,
source_revision,
gate,
@@ -2920,6 +2951,34 @@ def copy_release_bundle_artifact(descriptor_path: Path, bundle_dir: Path, eviden
evidence["sha256"] = source_sha
def copy_release_bundle_profile_raw_artifact(descriptor_path: Path, bundle_dir: Path, item: dict[str, object],
gate: str, field: str, artifact_kind: str) -> None:
wrapper = release_bundle_artifact_path(bundle_dir / "release-evidence.json", item.get("artifact"), gate, field)
payload = read_json(wrapper)
raw_source = release_bundle_descriptor_path(
descriptor_path,
payload.get("raw_profile_artifact"),
gate,
f"{field}.raw_profile",
)
raw_sha = digest(raw_source)
require(sha(payload.get("raw_profile_sha256")) and payload["raw_profile_sha256"] == raw_sha,
f"{gate}.{field} descriptor raw profile hash mismatch")
raw_bytes = evidence_integer(payload.get("raw_profile_bytes"), f"{gate}.{field}.raw_profile_bytes", 1, 2**63 - 1)
require(raw_bytes == raw_source.stat().st_size, f"{gate}.{field} descriptor raw profile bytes mismatch")
artifact_dir = bundle_dir / "artifacts"
safe_label = re.sub(r"[^A-Za-z0-9._-]", "-", f"{gate}-{field}-{artifact_kind}-raw")
raw_target = artifact_dir / f"{safe_label}{release_bundle_artifact_suffix(raw_source)}"
require(not raw_target.exists(), f"{gate}.{field} duplicate assembled raw profile path")
shutil.copyfile(raw_source, raw_target)
require(digest(raw_target) == raw_sha, f"{gate}.{field} assembled raw profile hash mismatch")
payload["raw_profile_artifact"] = raw_target.relative_to(bundle_dir).as_posix()
payload["raw_profile_sha256"] = raw_sha
payload["raw_profile_bytes"] = raw_target.stat().st_size
write_json(wrapper, payload)
item["sha256"] = digest(wrapper)
def assemble_scanner_heal_release_bundle_descriptors(
root: Path,
descriptor_paths: list[Path],
@@ -2987,6 +3046,15 @@ def assemble_scanner_heal_release_bundle_descriptors(
f"{field}.{artifact_kind}",
f"{gate}-{field}-{artifact_kind}",
)
if field == "profile_evidence":
copy_release_bundle_profile_raw_artifact(
descriptor_path,
bundle_dir,
item,
gate,
f"{field}.{artifact_kind}",
artifact_kind,
)
fields[field] = evidence
assembled_gates[gate] = {
"status": "pass",
@@ -3742,13 +3810,18 @@ class SelfTests(unittest.TestCase):
evidence["old_source_retained_until_successor"] = True
evidence["recovered_pending_migration"] = True
if field == "profile_evidence":
evidence["resolved_samples"] = 1
evidence["allocation_bytes"] = 1024
evidence["rss_peak_bytes"] = 4096
evidence["save_operations"] = 2
evidence["saved_bytes"] = 2048
profile_metrics = {
"resolved_samples": 1,
"allocation_bytes": 1024,
"rss_peak_bytes": 4096,
"save_operations": 2,
"saved_bytes": 2048,
}
evidence.update(profile_metrics)
artifacts = {}
for artifact_kind in RELEASE_PROFILE_ARTIFACTS:
raw_artifact = artifact_dir / f"{gate}-{field}-{artifact_kind}.raw"
raw_artifact.write_text(f"{artifact_kind} measured profile sample\n", encoding="utf-8")
profile_artifact = artifact_dir / f"{gate}-{field}-{artifact_kind}.json"
write_json(profile_artifact, {
"schema": 1,
@@ -3759,6 +3832,14 @@ class SelfTests(unittest.TestCase):
"gate": gate,
"field": field,
"artifact_kind": artifact_kind,
"raw_profile_artifact": raw_artifact.relative_to(bundle_dir).as_posix(),
"raw_profile_sha256": digest(raw_artifact),
"raw_profile_bytes": raw_artifact.stat().st_size,
"raw_profile_format": "raw",
**{
metric: profile_metrics[metric]
for metric in PROFILE_ARTIFACT_REQUIRED_METRICS[artifact_kind]
},
})
artifacts[artifact_kind] = {
"artifact": profile_artifact.relative_to(bundle_dir).as_posix(),
@@ -3845,6 +3926,14 @@ class SelfTests(unittest.TestCase):
self.assertEqual(evidence["artifact"], "artifacts/G01-root_authority_evidence.json")
self.assertEqual(evidence["sha256"], digest(bundle.parent / evidence["artifact"]))
self.assertTrue((bundle.parent / evidence["artifact"]).is_file())
profile = assembled["gates"]["P1"]["evidence_fields"]["profile_evidence"]
profile_artifact = profile["profile_artifacts"]["save-frequency"]
profile_wrapper = bundle.parent / profile_artifact["artifact"]
profile_payload = read_json(profile_wrapper)
raw_profile = bundle.parent / profile_payload["raw_profile_artifact"]
self.assertTrue(raw_profile.is_file())
self.assertEqual(profile_payload["raw_profile_sha256"], digest(raw_profile))
self.assertEqual(profile_artifact["sha256"], digest(profile_wrapper))
def test_scanner_heal_release_bundle_assembler_merges_measured_descriptors(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
@@ -4420,6 +4509,18 @@ class SelfTests(unittest.TestCase):
("P1", "profile_evidence", "rss-samples"),
"JSON artifact run_id mismatch",
),
(
"profile-raw-hash",
lambda payload: payload.update({"raw_profile_sha256": "0" * 64}),
("P1", "profile_evidence", "flamegraph"),
"raw profile hash mismatch",
),
(
"profile-save-cost",
lambda payload: payload.pop("saved_bytes"),
("P1", "profile_evidence", "save-frequency"),
"save-frequency.saved_bytes",
),
(
"fixture-marker",
lambda payload: payload.update({"fixture": True}),
scripts/run_scanner_heal_scheduler_pressure_evidence.py
+42 -9
View File
@@ -153,19 +153,48 @@ def post_stop_convergence_multiples(report: dict[str, Any]) -> list[float]:
return values
PROFILE_ARTIFACT_REQUIRED_METRICS = {
"allocation-profile": ("resolved_samples", "allocation_bytes"),
"flamegraph": ("resolved_samples",),
"rss-samples": ("resolved_samples", "rss_peak_bytes"),
"save-frequency": ("resolved_samples", "save_operations", "saved_bytes"),
}
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]:
run_id: str, window_id: str, profile_costs: dict[str, int]) -> dict[str, Any]:
copied: dict[str, Any] = {}
profile_dir = out_dir / "artifacts" / "profiles"
raw_dir = profile_dir / "raw"
profile_dir.mkdir(parents=True, exist_ok=True)
raw_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)
raw_target = raw_dir / f"{kind}{source.suffix or '.artifact'}"
shutil.copyfile(source, raw_target)
target = profile_dir / f"P1-profile_evidence-{kind}.json"
payload = {
"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": source.name,
"raw_profile_artifact": raw_target.relative_to(out_dir).as_posix(),
"raw_profile_sha256": digest(raw_target),
"raw_profile_bytes": raw_target.stat().st_size,
"raw_profile_format": artifact_format,
}
for metric in PROFILE_ARTIFACT_REQUIRED_METRICS[kind]:
payload[metric] = profile_costs[metric]
write_json(target, payload)
copied[kind] = {
"artifact": target.relative_to(out_dir).as_posix(),
"sha256": digest(target),
"artifact_format": artifact_format,
"artifact_format": "json",
"source_revision": source_revision,
"run_id": run_id,
"measurement_window_id": window_id,
@@ -265,6 +294,13 @@ def build_descriptor(args: argparse.Namespace) -> Path:
p2_limit = float(P2_WORK_MULTIPLE_LIMIT)
p2_worst = max(p2_multiples)
require(p2_worst <= p2_limit, "P2 post-stop convergence exceeded work multiple limit")
profile_costs = {
"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"),
}
common = {
"evidence_type": "measured",
@@ -286,6 +322,7 @@ def build_descriptor(args: argparse.Namespace) -> Path:
source_revision,
run_id,
window_id,
profile_costs,
)
gates: dict[str, Any] = {
@@ -355,11 +392,7 @@ def build_descriptor(args: argparse.Namespace) -> Path:
**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_costs,
"profile_artifacts": profile_refs,
}),
},
scripts/summarize_scanner_heal_perf.py
+28 -10
View File
@@ -10,6 +10,7 @@ from decimal import Decimal
import hashlib
import json
from pathlib import Path
import shutil
import sys
from typing import Any
@@ -518,9 +519,18 @@ def profile_wrapper_artifact(
window_id: str,
kind: str,
path: Path,
profile_measurements: dict[str, Any],
) -> dict[str, Any]:
required_metrics = {
"allocation-profile": ("resolved_samples", "allocation_bytes"),
"flamegraph": ("resolved_samples",),
"rss-samples": ("resolved_samples", "rss_peak_bytes"),
"save-frequency": ("resolved_samples", "save_operations", "saved_bytes"),
}[kind]
raw_artifact = artifact_dir / f"P1-profile_evidence-{kind}.raw{path.suffix}"
shutil.copyfile(path, raw_artifact)
wrapper = artifact_dir / f"P1-profile_evidence-{kind}.json"
write_json(wrapper, {
payload = {
"schema": 1,
"evidence_type": "measured",
"source_revision": source_revision,
@@ -530,9 +540,14 @@ def profile_wrapper_artifact(
"field": "profile_evidence",
"artifact_kind": kind,
"raw_profile_name": path.name,
"raw_profile_sha256": digest(path),
"raw_profile_bytes": path.stat().st_size,
})
"raw_profile_artifact": raw_artifact.relative_to(artifact_dir.parent).as_posix(),
"raw_profile_sha256": digest(raw_artifact),
"raw_profile_bytes": raw_artifact.stat().st_size,
"raw_profile_format": path.suffix.lower().lstrip(".") or "binary",
}
for metric in required_metrics:
payload[metric] = require_integer(profile_measurements.get(metric), metric, 1)
write_json(wrapper, payload)
return {
"artifact": wrapper.relative_to(artifact_dir.parent).as_posix(),
"sha256": digest(wrapper),
@@ -606,6 +621,13 @@ def write_release_bundle_descriptor(args: argparse.Namespace, summary: dict[str,
require(isinstance(profile, dict), "missing release_evidence.profile")
profile_measurements = profile.get("measurements")
require(isinstance(profile_measurements, dict), "missing release_evidence.profile.measurements")
profile_costs = {
"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 = profile_artifact_map(args.release_profile_artifact)
descriptor = args.release_bundle_descriptor_out
@@ -735,14 +757,10 @@ def write_release_bundle_descriptor(args: argparse.Namespace, summary: dict[str,
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_costs,
"profile_artifacts": {
kind: profile_wrapper_artifact(
artifact_dir, source_revision, run_id, window_id, kind, path
artifact_dir, source_revision, run_id, window_id, kind, path, profile_costs
)
for kind, path in sorted(profile_artifacts.items())
},
scripts/test_summarize_scanner_heal_perf.py
+7
View File
@@ -272,6 +272,13 @@ class ScannerHealPerfSummaryTest(unittest.TestCase):
gate,
)
self.assertEqual(status["verified_gate"], gate)
save_frequency = profile["profile_artifacts"]["save-frequency"]
wrapper = args.release_bundle_descriptor_out.parent / save_frequency["artifact"]
payload = summary.read_json(wrapper)
raw_profile = args.release_bundle_descriptor_out.parent / payload["raw_profile_artifact"]
self.assertTrue(raw_profile.is_file())
self.assertEqual(payload["saved_bytes"], 8192)
self.assertEqual(payload["raw_profile_sha256"], sha(raw_profile))
def test_release_descriptor_requires_profile_artifacts(self):
args = type("Args", (), {