71a7af5b3f
_resolve_incident() only ever fires from process_event() on a
service_healthy/service_recovered event. A service that is removed,
renamed, or was only ever a one-off test never emits that event again,
so its incident stays "active" in world/incidents.json forever — this
is what left 5 incidents wedged on VPS until a manual on-node edit
during the 2026-08-26 recon session (docs/sessions/2026-08-26.md).
Two independent unwedging mechanisms, both in observer._prune_stale_world
(runs every cycle, so no new event is required to trigger either):
(a) Time-based fallback: any active incident with last_occurrence older
than INCIDENT_STALE_RESOLVE_SECS (env, default 24h) auto-resolves
with resolved_reason="auto_stale_no_events_24h". Unlike the existing
orphan case (Case 2, 5-min guard, only unlinked incidents), this
also clears a service's lingering incident_id link — that link is
exactly what a decommissioned service's incident never gets a
chance to clear via the normal event path.
(b) Manual path: an operator touches
world/resolve-requests/<incident-id>; the observer consumes the
flag file each cycle, force-resolves with resolved_reason=
"manual_operator", and always removes the flag (even for an
unknown/already-resolved id) so a mistyped flag can't sit forever
looking unprocessed.
Chose a flag file over adding a mutation endpoint to operator_ui.py:
/action/mutate only knows actions/<status>/<id>.json, there is no
incidents equivalent, and world/incidents.json is exclusively
observer-owned (rewritten wholesale every cycle by _save_world) —
a second writer (the HTTP handler thread) would race the observer's
own writes. A flag file needs no new HTTP surface and reuses the
same "operator drops a file, the owning process consumes it"
pattern the actions pending/approved queue already uses. Smaller
diff, no new attack surface on a server with no auth on writes.
Tests added to test_incident_lifecycle.py: stale-resolve past the
threshold (service still linked), negative case (fresh active incident
stays active), configurable threshold, manual-flag resolve + flag
removal, flag for an unknown incident, flag for an already-resolved
incident. Full control-plane suite: 179 passed.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017WDKj5LRY8vdQMx57dfNnu
1084 lines
52 KiB
Python
1084 lines
52 KiB
Python
import os
|
||
import re
|
||
import sys
|
||
import json
|
||
import time
|
||
import glob
|
||
import logging
|
||
from logging.handlers import RotatingFileHandler
|
||
import urllib.request
|
||
import urllib.parse
|
||
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."""
|
||
tmp = path.with_suffix(".tmp")
|
||
with open(tmp, "w") as f:
|
||
json.dump(data, f, indent=2)
|
||
f.flush()
|
||
os.fsync(f.fileno())
|
||
os.replace(tmp, path)
|
||
|
||
|
||
def _parse_ts(ts) -> float:
|
||
"""Return a Unix timestamp float from ts, which may be int/float or an ISO-8601 string.
|
||
|
||
Events from node-agent use int(time.time()); events from stability-agent / events.py
|
||
use ISO format ('2026-06-03T10:30:00Z'). Both appear in incident fields such as
|
||
last_occurrence and resolved_at, so any arithmetic on them must go through here.
|
||
Returns 0.0 on None or unparseable input so callers can use plain comparisons.
|
||
"""
|
||
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
|
||
|
||
|
||
# Event filenames follow evt-<node>-<unixts>-<type>-<svc>.json (node_agent.py,
|
||
# ha_diag/event_emitter.py, and the observer's own _emit_node_transition). The
|
||
# <unixts> is the authoritative ordering key for checkpointing — matched the same
|
||
# way operator_ui.py::_event_file_ts does (a 9–11 digit run flanked by dashes;
|
||
# real epochs are 10 digits and stay so until year 2286).
|
||
_EVENT_TS_RE = re.compile(r"-(\d{9,11})-")
|
||
|
||
|
||
def _ts_from_event_name(name) -> "int | None":
|
||
"""Parse the embedded <unixts> from an event filename or full path.
|
||
|
||
Returns the int epoch, or None when the name does not carry one (foreign
|
||
prefix, events.py naming, junk file) so the caller can fall back to mtime.
|
||
"""
|
||
m = _EVENT_TS_RE.search(Path(name).stem)
|
||
return int(m.group(1)) if m else None
|
||
|
||
|
||
def _event_ts_from_path(file_path) -> int:
|
||
"""Ordering/checkpoint timestamp for an event file.
|
||
|
||
Primary: the <unixts> embedded in the filename. Fallback (name doesn't
|
||
parse): the file's mtime. NEVER returns 0 for an existing file — a value of
|
||
0 would make the file compare as "older than the checkpoint" and be skipped
|
||
forever, which is exactly the lexical-path poisoning this fix removes. If
|
||
even stat() fails, fall back to now() so the file is treated as new and gets
|
||
a chance to be processed (and quarantined if truly unreadable).
|
||
"""
|
||
ts = _ts_from_event_name(file_path)
|
||
if ts is not None:
|
||
return ts
|
||
try:
|
||
return int(os.stat(file_path).st_mtime)
|
||
except OSError:
|
||
return int(time.time())
|
||
|
||
|
||
def _checkpoint_ts_from_value(value) -> int:
|
||
"""Coerce a stored checkpoint value into an int epoch (migration helper).
|
||
|
||
Historical formats of observer_checkpoint.json values:
|
||
- int/float → already a timestamp (current format) — kept as-is
|
||
- path/name string → the pre-fix lexical checkpoint; parse the embedded
|
||
<unixts> out of it so the node resumes near where it
|
||
left off instead of reprocessing everything
|
||
- anything else / unparseable string → 0 (reprocess all events for that
|
||
node). Reprocessing is safe: process_event is idempotent w.r.t.
|
||
last_seen/world_state, so re-ingesting duplicates cannot corrupt state —
|
||
whereas guessing too high a checkpoint could silently drop events (the
|
||
failure mode being fixed). Bias to reprocess, never to skip.
|
||
"""
|
||
if isinstance(value, bool): # bool is an int subclass — exclude explicitly
|
||
return 0
|
||
if isinstance(value, (int, float)):
|
||
return int(value)
|
||
if isinstance(value, str) and value:
|
||
ts = _ts_from_event_name(value)
|
||
if ts is not None:
|
||
return ts
|
||
return 0
|
||
|
||
|
||
# Constants and Paths
|
||
RUNTIME_PATH = os.getenv("RUNTIME_PATH", "/opt/homelab")
|
||
EVENTS_DIR = Path(RUNTIME_PATH) / "events"
|
||
STATE_DIR = Path(RUNTIME_PATH) / "state"
|
||
LOGS_DIR = Path(RUNTIME_PATH) / "logs"
|
||
WORLD_DIR = Path(RUNTIME_PATH) / "world"
|
||
OBSERVER_STATE_FILE = STATE_DIR / "observer_checkpoint.json"
|
||
FAILED_EVENTS_DIR = STATE_DIR / "observer_failed_events"
|
||
|
||
# Manual incident-resolve flag drop: an operator (or a future UI action) touches
|
||
# world/resolve-requests/<incident-id> and the observer picks it up here, once
|
||
# per loop iteration — see _process_resolve_requests. Chosen over adding a new
|
||
# operator-ui/API mutation endpoint (see COMMIT 1b rationale in the commit
|
||
# message): /action/mutate only knows about actions/<status>/<id>.json, there is
|
||
# no equivalent for incidents.json, and incidents.json is written exclusively by
|
||
# the observer (world/*.json is observer-owned, overwritten wholesale every
|
||
# cycle by _save_world) — a second writer would race it. A flag file needs no
|
||
# new HTTP surface, no new writer of world/incidents.json, and follows the same
|
||
# "operator drops a file, the owning process consumes it" pattern the actions
|
||
# pending/approved queue already uses.
|
||
RESOLVE_REQUESTS_DIR = WORLD_DIR / "resolve-requests"
|
||
|
||
# Time-based fallback for incidents that can never receive the service_healthy
|
||
# event _resolve_incident() waits for (service removed/renamed/decommissioned
|
||
# from services.yaml, or a one-off test service) — see _prune_stale_world Case 3.
|
||
INCIDENT_STALE_RESOLVE_SECS = int(os.getenv("INCIDENT_STALE_RESOLVE_SECS", str(24 * 3600)))
|
||
|
||
REPO_ROOT = Path(__file__).parent.parent.parent
|
||
INVENTORY_TOPOLOGY = REPO_ROOT / "inventory" / "topology.yaml"
|
||
|
||
# --- SHADOW-READ: Prometheus up{} liveness (cutover etap 1) ----------------
|
||
# Optional parallel-run source. When PROM_SHADOW_URL is set, the observer ALSO
|
||
# queries Prometheus `up{}` each cycle and LOGS any disagreement with its own
|
||
# event-driven liveness — but NEVER acts on it. The authoritative liveness stays
|
||
# 100% event-driven (compute_liveness in _prune_stale_world). Empty/unset →
|
||
# shadow disabled, observer behaves exactly as before (graceful, fail-open).
|
||
# Target value (do NOT hardcode — set via env/host override): http://100.95.58.48:9090
|
||
PROM_SHADOW_URL = os.environ.get("PROM_SHADOW_URL", "").strip()
|
||
PROM_SHADOW_TIMEOUT = int(os.getenv("PROM_SHADOW_TIMEOUT", "5"))
|
||
|
||
# Logging setup
|
||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
||
logger = logging.getLogger("observer")
|
||
|
||
# --- PERSISTENT shadow-mismatch log (cutover etap 2) -----------------------
|
||
# SHADOW_LIVENESS_MISMATCH lines are cutover EVIDENCE and must survive a
|
||
# `docker rm`/recreate of the observer container. stdout (json-file) logs do
|
||
# NOT: an observer recreate on 2026-07-14 destroyed the 07-13/14 mismatch
|
||
# material mid-analysis. We therefore ALSO write each mismatch to a
|
||
# RotatingFileHandler on a HOST-mounted path so the file outlives the container.
|
||
#
|
||
# Path is the repo-conventional logs/<service>/ location under RUNTIME_PATH,
|
||
# i.e. /opt/homelab/logs/observer/shadow-liveness.log. No dedicated bind-mount
|
||
# is required: the base compose already mounts the whole of /opt/homelab into
|
||
# the observer, so this file is on the host by construction, and the observer
|
||
# (uid 1000) creates the dir itself — avoiding the root-owned bind-source
|
||
# ownership footgun a separate /var/log mount would introduce. Overridable via
|
||
# SHADOW_LOG_DIR for tests / non-container runs.
|
||
SHADOW_LOG_DIR = Path(os.getenv("SHADOW_LOG_DIR", str(LOGS_DIR / "observer")))
|
||
SHADOW_LOG_FILENAME = "shadow-liveness.log"
|
||
|
||
|
||
def _make_shadow_logger():
|
||
"""Build the dedicated PERSISTENT logger for SHADOW_LIVENESS_MISMATCH lines.
|
||
|
||
Returns a `logging.getLogger("observer.shadow")` wired to a
|
||
RotatingFileHandler (5 MiB x 5 backups — mismatches are short lines, that is
|
||
weeks of headroom) writing to SHADOW_LOG_DIR/shadow-liveness.log with the
|
||
same timestamped format as the main observer log.
|
||
|
||
propagate=False keeps these records OUT of root/stdout: the call site still
|
||
logs the mismatch to stdout via the ordinary `logger` (unchanged), so
|
||
without this each mismatch would appear twice in `docker logs`.
|
||
|
||
FAIL-SAFE: if the directory/file cannot be created or opened (permission,
|
||
read-only mount, …) we log ONE warning on the main observer logger and
|
||
return a handler-less logger. `.info()` on a handler-less, non-propagating
|
||
logger is a silent no-op, so a broken persistent log NEVER raises and NEVER
|
||
takes the observer down — the mismatch still reaches stdout at the call site.
|
||
Idempotent: any handler from a previous call is dropped first, so
|
||
re-instantiation (tests, re-import) never double-writes or pins a stale path.
|
||
"""
|
||
sl = logging.getLogger("observer.shadow")
|
||
sl.setLevel(logging.INFO)
|
||
sl.propagate = False
|
||
for h in list(sl.handlers):
|
||
sl.removeHandler(h)
|
||
try:
|
||
h.close()
|
||
except Exception:
|
||
pass
|
||
try:
|
||
os.makedirs(SHADOW_LOG_DIR, exist_ok=True)
|
||
handler = RotatingFileHandler(
|
||
SHADOW_LOG_DIR / SHADOW_LOG_FILENAME,
|
||
maxBytes=5 * 1024 * 1024,
|
||
backupCount=5,
|
||
)
|
||
handler.setFormatter(
|
||
logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
|
||
)
|
||
sl.addHandler(handler)
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"shadow-read: could not open persistent mismatch log at %s (%s) — "
|
||
"falling back to stdout only, observer continues",
|
||
SHADOW_LOG_DIR / SHADOW_LOG_FILENAME, exc,
|
||
)
|
||
return sl
|
||
|
||
|
||
class Observer:
|
||
def __init__(self):
|
||
# Per-node-directory checkpoint keyed on the last-processed event
|
||
# TIMESTAMP (int epoch): {"vps": 1784000000, "piha": 1784000123}.
|
||
# A file is "new" iff its event timestamp > the node's checkpoint.
|
||
# This replaces the earlier lexical-PATH comparison, which permanently
|
||
# poisoned a node the moment a single file with a lexically-larger name
|
||
# landed in its dir (e.g. evt-unknown-… > evt-piha-…): every genuinely
|
||
# newer event then sorted "before" the checkpoint and was skipped forever.
|
||
self.node_checkpoints: dict = {}
|
||
self.world_state = {
|
||
"nodes": {},
|
||
"services": {},
|
||
"deployments": {},
|
||
"incidents": {},
|
||
"summary": {
|
||
"last_update": datetime.now(timezone.utc).isoformat(),
|
||
"status": "initializing",
|
||
"active_incidents_count": 0
|
||
}
|
||
}
|
||
self.inventory = self._load_inventory()
|
||
# Nodes declared `status: dormant` in inventory/topology.yaml: expected
|
||
# offline (e.g. chelsty site, hardware down since ~2026-06-01). The
|
||
# observer keeps their last-known world state but never reclassifies
|
||
# liveness for them, so no node_offline/node_stale/node_online events
|
||
# are emitted while a node is dormant. See kb/decisions/architektura-2026-07-28.md.
|
||
self.dormant_nodes = {
|
||
name for name, info in self.inventory["nodes"].items()
|
||
if info.get("status") == "dormant"
|
||
}
|
||
if self.dormant_nodes:
|
||
logger.info("Dormant nodes (no liveness tracking): %s",
|
||
sorted(self.dormant_nodes))
|
||
self._ensure_dirs()
|
||
self._load_checkpoint()
|
||
# Persistent SHADOW_LIVENESS_MISMATCH sink (survives container recreate).
|
||
self.shadow_logger = _make_shadow_logger()
|
||
|
||
def _ensure_dirs(self):
|
||
WORLD_DIR.mkdir(parents=True, exist_ok=True)
|
||
STATE_DIR.mkdir(parents=True, exist_ok=True)
|
||
EVENTS_DIR.mkdir(parents=True, exist_ok=True)
|
||
LOGS_DIR.mkdir(parents=True, exist_ok=True)
|
||
FAILED_EVENTS_DIR.mkdir(parents=True, exist_ok=True)
|
||
RESOLVE_REQUESTS_DIR.mkdir(parents=True, exist_ok=True)
|
||
|
||
def _quarantine_event_file(self, file_path: str, node_dir: str, exc: Exception) -> None:
|
||
"""Move an unreadable/unprocessable event out of the hot path."""
|
||
src = Path(file_path)
|
||
dest_dir = FAILED_EVENTS_DIR / node_dir
|
||
dest_dir.mkdir(parents=True, exist_ok=True)
|
||
dest = dest_dir / src.name
|
||
if dest.exists():
|
||
dest = dest_dir / f"{src.stem}-{int(time.time())}{src.suffix}"
|
||
try:
|
||
os.replace(src, dest)
|
||
logger.error(
|
||
"Quarantined bad event for node_dir=%s: %s -> %s (%s: %s)",
|
||
node_dir, src, dest, type(exc).__name__, exc,
|
||
)
|
||
except Exception as move_exc:
|
||
logger.error(
|
||
"Failed to quarantine bad event for node_dir=%s: %s (%s: %s); move error=%s: %s",
|
||
node_dir, src, type(exc).__name__, exc, type(move_exc).__name__, move_exc,
|
||
)
|
||
|
||
def _load_inventory(self):
|
||
inventory = {"nodes": {}, "services": {}}
|
||
try:
|
||
if INVENTORY_TOPOLOGY.exists():
|
||
with open(INVENTORY_TOPOLOGY, "r") as f:
|
||
topo = yaml.safe_load(f)
|
||
for node_name, node_info in topo.get("nodes", {}).items():
|
||
node_info = node_info or {}
|
||
inventory["nodes"][node_name] = {
|
||
"roles": node_info.get("roles", []),
|
||
"connectivity": node_info.get("connectivity", {}),
|
||
# topology node status: active (default) | dormant
|
||
"status": node_info.get("status", "active"),
|
||
}
|
||
|
||
# Load service assignments from hosts files
|
||
hosts_dir = REPO_ROOT / "hosts"
|
||
for host_dir in hosts_dir.iterdir():
|
||
if host_dir.is_dir():
|
||
svc_file = host_dir / "services.yaml"
|
||
if svc_file.exists():
|
||
with open(svc_file, "r") as f:
|
||
svc_data = yaml.safe_load(f)
|
||
host_name = svc_data.get("host")
|
||
for svc_name, svc_info in svc_data.get("services", {}).items():
|
||
if host_name not in inventory["services"]:
|
||
inventory["services"][host_name] = {}
|
||
inventory["services"][host_name][svc_name] = {
|
||
"role": svc_info.get("role"),
|
||
"exposure": svc_info.get("exposure")
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"Failed to load inventory: {e}")
|
||
return inventory
|
||
|
||
def _load_checkpoint(self):
|
||
if OBSERVER_STATE_FILE.exists():
|
||
try:
|
||
with open(OBSERVER_STATE_FILE, "r") as f:
|
||
checkpoint = json.load(f)
|
||
|
||
if "node_checkpoints" in checkpoint:
|
||
# Per-directory checkpoints. Values may be int epochs (current
|
||
# format) OR pre-fix path strings — coerce every value to an
|
||
# int timestamp so a checkpoint file written by the old
|
||
# lexical-path observer migrates transparently on first start.
|
||
raw = checkpoint["node_checkpoints"] or {}
|
||
self.node_checkpoints = {
|
||
node: _checkpoint_ts_from_value(val)
|
||
for node, val in raw.items()
|
||
}
|
||
if any(not isinstance(v, (int, float)) for v in raw.values()):
|
||
logger.info(
|
||
"Migrated path-based node_checkpoints → timestamps: %s",
|
||
self.node_checkpoints,
|
||
)
|
||
elif "last_processed_file" in checkpoint:
|
||
# Migrate the very old single-file checkpoint: extract node dir
|
||
# from the path and the timestamp from the filename.
|
||
old = checkpoint["last_processed_file"]
|
||
if old:
|
||
try:
|
||
node_dir = Path(old).relative_to(EVENTS_DIR).parts[0]
|
||
self.node_checkpoints = {node_dir: _checkpoint_ts_from_value(old)}
|
||
logger.info(f"Migrated old checkpoint → node_checkpoints: {self.node_checkpoints}")
|
||
except Exception:
|
||
pass # Bad path — start fresh
|
||
|
||
self._load_world_from_disk()
|
||
except Exception as e:
|
||
logger.error(f"Failed to load checkpoint: {e}")
|
||
|
||
def _load_world_from_disk(self):
|
||
# Optional: Load existing state to resume faster
|
||
files = {
|
||
"nodes": WORLD_DIR / "nodes.json",
|
||
"services": WORLD_DIR / "services.json",
|
||
"deployments": WORLD_DIR / "deployments.json",
|
||
"incidents": WORLD_DIR / "incidents.json",
|
||
"summary": WORLD_DIR / "runtime-summary.json"
|
||
}
|
||
for key, path in files.items():
|
||
if path.exists():
|
||
try:
|
||
with open(path, "r") as f:
|
||
self.world_state[key] = json.load(f)
|
||
except Exception as e:
|
||
logger.error(f"Failed to load {key} state: {e}")
|
||
|
||
def _save_checkpoint(self):
|
||
try:
|
||
_atomic_write_json(OBSERVER_STATE_FILE, {"node_checkpoints": self.node_checkpoints})
|
||
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 _query_prometheus_liveness(self):
|
||
"""SHADOW-READ (cutover etap 1): Prometheus up{} → {node_name: up_bool}.
|
||
|
||
Parallel-run only: the result is compared against — and logged next to —
|
||
the authoritative event-driven liveness, but NEVER changes it. See
|
||
PROM_SHADOW_URL. Returns {} when shadow is disabled OR on ANY error, so
|
||
the observer can never crash or change behaviour because of this call:
|
||
every failure is fail-open (info/warning, never error, never raised).
|
||
|
||
Each `up` series is keyed by its `node` label (Prometheus fleet-node
|
||
targets carry node:vps/piha/solaria/lustro). Series without a node label
|
||
(e.g. the prometheus self-scrape up{job="prometheus"}) are ignored.
|
||
"""
|
||
url = PROM_SHADOW_URL
|
||
if not url:
|
||
return {} # shadow disabled — graceful no-op, observer unchanged
|
||
query_url = url.rstrip("/") + "/api/v1/query?" + urllib.parse.urlencode({"query": "up"})
|
||
try:
|
||
with urllib.request.urlopen(query_url, timeout=PROM_SHADOW_TIMEOUT) as resp:
|
||
payload = json.loads(resp.read().decode("utf-8"))
|
||
except Exception as exc:
|
||
# Fail-open: Prometheus down / timeout / bad JSON → no shadow data.
|
||
# Deliberately NOT logger.error — shadow-read must never look like a
|
||
# critical observer failure.
|
||
logger.warning(
|
||
"shadow-read: Prometheus query failed (%s: %s) — fail-open, "
|
||
"event liveness unaffected", type(exc).__name__, exc,
|
||
)
|
||
return {}
|
||
result: dict = {}
|
||
try:
|
||
for series in payload.get("data", {}).get("result", []):
|
||
node = series.get("metric", {}).get("node")
|
||
if not node:
|
||
continue # e.g. up{job="prometheus"} — no node label to map
|
||
value = series.get("value", [None, None])[1]
|
||
result[node] = (value == "1")
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"shadow-read: could not parse Prometheus response (%s: %s) — "
|
||
"fail-open", type(exc).__name__, exc,
|
||
)
|
||
return {}
|
||
return result
|
||
|
||
def _shadow_compare_liveness(self, node_name, event_liveness, node_info, prom_map, now):
|
||
"""SHADOW-READ comparison (cutover etap 1): log event-vs-Prometheus
|
||
liveness disagreement WITHOUT changing anything.
|
||
|
||
Maps both sources onto a shared up/down axis:
|
||
- event DEAD ≈ prom down
|
||
- event FRESH or STALE ≈ prom up (STALE = "seen recently, just
|
||
ageing" — still counts as up for this comparison; documented
|
||
assumption of etap 1)
|
||
|
||
A node Prometheus doesn't know (no series, e.g. chelsty-infra — not
|
||
scraped) is NOT a mismatch: missing data is not disagreement (debug only).
|
||
This method is read-only w.r.t. liveness/status and must never raise.
|
||
"""
|
||
if node_name not in prom_map:
|
||
logger.debug("shadow-read: no prom data for node=%s", node_name)
|
||
return
|
||
prom_up = prom_map[node_name]
|
||
event_up = event_liveness in (FRESH, STALE)
|
||
age = now - _parse_ts(node_info.get("last_seen"))
|
||
if event_up != prom_up:
|
||
logger.info(
|
||
"SHADOW_LIVENESS_MISMATCH node=%s event=%s prom=%s last_seen_age=%.0fs",
|
||
node_name, event_liveness, "up" if prom_up else "down", age,
|
||
)
|
||
# ALSO write to the persistent, host-mounted log so this evidence
|
||
# survives a container recreate (stdout json-file logs do not).
|
||
# Handler-less (fail-safe) → silent no-op; never affects the above.
|
||
self.shadow_logger.info(
|
||
"SHADOW_LIVENESS_MISMATCH node=%s event=%s prom=%s last_seen_age=%.0fs",
|
||
node_name, event_liveness, "up" if prom_up else "down", age,
|
||
)
|
||
else:
|
||
logger.debug(
|
||
"shadow-read agree node=%s event=%s prom=%s last_seen_age=%.0fs",
|
||
node_name, event_liveness, "up" if prom_up else "down", age,
|
||
)
|
||
|
||
def _process_resolve_requests(self, now):
|
||
"""Manual incident-resolve path (COMMIT 1b): consume flag files dropped
|
||
at world/resolve-requests/<incident-id>.
|
||
|
||
There is no operator-ui/API mutation endpoint for incidents (only for
|
||
actions, via /action/mutate — see RESOLVE_REQUESTS_DIR comment), and
|
||
incidents.json is exclusively observer-owned, so a flag file the
|
||
observer itself polls is the smallest addition that lets an operator
|
||
force-resolve a wedged incident without editing world/incidents.json
|
||
by hand.
|
||
|
||
The flag is always removed, even when the incident_id is unknown or
|
||
already resolved, so a stale/mistyped/duplicate flag cannot sit
|
||
forever looking like it hasn't been picked up yet.
|
||
"""
|
||
if not RESOLVE_REQUESTS_DIR.exists():
|
||
return
|
||
for flag_path in RESOLVE_REQUESTS_DIR.iterdir():
|
||
if not flag_path.is_file():
|
||
continue
|
||
incident_id = flag_path.name
|
||
inc = self.world_state["incidents"].get(incident_id)
|
||
if inc and inc.get("status") == "active":
|
||
logger.info(
|
||
f"Manually resolving incident {incident_id} "
|
||
f"(service={inc.get('service')}, node={inc.get('node')}) "
|
||
f"via resolve-request flag"
|
||
)
|
||
inc["status"] = "resolved"
|
||
inc["resolved_at"] = now
|
||
inc["resolved_reason"] = "manual_operator"
|
||
svc_key = f"{inc.get('node')}/{inc.get('service')}"
|
||
svc = self.world_state["services"].get(svc_key)
|
||
if svc and svc.get("incident_id") == incident_id:
|
||
svc["incident_id"] = None
|
||
elif inc:
|
||
logger.info(
|
||
f"Resolve-request flag for {incident_id} ignored "
|
||
f"(status already '{inc.get('status')}') — removing flag"
|
||
)
|
||
else:
|
||
logger.warning(
|
||
f"Resolve-request flag for unknown incident {incident_id} — removing flag"
|
||
)
|
||
try:
|
||
flag_path.unlink()
|
||
except OSError as exc:
|
||
logger.error(f"Failed to remove resolve-request flag {flag_path}: {exc}")
|
||
|
||
def _prune_stale_world(self):
|
||
"""Remove world-state entries for nodes absent from the topology inventory.
|
||
|
||
Root cause this guards against: when NODE_NAME env var is unset, node_agent.py
|
||
falls back to socket.gethostname(), which inside a Docker container returns the
|
||
12-char hex container ID (e.g. 'be17cb6eb0f6') instead of the canonical host name
|
||
('vps'). The observer ingests those events and creates ghost entries that never
|
||
expire on their own.
|
||
|
||
Also ages out resolved incidents older than 7 days to keep world state lean.
|
||
"""
|
||
known_nodes = set(self.inventory["nodes"].keys())
|
||
if not known_nodes:
|
||
# Inventory failed to load — don't prune to avoid wiping valid state.
|
||
return
|
||
|
||
stale_nodes = [n for n in list(self.world_state["nodes"].keys())
|
||
if n not in known_nodes]
|
||
for n in stale_nodes:
|
||
logger.info(f"Pruning stale node from world state: {n}")
|
||
del self.world_state["nodes"][n]
|
||
|
||
stale_svcs = [k for k in list(self.world_state["services"].keys())
|
||
if k.split("/")[0] in stale_nodes]
|
||
for k in stale_svcs:
|
||
logger.info(f"Pruning stale service from world state: {k}")
|
||
del self.world_state["services"][k]
|
||
|
||
# Prune ghost service keys whose service-name portion is a hash-prefixed
|
||
# Docker stale-state artifact (e.g. "9e36297651e7_control-plane-observer").
|
||
# These are created when node-agent incorrectly uses c.name instead of the
|
||
# compose label, and accumulate on every container rebuild.
|
||
# Pattern: <node>/<12hexchars>_<real-name>
|
||
ghost_svcs = [
|
||
k for k in list(self.world_state["services"].keys())
|
||
if len(k.split("/", 1)) == 2
|
||
and len(k.split("/", 1)[1]) > 13
|
||
and k.split("/", 1)[1][12] == "_"
|
||
and all(ch in "0123456789abcdef" for ch in k.split("/", 1)[1][:12])
|
||
]
|
||
for k in ghost_svcs:
|
||
logger.info(f"Pruning ghost (hash-prefixed) service key from world state: {k}")
|
||
del self.world_state["services"][k]
|
||
|
||
now = time.time()
|
||
|
||
# --- 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.
|
||
#
|
||
# SHADOW-READ (cutover etap 1): fetch Prometheus up{} ONCE per cycle
|
||
# (before the node loop, not per-node) for comparison-only logging. {}
|
||
# when disabled/unreachable — fail-open, event liveness is authoritative.
|
||
prom_liveness_map = self._query_prometheus_liveness()
|
||
for node_name, node_info in self.world_state["nodes"].items():
|
||
if node_name in self.dormant_nodes:
|
||
# Dormant node: keep last-known state frozen, emit nothing.
|
||
continue
|
||
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
|
||
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)
|
||
# SHADOW-READ (cutover etap 1): compare (log-only) the event-derived
|
||
# `liveness` computed above against Prometheus up{}. Purely additive —
|
||
# does NOT read back or mutate node_info["liveness"]/["status"], and
|
||
# the authoritative decision above is already committed.
|
||
self._shadow_compare_liveness(node_name, liveness, node_info, prom_liveness_map, now)
|
||
|
||
# Manual resolve-request flags: independent of the linked/orphan
|
||
# distinction below (an operator can force-resolve either kind), so
|
||
# process before the auto-resolve cases. Own try/except: a bad flag
|
||
# file must not block the auto-resolve cases below from running.
|
||
try:
|
||
self._process_resolve_requests(now)
|
||
except Exception as exc:
|
||
logger.error(f"Error processing resolve-request flags: {exc}")
|
||
|
||
try:
|
||
# Collect incident_ids currently referenced by any service entry.
|
||
linked_ids: set = {
|
||
svc.get("incident_id")
|
||
for svc in self.world_state["services"].values()
|
||
if svc.get("incident_id")
|
||
}
|
||
|
||
# Case 1 — service is healthy but still points at an active incident.
|
||
# process_event already calls _resolve_incident on service_healthy events,
|
||
# but if the observer restarted with on-disk state where the link was
|
||
# intact (inconsistency from a pre-atomic-write crash), it may not get
|
||
# resolved until the next service_healthy event is processed. Resolve
|
||
# immediately — a healthy service cannot have an ongoing incident.
|
||
for svc_key, svc in self.world_state["services"].items():
|
||
if svc.get("status") != "healthy":
|
||
continue
|
||
inc_id = svc.get("incident_id")
|
||
if not inc_id:
|
||
continue
|
||
inc = self.world_state["incidents"].get(inc_id, {})
|
||
if inc.get("status") == "active":
|
||
logger.info(
|
||
f"Auto-resolving incident {inc_id} for {svc_key}: "
|
||
f"service is healthy"
|
||
)
|
||
inc["status"] = "resolved"
|
||
inc["resolved_at"] = now
|
||
svc["incident_id"] = None
|
||
linked_ids.discard(inc_id)
|
||
|
||
# Case 2 — orphaned active incident: no service entry links to it and
|
||
# last_occurrence is older than 5 minutes (guard against creation races).
|
||
# These are the stale records left behind when on-disk state was
|
||
# inconsistent: the service entry had incident_id cleared but incidents.json
|
||
# still had the record as "active".
|
||
for inc_id, inc in self.world_state["incidents"].items():
|
||
if inc.get("status") != "active":
|
||
continue
|
||
if inc_id in linked_ids:
|
||
continue
|
||
age = now - _parse_ts(inc.get("last_occurrence"))
|
||
if age > 300: # 5-minute guard
|
||
logger.info(
|
||
f"Auto-resolving orphaned incident {inc_id} "
|
||
f"(service={inc.get('service')}, node={inc.get('node')}): "
|
||
f"no service references it, age={int(age)}s"
|
||
)
|
||
inc["status"] = "resolved"
|
||
inc["resolved_at"] = now
|
||
|
||
# Case 3 — long-idle active incident, time-based fallback (COMMIT 1a).
|
||
# _resolve_incident() only fires on a service_healthy/service_recovered
|
||
# event, which a removed/renamed/decommissioned or one-off test
|
||
# service will never emit again — such an incident would otherwise
|
||
# stay "active" forever, unlike Case 2 above it is NOT limited to
|
||
# orphaned (unlinked) incidents: a service entry can keep pointing at
|
||
# incident_id indefinitely too (that lingering link is exactly the
|
||
# case _resolve_incident() never got a chance to clear), so this
|
||
# checks every active incident regardless of linked_ids.
|
||
for inc_id, inc in self.world_state["incidents"].items():
|
||
if inc.get("status") != "active":
|
||
continue
|
||
age = now - _parse_ts(inc.get("last_occurrence"))
|
||
if age > INCIDENT_STALE_RESOLVE_SECS:
|
||
logger.info(
|
||
f"Auto-resolving stale incident {inc_id} "
|
||
f"(service={inc.get('service')}, node={inc.get('node')}): "
|
||
f"no events for {int(age)}s "
|
||
f"(threshold {INCIDENT_STALE_RESOLVE_SECS}s)"
|
||
)
|
||
inc["status"] = "resolved"
|
||
inc["resolved_at"] = now
|
||
inc["resolved_reason"] = "auto_stale_no_events_24h"
|
||
svc_key = f"{inc.get('node')}/{inc.get('service')}"
|
||
svc = self.world_state["services"].get(svc_key)
|
||
if svc and svc.get("incident_id") == inc_id:
|
||
svc["incident_id"] = None
|
||
|
||
except Exception as exc:
|
||
logger.error(f"Error during incident auto-resolve in _prune_stale_world: {exc}")
|
||
|
||
# Remove resolved incidents older than 7 days.
|
||
# Use _parse_ts so ISO-string resolved_at values are handled correctly.
|
||
stale_incidents = [
|
||
k for k, v in self.world_state["incidents"].items()
|
||
if v.get("status") == "resolved"
|
||
and now - _parse_ts(v.get("resolved_at")) > 7 * 86400
|
||
]
|
||
for k in stale_incidents:
|
||
del self.world_state["incidents"][k]
|
||
|
||
def _save_world(self):
|
||
self.world_state["summary"]["last_update"] = datetime.now(timezone.utc).isoformat()
|
||
active_incidents = [
|
||
k for k, v in self.world_state["incidents"].items() if v.get("status") == "active"
|
||
]
|
||
self.world_state["summary"]["active_incidents_count"] = len(active_incidents)
|
||
self.world_state["summary"]["node_count"] = len(self.world_state["nodes"])
|
||
self.world_state["summary"]["service_count"] = len(self.world_state["services"])
|
||
|
||
if active_incidents:
|
||
self.world_state["summary"]["status"] = "degraded"
|
||
else:
|
||
self.world_state["summary"]["status"] = "nominal"
|
||
|
||
files = {
|
||
"nodes.json": self.world_state["nodes"],
|
||
"services.json": self.world_state["services"],
|
||
"deployments.json": self.world_state["deployments"],
|
||
"incidents.json": self.world_state["incidents"],
|
||
"recommendations.json": [],
|
||
"runtime-summary.json": self.world_state["summary"]
|
||
}
|
||
for filename, data in files.items():
|
||
try:
|
||
_atomic_write_json(WORLD_DIR / filename, data)
|
||
except Exception as e:
|
||
logger.error(f"Failed to save {filename}: {e}")
|
||
|
||
def process_event(self, event):
|
||
etype = event.get("type")
|
||
node = event.get("node")
|
||
service = event.get("service")
|
||
severity = event.get("severity")
|
||
timestamp = event.get("timestamp")
|
||
cid = event.get("correlation_id")
|
||
payload = event.get("payload", {})
|
||
|
||
# 1. Update Node State
|
||
if node not in self.world_state["nodes"]:
|
||
self.world_state["nodes"][node] = {
|
||
"status": "unknown",
|
||
"last_seen": None,
|
||
"roles": self.inventory["nodes"].get(node, {}).get("roles", [])
|
||
}
|
||
self.world_state["nodes"][node]["last_seen"] = timestamp
|
||
|
||
if etype == "node_online":
|
||
self.world_state["nodes"][node]["status"] = "online"
|
||
elif etype == "node_offline":
|
||
self.world_state["nodes"][node]["status"] = "offline"
|
||
|
||
elif etype == "node_health":
|
||
# Regular heartbeat from node-agent; updates resource metrics.
|
||
# Clears disk_pressure if disk is now healthy (< warn threshold).
|
||
self.world_state["nodes"][node]["status"] = "online"
|
||
self.world_state["nodes"][node].update({
|
||
"disk_usage_pct": payload.get("disk_pct"),
|
||
"mem_usage_pct": payload.get("mem_pct"),
|
||
"cpu_usage_pct": payload.get("cpu_pct"),
|
||
})
|
||
if (payload.get("disk_pct") or 0) < 75:
|
||
self.world_state["nodes"][node].pop("disk_pressure", None)
|
||
|
||
elif etype == "disk_pressure":
|
||
# Emitted when disk usage crosses 75 % (medium) or 85 % (high).
|
||
# The supervisor reads disk_pressure to generate disk_cleanup actions.
|
||
self.world_state["nodes"][node]["disk_pressure"] = severity
|
||
self.world_state["nodes"][node]["disk_usage_pct"] = payload.get("usage_pct")
|
||
|
||
elif etype == "high_memory":
|
||
# Memory pressure observation; recorded on the node for correlation.
|
||
# No automated action — operator decides if a container restart helps.
|
||
self.world_state["nodes"][node]["memory_pressure"] = severity
|
||
self.world_state["nodes"][node]["mem_usage_pct"] = payload.get("usage_pct")
|
||
|
||
elif etype == "high_cpu":
|
||
# CPU pressure observation; recorded for visibility.
|
||
self.world_state["nodes"][node]["cpu_pressure"] = severity
|
||
self.world_state["nodes"][node]["cpu_usage_pct"] = payload.get("usage_pct")
|
||
|
||
# 2. Update Service State
|
||
if service and service != "all":
|
||
svc_key = f"{node}/{service}"
|
||
if svc_key not in self.world_state["services"]:
|
||
self.world_state["services"][svc_key] = {
|
||
"node": node,
|
||
"service": service,
|
||
"status": "unknown",
|
||
"last_check": None,
|
||
"incident_id": None
|
||
}
|
||
self.world_state["services"][svc_key]["last_check"] = timestamp
|
||
|
||
if etype == "service_recovered":
|
||
self.world_state["services"][svc_key]["status"] = "healthy"
|
||
self._resolve_incident(svc_key, timestamp)
|
||
elif etype == "service_healthy":
|
||
# Positive confirmation from node-agent that a managed container
|
||
# is running. This keeps services.json populated so the supervisor
|
||
# can correctly detect drift (absent entry = never reported = unknown,
|
||
# not the same as confirmed missing).
|
||
# Also resolve any active incident — if a service that had been
|
||
# unhealthy/crashing is now confirmed healthy, the incident is over.
|
||
self.world_state["services"][svc_key]["status"] = "healthy"
|
||
self._resolve_incident(svc_key, timestamp)
|
||
elif etype in ["service_unhealthy", "healthcheck_failed", "containers_not_running"]:
|
||
# containers_not_running: node-agent (node_agent.py) — and the
|
||
# stability-agent — report a *managed* container that has exited /
|
||
# dead or is crash-looping (restarting past RestartCount threshold).
|
||
# Treat it exactly like the other hard-failure signals: mark the
|
||
# service unhealthy and open an incident. The incident's
|
||
# trigger_type is the event type ("containers_not_running"), which
|
||
# the supervisor already recognises in CONTAINER_RESTART_TRIGGERS and
|
||
# remediates with a low-risk container_restart (vs. a full redeploy).
|
||
#
|
||
# Before this branch existed, containers_not_running fell through
|
||
# this if/elif chain entirely: node-agent saw the dead container and
|
||
# emitted the event, the observer bumped last_check but left status at
|
||
# its last "healthy" value and created NO incident. The supervisor
|
||
# then saw no drift and generated no action — so a dead / crash-looping
|
||
# container produced ZERO operator alerts. The monitoring loop was
|
||
# silently broken (matches the "action queue empty despite failures"
|
||
# symptom). (node_agent.py's own comment claims this event "rides the
|
||
# existing, supervisor-wired remediation path" — that path only exists
|
||
# once the observer opens the incident here.)
|
||
self.world_state["services"][svc_key]["status"] = "unhealthy"
|
||
self._handle_incident(svc_key, event)
|
||
elif etype in ["container_restarting", "container_state_unexpected"]:
|
||
# Intentionally observational — parity with node_agent.check_containers,
|
||
# which emits these as low/medium severity and explicitly does NOT wire
|
||
# them to any supervisor trigger. Causes: a transient post-deploy
|
||
# restart still below the crash-loop threshold, an operator `docker
|
||
# pause`, or an unhandled-but-not-fault docker state.
|
||
#
|
||
# We deliberately do NOT set status=unhealthy (that would make the
|
||
# supervisor generate remediation for a transient blip — a redeploy,
|
||
# since there is no CONTAINER_RESTART_TRIGGERS incident) and do NOT
|
||
# open an incident (noise). But the event must not vanish, so we
|
||
# record a lightweight observational trace on the service entry; the
|
||
# raw event also stays visible in the /events feed. If a restart is a
|
||
# real crash-loop it escalates on its own: node-agent re-emits it as
|
||
# containers_not_running once RestartCount crosses the threshold, and
|
||
# the branch above then alarms.
|
||
self.world_state["services"][svc_key]["last_observation"] = {
|
||
"type": etype,
|
||
"severity": severity,
|
||
"timestamp": timestamp,
|
||
"message": event.get("message"),
|
||
}
|
||
|
||
# 3. Update Deployment State
|
||
if etype.startswith("deployment_") and cid:
|
||
if cid not in self.world_state["deployments"]:
|
||
self.world_state["deployments"][cid] = {
|
||
"node": node,
|
||
"service": service,
|
||
"status": "unknown",
|
||
"started_at": None,
|
||
"finished_at": None,
|
||
"events": []
|
||
}
|
||
self.world_state["deployments"][cid]["events"].append({
|
||
"type": etype,
|
||
"timestamp": timestamp,
|
||
"payload": payload
|
||
})
|
||
if etype == "deployment_started":
|
||
self.world_state["deployments"][cid]["status"] = "in_progress"
|
||
self.world_state["deployments"][cid]["started_at"] = timestamp
|
||
elif etype == "deployment_completed":
|
||
self.world_state["deployments"][cid]["status"] = "completed"
|
||
self.world_state["deployments"][cid]["finished_at"] = timestamp
|
||
elif etype == "deployment_failed":
|
||
self.world_state["deployments"][cid]["status"] = "failed"
|
||
self.world_state["deployments"][cid]["finished_at"] = timestamp
|
||
# Deployment failure often creates an incident
|
||
self._handle_deployment_failure(event)
|
||
|
||
def _handle_incident(self, svc_key, event):
|
||
# Correlation: collapse repeated failures for the same service on the same node
|
||
active_incident = self.world_state["services"][svc_key].get("incident_id")
|
||
|
||
if active_incident and active_incident in self.world_state["incidents"]:
|
||
incident = self.world_state["incidents"][active_incident]
|
||
if incident["status"] == "active":
|
||
incident["last_occurrence"] = event["timestamp"]
|
||
incident["occurrence_count"] = incident.get("occurrence_count", 1) + 1
|
||
incident["events"].append(event["timestamp"])
|
||
return
|
||
|
||
# Create new incident
|
||
incident_id = f"inc-{int(time.time())}-{event.get('node')}-{event.get('service')}"
|
||
self.world_state["incidents"][incident_id] = {
|
||
"id": incident_id,
|
||
"node": event.get("node"),
|
||
"service": event.get("service"),
|
||
"status": "active",
|
||
"severity": event.get("severity"),
|
||
# trigger_type records the event type that opened this incident so that
|
||
# the supervisor can choose the appropriate remediation action
|
||
# (container_restart for containers_not_running / healthcheck_failed
|
||
# vs. a full redeploy for other causes).
|
||
"trigger_type": event.get("type"),
|
||
"started_at": event.get("timestamp"),
|
||
"last_occurrence": event.get("timestamp"),
|
||
"occurrence_count": 1,
|
||
"events": [event["timestamp"]],
|
||
"correlation_id": event.get("correlation_id")
|
||
}
|
||
self.world_state["services"][svc_key]["incident_id"] = incident_id
|
||
|
||
def _resolve_incident(self, svc_key, timestamp):
|
||
incident_id = self.world_state["services"][svc_key].get("incident_id")
|
||
if incident_id and incident_id in self.world_state["incidents"]:
|
||
if self.world_state["incidents"][incident_id]["status"] == "active":
|
||
self.world_state["incidents"][incident_id]["status"] = "resolved"
|
||
self.world_state["incidents"][incident_id]["resolved_at"] = timestamp
|
||
self.world_state["services"][svc_key]["incident_id"] = None
|
||
|
||
def _handle_deployment_failure(self, event):
|
||
# Specific logic for deployment failures
|
||
svc_key = f"{event.get('node')}/{event.get('service')}"
|
||
self._handle_incident(svc_key, event)
|
||
|
||
# Link diagnostics if available in payload
|
||
incident_id = self.world_state["services"][svc_key].get("incident_id")
|
||
if incident_id and incident_id in self.world_state["incidents"]:
|
||
payload = event.get("payload", {})
|
||
if "diagnostics_file" in payload:
|
||
self.world_state["incidents"][incident_id]["diagnostics_ref"] = payload["diagnostics_file"]
|
||
elif "error" in payload:
|
||
self.world_state["incidents"][incident_id]["last_error"] = payload["error"]
|
||
|
||
def run_once(self):
|
||
# Update heartbeat
|
||
heartbeat_file = STATE_DIR / "observer.heartbeat"
|
||
try:
|
||
heartbeat_file.touch()
|
||
except Exception as e:
|
||
logger.error(f"Failed to touch heartbeat file: {e}")
|
||
|
||
# Collect all event files grouped by node directory. A file is "new"
|
||
# when its event TIMESTAMP (from the filename, mtime fallback) is greater
|
||
# than the node's checkpoint timestamp — never a lexical path compare, so
|
||
# a lexically-smaller-but-newer filename (evt-piha-… after a stray
|
||
# evt-unknown-…) can no longer poison a node into skipping every event.
|
||
all_files = glob.glob(str(EVENTS_DIR / "**" / "*.json"), recursive=True)
|
||
|
||
new_files = []
|
||
for file_path in all_files:
|
||
try:
|
||
node_dir = str(Path(file_path).relative_to(EVENTS_DIR).parts[0])
|
||
except (IndexError, ValueError):
|
||
node_dir = "__unknown__"
|
||
ev_ts = _event_ts_from_path(file_path)
|
||
last_for_node = self.node_checkpoints.get(node_dir, 0)
|
||
if ev_ts > last_for_node:
|
||
new_files.append((ev_ts, node_dir, file_path))
|
||
|
||
# Process oldest-first (tie-break on path for determinism) so the
|
||
# checkpoint advances monotonically in time and a mid-batch crash resumes
|
||
# from the right place.
|
||
new_files.sort(key=lambda t: (t[0], t[2]))
|
||
|
||
if not new_files:
|
||
# Even if no new events, prune stale entries and refresh summary freshness.
|
||
self._prune_stale_world()
|
||
self._save_world()
|
||
return
|
||
|
||
logger.info(f"Processing {len(new_files)} new events across "
|
||
f"{len({n for _, n, _ in new_files})} node(s)")
|
||
for ev_ts, node_dir, file_path in new_files:
|
||
try:
|
||
with open(file_path, "r") as f:
|
||
event = json.load(f)
|
||
# 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 ev_ts > self.node_checkpoints.get(node_dir, 0):
|
||
self.node_checkpoints[node_dir] = ev_ts
|
||
continue
|
||
self.process_event(event)
|
||
# Advance per-node checkpoint by timestamp (only forward).
|
||
if ev_ts > self.node_checkpoints.get(node_dir, 0):
|
||
self.node_checkpoints[node_dir] = ev_ts
|
||
except Exception as e:
|
||
logger.error(
|
||
"Error processing node_dir=%s file=%s (%s: %s)",
|
||
node_dir, file_path, type(e).__name__, e,
|
||
)
|
||
self._quarantine_event_file(file_path, node_dir, e)
|
||
|
||
self._save_checkpoint()
|
||
self._prune_stale_world()
|
||
self._save_world()
|
||
|
||
def loop(self, interval=5):
|
||
logger.info("Starting observer loop")
|
||
while True:
|
||
self.run_once()
|
||
time.sleep(interval)
|
||
|
||
if __name__ == "__main__":
|
||
import sys
|
||
observer = Observer()
|
||
if "--run-once" in sys.argv:
|
||
observer.run_once()
|
||
else:
|
||
observer.loop()
|