andreas/heartbeat
Public Access
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases 50 Wiki Activity

fix: declare plugin interval so stale data waits two full intervals

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
andreas
2026-06-26 11:37:48 -04:00
co-authored by Claude Opus 4.8
parent 731545dd0e
commit f603ef9bd2
4 changed files with 47 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hbd/client/main.py
+5 -4
View File
@@ -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:
hbd/server/udp.py
+30 -12
View File
@@ -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:
plugin_interval = max(history[-1][0] - history[-2][0], 1)
host.reset_plugin_timer(plugin_name, plugin_interval * 3,
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))
# 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)
elif len(history) >= 2:
plugin_interval = max(history[-1][0] - history[-2][0], 1)
host.reset_plugin_timer(plugin_name, plugin_interval * _STALE_INTERVAL_MULTIPLIER,
_make_plugin_stale_callback(uname, ctx))
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":
scripts/c/hbc_mini.c
+11 -4
View File
@@ -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");
}
scripts/hbc_mini.py
+2 -2
View File
@@ -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