feat(host-agent): collect container logs (incremental push, byte-capped) [M79 step 1]

Agent-side of Docker container-log collection (milestone 79). Each interval the
agent fetches new log lines per running container over the local Docker socket,
demuxes the multiplexed stream, and folds sample["docker_logs"] into the same
push as metrics — no inbound channel needed.

- since-cursor per container kept in rate-state; a container's first interval
  seeds from a short tail, then fetches incrementally via ?since=<cursor>
- _docker_request_raw + _demux_docker_logs + RFC3339Nano ts parsing (stdlib,
  TTY/unframed fallback); one row per (stream, ts, line)
- per-batch byte cap with a truncation marker; logs are dropped (never buffered)
  across a backoff so an outage keeps metrics but not stale logs
- docker_logs_enabled (default on) + docker_log_exclude per-host config keys
- AGENT_VERSION 1.6.0 -> 1.7.0

Server ignores the new sample key until the ingest step lands, so this commit is
CI-safe on its own.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CAGR73DUowdVFVvYzLXC5C

plugins/host_agent/agent.py

@@ -20,7 +20,7 @@ from collections import deque
 from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime, timezone
 
-AGENT_VERSION = "1.6.0"
+AGENT_VERSION = "1.7.0"
 
 # Default path to the local Docker Engine socket. Overridable via the
 # `docker_socket` config key; collection is silently skipped if it's absent or
@@ -34,7 +34,10 @@ class ConfigError(Exception):
 
 REQUIRED_KEYS = ("url", "token")
 INT_KEYS = ("interval_seconds",)
-LIST_KEYS = ("mounts",)
+LIST_KEYS = ("mounts", "docker_log_exclude")
+# Truthy strings for bool keys; anything else (incl. empty) is False.
+BOOL_KEYS = ("docker_logs_enabled",)
+_BOOL_TRUE = ("1", "true", "yes", "on")
 
 
 def read_config(path: str) -> dict:
@@ -58,6 +61,8 @@ def read_config(path: str) -> dict:
                         raise ConfigError(f"{path}:{lineno}: {key} must be int")
                 elif key in LIST_KEYS:
                     cfg[key] = [v.strip() for v in value.split(",") if v.strip()]
+                elif key in BOOL_KEYS:
+                    cfg[key] = value.lower() in _BOOL_TRUE
                 else:
                     cfg[key] = value
     except FileNotFoundError:
@@ -69,6 +74,10 @@ def read_config(path: str) -> dict:
 
     cfg.setdefault("interval_seconds", 30)
     cfg.setdefault("docker_socket", DEFAULT_DOCKER_SOCKET)
+    # Container-log collection is on by default (operator preference); a host can
+    # opt out with `docker_logs_enabled = false` or thin it with docker_log_exclude.
+    cfg.setdefault("docker_logs_enabled", True)
+    cfg.setdefault("docker_log_exclude", [])
     return cfg
 
 
@@ -403,6 +412,15 @@ def _rates(cur: dict, prev: dict, dt: float) -> dict:
 
 DOCKER_API_TIMEOUT = 5.0
 
+# Container-log collection (m79). Logs ride in the same push as metrics, so one
+# interval's batch is capped to keep a chatty container from bloating a POST (and
+# the backoff ring): once the cap is hit a marker line is emitted and collection
+# stops — the deferred lines come on the next interval. On a container's first
+# interval we seed from a short tail, then switch to an incremental since-cursor
+# kept per container in the agent's rate-state.
+DOCKER_LOG_BATCH_MAX_BYTES = 262144   # 256 KiB of log text per push, all containers
+DOCKER_LOG_FIRST_TAIL = 50            # lines seeded on a container's first interval
+
 
 def _dechunk(body: bytes) -> bytes:
     """Decode an HTTP/1.1 chunked-transfer body into the raw payload."""
@@ -650,6 +668,198 @@ def collect_docker(socket_path: str) -> list:
         return list(ex.map(lambda c: _collect_one_container(socket_path, c), containers))
 
 
+# ─── container logs (m79) ─────────────────────────────────────────────────────
+#
+# The logs endpoint returns a raw multiplexed byte stream, not JSON, so it needs
+# a sibling of _docker_request that hands back bytes. Non-TTY containers frame
+# stdout/stderr with an 8-byte header (stream byte + 4-byte big-endian length);
+# TTY containers emit unframed bytes. We request timestamps=1, so every line is
+# prefixed with an RFC3339Nano timestamp we parse for the since-cursor.
+
+_LOG_STREAM_NAMES = {0: "stdout", 1: "stdout", 2: "stderr"}
+
+
+def _docker_request_raw(socket_path: str, path: str,
+                        timeout: float = DOCKER_API_TIMEOUT) -> bytes:
+    """GET `path` from the Docker API over the Unix socket; return the raw body.
+
+    Sibling of `_docker_request` for non-JSON endpoints (container logs). Reuses
+    the connect / send / de-chunk scaffolding; raises OSError on any transport
+    problem or non-2xx status so callers can silent-skip. A 200 with an empty
+    body (no new lines since the cursor) returns b"".
+    """
+    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    sock.settimeout(timeout)
+    try:
+        sock.connect(socket_path)
+        req = (
+            "GET " + path + " HTTP/1.1\r\n"
+            "Host: docker\r\n"
+            "Accept: application/octet-stream\r\n"
+            "Connection: close\r\n"
+            "\r\n"
+        )
+        sock.sendall(req.encode("ascii"))
+        chunks = []
+        while True:
+            buf = sock.recv(65536)
+            if not buf:
+                break
+            chunks.append(buf)
+    finally:
+        sock.close()
+
+    raw = b"".join(chunks)
+    head, _, body = raw.partition(b"\r\n\r\n")
+    header_text = head.decode("latin-1")
+    status_line = header_text.split("\r\n", 1)[0]
+    parts = status_line.split(None, 2)
+    status = int(parts[1]) if len(parts) >= 2 and parts[1].isdigit() else 0
+    if not (200 <= status < 300):
+        raise OSError(f"docker API {path} returned {status}")
+    if "transfer-encoding: chunked" in header_text.lower():
+        body = _dechunk(body)
+    return body
+
+
+def _demux_docker_logs(raw: bytes) -> list:
+    """Split a Docker logs stream into [(stream_name, payload_bytes), …].
+
+    Non-TTY containers multiplex with an 8-byte frame header (stream byte in
+    {0,1,2} + 4-byte big-endian length). The stream is treated as framed only if
+    the whole buffer parses as clean back-to-back frames; anything else (a TTY
+    container's raw stream, or a malformed header) falls back to a single stdout
+    blob so no bytes are lost.
+    """
+    if not raw:
+        return []
+    frames = []
+    i, n = 0, len(raw)
+    while i + 8 <= n:
+        stream = raw[i]
+        if stream > 2 or raw[i + 1] or raw[i + 2] or raw[i + 3]:
+            break  # not a valid frame header → not framed
+        size = int.from_bytes(raw[i + 4:i + 8], "big")
+        if i + 8 + size > n:
+            break  # frame overruns the buffer → not framed
+        frames.append((_LOG_STREAM_NAMES.get(stream, "stdout"),
+                       raw[i + 8:i + 8 + size]))
+        i += 8 + size
+    if frames and i == n:
+        return frames
+    return [("stdout", raw)]
+
+
+def _parse_log_ts(token: str):
+    """Parse a Docker RFC3339Nano timestamp token into a datetime, or None.
+
+    Normalises `Z` → `+00:00` and trims Docker's nanosecond precision to the
+    microseconds datetime.fromisoformat accepts (stdlib, Python 3.8+ safe).
+    """
+    t = token.strip()
+    if not t:
+        return None
+    if t.endswith("Z"):
+        t = t[:-1] + "+00:00"
+    m = re.match(r"^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})(\.\d+)?(.*)$", t)
+    if m:
+        frac = m.group(2) or ""
+        if len(frac) > 7:               # "." + up to 6 fractional digits
+            frac = frac[:7]
+        t = m.group(1) + frac + m.group(3)
+    try:
+        return datetime.fromisoformat(t)
+    except ValueError:
+        return None
+
+
+def _parse_container_logs(raw: bytes) -> list:
+    """Demux + line-split a logs stream → [(dt|None, stream, line), …] in order.
+
+    Each line is timestamp-prefixed (we request timestamps=1); an unparseable
+    prefix yields dt=None with the full raw line kept.
+    """
+    out = []
+    for stream, payload in _demux_docker_logs(raw):
+        text = payload.decode("utf-8", "replace")
+        for raw_line in text.split("\n"):
+            if not raw_line:
+                continue
+            token, _, msg = raw_line.partition(" ")
+            dt = _parse_log_ts(token)
+            if dt is not None:
+                out.append((dt, stream, msg))
+            else:
+                out.append((None, stream, raw_line))
+    return out
+
+
+def collect_docker_logs(socket_path: str, containers: list, state: dict,
+                        exclude=None,
+                        max_bytes: int = DOCKER_LOG_BATCH_MAX_BYTES) -> list:
+    """New container-log lines since the last interval, for running containers.
+
+    Keeps a per-container since-cursor in `state["docker_log_cursors"]`
+    (name → unix ts) so each interval fetches only what's new; a container's
+    first interval seeds from a short tail. The whole batch is capped at
+    `max_bytes` of log text — once exceeded a truncation marker is appended and
+    collection stops (logs must never balloon a push). `exclude` container names
+    are skipped entirely (a per-host bandwidth opt-out). Best-effort: a
+    per-container transport error just contributes nothing this interval.
+    Returns [{"container", "stream", "ts", "line"}, …].
+    """
+    exclude_set = set(exclude or ())
+    cursors = state.setdefault("docker_log_cursors", {})
+    running = {c.get("name") for c in (containers or [])
+               if c.get("name") and c.get("status") == "running"}
+    # Forget cursors for containers no longer running so state can't grow without
+    # bound over the agent's lifetime.
+    for gone in [k for k in cursors if k not in running]:
+        del cursors[gone]
+
+    out: list = []
+    total = 0
+    truncated = False
+    for c in (containers or []):
+        name = c.get("name")
+        if not name or name not in running or name in exclude_set:
+            continue
+        prev = cursors.get(name)
+        base = f"/containers/{name}/logs?stdout=1&stderr=1&timestamps=1"
+        # since accepts a fractional Unix ts; we still dedup the boundary line.
+        path = (f"{base}&tail={DOCKER_LOG_FIRST_TAIL}" if prev is None
+                else f"{base}&since={prev:.9f}")
+        try:
+            raw = _docker_request_raw(socket_path, path)
+        except (OSError, ValueError):
+            continue
+        newest = prev
+        for dt, stream, line in _parse_container_logs(raw):
+            ts_unix = dt.timestamp() if dt is not None else None
+            if prev is not None and ts_unix is not None and ts_unix <= prev:
+                continue  # boundary line already sent last interval
+            if total >= max_bytes:
+                truncated = True
+                break
+            out.append({"container": name, "stream": stream,
+                        "ts": dt.isoformat() if dt is not None else None,
+                        "line": line})
+            total += len(line)
+            if ts_unix is not None and (newest is None or ts_unix > newest):
+                newest = ts_unix
+        if newest is not None:
+            cursors[name] = newest
+        if truncated:
+            break
+
+    if truncated:
+        out.append({"container": "_steward", "stream": "stderr",
+                    "ts": datetime.now(timezone.utc).isoformat(),
+                    "line": (f"[steward] log batch truncated at {max_bytes} bytes; "
+                             "remaining lines deferred to the next interval")})
+    return out
+
+
 # ─── swarm (manager-only) ────────────────────────────────────────────────────
 
 
@@ -852,13 +1062,17 @@ class RingBuffer:
 
 
 def build_sample(mounts: list[str], state: dict,
-                 docker_socket: str = DEFAULT_DOCKER_SOCKET) -> dict:
+                 docker_socket: str = DEFAULT_DOCKER_SOCKET, *,
+                 docker_logs_enabled: bool = True,
+                 docker_log_exclude=None) -> dict:
     """Collect one full sample. Partial samples allowed if a collector fails.
 
     `state` carries the previous network/disk counters + monotonic timestamp so
-    throughput rates can be derived from deltas; it is mutated in place.
-    `docker_socket` is probed best-effort — the `docker` key is omitted entirely
-    when no containers are found, so non-Docker hosts add nothing to the payload.
+    throughput rates can be derived from deltas (and the per-container log
+    since-cursors); it is mutated in place. `docker_socket` is probed
+    best-effort — the `docker` key is omitted entirely when no containers are
+    found, so non-Docker hosts add nothing to the payload. `docker_logs_enabled`
+    / `docker_log_exclude` gate incremental container-log collection.
     """
     sample: dict = {"ts": datetime.now(timezone.utc).isoformat()}
     try:
@@ -940,6 +1154,18 @@ def build_sample(mounts: list[str], state: dict,
         if disk is not None:
             sample["docker_disk"] = disk
 
+    # Container logs (m79): incremental per-container tail folded into the same
+    # push as metrics. Skipped when there are no containers or logging is off;
+    # the key is omitted when nothing new arrived this interval.
+    if docker and docker_logs_enabled:
+        try:
+            logs = collect_docker_logs(docker_socket, docker, state,
+                                       exclude=docker_log_exclude)
+        except Exception:
+            logs = []
+        if logs:
+            sample["docker_logs"] = logs
+
     return sample
 
 
@@ -958,6 +1184,18 @@ def build_payload(samples: list[dict], hostname: str, metadata: dict) -> dict:
 BACKOFF_CAP = 300
 
 
+def _drop_logs(sample: dict) -> dict:
+    """Strip container logs from a sample before it's buffered for retry.
+
+    Logs are the one payload we never carry across a backoff — they'd bloat the
+    ring buffer and go stale — while metrics are kept so an outage doesn't lose
+    them. The since-cursor already advanced when the logs were collected, so the
+    dropped lines are simply not re-sent (accepted loss during an outage).
+    """
+    sample.pop("docker_logs", None)
+    return sample
+
+
 def next_backoff(current: int) -> int:
     if current <= 0:
         return 30
@@ -1022,6 +1260,8 @@ def main_loop(conf_path: str) -> int:
     hostname = cfg.get("hostname") or socket.gethostname()
     mounts = cfg.get("mounts") or default_mounts()
     docker_socket = cfg.get("docker_socket") or DEFAULT_DOCKER_SOCKET
+    docker_logs_enabled = cfg.get("docker_logs_enabled", True)
+    docker_log_exclude = cfg.get("docker_log_exclude") or []
     buffer = RingBuffer(maxlen=20)
     backoff = 0
     # Carries previous net/disk counters + monotonic ts for rate computation.
@@ -1036,13 +1276,17 @@ def main_loop(conf_path: str) -> int:
                 cfg = read_config(conf_path)
                 mounts = cfg.get("mounts") or default_mounts()
                 docker_socket = cfg.get("docker_socket") or DEFAULT_DOCKER_SOCKET
+                docker_logs_enabled = cfg.get("docker_logs_enabled", True)
+                docker_log_exclude = cfg.get("docker_log_exclude") or []
                 metadata = collect_metadata()  # refresh host_ip/distro on reload
                 _log("INFO", "config reloaded")
             except ConfigError as e:
                 _log("ERROR", f"reload failed: {e}")
             _reload_requested = False
 
-        sample = build_sample(mounts, rate_state, docker_socket)
+        sample = build_sample(mounts, rate_state, docker_socket,
+                              docker_logs_enabled=docker_logs_enabled,
+                              docker_log_exclude=docker_log_exclude)
         buffered = buffer.drain()
         payload = build_payload(
             samples=buffered + [sample],
@@ -1061,12 +1305,12 @@ def main_loop(conf_path: str) -> int:
                 _log("ERROR", "server rejected payload (400) — dropping sample")
             elif status == 401:
                 _log("ERROR", "token rejected (401) — check config + UI")
-                buffer.push(sample)
+                buffer.push(_drop_logs(sample))
             else:
                 _log("WARN", f"POST failed (status={status}); buffering")
                 for s in buffered:
-                    buffer.push(s)
-                buffer.push(sample)
+                    buffer.push(_drop_logs(s))
+                buffer.push(_drop_logs(sample))
             backoff = next_backoff(backoff)
             sleep_for = backoff
 

plugins/host_agent/templates/install.sh.j2

@@ -45,6 +45,10 @@ cat > "$CONF_FILE" <<EOF
 url = $STEWARD_URL
 token = $AGENT_TOKEN
 interval_seconds = 30
+# Container logs are collected by default. To opt this host out entirely:
+#   docker_logs_enabled = false
+# To skip specific noisy containers (comma-separated names):
+#   docker_log_exclude = watchtower, some-chatty-service
 EOF
 chown "root:$AGENT_USER" "$CONF_FILE"
 chmod 0640 "$CONF_FILE"