diff --git a/bot_bottle/backend/docker/cleanup.py b/bot_bottle/backend/docker/cleanup.py index 298d5fe6..4cc0f331 100644 --- a/bot_bottle/backend/docker/cleanup.py +++ b/bot_bottle/backend/docker/cleanup.py @@ -28,6 +28,7 @@ import subprocess from ...paths import bot_bottle_root from ...log import info, warn +from .. import EnumerationError from . import util as docker_mod from .bottle_cleanup_plan import DockerBottleCleanupPlan from ...bottle_state import bottle_state_dir, is_preserved @@ -36,15 +37,17 @@ from .compose import COMPOSE_PROJECT_PREFIX, list_compose_projects def _list_prefixed_containers() -> list[str]: """All bot-bottle-prefixed containers, running or stopped.""" - result = subprocess.run( - ["docker", "ps", "-a", - "--filter", f"name=^{COMPOSE_PROJECT_PREFIX}", - "--format", "{{.Names}}\t{{.Label \"com.docker.compose.project\"}}"], - capture_output=True, text=True, check=False, - ) + try: + result = subprocess.run( + ["docker", "ps", "-a", + "--filter", f"name=^{COMPOSE_PROJECT_PREFIX}", + "--format", "{{.Names}}\t{{.Label \"com.docker.compose.project\"}}"], + capture_output=True, text=True, check=False, + ) + except OSError as exc: + raise EnumerationError(f"docker ps failed: {exc}") from exc if result.returncode != 0: - warn(f"docker ps failed: {result.stderr.strip()}") - return [] + raise EnumerationError(f"docker ps failed: {result.stderr.strip()}") out: list[str] = [] for line in (result.stdout or "").splitlines(): if not line: @@ -63,15 +66,19 @@ def _list_prefixed_networks() -> list[str]: to a compose project. Compose-managed networks have a `com.docker.compose.project` label; bare ones (from pre-compose code paths) don't.""" - result = subprocess.run( - ["docker", "network", "ls", - "--filter", f"name={COMPOSE_PROJECT_PREFIX}", - "--format", "{{.Name}}\t{{.Label \"com.docker.compose.project\"}}"], - capture_output=True, text=True, check=False, - ) + try: + result = subprocess.run( + ["docker", "network", "ls", + "--filter", f"name={COMPOSE_PROJECT_PREFIX}", + "--format", "{{.Name}}\t{{.Label \"com.docker.compose.project\"}}"], + capture_output=True, text=True, check=False, + ) + except OSError as exc: + raise EnumerationError(f"docker network ls failed: {exc}") from exc if result.returncode != 0: - warn(f"docker network ls failed: {result.stderr.strip()}") - return [] + raise EnumerationError( + f"docker network ls failed: {result.stderr.strip()}" + ) out: list[str] = [] for line in (result.stdout or "").splitlines(): if not line: @@ -120,7 +127,10 @@ def prepare_cleanup() -> DockerBottleCleanupPlan: `enumerate_active_agents()` so the orphan-state-dir bucket doesn't include slugs whose non-docker bottle is still up.""" docker_mod.require_docker() - projects = list_compose_projects() + projects = list_compose_projects( + warn_on_error=False, + raise_on_error=True, + ) project_set = set(projects) # Late import to avoid a circular at module-load time — # the backend package's __init__ imports this module. diff --git a/bot_bottle/backend/firecracker/cleanup.py b/bot_bottle/backend/firecracker/cleanup.py index 21ec618c..f0f63b6f 100644 --- a/bot_bottle/backend/firecracker/cleanup.py +++ b/bot_bottle/backend/firecracker/cleanup.py @@ -11,10 +11,9 @@ Reaps *orphans* only — resources with no live VM behind them: — a VMM left lingering after its dir was removed. A run dir with a *live* firecracker process is a running bottle and is -left strictly alone: it is neither killed nor removed. (The backend's -`enumerate_active` registry is still a stub — #354 — so a live process -is the only reliable "this bottle is in use" signal we have. Once the -registry lands, registry-orphaned-but-running VMs can be reaped too.) +left strictly alone: it is neither killed nor removed. Active-agent +enumeration uses this same process snapshot, so cleanup and generic +backend consumers agree about which bottles are running. TAP slots free themselves (the flock drops when the launcher exits), so there is nothing to reclaim there. @@ -22,6 +21,7 @@ there is nothing to reclaim there. from __future__ import annotations +from collections.abc import Sequence import os import shutil import signal @@ -38,7 +38,7 @@ def _run_root() -> Path: return util.cache_dir() / "run" -def _run_dir_of(cmd: str, run_root: Path) -> Path | None: +def _run_dir_of(args: Sequence[str], run_root: Path) -> Path | None: """The bottle run dir a firecracker cmdline belongs to, or None. A bottle VM is launched with `--config-file //config.json`, @@ -46,15 +46,35 @@ def _run_dir_of(cmd: str, run_root: Path) -> Path | None: the run root. Anything else (a builder VM, the infra VM elsewhere) is not ours to reap here. """ - toks = cmd.split() - for i, tok in enumerate(toks): - if tok == "--config-file" and i + 1 < len(toks): - parent = Path(toks[i + 1]).parent + for i, arg in enumerate(args): + if arg == "--config-file" and i + 1 < len(args): + parent = Path(args[i + 1]).parent if parent.parent == run_root: return parent return None +def _decode_cmdline(raw: bytes) -> tuple[str, ...]: + """Decode Linux's NUL-delimited argv without losing embedded spaces.""" + return tuple( + value.decode(errors="surrogateescape") + for value in raw.split(b"\0") if value + ) + + +def _process_args(pid: int) -> tuple[str, ...] | None: + """Read one process's lossless argv, or None when it exited meanwhile.""" + try: + raw = Path(f"/proc/{pid}/cmdline").read_bytes() + except FileNotFoundError: + return None + except OSError as exc: + raise EnumerationError( + f"could not inspect Firecracker pid {pid}: {exc}" + ) from exc + return _decode_cmdline(raw) + + def _scan_processes(run_root: Path) -> tuple[set[str], list[int]]: """Inspect running firecracker VMs under ``run_root``. @@ -65,7 +85,7 @@ def _scan_processes(run_root: Path) -> tuple[set[str], list[int]]: """ try: result = subprocess.run( - ["pgrep", "-a", "firecracker"], + ["pgrep", "firecracker"], capture_output=True, text=True, check=False, ) except OSError as exc: @@ -83,14 +103,14 @@ def _scan_processes(run_root: Path) -> tuple[set[str], list[int]]: live: set[str] = set() orphan_pids: list[int] = [] for line in result.stdout.splitlines(): - parts = line.split(None, 1) - if len(parts) != 2: - continue try: - pid = int(parts[0]) + pid = int(line.strip()) except ValueError: continue - run_dir = _run_dir_of(parts[1], run_root) + args = _process_args(pid) + if args is None: + continue + run_dir = _run_dir_of(args, run_root) if run_dir is None: continue if run_dir.is_dir(): @@ -157,8 +177,8 @@ def _terminate_orphan(pid: int, run_root: Path) -> None: raise EnumerationError( f"could not revalidate Firecracker pid {pid}: {exc}" ) from exc - command = raw.replace(b"\0", b" ").decode(errors="replace") - run_dir = _run_dir_of(command, run_root) + args = _decode_cmdline(raw) + run_dir = _run_dir_of(args, run_root) if run_dir is None or run_dir.is_dir(): return info(f"kill firecracker VM pid {pid}") diff --git a/bot_bottle/backend/firecracker/infra_artifact.py b/bot_bottle/backend/firecracker/infra_artifact.py index 8695501b..078c67ed 100644 --- a/bot_bottle/backend/firecracker/infra_artifact.py +++ b/bot_bottle/backend/firecracker/infra_artifact.py @@ -41,6 +41,8 @@ from ... import resources from ...log import die, info from . import util +ARTIFACT_HTTP_TIMEOUT_SECONDS = 30.0 + # Bump if the on-disk artifact *format* changes (compression, layout) so a new # scheme can't collide with a cached/published artifact of the old one. _ARTIFACT_FORMAT = "1" @@ -164,7 +166,9 @@ def _download(url: str, dest: Path) -> None: """Stream `url` to `dest` (atomic via a `.part` sibling).""" tmp = dest.with_suffix(dest.suffix + ".part") try: - with urllib.request.urlopen(_open(url)) as resp, open(tmp, "wb") as out: + with urllib.request.urlopen( + _open(url), timeout=ARTIFACT_HTTP_TIMEOUT_SECONDS, + ) as resp, open(tmp, "wb") as out: shutil.copyfileobj(resp, out, _CHUNK) except urllib.error.HTTPError as e: tmp.unlink(missing_ok=True) diff --git a/bot_bottle/backend/firecracker/publish_infra.py b/bot_bottle/backend/firecracker/publish_infra.py index 6cfb651a..bbc84b1b 100644 --- a/bot_bottle/backend/firecracker/publish_infra.py +++ b/bot_bottle/backend/firecracker/publish_infra.py @@ -34,6 +34,7 @@ from pathlib import Path from . import infra_artifact, infra_vm, util _CHUNK = 1 << 20 +_REGISTRY_HTTP_TIMEOUT_SECONDS = 30.0 _GZ_NAME = "rootfs.ext4.gz" _SHA_NAME = "rootfs.ext4.gz.sha256" @@ -91,7 +92,9 @@ def _put(url: str, body: "bytes | Path", token: str) -> None: req.add_header("Authorization", f"token {token}") req.add_header("Content-Type", "application/octet-stream") try: - with urllib.request.urlopen(req) as resp: + with urllib.request.urlopen( + req, timeout=_REGISTRY_HTTP_TIMEOUT_SECONDS, + ) as resp: print(f" uploaded {url} (HTTP {resp.status})") except urllib.error.HTTPError as e: if e.code == 409: @@ -112,7 +115,9 @@ def _delete(url: str, token: str) -> None: if token: req.add_header("Authorization", f"token {token}") try: - with urllib.request.urlopen(req): + with urllib.request.urlopen( + req, timeout=_REGISTRY_HTTP_TIMEOUT_SECONDS, + ): pass except urllib.error.HTTPError as e: if e.code != 404: @@ -151,7 +156,10 @@ def _try_download_published(role: str, role_dir: Path) -> str | None: version = _role_version(role) sha_url = infra_artifact.artifact_url(version, _SHA_NAME, role=role) try: - with urllib.request.urlopen(infra_artifact._open(sha_url)): + with urllib.request.urlopen( + infra_artifact._open(sha_url), + timeout=_REGISTRY_HTTP_TIMEOUT_SECONDS, + ): pass except urllib.error.HTTPError as e: if e.code == 404: @@ -195,7 +203,10 @@ def _publish_bundle(role: str, role_dir: Path, token: str) -> str: # present, a re-publish is a no-op. Otherwise clear any partial upload left # by an interrupted prior attempt and upload the complete set. try: - with urllib.request.urlopen(infra_artifact._open(sha_url)) as resp: + with urllib.request.urlopen( + infra_artifact._open(sha_url), + timeout=_REGISTRY_HTTP_TIMEOUT_SECONDS, + ) as resp: remote_sha = resp.read().decode("utf-8").split()[0].strip().lower() except urllib.error.HTTPError as e: if e.code != 404: diff --git a/tests/unit/test_docker_cleanup.py b/tests/unit/test_docker_cleanup.py index 730f20ff..252098c8 100644 --- a/tests/unit/test_docker_cleanup.py +++ b/tests/unit/test_docker_cleanup.py @@ -14,9 +14,12 @@ from __future__ import annotations import tempfile import unittest from pathlib import Path +from unittest.mock import patch from tests.unit import use_bottle_root from bot_bottle import bottle_state +from bot_bottle.backend import EnumerationError +from bot_bottle.backend.docker import cleanup from bot_bottle.backend.docker.cleanup import _list_orphan_state_dirs @@ -116,5 +119,37 @@ class TestOrphanStateDirs(_FakeHomeMixin, unittest.TestCase): ) +class TestAuthoritativeDiscovery(unittest.TestCase): + def test_prepare_requires_authoritative_compose_projects(self): + with patch.object(cleanup.docker_mod, "require_docker"), \ + patch.object( + cleanup, "list_compose_projects", + side_effect=EnumerationError("compose unavailable"), + ) as projects, self.assertRaisesRegex( + EnumerationError, "compose unavailable", + ): + cleanup.prepare_cleanup() + projects.assert_called_once_with( + warn_on_error=False, + raise_on_error=True, + ) + + def test_container_query_failure_raises(self): + failed = cleanup.subprocess.CompletedProcess( + [], 1, stdout="", stderr="daemon unavailable", + ) + with patch.object(cleanup.subprocess, "run", return_value=failed), \ + self.assertRaisesRegex(EnumerationError, "daemon unavailable"): + cleanup._list_prefixed_containers() + + def test_network_query_failure_raises(self): + failed = cleanup.subprocess.CompletedProcess( + [], 1, stdout="", stderr="daemon unavailable", + ) + with patch.object(cleanup.subprocess, "run", return_value=failed), \ + self.assertRaisesRegex(EnumerationError, "daemon unavailable"): + cleanup._list_prefixed_networks() + + if __name__ == "__main__": unittest.main() diff --git a/tests/unit/test_firecracker_cleanup.py b/tests/unit/test_firecracker_cleanup.py index 069cfe38..f41256e8 100644 --- a/tests/unit/test_firecracker_cleanup.py +++ b/tests/unit/test_firecracker_cleanup.py @@ -32,15 +32,19 @@ class TestProcessScan(unittest.TestCase): self.assertEqual( Path("/cache/run/dev-a"), fc_cleanup._run_dir_of( - f"firecracker --config-file {run_root}/dev-a/config.json", run_root + ("firecracker", "--config-file", + f"{run_root}/dev-a/config.json"), run_root ), ) # infra/builder VMs elsewhere, or nested paths, are not ours. self.assertIsNone( - fc_cleanup._run_dir_of("firecracker --config-file /elsewhere/config.json", run_root) + fc_cleanup._run_dir_of( + ("firecracker", "--config-file", "/elsewhere/config.json"), + run_root, + ) ) self.assertIsNone( - fc_cleanup._run_dir_of("firecracker --no-config", run_root) + fc_cleanup._run_dir_of(("firecracker", "--no-config"), run_root) ) def test_scan_splits_live_dirs_from_orphan_pids(self): @@ -48,17 +52,40 @@ class TestProcessScan(unittest.TestCase): run_root = Path(tmp) (run_root / "live-a").mkdir() # dir present -> live VM, protected # "gone-b" dir intentionally absent -> lingering VMM, orphan pid - out = ( - f"111 firecracker --config-file {run_root}/live-a/config.json\n" - f"222 firecracker --config-file {run_root}/gone-b/config.json\n" - "333 firecracker --config-file /elsewhere/config.json\n" - "notanint firecracker --config-file x\n" - ) - with patch.object(fc_cleanup.subprocess, "run", return_value=_proc(out)): + args = { + 111: ("firecracker", "--config-file", + f"{run_root}/live-a/config.json"), + 222: ("firecracker", "--config-file", + f"{run_root}/gone-b/config.json"), + 333: ("firecracker", "--config-file", "/elsewhere/config.json"), + } + with patch.object( + fc_cleanup.subprocess, "run", + return_value=_proc("111\n222\n333\nnotanint\n"), + ), patch.object( + fc_cleanup, "_process_args", side_effect=args.get, + ): live, orphan_pids = fc_cleanup._scan_processes(run_root) self.assertEqual({str(run_root / "live-a")}, live) self.assertEqual([222], orphan_pids) + def test_scan_preserves_spaces_in_config_path(self): + with tempfile.TemporaryDirectory(prefix="fc cache ") as tmp: + run_root = Path(tmp) + live = run_root / "live bottle" + live.mkdir() + with patch.object( + fc_cleanup.subprocess, "run", return_value=_proc("111\n"), + ), patch.object( + fc_cleanup, "_process_args", + return_value=("firecracker", "--config-file", + str(live / "config.json")), + ): + self.assertEqual( + ({str(live)}, []), + fc_cleanup._scan_processes(run_root), + ) + def test_scan_empty_when_pgrep_finds_no_processes(self): with patch.object(fc_cleanup.subprocess, "run", return_value=_proc(returncode=1)): self.assertEqual((set(), []), fc_cleanup._scan_processes(Path("/x"))) @@ -117,9 +144,14 @@ class TestProcessScan(unittest.TestCase): run_root = Path(tmp) (run_root / "live-a").mkdir() (run_root / "dead-b").mkdir() - out = f"111 firecracker --config-file {run_root}/live-a/config.json\n" with patch.object(fc_cleanup, "_run_root", return_value=run_root), \ - patch.object(fc_cleanup.subprocess, "run", return_value=_proc(out)): + patch.object(fc_cleanup.subprocess, "run", + return_value=_proc("111\n")), \ + patch.object( + fc_cleanup, "_process_args", + return_value=("firecracker", "--config-file", + str(run_root / "live-a/config.json")), + ): plan = fc_cleanup.prepare_cleanup() self.assertEqual((), plan.vm_pids) self.assertEqual((str(run_root / "dead-b"),), plan.run_dirs) diff --git a/tests/unit/test_infra_artifact.py b/tests/unit/test_infra_artifact.py index 7f660344..69421154 100644 --- a/tests/unit/test_infra_artifact.py +++ b/tests/unit/test_infra_artifact.py @@ -284,6 +284,18 @@ class TestEnsureArtifact(_CacheMixin): ia.ensure_artifact_gz(version, role=_ROLE) self.assertEqual(first_calls, len(net.calls)) + def test_download_uses_network_deadline(self) -> None: + response = mock.MagicMock() + response.__enter__.return_value = io.BytesIO(b"payload") + with tempfile.TemporaryDirectory() as d, mock.patch.object( + ia.urllib.request, "urlopen", return_value=response, + ) as urlopen: + ia._download("https://registry/artifact", Path(d) / "artifact") + self.assertEqual( + ia.ARTIFACT_HTTP_TIMEOUT_SECONDS, + urlopen.call_args.kwargs["timeout"], + ) + def test_checksum_mismatch_fails_closed(self) -> None: version = "beefbeefbeefbeef" gz = _gz(b"payload") diff --git a/tests/unit/test_publish_infra.py b/tests/unit/test_publish_infra.py index 265b17b7..ddaf982c 100644 --- a/tests/unit/test_publish_infra.py +++ b/tests/unit/test_publish_infra.py @@ -49,6 +49,16 @@ class TestPut(unittest.TestCase): self.assertNotIsInstance(req.data, (bytes, bytearray)) self.assertEqual(str(len(payload)), req.get_header("Content-length")) + def test_put_uses_network_deadline(self) -> None: + with mock.patch.object( + pub.urllib.request, "urlopen", return_value=_Resp(), + ) as urlopen: + pub._put("https://reg/pkg", b"payload", token="") + self.assertEqual( + pub._REGISTRY_HTTP_TIMEOUT_SECONDS, + urlopen.call_args.kwargs["timeout"], + ) + def test_small_bytes_body_still_works(self) -> None: captured: list[urllib.request.Request] = []