mirror of
https://github.com/rustfs/rustfs.git
synced 2026-10-02 05:14:35 +08:00
hector
1efa86b8a3
## Related Issues
Follow-up to #8229.
## Summary of Changes
Locate the private auto-testing checkout from the lane root or the workspace root so the nested security checkout can record functional-chain evidence.
## Verification
The exact PR head b66129ab9f passed one mechanical correctness and simplicity review, nine real-Git layout and provenance checks, and sixteen existing evidence/envelope tests. The baseline sibling layout failed with git exit 128; the corrected layout succeeded while revision mismatches and missing checkouts stayed rejected. Current PR checks are completed with successful or skipped conclusions, including the aggregate.
## Impact
Both lane and private-script revision checks remain intact. No time limits, assertions, production behavior, or evidence validation requirements change. The synthetic layout checks do not execute the actual scheduled security suite; integrated main CI and release acceptance remain separate gates.
## Additional Notes
Approved review 5374559393 is bound to the exact head above. Reverting the single-file change restores the previous checkout lookup.
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()
|