menzeljandClaude Opus 4.8
5b59f5e8f9
Stacks overview and detail were doing an N+1 inspect storm: list_stacks called containers_for_stack AND compute_status (which re-fetched) per stack, and containers.list(sparse=False) full-inspects every container plus c.image triggered an image-inspect each. For N stacks that was ~2N*(1 list + M inspects + M image-inspects) sequential socket round trips (~1s for just 2 stacks, growing linearly). - compose_service.stack_status_summaries(): one low-level api.containers(all=True) summary call grouped by compose project label → whole list served in a single Docker round-trip (~10x faster). - compute_status() takes optional pre-fetched containers; get_stack and _stack_summary no longer double-fetch. - containers_for_stack() reads the image name from the inspect it already has instead of c.image (drops the per-container image-inspect). - Same batching applied to the agent's stack list/detail. Also: Forgejo (registry + git) moved to 10.10.6.10:3020 — updated image refs in docker-compose.yml, agent/Dockerfile, agent/docker-compose.yml. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
397 lines
12 KiB
Python
397 lines
12 KiB
Python
"""File-based stack storage and Docker Compose lifecycle.
|
||
|
||
The compose YAML on disk is always the source of truth. The database only
|
||
stores metadata (name, description, timestamps).
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import os
|
||
import re
|
||
import shutil
|
||
from dataclasses import dataclass, field
|
||
from datetime import datetime, timezone
|
||
from typing import Optional
|
||
|
||
from config import settings
|
||
from docker_client import DockerError, get_client, safe_call
|
||
|
||
COMPOSE_FILENAMES = ("compose.yaml", "compose.yml", "docker-compose.yml", "docker-compose.yaml")
|
||
DEFAULT_COMPOSE_NAME = "compose.yaml"
|
||
COMPOSE_LABEL = "com.docker.compose.project"
|
||
SERVICE_LABEL = "com.docker.compose.service"
|
||
|
||
|
||
class StackFileError(Exception):
|
||
pass
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Slug / paths
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
def slugify(name: str) -> str:
|
||
slug = re.sub(r"[^a-z0-9_-]+", "-", name.strip().lower())
|
||
slug = re.sub(r"-{2,}", "-", slug).strip("-_")
|
||
return slug or "stack"
|
||
|
||
|
||
def stacks_root(override: Optional[str] = None) -> str:
|
||
return override or settings.STACKS_DIR
|
||
|
||
|
||
def stack_dir(stack_id: str, override: Optional[str] = None) -> str:
|
||
return os.path.join(stacks_root(override), stack_id)
|
||
|
||
|
||
def find_compose_file(directory: str) -> Optional[str]:
|
||
for name in COMPOSE_FILENAMES:
|
||
candidate = os.path.join(directory, name)
|
||
if os.path.isfile(candidate):
|
||
return candidate
|
||
return None
|
||
|
||
|
||
def compose_path(stack_id: str, override: Optional[str] = None) -> str:
|
||
directory = stack_dir(stack_id, override)
|
||
return find_compose_file(directory) or os.path.join(directory, DEFAULT_COMPOSE_NAME)
|
||
|
||
|
||
def env_path(stack_id: str, override: Optional[str] = None) -> str:
|
||
return os.path.join(stack_dir(stack_id, override), ".env")
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Read / write files
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
def read_compose(stack_id: str, override: Optional[str] = None) -> str:
|
||
path = compose_path(stack_id, override)
|
||
if not os.path.isfile(path):
|
||
return ""
|
||
with open(path, "r", encoding="utf-8") as fh:
|
||
return fh.read()
|
||
|
||
|
||
def read_env(stack_id: str, override: Optional[str] = None) -> str:
|
||
path = env_path(stack_id, override)
|
||
if not os.path.isfile(path):
|
||
return ""
|
||
with open(path, "r", encoding="utf-8") as fh:
|
||
return fh.read()
|
||
|
||
|
||
def write_compose(stack_id: str, content: str, override: Optional[str] = None) -> None:
|
||
directory = stack_dir(stack_id, override)
|
||
os.makedirs(directory, exist_ok=True)
|
||
path = compose_path(stack_id, override)
|
||
# Non-destructive: back up existing file first.
|
||
if os.path.isfile(path):
|
||
shutil.copy2(path, path + ".bak")
|
||
tmp = path + ".tmp"
|
||
with open(tmp, "w", encoding="utf-8") as fh:
|
||
fh.write(content)
|
||
os.replace(tmp, path)
|
||
|
||
|
||
def write_env(stack_id: str, content: str, override: Optional[str] = None) -> None:
|
||
directory = stack_dir(stack_id, override)
|
||
os.makedirs(directory, exist_ok=True)
|
||
path = env_path(stack_id, override)
|
||
with open(path, "w", encoding="utf-8") as fh:
|
||
fh.write(content)
|
||
|
||
|
||
def delete_stack_files(stack_id: str, override: Optional[str] = None) -> None:
|
||
directory = stack_dir(stack_id, override)
|
||
if os.path.isdir(directory):
|
||
shutil.rmtree(directory)
|
||
|
||
|
||
def clone_stack_files(src_id: str, dst_id: str, override: Optional[str] = None) -> None:
|
||
src = stack_dir(src_id, override)
|
||
dst = stack_dir(dst_id, override)
|
||
if os.path.isdir(dst):
|
||
raise StackFileError(f"Target stack '{dst_id}' already exists")
|
||
shutil.copytree(src, dst)
|
||
|
||
|
||
def discover_stacks(override: Optional[str] = None) -> list[str]:
|
||
"""Return ids of all directories under STACKS_DIR that contain a compose file."""
|
||
root = stacks_root(override)
|
||
if not os.path.isdir(root):
|
||
return []
|
||
found = []
|
||
for entry in sorted(os.listdir(root)):
|
||
directory = os.path.join(root, entry)
|
||
if os.path.isdir(directory) and find_compose_file(directory):
|
||
found.append(entry)
|
||
return found
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Live status from Docker
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
@dataclass
|
||
class ContainerInfo:
|
||
id: str
|
||
name: str
|
||
service: str
|
||
image: str
|
||
state: str # running, exited, ...
|
||
status: str # human string
|
||
health: Optional[str] = None
|
||
ports: list[dict] = field(default_factory=list)
|
||
created: Optional[str] = None
|
||
|
||
|
||
def _parse_ports(attrs: dict) -> list[dict]:
|
||
ports = []
|
||
bindings = (attrs.get("NetworkSettings") or {}).get("Ports") or {}
|
||
for container_port, host in (bindings or {}).items():
|
||
if host:
|
||
for binding in host:
|
||
ports.append(
|
||
{
|
||
"container": container_port,
|
||
"host_ip": binding.get("HostIp"),
|
||
"host_port": binding.get("HostPort"),
|
||
}
|
||
)
|
||
else:
|
||
ports.append({"container": container_port, "host_port": None})
|
||
return ports
|
||
|
||
|
||
def containers_for_stack(stack_id: str) -> list[ContainerInfo]:
|
||
client = get_client()
|
||
raw = safe_call(
|
||
client.containers.list,
|
||
all=True,
|
||
filters={"label": f"{COMPOSE_LABEL}={stack_id}"},
|
||
)
|
||
result = []
|
||
for c in raw:
|
||
attrs = c.attrs
|
||
state = attrs.get("State", {}) or {}
|
||
health = (state.get("Health") or {}).get("Status")
|
||
result.append(
|
||
ContainerInfo(
|
||
id=c.id,
|
||
name=c.name,
|
||
service=c.labels.get(SERVICE_LABEL, c.name),
|
||
# Use the configured image name from the inspect we already have;
|
||
# reading c.image.tags would trigger a separate image-inspect call
|
||
# per container.
|
||
image=attrs.get("Config", {}).get("Image", "") or attrs.get("Image", ""),
|
||
state=state.get("Status", c.status),
|
||
status=attrs.get("State", {}).get("Status", c.status),
|
||
health=health,
|
||
ports=_parse_ports(attrs),
|
||
created=attrs.get("Created"),
|
||
)
|
||
)
|
||
return result
|
||
|
||
|
||
# in-memory set of stacks currently performing a pull/up
|
||
_BUSY: set[str] = set()
|
||
|
||
|
||
def mark_busy(stack_id: str) -> None:
|
||
_BUSY.add(stack_id)
|
||
|
||
|
||
def clear_busy(stack_id: str) -> None:
|
||
_BUSY.discard(stack_id)
|
||
|
||
|
||
def is_busy(stack_id: str) -> bool:
|
||
return stack_id in _BUSY
|
||
|
||
|
||
def _status_from_states(states: list[str]) -> str:
|
||
if not states:
|
||
return "stopped"
|
||
if any(s == "dead" for s in states):
|
||
return "error"
|
||
running = [s for s in states if s == "running"]
|
||
if len(running) == len(states):
|
||
return "running"
|
||
if running:
|
||
return "partial"
|
||
return "stopped"
|
||
|
||
|
||
def compute_status(stack_id: str, containers: Optional[list[ContainerInfo]] = None) -> str:
|
||
"""Status for one stack. Pass already-fetched ``containers`` to avoid a
|
||
redundant Docker round-trip (the detail view already has them)."""
|
||
if stack_id in _BUSY:
|
||
return "updating"
|
||
try:
|
||
if containers is None:
|
||
containers = containers_for_stack(stack_id)
|
||
except DockerError:
|
||
return "unknown"
|
||
return _status_from_states([c.state for c in containers])
|
||
|
||
|
||
def stack_status_summaries() -> dict[str, dict]:
|
||
"""One cheap Docker call → ``{project: {status, total, running}}`` for every
|
||
stack at once.
|
||
|
||
Uses the low-level container *summary* list (``GET /containers/json``, no
|
||
per-container inspect) grouped by the compose project label. This is far
|
||
cheaper than calling :func:`containers_for_stack` (which full-inspects every
|
||
container) once per stack for the overview — turning the stacks list from
|
||
O(stacks × containers) Docker round-trips into a single one.
|
||
"""
|
||
client = get_client()
|
||
raw = safe_call(client.api.containers, all=True)
|
||
by_project: dict[str, list[str]] = {}
|
||
for c in raw:
|
||
project = (c.get("Labels") or {}).get(COMPOSE_LABEL)
|
||
if project:
|
||
by_project.setdefault(project, []).append(c.get("State", ""))
|
||
return {
|
||
project: {
|
||
"status": _status_from_states(states),
|
||
"total": len(states),
|
||
"running": sum(1 for s in states if s == "running"),
|
||
}
|
||
for project, states in by_project.items()
|
||
}
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Compose CLI lifecycle (async subprocess)
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
def _compose_base_cmd(stack_id: str, override: Optional[str] = None) -> list[str]:
|
||
directory = stack_dir(stack_id, override)
|
||
compose_file = find_compose_file(directory) or os.path.join(directory, DEFAULT_COMPOSE_NAME)
|
||
return [
|
||
"docker",
|
||
"compose",
|
||
"-p",
|
||
stack_id,
|
||
"--project-directory",
|
||
directory,
|
||
"-f",
|
||
compose_file,
|
||
]
|
||
|
||
|
||
async def run_compose(
|
||
stack_id: str,
|
||
args: list[str],
|
||
override: Optional[str] = None,
|
||
timeout: float = 600.0,
|
||
) -> dict:
|
||
"""Run a `docker compose` subcommand. Returns {returncode, stdout, stderr}."""
|
||
cmd = _compose_base_cmd(stack_id, override) + args
|
||
proc = await asyncio.create_subprocess_exec(
|
||
*cmd,
|
||
stdout=asyncio.subprocess.PIPE,
|
||
stderr=asyncio.subprocess.PIPE,
|
||
)
|
||
try:
|
||
stdout_b, stderr_b = await asyncio.wait_for(proc.communicate(), timeout=timeout)
|
||
except asyncio.TimeoutError as exc:
|
||
proc.kill()
|
||
raise StackFileError(f"compose command timed out: {' '.join(args)}") from exc
|
||
return {
|
||
"returncode": proc.returncode,
|
||
"stdout": stdout_b.decode("utf-8", "replace"),
|
||
"stderr": stderr_b.decode("utf-8", "replace"),
|
||
"command": " ".join(args),
|
||
}
|
||
|
||
|
||
async def stream_compose(
|
||
stack_id: str, args: list[str], override: Optional[str] = None
|
||
):
|
||
"""Yield lines from a `docker compose` subcommand as they are produced."""
|
||
cmd = _compose_base_cmd(stack_id, override) + args
|
||
proc = await asyncio.create_subprocess_exec(
|
||
*cmd,
|
||
stdout=asyncio.subprocess.PIPE,
|
||
stderr=asyncio.subprocess.STDOUT,
|
||
)
|
||
assert proc.stdout is not None
|
||
async for raw in proc.stdout:
|
||
yield raw.decode("utf-8", "replace").rstrip("\n")
|
||
await proc.wait()
|
||
|
||
|
||
# Convenience lifecycle wrappers ------------------------------------------------
|
||
|
||
|
||
async def up(stack_id: str, override: Optional[str] = None) -> dict:
|
||
mark_busy(stack_id)
|
||
try:
|
||
return await run_compose(stack_id, ["up", "-d", "--remove-orphans"], override)
|
||
finally:
|
||
clear_busy(stack_id)
|
||
|
||
|
||
async def down(stack_id: str, override: Optional[str] = None) -> dict:
|
||
return await run_compose(stack_id, ["down"], override)
|
||
|
||
|
||
async def start(stack_id: str, override: Optional[str] = None) -> dict:
|
||
return await run_compose(stack_id, ["start"], override)
|
||
|
||
|
||
async def stop(stack_id: str, override: Optional[str] = None) -> dict:
|
||
return await run_compose(stack_id, ["stop"], override)
|
||
|
||
|
||
async def restart(stack_id: str, override: Optional[str] = None) -> dict:
|
||
return await run_compose(stack_id, ["restart"], override)
|
||
|
||
|
||
async def pull(stack_id: str, override: Optional[str] = None) -> dict:
|
||
mark_busy(stack_id)
|
||
try:
|
||
return await run_compose(stack_id, ["pull"], override)
|
||
finally:
|
||
clear_busy(stack_id)
|
||
|
||
|
||
async def update(stack_id: str, override: Optional[str] = None) -> dict:
|
||
"""Pull then up -d."""
|
||
mark_busy(stack_id)
|
||
try:
|
||
pull_res = await run_compose(stack_id, ["pull"], override)
|
||
up_res = await run_compose(stack_id, ["up", "-d", "--remove-orphans"], override)
|
||
return {
|
||
"returncode": up_res["returncode"],
|
||
"stdout": pull_res["stdout"] + "\n" + up_res["stdout"],
|
||
"stderr": pull_res["stderr"] + "\n" + up_res["stderr"],
|
||
"command": "pull + up -d",
|
||
}
|
||
finally:
|
||
clear_busy(stack_id)
|
||
|
||
|
||
async def logs(
|
||
stack_id: str,
|
||
service: Optional[str] = None,
|
||
tail: int = 200,
|
||
override: Optional[str] = None,
|
||
) -> dict:
|
||
args = ["logs", "--no-color", "--tail", str(tail), "--timestamps"]
|
||
if service:
|
||
args.append(service)
|
||
return await run_compose(stack_id, args, override, timeout=60.0)
|
||
|
||
|
||
def now() -> datetime:
|
||
return datetime.now(timezone.utc)
|