diff --git a/docs/observer-runtime.md b/docs/observer-runtime.md index fd65653..ab57a79 100644 --- a/docs/observer-runtime.md +++ b/docs/observer-runtime.md @@ -52,6 +52,42 @@ A single global checkpoint (`last_processed_file`) was replaced with this per-no - `node_online`, `node_offline` — update node status in nodes.json - `disk_pressure_*` — set `disk_pressure` field on the node record +## Node Liveness (TTL) + +Node `status` is derived from **freshness**, not from the last status event. +Every reconcile cycle (`_prune_stale_world`, runs even with no new events) the +observer computes `now - last_seen` per node and classifies it: + +| tier | age (always-on) | age (remote / LTE) | status written | panel health | +|---|---|---|---|---| +| fresh | ≤ 180 s | ≤ 900 s | `online` | nominal | +| stale | 180–600 s | 900–3600 s | `stale` | degraded | +| dead | > 600 s | > 3600 s | `offline` | error | +| unknown | `last_seen` missing | — | (unchanged) | (unchanged) | + +Thresholds (tied to the 60 s node-agent heartbeat: fresh = 3× interval) live in +**one place**: `services/control-plane/src/liveness.py`, imported by the observer +and both operator UIs (`compute_liveness` / `ttls_for` / `node_health`). Remote +nodes (topology role `remote`, or `chelsty-*`) use the wider TTLs above. +Override via env: `LIVENESS_TTL_FRESH`, `LIVENESS_TTL_DEAD`, +`LIVENESS_REMOTE_TTL_FRESH`, `LIVENESS_REMOTE_TTL_DEAD`. + +This fixes the "dead node shown NOMINAL" silent outage: previously `status` +stayed `online` forever because the only thing that flipped it to offline was a +`node_offline` event, which a crashed/partitioned node can never emit. + +**Transitions** are not silent. On crossing a boundary the observer writes an +event (`node_stale`, `node_offline`, or `node_online` on recovery) tagged +`source: "observer"` with the affected node in both `node` and +`payload.affected_node`. These are **skipped on re-ingest** (so they never reset +`last_seen`) and routed by the supervisor to `alert_only` actions (Telegram). + +**Read-time safety net:** both operator UIs recompute liveness from `last_seen` +at request time using the same helper, so even a *stalled observer* (frozen +`nodes.json`) still surfaces a dead node. Services inherit their node's liveness +(a service on a dead/stale node is never shown nominal) — computed read-time, +`services.json` is not mutated. + ## Incident Lifecycle 1. **Detection**: A `service_unhealthy` or `healthcheck_failed` event creates or increments an active incident. diff --git a/scripts/observer/observer.py b/scripts/observer/observer.py index dd6ee84..0ecfaa1 100644 --- a/scripts/observer/observer.py +++ b/scripts/observer/observer.py @@ -1,4 +1,5 @@ import os +import sys import json import time import glob @@ -7,6 +8,23 @@ import yaml from datetime import datetime, timezone from pathlib import Path +# Shared liveness logic (thresholds + fresh/stale/dead state machine) lives in +# the control-plane src so the observer and both operator UIs agree on what +# "dead" means. Added to sys.path relative to this file (works both in the +# container, where the repo is mounted at /repo, and in the test tree). +_CP_SRC = Path(__file__).resolve().parent.parent.parent / "services" / "control-plane" / "src" +if str(_CP_SRC) not in sys.path: + sys.path.insert(0, str(_CP_SRC)) +from liveness import ( # noqa: E402 + compute_liveness, + ttls_for, + liveness_to_status, + FRESH, + STALE, + DEAD, + UNKNOWN, +) + def _atomic_write_json(path: Path, data) -> None: """Write JSON atomically: write to a sibling .tmp, fsync, then os.replace.""" @@ -44,9 +62,6 @@ WORLD_DIR = Path(RUNTIME_PATH) / "world" OBSERVER_STATE_FILE = STATE_DIR / "observer_checkpoint.json" FAILED_EVENTS_DIR = STATE_DIR / "observer_failed_events" -# node_health events arrive every 60 s; 5 missed beats → declare offline -NODE_OFFLINE_TTL_SECS = int(os.getenv("NODE_OFFLINE_TTL_SECS", "300")) - REPO_ROOT = Path(__file__).parent.parent.parent INVENTORY_TOPOLOGY = REPO_ROOT / "inventory" / "topology.yaml" @@ -181,6 +196,63 @@ class Observer: except Exception as e: logger.error(f"Failed to save checkpoint: {e}") + def _emit_node_transition(self, node_name, prev, new, node_info, now): + """Write an event when a node crosses a liveness boundary. + + Makes liveness transitions visible (panel event feed) and actionable + (the supervisor routes node_offline/node_stale to an alert). Without + this the outage would be silent — exactly the failure mode being fixed. + + Recovery (stale/dead → fresh) emits node_online so a node coming back is + as visible as it going down. + + The affected node is the top-level ``node`` (so the supervisor can route + on it) AND ``payload.affected_node`` — note events.py::emit_event would + tag node=gethostname() (the observer host, vps), so the observer writes + the event directly instead of using that helper. + + ``source: "observer"`` marks the event so run_once skips re-ingesting it + (re-ingestion would reset the node's last_seen and resurrect it). + """ + if new == DEAD: + etype, severity = "node_offline", "high" + elif new == STALE: + etype, severity = "node_stale", "warning" + elif new == FRESH and prev in (STALE, DEAD): + etype, severity = "node_online", "info" + else: + return # not a transition we alert on + + ts = int(now) + event_id = f"evt-{node_name}-{ts}-{etype}-node" + age = int(now - _parse_ts(node_info.get("last_seen"))) + event = { + "id": event_id, + "timestamp": ts, + "date": datetime.now(timezone.utc).isoformat(), + "type": etype, + "severity": severity, + "node": node_name, + "service": None, + "source": "observer", + "message": f"Node {node_name} liveness {prev} -> {new} (last_seen {age}s ago)", + "payload": { + "affected_node": node_name, + "from": prev, + "to": new, + "last_seen": node_info.get("last_seen"), + "age_secs": age, + }, + } + try: + node_dir = EVENTS_DIR / node_name + node_dir.mkdir(parents=True, exist_ok=True) + _atomic_write_json(node_dir / f"{event_id}.json", event) + logger.warning("Node %s transition %s -> %s (emitted %s)", + node_name, prev, new, etype) + except Exception as exc: + logger.error("Failed to emit node transition for %s: %s", node_name, exc) + def _prune_stale_world(self): """Remove world-state entries for nodes absent from the topology inventory. @@ -227,21 +299,31 @@ class Observer: now = time.time() - # Mark nodes offline when last_seen exceeds NODE_OFFLINE_TTL_SECS. - # Handles crash/network-partition cases where node_offline is never emitted. + # --- Authoritative node liveness (fresh / stale / dead) ------------- + # Status is derived from how long ago the node last reported, NOT from + # the last status event. This runs every cycle (including cycles with + # no new events — see run_once), so a node that silently stops sending + # heartbeats transitions on its own without any node_offline event ever + # being emitted. This is the fix for the "dead node shown NOMINAL" + # class of silent outage. for node_name, node_info in self.world_state["nodes"].items(): - if node_info.get("status") != "online": + roles = (node_info.get("roles") + or self.inventory["nodes"].get(node_name, {}).get("roles", [])) + liveness = compute_liveness( + node_info.get("last_seen"), now=now, ttls=ttls_for(node_name, roles) + ) + if liveness == UNKNOWN: + # No last_seen yet — don't guess a node dead. Leave status as-is. continue - last_seen = _parse_ts(node_info.get("last_seen")) - if last_seen == 0.0: - continue # Never reported — keep status unknown, don't flip - age = now - last_seen - if age > NODE_OFFLINE_TTL_SECS: - logger.warning( - "Node %s last seen %ds ago (> TTL %ds) — marking offline", - node_name, int(age), NODE_OFFLINE_TTL_SECS, - ) - node_info["status"] = "offline" + prev = node_info.get("liveness") + node_info["liveness"] = liveness + new_status = liveness_to_status(liveness) + if new_status: + node_info["status"] = new_status + # Emit on a real transition only. prev is None on the very first + # classification after (re)start — that is a baseline, not an event. + if prev is not None and prev != liveness: + self._emit_node_transition(node_name, prev, liveness, node_info, now) try: # Collect incident_ids currently referenced by any service entry. @@ -535,7 +617,16 @@ class Observer: try: with open(file_path, "r") as f: event = json.load(f) - self.process_event(event) + # Skip events the observer emitted itself (node liveness + # transitions). Re-ingesting them would call process_event, + # which sets last_seen = event timestamp and would resurrect a + # node we just declared dead. They are still consumed by the + # supervisor (alerting) and the panel event feed. + if event.get("source") == "observer": + if file_path > self.node_checkpoints.get(node_dir, ""): + self.node_checkpoints[node_dir] = file_path + continue + self.process_event(event) # Advance per-node checkpoint (only forward — no regression). if file_path > self.node_checkpoints.get(node_dir, ""): self.node_checkpoints[node_dir] = file_path diff --git a/services/agent-system/docker-compose.yml b/services/agent-system/docker-compose.yml index f8ef1c0..dcb0e77 100644 --- a/services/agent-system/docker-compose.yml +++ b/services/agent-system/docker-compose.yml @@ -13,6 +13,10 @@ services: - "18180:8080" volumes: - /opt/homelab:/opt/homelab + # Shared liveness logic — single source of truth, bind-mounted read-only + # so this image does not carry a second copy of the thresholds/tier logic. + # web.py imports it defensively; absence degrades to observer status only. + - ../control-plane/src/liveness.py:/app/liveness.py:ro depends_on: - redis restart: unless-stopped diff --git a/services/agent-system/webui/web.py b/services/agent-system/webui/web.py index 6833da5..a380f6f 100644 --- a/services/agent-system/webui/web.py +++ b/services/agent-system/webui/web.py @@ -5,6 +5,18 @@ from datetime import datetime, timezone from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path +# Shared liveness logic — the SAME module the observer and operator_ui use, so +# all three agree on what "dead" means. This image only copies web.py + +# index.html, so liveness.py is bind-mounted in at /app/liveness.py (see this +# service's docker-compose.yml). Imported defensively: if the mount is missing +# we fall back to trusting the observer-written status rather than crashing the +# whole panel — the observer remains authoritative either way. +try: + from liveness import compute_liveness, ttls_for, node_health as _liveness_health, degrade_for_node + _LIVENESS_OK = True +except Exception: # pragma: no cover - only when liveness.py mount is absent + _LIVENESS_OK = False + STATE_DIR = Path(os.getenv("HOMELAB_STATE_ROOT", "/opt/homelab/state")) EVENTS_DIR = Path(os.getenv("HOMELAB_EVENTS_ROOT", "/opt/homelab/events")) @@ -47,12 +59,78 @@ def save_config(config): (STATE_DIR / "operator-config.json").write_text(json.dumps(config, indent=2)) +def _node_liveness_map(): + """Return {node_name: liveness_tier} from nodes.json (empty if unavailable).""" + if not _LIVENESS_OK: + return {} + raw = read_json_file(WORLD_DIR / "nodes.json", default={}) + if not isinstance(raw, dict): + return {} + return { + name: compute_liveness(info.get("last_seen"), + ttls=ttls_for(name, info.get("roles", []))) + for name, info in raw.items() + } + + def current_nodes(): - return read_json_file(WORLD_DIR / "nodes.json") + """Nodes shaped as a list with a freshness-aware health field. + + nodes.json is a dict keyed by name; index.html does nodes.map() and reads + node.health, so we shape to a list and compute health here. The health is + the read-time safety net (worse of persisted status and last_seen + freshness) so a stalled observer cannot make a dead node look nominal. + """ + raw = read_json_file(WORLD_DIR / "nodes.json", default={}) + if isinstance(raw, list): + return raw + result = [] + for name, info in raw.items(): + health = _liveness_health(info, name=name) if _LIVENESS_OK else ( + "error" if info.get("status") == "offline" + else "nominal" if info.get("status") == "online" else info.get("status", "unknown") + ) + result.append({ + "id": name, + "hostname": name, + "health": health, + "status": info.get("status", "unknown"), + "capabilities": info.get("roles", []), + "connectivity": "tailscale", + "incidents": 0, + "last_seen": info.get("last_seen"), + "disk_usage_pct": info.get("disk_usage_pct"), + "mem_usage_pct": info.get("mem_usage_pct"), + "cpu_usage_pct": info.get("cpu_usage_pct"), + "disk_pressure": info.get("disk_pressure"), + }) + return result def current_services(): - return read_json_file(WORLD_DIR / "services.json") + """Services shaped as a list, with the node-liveness cascade (variant B).""" + raw = read_json_file(WORLD_DIR / "services.json", default={}) + if isinstance(raw, list): + return raw + node_liveness = _node_liveness_map() + result = [] + for key, info in raw.items(): + svc_status = info.get("status", "unknown") + health = ("nominal" if svc_status == "healthy" + else ("error" if svc_status == "unhealthy" else svc_status)) + node_name = info.get("node", "") + if _LIVENESS_OK and node_name in node_liveness: + health = degrade_for_node(health, node_liveness[node_name]) + result.append({ + "id": key, + "name": info.get("service", key), + "node": node_name, + "health": health, + "actual_state": svc_status, + "last_check": info.get("last_check"), + "incident_id": info.get("incident_id"), + }) + return result def current_deployments(): diff --git a/services/control-plane/src/liveness.py b/services/control-plane/src/liveness.py new file mode 100644 index 0000000..212604a --- /dev/null +++ b/services/control-plane/src/liveness.py @@ -0,0 +1,151 @@ +"""Shared node-liveness logic — single source of truth for freshness. + +This module owns the fresh/stale/dead state machine and its TTL thresholds. +It is imported by every component that has to decide whether a node is alive: + + - scripts/observer/observer.py authoritative writer of world/nodes.json + - services/control-plane/src/operator_ui.py read-time safety net (panel) + - services/agent-system/webui/web.py read-time safety net (panel); + this file is bind-mounted into the webui image at /app/liveness.py — + see services/agent-system/docker-compose.yml. + +Why a single module: the previous bug ("dead node shown NOMINAL") was a missing +TTL. Thresholds and the tier logic must live in exactly one place so the +observer and both panels can never disagree about what "dead" means. + +Heartbeat interval is 60 s on every node (node-agent CHECK_INTERVAL=60), so the +fresh ceiling is 3× the interval. If the interval ever changes, raise the +thresholds here proportionally and nothing else needs to move. +""" +from __future__ import annotations + +import os +import time +from datetime import datetime + +# --- Liveness tiers --------------------------------------------------------- +FRESH = "fresh" +STALE = "stale" +DEAD = "dead" +UNKNOWN = "unknown" + +# --- TTL thresholds (seconds) ---------------------------------------------- +# Always-on nodes (vps/piha/solaria/lustro): heartbeat every 60 s. +# fresh ≤ 180 s → 3 missed beats tolerated (+ observer cycle + rsync latency) +# dead > 600 s → 10 missed beats == really gone +DEFAULT_TTLS = { + "fresh": int(os.getenv("LIVENESS_TTL_FRESH", "180")), + "dead": int(os.getenv("LIVENESS_TTL_DEAD", "600")), +} +# Remote / LTE nodes (chelsty-*): intermittent uplink, events buffer locally +# and ship on reconnect. Wider so a typical multi-minute LTE blip reads as +# stale (degraded), but a genuine outage still escalates to dead within an hour. +REMOTE_TTLS = { + "fresh": int(os.getenv("LIVENESS_REMOTE_TTL_FRESH", "900")), # 15 min + "dead": int(os.getenv("LIVENESS_REMOTE_TTL_DEAD", "3600")), # 1 h +} +# A node uses REMOTE_TTLS if its topology role is in REMOTE_ROLES, or its name +# is in REMOTE_NODES (fallback when roles are absent from the node record). +REMOTE_ROLES = {"remote"} +REMOTE_NODES = {"chelsty-infra", "chelsty-ha"} + +# --- Mappings --------------------------------------------------------------- +# Persisted node.status string written by the observer for each tier. +# UNKNOWN intentionally has no mapping: never overwrite status when we have no +# timing data (last_seen missing) — we don't guess a node dead. +_LIVENESS_TO_STATUS = {FRESH: "online", STALE: "stale", DEAD: "offline"} +# UI health bucket per liveness tier and per persisted status. +_LIVENESS_HEALTH = {FRESH: "nominal", STALE: "degraded", DEAD: "error", UNKNOWN: "unknown"} +_STATUS_HEALTH = {"online": "nominal", "stale": "degraded", "offline": "error"} +# Severity ordering for "take the worse of two health buckets". +_HEALTH_RANK = {"nominal": 0, "unknown": 1, "degraded": 2, "error": 3} + + +def parse_ts(ts): + """Epoch float from int/float or ISO-8601 string; 0.0 on None/garbage. + + node-agent emits int(time.time()); events.py / stability-agent emit ISO + strings. Both land in last_seen, so all arithmetic goes through here. + """ + if ts is None: + return 0.0 + if isinstance(ts, (int, float)): + return float(ts) + try: + return datetime.fromisoformat(str(ts).replace("Z", "+00:00")).timestamp() + except Exception: + return 0.0 + + +def ttls_for(node_name=None, roles=None): + """Pick the TTL set for a node by its name or topology roles.""" + roles = set(roles or ()) + if (node_name in REMOTE_NODES) or (roles & REMOTE_ROLES): + return REMOTE_TTLS + return DEFAULT_TTLS + + +def compute_liveness(last_seen, now=None, ttls=None): + """Classify a node from age = now - last_seen. + + Returns FRESH / STALE / DEAD, or UNKNOWN when last_seen is missing or + unparseable (no data → no opinion; callers must not flip status on UNKNOWN). + """ + ttls = ttls or DEFAULT_TTLS + ts = parse_ts(last_seen) + if ts <= 0: + return UNKNOWN + age = (now if now is not None else time.time()) - ts + if age <= ttls["fresh"]: + return FRESH + if age <= ttls["dead"]: + return STALE + return DEAD + + +def liveness_to_status(liveness): + """Persisted node.status for a tier, or None for UNKNOWN (leave as-is).""" + return _LIVENESS_TO_STATUS.get(liveness) + + +def _worse(a, b): + """Return the more severe of two UI health buckets.""" + return a if _HEALTH_RANK.get(a, 1) >= _HEALTH_RANK.get(b, 1) else b + + +def node_health(info, name=None, now=None): + """UI health bucket for a node record, applying the read-time freshness net. + + Takes the WORSE of (a) the observer-persisted status and (b) the freshness + derived from last_seen. Rationale: + - if the observer is alive it already wrote the right status; freshness + agrees and this is a no-op. + - if the observer itself is dead, nodes.json freezes with stale "online" + statuses — but last_seen keeps ageing, so freshness still surfaces the + outage. This is the whole point of duplicating the check at read time. + UNKNOWN freshness (no last_seen) never downgrades a node. + """ + stored_health = _STATUS_HEALTH.get(info.get("status", "unknown"), "unknown") + liveness = compute_liveness( + info.get("last_seen"), now=now, ttls=ttls_for(name, info.get("roles", [])) + ) + if liveness == UNKNOWN: + health = stored_health + else: + health = _worse(stored_health, _LIVENESS_HEALTH[liveness]) + if health == "nominal" and info.get("disk_pressure") == "high": + return "degraded" + return health + + +def degrade_for_node(service_health, node_liveness): + """Cascade node liveness onto a service's health (read-time, panel side). + + A service cannot be healthier than the node it runs on: a service on a dead + node is error, on a stale node at least degraded. + """ + if node_liveness == DEAD: + return "error" + if node_liveness == STALE: + return _worse(service_health, "degraded") + return service_health diff --git a/services/control-plane/src/operator_ui.py b/services/control-plane/src/operator_ui.py index bad983f..49ab73e 100644 --- a/services/control-plane/src/operator_ui.py +++ b/services/control-plane/src/operator_ui.py @@ -7,6 +7,10 @@ from datetime import datetime from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path +# Shared liveness logic (same module the observer uses). This file lives next +# to operator_ui.py in the control-plane image, so a plain import works. +from liveness import compute_liveness, ttls_for, node_health as _liveness_health, degrade_for_node + STATE_DIR = Path(os.getenv("HOMELAB_STATE_ROOT", "/opt/homelab/state")) EVENTS_DIR = Path(os.getenv("HOMELAB_EVENTS_ROOT", "/opt/homelab/events")) @@ -55,15 +59,15 @@ EVENTS_MAX_AGE_HOURS = int(os.getenv("EVENTS_MAX_AGE_HOURS", "24")) EVENTS_MAX_COUNT = int(os.getenv("EVENTS_MAX_COUNT", "200")) -def _node_health(info): - status = info.get("status", "unknown") - if status == "offline": - return "error" - if info.get("disk_pressure") == "high": - return "degraded" - if status == "online": - return "nominal" - return status +def _node_health(info, name=None): + """UI health bucket for a node, with the read-time freshness safety net. + + Delegates to the shared liveness helper: takes the worse of the observer's + persisted status and the freshness derived from last_seen, so a stalled + observer (frozen nodes.json) still surfaces a dead node as error instead of + nominal. Thresholds live in liveness.py — never here. + """ + return _liveness_health(info, name=name) def current_nodes(): @@ -81,7 +85,7 @@ def current_nodes(): result.append({ "id": name, "hostname": name, - "health": _node_health(info), + "health": _node_health(info, name), "status": info.get("status", "unknown"), "capabilities": info.get("roles", []), "connectivity": "tailscale", @@ -104,16 +108,34 @@ def current_services(): raw = read_json_file(WORLD_DIR / "services.json", default={}) if isinstance(raw, list): return raw + + # Read-time cascade (variant B): a service cannot be healthier than the node + # it runs on. Compute each node's liveness from nodes.json (NOT mutating + # it) and degrade the service health accordingly. Without this, a service + # on a dead node would still read "nominal" from its last health check. + nodes_raw = read_json_file(WORLD_DIR / "nodes.json", default={}) + node_liveness = {} + if isinstance(nodes_raw, dict): + for n_name, n_info in nodes_raw.items(): + node_liveness[n_name] = compute_liveness( + n_info.get("last_seen"), + ttls=ttls_for(n_name, n_info.get("roles", [])), + ) + result = [] for key, info in raw.items(): svc_status = info.get("status", "unknown") + health = ("nominal" if svc_status == "healthy" + else ("error" if svc_status == "unhealthy" + else svc_status)) + node_name = info.get("node", "") + if node_name in node_liveness: + health = degrade_for_node(health, node_liveness[node_name]) result.append({ "id": key, "name": info.get("service", key), - "node": info.get("node", ""), - "health": ("nominal" if svc_status == "healthy" - else ("error" if svc_status == "unhealthy" - else svc_status)), + "node": node_name, + "health": health, "desired_state": "running", "actual_state": svc_status, "deployment_state": "deployed", diff --git a/services/control-plane/src/supervisor.py b/services/control-plane/src/supervisor.py index a7db0b4..a9a4d05 100644 --- a/services/control-plane/src/supervisor.py +++ b/services/control-plane/src/supervisor.py @@ -85,6 +85,15 @@ HA_ALERT_COOLDOWN = 3600 # within this window — HA is in a planned restart/update and alerts would be noise. HA_TRANSITION_WINDOW = 300 # 5 minutes +# --------------------------------------------------------------------------- +# Node liveness event routing (observer-emitted node_offline/node_stale/node_online) +# --------------------------------------------------------------------------- +# A node we cannot reach cannot be auto-remediated (you can't docker-restart a +# host that is offline), so these are alert-only — they exist to make a silent +# outage loud. node_online is the recovery notice. +NODE_ALERT_EVENTS = {"node_offline", "node_stale", "node_online"} +NODE_ALERT_COOLDOWN = 3600 # 1-hour cooldown to avoid repeated Telegram noise + # When True, events that would generate container_restart are downgraded to alert_only # with a "[SHADOW MODE]" note. Safe default for initial deployment; set # HA_DIAG_SHADOW_MODE=false on the control-plane node when ready for live actions. @@ -539,7 +548,11 @@ class Supervisor: except Exception as e: logger.debug(f"Could not read event {event_file}: {e}") continue - if not event.get("type", "").startswith("ha_"): + etype = event.get("type", "") + if etype in NODE_ALERT_EVENTS: + self._route_node_event(event) + continue + if not etype.startswith("ha_"): continue self._route_ha_event(event) @@ -726,6 +739,49 @@ class Supervisor: } self._write_pending_action(action) + def _route_node_event(self, event: dict): + """Route an observer-emitted node liveness event to an alert_only action. + + node_offline / node_stale make a silent outage loud; node_online is the + recovery notice. No auto-remediation — an unreachable node can't be + restarted from here. Dedup via stable action_id + cooldown, mirroring + the HA alert path. + """ + event_type = event.get("type", "") + node = event.get("node") or event.get("payload", {}).get("affected_node") + if not node: + return + + action_id = f"alert-{event_type.replace('_', '-')}-{node}" + + for state in ("pending", "approved", "running"): + if (ACTIONS_DIR / state / f"{action_id}.json").exists(): + logger.debug(f"Skipping {action_id}: already in state '{state}'") + return + + if self._ha_action_recently_completed(action_id, NODE_ALERT_COOLDOWN): + logger.debug(f"Skipping {action_id}: within {NODE_ALERT_COOLDOWN}s cooldown") + return + + payload = dict(event.get("payload", {})) + payload["reason"] = event_type + + action = { + "action_id": action_id, + "timestamp": time.time(), + "type": "alert_only", + "node": node, + "service": None, + "risk_level": "info", + "confidence": 1.0, + "description": event.get( + "message", f"Node liveness alert: {event_type} on {node}" + ), + "status": "pending", + "payload": payload, + } + self._write_pending_action(action) + def _cancel_ha_container_restart(self, node: str): """Move a pending ha_websocket_dead container_restart to cancelled on recovery.""" action_id = f"container-restart-{node}-homeassistant" diff --git a/services/control-plane/tests/test_incident_lifecycle.py b/services/control-plane/tests/test_incident_lifecycle.py index 4960519..198f215 100644 --- a/services/control-plane/tests/test_incident_lifecycle.py +++ b/services/control-plane/tests/test_incident_lifecycle.py @@ -380,90 +380,193 @@ def test_run_once_quarantines_bad_event_and_processes_next_for_same_node(tmp_pat # --------------------------------------------------------------------------- -# 7. Node offline TTL — liveness detection via last_seen age +# 7. Node liveness — 3-state (fresh/stale/dead) authoritative classification # --------------------------------------------------------------------------- +# Default TTLs (liveness.py): fresh <=180s, dead >600s. "vps" is in the test +# topology so it survives stale-node pruning. -def test_prune_marks_online_node_offline_after_ttl(tmp_path): - """Node online with last_seen > NODE_OFFLINE_TTL_SECS must be flipped to offline.""" +def _node_events(obs_mod, node="vps"): + """List event files the observer emitted into / during a prune.""" + d = obs_mod.EVENTS_DIR / node + return sorted(d.glob("*.json")) if d.exists() else [] + + +def test_prune_keeps_fresh_node_online(tmp_path): obs = _make_observer_simple(tmp_path) - import observer.observer as obs_mod - - # Use "vps" — present in the test topology so it won't be pruned as stale. obs.world_state["nodes"]["vps"] = { - "status": "online", - "last_seen": time.time() - (obs_mod.NODE_OFFLINE_TTL_SECS + 60), # over TTL - "roles": [], + "status": "online", "last_seen": time.time() - 60, "roles": [], } - obs._prune_stale_world() - - assert obs.world_state["nodes"]["vps"]["status"] == "offline" - - -def test_prune_does_not_mark_node_offline_within_ttl(tmp_path): - """Node online with recent last_seen must stay online.""" - obs = _make_observer_simple(tmp_path) - import observer.observer as obs_mod - - obs.world_state["nodes"]["vps"] = { - "status": "online", - "last_seen": time.time() - (obs_mod.NODE_OFFLINE_TTL_SECS - 60), # within TTL - "roles": [], - } - - obs._prune_stale_world() - assert obs.world_state["nodes"]["vps"]["status"] == "online" + assert obs.world_state["nodes"]["vps"]["liveness"] == "fresh" -def test_prune_does_not_flip_already_offline_node(tmp_path): - """Node already offline must not be changed by TTL logic.""" +def test_prune_marks_node_stale_between_ttls(tmp_path): + """180s < age <= 600s → stale/degraded, not yet dead.""" obs = _make_observer_simple(tmp_path) - obs.world_state["nodes"]["vps"] = { - "status": "offline", - "last_seen": time.time() - 9999, - "roles": [], + "status": "online", "last_seen": time.time() - 300, "roles": [], } - obs._prune_stale_world() + assert obs.world_state["nodes"]["vps"]["status"] == "stale" + assert obs.world_state["nodes"]["vps"]["liveness"] == "stale" + +def test_prune_marks_node_dead_after_down_ttl(tmp_path): + """age > 600s → dead/offline.""" + obs = _make_observer_simple(tmp_path) + obs.world_state["nodes"]["vps"] = { + "status": "online", "last_seen": time.time() - 700, "roles": [], + } + obs._prune_stale_world() assert obs.world_state["nodes"]["vps"]["status"] == "offline" + assert obs.world_state["nodes"]["vps"]["liveness"] == "dead" -def test_prune_does_not_flip_never_seen_node(tmp_path): - """Node with last_seen=None must stay in its original status.""" - obs = _make_observer_simple(tmp_path) - - obs.world_state["nodes"]["vps"] = { - "status": "online", - "last_seen": None, - "roles": [], - } - - obs._prune_stale_world() - - # last_seen=None means we have no data — don't flip to offline - assert obs.world_state["nodes"]["vps"]["status"] == "online" - - -def test_prune_marks_node_offline_with_iso_last_seen(tmp_path): - """TTL check must work when last_seen is an ISO-8601 string.""" +def test_prune_unknown_when_no_last_seen(tmp_path): + """last_seen=None → UNKNOWN: status untouched, no liveness flip, no event.""" obs = _make_observer_simple(tmp_path) import observer.observer as obs_mod + obs.world_state["nodes"]["vps"] = { + "status": "online", "last_seen": None, "roles": [], + } + obs._prune_stale_world() + assert obs.world_state["nodes"]["vps"]["status"] == "online" + assert "liveness" not in obs.world_state["nodes"]["vps"] + assert _node_events(obs_mod) == [] + +def test_prune_marks_node_dead_with_iso_last_seen(tmp_path): + """3-state classification works when last_seen is an ISO-8601 string.""" + obs = _make_observer_simple(tmp_path) from datetime import datetime, timezone stale_ts = datetime.fromtimestamp( - time.time() - (obs_mod.NODE_OFFLINE_TTL_SECS + 120), - tz=timezone.utc, + time.time() - 700, tz=timezone.utc ).isoformat() - obs.world_state["nodes"]["vps"] = { - "status": "online", - "last_seen": stale_ts, - "roles": [], + "status": "online", "last_seen": stale_ts, "roles": [], } + obs._prune_stale_world() + assert obs.world_state["nodes"]["vps"]["status"] == "offline" + +# --- Transition emission -------------------------------------------------- + +def test_transition_emits_node_offline_event(tmp_path): + """fresh -> dead emits a node_offline event tagged source=observer.""" + obs = _make_observer_simple(tmp_path) + import observer.observer as obs_mod + obs.world_state["nodes"]["vps"] = { + "status": "online", "liveness": "fresh", + "last_seen": time.time() - 700, "roles": [], + } obs._prune_stale_world() + events = _node_events(obs_mod) + assert len(events) == 1 + ev = json.loads(events[0].read_text()) + assert ev["type"] == "node_offline" + assert ev["node"] == "vps" + assert ev["source"] == "observer" + assert ev["payload"]["affected_node"] == "vps" + assert ev["payload"]["to"] == "dead" + + +def test_transition_emits_recovery_on_dead_to_fresh(tmp_path): + """dead -> fresh emits node_online (recovery is as visible as the fall).""" + obs = _make_observer_simple(tmp_path) + import observer.observer as obs_mod + obs.world_state["nodes"]["vps"] = { + "status": "offline", "liveness": "dead", + "last_seen": time.time() - 10, "roles": [], + } + obs._prune_stale_world() + + assert obs.world_state["nodes"]["vps"]["status"] == "online" + events = _node_events(obs_mod) + assert len(events) == 1 + ev = json.loads(events[0].read_text()) + assert ev["type"] == "node_online" + assert ev["payload"]["from"] == "dead" + assert ev["payload"]["to"] == "fresh" + + +def test_no_emission_on_baseline_classification(tmp_path): + """First classification (prev liveness=None) is a baseline, not a transition.""" + obs = _make_observer_simple(tmp_path) + import observer.observer as obs_mod + obs.world_state["nodes"]["vps"] = { + "status": "online", "last_seen": time.time() - 700, "roles": [], + } + obs._prune_stale_world() + assert obs.world_state["nodes"]["vps"]["status"] == "offline" # still classified + assert _node_events(obs_mod) == [] # but no event + + +def test_no_emission_when_liveness_unchanged(tmp_path): + """Still-dead node must not re-emit on every cycle.""" + obs = _make_observer_simple(tmp_path) + import observer.observer as obs_mod + obs.world_state["nodes"]["vps"] = { + "status": "offline", "liveness": "dead", + "last_seen": time.time() - 5000, "roles": [], + } + obs._prune_stale_world() + assert _node_events(obs_mod) == [] + + +def test_remote_node_uses_wider_ttls(tmp_path): + """A remote-role node at age 700s is still only stale (remote dead TTL=3600).""" + obs = _make_observer_simple(tmp_path) + import observer.observer as obs_mod + # Add chelsty-infra to topology so it survives stale-node pruning. + obs_mod.INVENTORY_TOPOLOGY.write_text( + "nodes:\n" + " vps:\n roles: [control-plane]\n connectivity: {}\n" + " chelsty-infra:\n roles: [remote, infra]\n connectivity: {}\n" + ) + obs.inventory = obs._load_inventory() + obs.world_state["nodes"]["chelsty-infra"] = { + "status": "online", "last_seen": time.time() - 1200, + "roles": ["remote", "infra"], + } + obs._prune_stale_world() + # 1200s > default dead (600) but < remote dead (3600) → stale, not dead. + assert obs.world_state["nodes"]["chelsty-infra"]["liveness"] == "stale" + + +# --- Observer skips re-ingesting its own transition events ---------------- + +def test_run_once_skips_observer_emitted_events(tmp_path): + """An observer-source event must NOT be re-ingested (would reset last_seen).""" + obs = _make_observer_simple(tmp_path) + import observer.observer as obs_mod + + vps_dir = obs_mod.EVENTS_DIR / "vps" + vps_dir.mkdir(parents=True, exist_ok=True) + # Seed world with a dead vps so we can detect an unwanted resurrection. + obs.world_state["nodes"]["vps"] = { + "status": "offline", "liveness": "dead", + "last_seen": time.time() - 5000, "roles": [], + } + # An observer-emitted node_online event with a fresh timestamp — if it were + # ingested, process_event would set last_seen=now and flip vps back online. + ev_path = vps_dir / "evt-vps-9999999999-node_online-node.json" + ev_path.write_text(json.dumps({ + "id": "evt-vps-9999999999-node_online-node", + "timestamp": int(time.time()), + "type": "node_online", + "severity": "info", + "node": "vps", + "service": None, + "source": "observer", + "message": "synthetic", + "payload": {"affected_node": "vps"}, + })) + + obs.run_once() + + # Not ingested: last_seen stays old → vps stays dead/offline. assert obs.world_state["nodes"]["vps"]["status"] == "offline" + # Checkpoint still advanced over the skipped file. + assert obs.node_checkpoints.get("vps") == str(ev_path) diff --git a/services/control-plane/tests/test_liveness.py b/services/control-plane/tests/test_liveness.py new file mode 100644 index 0000000..183050a --- /dev/null +++ b/services/control-plane/tests/test_liveness.py @@ -0,0 +1,137 @@ +"""Unit tests for the shared liveness helper (thresholds + tier state machine).""" +from __future__ import annotations + +import sys +import time +from datetime import datetime, timezone +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +import liveness +from liveness import ( + compute_liveness, + ttls_for, + liveness_to_status, + node_health, + degrade_for_node, + parse_ts, + FRESH, + STALE, + DEAD, + UNKNOWN, +) + + +# --- parse_ts --------------------------------------------------------------- + +def test_parse_ts_int_float_iso_none(): + assert abs(parse_ts(1000) - 1000.0) < 0.001 + assert abs(parse_ts(1000.5) - 1000.5) < 0.001 + iso = "2026-06-01T00:00:00Z" + assert abs(parse_ts(iso) - datetime(2026, 6, 1, tzinfo=timezone.utc).timestamp()) < 1 + assert parse_ts(None) == 0.0 + assert parse_ts("garbage") == 0.0 + + +# --- compute_liveness (default TTLs: fresh<=180, dead>600) ------------------ + +def test_compute_liveness_fresh(): + assert compute_liveness(time.time() - 60) == FRESH + + +def test_compute_liveness_stale(): + assert compute_liveness(time.time() - 300) == STALE + + +def test_compute_liveness_dead(): + assert compute_liveness(time.time() - 700) == DEAD + + +def test_compute_liveness_boundaries(): + now = 1_000_000 + assert compute_liveness(now - 180, now=now) == FRESH # exactly fresh ceiling + assert compute_liveness(now - 181, now=now) == STALE + assert compute_liveness(now - 600, now=now) == STALE # exactly dead floor + assert compute_liveness(now - 601, now=now) == DEAD + + +def test_compute_liveness_unknown_on_missing(): + assert compute_liveness(None) == UNKNOWN + assert compute_liveness("garbage") == UNKNOWN + assert compute_liveness(0) == UNKNOWN + + +def test_compute_liveness_iso_string(): + iso = datetime.fromtimestamp(time.time() - 700, tz=timezone.utc).isoformat() + assert compute_liveness(iso) == DEAD + + +# --- ttls_for --------------------------------------------------------------- + +def test_ttls_for_default_vs_remote(): + assert ttls_for("vps", ["control-plane"]) == liveness.DEFAULT_TTLS + assert ttls_for("chelsty-infra", []) == liveness.REMOTE_TTLS # by name + assert ttls_for("anything", ["remote"]) == liveness.REMOTE_TTLS # by role + + +def test_remote_node_stale_where_default_would_be_dead(): + now = 1_000_000 + ts = now - 1200 # > default dead (600) and > remote fresh (900), < remote dead (3600) + assert compute_liveness(ts, now=now, ttls=ttls_for("chelsty-infra")) == STALE + assert compute_liveness(ts, now=now, ttls=ttls_for("vps")) == DEAD + + +# --- liveness_to_status ----------------------------------------------------- + +def test_liveness_to_status(): + assert liveness_to_status(FRESH) == "online" + assert liveness_to_status(STALE) == "stale" + assert liveness_to_status(DEAD) == "offline" + assert liveness_to_status(UNKNOWN) is None + + +# --- node_health (read-time safety net) ------------------------------------- + +def test_node_health_fresh_online_is_nominal(): + info = {"status": "online", "last_seen": time.time() - 30, "roles": []} + assert node_health(info, "vps") == "nominal" + + +def test_node_health_frozen_online_but_dead_is_error(): + """Observer dead → status frozen 'online' but last_seen old → error (the net).""" + info = {"status": "online", "last_seen": time.time() - 700, "roles": []} + assert node_health(info, "vps") == "error" + + +def test_node_health_stale_is_degraded(): + info = {"status": "online", "last_seen": time.time() - 300, "roles": []} + assert node_health(info, "vps") == "degraded" + + +def test_node_health_unknown_freshness_does_not_downgrade(): + info = {"status": "online", "last_seen": None, "roles": []} + assert node_health(info, "vps") == "nominal" + + +def test_node_health_disk_pressure_degrades_fresh_node(): + info = {"status": "online", "last_seen": time.time() - 30, + "roles": [], "disk_pressure": "high"} + assert node_health(info, "vps") == "degraded" + + +def test_node_health_takes_worse_of_status_and_freshness(): + # status already offline, freshness fresh → still error (worse wins) + info = {"status": "offline", "last_seen": time.time() - 5, "roles": []} + assert node_health(info, "vps") == "error" + + +# --- degrade_for_node (service cascade) ------------------------------------- + +def test_degrade_for_node(): + assert degrade_for_node("nominal", DEAD) == "error" + assert degrade_for_node("nominal", STALE) == "degraded" + assert degrade_for_node("nominal", FRESH) == "nominal" + # never upgrades a service that is already worse + assert degrade_for_node("error", STALE) == "error" + assert degrade_for_node("error", FRESH) == "error" diff --git a/services/control-plane/tests/test_operator_ui_liveness.py b/services/control-plane/tests/test_operator_ui_liveness.py new file mode 100644 index 0000000..0b95fbc --- /dev/null +++ b/services/control-plane/tests/test_operator_ui_liveness.py @@ -0,0 +1,77 @@ +"""Read-time liveness safety net + service cascade in the operator UI backend.""" +from __future__ import annotations + +import json +import sys +import time +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +import operator_ui + + +def _setup_world(tmp_path, monkeypatch, nodes, services): + world = tmp_path / "world" + world.mkdir(parents=True, exist_ok=True) + (world / "nodes.json").write_text(json.dumps(nodes)) + (world / "services.json").write_text(json.dumps(services)) + monkeypatch.setattr(operator_ui, "WORLD_DIR", world) + + +def test_current_nodes_fresh_is_nominal(tmp_path, monkeypatch): + _setup_world(tmp_path, monkeypatch, + {"vps": {"status": "online", "last_seen": time.time() - 30, "roles": []}}, + {}) + nodes = operator_ui.current_nodes() + assert nodes[0]["health"] == "nominal" + + +def test_current_nodes_frozen_online_but_dead_is_error(tmp_path, monkeypatch): + """The bug: observer wrote 'online' but last_seen is ancient → must be error.""" + _setup_world(tmp_path, monkeypatch, + {"chelsty-infra": {"status": "online", + "last_seen": time.time() - 16 * 86400, + "roles": ["remote"]}}, + {}) + nodes = operator_ui.current_nodes() + assert nodes[0]["health"] == "error" + + +def test_current_nodes_stale_is_degraded(tmp_path, monkeypatch): + _setup_world(tmp_path, monkeypatch, + {"piha": {"status": "online", "last_seen": time.time() - 300, "roles": []}}, + {}) + nodes = operator_ui.current_nodes() + assert nodes[0]["health"] == "degraded" + + +def test_service_cascade_dead_node_makes_service_error(tmp_path, monkeypatch): + """A 'healthy' service on a dead node must read error (cascade, read-time).""" + _setup_world( + tmp_path, monkeypatch, + {"piha": {"status": "online", "last_seen": time.time() - 700, "roles": []}}, + {"piha/vikunja": {"node": "piha", "service": "vikunja", "status": "healthy"}}, + ) + svcs = operator_ui.current_services() + assert svcs[0]["health"] == "error" + + +def test_service_cascade_stale_node_makes_service_degraded(tmp_path, monkeypatch): + _setup_world( + tmp_path, monkeypatch, + {"piha": {"status": "online", "last_seen": time.time() - 300, "roles": []}}, + {"piha/vikunja": {"node": "piha", "service": "vikunja", "status": "healthy"}}, + ) + svcs = operator_ui.current_services() + assert svcs[0]["health"] == "degraded" + + +def test_service_on_fresh_node_stays_nominal(tmp_path, monkeypatch): + _setup_world( + tmp_path, monkeypatch, + {"vps": {"status": "online", "last_seen": time.time() - 30, "roles": []}}, + {"vps/outline": {"node": "vps", "service": "outline", "status": "healthy"}}, + ) + svcs = operator_ui.current_services() + assert svcs[0]["health"] == "nominal" diff --git a/services/control-plane/tests/test_supervisor_ha.py b/services/control-plane/tests/test_supervisor_ha.py index 78d7b12..a0c98b0 100644 --- a/services/control-plane/tests/test_supervisor_ha.py +++ b/services/control-plane/tests/test_supervisor_ha.py @@ -384,7 +384,9 @@ def test_non_ha_events_not_routed(tmp_path, monkeypatch): sup = _setup_supervisor(tmp_path, monkeypatch) events_dir = tmp_path / "events" - for etype in ("service_unhealthy", "containers_not_running", "node_online", "deployment_failed"): + # node_online/node_offline/node_stale are now routed (node-liveness alerts), + # so use only events that are neither ha_* nor node-liveness events here. + for etype in ("service_unhealthy", "containers_not_running", "node_health", "deployment_failed"): e = _make_event(etype, service="mosquitto") e["type"] = etype _write_event(events_dir, e) diff --git a/services/control-plane/tests/test_supervisor_node_events.py b/services/control-plane/tests/test_supervisor_node_events.py new file mode 100644 index 0000000..659147a --- /dev/null +++ b/services/control-plane/tests/test_supervisor_node_events.py @@ -0,0 +1,122 @@ +"""Supervisor routing of observer-emitted node liveness events → alert_only.""" +from __future__ import annotations + +import json +import sys +import time +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) +import supervisor as supervisor_module +from supervisor import Supervisor + + +def _setup_supervisor(tmp_path, monkeypatch): + actions = tmp_path / "actions" + events = tmp_path / "events" + world = tmp_path / "world" + repo = tmp_path / "repo" + state = tmp_path / "state" + for d in (actions, events, world, repo / "hosts", state): + d.mkdir(parents=True, exist_ok=True) + monkeypatch.setattr(supervisor_module, "ACTIONS_DIR", actions) + monkeypatch.setattr(supervisor_module, "EVENTS_DIR", events) + monkeypatch.setattr(supervisor_module, "WORLD_DIR", world) + monkeypatch.setattr(supervisor_module, "REPO_ROOT", repo) + sup = Supervisor() + sup.desired_state = {"services": {}} + sup.actual_state = {"services": {}, "nodes": {}, "incidents": {}} + return sup + + +def _node_event(node, etype): + return { + "id": f"evt-{node}-{int(time.time())}-{etype}-node", + "type": etype, + "node": node, + "service": None, + "severity": "high", + "timestamp": int(time.time()), + "source": "observer", + "message": f"Node {node} {etype}", + "payload": {"affected_node": node, "from": "fresh", "to": "dead"}, + } + + +def _write(events_dir, event): + p = events_dir / f"{event['id']}.json" + p.write_text(json.dumps(event)) + return p + + +def _pending(tmp_path, action_id): + return tmp_path / "actions" / "pending" / f"{action_id}.json" + + +def test_node_offline_generates_alert_only(tmp_path, monkeypatch): + sup = _setup_supervisor(tmp_path, monkeypatch) + _write(tmp_path / "events", _node_event("chelsty-infra", "node_offline")) + + sup._process_ha_events() + + action_id = "alert-node-offline-chelsty-infra" + assert _pending(tmp_path, action_id).exists() + action = json.loads(_pending(tmp_path, action_id).read_text()) + assert action["type"] == "alert_only" + assert action["node"] == "chelsty-infra" + assert action["payload"]["reason"] == "node_offline" + + +def test_node_stale_generates_alert_only(tmp_path, monkeypatch): + sup = _setup_supervisor(tmp_path, monkeypatch) + _write(tmp_path / "events", _node_event("piha", "node_stale")) + + sup._process_ha_events() + + assert _pending(tmp_path, "alert-node-stale-piha").exists() + + +def test_node_online_recovery_generates_alert_only(tmp_path, monkeypatch): + sup = _setup_supervisor(tmp_path, monkeypatch) + _write(tmp_path / "events", _node_event("piha", "node_online")) + + sup._process_ha_events() + + assert _pending(tmp_path, "alert-node-online-piha").exists() + + +def test_node_alert_deduped_within_cooldown(tmp_path, monkeypatch): + """A second identical event must not create a second pending action.""" + sup = _setup_supervisor(tmp_path, monkeypatch) + _write(tmp_path / "events", _node_event("chelsty-infra", "node_offline")) + sup._process_ha_events() + + # Reset the in-memory processed-id set so the file is re-read, but the + # action already exists in pending → must be skipped. + sup._ha_processed_event_ids.clear() + sup._process_ha_events() + + pending = list((tmp_path / "actions" / "pending").glob("*.json")) + assert len(pending) == 1 + + +def test_node_alert_uses_affected_node_when_top_level_missing(tmp_path, monkeypatch): + """If node is absent at top level, fall back to payload.affected_node.""" + sup = _setup_supervisor(tmp_path, monkeypatch) + ev = _node_event("solaria", "node_offline") + ev["node"] = None + _write(tmp_path / "events", ev) + + sup._process_ha_events() + + assert _pending(tmp_path, "alert-node-offline-solaria").exists() + + +def test_non_node_non_ha_event_not_routed(tmp_path, monkeypatch): + sup = _setup_supervisor(tmp_path, monkeypatch) + ev = _node_event("vps", "node_health") # not in NODE_ALERT_EVENTS + _write(tmp_path / "events", ev) + + sup._process_ha_events() + + assert list((tmp_path / "actions" / "pending").glob("*.json")) == []