From f603ef9bd229fd8ad17feaf1a951a12c2e98a098 Mon Sep 17 00:00:00 2001 From: Andreas Wrede Date: Fri, 26 Jun 2026 11:37:48 -0400 Subject: [PATCH] fix: declare plugin interval so stale data waits two full intervals The server inferred each plugin's collection interval from the gap between the last two received PLG samples, then expired data at interval * 3. After an outage this guess was wrong: the request_update re-send of collect-once InfoPlugins produced two close samples, yielding a tiny inferred interval that purged permanent info data minutes after recovery. Clients now declare each plugin's interval in the PLG message (_interval). The server uses it directly (from the first post-recovery sample), expiring at interval * 3 so live data survives at least two full intervals; interval 0 marks collect-once InfoPlugins that never go stale. Legacy clients omit the field and fall back to the previous inferred-gap behavior. Adds _interval to all three clients: hbc, hbc_mini.py, hbc_mini.c. Co-Authored-By: Claude Opus 4.8 --- hbd/client/main.py | 9 +++++---- hbd/server/udp.py | 40 +++++++++++++++++++++++++++++----------- scripts/c/hbc_mini.c | 15 +++++++++++---- scripts/hbc_mini.py | 4 ++-- 4 files changed, 47 insertions(+), 21 deletions(-) diff --git a/hbd/client/main.py b/hbd/client/main.py index d09c7e9..6885da1 100644 --- a/hbd/client/main.py +++ b/hbd/client/main.py @@ -356,7 +356,8 @@ async def _info_plugin_refresh_loop(conn: AsyncConnection, info_plugins: List): try: data = await plugin.collect() if data: - await conn.sendto({"plugin": plugin.name, **data}, "PLG") + await conn.sendto( + {"plugin": plugin.name, **data, "_interval": plugin.interval}, "PLG") logger.info(f"Resent {plugin.name} data") except Exception as e: logger.error(f"Error re-collecting {plugin.name}: {e}", exc_info=True) @@ -377,8 +378,8 @@ async def plugin_collector(conn: AsyncConnection, registry: PluginRegistry): try: data = await plugin.collect() if data: - # Create PLG message with plugin name - plugin_msg = {"plugin": plugin.name, **data} + # Create PLG message with plugin name and declared interval + plugin_msg = {"plugin": plugin.name, **data, "_interval": plugin.interval} await conn.sendto(plugin_msg, "PLG") logger.info(f"Sent {plugin.name} data") except Exception as e: @@ -430,7 +431,7 @@ async def plugin_collector_interval( data = await plugin.collect() if data: # Don't use encode_plugin_data - create dict directly - plugin_msg = {"plugin": plugin.name, **data} + plugin_msg = {"plugin": plugin.name, **data, "_interval": plugin.interval} await conn.sendto(plugin_msg, "PLG") logger.debug(f"Sent {plugin.name} data") except asyncio.CancelledError: diff --git a/hbd/server/udp.py b/hbd/server/udp.py index 5fb0037..0ff209b 100644 --- a/hbd/server/udp.py +++ b/hbd/server/udp.py @@ -16,6 +16,11 @@ from . import notify as notify_mod logger = logging.getLogger(__name__) eventlog = notify_mod.eventlog +# Plugin data is declared stale after this many collection intervals without a +# fresh sample. >= 2 so a single missed sample never purges live data (the user +# directive: wait at least two full intervals after recovery before going stale). +_STALE_INTERVAL_MULTIPLIER = 3 + # SO_TIMESTAMP: kernel attaches a struct timeval to each received datagram. # Supported on Linux, FreeBSD, and macOS. The constant is not exposed by # Python's socket module on all platforms @@ -394,21 +399,36 @@ def handle_datagram(msg: dict, addr, transport, ctx: dict): plugin_name = msg.get("plugin") if plugin_name: # Extract plugin fields, dropping protocol metadata fields + # (_interval is the client-declared collection interval, not a metric). plugin_data = {k: v for k, v in msg.items() - if k not in ("ID", "plugin", "id", "name")} + if k not in ("ID", "plugin", "id", "name", "_interval")} # Store plugin data with timestamp host.add_plugin_data(plugin_name, plugin_data, timestamp=now) - # Reset stale timer using the observed send interval for this plugin. - # We need two samples to know the real interval; on the first sample - # we cancel any leftover timer but don't set a new one, to avoid - # false-stale firing for slow plugins (e.g. nagios_runner at 300 s). + + # Set the stale timer. Prefer the interval the client declares: it is + # the true collection cadence and is correct from the very first + # post-recovery sample, so we never purge data too early after an + # outage. interval == 0 means a collect-once InfoPlugin that never + # goes stale. Legacy clients omit _interval, so fall back to inferring + # the cadence from the gap between the last two received samples. history = host.plugin_data.get(plugin_name, []) - if len(history) >= 2: + declared = msg.get("_interval") + if declared is not None: + if declared > 0: + host.reset_plugin_timer(plugin_name, declared * _STALE_INTERVAL_MULTIPLIER, + _make_plugin_stale_callback(uname, ctx)) + else: + host.cancel_plugin_timer(plugin_name) + elif len(history) >= 2: plugin_interval = max(history[-1][0] - history[-2][0], 1) - host.reset_plugin_timer(plugin_name, plugin_interval * 3, + host.reset_plugin_timer(plugin_name, plugin_interval * _STALE_INTERVAL_MULTIPLIER, _make_plugin_stale_callback(uname, ctx)) - # Remove alert states for metrics present in the previous sample - # but absent now (e.g. a nagios check removed from configuration). + else: + host.cancel_plugin_timer(plugin_name) + + # Remove alert states for metrics present in the previous sample but + # absent now (e.g. a nagios check removed from configuration). + if len(history) >= 2: prev_keys = set(history[-2][1].keys()) curr_keys = set(plugin_data.keys()) for metric_name in prev_keys - curr_keys: @@ -417,8 +437,6 @@ def handle_datagram(msg: dict, addr, transport, ctx: dict): eventlog(uname, "INFO", f"stale check removed: {metric_path}") if (prev_keys - curr_keys) and msg_to_websockets: msg_to_websockets("host", host.stateinfo()) - else: - host.cancel_plugin_timer(plugin_name) # If os_info reports an owner and none is configured server-side, apply it if plugin_name == "os_info": diff --git a/scripts/c/hbc_mini.c b/scripts/c/hbc_mini.c index 155f650..f5ded92 100644 --- a/scripts/c/hbc_mini.c +++ b/scripts/c/hbc_mini.c @@ -667,6 +667,7 @@ static void plugin_os_info(conn_t *c, const config_t *cfg) { if (osver[0]) kv_set(&d, "distro_version_id", osver); } #endif + kv_set_int(&d, "_interval", 0); /* InfoPlugin: collect-once, never stale */ conn_send(c, "PLG", &d); LOGI("sent os_info"); } @@ -781,6 +782,7 @@ static void plugin_cpu_monitor(conn_t *c, const config_t *cfg) { } read_cpu_extras(&d); kv_set_dbl(&d, "_timestamp", now_ts()); + kv_set_int(&d, "_interval", cfg->cpu_interval); conn_send(c, "PLG", &d); LOGD("sent cpu_monitor"); } @@ -796,7 +798,7 @@ static void plugin_cpu_monitor(conn_t *c, const config_t *cfg) { static void mem_send(conn_t *c, long long tot, long long used, long long av, long long fr, long long act, long long ina, long long cac, long long buf, - long long stot, long long sused) { + long long stot, long long sused, int interval) { kvdict_t d; kv_clear(&d); kv_set(&d, "plugin", "memory_monitor"); kv_set_ull(&d, "memory_total", (unsigned long long)tot); @@ -819,6 +821,7 @@ static void mem_send(conn_t *c, kv_set(&d, "swap_percent", pct); } kv_set_dbl(&d, "_timestamp", now_ts()); + kv_set_int(&d, "_interval", interval); conn_send(c, "PLG", &d); LOGD("sent memory_monitor"); } @@ -863,7 +866,7 @@ static void plugin_memory_monitor(conn_t *c, const config_t *cfg) { /* values from /proc/meminfo are in kB */ mem_send(c, tot*1024, used*1024, av*1024, fr*1024, act*1024, ina*1024, cac*1024, buf*1024, - stot*1024, (stot-sfr)*1024); + stot*1024, (stot-sfr)*1024, cfg->mem_interval); } #elif defined(__FreeBSD__) || defined(__DragonFly__) @@ -889,7 +892,7 @@ static void plugin_memory_monitor(conn_t *c, const config_t *cfg) { long long cac = (long long)v_cache * ps; long long av = fr + ina + cac; if (av > tot) av = tot; long long used = tot - av; - mem_send(c, tot, used, av, fr, act, ina, cac, 0, 0, 0); + mem_send(c, tot, used, av, fr, act, ina, cac, 0, 0, 0, cfg->mem_interval); } #elif defined(__NetBSD__) @@ -910,7 +913,7 @@ static void plugin_memory_monitor(conn_t *c, const config_t *cfg) { long long used = tot - av; long long stot = (long long)uvm.swpages * ps; long long sinuse = (long long)uvm.swpginuse * ps; - mem_send(c, tot, used, av, fr, act, ina, 0, 0, stot, sinuse); + mem_send(c, tot, used, av, fr, act, ina, 0, 0, stot, sinuse, cfg->mem_interval); } #endif /* platform memory */ @@ -962,6 +965,7 @@ static void plugin_disk_monitor(conn_t *c, const config_t *cfg) { char *jval = malloc(MAX_VAL + 1); if (jval) { snprintf(jval, MAX_VAL, "@%s", json); kv_set(&d, "partitions", jval); free(jval); } kv_set_dbl(&d, "_timestamp", now_ts()); + kv_set_int(&d, "_interval", cfg->disk_interval); conn_send(c, "PLG", &d); free(json); LOGD("sent disk_monitor"); @@ -1067,6 +1071,7 @@ static void plugin_network_monitor(conn_t *c, const config_t *cfg) { char *jval = malloc(MAX_VAL + 1); if (jval) { snprintf(jval, MAX_VAL, "@%s", json); kv_set(&d, "interfaces", jval); free(jval); } kv_set_dbl(&d, "_timestamp", now_ts()); + kv_set_int(&d, "_interval", cfg->net_interval); conn_send(c, "PLG", &d); free(json); LOGD("sent network_monitor"); @@ -1125,6 +1130,7 @@ static void plugin_ping_monitor(conn_t *c, const config_t *cfg) { } } kv_set_dbl(&d, "_timestamp", now_ts()); + kv_set_int(&d, "_interval", cfg->ping_interval); conn_send(c, "PLG", &d); LOGD("sent ping_monitor"); } @@ -1194,6 +1200,7 @@ static void plugin_nagios_runner(conn_t *c, const config_t *cfg) { parse_perfdata(output, &d, name); } kv_set_dbl(&d, "_timestamp", now_ts()); + kv_set_int(&d, "_interval", cfg->nagios_interval); conn_send(c, "PLG", &d); LOGD("sent nagios_runner"); } diff --git a/scripts/hbc_mini.py b/scripts/hbc_mini.py index 80e73ee..de27545 100755 --- a/scripts/hbc_mini.py +++ b/scripts/hbc_mini.py @@ -955,7 +955,7 @@ async def _run_info_plugins(conn: AsyncConnection, plugins: List[Plugin]): try: data = await plugin.collect() if data: - await conn.sendto({"plugin": plugin.name, **data}, "PLG") + await conn.sendto({"plugin": plugin.name, **data, "_interval": plugin.interval}, "PLG") log.info("sent %s", plugin.name) except Exception as e: log.error("%s collect: %s", plugin.name, e) @@ -968,7 +968,7 @@ async def _run_monitor_group(conn: AsyncConnection, plugins: List[Plugin], inter try: data = await plugin.collect() if data: - await conn.sendto({"plugin": plugin.name, **data}, "PLG") + await conn.sendto({"plugin": plugin.name, **data, "_interval": plugin.interval}, "PLG") log.debug("sent %s", plugin.name) except asyncio.CancelledError: raise