fix(host_agent): every Docker read is bounded, and the agent reports its own footprint (#5764)

#5764 "The agent reads a container's whole log backlog into memory, on its
main loop, with no overall deadline". Docker03's 1.13.0 agent went silent
for a day at 2.1 GB and 46 h of CPU, then ignored SIGTERM.

- One shared reader, _docker_get: each answer has an overall deadline and a
  byte cap, not only a per-recv timeout. A trickling or endless body is cut
  off (15 s / 32 MiB for JSON, 5 s / 4 MiB for a log read).
- Container logs: since the cursor but only the newest 2,000 lines, all
  within 10 s. A note says when older lines were skipped, and a container
  whose answer is too big skips ahead to now instead of re-reading its
  backlog every interval.
- The agent reports its own RSS and CPU; Steward records agent_rss_bytes
  and agent_cpu_pct, so a runaway agent shows before it goes silent.

Agent 1.15.0.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

plugins/host_agent/__init__.py

@@ -65,10 +65,13 @@ def get_alert_metrics() -> dict[str, list[str]]:
     of the host's cores. inodes_used_pct is per mount (agent ≥ 1.13.0), and
     disk_used_pct_worst is the fullest mount by space or inodes.
     mounts_stalled counts the mounts that didn't answer the agent in time
-    (agent ≥ 1.14.0): "an NFS share stopped" is > 0."""
+    (agent ≥ 1.14.0): "an NFS share stopped" is > 0. agent_rss_bytes and
+    agent_cpu_pct are the agent's own memory and CPU (agent ≥ 1.15.0): a
+    healthy agent sits in tens of MB and a few %."""
     return {"host_agent": [
         "cpu_pct", "mem_used_pct", "mem_available_bytes", "swap_used_bytes",
         "disk_used_pct_worst", "disk_used_pct", "inodes_used_pct", "load_1m", "load_5m", "load_15m",
         "load_pct",
         "uptime_secs", "seconds_since_seen", "clock_off_secs", "mounts_stalled",
+        "agent_rss_bytes", "agent_cpu_pct",
     ]}

plugins/host_agent/agent.py

@@ -22,7 +22,7 @@ from collections import deque
 from concurrent.futures import ThreadPoolExecutor, wait as wait_futures
 from datetime import datetime, timezone
 
-AGENT_VERSION = "1.14.0"
+AGENT_VERSION = "1.15.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
@@ -510,6 +510,24 @@ DOCKER_LOG_BATCH_MAX_BYTES = 262144   # 256 KiB of log text per push, all contai
 DOCKER_LOG_FIRST_TAIL = 50            # lines seeded on a container's first interval
 DOCKER_LOG_FINAL_TAIL = 200           # lines read once from a container found already crashed
 DOCKER_LOG_FINAL_WINDOW_SECONDS = 600  # ... when it stopped this recently
+# A container that wrote more than this since its cursor is read from its last
+# DOCKER_LOG_MAX_LINES lines, and the older ones are skipped, not re-read every
+# interval (#5764). One log answer over DOCKER_LOG_RESPONSE_MAX_BYTES is dropped
+# and the container skipped ahead to now.
+DOCKER_LOG_MAX_LINES = 2000
+DOCKER_LOG_RESPONSE_MAX_BYTES = 4 * 1024 * 1024
+# All of one sample's log reads must finish inside this; the containers not
+# reached are read next interval.
+DOCKER_LOG_DEADLINE_SECONDS = 10.0
+# Any one Docker answer: past this size or this long, it is cut off rather than
+# waited on. A body that keeps trickling in never reaches EOF, and a per-read
+# timeout alone would wait on it forever (#5764).
+DOCKER_RESPONSE_MAX_BYTES = 32 * 1024 * 1024
+DOCKER_RESPONSE_DEADLINE_SECONDS = 15.0
+
+
+class DockerAnswerTooLarge(OSError):
+    """A Docker answer longer than the reader would hold."""
 
 
 def _dechunk(body: bytes) -> bytes:
@@ -530,30 +548,42 @@ def _dechunk(body: bytes) -> bytes:
     return bytes(out)
 
 
-def _docker_request(socket_path: str, path: str, timeout: float = DOCKER_API_TIMEOUT):
-    """GET `path` from the Docker Engine API over a Unix socket; return JSON.
+def _docker_get(socket_path: str, path: str, accept: str, timeout: float,
+                deadline: float, max_bytes: int) -> tuple:
+    """GET `path` from the Docker Engine API over a Unix socket.
 
-    Raises OSError on any socket/transport problem (absent socket, permission
-    denied, non-200) and ValueError on a non-JSON body — both are caught by the
-    caller and treated as "no docker here", so collection degrades silently.
+    Returns (status, body), the body de-chunked. Each read waits at most
+    `timeout`, the whole answer at most `deadline`, and an answer longer than
+    `max_bytes` raises DockerAnswerTooLarge without being held whole: a slow or
+    endless body is cut off, never waited on (#5764). Raises OSError on any
+    transport problem.
     """
+    end = time.monotonic() + deadline
     sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
-    sock.settimeout(timeout)
+    sock.settimeout(min(timeout, deadline))
     try:
         sock.connect(socket_path)
         req = (
             "GET " + path + " HTTP/1.1\r\n"
             "Host: docker\r\n"
-            "Accept: application/json\r\n"
+            "Accept: " + accept + "\r\n"
             "Connection: close\r\n"
             "\r\n"
         )
         sock.sendall(req.encode("ascii"))
         chunks = []
+        size = 0
         while True:
+            left = end - time.monotonic()
+            if left <= 0:
+                raise socket.timeout(f"docker API {path} still answering after {deadline:g} s")
+            sock.settimeout(min(timeout, left))
             buf = sock.recv(65536)
             if not buf:
                 break
+            size += len(buf)
+            if size > max_bytes:
+                raise DockerAnswerTooLarge(f"docker API {path} answered more than {max_bytes} bytes")
             chunks.append(buf)
     finally:
         sock.close()
@@ -564,10 +594,23 @@ def _docker_request(socket_path: str, path: str, timeout: float = DOCKER_API_TIM
     status_line = header_text.split("\r\n", 1)[0]
     parts = status_line.split(None, 2)  # "HTTP/1.1 200 OK"
     status = int(parts[1]) if len(parts) >= 2 and parts[1].isdigit() else 0
-    if status != 200:
-        raise OSError(f"docker API {path} returned {status}")
     if "transfer-encoding: chunked" in header_text.lower():
         body = _dechunk(body)
+    return status, body
+
+
+def _docker_request(socket_path: str, path: str, timeout: float = DOCKER_API_TIMEOUT):
+    """GET `path` from the Docker Engine API over a Unix socket; return JSON.
+
+    Raises OSError on any socket/transport problem (absent socket, permission
+    denied, non-200, an answer too slow or too large) and ValueError on a
+    non-JSON body — both are caught by the caller and treated as "no docker
+    here", so collection degrades silently.
+    """
+    status, body = _docker_get(socket_path, path, "application/json", timeout,
+                               DOCKER_RESPONSE_DEADLINE_SECONDS, DOCKER_RESPONSE_MAX_BYTES)
+    if status != 200:
+        raise OSError(f"docker API {path} returned {status}")
     return json.loads(body.decode("utf-8"))
 
 
@@ -817,42 +860,17 @@ 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"".
+    Sibling of `_docker_request` for non-JSON endpoints (container logs). The
+    whole answer must arrive within `timeout` and fit in
+    DOCKER_LOG_RESPONSE_MAX_BYTES (#5764); raises OSError on any transport
+    problem or non-2xx status so callers can silent-skip, and
+    DockerAnswerTooLarge past the size. 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
+    status, body = _docker_get(socket_path, path, "application/octet-stream", timeout,
+                               timeout, DOCKER_LOG_RESPONSE_MAX_BYTES)
     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
 
 
@@ -928,10 +946,18 @@ def _parse_container_logs(raw: bytes) -> list:
     return out
 
 
+def _log_note(line: str) -> dict:
+    """A line Steward's own log collection adds to the batch, shown under
+    the "_steward" container."""
+    return {"container": "_steward", "stream": "stderr",
+            "ts": datetime.now(timezone.utc).isoformat(), "line": "[steward] " + line}
+
+
 def collect_docker_logs(socket_path: str, containers: list, state: dict,
                         exclude=None,
                         max_bytes: int = DOCKER_LOG_BATCH_MAX_BYTES,
-                        now: float | None = None) -> list:
+                        now: float | None = None,
+                        deadline: float = DOCKER_LOG_DEADLINE_SECONDS) -> list:
     """New container-log lines since the last interval.
 
     Keeps a per-container since-cursor in `state["docker_log_cursors"]`
@@ -950,6 +976,13 @@ def collect_docker_logs(socket_path: str, containers: list, state: dict,
     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.
+
+    What is read is bounded too (#5764): at most DOCKER_LOG_MAX_LINES lines
+    per container (the newest; a note says older ones were skipped), at most
+    DOCKER_LOG_RESPONSE_MAX_BYTES per answer (past it the container skips
+    ahead to now), and all of it within `deadline` seconds. A container that
+    writes faster than it can be read costs one bounded read per interval,
+    never a backlog re-read every time.
     Returns [{"container", "stream", "ts", "line"}, …].
     """
     exclude_set = set(exclude or ())
@@ -966,21 +999,29 @@ def collect_docker_logs(socket_path: str, containers: list, state: dict,
     out: list = []
     total = 0
     truncated = False
+    end = time.monotonic() + deadline
+    # since + tail: of the lines since the cursor, the newest MAX_LINES.
+    since_tail = f"&tail={DOCKER_LOG_MAX_LINES}"
     for name, c in listed.items():
         if name in exclude_set:
             continue
+        left = end - time.monotonic()
+        if left <= 0:
+            out.append(_log_note(f"log reading stopped at its {deadline:g} s limit; "
+                                 "the other containers are read next interval"))
+            break
         prev = cursors.get(name)
         base = f"/containers/{name}/logs?stdout=1&stderr=1&timestamps=1"
         if c.get("status") == "running":
             # 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}")
+                    else f"{base}&since={prev:.9f}{since_tail}")
         else:
             stopped = c.get("finished_at")
             if not stopped or finals.get(name) == stopped:
                 continue
             if prev is not None:
-                path = f"{base}&since={prev:.9f}"
+                path = f"{base}&since={prev:.9f}{since_tail}"
             else:
                 at = _parse_log_ts(stopped)
                 if (c.get("exit_code") in (None, 0) or at is None
@@ -989,11 +1030,22 @@ def collect_docker_logs(socket_path: str, containers: list, state: dict,
                 path = f"{base}&tail={DOCKER_LOG_FINAL_TAIL}"
             finals[name] = stopped
         try:
-            raw = _docker_request_raw(socket_path, path)
+            raw = _docker_request_raw(socket_path, path, timeout=min(DOCKER_API_TIMEOUT, left))
+        except DockerAnswerTooLarge:
+            # More than one answer can hold: skip to now rather than re-asking
+            # for the same backlog next interval.
+            cursors[name] = now
+            out.append(_log_note(f"{name} wrote more than {DOCKER_LOG_RESPONSE_MAX_BYTES} "
+                                 "bytes of log since the last read; skipped ahead to now"))
+            continue
         except (OSError, ValueError):
             continue
+        parsed = _parse_container_logs(raw)
+        if prev is not None and len(parsed) >= DOCKER_LOG_MAX_LINES:
+            out.append(_log_note(f"{name} wrote more than {DOCKER_LOG_MAX_LINES} lines "
+                                 "since the last read; older ones skipped"))
         newest = prev
-        for dt, stream, line in _parse_container_logs(raw):
+        for dt, stream, line in parsed:
             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
@@ -1012,10 +1064,8 @@ def collect_docker_logs(socket_path: str, containers: list, state: dict,
             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")})
+        out.append(_log_note(f"log batch truncated at {max_bytes} bytes; "
+                             "remaining lines deferred to the next interval"))
     return out
 
 
@@ -1253,6 +1303,24 @@ class RingBuffer:
         return len(self._dq)
 
 
+def collect_self(state: dict) -> dict:
+    """The agent's own footprint: resident memory and CPU % of one core since
+    the last sample (#5764). A 1.13.0 agent once grew to 2.1 GB and spun for a
+    day before going silent; this is how Steward sees the next one coming."""
+    out: dict = {}
+    for line in _read_file("/proc/self/status").splitlines():
+        if line.startswith("VmRSS:"):
+            out["rss_bytes"] = int(line.split()[1]) * 1024
+            break
+    t = os.times()
+    cpu, mono = t.user + t.system, time.monotonic()
+    prev_cpu, prev_mono = state.get("self_cpu"), state.get("self_mono")
+    if prev_cpu is not None and mono > prev_mono:
+        out["cpu_pct"] = round(100.0 * max(0.0, cpu - prev_cpu) / (mono - prev_mono), 2)
+    state["self_cpu"], state["self_mono"] = cpu, mono
+    return out
+
+
 # ─── payload ─────────────────────────────────────────────────────────────────
 
 
@@ -1368,6 +1436,12 @@ def build_sample(mounts: list[str], state: dict,
         if logs:
             sample["docker_logs"] = logs
 
+    # Last, so its CPU covers the whole of this sample's work.
+    try:
+        sample["agent"] = collect_self(state)
+    except Exception:
+        pass
+
     return sample
 
 

plugins/host_agent/routes.py

@@ -149,6 +149,14 @@ def _expand_sample_to_metrics(
     if sample.get("uptime_secs") is not None:
         row("uptime_secs", host_name, sample["uptime_secs"])
 
+    # The agent's own footprint (#5764), so a runaway agent shows before it
+    # goes silent.
+    me = sample.get("agent") or {}
+    if me.get("rss_bytes") is not None:
+        row("agent_rss_bytes", host_name, me["rss_bytes"])
+    if me.get("cpu_pct") is not None:
+        row("agent_cpu_pct", host_name, me["cpu_pct"])
+
     # The fullest disk is the fullest by space or by inodes (#5745): out of
     # inodes, every file create fails with space to spare, so the disk rules
     # and the overview, which read disk_used_pct_worst, see either.