Host-side slot execution — image assembly, build, run, teardown.
The shared Containerfile templates ship inside this package; a consuming
repo customises them declaratively (extra-packages becomes the
EXTRA_PACKAGES build arg) or, for genuinely repo-specific image
tweaks, drops a fragment next to its matrix.yml
(fragments/Containerfile.<slot>) that the engine appends verbatim.
Process output streams straight to the operator's terminal (podman build
and the test run are long and live); all narration around it belongs to
cli.
SlotResult(passed, observed='?')
dataclass
Outcome of one slot run: pass/fail plus the observed version.
passed
instance-attribute
observed = '?'
class-attribute
instance-attribute
render_containerfile(config, slot_name)
Shared template plus the repo's optional fragment, ready to build.
Source code in src/terok_util/matrix/runner.py
| def render_containerfile(config: MatrixConfig, slot_name: str) -> str:
"""Shared template plus the repo's optional fragment, ready to build."""
kind = SLOTS[slot_name].kind
flavor = "nix" if kind is SlotKind.NIX else config.flavor
rendered = _TEMPLATES.get_template(f"{flavor}/Containerfile.{slot_name}").render(
uv_tag=UV_IMAGE_TAG,
uv_python_dir=UV_MANAGED_PYTHON_DIR,
)
fragment = config.containers_dir / "fragments" / f"Containerfile.{slot_name}"
if fragment.is_file():
rendered += "\n" + fragment.read_text(encoding="utf-8")
return rendered
|
build_image(config, slot_name, results_dir, no_cache=False)
Assemble and build the slot image; the repo root is the build context.
Source code in src/terok_util/matrix/runner.py
| def build_image(
config: MatrixConfig, slot_name: str, results_dir: Path, no_cache: bool = False
) -> bool:
"""Assemble and build the slot image; the repo root is the build context."""
containerfile = results_dir / f"Containerfile.{slot_name}"
containerfile.write_text(render_containerfile(config, slot_name), encoding="utf-8")
argv = [
"podman",
"build",
# Base tags are moving targets (quay.io/podman/stable:latest,
# distro :latest bases) but a cached FROM never re-checks the
# registry -- the podman slot sat on 5.8.2 while quay shipped
# newer. --pull=newer is a digest check per build: pulls only
# when the registry actually has a fresher base.
"--pull=newer",
*(["--no-cache"] if no_cache else []),
"--build-arg",
f"IMAGE_PREFIX={config.image_prefix}",
"--build-arg",
f"EXTRA_PACKAGES={' '.join(config.slots[slot_name].extra_packages)}",
"-t",
f"{config.image_prefix}:{slot_name}",
"-f",
str(containerfile),
str(config.repo_root),
]
return subprocess.run(argv, check=False).returncode == 0 # nosec B603
|
run_slot(config, slot_name, results_dir, scope='all', line_prefix=None)
Run one slot's test container and collect its observed version.
With line_prefix, the container's combined output streams through
this process line by line, each line tagged with the prefix — live
and attributable when several slots run concurrently. Each tagged
line is emitted as a single write call so concurrent slots can
interleave only between lines, never inside one.
Source code in src/terok_util/matrix/runner.py
| def run_slot(
config: MatrixConfig,
slot_name: str,
results_dir: Path,
scope: str = "all",
line_prefix: str | None = None,
) -> SlotResult:
"""Run one slot's test container and collect its observed version.
With *line_prefix*, the container's combined output streams through
this process line by line, each line tagged with the prefix — live
and attributable when several slots run concurrently. Each tagged
line is emitted as a single ``write`` call so concurrent slots can
interleave only between lines, never inside one.
"""
_write_scripts(config, slot_name, results_dir, scope)
argv = _run_argv(config, slot_name, results_dir)
if line_prefix is None:
status = subprocess.run(argv, check=False).returncode # nosec B603
else:
with subprocess.Popen( # nosec B603
argv, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, errors="replace"
) as proc:
stdout = cast(IO[str], proc.stdout) # guaranteed non-None by stdout=PIPE
for line in stdout:
sys.stdout.write(f"{line_prefix}{line}")
sys.stdout.flush()
status = proc.wait()
return SlotResult(passed=status == 0, observed=_observed_version(slot_name, results_dir))
|
sweep_containers(config)
Remove leftover run containers owned by this harness.
podman run --rm cleans up after itself — but an interrupted run
leaves the container behind, and a leftover container pins its (by
then dangling) image generation, immune to any prune. The sweep is
what lets the prune that follows actually reclaim the bytes.
Returns:
| Type |
Description |
int
|
The number of containers removed.
|
Source code in src/terok_util/matrix/runner.py
| def sweep_containers(config: MatrixConfig) -> int:
"""Remove leftover run containers owned by this harness.
``podman run --rm`` cleans up after itself — but an interrupted run
leaves the container behind, and a leftover container pins its (by
then dangling) image generation, immune to any prune. The sweep is
what lets the prune that follows actually reclaim the bytes.
Returns:
The number of containers removed.
"""
listing = subprocess.run( # nosec B603 B607 - fixed argv, podman from PATH by design
["podman", "ps", "-aq", "--filter", f"label={OWNERSHIP_LABEL}={config.image_prefix}"],
check=False,
capture_output=True,
text=True,
)
container_ids = listing.stdout.split()
if not container_ids:
return 0
subprocess.run( # nosec B603 B607 - fixed argv, podman from PATH by design
["podman", "rm", "-f", "-t", _REMOVAL_GRACE_SECONDS, *container_ids],
check=False,
capture_output=True,
text=True,
)
return len(container_ids)
|
prune_dangling(config)
Prune dangling generations of exactly this harness's images.
Idle CPU/IO priority when available — small per-run increments instead
of an hours-long backlog. A failed prune is warned about, never
raised: teardown must always run to completion. Returns the number
of pruned image records.
Source code in src/terok_util/matrix/runner.py
| def prune_dangling(config: MatrixConfig) -> int:
"""Prune dangling generations of exactly this harness's images.
Idle CPU/IO priority when available — small per-run increments instead
of an hours-long backlog. A failed prune is warned about, never
raised: teardown must always run to completion. Returns the number
of pruned image records.
"""
argv = [
"podman",
"image",
"prune",
"-f",
"--filter",
f"label={OWNERSHIP_LABEL}={config.image_prefix}",
]
for wrapper in (["ionice", "-c3"], ["nice", "-n19"]):
if which(wrapper[0]):
argv = wrapper + argv
pruned = subprocess.run(argv, check=False, capture_output=True, text=True) # nosec B603
if pruned.returncode != 0:
_warn_prune_failure(pruned.stderr)
return len(pruned.stdout.splitlines())
|
external_storage_leftovers()
Name the external (buildah) containers stuck holding storage.
These pin dangling image layers but were not created by this engine —
removing them here could tear a live podman build out from under
another process, so recovery is superbuild's job; the CLI only names
them. Returns the container names (or ids, for nameless entries).
Source code in src/terok_util/matrix/runner.py
| def external_storage_leftovers() -> list[str]:
"""Name the external (buildah) containers stuck holding storage.
These pin dangling image layers but were not created by this engine —
removing them here could tear a live ``podman build`` out from under
another process, so recovery is superbuild's job; the CLI only names
them. Returns the container names (or ids, for nameless entries).
"""
listing = subprocess.run( # nosec B603 B607 - fixed argv, podman from PATH by design
["podman", "ps", "-a", "--external", "--format", _EXTERNAL_PS_FORMAT],
check=False,
capture_output=True,
text=True,
)
leftovers = []
for line in listing.stdout.splitlines():
fields = line.split()
if len(fields) >= 2 and fields[1].lower() == _EXTERNAL_STORAGE_STATE:
leftovers.append(sanitize_tty(fields[2] if len(fields) > 2 else fields[0]))
return leftovers
|