mirror of
https://github.com/rustfs/rustfs.git
synced 2026-10-04 04:21:35 +00:00
hector
b66129ab9f
#8229 moved the security lane's evidence steps onto its own rustfs checkout (rustfs-repo/), which changed what ROOT means for those runs: record() resolves 'auto-testing' relative to ROOT, so it looked for rustfs-repo/auto-testing while the lane checks the private scripts out at the workspace root. The git call exited 128 and the security lane went red with zero cases in the Sep 29 and Sep 30 overnight chains (36601455596, 36747961149) - the first failure mode after the checkout itself stopped being the problem. Resolve the checkout directory as the first of ROOT/auto-testing and GITHUB_WORKSPACE/auto-testing that exists; the testing_sha provenance check is unchanged.
268 lines
15 KiB
Python
268 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""Bind functional-suite evidence to one candidate and one chain attempt."""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
from datetime import datetime, timezone
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import subprocess
|
|
|
|
from resolve_functional_candidate import ROOT, positive, require, sha, validate_manifest
|
|
|
|
SUITES = ("upgrade", "s3", "kms", "tier", "storage", "heal", "pool", "security", "replication", "fault-tolerance", "table", "performance")
|
|
MAX_REPORT = 8 * 1024 * 1024
|
|
|
|
|
|
def current_chain():
|
|
chain = json.loads(os.environ["CHAIN_MANIFEST"])
|
|
require(isinstance(chain, dict) and set(chain) == {"schema", "run_id", "attempt", "workflow_sha", "testing_sha", "candidate"}, "invalid chain envelope")
|
|
require(type(chain["schema"]) is int and chain["schema"] == 1, "unsupported chain schema")
|
|
require(positive(chain["run_id"]) and positive(chain["attempt"]), "invalid chain run identity")
|
|
require(chain["run_id"] == int(os.environ["GITHUB_RUN_ID"]) and chain["attempt"] == int(os.environ["GITHUB_RUN_ATTEMPT"]), "chain belongs to another run attempt; rerun all jobs")
|
|
require(sha(chain["workflow_sha"]) and chain["workflow_sha"] == os.environ["GITHUB_SHA"], "chain workflow source mismatch")
|
|
head = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=ROOT, text=True).strip()
|
|
require(head == chain["workflow_sha"], "lane checkout differs from chain workflow source")
|
|
# testing_sha is either the committed pin or auto-testing main HEAD via
|
|
# resolve_functional_candidate.py's >24h staleness fallback, so pin
|
|
# equality is no longer an invariant (the 09-21 chain died on exactly
|
|
# that check once the fallback finally fired). Lanes check out exactly
|
|
# this sha, which is what the format check guards.
|
|
require(sha(chain["testing_sha"]), "private script revision is not a valid commit sha")
|
|
candidate = chain["candidate"]
|
|
require(isinstance(candidate, dict) and set(candidate) == {"manifest", "artifact_id", "artifact_digest", "workflow_sha", "workflow_ref", "build_started_at"}, "invalid candidate envelope")
|
|
manifest = candidate["manifest"]
|
|
require(positive(candidate["artifact_id"]) and isinstance(candidate["artifact_digest"], str) and bool(re.fullmatch(r"sha256:[0-9a-f]{64}", candidate["artifact_digest"])), "invalid candidate artifact identity")
|
|
require(sha(candidate["workflow_sha"]) and candidate["workflow_ref"] == "main", "candidate workflow source is invalid")
|
|
validate_manifest(manifest, {"id": manifest["build_run_id"], "run_attempt": manifest["build_run_attempt"], "head_sha": candidate["workflow_sha"]})
|
|
return chain
|
|
|
|
|
|
def consume(chain):
|
|
manifest = chain["candidate"]["manifest"]
|
|
with open(os.environ["GITHUB_ENV"], "a") as output:
|
|
# Every pinned installer already verifies these hashes before dpkg.
|
|
for key, value in (("RUSTFS_NIGHTLY_PACKAGE_URL", manifest["package_url"]),
|
|
("PACKAGE_SHA256", manifest["package_sha256"]), ("TO_SHA256", manifest["package_sha256"])):
|
|
output.write(key + "=" + value + "\n")
|
|
with open(os.environ["GITHUB_OUTPUT"], "a") as output:
|
|
output.write("testing_sha=" + chain["testing_sha"] + "\n")
|
|
|
|
|
|
def report_counts(text, performance=False):
|
|
counts = {"PASS": 0, "FAIL": 0, "SKIP": 0, "UNSUPPORTED": 0, "RUNNING": 0}
|
|
if performance:
|
|
rows = list(csv.DictReader(io.StringIO(text), delimiter="\t"))
|
|
seen = set()
|
|
for row in rows:
|
|
key = (row.get("method"), row.get("size"))
|
|
require(key[0] in ("get", "put", "mixed") and key[1] and key not in seen, "invalid or duplicate performance round")
|
|
seen.add(key)
|
|
fields = ("throughput", "obj_per_s", "req_avg", "req_p50")
|
|
if key[0] != "mixed":
|
|
fields += ("req_p90", "req_p99")
|
|
require(all(isinstance(row.get(field), str) and row[field].strip() for field in fields), "missing benchmark metrics")
|
|
counts["PASS"] = len(rows)
|
|
return counts
|
|
column = None
|
|
for line in text.splitlines():
|
|
if not line.startswith("|"):
|
|
column = None
|
|
continue
|
|
cells = [cell.strip().strip("*") for cell in line.strip().strip("|").split("|")]
|
|
for label in ("Status", "Result"):
|
|
if cells[0] in ("ID", "Case", "Topology", "Step") and label in cells:
|
|
column = cells.index(label)
|
|
break
|
|
else:
|
|
if column is not None:
|
|
require(len(cells) > column, "incomplete report row")
|
|
status = cells[column]
|
|
if re.fullmatch(r":?-+:?", status):
|
|
continue
|
|
require(status in counts, "unknown case result")
|
|
counts[status] += 1
|
|
return counts
|
|
|
|
|
|
def fault_tolerance_counts(text):
|
|
counts = {"PASS": 0, "FAIL": 0, "SKIP": 0, "UNSUPPORTED": 0, "RUNNING": 0}
|
|
statuses = {"pass": "PASS", "known-divergence": "UNSUPPORTED", "UNEXPECTED": "FAIL"}
|
|
cases = set()
|
|
summary = None
|
|
for line in text.splitlines():
|
|
if line.startswith("FT-CASE:"):
|
|
match = re.fullmatch(r"FT-CASE:\s+(\S+)\s+verdict=(\S+)\s+.*", line)
|
|
require(match is not None and summary is None, "invalid or late fault-tolerance case")
|
|
case, status = match.groups()
|
|
require(case not in cases and status in statuses, "duplicate or unknown fault-tolerance case result")
|
|
cases.add(case)
|
|
counts[statuses[status]] += 1
|
|
elif line.startswith("FT-SUMMARY:"):
|
|
match = re.fullmatch(r"FT-SUMMARY: unexpected=(\d+) known-divergence=(\d+) strict=([01])", line)
|
|
require(match is not None and summary is None, "invalid or duplicate fault-tolerance summary")
|
|
summary = tuple(map(int, match.groups()))
|
|
require(cases and summary is not None, "missing completed fault-tolerance evidence")
|
|
require(summary[:2] == (counts["FAIL"], counts["UNSUPPORTED"]), "fault-tolerance summary disagrees with cases")
|
|
require(not summary[2] or not counts["UNSUPPORTED"], "strict fault-tolerance run has known divergence")
|
|
return counts
|
|
|
|
|
|
def auto_testing_dir():
|
|
"""Locate the auto-testing checkout. Lanes that run this script from a
|
|
subdirectory checkout (security keeps its rustfs clone in rustfs-repo/)
|
|
still check auto-testing out at the workspace root, so ROOT alone is not
|
|
always the right base."""
|
|
candidates = [ROOT / "auto-testing"]
|
|
workspace = os.environ.get("GITHUB_WORKSPACE")
|
|
if workspace:
|
|
candidates.append(Path(workspace) / "auto-testing")
|
|
for candidate in candidates:
|
|
if (candidate / ".git").exists():
|
|
return candidate
|
|
return candidates[0]
|
|
|
|
|
|
def record(chain, suite, report, output):
|
|
require(suite in SUITES, "unknown suite")
|
|
result = {"schema": 1, "suite": suite, "chain": chain, "valid": False, "counts": {}, "report_sha256": None}
|
|
error = None
|
|
try:
|
|
private_head = subprocess.check_output(["git", "-C", str(auto_testing_dir()), "rev-parse", "HEAD"], cwd=ROOT, text=True).strip()
|
|
require(private_head == chain["testing_sha"], "suite used a different private script revision")
|
|
require(report.is_file() and 0 < report.stat().st_size <= MAX_REPORT, "missing, empty or oversized report")
|
|
data = report.read_bytes()
|
|
result["report_sha256"] = hashlib.sha256(data).hexdigest()
|
|
text = data.decode("utf-8")
|
|
result["counts"] = fault_tolerance_counts(text) if suite == "fault-tolerance" else report_counts(text, suite == "performance")
|
|
require(result["counts"]["PASS"] > 0 and not result["counts"]["FAIL"] and not result["counts"]["RUNNING"], "no passing executions or incomplete/failed cases")
|
|
require(all(os.environ[key] == "success" for key in ("CHAIN_JOB_STATUS", "CHAIN_TEST_OUTCOME", "CHAIN_REPORT_OUTCOME")), "suite, report or job did not succeed")
|
|
result["valid"] = True
|
|
except (OSError, ValueError, subprocess.SubprocessError) as exc:
|
|
error = exc
|
|
result["error"] = str(exc)
|
|
output.parent.mkdir(parents=True, exist_ok=False)
|
|
output.write_text(json.dumps(result, sort_keys=True) + "\n")
|
|
# A failed validation may still produce fresh diagnostics. A failed write
|
|
# or directory collision must never authorize uploading a leftover file.
|
|
with open(os.environ["GITHUB_OUTPUT"], "a") as step_output:
|
|
step_output.write("written=true\n")
|
|
if error:
|
|
raise error
|
|
|
|
|
|
def summarize(chain, directory, needs):
|
|
"""Retain failed/missing lanes without granting complete-success evidence."""
|
|
lanes = []
|
|
for suite in SUITES:
|
|
lane = {"suite": suite, "result": needs.get(suite, {}).get("result", "missing"),
|
|
"evidence": None, "error": None}
|
|
try:
|
|
record = json.loads((directory / (suite + ".json")).read_text())
|
|
require(chain is not None and record.get("chain") == chain and record.get("suite") == suite,
|
|
"suite evidence identity mismatch")
|
|
lane["evidence"] = record
|
|
except (OSError, ValueError, AttributeError) as error:
|
|
lane["error"] = str(error)
|
|
lanes.append(lane)
|
|
result = {"schema": 1, "chain": chain, "needs": needs, "lanes": lanes,
|
|
"complete": False, "error": None, "completed_at": datetime.now(timezone.utc).isoformat()}
|
|
try:
|
|
require(chain is not None, "candidate preparation did not complete")
|
|
aggregate(chain, directory, {suite: value for suite, value in needs.items() if suite != "prepare"})
|
|
result["complete"] = True
|
|
except (OSError, ValueError, KeyError, TypeError, AttributeError) as error:
|
|
result["error"] = str(error)
|
|
return result
|
|
|
|
|
|
def render_summary(result):
|
|
lines = ["# RustFS functional chain report", "",
|
|
"- All required suites passed with verified evidence: " + str(result["complete"]).lower(),
|
|
"- Preparation: " + result["needs"].get("prepare", {}).get("result", "missing")]
|
|
if result["chain"]:
|
|
chain = result["chain"]
|
|
manifest = chain["candidate"]["manifest"]
|
|
lines += [f"- Chain run / attempt: {chain['run_id']} / {chain['attempt']}",
|
|
f"- Build run / attempt: {manifest['build_run_id']} / {manifest['build_run_attempt']}",
|
|
f"- Source: {manifest.get('source_ref', 'main')} @ {manifest['source_sha']}",
|
|
f"- Package SHA256: {manifest['package_sha256']}", f"- Test scripts: {chain['testing_sha']}"]
|
|
lines += ["", "| Suite | Job result | Evidence | PASS | FAIL | SKIP | UNSUPPORTED | RUNNING |",
|
|
"| --- | --- | --- | --- | --- | --- | --- | --- |"]
|
|
for lane in result["lanes"]:
|
|
record = lane["evidence"] or {}
|
|
counts = record.get("counts", {})
|
|
state = "valid" if record.get("valid") is True else ("invalid" if record else "missing")
|
|
values = [lane["suite"], lane["result"], state] + [str(counts.get(key, "—")) for key in
|
|
("PASS", "FAIL", "SKIP", "UNSUPPORTED", "RUNNING")]
|
|
lines.append("| " + " | ".join(values) + " |")
|
|
lines += ["", "Missing, skipped, cancelled or invalid evidence is NOT a passing test or proof of a fix.",
|
|
"See the suite artifacts for case results and diagnostics; this report does not replace the complete-success gate."]
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|
|
def aggregate(chain, directory, needs, allow_skipped=()):
|
|
allowed = {name for name in allow_skipped if name}
|
|
require(allowed <= set(SUITES), "aggregate allow-list names an unknown lane")
|
|
require(set(needs) == set(SUITES), "aggregate is missing a required lane")
|
|
skipped = {name for name, value in needs.items() if value.get("result") == "skipped"}
|
|
require(skipped <= allowed, "a lane was skipped without preflight permission: " + ", ".join(sorted(skipped - allowed)))
|
|
require(all(value.get("result") == "success" for name, value in needs.items() if name not in skipped), "a required suite did not succeed")
|
|
expected = [suite for suite in SUITES if suite not in skipped]
|
|
require({path.name for path in directory.iterdir()} == {suite + ".json" for suite in expected}, "missing or unexpected suite evidence")
|
|
records = [json.loads((directory / (suite + ".json")).read_text()) for suite in expected]
|
|
validate_records(chain, records, expected)
|
|
return {"schema": 1, "chain": chain, "suites": records, "complete": True,
|
|
"skipped_lanes": sorted(skipped), "completed_at": datetime.now(timezone.utc).isoformat()}
|
|
|
|
|
|
def validate_records(chain, records, expected_suites=SUITES):
|
|
require(isinstance(records, list) and len(records) == len(expected_suites), "missing suite evidence")
|
|
require([record.get("suite") for record in records] == list(expected_suites), "missing, duplicate or reordered suite evidence")
|
|
for suite, result in zip(SUITES, records):
|
|
require(type(result.get("schema")) is int and result["schema"] == 1 and result.get("suite") == suite and result.get("chain") == chain, "suite evidence identity mismatch")
|
|
require(result.get("valid") is True and sha(result.get("report_sha256"), 64), "suite evidence is invalid")
|
|
counts = result.get("counts", {})
|
|
require(set(counts) == {"PASS", "FAIL", "SKIP", "UNSUPPORTED", "RUNNING"}, "missing suite counts")
|
|
require(all(type(value) is int and value >= 0 for value in counts.values()) and counts["PASS"] > 0 and counts["FAIL"] == counts["RUNNING"] == 0, "suite has no complete passing evidence")
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("mode", choices=("consume", "record", "aggregate", "summarize"))
|
|
parser.add_argument("--suite", choices=SUITES)
|
|
parser.add_argument("--report", type=Path)
|
|
parser.add_argument("--output", type=Path)
|
|
parser.add_argument("--directory", type=Path)
|
|
parser.add_argument("--allow-skipped", default="",
|
|
help="comma-separated lanes the preflight deliberately skipped (e.g. performance)")
|
|
args = parser.parse_args()
|
|
if args.mode == "summarize":
|
|
chain = current_chain() if os.environ.get("CHAIN_MANIFEST") else None
|
|
result = summarize(chain, args.directory, json.loads(os.environ["CHAIN_NEEDS"]))
|
|
args.output.write_text(json.dumps(result, sort_keys=True) + "\n")
|
|
args.output.with_suffix(".md").write_text(render_summary(result))
|
|
if os.environ.get("GITHUB_STEP_SUMMARY"):
|
|
with open(os.environ["GITHUB_STEP_SUMMARY"], "a") as summary:
|
|
summary.write(render_summary(result))
|
|
return
|
|
chain = current_chain()
|
|
if args.mode == "consume":
|
|
consume(chain)
|
|
elif args.mode == "record":
|
|
record(chain, args.suite, args.report, args.output)
|
|
else:
|
|
needs = json.loads(os.environ["CHAIN_NEEDS"])
|
|
needs.pop("prepare", None)
|
|
result = aggregate(chain, args.directory, needs, args.allow_skipped.split(","))
|
|
args.output.write_text(json.dumps(result, sort_keys=True) + "\n")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|