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) <noreply@anthropic.com>
This commit is contained in:
parent
4965866644
commit
e5ecefe2b7
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
254
services/control-plane/tests/test_observer_shadow_liveness.py
Normal file
254
services/control-plane/tests/test_observer_shadow_liveness.py
Normal file
|
|
@ -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
|
||||
Loading…
Reference in a new issue