226 lines
7.3 KiB
Python
226 lines
7.3 KiB
Python
"""
|
|
brain-watchdog: external watchdog for the control-plane on VPS.
|
|
|
|
Runs on PIHA; queries /summary directly over Tailscale and alerts via
|
|
Telegram Bot API without going through the control-plane itself.
|
|
Never trusts the self-reported "status" field — freshness is computed
|
|
locally from last_update epoch vs. time.time().
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
CONTROL_PLANE_URL = os.environ["CONTROL_PLANE_URL"].rstrip("/")
|
|
STALE_THRESHOLD = int(os.environ.get("STALE_THRESHOLD", "600"))
|
|
INTERVAL = int(os.environ.get("INTERVAL", "60"))
|
|
FAILS_BEFORE_ALERT = int(os.environ.get("FAILS_BEFORE_ALERT", "3"))
|
|
TG_TOKEN = os.environ["TG_TOKEN"]
|
|
TG_CHAT_ID = os.environ["TG_CHAT_ID"]
|
|
HEALTHCHECKS_URL = os.environ.get("HEALTHCHECKS_URL", "").strip()
|
|
PROMETHEUS_URL = os.environ.get("PROMETHEUS_URL", "").strip()
|
|
|
|
STATE_FILE = Path("/data/state.json")
|
|
|
|
|
|
def load_state() -> dict:
|
|
if STATE_FILE.exists():
|
|
try:
|
|
return json.loads(STATE_FILE.read_text())
|
|
except Exception:
|
|
pass
|
|
return {"fail_count": 0, "alerted": False, "last_ok": 0.0, "prom_alerted": {}}
|
|
|
|
|
|
def save_state(state: dict) -> None:
|
|
STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
|
|
STATE_FILE.write_text(json.dumps(state))
|
|
|
|
|
|
def http_get(url: str, timeout: int = 10) -> tuple[int | None, dict | None]:
|
|
try:
|
|
with urllib.request.urlopen(url, timeout=timeout) as resp:
|
|
return resp.status, json.loads(resp.read())
|
|
except urllib.error.HTTPError as exc:
|
|
return exc.code, None
|
|
except Exception:
|
|
return None, None
|
|
|
|
|
|
def send_telegram(message: str) -> bool:
|
|
url = f"https://api.telegram.org/bot{TG_TOKEN}/sendMessage"
|
|
payload = json.dumps(
|
|
{"chat_id": TG_CHAT_ID, "text": message, "parse_mode": "HTML"}
|
|
).encode()
|
|
req = urllib.request.Request(
|
|
url, data=payload, headers={"Content-Type": "application/json"}
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=10) as resp:
|
|
return resp.status == 200
|
|
except Exception as exc:
|
|
print(f"[telegram] send failed: {exc}", flush=True)
|
|
return False
|
|
|
|
|
|
def ping_healthchecks() -> None:
|
|
if not HEALTHCHECKS_URL:
|
|
return
|
|
try:
|
|
urllib.request.urlopen(HEALTHCHECKS_URL, timeout=10)
|
|
except Exception as exc:
|
|
print(f"[healthchecks] ping failed: {exc}", flush=True)
|
|
|
|
|
|
def check_prometheus_alerts() -> list[dict]:
|
|
if not PROMETHEUS_URL:
|
|
return []
|
|
status, body = http_get(f"{PROMETHEUS_URL}/api/v1/alerts")
|
|
if status is None or status != 200 or not body:
|
|
print(f"[prometheus] poll failed (status={status})", flush=True)
|
|
return []
|
|
try:
|
|
alerts = body.get("data", {}).get("alerts", [])
|
|
firing = []
|
|
for alert in alerts:
|
|
if alert.get("state") != "firing":
|
|
continue
|
|
labels = alert.get("labels", {})
|
|
annotations = alert.get("annotations", {})
|
|
alertname = labels.get("alertname", "unknown")
|
|
node = labels.get("node", labels.get("instance", "unknown"))
|
|
firing.append({
|
|
"key": f"{alertname}:{node}",
|
|
"alertname": alertname,
|
|
"node": node,
|
|
"summary": annotations.get("summary", ""),
|
|
"description": annotations.get("description", ""),
|
|
})
|
|
return firing
|
|
except Exception as exc:
|
|
print(f"[prometheus] parse error: {exc}", flush=True)
|
|
return []
|
|
|
|
|
|
def handle_prometheus_alerts(state: dict) -> None:
|
|
firing = check_prometheus_alerts()
|
|
prom_alerted: dict = state.get("prom_alerted", {})
|
|
firing_keys = {a["key"] for a in firing}
|
|
|
|
for alert in firing:
|
|
key = alert["key"]
|
|
if key not in prom_alerted:
|
|
msg = (
|
|
f"🚨 <b>Prometheus alert: {alert['alertname']}</b>\n"
|
|
f"Node: <code>{alert['node']}</code>\n"
|
|
f"Summary: {alert['summary']}"
|
|
)
|
|
if alert["description"]:
|
|
msg += f"\n{alert['description']}"
|
|
sent = send_telegram(msg)
|
|
if sent:
|
|
prom_alerted[key] = {"alertname": alert["alertname"], "node": alert["node"]}
|
|
print(f"[prometheus] sent alert: {key}", flush=True)
|
|
|
|
for key in list(prom_alerted.keys()):
|
|
if key not in firing_keys:
|
|
info = prom_alerted[key]
|
|
send_telegram(
|
|
f"✅ <b>Prometheus alert resolved: {info['alertname']}</b>\n"
|
|
f"Node: <code>{info['node']}</code>"
|
|
)
|
|
del prom_alerted[key]
|
|
print(f"[prometheus] sent recovery: {key}", flush=True)
|
|
|
|
state["prom_alerted"] = prom_alerted
|
|
|
|
|
|
def check() -> tuple[bool, str]:
|
|
"""Return (ok, human-readable reason). Never reads 'status' field."""
|
|
status, body = http_get(f"{CONTROL_PLANE_URL}/summary")
|
|
|
|
if status is None:
|
|
return False, "panel unreachable (connection error)"
|
|
|
|
if status != 200:
|
|
return False, f"panel returned HTTP {status}"
|
|
|
|
if not body:
|
|
return False, "panel returned empty / invalid JSON"
|
|
|
|
raw = body.get("last_update")
|
|
if raw is None:
|
|
return False, "summary missing last_update field"
|
|
|
|
try:
|
|
last_update_ts = float(raw)
|
|
except (TypeError, ValueError):
|
|
return False, f"last_update not parseable: {raw!r}"
|
|
|
|
age = time.time() - last_update_ts
|
|
if age > STALE_THRESHOLD:
|
|
return False, (
|
|
f"brain stale: last update {int(age // 60)}m ago "
|
|
f"(threshold {STALE_THRESHOLD // 60}m)"
|
|
)
|
|
|
|
return True, f"ok (age {int(age)}s)"
|
|
|
|
|
|
def main() -> None:
|
|
print(
|
|
f"[brain-watchdog] starting — "
|
|
f"url={CONTROL_PLANE_URL} "
|
|
f"stale_threshold={STALE_THRESHOLD}s "
|
|
f"interval={INTERVAL}s "
|
|
f"fails_before_alert={FAILS_BEFORE_ALERT}",
|
|
flush=True,
|
|
)
|
|
state = load_state()
|
|
|
|
while True:
|
|
ok, reason = check()
|
|
ts = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
|
|
print(f"[{ts}] {'OK ' if ok else 'FAIL'} — {reason}", flush=True)
|
|
|
|
if ok:
|
|
if state["alerted"]:
|
|
send_telegram(
|
|
"✅ <b>brain-watchdog: control-plane RECOVERED</b>\n"
|
|
f"{reason}"
|
|
)
|
|
print("[telegram] sent recovery alert", flush=True)
|
|
state["fail_count"] = 0
|
|
state["alerted"] = False
|
|
state["last_ok"] = time.time()
|
|
save_state(state)
|
|
ping_healthchecks()
|
|
else:
|
|
state["fail_count"] = state.get("fail_count", 0) + 1
|
|
save_state(state)
|
|
|
|
if state["fail_count"] >= FAILS_BEFORE_ALERT and not state["alerted"]:
|
|
sent = send_telegram(
|
|
"🚨 <b>brain-watchdog: control-plane DOWN</b>\n"
|
|
f"Reason: {reason}\n"
|
|
f"Consecutive failures: {state['fail_count']}\n"
|
|
f"URL: <code>{CONTROL_PLANE_URL}</code>"
|
|
)
|
|
if sent:
|
|
state["alerted"] = True
|
|
save_state(state)
|
|
print("[telegram] sent alert", flush=True)
|
|
|
|
# --- Prometheus alerts (independent of brain watchdog) ---
|
|
handle_prometheus_alerts(state)
|
|
save_state(state)
|
|
|
|
time.sleep(INTERVAL)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|