jbod-monitor/poller.py

"""Dedicated hardware poller — the sole owner of SAS/SMART/SES I/O.

Runs as its own (privileged, pid:host) container. Everything that touches
the bus happens here, serialized behind the hardware gate and paced, then
written to Redis. The API and MQTT services are pure Redis readers and
never issue a hardware command — so they can crash-loop freely without
ever storming the backplanes.

Design points that matter for fragile SAS expanders:
  - one global semaphore (POLL_CONCURRENCY=1) serializes all hardware I/O
  - a gap (POLL_DRIVE_GAP) between each drive keeps the bus mostly quiet
  - startup jitter avoids a thundering herd on (re)start
  - a single-instance Redis lock means two pollers can't both poll
  - last-good results stay in Redis; a failed sweep never blanks them
"""
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.enclosure import discover_enclosures, get_enclosure_status, list_slots
from services.leds import run_led
from services.smart import _run_smartctl, sg_ses_available, smartctl_available
from services.zfs import get_zfs_pool_map

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

SWEEP_INTERVAL = int(os.environ.get("POLL_SWEEP_INTERVAL", "300"))
STARTUP_JITTER = float(os.environ.get("POLL_STARTUP_JITTER", "10"))
LOCK_TTL = 30
# Pacing between hardware ops (POLL_DRIVE_GAP) is enforced centrally by the
# hardware gate, so every op — SMART, SES, host, MegaRAID, ledctl — is spaced.

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


async def sweep() -> None:
    """One full, paced pass over all hardware → Redis."""
    t0 = time.monotonic()
    errors: list[str] = []

    # 1. Discover enclosures + slots (sysfs, no bus I/O).
    enclosures = discover_enclosures()
    inv_enclosures = []
    drive_devices: list[str] = []
    for enc in enclosures:
        slots = list_slots(enc["id"])
        inv_enclosures.append({**enc, "slots": slots})
        for s in slots:
            if s["device"]:
                drive_devices.append(s["device"])

    # 2. ZFS topology (gated).
    pool_map = await store.get_zfs_map()
    try:
        pool_map = await get_zfs_pool_map()
        await store.set_zfs_map(pool_map)
    except Exception as e:
        errors.append(f"zfs:{e}")

    # 3. Per-drive SMART (gate serializes + paces each call).
    for dev in drive_devices:
        try:
            data = await _run_smartctl(dev)
            await store.set_smart(dev, data)
        except Exception as e:
            errors.append(f"smart:{dev}:{e}")

    # 4. Per-enclosure SES status (gate serializes + paces each call).
    for enc in enclosures:
        if not enc.get("sg_device"):
            continue
        try:
            ses = await get_enclosure_status(enc["sg_device"])
            if ses is not None:
                await store.set_ses(enc["id"], ses)
        except Exception as e:
            errors.append(f"ses:{enc['id']}:{e}")

    # NOTE: host/on-metal drives are owned by the separate host-poller now.

    # 6. Inventory + heartbeat.
    await store.set_inventory({"enclosures": inv_enclosures, "ts": store.now()})
    duration = round(time.monotonic() - t0, 1)
    await store.set_meta({
        "last_sweep_ts": store.now(),
        "last_sweep_duration": duration,
        "drive_count": len(drive_devices),
        "enclosure_count": len(enclosures),
        "errors": errors[:20],
        "sweep_interval": SWEEP_INTERVAL,
    })
    logger.info(
        "sweep done: %d drives, %d enclosures, %.1fs, %d error(s)",
        len(drive_devices), len(enclosures), duration, len(errors),
    )


async def led_consumer() -> None:
    """Drain locate-LED requests from Redis and run them (gated)."""
    while True:
        try:
            cmd = await store.pop_led(timeout=5)
            if cmd:
                try:
                    await run_led(cmd["device"], cmd["state"])
                    logger.info("LED %s -> %s", cmd.get("device"), cmd.get("state"))
                except Exception as e:
                    logger.warning("LED command failed %s: %s", cmd, e)
        except asyncio.CancelledError:
            raise
        except Exception as e:
            logger.warning("LED consumer error: %s", e)
            await asyncio.sleep(2)


async def lock_refresher() -> None:
    while True:
        await asyncio.sleep(LOCK_TTL / 3)
        # Any failure to confirm we still hold the lock is fatal — keeping the
        # lock alive is what prevents a second poller from sweeping concurrently.
        try:
            ok = await store.refresh_lock(TOKEN, LOCK_TTL)
        except Exception as e:
            logger.error("lock refresh error (%s) — exiting to avoid double-polling", e)
            os._exit(1)
        if not ok:
            logger.error("lost poller lock — exiting to avoid double-polling")
            os._exit(1)


def _critical_task_done(task: asyncio.Task) -> None:
    """Fail-safe: if a critical task ends for any reason other than a clean
    cancel (shutdown), exit so we never run unguarded."""
    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 — poller needs Redis; retrying in 5s")
        await asyncio.sleep(5)
        await init_cache()

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

    # Start the refresher IMMEDIATELY — before any pre-sweep sleep. The jitter
    # and restart-storm deferral below can together exceed LOCK_TTL, so without
    # an active refresher the lock would expire and a second poller could begin
    # sweeping concurrently (a hardware-storm risk).
    refresher = asyncio.create_task(lock_refresher())
    refresher.add_done_callback(_critical_task_done)

    if not smartctl_available():
        logger.warning("smartctl not found — install smartmontools")
    if not sg_ses_available():
        logger.warning("sg_ses not found — install sg3-utils")
    if os.geteuid() != 0:
        logger.warning("not running as root — smartctl/sg_ses may fail")

    # Stagger startup so a restart doesn't immediately storm the bus.
    jitter = random.uniform(0, STARTUP_JITTER)
    logger.info("startup jitter %.1fs", jitter)
    await asyncio.sleep(jitter)

    # Restart-storm guard: if a sweep ran recently (persisted in Redis across
    # restarts), wait out the remainder of the interval instead of immediately
    # re-sweeping. This is the key protection against deploy-cycling storms.
    meta = await store.get_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 sweep %.0fs ago — deferring first sweep %.0fs", since, wait)
            await asyncio.sleep(wait)

    leds = asyncio.create_task(led_consumer())
    try:
        while True:
            t0 = time.monotonic()
            try:
                await sweep()
            except Exception as e:
                logger.error("sweep error: %s", e)
            elapsed = time.monotonic() - t0
            await asyncio.sleep(max(5, SWEEP_INTERVAL - elapsed))
    finally:
        refresher.cancel()
        leds.cancel()
        await close_cache()


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