jbod-monitor/host_poller.py

"""Host-poller — server chassis temps, split into two independent loops.

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:

  - 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
import os
import random
import socket
import time

from services import store
from services.cache import close_cache, init_cache, redis_available
from services.host import get_host_drives
from services.host_sensors import read_coretemp, read_ipmi_temps

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = logging.getLogger("host-poller")

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_env() -> None:
    """IPMI/iDRAC environmental + CPU temps (the responsive loop)."""
    t0 = time.monotonic()
    errors: list[str] = []
    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"]
    cpu = read_coretemp() or [s for s in ipmi if s.get("kind") == "cpu"]

    await store.set_host_sensors({
        "node": socket.gethostname(),
        "env": env,
        "cpu": cpu,
        "ts": store.now(),
    })
    await store.set_meta({
        "last_sweep_ts": store.now(),
        "last_sweep_duration": round(time.monotonic() - t0, 1),
        "env_count": len(env),
        "cpu_count": len(cpu),
        "errors": errors[:20],
        "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:
    while True:
        await asyncio.sleep(LOCK_TTL / 3)
        try:
            ok = await store.refresh_lock(TOKEN, LOCK_TTL, key=store.K_HOST_LOCK)
        except Exception as e:
            logger.error("lock refresh error (%s) — exiting", e)
            os._exit(1)
        if not ok:
            logger.error("lost host-poller lock — exiting")
            os._exit(1)


def _critical_task_done(task: asyncio.Task) -> None:
    if task.cancelled():
        return
    logger.error("critical task ended unexpectedly — exiting")
    os._exit(1)


async def main() -> None:
    await init_cache()
    while not redis_available():
        logger.warning("Redis unavailable — host-poller needs Redis; retrying in 5s")
        await asyncio.sleep(5)
        await init_cache()

    while not await store.acquire_lock(TOKEN, LOCK_TTL, key=store.K_HOST_LOCK):
        logger.warning("another host-poller holds the lock; waiting %ds", LOCK_TTL // 2)
        await asyncio.sleep(LOCK_TTL // 2)
    logger.info("host-poller lock acquired (%s)", TOKEN)

    refresher = asyncio.create_task(lock_refresher())
    refresher.add_done_callback(_critical_task_done)

    if os.geteuid() != 0:
        logger.warning("not running as root — ipmitool/smartctl may fail")

    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:
        await asyncio.gather(env_task, drive_task)
    finally:
        for t in (refresher, env_task, drive_task):
            t.cancel()
        await close_cache()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass