feat(server): monitor only present protocols; host-level overdue and simplex alerts
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Uc9V1v7iXruubpoLiSE24M
This commit is contained in:
+89
-22
@@ -194,6 +194,40 @@ def _set_connectivity_alert(host, afam, level_name):
|
||||
state.update(level, level_name)
|
||||
|
||||
|
||||
_PRESENCE_KEYS = {"p4": "IPv4", "p6": "IPv6"}
|
||||
|
||||
|
||||
def _apply_presence(host, msg, now, uname, ctx, skip_conn):
|
||||
"""Record client-reported protocol presence and settle the affected connections.
|
||||
|
||||
A family reported absent whose connection is already overdue/unknown is parked as
|
||||
ABSENT (timer cancelled, connectivity alert cleared). One still UP is left alone; its
|
||||
timer expiry parks it. A family reported present again while ABSENT restarts a grace
|
||||
timer. The connection that carried this very heartbeat (skip_conn) is never touched.
|
||||
"""
|
||||
cfg = ctx.get("config", {})
|
||||
for key, afam in _PRESENCE_KEYS.items():
|
||||
if key not in msg:
|
||||
continue
|
||||
present = bool(msg[key])
|
||||
host.set_presence(afam, present, now)
|
||||
conn = host.connections.get(afam)
|
||||
if conn is None or conn is skip_conn:
|
||||
continue
|
||||
cls = conn.__class__
|
||||
if not present:
|
||||
if conn.getstate() in (cls.OVERDUE, cls.UNKNOWN):
|
||||
conn.cancel_overdue_timer()
|
||||
conn.newstate(cls.ABSENT, now)
|
||||
_set_connectivity_alert(host, afam, "OK")
|
||||
elif conn.getstate() == cls.ABSENT:
|
||||
conn.newstate(cls.UNKNOWN, now)
|
||||
interval = int(msg.get("interval", 0) or 0) or host.interval
|
||||
if interval > 0:
|
||||
on_overdue, _ = _make_timer_callbacks(uname, host, ctx)
|
||||
conn.reset_overdue_timer(interval + cfg.get("grace", 2), on_overdue)
|
||||
|
||||
|
||||
def _make_timer_callbacks(uname, host, ctx):
|
||||
"""Return (on_overdue, on_unknown) async callbacks for connection timer logic.
|
||||
|
||||
@@ -210,26 +244,46 @@ def _make_timer_callbacks(uname, host, ctx):
|
||||
msg_to_websockets("host", host.stateinfo())
|
||||
|
||||
async def on_overdue(connection):
|
||||
if connection.getstate() != connection.__class__.UP:
|
||||
cls = connection.__class__
|
||||
if connection.getstate() not in (cls.UP, cls.UNKNOWN):
|
||||
return
|
||||
now = time.time()
|
||||
connection.newstate(connection.__class__.OVERDUE, now, cfg.get("grace", 2))
|
||||
msg = f"{connection.afam} overdue"
|
||||
eventlog(uname, "CRITICAL", msg)
|
||||
if host.watched:
|
||||
asyncio.create_task(notify_mod.send_notification(
|
||||
uname,
|
||||
notify_mod.Notification(title=f"[CRITICAL] {uname}", body=msg, level="CRITICAL"),
|
||||
))
|
||||
# Track in alert_states so the Alerts Dashboard shows this
|
||||
_set_connectivity_alert(host, connection.afam, "CRITICAL")
|
||||
if threshold_checker:
|
||||
threshold_checker.check_value(
|
||||
host_name=uname,
|
||||
metric_path="rtt",
|
||||
value=float("inf"),
|
||||
alert_states=host.alert_states,
|
||||
)
|
||||
afam = connection.afam
|
||||
if not host.flag_present(afam):
|
||||
# The client says this protocol has no connectivity: not an outage.
|
||||
connection.newstate(cls.ABSENT, now)
|
||||
connection.cancel_overdue_timer()
|
||||
_set_connectivity_alert(host, afam, "OK")
|
||||
if msg_to_websockets:
|
||||
msg_to_websockets("host", host.stateinfo())
|
||||
return
|
||||
|
||||
prev = host.status()
|
||||
connection.newstate(cls.OVERDUE, now, cfg.get("grace", 2))
|
||||
new = host.status()
|
||||
if new == "overdue" and prev != "overdue":
|
||||
names = ", ".join(sorted(af for af, c in host.connections.items()
|
||||
if c.getstate() == cls.OVERDUE))
|
||||
msg = f"{names} overdue"
|
||||
eventlog(uname, "CRITICAL", msg)
|
||||
if host.watched:
|
||||
asyncio.create_task(notify_mod.send_notification(
|
||||
uname,
|
||||
notify_mod.Notification(title=f"[CRITICAL] {uname}", body=msg, level="CRITICAL"),
|
||||
))
|
||||
# Track in alert_states so the Alerts Dashboard shows this
|
||||
for af, c in host.connections.items():
|
||||
if c.getstate() == cls.OVERDUE:
|
||||
_set_connectivity_alert(host, af, "CRITICAL")
|
||||
if threshold_checker:
|
||||
threshold_checker.check_value(
|
||||
host_name=uname,
|
||||
metric_path="rtt",
|
||||
value=float("inf"),
|
||||
alert_states=host.alert_states,
|
||||
)
|
||||
elif new == "simplex":
|
||||
eventlog(uname, "INFO", f"{afam} overdue (simplex)")
|
||||
if msg_to_websockets:
|
||||
msg_to_websockets("host", host.stateinfo())
|
||||
connection.reset_overdue_timer(DROPOVERDUE, on_unknown)
|
||||
@@ -270,6 +324,8 @@ def restore_connection_timers(hbdclass, ctx):
|
||||
interval = host.interval
|
||||
for afam, conn in list(host.connections.items()):
|
||||
state = conn.getstate()
|
||||
if state == hbdclass.Connection.ABSENT:
|
||||
continue
|
||||
if state == hbdclass.Connection.DOWN:
|
||||
_set_connectivity_alert(host, afam, "CRITICAL")
|
||||
continue
|
||||
@@ -313,7 +369,8 @@ def restore_connection_timers(hbdclass, ctx):
|
||||
# Ensure the connectivity alert is set — it may be missing if
|
||||
# hbd was shut down before the on_overdue callback had a chance
|
||||
# to record it.
|
||||
_set_connectivity_alert(host, afam, "CRITICAL")
|
||||
if host.status() == "overdue":
|
||||
_set_connectivity_alert(host, afam, "CRITICAL")
|
||||
restored += 1
|
||||
|
||||
logger.info("Restored timers for %d connection(s)", restored)
|
||||
@@ -517,6 +574,9 @@ def handle_datagram(msg: dict, addr, transport, ctx: dict):
|
||||
if message:
|
||||
eventlog(uname, "INFO", message, service=service)
|
||||
|
||||
_apply_presence(host, msg, now, uname, ctx, conn)
|
||||
prev_status = host.status()
|
||||
|
||||
if conn.getstate() != hbdcls.Connection.UP:
|
||||
# Transition to UP and log/notify if appropriate
|
||||
lasts = conn.state
|
||||
@@ -538,8 +598,12 @@ def handle_datagram(msg: dict, addr, transport, ctx: dict):
|
||||
]
|
||||
for k in stale_plugin_keys:
|
||||
del host.alert_states[k]
|
||||
# Clear connectivity alert now that the host is back up
|
||||
# Clear connectivity alert now that the host is back up; once the host is no longer
|
||||
# overdue no family may keep a connectivity alert (e.g. the simplex case).
|
||||
_set_connectivity_alert(host, conn.afam, "OK")
|
||||
if host.status() in ("up", "simplex"):
|
||||
for af in host.connections:
|
||||
_set_connectivity_alert(host, af, "OK")
|
||||
# Don't log/notify RECOVER for a brand-new host seen for the first time —
|
||||
# it was never down, it just hasn't been seen before.
|
||||
if not newh:
|
||||
@@ -551,8 +615,11 @@ def handle_datagram(msg: dict, addr, transport, ctx: dict):
|
||||
else:
|
||||
m = "%s back after being %s for %s" % (conn.afam, lasts, dur(d))
|
||||
if m:
|
||||
eventlog(uname, "RECOVER", m)
|
||||
if host.watched:
|
||||
# RECOVER (and a notification) only if the host itself was overdue/down;
|
||||
# a single family coming back while another stayed up is informational.
|
||||
level = "RECOVER" if prev_status in ("overdue", "down", "unknown") else "INFO"
|
||||
eventlog(uname, level, m)
|
||||
if host.watched and level == "RECOVER":
|
||||
asyncio.create_task(notify_mod.send_notification(
|
||||
uname,
|
||||
notify_mod.Notification(title=f"[RECOVER] {uname}", body=m, level="RECOVER"),
|
||||
|
||||
@@ -0,0 +1,173 @@
|
||||
"""Host-level overdue alerting driven by client-reported protocol presence."""
|
||||
import asyncio
|
||||
import time
|
||||
|
||||
from hbd.common.proto import dicttos
|
||||
from hbd.server import hbdclass, udp
|
||||
from hbd.server.hbdclass import Connection as C
|
||||
from hbd.server.udp import handle_datagram, parse_message
|
||||
|
||||
|
||||
class _FakeTransport:
|
||||
def sendto(self, data, addr):
|
||||
pass
|
||||
|
||||
|
||||
def _ctx():
|
||||
return {"config": {}, "hbdclass": hbdclass, "msg_to_websockets": None,
|
||||
"DEBUG": 0, "verbose": False}
|
||||
|
||||
|
||||
def _fresh(name):
|
||||
hbdclass.Host.hosts.pop(name, None)
|
||||
|
||||
|
||||
def _send(name, addr, ctx, **fields):
|
||||
msg = parse_message(dicttos("HTB", {"name": name, "id": 0, **fields}))
|
||||
handle_datagram(msg, (addr, 50000), _FakeTransport(), ctx)
|
||||
|
||||
|
||||
def _events(monkeypatch):
|
||||
events = []
|
||||
monkeypatch.setattr(udp, "eventlog", lambda u, level, msg, **kw: events.append((level, msg)))
|
||||
return events
|
||||
|
||||
|
||||
def _dual_stack(name, ctx):
|
||||
_fresh(name)
|
||||
_send(name, "127.0.0.1", ctx)
|
||||
_send(name, "::1", ctx)
|
||||
return hbdclass.Host.hosts[name]
|
||||
|
||||
|
||||
def test_flags_are_stored_on_host():
|
||||
ctx = _ctx()
|
||||
_fresh("pres-store")
|
||||
_send("pres-store", "127.0.0.1", ctx, p4=1, p6=0)
|
||||
h = hbdclass.Host.hosts["pres-store"]
|
||||
assert h.presence["IPv4"]["present"] is True
|
||||
assert h.presence["IPv6"]["present"] is False
|
||||
|
||||
|
||||
def test_no_flags_leaves_presence_unknown():
|
||||
ctx = _ctx()
|
||||
_fresh("pres-legacy")
|
||||
_send("pres-legacy", "127.0.0.1", ctx)
|
||||
assert hbdclass.Host.hosts["pres-legacy"].presence == {}
|
||||
|
||||
|
||||
def test_absent_flag_moves_overdue_family_to_absent_and_clears_alert(monkeypatch):
|
||||
_events(monkeypatch)
|
||||
ctx = _ctx()
|
||||
h = _dual_stack("pres-absent", ctx)
|
||||
v6 = h.connections["IPv6"]
|
||||
v6.state = C.OVERDUE
|
||||
udp._set_connectivity_alert(h, "IPv6", "CRITICAL")
|
||||
_send("pres-absent", "127.0.0.1", ctx, p4=1, p6=0)
|
||||
assert v6.state == C.ABSENT
|
||||
assert "connectivity.IPv6" not in h.alert_states
|
||||
assert h.status() == "up"
|
||||
|
||||
|
||||
def test_absent_flag_does_not_touch_family_that_is_up():
|
||||
ctx = _ctx()
|
||||
h = _dual_stack("pres-up-stays", ctx)
|
||||
_send("pres-up-stays", "127.0.0.1", ctx, p4=1, p6=0)
|
||||
assert h.connections["IPv6"].state == C.UP # traffic overrides the flag
|
||||
|
||||
|
||||
def test_present_again_arms_grace_timer():
|
||||
ctx = _ctx()
|
||||
|
||||
async def scenario():
|
||||
_fresh("pres-back")
|
||||
_send("pres-back", "127.0.0.1", ctx, interval=10)
|
||||
_send("pres-back", "::1", ctx, interval=10)
|
||||
h = hbdclass.Host.hosts["pres-back"]
|
||||
v6 = h.connections["IPv6"]
|
||||
v6.cancel_overdue_timer()
|
||||
v6.state = C.ABSENT
|
||||
_send("pres-back", "127.0.0.1", ctx, interval=10, p6=1)
|
||||
assert v6.state == C.UNKNOWN and v6.overdue_timer is not None
|
||||
for c in h.connections.values():
|
||||
c.cancel_overdue_timer()
|
||||
|
||||
asyncio.run(scenario())
|
||||
|
||||
|
||||
def test_one_family_overdue_is_simplex_all_overdue_is_critical(monkeypatch):
|
||||
events = _events(monkeypatch)
|
||||
ctx = _ctx()
|
||||
h = _dual_stack("pres-simplex", ctx)
|
||||
on_overdue, _ = udp._make_timer_callbacks("pres-simplex", h, ctx)
|
||||
|
||||
asyncio.run(on_overdue(h.connections["IPv6"]))
|
||||
assert h.status() == "simplex"
|
||||
assert ("INFO", "IPv6 overdue (simplex)") in events
|
||||
assert not any(level == "CRITICAL" for level, _ in events)
|
||||
assert "connectivity.IPv6" not in h.alert_states
|
||||
|
||||
asyncio.run(on_overdue(h.connections["IPv4"]))
|
||||
assert h.status() == "overdue"
|
||||
assert [e for e in events if e[0] == "CRITICAL"] == [("CRITICAL", "IPv4, IPv6 overdue")]
|
||||
assert {"connectivity.IPv4", "connectivity.IPv6"} <= set(h.alert_states)
|
||||
|
||||
|
||||
def test_timer_expiry_on_absent_family_is_silent(monkeypatch):
|
||||
events = _events(monkeypatch)
|
||||
ctx = _ctx()
|
||||
h = _dual_stack("pres-quiet", ctx)
|
||||
_send("pres-quiet", "127.0.0.1", ctx, p4=1, p6=0) # v6 still UP => left alone
|
||||
events.clear()
|
||||
on_overdue, _ = udp._make_timer_callbacks("pres-quiet", h, ctx)
|
||||
asyncio.run(on_overdue(h.connections["IPv6"]))
|
||||
assert h.connections["IPv6"].state == C.ABSENT
|
||||
assert events == []
|
||||
assert h.status() == "up"
|
||||
|
||||
|
||||
def test_recovery_from_simplex_is_info_only(monkeypatch):
|
||||
events = _events(monkeypatch)
|
||||
ctx = _ctx()
|
||||
h = _dual_stack("pres-rec-simplex", ctx)
|
||||
v6 = h.connections["IPv6"]
|
||||
v6.state, v6.statetime = C.OVERDUE, time.time() - 100
|
||||
events.clear()
|
||||
_send("pres-rec-simplex", "::1", ctx)
|
||||
assert v6.state == C.UP
|
||||
assert [e[0] for e in events] == ["INFO"] and "back after" in events[0][1]
|
||||
|
||||
|
||||
def test_recovery_from_overdue_is_recover_and_clears_all_alerts(monkeypatch):
|
||||
events = _events(monkeypatch)
|
||||
ctx = _ctx()
|
||||
h = _dual_stack("pres-rec-overdue", ctx)
|
||||
for afam in ("IPv4", "IPv6"):
|
||||
c = h.connections[afam]
|
||||
c.state, c.statetime = C.OVERDUE, time.time() - 100
|
||||
udp._set_connectivity_alert(h, afam, "CRITICAL")
|
||||
events.clear()
|
||||
_send("pres-rec-overdue", "127.0.0.1", ctx)
|
||||
assert [e[0] for e in events] == ["RECOVER"]
|
||||
assert h.status() == "simplex"
|
||||
assert "connectivity.IPv6" not in h.alert_states # left overdue => all cleared
|
||||
assert "connectivity.IPv4" not in h.alert_states
|
||||
|
||||
|
||||
def test_restore_skips_absent_and_sets_no_alert_for_simplex():
|
||||
ctx = _ctx()
|
||||
|
||||
async def scenario():
|
||||
h = _dual_stack("pres-restore", ctx)
|
||||
h.connections["IPv6"].state = C.OVERDUE # v4 UP + v6 OVERDUE => simplex
|
||||
udp.restore_connection_timers(hbdclass, ctx)
|
||||
assert "connectivity.IPv6" not in h.alert_states
|
||||
h.connections["IPv6"].cancel_overdue_timer()
|
||||
h.connections["IPv6"].state = C.ABSENT
|
||||
h.set_presence("IPv6", False, time.time())
|
||||
udp.restore_connection_timers(hbdclass, ctx)
|
||||
assert h.connections["IPv6"].state == C.ABSENT
|
||||
for c in h.connections.values():
|
||||
c.cancel_overdue_timer()
|
||||
|
||||
asyncio.run(scenario())
|
||||
Reference in New Issue
Block a user