From c3736f47ee433dc7378207d94a2654c7a935ffbc Mon Sep 17 00:00:00 2001 From: Andreas Wrede Date: Mon, 21 Sep 2026 09:28:26 -0400 Subject: [PATCH] feat(server): monitor only present protocols; host-level overdue and simplex alerts Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01Uc9V1v7iXruubpoLiSE24M --- hbd/server/udp.py | 111 +++++++++++++++++++----- tests/test_udp_presence.py | 173 +++++++++++++++++++++++++++++++++++++ 2 files changed, 262 insertions(+), 22 deletions(-) create mode 100644 tests/test_udp_presence.py diff --git a/hbd/server/udp.py b/hbd/server/udp.py index f3bcd63..76847a5 100644 --- a/hbd/server/udp.py +++ b/hbd/server/udp.py @@ -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"), diff --git a/tests/test_udp_presence.py b/tests/test_udp_presence.py new file mode 100644 index 0000000..260d894 --- /dev/null +++ b/tests/test_udp_presence.py @@ -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())