From ea045433e5373c2274eb74d9a80daaa1d12a207b Mon Sep 17 00:00:00 2001 From: adam Date: Tue, 16 Jun 2026 18:50:16 +0000 Subject: [PATCH] Split host-poller into separate env (IPMI/CPU) and drive (SMART) loops Two concurrent loops in one process sharing the lock + hardware gate (IPMI and SMART never run concurrently) but on separate schedules: env every HOST_ENV_INTERVAL (60s, responsive inlet/CPU) and drives every HOST_DRIVE_INTERVAL (300s, gentle SMART). Each loop has its own heartbeat (env_meta/drive_meta) + restart-storm guard. --- compose.infra.yml | 3 +- host_poller.py | 121 ++++++++++++++++++++++++++-------------------- services/store.py | 3 +- 3 files changed, 73 insertions(+), 54 deletions(-) diff --git a/compose.infra.yml b/compose.infra.yml index 70379c5..f9edefc 100644 --- a/compose.infra.yml +++ b/compose.infra.yml @@ -52,7 +52,8 @@ services: - REDIS_HOST=127.0.0.1 - REDIS_PORT=6379 - REDIS_DB=0 - - HOST_POLL_INTERVAL=60 + - HOST_ENV_INTERVAL=60 # IPMI/iDRAC + CPU — responsive (inlet temp 1/min) + - HOST_DRIVE_INTERVAL=300 # on-metal SMART — gentle, isolated from env - POLL_DRIVE_GAP=0.5 - POLL_CONCURRENCY=1 - POLL_STARTUP_JITTER=10 diff --git a/host_poller.py b/host_poller.py index ff6a407..f8e1657 100644 --- a/host_poller.py +++ b/host_poller.py @@ -1,12 +1,16 @@ -"""Host-poller — server chassis temps (IPMI/iDRAC + CPU) and on-metal drives. +"""Host-poller — server chassis temps, split into two independent loops. -Separate process/container from the SAS poller: this owns the chassis -subsystem (in-band IPMI, CPU coretemp, PERC/onboard drives), which is -independent of the external SAS shelves and shouldn't share their hardware -gate's bus. Writes to Redis; the web/MQTT consumer publishes these on the hub -device. Own single-instance lock + heartbeat; paced via the per-process gate. +Separate process/container from the SAS poller. Two concurrent loops in one +process, sharing the single-instance lock AND the per-process hardware gate +(so IPMI and SMART never hit hardware at the same instant), but on separate +schedules: -Host-drive ZFS annotation reuses jbod:zfs_map written by the SAS poller. + - env loop (fast): IPMI/iDRAC env + CPU coretemp -> jbod:host_sensors + - drive loop (slow): on-metal drive SMART -> jbod:host_drives + +Keeping SMART on a gentler cadence than the responsive inlet/CPU reads. Each +loop has its own heartbeat + restart-storm guard. Host-drive ZFS annotation +reuses jbod:zfs_map written by the SAS poller. """ import asyncio import logging @@ -26,25 +30,24 @@ logging.basicConfig( ) logger = logging.getLogger("host-poller") -SWEEP_INTERVAL = int(os.environ.get("HOST_POLL_INTERVAL", "120")) +ENV_INTERVAL = int(os.environ.get("HOST_ENV_INTERVAL", "60")) +DRIVE_INTERVAL = int(os.environ.get("HOST_DRIVE_INTERVAL", "300")) STARTUP_JITTER = float(os.environ.get("POLL_STARTUP_JITTER", "10")) LOCK_TTL = 30 TOKEN = f"{socket.gethostname()}-hostpoll-{os.getpid()}" -async def sweep() -> None: +async def sweep_env() -> None: + """IPMI/iDRAC environmental + CPU temps (the responsive loop).""" t0 = time.monotonic() errors: list[str] = [] - - # 1. IPMI (iDRAC) environmental + CPU temps. try: ipmi = await read_ipmi_temps() except Exception as e: errors.append(f"ipmi:{e}") ipmi = [] env = [s for s in ipmi if s.get("kind") == "env"] - # Prefer kernel coretemp for CPU package temps; fall back to IPMI CPU temps. cpu = read_coretemp() or [s for s in ipmi if s.get("kind") == "cpu"] await store.set_host_sensors({ @@ -53,30 +56,57 @@ async def sweep() -> None: "cpu": cpu, "ts": store.now(), }) - - # 2. On-metal (non-enclosure) drives — annotated from the SAS poller's zfs map. - try: - pool_map = await store.get_zfs_map() - host_drives = await get_host_drives(pool_map) - await store.set_host_drives(host_drives) - except Exception as e: - errors.append(f"hostdrives:{e}") - host_drives = [] - - duration = round(time.monotonic() - t0, 1) await store.set_meta({ "last_sweep_ts": store.now(), - "last_sweep_duration": duration, + "last_sweep_duration": round(time.monotonic() - t0, 1), "env_count": len(env), "cpu_count": len(cpu), - "host_drive_count": len(host_drives), "errors": errors[:20], - "sweep_interval": SWEEP_INTERVAL, - }, key=store.K_HOST_META) - logger.info( - "host sweep done: %d env, %d cpu, %d drives, %.1fs, %d error(s)", - len(env), len(cpu), len(host_drives), duration, len(errors), - ) + "interval": ENV_INTERVAL, + }, key=store.K_HOST_ENV_META) + logger.info("env sweep: %d env, %d cpu, %d error(s)", len(env), len(cpu), len(errors)) + + +async def sweep_drives() -> None: + """On-metal (non-enclosure) drive SMART (the gentle loop).""" + t0 = time.monotonic() + errors: list[str] = [] + drives: list = [] + try: + pool_map = await store.get_zfs_map() + drives = await get_host_drives(pool_map) + await store.set_host_drives(drives) + except Exception as e: + errors.append(f"hostdrives:{e}") + await store.set_meta({ + "last_sweep_ts": store.now(), + "last_sweep_duration": round(time.monotonic() - t0, 1), + "host_drive_count": len(drives), + "errors": errors[:20], + "interval": DRIVE_INTERVAL, + }, key=store.K_HOST_DRIVE_META) + logger.info("drive sweep: %d drives, %.1fs, %d error(s)", + len(drives), round(time.monotonic() - t0, 1), len(errors)) + + +async def run_loop(name: str, fn, interval: int, meta_key: str) -> None: + """Run `fn` every `interval`s, with a restart-storm guard from its meta.""" + meta = await store.get_meta(key=meta_key) + if meta and meta.get("last_sweep_ts"): + since = store.now() - meta["last_sweep_ts"] + if 0 <= since < interval: + wait = interval - since + logger.info("%s: last sweep %.0fs ago — deferring first %.0fs", name, since, wait) + await asyncio.sleep(wait) + while True: + t0 = time.monotonic() + try: + await fn() + except asyncio.CancelledError: + raise + except Exception as e: + logger.error("%s loop error: %s", name, e) + await asyncio.sleep(max(5, interval - (time.monotonic() - t0))) async def lock_refresher() -> None: @@ -117,30 +147,17 @@ async def main() -> None: if os.geteuid() != 0: logger.warning("not running as root — ipmitool/smartctl may fail") - jitter = random.uniform(0, STARTUP_JITTER) - logger.info("startup jitter %.1fs", jitter) - await asyncio.sleep(jitter) - - # Restart-storm guard (mirrors the SAS poller). - meta = await store.get_meta(key=store.K_HOST_META) - if meta and meta.get("last_sweep_ts"): - since = store.now() - meta["last_sweep_ts"] - if 0 <= since < SWEEP_INTERVAL: - wait = SWEEP_INTERVAL - since - logger.info("last host sweep %.0fs ago — deferring first sweep %.0fs", since, wait) - await asyncio.sleep(wait) + await asyncio.sleep(random.uniform(0, STARTUP_JITTER)) + env_task = asyncio.create_task( + run_loop("env", sweep_env, ENV_INTERVAL, store.K_HOST_ENV_META)) + drive_task = asyncio.create_task( + run_loop("drive", sweep_drives, DRIVE_INTERVAL, store.K_HOST_DRIVE_META)) try: - while True: - t0 = time.monotonic() - try: - await sweep() - except Exception as e: - logger.error("host sweep error: %s", e) - elapsed = time.monotonic() - t0 - await asyncio.sleep(max(5, SWEEP_INTERVAL - elapsed)) + await asyncio.gather(env_task, drive_task) finally: - refresher.cancel() + for t in (refresher, env_task, drive_task): + t.cancel() await close_cache() diff --git a/services/store.py b/services/store.py index 6ed0e1a..f3ec0a9 100644 --- a/services/store.py +++ b/services/store.py @@ -25,7 +25,8 @@ K_INVENTORY = "jbod:inventory" K_HOST_DRIVES = "jbod:host_drives" K_HOST_SENSORS = "jbod:host_sensors" K_META = "jbod:poll:meta" -K_HOST_META = "jbod:host_poll:meta" +K_HOST_ENV_META = "jbod:host_poll:env_meta" +K_HOST_DRIVE_META = "jbod:host_poll:drive_meta" K_LED_QUEUE = "jbod:led:queue" K_LOCK = "jbod:poll:lock" K_HOST_LOCK = "jbod:host_poll:lock"