vhaudiquet
ed6d88008d
resolve_pkh preferred target/release/pkh however old it was: after a commit without a release rebuild, the harness measured the previous code — the 54-package run recorded before this change exercised the pre-nested-namespace builds and recorded no phase profiles, the profile feature itself being newer than that binary. A repo binary is now rebuilt when it is older than HEAD or than the newest source edit; an explicit --pkh still pins whatever binary it names.
970 lines
38 KiB
Python
Executable File
970 lines
38 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""pkh end-to-end package harness.
|
|
|
|
Runs real packages from catalog.yml through the full pkh workflow:
|
|
|
|
pkh pull (from the archive, by default)
|
|
pkh chlog --rebuild (no-change rebuild entry; EDITOR is stubbed out)
|
|
pkh build (.dsc + _source.changes next to the tree)
|
|
pkh deb (binary packages in an ephemeral chroot)
|
|
|
|
Every stage is logged per package, timed, and checked for its artifacts;
|
|
results land in <work>/runs/<id>/results.json and results.md. Pulls run in
|
|
parallel (network-bound), the build pipeline runs sequentially (one heavy
|
|
build at a time, and pkh's shared chroot/apt/ccache caches stay warm).
|
|
|
|
Usage:
|
|
run.py --list
|
|
run.py --dry-run
|
|
run.py # smoke + standard tiers
|
|
run.py --tier long # the multi-hour builds
|
|
run.py --only hello,dracut # exact names
|
|
run.py --resume runs/<id>/results.json # continue an interrupted run
|
|
|
|
See README.md in this directory for the full documentation.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import concurrent.futures
|
|
import datetime as dt
|
|
import json
|
|
import os
|
|
import platform
|
|
import shlex
|
|
import shutil
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
|
|
TIERS = ("smoke", "standard", "long")
|
|
DEFAULT_TIERS = ("smoke", "standard")
|
|
TIER_TIMEOUT_MIN = {"smoke": 30, "standard": 120, "long": 360}
|
|
STAGES = ("pull", "chlog", "build", "deb")
|
|
PIPELINE_STAGES = ("chlog", "build", "deb")
|
|
TERMINAL_OUTCOMES = ("pass", "xpass", "xfail")
|
|
REPO_ROOT = Path(__file__).resolve().parents[2]
|
|
SCRIPT_DIR = Path(__file__).resolve().parent
|
|
|
|
|
|
class CatalogError(Exception):
|
|
"""Invalid catalog content or unusable options."""
|
|
|
|
|
|
@dataclass
|
|
class Entry:
|
|
"""One catalog entry: a source package and how to build it."""
|
|
|
|
name: str
|
|
category: str = "misc"
|
|
tier: str = "standard"
|
|
dist: str = "debian"
|
|
series: str = ""
|
|
arch: str = ""
|
|
cross: bool = False
|
|
pocket: str = ""
|
|
ppa: list[str] = field(default_factory=list)
|
|
timeout: float = 0.0 # minutes; 0 = the tier default
|
|
xfail: str = ""
|
|
notes: str = ""
|
|
tags: list[str] = field(default_factory=list)
|
|
|
|
def timeout_min(self) -> float:
|
|
return self.timeout if self.timeout else TIER_TIMEOUT_MIN.get(self.tier, 120)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Catalog loading: PyYAML when available, a schema-specific fallback parser
|
|
# otherwise (the subset it accepts is documented in catalog.yml).
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _scalar(text: str):
|
|
text = text.strip()
|
|
if text.startswith("[") and text.endswith("]"):
|
|
inner = text[1:-1].strip()
|
|
return [v.strip().strip("'\"") for v in inner.split(",") if v.strip()]
|
|
if len(text) >= 2 and text[0] == text[-1] and text[0] in "'\"":
|
|
return text[1:-1]
|
|
if text in ("true", "false"):
|
|
return text == "true"
|
|
if text.isdigit():
|
|
return int(text)
|
|
return text
|
|
|
|
|
|
def _parse_mini_yaml(text: str) -> dict:
|
|
"""Parse the flat 'defaults' mapping + flat 'packages' list subset."""
|
|
doc: dict = {"defaults": {}, "packages": []}
|
|
mode = None # None | "defaults" | "packages"
|
|
item: dict | None = None
|
|
for lineno, raw in enumerate(text.splitlines(), 1):
|
|
if not raw.strip() or raw.lstrip().startswith("#"):
|
|
continue
|
|
indent = len(raw) - len(raw.lstrip(" "))
|
|
stripped = raw.strip()
|
|
if indent == 0:
|
|
if stripped in ("defaults:", "packages:"):
|
|
mode = stripped[:-1]
|
|
item = None
|
|
continue
|
|
raise CatalogError(f"catalog:{lineno}: unexpected top-level '{stripped}'")
|
|
if mode is None:
|
|
raise CatalogError(f"catalog:{lineno}: content outside a top-level key")
|
|
if mode == "defaults":
|
|
key, sep, value = stripped.partition(":")
|
|
if not sep:
|
|
raise CatalogError(f"catalog:{lineno}: expected 'key: value'")
|
|
doc["defaults"][key.strip()] = _scalar(value)
|
|
else:
|
|
if stripped.startswith("- "):
|
|
item = {}
|
|
doc["packages"].append(item)
|
|
stripped = stripped[2:].strip()
|
|
if not stripped:
|
|
continue
|
|
if item is None:
|
|
raise CatalogError(f"catalog:{lineno}: list items must start with '- '")
|
|
key, sep, value = stripped.partition(":")
|
|
if not sep:
|
|
raise CatalogError(f"catalog:{lineno}: expected 'key: value'")
|
|
item[key.strip()] = _scalar(value)
|
|
return doc
|
|
|
|
|
|
def load_catalog(path: Path) -> list[Entry]:
|
|
text = path.read_text(encoding="utf-8")
|
|
use_pyyaml = not os.environ.get("PKH_E2E_NO_PYYAML")
|
|
try:
|
|
if use_pyyaml:
|
|
import yaml # type: ignore import-not-found
|
|
|
|
doc = yaml.safe_load(text)
|
|
else:
|
|
doc = _parse_mini_yaml(text)
|
|
except CatalogError:
|
|
raise
|
|
except Exception as exc:
|
|
if use_pyyaml:
|
|
raise CatalogError(f"cannot parse {path}: {exc}") from exc
|
|
raise
|
|
|
|
defaults = doc.get("defaults") or {}
|
|
raw_entries = doc.get("packages") or []
|
|
if not raw_entries:
|
|
raise CatalogError(f"{path}: no packages listed")
|
|
|
|
entries: list[Entry] = []
|
|
seen: set[str] = set()
|
|
for i, raw in enumerate(raw_entries):
|
|
name = str(raw.get("name", "")).strip()
|
|
if not name:
|
|
raise CatalogError(f"{path}: package #{i + 1} has no name")
|
|
if name in seen:
|
|
raise CatalogError(f"{path}: duplicate package '{name}'")
|
|
seen.add(name)
|
|
|
|
entry = Entry(
|
|
name=name,
|
|
category=str(raw.get("category", "misc")),
|
|
tier=str(raw.get("tier", defaults.get("tier", "standard"))),
|
|
dist=str(raw.get("dist", defaults.get("dist", "debian"))),
|
|
series=str(raw.get("series", "")),
|
|
arch=str(raw.get("arch", "")),
|
|
cross=bool(raw.get("cross", False)),
|
|
pocket=str(raw.get("pocket", "")),
|
|
ppa=[str(p) for p in (raw.get("ppa") or [])],
|
|
timeout=float(raw.get("timeout", 0)),
|
|
xfail=str(raw.get("xfail", "")),
|
|
notes=str(raw.get("notes", "")),
|
|
tags=[str(t) for t in (raw.get("tags") or [])],
|
|
)
|
|
if entry.tier not in TIERS:
|
|
raise CatalogError(f"{path}: {name}: unknown tier '{entry.tier}'")
|
|
if entry.dist not in ("debian", "ubuntu"):
|
|
raise CatalogError(f"{path}: {name}: unknown dist '{entry.dist}'")
|
|
if entry.cross and not entry.arch:
|
|
raise CatalogError(f"{path}: {name}: cross: true needs an arch")
|
|
entries.append(entry)
|
|
return entries
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Selection and rendering
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def select_entries(
|
|
entries: list[Entry], args: argparse.Namespace
|
|
) -> tuple[list[Entry], list[str]]:
|
|
"""Apply the CLI filters; returns the selection and the filtered-out names."""
|
|
tiers = (
|
|
list(TIERS) if args.tier == "all" else [t.strip() for t in args.tier.split(",")]
|
|
)
|
|
for tier in tiers:
|
|
if tier not in TIERS:
|
|
raise CatalogError(f"unknown tier '{tier}' (expected: {', '.join(TIERS)})")
|
|
|
|
selected = [e for e in entries if e.tier in tiers]
|
|
dropped = [e.name for e in entries if e.tier not in tiers]
|
|
|
|
if args.category:
|
|
wanted = {c.strip() for c in args.category.split(",") if c.strip()}
|
|
dropped += [e.name for e in selected if e.category not in wanted]
|
|
selected = [e for e in selected if e.category in wanted]
|
|
if args.tag:
|
|
wanted = {t.strip() for t in args.tag.split(",") if t.strip()}
|
|
dropped += [e.name for e in selected if not wanted.intersection(e.tags)]
|
|
selected = [e for e in selected if wanted.intersection(e.tags)]
|
|
if args.only:
|
|
names = {n.strip() for n in args.only.split(",") if n.strip()}
|
|
unknown = sorted(names - {e.name for e in entries})
|
|
if unknown:
|
|
raise CatalogError(f"--only: not in the catalog: {', '.join(unknown)}")
|
|
dropped += [e.name for e in selected if e.name not in names]
|
|
selected = [e for e in selected if e.name in names]
|
|
if args.skip:
|
|
names = {n.strip() for n in args.skip.split(",") if n.strip()}
|
|
dropped += [e.name for e in selected if e.name in names]
|
|
selected = [e for e in selected if e.name not in names]
|
|
return selected, dropped
|
|
|
|
|
|
def fmt_duration(seconds: float | None) -> str:
|
|
if seconds is None:
|
|
return "-"
|
|
seconds = int(seconds)
|
|
if seconds < 60:
|
|
return f"{seconds}s"
|
|
if seconds < 3600:
|
|
return f"{seconds // 60}m{seconds % 60:02d}s"
|
|
return f"{seconds // 3600}h{(seconds % 3600) // 60:02d}m"
|
|
|
|
|
|
def render_list(entries: list[Entry]) -> str:
|
|
lines = [
|
|
f"{'PACKAGE':<20} {'TIER':<9} {'DIST':<7} {'CATEGORY':<14} "
|
|
f"{'TAGS':<30} NOTES"
|
|
]
|
|
for e in entries:
|
|
extra = []
|
|
if e.cross:
|
|
extra.append(f"cross {e.arch}")
|
|
if e.series:
|
|
extra.append(f"series {e.series}")
|
|
note = " ".join(extra + [e.xfail or e.notes])
|
|
lines.append(
|
|
f"{e.name:<20} {e.tier:<9} {e.dist:<7} {e.category:<14} "
|
|
f"{','.join(e.tags):<30} {note}"
|
|
)
|
|
return "\n".join(lines) + f"\n\n{len(entries)} package(s)\n"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Command construction and child-process plumbing
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def stage_command(args: argparse.Namespace, entry: Entry, stage: str, pkh: Path) -> list[str]:
|
|
cmd = [str(pkh), stage]
|
|
if stage == "pull":
|
|
cmd += [entry.name, "-d", args.dist or entry.dist]
|
|
if args.series or entry.series:
|
|
cmd += ["-s", args.series or entry.series]
|
|
if args.pocket or entry.pocket:
|
|
cmd += ["-p", args.pocket or entry.pocket]
|
|
if not args.git:
|
|
cmd += ["--archive"]
|
|
elif stage == "chlog":
|
|
cmd += ["--rebuild"]
|
|
if args.series or entry.series:
|
|
cmd += ["-s", args.series or entry.series]
|
|
elif stage == "build":
|
|
cmd += ["--verbose"]
|
|
elif stage == "deb":
|
|
cmd += ["--verbose"]
|
|
if args.series or entry.series:
|
|
cmd += ["-s", args.series or entry.series]
|
|
arch = args.arch or entry.arch
|
|
if arch:
|
|
cmd += ["-a", arch]
|
|
if entry.cross or args.cross:
|
|
cmd += ["--cross"]
|
|
if args.pocket or entry.pocket:
|
|
cmd += ["-p", args.pocket or entry.pocket]
|
|
for ppa in entry.ppa + args.ppa:
|
|
cmd += ["--ppa", ppa]
|
|
if args.jobs:
|
|
cmd += ["-j", str(args.jobs)]
|
|
if args.no_ccache:
|
|
cmd += ["--no-ccache"]
|
|
return cmd
|
|
|
|
|
|
def child_env() -> dict:
|
|
"""Environment for pkh children: no editor, a deterministic identity."""
|
|
env = dict(os.environ)
|
|
env["EDITOR"] = "true" # chlog launches it after generating; nothing to edit
|
|
env.setdefault("DEBEMAIL", "pkh-e2e@invalid")
|
|
env.setdefault("DEBFULLNAME", "pkh e2e")
|
|
return env
|
|
|
|
|
|
def run_stage(
|
|
cmd: list[str], cwd: Path, log_path: Path, timeout_min: float
|
|
) -> tuple[str, float]:
|
|
"""Run one stage, logging to log_path; returns (status, seconds).
|
|
|
|
Status is 'ok', 'fail' or 'timeout'. The whole process group is killed
|
|
on timeout: builds fan out into make/compiler children, and killing only
|
|
the direct child would leave them running. On Ctrl+C the group gets a
|
|
graceful SIGTERM first, mirroring pkh's own interrupt cleanup.
|
|
"""
|
|
with log_path.open("ab") as log:
|
|
log.write(f"\n$ {shlex.join(cmd)}\n(cwd: {cwd})\n".encode())
|
|
log.flush()
|
|
start = time.monotonic()
|
|
proc = subprocess.Popen(
|
|
cmd,
|
|
cwd=cwd,
|
|
env=child_env(),
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=log,
|
|
stderr=subprocess.STDOUT,
|
|
start_new_session=True,
|
|
)
|
|
try:
|
|
proc.wait(timeout=timeout_min * 60)
|
|
except subprocess.TimeoutExpired:
|
|
_kill_group(proc, signal.SIGKILL)
|
|
return "timeout", time.monotonic() - start
|
|
except KeyboardInterrupt:
|
|
# A graceful stop first: SIGTERM lets pkh run its own interrupt
|
|
# cleanup; a child ignoring it is killed after a grace period.
|
|
_kill_group(proc, signal.SIGTERM, wait=10)
|
|
_kill_group(proc, signal.SIGKILL)
|
|
raise
|
|
return ("ok" if proc.returncode == 0 else "fail"), time.monotonic() - start
|
|
|
|
|
|
def _kill_group(
|
|
proc: subprocess.Popen, sig: signal.Signals, wait: float | None = None
|
|
) -> None:
|
|
try:
|
|
os.killpg(os.getpgid(proc.pid), sig)
|
|
except ProcessLookupError:
|
|
pass
|
|
try:
|
|
proc.wait(timeout=wait)
|
|
except subprocess.TimeoutExpired:
|
|
pass
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# The harness proper
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class Harness:
|
|
def __init__(self, args: argparse.Namespace, entries: list[Entry], pkh: Path):
|
|
self.args = args
|
|
self.entries = entries
|
|
self.pkh = pkh
|
|
self.stages = [s.strip() for s in args.stages.split(",") if s.strip()]
|
|
self.records: dict[str, dict] = {}
|
|
self.pull_results: dict[str, tuple[str, float]] = {}
|
|
self.lock = threading.Lock()
|
|
self.run_dir = self._open_run_dir()
|
|
self.results_path = self.run_dir / "results.json"
|
|
(self.run_dir / "logs").mkdir(parents=True, exist_ok=True)
|
|
|
|
# -- layout -------------------------------------------------------------
|
|
|
|
def _open_run_dir(self) -> Path:
|
|
if self.args.resume:
|
|
try:
|
|
data = json.loads(Path(self.args.resume).read_text())
|
|
run_dir = Path(data["run_dir"])
|
|
except (OSError, KeyError, json.JSONDecodeError) as exc:
|
|
raise CatalogError(f"--resume: cannot read {self.args.resume}: {exc}")
|
|
if not run_dir.is_dir():
|
|
raise CatalogError(f"--resume: missing run directory {run_dir}")
|
|
for record in data.get("entries", []):
|
|
self.records[record["name"]] = record
|
|
return run_dir
|
|
stamp = dt.datetime.now().strftime("%Y%m%d-%H%M%S")
|
|
return SCRIPT_DIR / "runs" / f"{stamp}-{os.getpid()}"
|
|
|
|
def log_path(self, name: str) -> Path:
|
|
return self.run_dir / "logs" / f"{name}.log"
|
|
|
|
def workroot(self, entry: Entry) -> Path:
|
|
# pkh pull (run from the run root) puts the tree at
|
|
# ./<pkg>/<pkg>/: this directory is both the tree's parent and
|
|
# where pkh build / pkh deb drop their artifacts.
|
|
return self.run_dir / entry.name
|
|
|
|
def tree(self, entry: Entry) -> Path:
|
|
return self.workroot(entry) / entry.name
|
|
|
|
# -- stages ---------------------------------------------------------------
|
|
|
|
def do_pull(self, entry: Entry) -> tuple[str, float]:
|
|
# Pull from the run root: pkh pull (cwd-less) extracts to
|
|
# ./<pkg>/<pkg>/ under its cwd, so every package still gets its own
|
|
# subtree and parallel pulls never collide. A failed pull leaves a
|
|
# partial tree that pkh pull would refuse to overwrite and that a
|
|
# later --resume must not trust: wipe it before retrying and after
|
|
# the last attempt.
|
|
for attempt in range(self.args.pull_retries + 1):
|
|
if attempt:
|
|
shutil.rmtree(self.workroot(entry), ignore_errors=True)
|
|
status, seconds = run_stage(
|
|
stage_command(self.args, entry, "pull", self.pkh),
|
|
self.run_dir,
|
|
self.log_path(entry.name),
|
|
entry.timeout_min(),
|
|
)
|
|
if status == "ok":
|
|
return status, seconds
|
|
shutil.rmtree(self.workroot(entry), ignore_errors=True)
|
|
return status, seconds
|
|
|
|
def do_chlog(self, entry: Entry) -> tuple[str, float]:
|
|
return run_stage(
|
|
stage_command(self.args, entry, "chlog", self.pkh),
|
|
self.tree(entry),
|
|
self.log_path(entry.name),
|
|
entry.timeout_min(),
|
|
)
|
|
|
|
def do_build(self, entry: Entry) -> tuple[tuple[str, float], list[str]]:
|
|
record = run_stage(
|
|
stage_command(self.args, entry, "build", self.pkh),
|
|
self.tree(entry),
|
|
self.log_path(entry.name),
|
|
entry.timeout_min(),
|
|
)
|
|
artifacts = self.build_artifacts(entry)
|
|
return self._check_artifacts(entry, record, artifacts, "dsc/changes"), artifacts
|
|
|
|
def do_deb(self, entry: Entry) -> tuple[tuple[str, float], list[str]]:
|
|
record = run_stage(
|
|
stage_command(self.args, entry, "deb", self.pkh),
|
|
self.tree(entry),
|
|
self.log_path(entry.name),
|
|
entry.timeout_min(),
|
|
)
|
|
debs = self.deb_artifacts(entry)
|
|
return self._check_artifacts(entry, record, debs, ".deb"), debs
|
|
|
|
def build_artifacts(self, entry: Entry) -> list[str]:
|
|
return sorted(
|
|
p.name
|
|
for pattern in ("*.dsc", "*_source.changes")
|
|
for p in self.workroot(entry).glob(pattern)
|
|
)
|
|
|
|
def deb_artifacts(self, entry: Entry) -> list[str]:
|
|
return sorted(p.name for p in self.workroot(entry).glob("*.deb"))
|
|
|
|
def _check_artifacts(
|
|
self, entry: Entry, record: tuple[str, float], artifacts: list[str], what: str
|
|
) -> tuple[str, float]:
|
|
"""A green stage with no artifacts is a failure: log it and flip."""
|
|
if record[0] == "ok" and not artifacts:
|
|
with self.log_path(entry.name).open("ab") as log:
|
|
log.write(
|
|
f"\npkh-e2e: stage reported success but produced no {what}\n".encode()
|
|
)
|
|
return ("fail", record[1])
|
|
return record
|
|
|
|
def changelog_version(self, entry: Entry) -> str:
|
|
try:
|
|
out = subprocess.run(
|
|
["dpkg-parsechangelog", "-S", "Version"],
|
|
cwd=self.tree(entry),
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
return out.stdout.strip()
|
|
except Exception:
|
|
return ""
|
|
|
|
# -- per-entry pipeline ------------------------------------------------------
|
|
|
|
def run_entry(self, entry: Entry) -> dict:
|
|
"""Full pipeline for one entry; never raises."""
|
|
# A resumed run keeps what the previous record already achieved
|
|
previous = self.records.get(entry.name)
|
|
record = {
|
|
"name": entry.name,
|
|
"category": entry.category,
|
|
"tier": entry.tier,
|
|
"dist": self.args.dist or entry.dist,
|
|
"series": self.args.series or entry.series,
|
|
"arch": self.args.arch or entry.arch or platform.machine(),
|
|
"cross": entry.cross or self.args.cross,
|
|
"tags": entry.tags,
|
|
"xfail": entry.xfail,
|
|
"notes": entry.notes,
|
|
"outcome": "fail",
|
|
"version": "",
|
|
"stages": {},
|
|
"artifacts": [],
|
|
"total_seconds": 0.0,
|
|
"log": str(self.log_path(entry.name).relative_to(self.run_dir)),
|
|
}
|
|
started = time.monotonic()
|
|
try:
|
|
self._run_entry(record, entry, previous)
|
|
finally:
|
|
record["total_seconds"] = time.monotonic() - started
|
|
if entry.xfail:
|
|
record["outcome"] = "xfail" if record["outcome"] == "fail" else "xpass"
|
|
with self.lock:
|
|
self.records[entry.name] = record
|
|
self.flush()
|
|
color = {"pass": 32, "xfail": 33, "xpass": 33}.get(record["outcome"], 31)
|
|
print(
|
|
f"[{entry.name}] \033[{color}m{record['outcome']}\033[0m "
|
|
f"({fmt_duration(record['total_seconds'])})"
|
|
+ (f" — {record['error']}" if record.get("error") else "")
|
|
)
|
|
return record
|
|
|
|
def _run_entry(self, record: dict, entry: Entry, previous: dict | None) -> None:
|
|
# Stages the resumed record already completed are reused rather than
|
|
# re-run: a resumed chlog must not bump the version a second time.
|
|
# Reuse only while the build selection is unchanged.
|
|
reusable = previous is not None and all(
|
|
previous.get(key) == record.get(key)
|
|
for key in ("dist", "series", "arch", "cross")
|
|
)
|
|
prev_stages = previous.get("stages", {}) if previous else {}
|
|
|
|
if "pull" in self.stages:
|
|
if (self.tree(entry) / "debian").is_dir():
|
|
# Resumed run: the tree survived, keep it (pkh pull refuses
|
|
# existing directories anyway).
|
|
record["stages"]["pull"] = {"status": "ok", "seconds": 0.0,
|
|
"cached": True}
|
|
print(f"[{entry.name}] pull ok (tree already present)")
|
|
else:
|
|
status, seconds = self.pull_results.get(entry.name, ("fail", 0.0))
|
|
record["stages"]["pull"] = {"status": status, "seconds": seconds}
|
|
if status != "ok":
|
|
record["error"] = f"pull {status}"
|
|
return
|
|
# Already reported by the pull phase; no duplicate line.
|
|
|
|
for stage in PIPELINE_STAGES:
|
|
if stage not in self.stages:
|
|
continue
|
|
prior = prev_stages.get(stage) if reusable else None
|
|
if prior and prior.get("status") == "ok":
|
|
record["stages"][stage] = {**prior, "reused": True}
|
|
if stage == "chlog":
|
|
record["version"] = previous.get("version", "")
|
|
elif stage == "build":
|
|
record["artifacts"] += self.build_artifacts(entry)
|
|
elif stage == "deb":
|
|
record["artifacts"] += self.deb_artifacts(entry)
|
|
print(f"[{entry.name}] {stage} ok (reused)")
|
|
continue
|
|
if not (self.tree(entry) / "debian").is_dir():
|
|
record["error"] = (
|
|
f"no source tree for {stage}; include the pull stage or "
|
|
"resume the original run directory"
|
|
)
|
|
return
|
|
|
|
if stage == "chlog":
|
|
status, seconds = self.do_chlog(entry)
|
|
record["stages"]["chlog"] = {"status": status, "seconds": seconds}
|
|
if status != "ok":
|
|
record["error"] = f"chlog {status}"
|
|
return
|
|
record["version"] = self.changelog_version(entry)
|
|
print(
|
|
f"[{entry.name}] chlog ok ({fmt_duration(seconds)})"
|
|
+ (f" -> {record['version']}" if record["version"] else "")
|
|
)
|
|
elif stage == "build":
|
|
(status, seconds), artifacts = self.do_build(entry)
|
|
record["stages"]["build"] = {"status": status, "seconds": seconds}
|
|
record["artifacts"] += artifacts
|
|
if status != "ok":
|
|
record["error"] = f"build {status}"
|
|
return
|
|
print(f"[{entry.name}] build ok ({fmt_duration(seconds)})")
|
|
elif stage == "deb":
|
|
(status, seconds), debs = self.do_deb(entry)
|
|
record["stages"]["deb"] = {"status": status, "seconds": seconds}
|
|
record["artifacts"] += debs
|
|
if status != "ok":
|
|
record["error"] = f"deb {status}"
|
|
return
|
|
print(
|
|
f"[{entry.name}] deb ok ({fmt_duration(seconds)}), "
|
|
f"{len(debs)} binary package(s)"
|
|
)
|
|
|
|
record["outcome"] = "pass"
|
|
|
|
# -- run orchestration ---------------------------------------------------------
|
|
|
|
def preflight(self) -> None:
|
|
version = subprocess.run(
|
|
[str(self.pkh), "--version"], capture_output=True, text=True
|
|
)
|
|
print(f"[pkh-e2e] pkh: {self.pkh} ({version.stdout.strip() or '?'})")
|
|
if "deb" in self.stages:
|
|
missing = [
|
|
tool
|
|
for tool in ("mmdebstrap", "unshare", "sudo")
|
|
if shutil.which(tool) is None
|
|
]
|
|
if missing:
|
|
print(f"[pkh-e2e] WARNING: deb needs: {', '.join(missing)}")
|
|
if subprocess.run(["sudo", "-n", "true"], capture_output=True).returncode:
|
|
print(
|
|
"[pkh-e2e] WARNING: passwordless sudo unavailable; "
|
|
"ephemeral chroot builds will fail"
|
|
)
|
|
free_g = shutil.disk_usage(self.run_dir).free / 2**30
|
|
if free_g < 20:
|
|
print(f"[pkh-e2e] WARNING: only {free_g:.0f}G free on {self.run_dir}")
|
|
print(
|
|
f"[pkh-e2e] {len(self.entries)} package(s), stages: "
|
|
f"{'+'.join(self.stages)} -> {self.run_dir}"
|
|
)
|
|
|
|
def resume_skip(self, entry: Entry) -> bool:
|
|
previous = self.records.get(entry.name)
|
|
if not previous or previous.get("outcome") not in TERMINAL_OUTCOMES:
|
|
return False
|
|
return all(s in previous.get("stages", {}) for s in self.stages)
|
|
|
|
def pull_phase(self) -> None:
|
|
if "pull" not in self.stages:
|
|
return
|
|
todo = [
|
|
e
|
|
for e in self.entries
|
|
if not self.resume_skip(e) and not (self.tree(e) / "debian").is_dir()
|
|
]
|
|
if not todo:
|
|
return
|
|
print(
|
|
f"[pkh-e2e] pulling {len(todo)} package(s) "
|
|
f"({self.args.pull_jobs} at a time)"
|
|
)
|
|
with concurrent.futures.ThreadPoolExecutor(
|
|
max_workers=self.args.pull_jobs
|
|
) as pool:
|
|
futures = {pool.submit(self.do_pull, e): e for e in todo}
|
|
for future in concurrent.futures.as_completed(futures):
|
|
entry = futures[future]
|
|
try:
|
|
status, seconds = future.result()
|
|
except Exception as exc: # keep the run going on worker errors
|
|
status, seconds = "fail", 0.0
|
|
print(f"[{entry.name}] pull crashed: {exc}")
|
|
self.pull_results[entry.name] = (status, seconds)
|
|
print(
|
|
f"[{entry.name}] pull {'ok' if status == 'ok' else status} "
|
|
f"({fmt_duration(seconds)})"
|
|
)
|
|
|
|
def run(self) -> int:
|
|
self.preflight()
|
|
self.flush()
|
|
|
|
self.pull_phase()
|
|
|
|
failures = 0
|
|
for entry in self.entries:
|
|
if self.resume_skip(entry):
|
|
print(f"[{entry.name}] already {self.records[entry.name]['outcome']},"
|
|
" skipping")
|
|
continue
|
|
if self.args.fail_fast and failures:
|
|
print(f"[{entry.name}] skipped (fail-fast)")
|
|
continue
|
|
record = self.run_entry(entry)
|
|
if record["outcome"] in ("fail", "timeout"):
|
|
failures += 1
|
|
|
|
self.write_report()
|
|
self.print_summary()
|
|
return self.exit_code(failures)
|
|
|
|
def print_summary(self) -> None:
|
|
rows = sorted(self.records.values(), key=lambda r: r["name"])
|
|
print(f"\n=== summary ({len(rows)} recorded) ===")
|
|
print(
|
|
f"{'PACKAGE':<20} {'TIER':<9} {'RESULT':<8} "
|
|
f"{'PULL':>7} {'CHLOG':>7} {'BUILD':>8} {'DEB':>8} {'TOTAL':>8} VERSION"
|
|
)
|
|
for r in rows:
|
|
def sec(stage):
|
|
data = r["stages"].get(stage)
|
|
return fmt_duration(data["seconds"]) if data else "-"
|
|
|
|
print(
|
|
f"{r['name']:<20} {r['tier']:<9} {r['outcome']:<8} "
|
|
f"{sec('pull'):>7} {sec('chlog'):>7} {sec('build'):>8} "
|
|
f"{sec('deb'):>8} {fmt_duration(r['total_seconds']):>8} "
|
|
f"{r['version']}"
|
|
)
|
|
tally: dict[str, int] = {}
|
|
for r in rows:
|
|
tally[r["outcome"]] = tally.get(r["outcome"], 0) + 1
|
|
print("\n" + ", ".join(f"{v} {k}" for k, v in sorted(tally.items())))
|
|
|
|
def exit_code(self, failures: int) -> int:
|
|
if failures:
|
|
return 1
|
|
if self.args.strict and any(
|
|
r["outcome"] == "xpass" for r in self.records.values()
|
|
):
|
|
return 1
|
|
return 0
|
|
|
|
# -- persistence ---------------------------------------------------------------
|
|
|
|
def flush(self) -> None:
|
|
"""Persist results progressively: a killed run keeps its records."""
|
|
if not hasattr(self, "_started"):
|
|
self._started = dt.datetime.now().astimezone().isoformat()
|
|
meta = {
|
|
"run_dir": str(self.run_dir),
|
|
"pkh": str(self.pkh),
|
|
"filters": {
|
|
"tier": self.args.tier,
|
|
"category": self.args.category,
|
|
"tag": self.args.tag,
|
|
"only": self.args.only,
|
|
"skip": self.args.skip,
|
|
"stages": self.stages,
|
|
},
|
|
"host": {
|
|
"kernel": platform.release(),
|
|
"arch": platform.machine(),
|
|
"python": platform.python_version(),
|
|
},
|
|
"started": self._started,
|
|
"updated": dt.datetime.now().astimezone().isoformat(),
|
|
"entries": sorted(self.records.values(), key=lambda r: r["name"]),
|
|
}
|
|
tmp = self.results_path.with_suffix(".json.tmp")
|
|
tmp.write_text(json.dumps(meta, indent=2) + "\n")
|
|
tmp.replace(self.results_path)
|
|
|
|
def write_report(self) -> None:
|
|
rows = sorted(self.records.values(), key=lambda r: (r["outcome"], r["name"]))
|
|
lines = [
|
|
f"# pkh e2e run — {self.run_dir.name}",
|
|
"",
|
|
f"- pkh: `{self.pkh}`",
|
|
f"- started: {getattr(self, '_started', '?')}",
|
|
f"- stages: {'+'.join(self.stages)}",
|
|
"",
|
|
"| package | tier | result | pull | chlog | build | deb | total |"
|
|
" version |",
|
|
"|---|---|---|---|---|---|---|---|---|",
|
|
]
|
|
for r in rows:
|
|
def sec(stage):
|
|
data = r["stages"].get(stage)
|
|
return fmt_duration(data["seconds"]) if data else "-"
|
|
|
|
lines.append(
|
|
f"| {r['name']} | {r['tier']} | {r['outcome']} | {sec('pull')} | "
|
|
f"{sec('chlog')} | {sec('build')} | {sec('deb')} | "
|
|
f"{fmt_duration(r['total_seconds'])} | {r['version']} |"
|
|
)
|
|
(self.run_dir / "results.md").write_text("\n".join(lines) + "\n")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Entry point
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def parse_args(argv: list[str]) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
|
|
)
|
|
parser.add_argument("--catalog", type=Path, default=SCRIPT_DIR / "catalog.yml",
|
|
help="catalog file (default: catalog.yml)")
|
|
parser.add_argument("--tier", default=",".join(DEFAULT_TIERS),
|
|
help="tiers to run, comma-separated: smoke, standard, long, "
|
|
"or 'all' (default: smoke,standard)")
|
|
parser.add_argument("--category", default="",
|
|
help="comma-separated categories to keep")
|
|
parser.add_argument("--tag", default="",
|
|
help="comma-separated tags; keep packages carrying any")
|
|
parser.add_argument("--only", default="",
|
|
help="comma-separated exact package names to run")
|
|
parser.add_argument("--skip", default="",
|
|
help="comma-separated exact package names to skip")
|
|
parser.add_argument("--stages", default=",".join(STAGES),
|
|
help=f"stages to run (default: {','.join(STAGES)})")
|
|
parser.add_argument("--pkh", default="",
|
|
help="pkh binary to exercise (default: the repo build, "
|
|
"then PATH)")
|
|
parser.add_argument("--build-pkh", action="store_true",
|
|
help="run 'cargo build --release' before resolving pkh")
|
|
parser.add_argument("--git", action="store_true",
|
|
help="pull through the package VCS instead of the archive")
|
|
parser.add_argument("--dist", default="",
|
|
help="override the catalog default dist (per-entry dist wins)")
|
|
parser.add_argument("--series", default="",
|
|
help="override the series (per-entry series wins)")
|
|
parser.add_argument("--arch", default="",
|
|
help="force the 'pkh deb -a' architecture on every entry")
|
|
parser.add_argument("--cross", action="store_true",
|
|
help="pass --cross to every deb build")
|
|
parser.add_argument("--pocket", default="",
|
|
help="distribution pocket passed to pull and deb")
|
|
parser.add_argument("--ppa", action="append", default=[],
|
|
help="PPA passed to 'pkh deb --ppa' (repeatable)")
|
|
parser.add_argument("--jobs", type=int, default=0,
|
|
help="pass -j N to 'pkh deb' (default: pkh chooses)")
|
|
parser.add_argument("--no-ccache", action="store_true",
|
|
help="pass --no-ccache to 'pkh deb'")
|
|
parser.add_argument("--pull-jobs", type=int, default=4,
|
|
help="parallel pulls (default: 4)")
|
|
parser.add_argument("--pull-retries", type=int, default=1,
|
|
help="retries per failed pull (default: 1)")
|
|
parser.add_argument("--fail-fast", action="store_true",
|
|
help="stop the pipeline at the first failure")
|
|
parser.add_argument("--strict", action="store_true",
|
|
help="also fail the run on xpass")
|
|
parser.add_argument("--resume", default="",
|
|
help="results.json of a previous run: reuse its run "
|
|
"directory and skip finished packages")
|
|
parser.add_argument("--list", action="store_true",
|
|
help="print the selected packages and exit")
|
|
parser.add_argument("--dry-run", action="store_true",
|
|
help="print the plan and the exact commands, run nothing")
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
def main(argv: list[str]) -> int:
|
|
args = parse_args(argv)
|
|
try:
|
|
entries = load_catalog(args.catalog)
|
|
selected, dropped = select_entries(entries, args)
|
|
|
|
stages = [s.strip() for s in args.stages.split(",") if s.strip()]
|
|
for stage in stages:
|
|
if stage not in STAGES:
|
|
raise CatalogError(
|
|
f"unknown stage '{stage}' (expected: {', '.join(STAGES)})"
|
|
)
|
|
if args.pull_jobs < 1:
|
|
raise CatalogError("--pull-jobs must be at least 1")
|
|
|
|
if args.list:
|
|
print(render_list(selected))
|
|
if dropped:
|
|
print(f"(filtered out: {len(dropped)})")
|
|
return 0
|
|
if not selected:
|
|
print("nothing selected", file=sys.stderr)
|
|
return 2
|
|
|
|
pkh = resolve_pkh(args)
|
|
if args.dry_run:
|
|
print(f"pkh: {pkh}")
|
|
print(f"selected {len(selected)} package(s), stages: {args.stages}")
|
|
for entry in selected:
|
|
where = f" {entry.series}" if entry.series else ""
|
|
print(f"\n[{entry.name}] {entry.tier}/{entry.category} "
|
|
f"({entry.dist}{where})")
|
|
for stage in stages:
|
|
cmd = stage_command(args, entry, stage, pkh)
|
|
print(f" {shlex.join(cmd)}"
|
|
f" # timeout {entry.timeout_min():.0f}m")
|
|
return 0
|
|
|
|
harness = Harness(args, selected, pkh)
|
|
return harness.run()
|
|
except CatalogError as exc:
|
|
print(f"error: {exc}", file=sys.stderr)
|
|
return 2
|
|
except KeyboardInterrupt:
|
|
print("\ninterrupted", file=sys.stderr)
|
|
return 130
|
|
|
|
|
|
def repo_source_epoch() -> float | None:
|
|
"""The newest evidence of source changes: HEAD's commit time, or the
|
|
newest mtime under src/ for uncommitted edits. None outside a git
|
|
checkout with no src/ tree."""
|
|
stamps = []
|
|
try:
|
|
out = subprocess.run(
|
|
["git", "log", "-1", "--format=%ct"],
|
|
cwd=REPO_ROOT, capture_output=True, text=True, check=True,
|
|
)
|
|
stamps.append(int(out.stdout.strip()))
|
|
except Exception:
|
|
pass
|
|
src = REPO_ROOT / "src"
|
|
if src.is_dir():
|
|
stamps.extend(path.stat().st_mtime for path in src.rglob("*.rs"))
|
|
return max(stamps) if stamps else None
|
|
|
|
|
|
def resolve_pkh(args: argparse.Namespace) -> Path:
|
|
if args.pkh:
|
|
path = Path(args.pkh).resolve()
|
|
if not path.is_file():
|
|
raise CatalogError(f"--pkh: no such file: {path}")
|
|
return path
|
|
|
|
# A repo binary older than the sources measures whatever the last
|
|
# `cargo build --release` happened to include — yesterday's code.
|
|
# Rebuild unless the caller pinned a binary or asked for it explicitly.
|
|
if not args.build_pkh:
|
|
head = repo_source_epoch()
|
|
fresh = head is not None and any(
|
|
candidate.is_file() and candidate.stat().st_mtime >= head
|
|
for candidate in (REPO_ROOT / "target/release/pkh",
|
|
REPO_ROOT / "target/debug/pkh")
|
|
)
|
|
if fresh:
|
|
return next(candidate
|
|
for candidate in (REPO_ROOT / "target/release/pkh",
|
|
REPO_ROOT / "target/debug/pkh")
|
|
if candidate.is_file()
|
|
and candidate.stat().st_mtime >= head)
|
|
print("[pkh-e2e] the repo pkh binary is older than the sources; "
|
|
"running 'cargo build --release'")
|
|
|
|
subprocess.run(["cargo", "build", "--release"], cwd=REPO_ROOT, check=True)
|
|
for candidate in (REPO_ROOT / "target/release/pkh", REPO_ROOT / "target/debug/pkh"):
|
|
if candidate.is_file():
|
|
return candidate
|
|
found = shutil.which("pkh")
|
|
if not found:
|
|
raise CatalogError(
|
|
"no pkh binary: pass --pkh, use --build-pkh, or install pkh on PATH"
|
|
)
|
|
return Path(found)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main(sys.argv[1:]))
|