"""Read-only Docker Engine API client. The bot talks to docker-socket-proxy (GET only), never to the socket itself. Only containers carrying the managed label are visible: the host also runs other projects (Vobot, meditron), and an id outside the list is refused before any request goes out. """ import json import re import struct from dataclasses import dataclass from datetime import UTC, datetime from enum import StrEnum from types import TracebackType from typing import Any, Self import httpx from app.config import DockerSettings SHA_LABEL = "tgit.sha" DEPLOYED_LABEL = "tgit.deployed_at" HEALTH_URL_LABEL = "tgit.health_url" SERVICE_LABEL = "com.docker.compose.service" _FRACTION = re.compile(r"(\.\d{6})\d+") # Docker gives nanoseconds, datetime takes microseconds class DockerError(Exception): pass class Light(StrEnum): """S-1: 🟢 healthy / 🟡 starting / 🔴 unhealthy or crashed / ⚪ stopped.""" GREEN = "🟢" YELLOW = "🟡" RED = "🔴" WHITE = "⚪" def parse_time(value: str | None) -> datetime | None: if not value or value.startswith("0001-"): # Docker's zero time: never started / no deploy label return None try: return datetime.fromisoformat(_FRACTION.sub(r"\1", value).replace("Z", "+00:00")).astimezone(UTC) except ValueError: return None def light(state: str, health: str | None, exit_code: int = 0) -> Light: if state == "running": if health == "unhealthy": return Light.RED if health == "starting": return Light.YELLOW return Light.GREEN if state in ("restarting", "dead") or (state == "exited" and exit_code != 0): return Light.RED if state == "created": return Light.YELLOW return Light.WHITE @dataclass(frozen=True) class Container: """A list entry (GET /containers/json): enough for the overview, no extra request per container.""" id: str name: str service: str image: str state: str # created | running | paused | restarting | removing | exited | dead health: str | None # starting | healthy | unhealthy | None — no healthcheck exit_code: int labels: dict[str, str] @property def short_id(self) -> str: return self.id[:12] @property def light(self) -> Light: return light(self.state, self.health, self.exit_code) _HEALTH_IN_STATUS = re.compile(r"\((healthy|unhealthy|health: starting)\)") _EXIT_IN_STATUS = re.compile(r"^Exited \((-?\d+)\)") def _health_from_status(status: str) -> str | None: """`Up 2 hours (healthy)` → healthy. The list endpoint has no structured health field.""" m = _HEALTH_IN_STATUS.search(status) if m is None: return None return "starting" if m.group(1) == "health: starting" else m.group(1) @dataclass(frozen=True) class Details: """GET /containers/{id}/json — the card (S-2) and most checks.""" id: str name: str service: str image: str state: str health: str | None exit_code: int restart_count: int started_at: datetime | None finished_at: datetime | None oom_killed: bool labels: dict[str, str] @property def light(self) -> Light: return light(self.state, self.health, self.exit_code) @property def sha(self) -> str | None: return self.labels.get(SHA_LABEL) or None @property def deployed_at(self) -> datetime | None: return parse_time(self.labels.get(DEPLOYED_LABEL)) @property def health_url(self) -> str | None: return self.labels.get(HEALTH_URL_LABEL) or None @dataclass(frozen=True) class Stats: cpu_pct: float # of one CPU × online CPUs, as `docker stats` shows it mem_bytes: int mem_limit: int # the cgroup limit; the host's RAM when the container has none @property def mem_pct(self) -> float: return 100.0 * self.mem_bytes / self.mem_limit if self.mem_limit else 0.0 def parse_stats(data: dict[str, Any]) -> Stats: cpu, pre = data.get("cpu_stats") or {}, data.get("precpu_stats") or {} cpu_delta = (cpu.get("cpu_usage") or {}).get("total_usage", 0) - (pre.get("cpu_usage") or {}).get("total_usage", 0) sys_delta = cpu.get("system_cpu_usage", 0) - pre.get("system_cpu_usage", 0) cpus = cpu.get("online_cpus") or len((cpu.get("cpu_usage") or {}).get("percpu_usage") or []) or 1 cpu_pct = 100.0 * cpu_delta / sys_delta * cpus if cpu_delta > 0 and sys_delta > 0 else 0.0 mem = data.get("memory_stats") or {} extra = mem.get("stats") or {} # Page cache isn't pressure: cgroup v2 reports inactive_file, v1 total_inactive_file (docker CLI does the same) cache = extra.get("inactive_file", extra.get("total_inactive_file", 0)) usage = max(int(mem.get("usage", 0)) - int(cache), 0) return Stats(cpu_pct=round(cpu_pct, 1), mem_bytes=usage, mem_limit=int(mem.get("limit", 0))) def demux_logs(raw: bytes) -> str: """Split Docker's multiplexed stream (8-byte header per frame) — containers without a TTY send it this way.""" if raw[:1] not in (b"\x00", b"\x01", b"\x02") or len(raw) < 8 or raw[1:4] != b"\x00\x00\x00": return raw.decode(errors="replace") # TTY container: plain bytes out: list[bytes] = [] pos = 0 while pos + 8 <= len(raw): (size,) = struct.unpack(">I", raw[pos + 4 : pos + 8]) out.append(raw[pos + 8 : pos + 8 + size]) pos += 8 + size return b"".join(out).decode(errors="replace") class DockerClient: def __init__(self, settings: DockerSettings, transport: httpx.AsyncBaseTransport | None = None) -> None: self._s = settings self._http = httpx.AsyncClient(base_url=settings.url, timeout=settings.timeout, transport=transport) self._known: dict[str, Container] = {} # the last listing: the only ids card/logs/stats may touch async def __aenter__(self) -> Self: return self async def __aexit__( self, exc_type: type[BaseException] | None, exc: BaseException | None, tb: TracebackType | None ) -> None: await self.aclose() async def aclose(self) -> None: await self._http.aclose() async def containers(self) -> list[Container]: """All managed containers, stopped ones too, sorted by name.""" filters = json.dumps({"label": [self._s.label]}) data = await self._get_json("/containers/json", params={"all": "1", "filters": filters}) result = [] for c in data: labels = c.get("Labels") or {} names = c.get("Names") or [c["Id"][:12]] result.append( Container( id=c["Id"], name=names[0].lstrip("/"), service=labels.get(SERVICE_LABEL, ""), image=c.get("Image", ""), state=c.get("State", ""), health=_health_from_status(c.get("Status", "")), exit_code=int(m.group(1)) if (m := _EXIT_IN_STATUS.match(c.get("Status", ""))) else 0, labels=labels, ) ) self._known = {c.id: c for c in result} return sorted(result, key=lambda c: c.name) async def get(self, short_id: str) -> Container: """A managed container by its 12-char id prefix (callback data carries only that).""" for attempt in range(2): if attempt or not self._known: await self.containers() # first use, or recreated since the last listing for full, container in self._known.items(): if short_id and full.startswith(short_id): return container raise DockerError(f"container {short_id} is not managed by tgit") async def resolve(self, short_id: str) -> str: return (await self.get(short_id)).id async def inspect(self, container_id: str) -> Details: self._guard(container_id) d = await self._get_json(f"/containers/{container_id}/json") state = d.get("State") or {} labels = (d.get("Config") or {}).get("Labels") or {} return Details( id=d["Id"], name=d.get("Name", "").lstrip("/"), service=labels.get(SERVICE_LABEL, ""), image=(d.get("Config") or {}).get("Image", ""), state=state.get("Status", ""), health=(state.get("Health") or {}).get("Status"), exit_code=int(state.get("ExitCode", 0)), restart_count=int(d.get("RestartCount", 0)), started_at=parse_time(state.get("StartedAt")), finished_at=parse_time(state.get("FinishedAt")), oom_killed=bool(state.get("OOMKilled", False)), labels=labels, ) async def stats(self, container_id: str) -> Stats: """One sample with a previous one to diff against (stream=false waits ~1 s for it).""" self._guard(container_id) return parse_stats(await self._get_json(f"/containers/{container_id}/stats", params={"stream": "false"})) async def logs(self, container_id: str, since: datetime | None = None, tail: int | None = None) -> str: """stdout+stderr in order; at most max_log_lines lines and max_log_bytes — the newest ones are kept.""" self._guard(container_id) lines = min(tail, self._s.max_log_lines) if tail is not None else self._s.max_log_lines params = {"stdout": "1", "stderr": "1", "tail": str(lines)} if since is not None: params["since"] = str(int(since.timestamp())) response = await self._request(f"/containers/{container_id}/logs", params) text = demux_logs(response.content) if len(text) > self._s.max_log_bytes: cut = text[-self._s.max_log_bytes :] text = cut[cut.find("\n") + 1 :] # drop the torn first line return text async def ping(self) -> bool: try: return (await self._request("/_ping", {})).text == "OK" except DockerError: return False def _guard(self, container_id: str) -> None: if container_id not in self._known: raise DockerError(f"container {container_id[:12]} is not managed by tgit") async def _get_json(self, path: str, params: dict[str, str] | None = None) -> Any: response = await self._request(path, params or {}) try: return response.json() except ValueError as e: raise DockerError(f"{path}: not JSON") from e async def _request(self, path: str, params: dict[str, str]) -> httpx.Response: try: response = await self._http.get(path, params=params) except httpx.TransportError as e: # timeouts included raise DockerError(f"docker unreachable: {type(e).__name__}") from e if response.status_code >= 400: raise DockerError(f"{path}: HTTP {response.status_code} {response.text[:200]}") return response