FabledSteward/tests/plugins/host_agent/test_agent_docker_bounds.py

# tests/plugins/host_agent/test_agent_docker_bounds.py
"""#5764: everything the agent reads from Docker is bounded in time and size.

A 1.13.0 agent read a container's whole log backlog into memory on its main
loop, with only a per-read timeout, and went silent for a day at 2.1 GB. The
reader tests here talk to a real Unix socket that trickles or floods, so the
bound is the one the agent actually has, not a mocked one.
"""
import os
import socket
import tempfile
import threading
import time

import pytest

from plugins.host_agent import agent as a


@pytest.fixture
def docker_sock():
    """A Unix socket that answers each connection with `serve(conn)`."""
    d = tempfile.mkdtemp(dir="/tmp")
    path = os.path.join(d, "d.sock")
    srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    srv.bind(path)
    srv.listen(1)
    holder = {}

    def run():
        conn, _ = srv.accept()
        conn.recv(4096)
        try:
            holder["serve"](conn)
        except OSError:
            pass                       # the agent hung up on us: expected
        finally:
            conn.close()

    def start(serve):
        holder["serve"] = serve
        threading.Thread(target=run, daemon=True).start()
        return path

    yield start
    srv.close()


def test_an_answer_that_keeps_trickling_is_cut_off_at_its_deadline(docker_sock):
    def trickle(conn):
        conn.sendall(b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n\r\n[")
        while True:                     # a byte every 50 ms: never idle 5 s, never done
            conn.sendall(b" ")
            time.sleep(0.05)

    path = docker_sock(trickle)
    began = time.monotonic()
    with pytest.raises(OSError):
        a._docker_get(path, "/containers/json", "application/json",
                      timeout=5.0, deadline=0.4, max_bytes=1 << 20)
    assert time.monotonic() - began < 2.0


def test_an_answer_over_its_size_is_refused_not_held(docker_sock):
    def flood(conn):
        conn.sendall(b"HTTP/1.1 200 OK\r\n\r\n")
        block = b"x" * 65536
        for _ in range(64):             # 4 MiB on offer
            conn.sendall(block)

    path = docker_sock(flood)
    with pytest.raises(a.DockerAnswerTooLarge):
        a._docker_get(path, "/containers/web/logs", "application/octet-stream",
                      timeout=5.0, deadline=5.0, max_bytes=256 * 1024)


def test_a_whole_answer_still_reads(docker_sock):
    def ok(conn):
        conn.sendall(b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n"
                     b"2\r\n[]\r\n0\r\n\r\n")

    assert a._docker_request(docker_sock(ok), "/containers/json") == []


def _frame(stream: int, payload: bytes) -> bytes:
    return bytes([stream, 0, 0, 0]) + len(payload).to_bytes(4, "big") + payload


def _log_lines(n: int) -> bytes:
    return b"".join(_frame(1, f"2026-10-14T12:00:{i % 60:02d}.{i:09d}Z line {i}\n".encode())
                    for i in range(n))


def test_a_busy_container_is_read_from_its_newest_lines_and_says_what_it_skipped(monkeypatch):
    asked = []

    def fake_raw(socket_path, path, timeout=a.DOCKER_API_TIMEOUT):
        asked.append(path)
        return _log_lines(a.DOCKER_LOG_MAX_LINES)

    monkeypatch.setattr(a, "_docker_request_raw", fake_raw)
    state = {"docker_log_cursors": {"web": 1791979200.0}}     # 2026-10-14T12:00:00Z
    out = a.collect_docker_logs("/sock", [{"name": "web", "status": "running"}], state,
                                max_bytes=1 << 30)
    assert "since=" in asked[0] and f"tail={a.DOCKER_LOG_MAX_LINES}" in asked[0]
    notes = [r["line"] for r in out if r["container"] == "_steward"]
    assert any("older ones skipped" in n for n in notes)


def test_a_container_whose_log_is_too_big_skips_ahead_instead_of_rereading(monkeypatch):
    def too_big(socket_path, path, timeout=a.DOCKER_API_TIMEOUT):
        raise a.DockerAnswerTooLarge("too big")

    monkeypatch.setattr(a, "_docker_request_raw", too_big)
    state = {"docker_log_cursors": {"web": 1791979200.0}}
    out = a.collect_docker_logs("/sock", [{"name": "web", "status": "running"}], state,
                                now=1791979500.0)
    assert state["docker_log_cursors"]["web"] == 1791979500.0   # next read starts at now
    assert "skipped ahead to now" in out[-1]["line"]


def test_log_reading_stops_at_its_deadline(monkeypatch):
    asked = []
    monkeypatch.setattr(a, "_docker_request_raw",
                        lambda s, p, timeout=a.DOCKER_API_TIMEOUT: asked.append(p) or b"")
    out = a.collect_docker_logs("/sock", [{"name": "web", "status": "running"}], {},
                                deadline=0.0)
    assert asked == []
    assert "limit" in out[-1]["line"]


def test_the_agent_reports_its_own_memory_and_cpu():
    state: dict = {}
    first = a.collect_self(state)
    assert first["rss_bytes"] > 1024 * 1024
    assert "cpu_pct" not in first               # no previous sample to measure from
    sum(range(200_000))
    second = a.collect_self(state)
    assert second["cpu_pct"] >= 0.0