diff --git a/plugins/host_agent/agent.py b/plugins/host_agent/agent.py index 98a237e..653f073 100644 --- a/plugins/host_agent/agent.py +++ b/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×tamps=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 diff --git a/plugins/host_agent/templates/install.sh.j2 b/plugins/host_agent/templates/install.sh.j2 index 02d55be..5def950 100644 --- a/plugins/host_agent/templates/install.sh.j2 +++ b/plugins/host_agent/templates/install.sh.j2 @@ -45,6 +45,10 @@ cat > "$CONF_FILE" < bytes: + """Build one Docker multiplexed-log frame (8-byte header + payload).""" + return bytes([stream, 0, 0, 0]) + len(payload).to_bytes(4, "big") + payload + + +def test_demux_docker_logs_framed(): + raw = _frame(1, b"out line\n") + _frame(2, b"err line\n") + assert a._demux_docker_logs(raw) == [ + ("stdout", b"out line\n"), ("stderr", b"err line\n")] + + +def test_demux_docker_logs_tty_fallback(): + # A TTY container's stream isn't framed — the leading byte ('2') is not a + # valid stream id, so the whole blob is treated as one stdout payload. + raw = b"2023-11-14T12:00:00.000000000Z hi\n" + assert a._demux_docker_logs(raw) == [("stdout", raw)] + + +def test_demux_docker_logs_empty(): + assert a._demux_docker_logs(b"") == [] + + +def test_parse_log_ts_handles_nanoseconds_and_z(): + dt = a._parse_log_ts("2023-11-14T12:00:00.123456789Z") + assert dt is not None + assert dt.year == 2023 and dt.microsecond == 123456 # ns trimmed to µs + assert dt.utcoffset().total_seconds() == 0 + assert a._parse_log_ts("not-a-time") is None + assert a._parse_log_ts("") is None + + +def test_parse_container_logs_splits_streams_and_ts(): + raw = (_frame(1, b"2023-11-14T12:00:00.000000001Z hello world\n") + + _frame(2, b"2023-11-14T12:00:01.000000000Z oops\n")) + parsed = a._parse_container_logs(raw) + assert len(parsed) == 2 + dt0, s0, l0 = parsed[0] + dt1, s1, l1 = parsed[1] + assert (s0, l0) == ("stdout", "hello world") + assert (s1, l1) == ("stderr", "oops") + assert dt1 > dt0 + + +def test_parse_container_logs_keeps_unparseable_line(): + # No timestamp prefix (e.g. a partial write) → dt is None, full line kept. + parsed = a._parse_container_logs(_frame(1, b"no-timestamp-here\n")) + assert parsed == [(None, "stdout", "no-timestamp-here")] + + +def test_collect_docker_logs_since_cursor_and_dedup(monkeypatch): + containers = [{"name": "web", "status": "running"}] + calls = [] + t1 = "2023-11-14T12:00:00.000000000Z" + t2 = "2023-11-14T12:00:01.000000000Z" + t3 = "2023-11-14T12:00:02.000000000Z" + + def fake_raw(socket_path, path, timeout=a.DOCKER_API_TIMEOUT): + calls.append(path) + if "tail=" in path: # first interval seeds from a tail + return _frame(1, f"{t1} a\n".encode()) + _frame(1, f"{t2} b\n".encode()) + # second interval: daemon re-returns the boundary line (t2) + a new one + return _frame(1, f"{t2} b\n".encode()) + _frame(1, f"{t3} c\n".encode()) + + monkeypatch.setattr(a, "_docker_request_raw", fake_raw) + state: dict = {} + first = a.collect_docker_logs("/sock", containers, state) + assert [r["line"] for r in first] == ["a", "b"] + assert "tail=" in calls[0] + + second = a.collect_docker_logs("/sock", containers, state) + # boundary line b (t2) deduped against the cursor; only c (t3) is new + assert [r["line"] for r in second] == ["c"] + assert "since=" in calls[1] + + +def test_collect_docker_logs_excludes_and_skips_non_running(monkeypatch): + containers = [ + {"name": "web", "status": "running"}, + {"name": "noisy", "status": "running"}, + {"name": "db", "status": "exited"}, + ] + seen = [] + + def fake_raw(socket_path, path, timeout=a.DOCKER_API_TIMEOUT): + seen.append(path) + return _frame(1, b"2023-11-14T12:00:00.000000000Z x\n") + + monkeypatch.setattr(a, "_docker_request_raw", fake_raw) + out = a.collect_docker_logs("/sock", containers, {}, exclude=["noisy"]) + # only web is fetched: noisy is excluded, db isn't running + assert seen and all("/containers/web/" in p for p in seen) + assert {r["container"] for r in out} == {"web"} + + +def test_collect_docker_logs_byte_cap_truncates(monkeypatch): + containers = [{"name": "web", "status": "running"}] + blob = b"".join( + _frame(1, f"2023-11-14T12:00:{i:02d}.000000000Z {'x' * 20}\n".encode()) + for i in range(10)) + monkeypatch.setattr(a, "_docker_request_raw", + lambda s, p, timeout=a.DOCKER_API_TIMEOUT: blob) + out = a.collect_docker_logs("/sock", containers, {}, max_bytes=50) + # Once the cap is crossed a single marker line is appended and we stop. + assert out[-1]["container"] == "_steward" + assert "truncated" in out[-1]["line"] + real = [r for r in out if r["container"] == "web"] + assert 0 < len(real) < 10 + + +def test_collect_docker_logs_forgets_gone_container_cursors(monkeypatch): + monkeypatch.setattr( + a, "_docker_request_raw", + lambda s, p, timeout=a.DOCKER_API_TIMEOUT: + _frame(1, b"2023-11-14T12:00:00.000000000Z x\n")) + state: dict = {} + a.collect_docker_logs("/sock", [{"name": "web", "status": "running"}], state) + assert "web" in state["docker_log_cursors"] + # web is gone next interval → its cursor is pruned so state can't grow forever + a.collect_docker_logs("/sock", [{"name": "db", "status": "running"}], state) + assert "web" not in state["docker_log_cursors"] + assert "db" in state["docker_log_cursors"] + + +def test_build_sample_includes_logs_when_present(monkeypatch): + monkeypatch.setattr(a, "collect_docker", + lambda _s: [{"name": "web", "status": "running"}]) + monkeypatch.setattr(a, "collect_swarm", lambda _s: None) + monkeypatch.setattr(a, "collect_disk_usage", lambda _s: None) + monkeypatch.setattr( + a, "collect_docker_logs", + lambda *args, **kw: [{"container": "web", "stream": "stdout", + "ts": "t", "line": "hello"}]) + sample = a.build_sample(["/"], {}, "/var/run/docker.sock") + assert sample["docker_logs"][0]["line"] == "hello" + + +def test_build_sample_omits_logs_when_disabled(monkeypatch): + monkeypatch.setattr(a, "collect_docker", + lambda _s: [{"name": "web", "status": "running"}]) + monkeypatch.setattr(a, "collect_swarm", lambda _s: None) + monkeypatch.setattr(a, "collect_disk_usage", lambda _s: None) + called = {"logs": False} + + def _logs(*args, **kw): + called["logs"] = True + return [{"container": "web", "line": "x"}] + + monkeypatch.setattr(a, "collect_docker_logs", _logs) + sample = a.build_sample(["/"], {}, "/var/run/docker.sock", + docker_logs_enabled=False) + assert "docker_logs" not in sample + assert called["logs"] is False # collection skipped entirely, not just dropped + + +def test_drop_logs_strips_only_logs(): + sample = {"ts": "t", "cpu_pct": 5.0, + "docker": [{"name": "web"}], "docker_logs": [{"line": "x"}]} + out = a._drop_logs(sample) + assert "docker_logs" not in out + assert out["docker"] == [{"name": "web"}] and out["cpu_pct"] == 5.0 # metrics kept + assert a._drop_logs(out) is out # idempotent + + +def test_read_config_docker_logs_default_on(tmp_path): + p = tmp_path / "agent.conf" + p.write_text("url = x\ntoken = y\n") + cfg = a.read_config(str(p)) + assert cfg["docker_logs_enabled"] is True + assert cfg["docker_log_exclude"] == [] + + +def test_read_config_disables_docker_logs(tmp_path): + p = tmp_path / "agent.conf" + p.write_text("url = x\ntoken = y\ndocker_logs_enabled = false\n") + cfg = a.read_config(str(p)) + assert cfg["docker_logs_enabled"] is False + + +def test_read_config_parses_docker_log_exclude(tmp_path): + p = tmp_path / "agent.conf" + p.write_text("url = x\ntoken = y\ndocker_log_exclude = watchtower, foo\n") + cfg = a.read_config(str(p)) + assert cfg["docker_log_exclude"] == ["watchtower", "foo"]