dozzle_n8n_logging/log-proxy/app.py

"""log-proxy: read-only Docker logs API for n8n.

Only endpoint surface:
  GET /healthz          liveness
  GET /services         list allow-listed Coolify service groups + their containers
  GET /logs/<group>     pull recent log lines from every container in the group

No write endpoints. No shell-out. Docker socket is RO-mounted at /var/run/docker.sock.
"""

from __future__ import annotations

import os
import re
import time
from datetime import datetime
from typing import Iterable

import docker
import yaml
from fastapi import FastAPI, HTTPException, Query
from fastapi.responses import JSONResponse

GROUPS_PATH = os.getenv("GROUPS_PATH", "/config/groups.yml")
DOCKER_SOCK = os.getenv("DOCKER_SOCK", "unix:///var/run/docker.sock")
COOLIFY_UUID_ENV = "COOLIFY_RESOURCE_UUID"

app = FastAPI(title="log-proxy", version="0.1.0")
docker_client = docker.DockerClient(base_url=DOCKER_SOCK, timeout=30)

# ISO timestamp prefix Docker emits when timestamps=True
TS_NANO_RE = re.compile(r"^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z) (.*)$")


def load_groups() -> dict[str, str]:
    """Read the UUID -> friendly-name allow-list. Empty if file missing."""
    try:
        with open(GROUPS_PATH) as fh:
            data = yaml.safe_load(fh) or {}
    except FileNotFoundError:
        return {}
    return {k: v for k, v in data.items() if isinstance(k, str) and isinstance(v, str)}


def container_uuid(container) -> str | None:
    env_list = container.attrs.get("Config", {}).get("Env") or []
    for item in env_list:
        if item.startswith(f"{COOLIFY_UUID_ENV}="):
            return item.split("=", 1)[1]
    return None


def resolve_group(name_or_uuid: str, allowed: dict[str, str]) -> str | None:
    """Accept either the UUID or the friendly name. Return UUID, or None if unknown."""
    if name_or_uuid in allowed:
        return name_or_uuid
    for uuid, friendly in allowed.items():
        if friendly == name_or_uuid:
            return uuid
    return None


def monitored_containers(allowed: dict[str, str]) -> Iterable[tuple[object, str]]:
    """Yield (container, uuid) for every running container whose UUID is allow-listed."""
    for c in docker_client.containers.list(all=False):
        uuid = container_uuid(c)
        if uuid and uuid in allowed:
            yield c, uuid


def parse_ts(prefix: str) -> float | None:
    """Docker emits nanosecond precision; Python only takes microseconds. Truncate."""
    iso = prefix
    if iso.endswith("Z"):
        iso = iso[:-1] + "+00:00"
    iso = re.sub(r"(\.\d{6})\d+", r"\1", iso)
    try:
        return datetime.fromisoformat(iso).timestamp()
    except ValueError:
        return None


@app.get("/healthz")
def healthz():
    try:
        docker_client.ping()
    except Exception as exc:
        raise HTTPException(status_code=503, detail=f"docker unreachable: {exc}") from exc
    return {"ok": True}


@app.get("/services")
def services():
    allowed = load_groups()
    by_uuid: dict[str, dict] = {}
    for c, uuid in monitored_containers(allowed):
        entry = by_uuid.setdefault(uuid, {"group": uuid, "name": allowed[uuid], "containers": []})
        entry["containers"].append(c.name)
    return JSONResponse([by_uuid[u] for u in sorted(by_uuid)])


@app.get("/logs/{group}")
def logs(
    group: str,
    since: int | None = Query(None, description="Unix seconds; default now-60"),
    until: int | None = Query(None, description="Unix seconds; default now"),
    limit: int = Query(2000, ge=1, le=10000),
):
    allowed = load_groups()
    target_uuid = resolve_group(group, allowed)
    if target_uuid is None:
        raise HTTPException(status_code=404, detail=f"Unknown group: {group}")

    now = int(time.time())
    since_ts = since if since is not None else now - 60
    until_ts = until if until is not None else now

    out: list[dict] = []
    for c in docker_client.containers.list(all=False):
        if container_uuid(c) != target_uuid:
            continue
        try:
            raw = c.logs(
                stdout=True,
                stderr=True,
                since=since_ts,
                until=until_ts,
                timestamps=True,
                tail=limit,
            )
        except Exception:
            continue
        if not raw:
            continue
        for raw_line in raw.decode("utf-8", errors="replace").splitlines():
            if not raw_line.strip():
                continue
            match = TS_NANO_RE.match(raw_line)
            if match:
                ts_val = parse_ts(match.group(1)) or float(since_ts)
                msg = match.group(2)
            else:
                ts_val = float(since_ts)
                msg = raw_line
            out.append({"container": c.name, "ts": ts_val, "line": msg})

    out.sort(key=lambda m: m["ts"])
    return JSONResponse(out[:limit])