didericis/bot-bottle
1
0
Fork 0
Code Issues 34 Pull Requests 11 Actions Packages Releases Wiki Activity

fix(cleanup): use authoritative resource identities

Browse Source
This commit is contained in:
didericis (codex)
2026-07-27 03:31:34 +00:00
committed by didericis
parent d04bbf1454
commit cdc5c8203d
8 changed files with 185 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bot_bottle/backend/docker/cleanup.py
+27 -17
View File
@@ -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.
bot_bottle/backend/firecracker/cleanup.py
+37 -17
View File
@@ -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 <run_root>/<slug>/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}")
bot_bottle/backend/firecracker/infra_artifact.py
+5 -1
View File
@@ -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)
bot_bottle/backend/firecracker/publish_infra.py
+15 -4
View File
@@ -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:
tests/unit/test_docker_cleanup.py
+35
View File
@@ -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()
tests/unit/test_firecracker_cleanup.py
+44 -12
View File
@@ -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)
tests/unit/test_infra_artifact.py
+12
View File
@@ -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")
tests/unit/test_publish_infra.py
+10
View File
@@ -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] = []