389 lines
16 KiB
Python
389 lines
16 KiB
Python
"""Inventory stale candidate Compose stacks and optionally stop one exact, quiescent project.
|
|
|
|
The default inventory mode is read-only. The only mutation available is ``docker stop`` for
|
|
containers inspected under one exact ``modelforge-validation-*`` Compose project label. Volumes and
|
|
images are never removed.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import stat
|
|
import subprocess
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path, PurePosixPath
|
|
from typing import Any
|
|
|
|
PROJECT_PREFIX = "modelforge-validation-"
|
|
PROJECT_PATTERN = re.compile(
|
|
r"modelforge-validation-[a-z0-9](?:[a-z0-9-]{0,39}[a-z0-9])?\Z"
|
|
)
|
|
PROJECT_LABEL = "com.docker.compose.project"
|
|
SERVICE_LABEL = "com.docker.compose.service"
|
|
ONEOFF_LABEL = "com.docker.compose.oneoff"
|
|
CONFIG_TARGET = "/app/config"
|
|
CONFIG_MANIFEST = "models/initial-candidates.yaml"
|
|
DEFAULT_STALE_AFTER_HOURS = 24
|
|
STOPPABLE_RUNNING_SERVICES = {"postgres", "redis", "web"}
|
|
NEVER_STOP_RUNNING_SERVICES = {"api", "migrate", "node-agent", "runtime-worker"}
|
|
|
|
|
|
class RetentionError(ValueError):
|
|
"""Docker state or operator input cannot be used safely."""
|
|
|
|
|
|
def validate_project(project: str) -> str:
|
|
if PROJECT_PATTERN.fullmatch(project) is None:
|
|
raise RetentionError(
|
|
"project must be an exact bounded modelforge-validation-* name"
|
|
)
|
|
return project
|
|
|
|
|
|
def _mapping(value: object, label: str) -> dict[str, Any]:
|
|
if not isinstance(value, dict):
|
|
raise RetentionError(f"Docker returned invalid {label}")
|
|
return value
|
|
|
|
|
|
def _unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
|
|
result: dict[str, Any] = {}
|
|
for key, value in pairs:
|
|
if key in result:
|
|
raise RetentionError(f"evidence record has duplicate field: {key}")
|
|
result[key] = value
|
|
return result
|
|
|
|
|
|
def _docker(*args: str) -> str:
|
|
try:
|
|
completed = subprocess.run(
|
|
("docker", *args),
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
)
|
|
except OSError as exc:
|
|
raise RetentionError("Docker CLI could not be started") from exc
|
|
if completed.returncode:
|
|
raise RetentionError(f"Docker command failed: {' '.join(args)}")
|
|
return completed.stdout
|
|
|
|
|
|
def _parse_inspection(raw: str, identifier: str) -> dict[str, Any]:
|
|
try:
|
|
value = json.loads(raw)
|
|
except json.JSONDecodeError as exc:
|
|
raise RetentionError("Docker returned invalid container inspection JSON") from exc
|
|
if not isinstance(value, list) or len(value) != 1:
|
|
raise RetentionError("Docker returned ambiguous container inspection data")
|
|
item = _mapping(value[0], "container inspection")
|
|
inspected_id = item.get("Id")
|
|
if not isinstance(inspected_id, str) or inspected_id != identifier:
|
|
raise RetentionError("Docker returned a different container than requested")
|
|
return item
|
|
|
|
|
|
def _created_at(value: object) -> datetime:
|
|
if not isinstance(value, str) or not value:
|
|
raise RetentionError("container creation time is missing")
|
|
try:
|
|
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
except ValueError as exc:
|
|
raise RetentionError("container creation time is invalid") from exc
|
|
if parsed.tzinfo is None:
|
|
raise RetentionError("container creation time has no timezone")
|
|
return parsed.astimezone(UTC)
|
|
|
|
|
|
def _path_stat(path: Path) -> object | None:
|
|
try:
|
|
return path.stat()
|
|
except (FileNotFoundError, NotADirectoryError):
|
|
return None
|
|
except OSError as exc:
|
|
raise RetentionError(f"cannot inspect bind source path: {path}") from exc
|
|
|
|
|
|
def _bind_sources(container: dict[str, Any]) -> list[dict[str, object]]:
|
|
mounts = container.get("Mounts")
|
|
if not isinstance(mounts, list):
|
|
raise RetentionError("container mount data is missing or ambiguous")
|
|
result: list[dict[str, object]] = []
|
|
for raw_mount in mounts:
|
|
mount = _mapping(raw_mount, "mount")
|
|
if mount.get("Type") != "bind":
|
|
continue
|
|
source = mount.get("Source")
|
|
destination = mount.get("Destination")
|
|
if not isinstance(source, str) or not source or not Path(source).is_absolute():
|
|
raise RetentionError("container has an invalid bind source")
|
|
if not isinstance(destination, str) or not destination.startswith("/"):
|
|
raise RetentionError("container has an invalid bind destination")
|
|
source_path = Path(source)
|
|
source_stat = _path_stat(source_path)
|
|
source_exists = source_stat is not None
|
|
entry: dict[str, object] = {
|
|
"destination": destination,
|
|
"source": str(source_path),
|
|
"source_exists": source_exists,
|
|
}
|
|
if PurePosixPath(destination).as_posix().rstrip("/") == CONFIG_TARGET:
|
|
manifest = source_path / CONFIG_MANIFEST
|
|
manifest_stat = _path_stat(manifest) if source_exists else None
|
|
entry["manifest"] = str(manifest)
|
|
entry["manifest_exists"] = (
|
|
manifest_stat is not None and stat.S_ISREG(manifest_stat.st_mode)
|
|
)
|
|
result.append(entry)
|
|
return result
|
|
|
|
|
|
def _inspect_container(identifier: str) -> dict[str, Any]:
|
|
item = _parse_inspection(_docker("container", "inspect", identifier), identifier)
|
|
config = _mapping(item.get("Config"), "container config")
|
|
labels = config.get("Labels")
|
|
if not isinstance(labels, dict):
|
|
raise RetentionError("container Compose labels are missing or ambiguous")
|
|
project = labels.get(PROJECT_LABEL)
|
|
if not isinstance(project, str) or not project:
|
|
raise RetentionError("container project label is invalid")
|
|
if not project.startswith(PROJECT_PREFIX):
|
|
return {"project": project}
|
|
validate_project(project)
|
|
|
|
name = item.get("Name")
|
|
if not isinstance(name, str) or not name.startswith("/"):
|
|
raise RetentionError("container project label or name is invalid")
|
|
state = _mapping(item.get("State"), "container state")
|
|
status = state.get("Status")
|
|
if not isinstance(status, str) or not status:
|
|
raise RetentionError("container state status is missing")
|
|
service = labels.get(SERVICE_LABEL)
|
|
if not isinstance(service, str) or not service:
|
|
raise RetentionError("container Compose service label is missing")
|
|
return {
|
|
"id": identifier,
|
|
"name": name.lstrip("/"),
|
|
"project": project,
|
|
"service": service,
|
|
"oneoff": labels.get(ONEOFF_LABEL),
|
|
"status": status.lower(),
|
|
"created_at": _created_at(item.get("Created")),
|
|
"bind_sources": _bind_sources(item),
|
|
}
|
|
|
|
|
|
def collect_projects(
|
|
*, now: datetime | None = None, stale_after_hours: int = DEFAULT_STALE_AFTER_HOURS
|
|
) -> list[dict[str, Any]]:
|
|
"""Inventory Compose containers and group only exact validation project labels."""
|
|
|
|
if stale_after_hours < 1 or stale_after_hours > 8760:
|
|
raise RetentionError("stale age must be between 1 and 8760 hours")
|
|
identifiers = [
|
|
line.strip()
|
|
for line in _docker(
|
|
"container", "ls", "--all", "--quiet", "--no-trunc", "--filter", f"label={PROJECT_LABEL}"
|
|
).splitlines()
|
|
if line.strip()
|
|
]
|
|
if len(set(identifiers)) != len(identifiers):
|
|
raise RetentionError("Docker returned duplicate container identifiers")
|
|
|
|
projects: dict[str, list[dict[str, Any]]] = {}
|
|
for identifier in identifiers:
|
|
container = _inspect_container(identifier)
|
|
project = container["project"]
|
|
if project.startswith(PROJECT_PREFIX):
|
|
validate_project(project)
|
|
if not container["name"].startswith(f"{project}-"):
|
|
raise RetentionError(f"container name is outside exact project {project}")
|
|
projects.setdefault(project, []).append(container)
|
|
|
|
current = (now or datetime.now(UTC)).astimezone(UTC)
|
|
result: list[dict[str, Any]] = []
|
|
for project, containers in sorted(projects.items()):
|
|
created_values = [item["created_at"] for item in containers]
|
|
oldest = min(created_values)
|
|
age_seconds = int((current - oldest).total_seconds())
|
|
if age_seconds < 0:
|
|
raise RetentionError(f"container creation time is in the future for {project}")
|
|
sources = [
|
|
{"container": item["name"], **source}
|
|
for item in containers
|
|
for source in item["bind_sources"]
|
|
]
|
|
result.append(
|
|
{
|
|
"project": project,
|
|
"oldest_container_created_at": oldest.isoformat().replace("+00:00", "Z"),
|
|
"age_hours": round(age_seconds / 3600, 2),
|
|
"stale": age_seconds >= stale_after_hours * 3600,
|
|
"missing_bind_sources": [source for source in sources if not source["source_exists"]],
|
|
"missing_manifests": [
|
|
source for source in sources if source.get("manifest") and not source["manifest_exists"]
|
|
],
|
|
"containers": [
|
|
{
|
|
key: (
|
|
item[key].isoformat().replace("+00:00", "Z")
|
|
if key == "created_at"
|
|
else item[key]
|
|
)
|
|
for key in ("id", "name", "service", "status", "created_at")
|
|
}
|
|
for item in sorted(containers, key=lambda row: row["name"])
|
|
],
|
|
}
|
|
)
|
|
return result
|
|
|
|
|
|
def _evidence_record(path: Path | None, *, project: str) -> Path:
|
|
if path is None:
|
|
raise RetentionError("stop requires --evidence-record for retained evidence")
|
|
try:
|
|
resolved = path.resolve(strict=True)
|
|
record_stat = resolved.stat()
|
|
if not stat.S_ISREG(record_stat.st_mode) or record_stat.st_size == 0:
|
|
raise RetentionError("evidence record must be a nonempty regular file")
|
|
record = json.loads(resolved.read_text(encoding="utf-8"), object_pairs_hook=_unique_object)
|
|
except OSError as exc:
|
|
raise RetentionError("evidence record cannot be read") from exc
|
|
except RuntimeError as exc:
|
|
raise RetentionError("evidence record path is ambiguous") from exc
|
|
except json.JSONDecodeError as exc:
|
|
raise RetentionError("evidence record must be valid JSON") from exc
|
|
except UnicodeError as exc:
|
|
raise RetentionError("evidence record must be valid UTF-8 JSON") from exc
|
|
if (
|
|
not isinstance(record, dict)
|
|
or set(record) != {"schema_version", "project", "retained", "evidence_reference"}
|
|
or type(record["schema_version"]) is not int
|
|
or record["schema_version"] != 1
|
|
or record["project"] != project
|
|
or record["retained"] is not True
|
|
or not isinstance(record["evidence_reference"], str)
|
|
or not record["evidence_reference"].strip()
|
|
):
|
|
raise RetentionError("evidence record must identify this project and retained evidence")
|
|
return resolved
|
|
|
|
|
|
def stop_project(
|
|
project: str,
|
|
*,
|
|
confirm_stop: str | None,
|
|
confirm_no_active_jobs: bool,
|
|
evidence_record: Path | None,
|
|
stale_after_hours: int,
|
|
now: datetime | None = None,
|
|
) -> dict[str, object]:
|
|
"""Stop running containers for one exact stale or incomplete validation project."""
|
|
|
|
validate_project(project)
|
|
if confirm_stop != project:
|
|
raise RetentionError("stop requires --confirm-stop with the exact project name")
|
|
if not confirm_no_active_jobs:
|
|
raise RetentionError("stop requires --confirm-no-active-jobs")
|
|
evidence = _evidence_record(evidence_record, project=project)
|
|
|
|
projects = collect_projects(now=now, stale_after_hours=stale_after_hours)
|
|
selected = [item for item in projects if item["project"] == project]
|
|
if len(selected) != 1:
|
|
raise RetentionError(f"exact validation project was not found unambiguously: {project}")
|
|
details = selected[0]
|
|
if not details["stale"] and not details["missing_bind_sources"] and not details["missing_manifests"]:
|
|
raise RetentionError("stop is limited to a stale project or one with missing bind-source files")
|
|
|
|
project_containers = [
|
|
_inspect_container(container["id"])
|
|
for container in details["containers"]
|
|
]
|
|
running: list[dict[str, Any]] = []
|
|
for container in project_containers:
|
|
if container["status"] not in {"running", "restarting", "paused"}:
|
|
continue
|
|
service = container["service"]
|
|
if service == "api":
|
|
raise RetentionError("refusing stop while the project API is running")
|
|
if container["status"] != "running":
|
|
raise RetentionError(f"refusing stop while service state is ambiguous: {container['status']}")
|
|
if str(container["oneoff"]).lower() in {"true", "1", "yes"}:
|
|
raise RetentionError("refusing stop while a one-off job container is running")
|
|
if service in NEVER_STOP_RUNNING_SERVICES or service not in STOPPABLE_RUNNING_SERVICES:
|
|
raise RetentionError(f"refusing stop while service activity is unknown: {service}")
|
|
running.append(container)
|
|
|
|
stopped: list[str] = []
|
|
if running:
|
|
identifiers = [item["id"] for item in running]
|
|
_docker("stop", "--time", "30", *identifiers)
|
|
stopped = identifiers
|
|
# Confirm every ID still has the same exact project label and is no longer running.
|
|
for item in running:
|
|
identifier = item["id"]
|
|
after = _inspect_container(identifier)
|
|
if (
|
|
after["project"] != project
|
|
or after["name"] != item["name"]
|
|
or after["service"] != item["service"]
|
|
):
|
|
raise RetentionError("container project label changed during stop")
|
|
if after["status"] in {"running", "restarting", "paused"}:
|
|
raise RetentionError("Docker reported success but a selected container is still running")
|
|
return {
|
|
"action": "stop",
|
|
"project": project,
|
|
"stopped_container_ids": stopped,
|
|
"evidence_record": str(evidence),
|
|
"volumes_or_images_removed": False,
|
|
}
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--stale-after-hours", type=int, default=DEFAULT_STALE_AFTER_HOURS)
|
|
parser.add_argument(
|
|
"--stop-project", help="stop eligible containers for one exact validation project"
|
|
)
|
|
parser.add_argument("--confirm-stop", help="must exactly match --stop-project")
|
|
parser.add_argument("--confirm-no-active-jobs", action="store_true")
|
|
parser.add_argument("--evidence-record", type=Path, help="nonempty file proving evidence was retained")
|
|
args = parser.parse_args(argv)
|
|
try:
|
|
if args.stop_project is None:
|
|
if args.confirm_stop or args.confirm_no_active_jobs or args.evidence_record is not None:
|
|
raise RetentionError("stop confirmation options require --stop-project")
|
|
result: dict[str, object] = {
|
|
"result": "PASS",
|
|
"mode": "read-only",
|
|
"stale_after_hours": args.stale_after_hours,
|
|
"projects": collect_projects(stale_after_hours=args.stale_after_hours),
|
|
}
|
|
else:
|
|
result = {
|
|
"result": "PASS",
|
|
**stop_project(
|
|
args.stop_project,
|
|
confirm_stop=args.confirm_stop,
|
|
confirm_no_active_jobs=args.confirm_no_active_jobs,
|
|
evidence_record=args.evidence_record,
|
|
stale_after_hours=args.stale_after_hours,
|
|
),
|
|
}
|
|
except (RetentionError, OSError) as exc:
|
|
print(json.dumps({"result": "FAIL", "error": str(exc)}, sort_keys=True))
|
|
return 2
|
|
print(json.dumps(result, sort_keys=True))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|