From e5ecefe2b74b93c881f11b3b5bb0b70a6b11248f Mon Sep 17 00:00:00 2001 From: Oskar Kapala Date: Thu, 9 Jul 2026 15:38:20 +0200 Subject: [PATCH] feat(observer): shadow-read Prometheus up{} liveness with mismatch logging (cutover etap 1, no switching) Observer now optionally (PROM_SHADOW_URL) queries Prometheus up{} once per cycle and LOGS SHADOW_LIVENESS_MISMATCH when its event-driven liveness disagrees. Parallel-run only: compute_liveness and _emit_node_transition are untouched; authoritative liveness stays 100% event-driven. Fail-open on any Prometheus error (down/timeout/bad JSON -> {}). 9 new tests, incl. proof that shadow-read does not change node_info liveness/status. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../control-plane/docker-compose.override.yml | 7 + hosts/vps/runtime/control-plane/env.example | 11 + scripts/observer/observer.py | 98 +++++++ .../tests/test_observer_shadow_liveness.py | 254 ++++++++++++++++++ 4 files changed, 370 insertions(+) create mode 100644 services/control-plane/tests/test_observer_shadow_liveness.py diff --git a/hosts/vps/runtime/control-plane/docker-compose.override.yml b/hosts/vps/runtime/control-plane/docker-compose.override.yml index 7834b87..df6f8a1 100644 --- a/hosts/vps/runtime/control-plane/docker-compose.override.yml +++ b/hosts/vps/runtime/control-plane/docker-compose.override.yml @@ -27,6 +27,13 @@ services: observer: mem_limit: 192m oom_score_adj: -900 + # SHADOW-READ (Prometheus liveness cutover — etap 1). Optional: when set, the + # observer ALSO queries Prometheus up{} each cycle and LOGS mismatches vs its + # event-driven liveness (grep SHADOW_LIVENESS_MISMATCH), but NEVER switches + # source. Unset → disabled, observer unchanged (fail-open on any prom error). + # Uncomment to enable the ≥7-day parallel-run: + # environment: + # - PROM_SHADOW_URL=http://100.95.58.48:9090 supervisor: mem_limit: 400m diff --git a/hosts/vps/runtime/control-plane/env.example b/hosts/vps/runtime/control-plane/env.example index 89985a1..24fa983 100644 --- a/hosts/vps/runtime/control-plane/env.example +++ b/hosts/vps/runtime/control-plane/env.example @@ -5,3 +5,14 @@ HOMELAB_EVENTS_ROOT=/opt/homelab/events HOMELAB_WORLD_ROOT=/opt/homelab/world HOMELAB_ACTIONS_ROOT=/opt/homelab/actions HOMELAB_CONFIG_ROOT=/opt/homelab/config + +# --- SHADOW-READ: Prometheus up{} liveness (cutover etap 1) --------------- +# OPTIONAL. Consumed only by the observer service. When set, the observer ALSO +# queries Prometheus up{} each cycle and LOGS disagreements with its own +# event-driven liveness (grep-able: SHADOW_LIVENESS_MISMATCH). It NEVER switches +# the liveness source — this is a parallel-run to collect ≥7 days of evidence. +# Empty/unset → shadow disabled, observer behaves exactly as today. Any +# Prometheus error (down/timeout/bad JSON) is fail-open (no effect on liveness). +# Real value is the fleet-prometheus Tailscale URL, e.g.: +# PROM_SHADOW_URL=http://100.95.58.48:9090 +# PROM_SHADOW_TIMEOUT=5 diff --git a/scripts/observer/observer.py b/scripts/observer/observer.py index 0ecfaa1..5eb240f 100644 --- a/scripts/observer/observer.py +++ b/scripts/observer/observer.py @@ -4,6 +4,8 @@ import json import time import glob import logging +import urllib.request +import urllib.parse import yaml from datetime import datetime, timezone from pathlib import Path @@ -65,6 +67,16 @@ FAILED_EVENTS_DIR = STATE_DIR / "observer_failed_events" 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") @@ -253,6 +265,82 @@ class Observer: 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, + ) + 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 _prune_stale_world(self): """Remove world-state entries for nodes absent from the topology inventory. @@ -306,6 +394,11 @@ class Observer: # 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(): roles = (node_info.get("roles") or self.inventory["nodes"].get(node_name, {}).get("roles", [])) @@ -324,6 +417,11 @@ class Observer: # 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) try: # Collect incident_ids currently referenced by any service entry. diff --git a/services/control-plane/tests/test_observer_shadow_liveness.py b/services/control-plane/tests/test_observer_shadow_liveness.py new file mode 100644 index 0000000..e1e7dfe --- /dev/null +++ b/services/control-plane/tests/test_observer_shadow_liveness.py @@ -0,0 +1,254 @@ +"""Tests for the observer SHADOW-READ of Prometheus up{} liveness (cutover etap 1). + +The whole point of etap 1 is a PARALLEL-RUN: the observer additionally queries +Prometheus and LOGS disagreements, but authoritative liveness stays 100% +event-driven (compute_liveness). These tests prove exactly that: shadow-read +never mutates node_info["liveness"]/["status"], is fail-open on any Prometheus +error, and logs a grep-able SHADOW_LIVENESS_MISMATCH line on disagreement. +""" +from __future__ import annotations + +import logging +import sys +import time +from pathlib import Path + +import pytest + +# Observer lives outside the control-plane package; add scripts/ to path. +sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent / "scripts")) +import observer.observer as obs_mod +from observer.observer import Observer +from liveness import FRESH, DEAD + + +# --------------------------------------------------------------------------- +# Fixtures / helpers +# --------------------------------------------------------------------------- + +@pytest.fixture(autouse=True) +def _redirect_observer_paths(tmp_path, monkeypatch): + """Redirect every observer runtime path into a per-test tmp_path. + + Mirrors the fixture in test_incident_lifecycle.py so these tests never touch + the real /opt/homelab state. Also defaults PROM_SHADOW_URL to "" (disabled) + so a test that forgets to set it exercises the graceful no-op path. + """ + world = tmp_path / "world" + state = tmp_path / "state" + events = tmp_path / "events" + logs = tmp_path / "logs" + repo = tmp_path / "repo" + + for d in (world, state, events, logs, repo / "inventory", repo / "hosts"): + d.mkdir(parents=True, exist_ok=True) + + (repo / "inventory" / "topology.yaml").write_text( + "nodes:\n" + " vps:\n roles: [control-plane]\n connectivity: {}\n" + " piha:\n roles: [infra]\n connectivity: {}\n" + ) + + monkeypatch.setattr(obs_mod, "WORLD_DIR", world) + monkeypatch.setattr(obs_mod, "STATE_DIR", state) + monkeypatch.setattr(obs_mod, "EVENTS_DIR", events) + monkeypatch.setattr(obs_mod, "LOGS_DIR", logs) + monkeypatch.setattr(obs_mod, "INVENTORY_TOPOLOGY", repo / "inventory" / "topology.yaml") + monkeypatch.setattr(obs_mod, "REPO_ROOT", repo) + monkeypatch.setattr(obs_mod, "FAILED_EVENTS_DIR", state / "observer_failed_events") + monkeypatch.setattr(obs_mod, "OBSERVER_STATE_FILE", state / "observer_checkpoint.json") + monkeypatch.setattr(obs_mod, "PROM_SHADOW_URL", "") + + +class _FakeResp: + """Minimal context-manager stand-in for urllib.request.urlopen()'s return.""" + + def __init__(self, body: str): + self._body = body + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + def read(self): + return self._body.encode("utf-8") + + +def _mock_prom(monkeypatch, body): + """Make urllib.request.urlopen return `body` (str) or raise (if Exception).""" + def _fake_urlopen(url, timeout=None): + if isinstance(body, Exception): + raise body + return _FakeResp(body) + monkeypatch.setattr(obs_mod.urllib.request, "urlopen", _fake_urlopen) + + +_SAMPLE_UP_JSON = ( + '{"status":"success","data":{"resultType":"vector","result":[' + '{"metric":{"__name__":"up","job":"fleet-node","node":"vps"},"value":[1720000000,"1"]},' + '{"metric":{"__name__":"up","job":"fleet-node","node":"piha"},"value":[1720000000,"0"]},' + '{"metric":{"__name__":"up","job":"prometheus"},"value":[1720000000,"1"]}' + ']}}' +) + + +# --------------------------------------------------------------------------- +# 1. _query_prometheus_liveness — disabled / parsing / fail-open +# --------------------------------------------------------------------------- + +def test_query_returns_empty_when_url_unset(monkeypatch): + """PROM_SHADOW_URL empty → {} (shadow disabled), no HTTP attempted.""" + monkeypatch.setattr(obs_mod, "PROM_SHADOW_URL", "") + + def _boom(*a, **k): # pragma: no cover - must never be called + raise AssertionError("urlopen must not be called when shadow is disabled") + monkeypatch.setattr(obs_mod.urllib.request, "urlopen", _boom) + + assert Observer()._query_prometheus_liveness() == {} + + +def test_query_parses_up_series_by_node_label(monkeypatch): + """Parses up{} into {node: bool}; series without a node label are ignored.""" + monkeypatch.setattr(obs_mod, "PROM_SHADOW_URL", "http://prom:9090") + _mock_prom(monkeypatch, _SAMPLE_UP_JSON) + + result = Observer()._query_prometheus_liveness() + + assert result == {"vps": True, "piha": False} + assert "prometheus" not in result # up{job="prometheus"} has no node label + + +def test_query_fail_open_on_unreachable(monkeypatch): + """Prometheus unreachable → {} and NO exception propagates (fail-open).""" + monkeypatch.setattr(obs_mod, "PROM_SHADOW_URL", "http://prom:9090") + _mock_prom(monkeypatch, ConnectionRefusedError("nope")) + + assert Observer()._query_prometheus_liveness() == {} + + +def test_query_fail_open_on_bad_json(monkeypatch): + """Malformed JSON body → {} and no raise.""" + monkeypatch.setattr(obs_mod, "PROM_SHADOW_URL", "http://prom:9090") + _mock_prom(monkeypatch, "this is not json") + + assert Observer()._query_prometheus_liveness() == {} + + +# --------------------------------------------------------------------------- +# 2. The load-bearing invariant: shadow-read NEVER changes liveness +# --------------------------------------------------------------------------- + +def _fresh_node_world(obs: Observer): + """Seed one node whose event-driven liveness computes to FRESH.""" + obs.world_state["nodes"] = { + "vps": {"status": "unknown", "last_seen": int(time.time()), "roles": ["control-plane"]}, + } + + +def test_shadow_does_not_change_liveness(monkeypatch): + """Event liveness after a cycle WITH shadow (mismatching prom) is byte-for-byte + identical to a cycle WITHOUT shadow. Proof that shadow switches nothing.""" + # Baseline: shadow disabled. + baseline = Observer() + _fresh_node_world(baseline) + baseline._prune_stale_world() + base_liveness = baseline.world_state["nodes"]["vps"]["liveness"] + base_status = baseline.world_state["nodes"]["vps"]["status"] + + assert base_liveness == FRESH + assert base_status == "online" + + # Shadow enabled, and Prometheus DISAGREES (says vps is down). + monkeypatch.setattr(obs_mod, "PROM_SHADOW_URL", "http://prom:9090") + _mock_prom( + monkeypatch, + '{"data":{"result":[' + '{"metric":{"node":"vps"},"value":[1720000000,"0"]}]}}', + ) + shadow = Observer() + _fresh_node_world(shadow) + shadow._prune_stale_world() + + # Authoritative liveness/status must be IDENTICAL to the no-shadow baseline. + assert shadow.world_state["nodes"]["vps"]["liveness"] == base_liveness + assert shadow.world_state["nodes"]["vps"]["status"] == base_status + # And no prom-derived field leaked into the node record. + assert "prom_up" not in shadow.world_state["nodes"]["vps"] + + +# --------------------------------------------------------------------------- +# 3. Mismatch logging +# --------------------------------------------------------------------------- + +def test_mismatch_is_logged(monkeypatch, caplog): + """event=fresh but prom=down → grep-able SHADOW_LIVENESS_MISMATCH at INFO.""" + monkeypatch.setattr(obs_mod, "PROM_SHADOW_URL", "http://prom:9090") + _mock_prom( + monkeypatch, + '{"data":{"result":[' + '{"metric":{"node":"vps"},"value":[1720000000,"0"]}]}}', + ) + obs = Observer() + _fresh_node_world(obs) + + with caplog.at_level(logging.INFO, logger="observer"): + obs._prune_stale_world() + + assert "SHADOW_LIVENESS_MISMATCH" in caplog.text + assert "node=vps" in caplog.text + assert "event=fresh" in caplog.text + assert "prom=down" in caplog.text + + +def test_agreement_does_not_log_mismatch(monkeypatch, caplog): + """event=fresh and prom=up → agreement, no mismatch line at INFO.""" + monkeypatch.setattr(obs_mod, "PROM_SHADOW_URL", "http://prom:9090") + _mock_prom( + monkeypatch, + '{"data":{"result":[' + '{"metric":{"node":"vps"},"value":[1720000000,"1"]}]}}', + ) + obs = Observer() + _fresh_node_world(obs) + + with caplog.at_level(logging.INFO, logger="observer"): + obs._prune_stale_world() + + assert "SHADOW_LIVENESS_MISMATCH" not in caplog.text + + +def test_dead_node_agrees_with_prom_down(monkeypatch, caplog): + """event=dead and prom=down → agreement (DEAD maps to down).""" + monkeypatch.setattr(obs_mod, "PROM_SHADOW_URL", "http://prom:9090") + _mock_prom( + monkeypatch, + '{"data":{"result":[' + '{"metric":{"node":"vps"},"value":[1720000000,"0"]}]}}', + ) + obs = Observer() + obs.world_state["nodes"] = { + "vps": {"status": "unknown", "last_seen": int(time.time()) - 700, "roles": ["control-plane"]}, + } + + with caplog.at_level(logging.INFO, logger="observer"): + obs._prune_stale_world() + + assert obs.world_state["nodes"]["vps"]["liveness"] == DEAD + assert "SHADOW_LIVENESS_MISMATCH" not in caplog.text + + +def test_node_unknown_to_prometheus_is_not_mismatch(monkeypatch, caplog): + """A node with no up{} series (e.g. chelsty-infra) is not a disagreement.""" + monkeypatch.setattr(obs_mod, "PROM_SHADOW_URL", "http://prom:9090") + _mock_prom(monkeypatch, '{"data":{"result":[]}}') # empty — knows no node + obs = Observer() + _fresh_node_world(obs) + + with caplog.at_level(logging.INFO, logger="observer"): + obs._prune_stale_world() + + assert "SHADOW_LIVENESS_MISMATCH" not in caplog.text + # And liveness still computed from events. + assert obs.world_state["nodes"]["vps"]["liveness"] == FRESH