From c4f09e9ced9478fa87e2f42020f9f02efcb5b1a8 Mon Sep 17 00:00:00 2001 From: Andreas Wrede Date: Fri, 1 May 2026 05:33:27 -0400 Subject: [PATCH] version 5.1.8 - fix: matrix/sms_voipms notifications blocked the event loop on timeout; make send_notification async, dispatch all channel drivers as non-blocking tasks (asyncio.to_thread for sync drivers, asyncio.wait_for for async); update all call sites to fire-and-forget via create_task - feat: add /about page with version, runtime, uptime counter, and repo link - fix: hbc_mini plugin data format now matches full hbc client so Host Overview displays memory, disk, and network metrics correctly Co-Authored-By: Claude Sonnet 4.6 (1M context) --- hbd/__init__.py | 2 +- hbd/server/cli.py | 11 +- hbd/server/http.py | 48 ++++++++ hbd/server/notify.py | 86 +++++--------- hbd/server/templates/about.html | 199 ++++++++++++++++++++++++++++++++ hbd/server/templates/nav.html | 1 + hbd/server/threshold.py | 45 +++----- hbd/server/udp.py | 20 ++-- pyproject.toml | 2 +- scripts/hbc_mini.py | 58 ++++++---- 10 files changed, 349 insertions(+), 123 deletions(-) create mode 100644 hbd/server/templates/about.html diff --git a/hbd/__init__.py b/hbd/__init__.py index 7aa2e6a..3622136 100644 --- a/hbd/__init__.py +++ b/hbd/__init__.py @@ -14,4 +14,4 @@ Install options: """ __all__ = ["__version__"] -__version__ = "5.1.7" +__version__ = "5.1.8" diff --git a/hbd/server/cli.py b/hbd/server/cli.py index 23b35b8..96f687a 100644 --- a/hbd/server/cli.py +++ b/hbd/server/cli.py @@ -144,17 +144,16 @@ def cmd_notify(args): url=f"{base_url}/plugins" if base_url else "", ) - # Bypass min_level for explicit test sends; run async channels directly import asyncio + from .notify import _send_matrix_async, _send_sms_voipms_async, _DRIVERS ch_type = channel_cfg.get("type", "") print(f"Sending via {args.channel} ({ch_type}): {title} — {args.message}") - if ch_type in ("matrix", "sms_voipms"): - from .notify import _send_matrix_async, _send_sms_voipms_async - driver_async = _send_matrix_async if ch_type == "matrix" else _send_sms_voipms_async - ok = asyncio.run(driver_async(channel_cfg, notif)) + if ch_type == "matrix": + ok = asyncio.run(_send_matrix_async(channel_cfg, notif)) + elif ch_type == "sms_voipms": + ok = asyncio.run(_send_sms_voipms_async(channel_cfg, notif)) else: - from .notify import _DRIVERS driver = _DRIVERS.get(ch_type) if driver is None: print(f"Error: unknown channel type '{ch_type}'", file=sys.stderr) diff --git a/hbd/server/http.py b/hbd/server/http.py index eee99b5..0c2b76d 100644 --- a/hbd/server/http.py +++ b/hbd/server/http.py @@ -1,7 +1,11 @@ """HTTP server implementation using aiohttp and jinja2.""" import asyncio +import datetime import json +import platform +import socket +import sys import time import urllib.parse import os @@ -111,6 +115,7 @@ async def start( This function is intended to be awaited inside the main asyncio event loop. """ get_now = get_now or (lambda: time.time()) + _start_epoch = time.time() async def old_index(request): _require_auth_redirect(request) @@ -806,6 +811,48 @@ async def start( ) return web.Response(text=body, content_type="text/html") + # ------------------------------------------------------------------------- + # About page + # ------------------------------------------------------------------------- + + async def about_page(request): + """GET /about — version, runtime, and project information.""" + current_user, _ = _require_auth_redirect(request) + pkg_dir = os.path.dirname(__file__) + templates_dir = config.get("templates_dir", os.path.join(pkg_dir, "templates")) + env = jinja2.Environment(loader=jinja2.FileSystemLoader(templates_dir)) + from hbd import __version__ as hbd_version + + uptime_secs = int(time.time() - _start_epoch) + days, rem = divmod(uptime_secs, 86400) + hours, rem = divmod(rem, 3600) + mins, secs = divmod(rem, 60) + if days: + uptime_str = f"{days}d {hours}h {mins}m" + elif hours: + uptime_str = f"{hours}h {mins}m {secs}s" + else: + uptime_str = f"{mins}m {secs}s" + + start_dt = datetime.datetime.fromtimestamp(_start_epoch) + start_time_str = start_dt.strftime("%Y-%m-%d %H:%M:%S") + + tmpl = env.get_template("about.html") + body = tmpl.render( + title="About - Heartbeat", + header="About", + hbd_version=hbd_version, + python_version=f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro} ({platform.python_implementation()})", + server_hostname=socket.gethostname(), + start_epoch=int(_start_epoch), + start_time_str=start_time_str, + uptime_str=uptime_str, + host_count=len(hbdclass.Host.hosts), + current_user=current_user.to_dict() if current_user else None, + active_page="about", + ) + return web.Response(text=body, content_type="text/html") + # ------------------------------------------------------------------------- # Settings page (admin only) # ------------------------------------------------------------------------- @@ -859,6 +906,7 @@ async def start( web.get("/live", live), web.get("/plugins", plugins_page), web.get("/alerts", alerts_page), + web.get("/about", about_page), web.get("/profile", profile_page), web.get("/settings", settings_page), web.get("/static/{path:.*}", static), diff --git a/hbd/server/notify.py b/hbd/server/notify.py index 172b3f2..a630e92 100644 --- a/hbd/server/notify.py +++ b/hbd/server/notify.py @@ -15,7 +15,6 @@ their own ``notification_channels`` list. When no users are configured the server runs silently (no notifications sent). """ -import asyncio import asyncio import logging import smtplib @@ -30,13 +29,10 @@ from . import ws as ws_mod logger = logging.getLogger(__name__) -logger = logging.getLogger(__name__) - msg_to_websockets = ws_mod.broadcast # Module-level state set via setup() _config: dict = {} -_loop: Optional[asyncio.AbstractEventLoop] = None # Tracks which channels fired a WARNING/CRITICAL per host. # {host_name: set of channel_names} — used to route RECOVER to the same channels. @@ -73,11 +69,9 @@ class Notification: # --------------------------------------------------------------------------- def setup(cfg: dict, loop: Optional[asyncio.AbstractEventLoop] = None): - """Initialize notifier from configuration dict and event loop.""" - global _config, _loop + """Initialize notifier from configuration dict.""" + global _config _config = dict(cfg) - if loop is not None: - _loop = loop def reload_config(cfg: dict): @@ -299,17 +293,6 @@ async def _send_sms_voipms_async(channel_cfg: dict, notif: Notification) -> bool return False -def _send_sms_voipms(channel_cfg: dict, notif: Notification) -> bool: - """Dispatch voip.ms SMS send onto the shared event loop.""" - if _loop is None: - logger.warning("sms_voipms: event loop not available") - return False - future = asyncio.run_coroutine_threadsafe(_send_sms_voipms_async(channel_cfg, notif), _loop) - try: - return future.result(timeout=15) - except Exception as e: - logger.error("sms_voipms send timed out or failed: %s", e) - return False async def _send_matrix_async(channel_cfg: dict, notif: Notification) -> bool: @@ -357,40 +340,23 @@ async def _send_matrix_async(channel_cfg: dict, notif: Notification) -> bool: await client.close() -def _send_matrix(channel_cfg: dict, notif: Notification) -> bool: - """Dispatch matrix send onto the shared event loop.""" - if _loop is None: - logger.warning("matrix: event loop not available") - return False - future = asyncio.run_coroutine_threadsafe(_send_matrix_async(channel_cfg, notif), _loop) - try: - return future.result(timeout=15) - except Exception as e: - logger.error("matrix send timed out or failed: %s", e) - return False - - # --------------------------------------------------------------------------- -# Channel dispatcher +# Channel dispatcher (all async — sync drivers run in a thread executor) # --------------------------------------------------------------------------- +# Sync drivers kept for `hbd notify` CLI usage (asyncio.run wraps them there). _DRIVERS = { "pushover": _send_pushover, "email": _send_email, "mattermost": _send_mattermost, "signal": _send_signal, - "sms_voipms": _send_sms_voipms, - "matrix": _send_matrix, } +_TIMEOUT = 15 # seconds per channel send -def _dispatch_to_channel(channel_name: str, channel_cfg: dict, notif: Notification) -> bool: - """Send *notif* to a single named channel, honouring min_level. - RECOVER always bypasses min_level — a recovery is always relevant if the - channel was configured for any alerting (handles the restart-then-recover case - where _alerted_channels is empty and we fall through to the normal loop). - """ +async def _dispatch_to_channel(channel_name: str, channel_cfg: dict, notif: Notification) -> bool: + """Send *notif* to a single named channel, honouring min_level.""" level = notif.level.upper() if level != "RECOVER": min_level = channel_cfg.get("min_level", "WARNING").upper() @@ -398,14 +364,24 @@ def _dispatch_to_channel(channel_name: str, channel_cfg: dict, notif: Notificati logger.debug( "channel '%s': skipping level %s (min_level=%s)", channel_name, level, min_level ) - return True # not an error — filtered intentionally + return True # filtered intentionally ch_type = channel_cfg.get("type", "") - driver = _DRIVERS.get(ch_type) - if driver is None: - logger.warning("unknown channel type '%s' for channel '%s'", ch_type, channel_name) + try: + if ch_type == "matrix": + return await asyncio.wait_for(_send_matrix_async(channel_cfg, notif), timeout=_TIMEOUT) + if ch_type == "sms_voipms": + return await asyncio.wait_for(_send_sms_voipms_async(channel_cfg, notif), timeout=_TIMEOUT) + sync_driver = _DRIVERS.get(ch_type) + if sync_driver is None: + logger.warning("unknown channel type '%s' for channel '%s'", ch_type, channel_name) + return False + return await asyncio.wait_for( + asyncio.to_thread(sync_driver, channel_cfg, notif), timeout=_TIMEOUT + ) + except asyncio.TimeoutError: + logger.error("channel '%s' timed out after %ds", channel_name, _TIMEOUT) return False - return driver(channel_cfg, notif) # --------------------------------------------------------------------------- @@ -419,7 +395,7 @@ def _build_url(host_name: str) -> str: return f"{base_url}/plugins#{host_name}" -def send_notification(host_name: str, notif: Notification) -> dict: +async def send_notification(host_name: str, notif: Notification) -> dict: """Dispatch *notif* to all managers/owner of *host_name*. Looks up the host's owner + managers, resolves each user's @@ -469,16 +445,12 @@ def send_notification(host_name: str, notif: Notification) -> dict: if not channel_cfg: continue try: - ch_type = channel_cfg.get("type", "") - driver = _DRIVERS.get(ch_type) - if driver: - ok = driver(channel_cfg, notif) - results[channel_name] = ok - if ok: - logger.info("recover sent to channel '%s': %s", channel_name, notif.title) + ok = await _dispatch_to_channel(channel_name, channel_cfg, notif) + results[channel_name] = ok + if ok: + logger.info("recover sent to channel '%s': %s", channel_name, notif.title) except Exception as e: logger.error("error sending recover to channel '%s': %s", channel_name, e) - # Clear the alerted set once recovery is delivered del _alerted_channels[host_name] return results @@ -489,14 +461,14 @@ def send_notification(host_name: str, notif: Notification) -> dict: continue for channel_name in user.notification_channels: if channel_name in results: - continue # already dispatched to this channel this notification + continue channel_cfg = global_channels.get(channel_name) if not channel_cfg: logger.warning("channel '%s' not defined in notification_channels", channel_name) results[channel_name] = False continue try: - ok = _dispatch_to_channel(channel_name, channel_cfg, notif) + ok = await _dispatch_to_channel(channel_name, channel_cfg, notif) results[channel_name] = ok if ok: logger.info("notification sent to channel '%s': %s", channel_name, notif.title) diff --git a/hbd/server/templates/about.html b/hbd/server/templates/about.html new file mode 100644 index 0000000..7756a4d --- /dev/null +++ b/hbd/server/templates/about.html @@ -0,0 +1,199 @@ + + + {% include 'head.html' %} + + + + + {% include 'nav.html' %} + +
+

{{ header }}

+

Heartbeat monitoring system

+ +
+
+
+ +
Lightweight host monitoring over UDP
+
+ v{{ hbd_version }} +
+
+ +
+

Version

+
+ Server version + {{ hbd_version }} +
+
+ Python + {{ python_version }} +
+
+ License + MIT +
+
+ +
+

Runtime

+
+ Host + {{ server_hostname }} +
+
+ Started + {{ start_time_str }} +
+
+ Uptime + {{ uptime_str }} +
+
+ Hosts monitored + {{ host_count }} +
+
+ +
+

Contact & Source

+
+ Author + Andreas Wrede +
+
+ Email + aew@wrede.ca +
+
+ Repository + git.wrede.ca/andreas/heartbeat +
+
+ +
+ + + + diff --git a/hbd/server/templates/nav.html b/hbd/server/templates/nav.html index 70467cb..88983df 100644 --- a/hbd/server/templates/nav.html +++ b/hbd/server/templates/nav.html @@ -9,6 +9,7 @@ {% if current_user and current_user.admin %} Settings {% endif %} + About
diff --git a/hbd/server/threshold.py b/hbd/server/threshold.py index b87e7d7..3e25060 100644 --- a/hbd/server/threshold.py +++ b/hbd/server/threshold.py @@ -987,18 +987,14 @@ class ThresholdChecker: value: Any, ): """Send notification and log to journal/eventlog.""" - try: - notify_mod.send_notification( - host_name, - notify_mod.Notification( - title=f"[{lvl}] {host_name}", - body=message, - level=lvl, - ), - ) - logger.info("Notification sent: %s", message) - except Exception as e: - logger.error("Failed to send notification: %s", e) + asyncio.get_event_loop().create_task(notify_mod.send_notification( + host_name, + notify_mod.Notification( + title=f"[{lvl}] {host_name}", + body=message, + level=lvl, + ), + )) # Log to journal if self.journal is not None: @@ -1195,20 +1191,17 @@ class ThresholdChecker: else: message = f"REMINDER ({alert_state.level.name}): {host_name} - {metric_path} = {value} (ongoing for {int(now - alert_state.since)}s)" - try: - notify_mod.send_notification( - host_name, - notify_mod.Notification( - title=f"[REMINDER/{alert_state.level.name}] {host_name}", - body=message, - level=alert_state.level.name, - ), - ) - alert_state.last_notification = now - alert_state.notification_count += 1 - logger.info("Re-notification sent: %s", message) - except Exception as e: - logger.error("Failed to send re-notification: %s", e) + asyncio.get_event_loop().create_task(notify_mod.send_notification( + host_name, + notify_mod.Notification( + title=f"[REMINDER/{alert_state.level.name}] {host_name}", + body=message, + level=alert_state.level.name, + ), + )) + alert_state.last_notification = now + alert_state.notification_count += 1 + logger.info("Re-notification sent: %s", message) def get_active_alerts(self, alert_states: Dict[str, AlertState]) -> list: """ diff --git a/hbd/server/udp.py b/hbd/server/udp.py index da6d6a6..1dc8734 100644 --- a/hbd/server/udp.py +++ b/hbd/server/udp.py @@ -211,10 +211,10 @@ def _make_timer_callbacks(uname, host, ctx): connection.newstate(connection.__class__.OVERDUE, now, cfg.get("grace", 2)) msg = f"{connection.afam} overdue" eventlog(uname, "CRITICAL", msg) - notify_mod.send_notification( + 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: @@ -407,10 +407,10 @@ def handle_datagram(msg: dict, addr, transport, ctx: dict): if res: eventlog(uname, "WARNING", res) - notify_mod.send_notification( + asyncio.create_task(notify_mod.send_notification( uname, notify_mod.Notification(title=f"[WARNING] {uname}", body=res, level="WARNING"), - ) + )) interval = int(msg.get("interval", 0) or 0) shutdown = msg.get("shutdown", 0) @@ -420,10 +420,10 @@ def handle_datagram(msg: dict, addr, transport, ctx: dict): if boot: eventlog(uname, "INFO", "booted") - notify_mod.send_notification( + asyncio.create_task(notify_mod.send_notification( uname, notify_mod.Notification(title=f"[INFO] {uname}", body=f"{host.name} booted", level="INFO"), - ) + )) if message: eventlog(uname, "INFO", "msg: %s" % message, service=service) @@ -440,10 +440,10 @@ def handle_datagram(msg: dict, addr, transport, ctx: dict): else: m = "%s back after being %s for %s" % (conn.afam, lasts, dur(d)) eventlog(uname, "RECOVER", m) - notify_mod.send_notification( + asyncio.create_task(notify_mod.send_notification( uname, notify_mod.Notification(title=f"[RECOVER] {uname}", body=m, level="RECOVER"), - ) + )) if boot or newh: host.upcount = host.doesack @@ -453,10 +453,10 @@ def handle_datagram(msg: dict, addr, transport, ctx: dict): if shutdown: m = "%s shutdown" % conn.afam eventlog(uname, "INFO", m) - notify_mod.send_notification( + asyncio.create_task(notify_mod.send_notification( uname, notify_mod.Notification(title=f"[INFO] {uname}", body=m, level="INFO"), - ) + )) conn.newstate(hbdcls.Connection.DOWN, now) _set_connectivity_alert(host, conn.afam, "CRITICAL") diff --git a/pyproject.toml b/pyproject.toml index e426cae..948a8e1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "hbd" -version = "5.1.7" +version = "5.1.8" description = "Heartbeat monitoring system — client (hbc) and server (hbd)" readme = "README.md" requires-python = ">=3.11" diff --git a/scripts/hbc_mini.py b/scripts/hbc_mini.py index 7ec9206..1fdb94d 100755 --- a/scripts/hbc_mini.py +++ b/scripts/hbc_mini.py @@ -233,7 +233,7 @@ class OSInfoPlugin(InfoPlugin): "machine": platform.machine(), "architecture": platform.architecture()[0], "python_version": platform.python_version(), - "hbc_version": "5.1.7", + "hbc_version": "5.1.8", "hbc_type": "mini", } if platform.system() == "Linux": @@ -531,19 +531,27 @@ class MemoryMonitorPlugin(MonitorPlugin): return {} total = mi.get("MemTotal", 0) avail = mi.get("MemAvailable", mi.get("MemFree", 0)) + free = mi.get("MemFree", 0) used = total - avail data: Dict[str, Any] = { - "mem_total_kb": total, - "mem_used_kb": used, - "mem_available_kb": avail, - "mem_percent": round(100.0 * used / total, 1) if total else 0.0, + "memory_total": total * 1024, + "memory_used": used * 1024, + "memory_available": avail * 1024, + "memory_free": free * 1024, + "memory_percent": round(100.0 * used / total, 1) if total else 0.0, } + for field, key in (("Buffers", "memory_buffers"), ("Cached", "memory_cached"), + ("Active", "memory_active"), ("Inactive", "memory_inactive")): + if field in mi: + data[key] = mi[field] * 1024 stotal = mi.get("SwapTotal", 0) if stotal: sfree = mi.get("SwapFree", 0) - data["swap_total_kb"] = stotal - data["swap_used_kb"] = stotal - sfree - data["swap_percent"] = round(100.0 * (stotal - sfree) / stotal, 1) + sused = stotal - sfree + data["swap_total"] = stotal * 1024 + data["swap_used"] = sused * 1024 + data["swap_free"] = sfree * 1024 + data["swap_percent"] = round(100.0 * sused / stotal, 1) return data @@ -579,7 +587,7 @@ class DiskMonitorPlugin(MonitorPlugin): except Exception as e: self.logger.warning("df failed: %s", e) return {} - data: Dict[str, Any] = {} + partitions: Dict[str, Any] = {} for line in out.decode(errors="replace").splitlines()[1:]: parts = line.split() if len(parts) < 6: @@ -588,14 +596,19 @@ class DiskMonitorPlugin(MonitorPlugin): if self.mounts and mount not in self.mounts: continue try: - key = re.sub(r"[^a-zA-Z0-9_]", "_", mount).strip("_") or "root" - data[f"{key}_total_kb"] = int(parts[1]) - data[f"{key}_used_kb"] = int(parts[2]) - data[f"{key}_avail_kb"] = int(parts[3]) - data[f"{key}_percent"] = int(parts[4].rstrip("%")) + total_kb = int(parts[1]) + used_kb = int(parts[2]) + avail_kb = int(parts[3]) + pct = int(parts[4].rstrip("%")) + partitions[mount] = { + "total": total_kb * 1024, + "used": used_kb * 1024, + "free": avail_kb * 1024, + "percent": pct, + } except (ValueError, IndexError): continue - return data + return {"partitions": partitions} if partitions else {} # --------------------------------------------------------------------------- @@ -651,17 +664,18 @@ class NetworkMonitorPlugin(MonitorPlugin): self._prev = (now, curr) if dt <= 0: return {} - data: Dict[str, Any] = {} + interfaces: Dict[str, Any] = {} for iface, (rx, tx) in curr.items(): if iface in self.skip_ifaces or iface not in prev: continue prx, ptx = prev[iface] - key = re.sub(r"[^a-zA-Z0-9_]", "_", iface) - data[f"{key}_rx_bps"] = round((rx - prx) / dt) - data[f"{key}_tx_bps"] = round((tx - ptx) / dt) - data[f"{key}_rx_bytes"] = rx - data[f"{key}_tx_bytes"] = tx - return data + interfaces[iface] = { + "bytes_recv": rx, + "bytes_sent": tx, + "bytes_recv_delta": rx - prx, + "bytes_sent_delta": tx - ptx, + } + return {"interfaces": interfaces} if interfaces else {} # ---------------------------------------------------------------------------