diff --git a/Dockerfile b/Dockerfile index c09ea5e..ef2c990 100644 --- a/Dockerfile +++ b/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 diff --git a/build.sh b/build.sh index 7519654..0e00e2a 100755 --- a/build.sh +++ b/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}" ;; - web) TARGETS="web"; MSG="${2:-Build and push images}" ;; - all|"") TARGETS="poller web"; MSG="${2:-Build and push images}" ;; - *) TARGETS="poller web"; MSG="${1}" ;; + poller) TARGETS="poller"; MSG="${2:-Build and push images}" ;; + host-poller) TARGETS="host-poller"; MSG="${2:-Build and push images}" ;; + web) TARGETS="web"; MSG="${2:-Build and push images}" ;; + all|"") TARGETS="poller host-poller web"; MSG="${2:-Build and push images}" ;; + *) TARGETS="poller host-poller web"; MSG="${1}" ;; esac TARGET="${TARGETS// /+}" diff --git a/compose.infra.yml b/compose.infra.yml index a2eb852..0533a23 100644 --- a/compose.infra.yml +++ b/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 diff --git a/host_poller.py b/host_poller.py new file mode 100644 index 0000000..ff6a407 --- /dev/null +++ b/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 diff --git a/poller.py b/poller.py index 4ea1820..ce242fc 100644 --- a/poller.py +++ b/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). - 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}") + # 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()}) diff --git a/services/host_sensors.py b/services/host_sensors.py new file mode 100644 index 0000000..9bdc191 --- /dev/null +++ b/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 diff --git a/services/mqtt_publisher.py b/services/mqtt_publisher.py index c3c9b59..b5bf246 100644 --- a/services/mqtt_publisher.py +++ b/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 @@ -242,6 +243,61 @@ class MqttPublisher: 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 @@ -249,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 diff --git a/services/store.py b/services/store.py index ee349d9..6ed0e1a 100644 --- a/services/store.py +++ b/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: - return await cache_get(K_META) +async def get_host_sensors() -> dict | None: + 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: - """Atomically claim the poller lock (or re-own it). True if held.""" +async def acquire_lock(token: str, ttl: int = 30, key: str = K_LOCK) -> bool: + """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: