tgit/app/runtime/docker.py

"""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