From dff76ecd308db212bb09a746a80c2c30ba00dbfe Mon Sep 17 00:00:00 2001 From: oskar Date: Thu, 16 Jul 2026 20:20:17 +0200 Subject: [PATCH] =?UTF-8?q?fix(events):=20service=5Fhealthy=20tylko=20na?= =?UTF-8?q?=20transition=20+=20cleanup=20358k=20backlog=20+=20retencja=20?= =?UTF-8?q?=E2=80=94=20glob=20zalewu=20paralizowal=20reconcile=20superviso?= =?UTF-8?q?ra?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - node-agent: service_healthy emitowany tylko przy przejsciu w stan zdrowy (per-service in-memory state), nie co cykl dla kazdego zdrowego serwisu. To samo dla control-plane HTTP probe. healthcheck_failed/containers_not_running/ incydenty pozostaja emitowane bez zmian (realne sygnaly). - node-agent: naprawiono _cleanup_control_plane_fs — istniejaca retencja eventow byla martwa od migracji observer_checkpoint.json na epoch-int (str(path) <= int rzucal TypeError, lapane cicho przez broad except). Teraz porownuje epoch-do-epoch i czysci wylacznie service_healthy/node_health starsze niz checkpoint + 3-dniowy bufor; healthcheck_failed/incydenty/ha_* zachowane bezterminowo. - scripts/maintenance/cleanup_event_backlog.py: jednorazowy skrypt do bezpiecznego czyszczenia backlogu na VPS (dry-run domyslnie, --apply do usuniecia). Ten sam warunek: typ szumu + starsze niz checkpoint + 1h bufor. --- scripts/maintenance/cleanup_event_backlog.py | 206 ++++++++++++++++++ .../tests/test_cleanup_event_backlog.py | 111 ++++++++++ services/node-agent/src/node_agent.py | 158 ++++++++++++-- .../node-agent/tests/test_check_containers.py | 53 +++++ .../tests/test_cleanup_control_plane_fs.py | 152 +++++++++++++ .../tests/test_control_plane_health.py | 64 ++++++ 6 files changed, 725 insertions(+), 19 deletions(-) create mode 100644 scripts/maintenance/cleanup_event_backlog.py create mode 100644 scripts/maintenance/tests/test_cleanup_event_backlog.py create mode 100644 services/node-agent/tests/test_cleanup_control_plane_fs.py create mode 100644 services/node-agent/tests/test_control_plane_health.py diff --git a/scripts/maintenance/cleanup_event_backlog.py b/scripts/maintenance/cleanup_event_backlog.py new file mode 100644 index 0000000..360dbe3 --- /dev/null +++ b/scripts/maintenance/cleanup_event_backlog.py @@ -0,0 +1,206 @@ +#!/usr/bin/env python3 +""" +cleanup_event_backlog.py — safe backlog cleanup for /opt/homelab/events/. + +Removes ONLY processed noise events (service_healthy, node_health by default) +that are both: + 1. Strictly older than the observer's per-node checkpoint timestamp + (observer_checkpoint.json) — i.e. guaranteed already processed into + world_state, per the same ordering guarantee observer.py itself relies on + (oldest-first, monotonically-advancing checkpoint). + 2. Older than a minimum safety age (default 3600s / 1h) — defense in depth + against clock skew or a checkpoint write racing this script, mirroring + the dual-condition pattern already used in node_agent.py's + _cleanup_control_plane_fs. + +Everything else — healthcheck_failed, containers_not_running, incidents, +ha_* events, node liveness transitions, disk/memory/cpu pressure, and any +event newer than the checkpoint — is never touched. + +Defaults to DRY RUN: prints counts per node/type and does not delete +anything. Pass --apply to actually delete. + +Usage: + python3 cleanup_event_backlog.py # dry run, defaults + python3 cleanup_event_backlog.py --apply # actually delete + python3 cleanup_event_backlog.py --types service_healthy,node_health + python3 cleanup_event_backlog.py --min-age-secs 86400 # extra-cautious +""" +import argparse +import json +import os +import re +import sys +import time +from pathlib import Path + +RUNTIME_PATH = Path(os.getenv("RUNTIME_PATH", "/opt/homelab")) +EVENTS_DIR = RUNTIME_PATH / "events" +CHECKPOINT_FILE = RUNTIME_PATH / "state" / "observer_checkpoint.json" + +# Matches observer.py's _EVENT_TS_RE: evt----.json +_EVENT_TS_RE = re.compile(r"-(\d{9,11})-") + +DEFAULT_TYPES = {"service_healthy", "node_health"} + + +def _ts_from_name(name: str): + m = _EVENT_TS_RE.search(Path(name).stem) + return int(m.group(1)) if m else None + + +def _type_from_name(name: str): + """Best-effort event type extraction from evt----.json. + + Mirrors the informal convention node-agent/observer use when building + filenames; falls back to 'unknown' rather than guessing wrong. + """ + stem = Path(name).stem + m = re.match(r"evt-.+?-\d{9,11}-(.+)$", stem) + if not m: + return "unknown" + rest = m.group(1) + # rest is "-"; type itself may contain underscores but not + # dashes in this codebase's event vocabulary, so split on first dash after + # the known type token is ambiguous — instead check against DEFAULT_TYPES + # and a couple of other well-known multi-word-free types directly. + for t in KNOWN_TYPES: + if rest == t or rest.startswith(t + "-"): + return t + return rest.split("-")[0] + + +# Every event type this repo's emitters are known to produce (node_agent.py, +# ha_diag/event_emitter.py, observer.py's own _emit_node_transition), longest +# first so startswith() matching in _type_from_name doesn't shadow a longer +# type with a shorter prefix (e.g. "node_health" vs "node_offline" share no +# prefix, but keep this list explicit rather than relying on luck). +KNOWN_TYPES = sorted( + { + "service_healthy", "service_unhealthy", "service_recovered", + "node_health", "node_online", "node_offline", "node_stale", + "healthcheck_failed", "containers_not_running", "container_restarting", + "container_state_unexpected", "disk_pressure", "high_memory", "high_cpu", + "mqtt_unreachable", + "ha_websocket_dead", "ha_websocket_recovered", "ha_integration_failed", + "ha_entity_unavailable_long", "ha_automation_failing", "ha_update_available", + "ha_recorder_lag", "ha_system_health_degraded", + "deployment_started", "deployment_completed", "deployment_failed", + "remediation_started", "remediation_completed", + }, + key=len, reverse=True, +) + + +def load_checkpoints(checkpoint_file: Path = CHECKPOINT_FILE): + if not checkpoint_file.exists(): + print(f"WARNING: no checkpoint file at {checkpoint_file} — refusing to " + f"guess; nothing is safe to delete without it.", file=sys.stderr) + return {} + data = json.loads(checkpoint_file.read_text()) + raw = data.get("node_checkpoints", {}) + out = {} + for node, val in raw.items(): + if isinstance(val, bool): + continue + if isinstance(val, (int, float)): + out[node] = int(val) + return out + + +def scan(types_to_clean, min_age_secs, checkpoints, now, events_dir: Path = EVENTS_DIR): + """Return (candidates: list[Path], counts: dict[node][type] -> (delete, keep)).""" + candidates = [] + counts = {} + if not events_dir.exists(): + return candidates, counts + + for node_dir in sorted(p for p in events_dir.iterdir() if p.is_dir()): + node = node_dir.name + checkpoint = checkpoints.get(node) + counts.setdefault(node, {}) + for f in node_dir.glob("*.json"): + etype = _type_from_name(f.name) + ts = _ts_from_name(f.name) + if ts is None: + try: + ts = int(f.stat().st_mtime) + except OSError: + continue + + bucket = counts[node].setdefault(etype, {"delete": 0, "keep": 0}) + + if etype not in types_to_clean: + bucket["keep"] += 1 + continue + if checkpoint is None: + # No checkpoint for this node dir at all (e.g. a ghost/foreign + # dir) — never guess; keep everything. + bucket["keep"] += 1 + continue + if ts >= checkpoint: + # Not guaranteed processed yet — never delete. + bucket["keep"] += 1 + continue + if now - ts < min_age_secs: + # Within the safety buffer even though past checkpoint. + bucket["keep"] += 1 + continue + + bucket["delete"] += 1 + candidates.append(f) + + return candidates, counts + + +def print_report(counts, min_age_secs, checkpoints): + total_delete = 0 + total_keep = 0 + for node in sorted(counts): + cp = checkpoints.get(node) + cp_str = f"checkpoint={cp}" if cp is not None else "NO CHECKPOINT (nothing eligible)" + print(f"\n=== {node} ({cp_str}) ===") + for etype, c in sorted(counts[node].items(), key=lambda kv: -sum(kv[1].values())): + if c["delete"] or c["keep"]: + print(f" {etype:32s} delete={c['delete']:>7d} keep={c['keep']:>7d}") + total_delete += c["delete"] + total_keep += c["keep"] + print(f"\nTOTAL: delete={total_delete} keep={total_keep}") + print(f"(min_age_secs={min_age_secs}, i.e. only files older than checkpoint AND " + f"more than {min_age_secs}s old are ever deleted)") + + +def main(): + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("--apply", action="store_true", + help="Actually delete. Without this flag, dry-run only.") + ap.add_argument("--types", default=",".join(sorted(DEFAULT_TYPES)), + help=f"Comma-separated event types to clean (default: {DEFAULT_TYPES})") + ap.add_argument("--min-age-secs", type=int, default=3600, + help="Extra safety buffer past the checkpoint (default 3600s = 1h)") + args = ap.parse_args() + + types_to_clean = {t.strip() for t in args.types.split(",") if t.strip()} + checkpoints = load_checkpoints(CHECKPOINT_FILE) + now = int(time.time()) + + candidates, counts = scan(types_to_clean, args.min_age_secs, checkpoints, now, EVENTS_DIR) + print_report(counts, args.min_age_secs, checkpoints) + + if not args.apply: + print(f"\nDRY RUN — no files deleted. Re-run with --apply to delete " + f"the {len(candidates)} files listed above as 'delete'.") + return + + deleted = 0 + for f in candidates: + try: + f.unlink(missing_ok=True) + deleted += 1 + except Exception as exc: + print(f"Failed to remove {f}: {exc}", file=sys.stderr) + print(f"\nDeleted {deleted}/{len(candidates)} files.") + + +if __name__ == "__main__": + main() diff --git a/scripts/maintenance/tests/test_cleanup_event_backlog.py b/scripts/maintenance/tests/test_cleanup_event_backlog.py new file mode 100644 index 0000000..6f5613f --- /dev/null +++ b/scripts/maintenance/tests/test_cleanup_event_backlog.py @@ -0,0 +1,111 @@ +"""Tests for cleanup_event_backlog.py — the event-flood backlog cleanup script. + +Covers the safety invariant the whole cleanup depends on: only events that are +(a) a noise type, (b) strictly older than the observer's per-node checkpoint, +and (c) past the extra min-age safety buffer are ever deleted. Everything +else — unprocessed events, real signals, foreign/ghost node dirs — survives. +""" +from __future__ import annotations + +import json +import sys +import time +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +import cleanup_event_backlog as cleanup # noqa: E402 + + +def _write(node_dir: Path, ts: int, etype: str, svc: str = "svc") -> Path: + p = node_dir / f"evt-{node_dir.name}-{ts}-{etype}-{svc}.json" + p.write_text("{}") + return p + + +def test_deletes_only_processed_noise_past_buffer(tmp_path): + events_dir = tmp_path / "events" + node_dir = events_dir / "piha" + node_dir.mkdir(parents=True) + now = int(time.time()) + checkpoint = now - 100 + + old_healthy = _write(node_dir, now - 7200, "service_healthy") + old_node_health = _write(node_dir, now - 7200, "node_health") + recent_healthy = _write(node_dir, now - 30, "service_healthy", svc="svc2") # within buffer + unprocessed = _write(node_dir, now + 50, "service_healthy", svc="svc3") # after checkpoint + real_signal = _write(node_dir, now - 999999, "healthcheck_failed") + + checkpoints = {"piha": checkpoint} + candidates, counts = cleanup.scan( + {"service_healthy", "node_health"}, 3600, checkpoints, now, events_dir + ) + + assert set(candidates) == {old_healthy, old_node_health} + for f in (recent_healthy, unprocessed, real_signal): + assert f.exists() + assert counts["piha"]["healthcheck_failed"]["delete"] == 0 + assert counts["piha"]["healthcheck_failed"]["keep"] == 1 + + +def test_never_deletes_without_checkpoint_for_node(tmp_path): + """A node dir with no checkpoint entry (ghost/foreign dir) must be left alone.""" + events_dir = tmp_path / "events" + node_dir = events_dir / "be17cb6eb0f6" + node_dir.mkdir(parents=True) + now = int(time.time()) + _write(node_dir, now - 999999, "service_healthy") + + candidates, counts = cleanup.scan( + {"service_healthy", "node_health"}, 3600, {}, now, events_dir + ) + assert candidates == [] + assert counts["be17cb6eb0f6"]["service_healthy"]["keep"] == 1 + + +def test_never_deletes_incidents_or_unknown_types(tmp_path): + events_dir = tmp_path / "events" + node_dir = events_dir / "vps" + node_dir.mkdir(parents=True) + now = int(time.time()) + checkpoint = now - 100 + ha_event = _write(node_dir, now - 999999, "ha_entity_unavailable_long") + incident_like = _write(node_dir, now - 999999, "containers_not_running") + + candidates, _counts = cleanup.scan( + {"service_healthy", "node_health"}, 3600, {"vps": checkpoint}, now, events_dir + ) + assert candidates == [] + assert ha_event.exists() + assert incident_like.exists() + + +def test_apply_actually_deletes_and_dry_run_does_not(tmp_path, capsys, monkeypatch): + events_dir = tmp_path / "events" + state_dir = tmp_path / "state" + node_dir = events_dir / "piha" + node_dir.mkdir(parents=True) + state_dir.mkdir(parents=True) + now = int(time.time()) + checkpoint = now - 100 + old_healthy = _write(node_dir, now - 7200, "service_healthy") + (state_dir / "observer_checkpoint.json").write_text( + json.dumps({"node_checkpoints": {"piha": checkpoint}}) + ) + + monkeypatch.setattr(cleanup, "EVENTS_DIR", events_dir) + monkeypatch.setattr(cleanup, "CHECKPOINT_FILE", state_dir / "observer_checkpoint.json") + monkeypatch.setattr(sys, "argv", ["cleanup_event_backlog.py"]) + cleanup.main() + assert old_healthy.exists() # dry run: untouched + + monkeypatch.setattr(sys, "argv", ["cleanup_event_backlog.py", "--apply"]) + cleanup.main() + assert not old_healthy.exists() # applied: deleted + + +def test_type_extraction_matches_node_agent_filename_convention(): + assert cleanup._type_from_name("evt-piha-1784215978-service_healthy-mosquitto.json") == "service_healthy" + assert cleanup._type_from_name("evt-vps-1784215978-node_health-node.json") == "node_health" + assert cleanup._type_from_name("evt-piha-1784215978-healthcheck_failed-z2m.json") == "healthcheck_failed" + assert cleanup._type_from_name("evt-piha-1784215978-ha_entity_unavailable_long-node.json") == \ + "ha_entity_unavailable_long" diff --git a/services/node-agent/src/node_agent.py b/services/node-agent/src/node_agent.py index 337290f..a457914 100644 --- a/services/node-agent/src/node_agent.py +++ b/services/node-agent/src/node_agent.py @@ -128,6 +128,66 @@ def _utc_iso() -> str: return datetime.now(timezone.utc).isoformat() +# Matches evt----.json (this module's own emit_event +# naming, mirrored by observer.py's _EVENT_TS_RE / _ts_from_event_name so both +# sides of the checkpoint agree on how to read a filename's embedded epoch). +_EVENT_TS_RE = re.compile(r"-(\d{9,11})-") +_EVENT_TYPE_RE = re.compile(r"^evt-.+?-\d{9,11}-(.+)$") + + +def _event_ts_from_filename(name: str): + """Return the embedded from an event filename, or None if absent.""" + m = _EVENT_TS_RE.search(Path(name).stem) + return int(m.group(1)) if m else None + + +def _event_type_from_filename(name: str) -> str: + """Best-effort event type from evt----.json. + + Used only to decide retention eligibility (never for dispatch), so an + imprecise match on an unrecognized type is safe: it simply keeps the file. + """ + stem = Path(name).stem + m = _EVENT_TYPE_RE.match(stem) + if not m: + return "unknown" + rest = m.group(1) + for t in _RETENTION_NOISE_TYPES: + if rest == t or rest.startswith(t + "-"): + return t + return rest + + +def _checkpoint_ts_from_value(value) -> int: + """Coerce a stored observer_checkpoint.json value into an int epoch. + + Mirrors observer.py's own _checkpoint_ts_from_value: current format is an + int/float epoch; older observer builds stored a lexical path string, from + which the embedded is extracted. Unparseable/absent -> 0, which + compares as "nothing processed yet" and safely keeps every file for that + node rather than guessing a checkpoint too high and deleting unprocessed + events. + """ + if isinstance(value, bool): + return 0 + if isinstance(value, (int, float)): + return int(value) + if isinstance(value, str) and value: + ts = _event_ts_from_filename(value) + if ts is not None: + return ts + return 0 + + +# Event types eligible for the ongoing retention sweep in +# _cleanup_control_plane_fs: pure positive-confirmation noise, never the +# actionable signals (healthcheck_failed, containers_not_running, ha_*, +# incidents, disk/memory/cpu pressure, deployment/remediation records, node +# liveness transitions), which are kept indefinitely by this method — an +# operator or the panel event feed may need to look back at those. +_RETENTION_NOISE_TYPES = frozenset({"service_healthy", "node_health"}) + + # --------------------------------------------------------------------------- # NodeAgent # --------------------------------------------------------------------------- @@ -138,6 +198,16 @@ class NodeAgent: self.node_type = _resolve_node_type() self._ensure_dirs() + # Tracks the last known health classification per service (by canonical + # name — container service names AND the synthetic "control-plane" key + # used by _check_control_plane_health share this dict) so service_healthy + # is emitted only on the unhealthy->healthy transition, not once per cycle + # for a service that is already healthy. In-memory only: reset on process + # restart is acceptable — it just re-emits one confirmation event, not a + # flood, and the observer treats that re-confirmation as idempotent (see + # process_event). + self._service_health_state: dict = {} + self.docker_client = None if DOCKER_AVAILABLE: try: @@ -361,6 +431,7 @@ class NodeAgent: # Exited container that carries an auto-restart policy if status in ("exited", "dead"): + self._service_health_state[name] = False logger.warning(f"Container exited: {name} (restart={restart_policy})") self.emit_event( "containers_not_running", "high", name, @@ -378,6 +449,7 @@ class NodeAgent: # alarm, while a real crash-loop escalates to the same actionable # signal as an exited container. elif status == "restarting": + self._service_health_state[name] = False if restart_count >= CRASH_LOOP_RESTART_THRESHOLD: # Genuine crash-loop: reuse containers_not_running so it # rides the existing, supervisor-wired remediation path @@ -414,6 +486,7 @@ class NodeAgent: # Running container with a failing health check elif status == "running" and health_status == "unhealthy": + self._service_health_state[name] = False logger.warning(f"Container unhealthy: {name}") self.emit_event( "healthcheck_failed", "high", name, @@ -425,18 +498,31 @@ class NodeAgent: # services.json stays populated for the supervisor's drift detection. # Without this, the supervisor sees services.json as empty and treats # all desired services as "missing", flooding the action queue. + # + # Emitted ONLY on the unhealthy/unknown -> healthy transition, not + # every cycle: a healthy service otherwise re-confirms itself once + # per health-check interval forever, which is exactly the flood + # that paralyzed the supervisor's reconcile() glob (358k backlog + # files, 91% service_healthy). The positive "still healthy" state is + # already carried by the presence of a `healthy` services.json entry + # (written once here) plus the node_health heartbeat every cycle — + # so a stuck/frozen node_agent is still visible via node liveness, + # without needing a fresh service_healthy file per service per cycle. elif status == "running": - self.emit_event( - "service_healthy", "info", name, - f"Container '{name}' is running", - {**base_payload, "health_status": health_status or "none"}, - ) + if self._service_health_state.get(name) is not True: + self.emit_event( + "service_healthy", "info", name, + f"Container '{name}' is running", + {**base_payload, "health_status": health_status or "none"}, + ) + self._service_health_state[name] = True # Deliberately paused. docker pause is always an operator/tool # action (Docker never auto-pauses), so it is not a fault — but a # managed service should not normally sit paused, so surface it # observationally rather than swallowing it. Non-actionable. elif status == "paused": + self._service_health_state[name] = False logger.info(f"Container paused: {name}") self.emit_event( "container_state_unexpected", "medium", name, @@ -456,6 +542,7 @@ class NodeAgent: # silently: emit a diagnostic so a new/unknown state becomes visible # instead of quietly recreating the very blind spot this fix closes. else: + self._service_health_state[name] = False logger.warning( f"Container in unhandled Docker state: {name} (state='{status}')" ) @@ -587,23 +674,42 @@ class NodeAgent: except Exception as exc: logger.error(f"Failed to remove {f}: {exc}") - # 3. Event files older than 3 days AND already past observer checkpoint. - # The dual condition guarantees we never delete an unprocessed event. - # Checkpoint format: {"node_checkpoints": {"piha": "/path/last", ...}} + # 3. Noise event files (service_healthy / node_health) that are BOTH + # already past the observer's per-node checkpoint (guaranteed + # processed into world_state — see _checkpoint_ts_from_value) AND + # older than 3 days (safety buffer past the checkpoint). The dual + # condition guarantees we never delete an unprocessed event, and + # the type restriction guarantees healthcheck_failed / incidents / + # ha_* / node-liveness events are kept indefinitely by this method. + # + # BUG FIX (event-flood cleanup): this previously compared str(f) + # (a full file path) against the checkpoint value with <=, which + # only ever worked when the checkpoint stored the pre-fix lexical + # path format. Since observer.py migrated node_checkpoints to int + # epoch timestamps, every comparison here raised + # "'<=' not supported between instances of 'str' and 'int'", + # silently caught by the broad except below — so this cleanup has + # been a no-op ever since, which is the direct cause of the 358k-file + # backlog that paralyzed the supervisor's reconcile(). Fixed by + # comparing epoch-to-epoch, same as observer.py's own checkpoint + # logic. checkpoint_file = STATE_DIR / "observer_checkpoint.json" node_checkpoints: dict = {} if checkpoint_file.exists(): try: cp = json.loads(checkpoint_file.read_text()) if "node_checkpoints" in cp: - node_checkpoints = cp["node_checkpoints"] + node_checkpoints = { + node: _checkpoint_ts_from_value(val) + for node, val in (cp["node_checkpoints"] or {}).items() + } elif "last_processed_file" in cp: # Migrate old single-file format old = cp.get("last_processed_file", "") if old: try: node_dir = Path(old).relative_to(EVENTS_DIR).parts[0] - node_checkpoints = {node_dir: old} + node_checkpoints = {node_dir: _checkpoint_ts_from_value(old)} except Exception: pass except Exception as exc: @@ -612,13 +718,20 @@ class NodeAgent: if node_checkpoints: for f in EVENTS_DIR.glob("**/*.json"): try: + etype = _event_type_from_filename(f.name) + if etype not in _RETENTION_NOISE_TYPES: + continue # Determine which node directory this event belongs to rel = Path(f).relative_to(EVENTS_DIR) node_dir = str(rel.parts[0]) if rel.parts else "__unknown__" - last_for_node = node_checkpoints.get(node_dir, "") - if (last_for_node - and now - f.stat().st_mtime > three_days - and str(f) <= last_for_node): + checkpoint_ts = node_checkpoints.get(node_dir) + if not checkpoint_ts: + continue + event_ts = _event_ts_from_filename(f.name) + if event_ts is None: + event_ts = int(f.stat().st_mtime) + if (event_ts < checkpoint_ts + and now - f.stat().st_mtime > three_days): f.unlink(missing_ok=True) logger.info(f"Cleaned old event: {f.name}") except Exception as exc: @@ -711,18 +824,25 @@ class NodeAgent: try: resp = urllib.request.urlopen(endpoint, timeout=5) if resp.status == 200: - self.emit_event( - "service_healthy", "info", "control-plane", - "Control-plane HTTP endpoint is reachable", - {"endpoint": endpoint}, - ) + # Transition-only, same rationale as check_containers(): without + # this gate, control-plane re-confirmed itself every cycle + # forever, contributing to the same service_healthy flood. + if self._service_health_state.get("control-plane") is not True: + self.emit_event( + "service_healthy", "info", "control-plane", + "Control-plane HTTP endpoint is reachable", + {"endpoint": endpoint}, + ) + self._service_health_state["control-plane"] = True else: + self._service_health_state["control-plane"] = False self.emit_event( "service_unhealthy", "high", "control-plane", f"Control-plane HTTP endpoint returned HTTP {resp.status}", {"endpoint": endpoint, "http_status": resp.status}, ) except Exception as exc: + self._service_health_state["control-plane"] = False self.emit_event( "service_unhealthy", "high", "control-plane", f"Control-plane HTTP endpoint unreachable: {exc}", diff --git a/services/node-agent/tests/test_check_containers.py b/services/node-agent/tests/test_check_containers.py index e6274dc..1a584d8 100644 --- a/services/node-agent/tests/test_check_containers.py +++ b/services/node-agent/tests/test_check_containers.py @@ -127,6 +127,59 @@ def test_running_healthy_emits_service_healthy(agent, monkeypatch): assert events[0][1] == "info" +def test_running_healthy_second_cycle_is_silent(agent, monkeypatch): + """Regression for the event-flood fix: a service that stays healthy across + cycles must NOT re-emit service_healthy every cycle — only on the + transition into healthy.""" + c = make_container("running") + run_check(agent, [c], monkeypatch) # first cycle: transition, emits once + events = run_check(agent, [c], monkeypatch) # second cycle: still healthy + assert events == [] + + +def test_running_healthy_after_unhealthy_reemits_service_healthy(agent, monkeypatch): + """Recovery (unhealthy -> healthy) must still be visible exactly once.""" + unhealthy = make_container("running", health="unhealthy", name="svc") + events = run_check(agent, [unhealthy], monkeypatch) + assert events[0][0] == "healthcheck_failed" + + healthy = make_container("running", name="svc") + events = run_check(agent, [healthy], monkeypatch) + assert len(events) == 1 + assert events[0][0] == "service_healthy" + + # Third cycle, still healthy: silent again. + events = run_check(agent, [healthy], monkeypatch) + assert events == [] + + +def test_running_healthy_after_crash_loop_reemits_service_healthy(agent, monkeypatch): + """Recovery from a crash-loop (containers_not_running) is also a real + transition and must re-confirm health exactly once.""" + crashing = make_container("restarting", restart_count=7, name="svc") + events = run_check(agent, [crashing], monkeypatch) + assert events[0][0] == "containers_not_running" + + recovered = make_container("running", name="svc") + events = run_check(agent, [recovered], monkeypatch) + assert len(events) == 1 + assert events[0][0] == "service_healthy" + + +def test_different_services_tracked_independently(agent, monkeypatch): + """One service's health transition must not gate another service's emission.""" + svc_a = make_container("running", name="svc-a", compose_service="svc-a") + svc_b = make_container("running", name="svc-b", compose_service="svc-b") + + events = run_check(agent, [svc_a, svc_b], monkeypatch) + assert {e[0] for e in events} == {"service_healthy"} + assert len(events) == 2 + + # Both still healthy next cycle: both silent. + events = run_check(agent, [svc_a, svc_b], monkeypatch) + assert events == [] + + def test_running_unhealthy_emits_healthcheck_failed(agent, monkeypatch): c = make_container("running", health="unhealthy") events = run_check(agent, [c], monkeypatch) diff --git a/services/node-agent/tests/test_cleanup_control_plane_fs.py b/services/node-agent/tests/test_cleanup_control_plane_fs.py new file mode 100644 index 0000000..7209f14 --- /dev/null +++ b/services/node-agent/tests/test_cleanup_control_plane_fs.py @@ -0,0 +1,152 @@ +"""Tests for NodeAgent._cleanup_control_plane_fs() event-retention sweep. + +Regression coverage for the event-flood root cause: this method's event +cleanup previously compared str(file_path) against the observer's checkpoint +value with <=, which only worked for the old lexical-path checkpoint format. +Once observer.py migrated node_checkpoints to int epoch timestamps, every +comparison raised a TypeError (caught silently by the broad except), so this +retention sweep was a no-op — the direct cause of the 358k-file backlog that +paralyzed the supervisor's reconcile(). +""" +from __future__ import annotations + +import json +import os +import shutil +import time + +import pytest + +import node_agent + + +@pytest.fixture(autouse=True) +def _isolated_runtime_state(): + """EVENTS_DIR/STATE_DIR are a single temp dir shared for the whole test + session (conftest.py sets RUNTIME_PATH once at import time). Clear them + before and after each test here so tests that write real event/checkpoint + files don't leak state into each other or into other test modules.""" + def _clear(): + if node_agent.EVENTS_DIR.exists(): + shutil.rmtree(node_agent.EVENTS_DIR) + node_agent.EVENTS_DIR.mkdir(parents=True, exist_ok=True) + checkpoint = node_agent.STATE_DIR / "observer_checkpoint.json" + checkpoint.unlink(missing_ok=True) + + _clear() + yield + _clear() + + +def _write_event(events_dir, node, ts, etype, svc="svc"): + """Write an event file whose filename AND on-disk mtime both reflect `ts`. + + _cleanup_control_plane_fs's age check reads f.stat().st_mtime (real + filesystem mtime), not the filename-embedded timestamp — matching + production, where a file's mtime is set once at the moment node-agent + writes it. Tests must set mtime explicitly to simulate an "old" event. + """ + node_dir = events_dir / node + node_dir.mkdir(parents=True, exist_ok=True) + p = node_dir / f"evt-{node}-{ts}-{etype}-{svc}.json" + p.write_text("{}") + os.utime(p, (ts, ts)) + return p + + +def _write_checkpoint(state_dir, node_checkpoints): + (state_dir / "observer_checkpoint.json").write_text( + json.dumps({"node_checkpoints": node_checkpoints}) + ) + + +def test_int_checkpoint_deletes_old_processed_noise(agent): + """The bug: this must NOT raise, and must actually delete eligible files.""" + events_dir = node_agent.EVENTS_DIR + state_dir = node_agent.STATE_DIR + now = int(time.time()) + checkpoint = now - 100 + + old_healthy = _write_event(events_dir, "piha", now - 4 * 86_400, "service_healthy") + old_node_health = _write_event(events_dir, "piha", now - 4 * 86_400, "node_health") + _write_checkpoint(state_dir, {"piha": checkpoint}) + + agent._cleanup_control_plane_fs() + + assert not old_healthy.exists() + assert not old_node_health.exists() + + +def test_keeps_real_signals_regardless_of_age(agent): + """healthcheck_failed / incidents must survive this sweep indefinitely.""" + events_dir = node_agent.EVENTS_DIR + state_dir = node_agent.STATE_DIR + now = int(time.time()) + checkpoint = now - 100 + + healthcheck = _write_event(events_dir, "piha", now - 30 * 86_400, "healthcheck_failed") + ha_event = _write_event(events_dir, "piha", now - 30 * 86_400, "ha_entity_unavailable_long") + _write_checkpoint(state_dir, {"piha": checkpoint}) + + agent._cleanup_control_plane_fs() + + assert healthcheck.exists() + assert ha_event.exists() + + +def test_keeps_unprocessed_noise_even_if_old(agent): + """An event newer than the checkpoint is not guaranteed processed — keep it.""" + events_dir = node_agent.EVENTS_DIR + state_dir = node_agent.STATE_DIR + now = int(time.time()) + checkpoint = now - 100 + + unprocessed = _write_event(events_dir, "piha", now - 4 * 86_400, "service_healthy", svc="future") + # Force it to sort after the checkpoint despite an old mtime by using a + # timestamp greater than checkpoint in the filename itself. + unprocessed.rename(events_dir / "piha" / f"evt-piha-{checkpoint + 50}-service_healthy-future.json") + _write_checkpoint(state_dir, {"piha": checkpoint}) + + agent._cleanup_control_plane_fs() + + assert (events_dir / "piha" / f"evt-piha-{checkpoint + 50}-service_healthy-future.json").exists() + + +def test_keeps_recent_processed_noise_within_three_day_buffer(agent): + """Past checkpoint but within the 3-day safety buffer: not yet eligible.""" + events_dir = node_agent.EVENTS_DIR + state_dir = node_agent.STATE_DIR + now = int(time.time()) + checkpoint = now - 100 + + recent = _write_event(events_dir, "piha", now - 3600, "service_healthy") + _write_checkpoint(state_dir, {"piha": checkpoint}) + + agent._cleanup_control_plane_fs() + + assert recent.exists() + + +def test_legacy_path_string_checkpoint_still_migrates_correctly(agent): + """Old observer builds stored a lexical path string per node; the embedded + timestamp must still be extracted so retention works during a rolling + upgrade instead of silently no-op'ing again.""" + events_dir = node_agent.EVENTS_DIR + state_dir = node_agent.STATE_DIR + now = int(time.time()) + checkpoint_ts = now - 100 + legacy_path_value = f"/opt/homelab/events/piha/evt-piha-{checkpoint_ts}-node_health-node.json" + + old_healthy = _write_event(events_dir, "piha", now - 4 * 86_400, "service_healthy") + _write_checkpoint(state_dir, {"piha": legacy_path_value}) + + agent._cleanup_control_plane_fs() + + assert not old_healthy.exists() + + +def test_no_checkpoint_file_is_noop_not_crash(agent): + events_dir = node_agent.EVENTS_DIR + _write_event(events_dir, "piha", int(time.time()) - 999999, "service_healthy") + # No checkpoint file written at all. + agent._cleanup_control_plane_fs() # must not raise diff --git a/services/node-agent/tests/test_control_plane_health.py b/services/node-agent/tests/test_control_plane_health.py new file mode 100644 index 0000000..889fcb6 --- /dev/null +++ b/services/node-agent/tests/test_control_plane_health.py @@ -0,0 +1,64 @@ +"""Tests for NodeAgent._check_control_plane_health() transition-only emission. + +Regression coverage for the event-flood fix: the VPS control-plane HTTP probe +used to emit service_healthy every single cycle it was reachable, which was +part of the same flood as check_containers()'s per-service confirmation. +""" +from __future__ import annotations + +from unittest.mock import MagicMock + +import node_agent + + +def run_probe(agent, monkeypatch, *, status=200, raise_exc=None): + """Run _check_control_plane_health() with urlopen mocked, capturing emitted events.""" + emitted = [] + + def fake_emit(event_type, severity, service, message, payload=None): + emitted.append((event_type, severity, service, message, payload or {})) + + monkeypatch.setattr(agent, "emit_event", fake_emit) + + if raise_exc is not None: + def fake_urlopen(*a, **kw): + raise raise_exc + else: + resp = MagicMock() + resp.status = status + + def fake_urlopen(*a, **kw): + return resp + + monkeypatch.setattr("urllib.request.urlopen", fake_urlopen) + agent._check_control_plane_health() + return emitted + + +def test_reachable_emits_service_healthy(agent, monkeypatch): + events = run_probe(agent, monkeypatch, status=200) + assert len(events) == 1 + assert events[0][0] == "service_healthy" + assert events[0][2] == "control-plane" + + +def test_reachable_second_cycle_is_silent(agent, monkeypatch): + run_probe(agent, monkeypatch, status=200) + events = run_probe(agent, monkeypatch, status=200) + assert events == [] + + +def test_unreachable_reemits_service_unhealthy_every_cycle(agent, monkeypatch): + """service_unhealthy is a real fault signal — must stay unconditional.""" + run_probe(agent, monkeypatch, status=200) + events = run_probe(agent, monkeypatch, raise_exc=OSError("connection refused")) + assert events[0][0] == "service_unhealthy" + events = run_probe(agent, monkeypatch, raise_exc=OSError("connection refused")) + assert events[0][0] == "service_unhealthy" + + +def test_recovery_after_unreachable_reemits_service_healthy(agent, monkeypatch): + run_probe(agent, monkeypatch, raise_exc=OSError("connection refused")) + events = run_probe(agent, monkeypatch, status=200) + assert len(events) == 1 + assert events[0][0] == "service_healthy"