Add host-poller: chassis IPMI/iDRAC + CPU + on-metal drive temps

New third producer container (host-poller) owning the server chassis
subsystem, separate from the SAS-shelf poller:
- services/host_sensors.py: in-band IPMI (ipmitool sdr) for Inlet/Exhaust/etc.
  + CPU package temps from coretemp hwmon
- host_poller.py: own single-instance lock + heartbeat (host_poll keys),
  restart-storm guard, paced; writes jbod:host_sensors + jbod:host_drives
- move host/on-metal drive collection off the SAS poller to host-poller
  (reads jbod:zfs_map for annotation)
- store: generalized lock/meta with a key param; host_sensors + host-poll keys
- mqtt_publisher: publish env/CPU/host-drive temps as entities on the hub
  device (JBOD Monitor)
- Dockerfile host-poller target (ipmitool+smartmontools); compose.infra adds
  host-poller; build.sh gains host-poller/all

Dockerfile

@@ -30,6 +30,27 @@ COPY services/ services/
 
 CMD ["python", "-u", "poller.py"]
 
+# ---- Target: host-poller (chassis IPMI/iDRAC + CPU + on-metal drives) ----
+# Privileged at runtime (needs /dev/ipmi0 + raw disks). Carries ipmitool +
+# smartmontools; no SAS-shelf tooling.
+FROM python:3.13-slim AS host-poller
+
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    ipmitool \
+    smartmontools \
+    util-linux \
+    && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /app
+
+COPY requirements-poller.txt .
+RUN pip install --no-cache-dir -r requirements-poller.txt
+
+COPY host_poller.py .
+COPY services/ services/
+
+CMD ["python", "-u", "host_poller.py"]
+
 # ---- Target: web (read-only consumer) ----
 # Unprivileged at runtime; no hardware tools. Serves REST/UI + MQTT, reads Redis.
 FROM python:3.13-slim AS web

build.sh

@@ -16,10 +16,11 @@ cd "$(dirname "$0")"
 # First arg is the target if it's a known one; otherwise it's treated as the
 # commit message (backward-compatible with `./build.sh "message"`).
 case "${1:-}" in
-    poller) TARGETS="poller"; MSG="${2:-Build and push images}" ;;
+    poller)      TARGETS="poller";      MSG="${2:-Build and push images}" ;;
-    web)    TARGETS="web";    MSG="${2:-Build and push images}" ;;
+    host-poller) TARGETS="host-poller"; MSG="${2:-Build and push images}" ;;
-    all|"") TARGETS="poller web"; MSG="${2:-Build and push images}" ;;
+    web)         TARGETS="web";         MSG="${2:-Build and push images}" ;;
-    *)      TARGETS="poller web"; MSG="${1}" ;;
+    all|"")      TARGETS="poller host-poller web"; MSG="${2:-Build and push images}" ;;
+    *)           TARGETS="poller host-poller web"; MSG="${1}" ;;
 esac
 TARGET="${TARGETS// /+}"
 

compose.infra.yml

@@ -34,6 +34,32 @@ services:
     depends_on:
       - redis
 
+  # Chassis producer: IPMI/iDRAC + CPU coretemp + on-metal drives → Redis.
+  # Separate from the SAS poller (different subsystem); writes host_sensors +
+  # host_drives. The web consumer publishes these on the hub device.
+  host-poller:
+    image: docker.adamksmith.xyz/jbod-host-poller:latest   # pin to a SHA in deploy
+    container_name: jbod-host-poller
+    restart: unless-stopped
+    privileged: true
+    network_mode: host
+    volumes:
+      - /dev:/dev
+      - /sys:/sys:ro
+      - /run/udev:/run/udev:ro
+    environment:
+      - TZ=America/Denver
+      - REDIS_HOST=127.0.0.1
+      - REDIS_PORT=6379
+      - REDIS_DB=0
+      - HOST_POLL_INTERVAL=120
+      - POLL_DRIVE_GAP=0.5
+      - POLL_CONCURRENCY=1
+      - POLL_STARTUP_JITTER=10
+      - POLL_DATA_TTL=900
+    depends_on:
+      - redis
+
   redis:
     image: redis:7-alpine
     container_name: jbod-redis

host_poller.py

@@ -0,0 +1,151 @@
+"""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

poller.py

@@ -23,7 +23,6 @@ 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.host import get_host_drives
 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
@@ -86,12 +85,7 @@ async def sweep() -> None:
         except Exception as e:
             errors.append(f"ses:{enc['id']}:{e}")
 
-    # 5. Host (non-enclosure) drives + MegaRAID (gated, reuses pool_map).
+    # NOTE: host/on-metal drives are owned by the separate host-poller now.
-    try:
-        host_drives = await get_host_drives(pool_map)
-        await store.set_host_drives(host_drives)
-    except Exception as e:
-        errors.append(f"host:{e}")
 
     # 6. Inventory + heartbeat.
     await store.set_inventory({"enclosures": inv_enclosures, "ts": store.now()})

services/host_sensors.py

@@ -0,0 +1,95 @@
+"""Host/chassis temperature collectors: IPMI (iDRAC) + CPU coretemp.
+
+Used by the host-poller (separate process from the SAS poller). In-band IPMI
+via ipmitool (/dev/ipmi0) gives environmental sensors (Inlet/Exhaust/…); the
+kernel coretemp hwmon gives CPU package temps. IPMI is gated for serialization;
+coretemp is a cheap sysfs read.
+"""
+import asyncio
+import logging
+import re
+from pathlib import Path
+
+from services.hwgate import gate
+
+logger = logging.getLogger(__name__)
+
+HWMON = Path("/sys/class/hwmon")
+
+
+async def read_ipmi_temps() -> list[dict]:
+    """`ipmitool sdr type temperature` → [{name, temperature_c, status, kind}]."""
+    try:
+        async with gate():
+            proc = await asyncio.create_subprocess_exec(
+                "ipmitool", "sdr", "type", "temperature",
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+            )
+            out, err = await proc.communicate()
+    except FileNotFoundError:
+        logger.warning("ipmitool not found")
+        return []
+    if proc.returncode != 0:
+        logger.warning("ipmitool failed: %s", err.decode(errors="replace").strip())
+        return []
+    return _parse_ipmi(out.decode(errors="replace"))
+
+
+def _parse_ipmi(text: str) -> list[dict]:
+    # e.g.  "Inlet Temp       | 04h | ok  |  7.1 | 23 degrees C"
+    #       "Temp             | 0Eh | ok  |  3.1 | 40 degrees C"  (CPU1)
+    sensors: list[dict] = []
+    for line in text.splitlines():
+        parts = [p.strip() for p in line.split("|")]
+        if len(parts) < 5:
+            continue
+        name, _sid, status, entity, reading = parts[:5]
+        m = re.search(r"(-?\d+(?:\.\d+)?)\s*degrees?\s*C", reading, re.IGNORECASE)
+        if not m:  # "no reading", "disabled", "ns", etc.
+            continue
+        temp = float(m.group(1))
+        kind = "env"
+        # Dell reports CPU temps as "Temp" under entity 3.x — relabel them.
+        if name.lower() == "temp" and entity.startswith("3."):
+            name = f"CPU {entity.split('.')[-1]}"
+            kind = "cpu"
+        sensors.append({
+            "name": name,
+            "temperature_c": temp,
+            "status": status or "ok",
+            "kind": kind,
+        })
+    return sensors
+
+
+def read_coretemp() -> list[dict]:
+    """CPU package temps from coretemp hwmon → [{name, temperature_c, kind}]."""
+    cpus: list[dict] = []
+    if not HWMON.is_dir():
+        return cpus
+    for hw in sorted(HWMON.iterdir()):
+        try:
+            if (hw / "name").read_text().strip() != "coretemp":
+                continue
+        except OSError:
+            continue
+        for label_f in sorted(hw.glob("temp*_label")):
+            try:
+                label = label_f.read_text().strip()
+            except OSError:
+                continue
+            if not label.lower().startswith("package id"):
+                continue
+            input_f = label_f.with_name(label_f.name.replace("_label", "_input"))
+            try:
+                milli = int(input_f.read_text().strip())
+            except (OSError, ValueError):
+                continue
+            pkg = label.split("id")[-1].strip()
+            cpus.append({
+                "name": f"CPU {pkg}",
+                "temperature_c": round(milli / 1000),
+                "kind": "cpu",
+            })
+    return cpus

services/mqtt_publisher.py

@@ -19,6 +19,7 @@ import logging
 import os
 import re
 
+from services import store
 from services.secrets import read_kv2
 from services.temps import build_temp_snapshot, get_node_id
 
@@ -80,7 +81,10 @@ class MqttPublisher:
         self.discovery_prefix = os.environ.get("MQTT_DISCOVERY_PREFIX", "homeassistant")
         self.base_topic = os.environ.get("MQTT_BASE_TOPIC", "jbod-monitor")
         self.interval = int(os.environ.get("MQTT_PUBLISH_INTERVAL", "60"))
-        self.node = _slug(get_node_id())
+        self.node_raw = get_node_id()
+        self.node = _slug(self.node_raw)
+        # Parent "hub" device that the per-enclosure devices nest under.
+        self.hub_id = f"{self.node}_jbod_monitor"
 
         self.availability_topic = f"{self.base_topic}/{self.node}/status"
         # unique_ids for which we've already published discovery this connection.
@@ -127,6 +131,16 @@ class MqttPublisher:
         logger.warning("MQTT disconnected: %s", reason_code)
 
     # discovery / state ------------------------------------------------------
+    def _hub_device_block(self) -> dict:
+        """The parent device the enclosures nest under (named, so HA doesn't
+        show an 'Unnamed Device')."""
+        return {
+            "identifiers": [self.hub_id],
+            "name": f"JBOD Monitor ({self.node_raw})",
+            "manufacturer": "jbod-monitor",
+            "model": "SES/SMART poller",
+        }
+
     def _device_block(self, enc: dict) -> dict:
         enc_id = enc["id"]
         vendor = enc.get("vendor", "").strip()
@@ -137,9 +151,29 @@ class MqttPublisher:
             "name": f"JBOD {name} ({enc_id})",
             "manufacturer": vendor or None,
             "model": model or None,
-            "via_device": f"{self.node}_jbod_monitor",
+            "via_device": self.hub_id,
         }
 
+    def _publish_hub(self) -> None:
+        """Announce the hub device via a connectivity binary_sensor, so the
+        parent device is named and shows monitor online/offline status."""
+        uid = f"{self.hub_id}_status"
+        if uid in self._discovered:
+            return
+        payload = {
+            "name": "Status",
+            "unique_id": uid,
+            "device_class": "connectivity",
+            "state_topic": self.availability_topic,
+            "payload_on": "online",
+            "payload_off": "offline",
+            "entity_category": "diagnostic",
+            "device": self._hub_device_block(),
+        }
+        topic = f"{self.discovery_prefix}/binary_sensor/{self.hub_id}/status/config"
+        self._client.publish(topic, json.dumps(payload), qos=1, retain=True)
+        self._discovered.add(uid)
+
     def _publish_discovery(self, enc: dict) -> None:
         enc_id = enc["id"]
         enc_slug = _slug(enc_id)
@@ -204,10 +238,66 @@ class MqttPublisher:
         self._client.publish(state_topic, json.dumps(payload), qos=0, retain=True)
 
     def publish_snapshot(self, snapshot: dict) -> None:
+        self._publish_hub()  # name the parent device before nesting enclosures
         for enc in snapshot.get("enclosures", []):
             self._publish_discovery(enc)
             self._publish_state(enc)
 
+    def publish_host(self, sensors: dict, drives: list) -> None:
+        """Publish chassis temps (IPMI env, CPU, on-metal drives) on the hub."""
+        self._publish_hub()
+        state_topic = f"{self.base_topic}/{self.node}/host/state"
+        device = self._hub_device_block()
+        payload: dict = {}
+
+        def announce(object_id: str, cfg: dict) -> None:
+            uid = cfg["unique_id"]
+            if uid not in self._discovered:
+                topic = f"{self.discovery_prefix}/sensor/{self.hub_id}/{object_id}/config"
+                self._client.publish(topic, json.dumps(cfg), qos=1, retain=True)
+                self._discovered.add(uid)
+
+        common = {
+            "device_class": "temperature",
+            "unit_of_measurement": "°C",
+            "state_class": "measurement",
+            "state_topic": state_topic,
+            "availability_topic": self.availability_topic,
+            "device": device,
+        }
+
+        def add(kind: str, name: str, temp, icon: str | None = None) -> None:
+            if temp is None or not name:
+                return
+            key = f"{kind}_{_slug(name)}"
+            payload[key] = temp
+            cfg = {
+                **common,
+                "name": name,
+                "unique_id": f"{self.hub_id}_{key}",
+                "value_template": f"{{{{ value_json.{key} }}}}",
+            }
+            if icon:
+                cfg["icon"] = icon
+            announce(key, cfg)
+
+        for s in sensors.get("env", []):
+            add("env", s.get("name"), s.get("temperature_c"))
+        for c in sensors.get("cpu", []):
+            add("cpu", c.get("name"), c.get("temperature_c"), icon="mdi:cpu-64-bit")
+
+        def add_drive(d: dict) -> None:
+            dev = d.get("device")
+            if dev:
+                add("drive", f"Drive {dev}", d.get("temperature_c"), icon="mdi:harddisk")
+
+        for d in drives:
+            add_drive(d)
+            for pd in d.get("physical_drives", []):
+                add_drive(pd)
+
+        self._client.publish(state_topic, json.dumps(payload), qos=0, retain=True)
+
     async def run(self) -> None:
         """Background loop: build a snapshot and publish on an interval."""
         await asyncio.sleep(3)  # let the first SMART poll populate the cache
@@ -215,9 +305,14 @@ class MqttPublisher:
             try:
                 snapshot = await build_temp_snapshot()
                 self.publish_snapshot(snapshot)
+                host_sensors = await store.get_host_sensors() or {}
+                host_drives = await store.get_host_drives() or []
+                self.publish_host(host_sensors, host_drives)
                 logger.info(
-                    "MQTT published temps for %d enclosure(s)",
+                    "MQTT published temps for %d enclosure(s) + host (%d env, %d cpu)",
                     len(snapshot.get("enclosures", [])),
+                    len(host_sensors.get("env", [])),
+                    len(host_sensors.get("cpu", [])),
                 )
             except asyncio.CancelledError:
                 raise

services/store.py

@@ -23,9 +23,12 @@ K_SES = "jbod:ses:{}"
 K_ZFS = "jbod:zfs_map"
 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_LED_QUEUE = "jbod:led:queue"
 K_LOCK = "jbod:poll:lock"
+K_HOST_LOCK = "jbod:host_poll:lock"
 
 
 # ── writes (poller) ─────────────────────────────────────────────────────
@@ -49,9 +52,13 @@ async def set_host_drives(host_drives: list) -> None:
     await cache_set(K_HOST_DRIVES, host_drives, DATA_TTL)
 
 
-async def set_meta(meta: dict) -> None:
+async def set_host_sensors(sensors: dict) -> None:
+    await cache_set(K_HOST_SENSORS, sensors, DATA_TTL)
+
+
+async def set_meta(meta: dict, key: str = K_META) -> None:
     # Meta has no TTL — it's the heartbeat; staleness is judged from its ts.
-    await cache_set(K_META, meta, 0)
+    await cache_set(key, meta, 0)
 
 
 # ── reads (consumers) ───────────────────────────────────────────────────
@@ -75,8 +82,12 @@ async def get_host_drives() -> list:
     return await cache_get(K_HOST_DRIVES) or []
 
 
-async def get_meta() -> dict | None:
+async def get_host_sensors() -> dict | None:
-    return await cache_get(K_META)
+    return await cache_get(K_HOST_SENSORS)
+
+
+async def get_meta(key: str = K_META) -> dict | None:
+    return await cache_get(key)
 
 
 # ── locate-LED command queue ────────────────────────────────────────────
@@ -126,20 +137,20 @@ _REFRESH_LUA = (
 )
 
 
-async def acquire_lock(token: str, ttl: int = 30) -> bool:
+async def acquire_lock(token: str, ttl: int = 30, key: str = K_LOCK) -> bool:
-    """Atomically claim the poller lock (or re-own it). True if held."""
+    """Atomically claim a single-instance lock (or re-own it). True if held."""
     client = get_client()
     if client is None:
         return True  # no Redis → no coordination possible; run anyway
-    return bool(await client.eval(_ACQUIRE_LUA, 1, K_LOCK, token, ttl))
+    return bool(await client.eval(_ACQUIRE_LUA, 1, key, token, ttl))
 
 
-async def refresh_lock(token: str, ttl: int = 30) -> bool:
+async def refresh_lock(token: str, ttl: int = 30, key: str = K_LOCK) -> bool:
     """Atomically extend the lock iff we still own it."""
     client = get_client()
     if client is None:
         return True
-    return bool(await client.eval(_REFRESH_LUA, 1, K_LOCK, token, ttl))
+    return bool(await client.eval(_REFRESH_LUA, 1, key, token, ttl))
 
 
 def now() -> float: