jbod-monitor/host_poller.py

"""Host-poller — server chassis temps (IPMI/iDRAC + CPU) and on-metal drives.

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.

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")

SWEEP_INTERVAL = int(os.environ.get("HOST_POLL_INTERVAL", "120"))
STARTUP_JITTER = float(os.environ.get("POLL_STARTUP_JITTER", "10"))
LOCK_TTL = 30

TOKEN = f"{socket.gethostname()}-hostpoll-{os.getpid()}"


async def sweep() -> None:
    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({
        "node": socket.gethostname(),
        "env": env,
        "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,
        "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),
    )


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")

    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)

    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))
    finally:
        refresher.cancel()
        await close_cache()


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