Roundtable-plugins/docker/scraper.py

# plugins/docker/scraper.py
"""Docker API client via Unix socket."""
from __future__ import annotations
import asyncio
import logging
from datetime import datetime, timezone

import httpx

logger = logging.getLogger(__name__)


def _short_id(container_id: str) -> str:
    return container_id[:12] if container_id else ""


def _calc_cpu_pct(stats: dict) -> float:
    """Calculate CPU % from a Docker stats snapshot (one-shot)."""
    try:
        cpu = stats.get("cpu_stats", {})
        precpu = stats.get("precpu_stats", {})
        cpu_delta = (
            cpu["cpu_usage"]["total_usage"] - precpu["cpu_usage"]["total_usage"]
        )
        sys_delta = cpu.get("system_cpu_usage", 0) - precpu.get("system_cpu_usage", 0)
        num_cpus = cpu.get("online_cpus") or len(
            cpu["cpu_usage"].get("percpu_usage") or [None]
        )
        if sys_delta <= 0 or cpu_delta < 0:
            return 0.0
        return round((cpu_delta / sys_delta) * num_cpus * 100.0, 2)
    except (KeyError, TypeError, ZeroDivisionError):
        return 0.0


def _calc_mem(stats: dict) -> tuple[int, int, float]:
    """Return (usage_bytes, limit_bytes, mem_pct) from Docker stats."""
    try:
        mem = stats.get("memory_stats", {})
        usage = mem.get("usage", 0)
        limit = mem.get("limit", 0)
        # Subtract page cache for working set (cgroup v1: "cache", v2: "inactive_file")
        cache = mem.get("stats", {}).get("cache", 0) or \
                mem.get("stats", {}).get("inactive_file", 0)
        actual = max(0, usage - cache)
        pct = round(actual / limit * 100.0, 2) if limit > 0 else 0.0
        return actual, limit, pct
    except (KeyError, TypeError):
        return 0, 0, 0.0


def _parse_ports(ports: list) -> list[dict]:
    """Normalise Docker port bindings to a compact list."""
    result = []
    for p in (ports or []):
        if p.get("PublicPort"):
            result.append({
                "host_port": p["PublicPort"],
                "container_port": p["PrivatePort"],
                "protocol": p.get("Type", "tcp"),
            })
    return result


async def scrape_docker(socket_path: str, include_stopped: bool) -> list[dict]:
    """
    Fetch all containers + resource stats from the Docker daemon.
    Returns a list of normalised dicts ready for the scheduler to persist.
    Raises ConnectionError if the socket is unreachable.
    """
    transport = httpx.AsyncHTTPTransport(uds=socket_path)
    try:
        async with httpx.AsyncClient(
            transport=transport,
            base_url="http://docker",
            timeout=10.0,
        ) as client:
            resp = await client.get(
                "/containers/json",
                params={"all": "true" if include_stopped else "false"},
            )
            resp.raise_for_status()
            containers = resp.json()

            async def _get_stats(c: dict) -> tuple[dict, dict | None]:
                if c.get("State") != "running":
                    return c, None
                try:
                    r = await client.get(
                        f"/containers/{c['Id']}/stats",
                        params={"stream": "false", "one-shot": "true"},
                        timeout=5.0,
                    )
                    return c, r.json()
                except Exception as exc:
                    logger.debug(
                        "Stats unavailable for %s: %s", _short_id(c.get("Id", "")), exc
                    )
                    return c, None

            pairs = await asyncio.gather(*[_get_stats(c) for c in containers])

    except httpx.ConnectError as exc:
        raise ConnectionError(
            f"Cannot connect to Docker socket at {socket_path}: {exc}"
        ) from exc

    results = []
    now = datetime.now(timezone.utc)
    for container, stats in pairs:
        names = container.get("Names") or []
        name = names[0].lstrip("/") if names else _short_id(container.get("Id", ""))

        cpu_pct = mem_usage = mem_limit = mem_pct = None
        if stats is not None:
            cpu_pct = _calc_cpu_pct(stats)
            mem_usage, mem_limit, mem_pct = _calc_mem(stats)

        # Created is a Unix epoch int from /containers/json
        created_ts = container.get("Created")
        started_at = (
            datetime.fromtimestamp(created_ts, tz=timezone.utc)
            if isinstance(created_ts, (int, float)) else None
        )

        results.append({
            "name": name,
            "container_id": _short_id(container.get("Id", "")),
            "image": container.get("Image", ""),
            "status": container.get("State", "unknown"),
            "cpu_pct": cpu_pct,
            "mem_usage_bytes": mem_usage,
            "mem_limit_bytes": mem_limit,
            "mem_pct": mem_pct,
            "restart_count": 0,  # would require /inspect; left as 0 for now
            "ports": _parse_ports(container.get("Ports", [])),
            "started_at": started_at,
        })

    return results