Compare commits
9 Commits
226f73851c
...
feat/dedic
| Author | SHA1 | Date | |
|---|---|---|---|
| d7f2ae69ca | |||
| 0f0bdf4f6c | |||
| 5032c58f7d | |||
| 4c68bff964 | |||
| 514502e045 | |||
| ea045433e5 | |||
| 822829aae4 | |||
| 1839e9d5f2 | |||
| 44a7824425 |
21
Dockerfile
21
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
|
||||
|
||||
9
build.sh
9
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// /+}"
|
||||
|
||||
|
||||
@@ -34,6 +34,33 @@ 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_ENV_INTERVAL=60 # IPMI/iDRAC + CPU — responsive (inlet temp 1/min)
|
||||
- HOST_DRIVE_INTERVAL=300 # on-metal SMART — gentle, isolated from env
|
||||
- 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
|
||||
|
||||
168
host_poller.py
Normal file
168
host_poller.py
Normal file
@@ -0,0 +1,168 @@
|
||||
"""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_hwmon_env, 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"] + read_hwmon_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
|
||||
66
k8s/README.md
Normal file
66
k8s/README.md
Normal file
@@ -0,0 +1,66 @@
|
||||
# jbod-monitor on RKE2 (DaemonSet)
|
||||
|
||||
Per-node self-contained deployment for Kubernetes worker nodes that can't run
|
||||
the docker-compose stack. Each opted-in node runs one pod (`host-poller` +
|
||||
`redis` + `web`) and publishes its own Home Assistant device
|
||||
(`JBOD Monitor (<node>)`) over MQTT.
|
||||
|
||||
This mirrors the docker-compose stack with **no code changes** — each node is
|
||||
independent, exactly like the standalone hosts.
|
||||
|
||||
## What it collects
|
||||
|
||||
- **host-poller** (privileged, hostPath `/dev`+`/sys`): IPMI/iDRAC env, CPU
|
||||
coretemp, on-metal drive SMART → node-local redis.
|
||||
- **web**: reads redis, publishes per-node MQTT discovery to the broker.
|
||||
- No SAS poller (no SES enclosures on these nodes). Consequence: host-drive
|
||||
entities have no ZFS pool label (no zfs_map producer).
|
||||
|
||||
## Prereqs — verified 2026-06-16
|
||||
|
||||
1. **Registry pull**: registry requires auth. `registry-cred-vaultstaticsecret.yaml`
|
||||
syncs `secret/registry/nexus` → `docker-registry-cred` (dockerconfigjson),
|
||||
referenced as `imagePullSecrets` in the DaemonSet. (Mirrors the cluster's
|
||||
existing `*-registry-cred` pattern.)
|
||||
2. **VSO**: v1.3.0; `openbao-kubernetes` VaultAuth healthy; `transformation.templates`
|
||||
supported. No existing VSS in the cluster uses transformation, so watch the
|
||||
`jbod-mqtt` VSS status on first apply for template errors.
|
||||
3. **MQTT egress**: confirmed — a cluster pod reached `10.5.30.3:1883`.
|
||||
4. **Nodes**: pascal/currie/fermi are untainted → no tolerations needed.
|
||||
|
||||
## Apply order
|
||||
|
||||
```bash
|
||||
# 1. Label ONLY the intended nodes (do not blanket-label all workers):
|
||||
kubectl label node usco-dc-pascal usco-dc-currie usco-dc-fermi jbod-monitor=enabled
|
||||
# (canary: label just usco-dc-fermi first)
|
||||
|
||||
# 2. Namespace + DaemonSet:
|
||||
kubectl apply -f k8s/daemonset.yaml
|
||||
|
||||
# 3. Secrets (VSO → docker-registry-cred + jbod-mqtt). Apply, then confirm sync:
|
||||
kubectl apply -f k8s/registry-cred-vaultstaticsecret.yaml
|
||||
kubectl apply -f k8s/mqtt-vaultstaticsecret.yaml
|
||||
kubectl -n jbod-monitor get vaultstaticsecret # SYNCED=True for both
|
||||
kubectl -n jbod-monitor get secret jbod-mqtt -o jsonpath='{.data.MQTT_USERNAME}' | base64 -d
|
||||
|
||||
# 4. Roll the DaemonSet so pods pick up the secrets if they raced ahead:
|
||||
kubectl -n jbod-monitor rollout restart daemonset/jbod-monitor
|
||||
```
|
||||
|
||||
## Verify
|
||||
|
||||
```bash
|
||||
kubectl -n jbod-monitor get pods -o wide # one per labeled node
|
||||
kubectl -n jbod-monitor logs <pod> -c host-poller | grep sweep
|
||||
kubectl -n jbod-monitor logs <pod> -c web | grep -i mqtt # "MQTT connected"
|
||||
```
|
||||
|
||||
Then check Home Assistant for `JBOD Monitor (usco-dc-pascal|currie|fermi)`.
|
||||
|
||||
## Notes
|
||||
|
||||
- pascal (R620) already has an iDRAC "System Board Inlet Temp" in HA via another
|
||||
integration — its env temp will partially duplicate; drives + CPU are net-new.
|
||||
- Pin image tags by SHA (already done: host-poller `514502e`, web `1839e9d`).
|
||||
- Removal: `kubectl delete -f k8s/daemonset.yaml` and unlabel the nodes.
|
||||
103
k8s/daemonset.yaml
Normal file
103
k8s/daemonset.yaml
Normal file
@@ -0,0 +1,103 @@
|
||||
# jbod-monitor on RKE2 nodes — per-node self-contained DaemonSet (Model A).
|
||||
#
|
||||
# Each scheduled node runs one pod = host-poller + redis + web, sharing the
|
||||
# pod's localhost (no hostNetwork needed). Only host-poller is privileged and
|
||||
# mounts /dev + /sys to read IPMI / SMART / coretemp. web egresses to the MQTT
|
||||
# broker and publishes a per-node HA device (MQTT_NODE_ID = spec.nodeName).
|
||||
#
|
||||
# No SAS poller (these nodes have no SES enclosures). NOTE: without it there is
|
||||
# no zfs_map, so host-drive entities won't carry a ZFS pool label.
|
||||
#
|
||||
# Target nodes are chosen by label — see k8s/README.md (label only
|
||||
# pascal/currie/fermi). Apply order: namespace -> VaultStaticSecret -> this.
|
||||
---
|
||||
apiVersion: v1
|
||||
kind: Namespace
|
||||
metadata:
|
||||
name: jbod-monitor
|
||||
---
|
||||
apiVersion: apps/v1
|
||||
kind: DaemonSet
|
||||
metadata:
|
||||
name: jbod-monitor
|
||||
namespace: jbod-monitor
|
||||
labels:
|
||||
app: jbod-monitor
|
||||
spec:
|
||||
selector:
|
||||
matchLabels:
|
||||
app: jbod-monitor
|
||||
template:
|
||||
metadata:
|
||||
labels:
|
||||
app: jbod-monitor
|
||||
spec:
|
||||
# Only nodes explicitly opted in (see README: kubectl label node ...).
|
||||
nodeSelector:
|
||||
jbod-monitor: "enabled"
|
||||
imagePullSecrets:
|
||||
- name: docker-registry-cred # synced by registry-cred-vaultstaticsecret.yaml
|
||||
containers:
|
||||
- name: redis
|
||||
image: redis:7-alpine
|
||||
args: ["redis-server", "--save", "60", "1", "--loglevel", "warning"]
|
||||
volumeMounts:
|
||||
- name: redis-data
|
||||
mountPath: /data
|
||||
resources:
|
||||
requests: {cpu: 25m, memory: 32Mi}
|
||||
limits: {memory: 128Mi}
|
||||
|
||||
- name: host-poller
|
||||
image: docker.adamksmith.xyz/jbod-host-poller:514502e
|
||||
securityContext:
|
||||
privileged: true # raw /dev access for ipmitool + smartctl
|
||||
env:
|
||||
- {name: TZ, value: "America/Denver"}
|
||||
- {name: REDIS_HOST, value: "127.0.0.1"}
|
||||
- {name: REDIS_PORT, value: "6379"}
|
||||
- {name: HOST_ENV_INTERVAL, value: "60"}
|
||||
- {name: HOST_DRIVE_INTERVAL, value: "300"}
|
||||
- {name: POLL_CONCURRENCY, value: "1"}
|
||||
- {name: POLL_DRIVE_GAP, value: "0.5"}
|
||||
- {name: POLL_STARTUP_JITTER, value: "10"}
|
||||
- {name: POLL_DATA_TTL, value: "900"}
|
||||
volumeMounts:
|
||||
- {name: dev, mountPath: /dev}
|
||||
- {name: sys, mountPath: /sys, readOnly: true}
|
||||
- {name: udev, mountPath: /run/udev, readOnly: true}
|
||||
resources:
|
||||
requests: {cpu: 25m, memory: 64Mi}
|
||||
limits: {memory: 256Mi}
|
||||
|
||||
- name: web
|
||||
image: docker.adamksmith.xyz/jbod-web:1839e9d
|
||||
env:
|
||||
- {name: TZ, value: "America/Denver"}
|
||||
- {name: REDIS_HOST, value: "127.0.0.1"}
|
||||
- {name: REDIS_PORT, value: "6379"}
|
||||
- {name: MQTT_HOST, value: "10.5.30.3"}
|
||||
- {name: MQTT_PORT, value: "1883"}
|
||||
- {name: MQTT_DISCOVERY_PREFIX, value: "homeassistant"}
|
||||
- {name: MQTT_BASE_TOPIC, value: "jbod-monitor"}
|
||||
- {name: MQTT_PUBLISH_INTERVAL, value: "60"}
|
||||
- name: MQTT_NODE_ID
|
||||
valueFrom:
|
||||
fieldRef:
|
||||
fieldPath: spec.nodeName
|
||||
envFrom:
|
||||
- secretRef:
|
||||
name: jbod-mqtt # MQTT_USERNAME / MQTT_PASSWORD (see VaultStaticSecret)
|
||||
resources:
|
||||
requests: {cpu: 25m, memory: 64Mi}
|
||||
limits: {memory: 256Mi}
|
||||
|
||||
volumes:
|
||||
- name: redis-data
|
||||
emptyDir: {}
|
||||
- name: dev
|
||||
hostPath: {path: /dev}
|
||||
- name: sys
|
||||
hostPath: {path: /sys}
|
||||
- name: udev
|
||||
hostPath: {path: /run/udev}
|
||||
35
k8s/mqtt-vaultstaticsecret.yaml
Normal file
35
k8s/mqtt-vaultstaticsecret.yaml
Normal file
@@ -0,0 +1,35 @@
|
||||
# MQTT broker creds for the web container, synced from OpenBao by VSO.
|
||||
#
|
||||
# Reads secret/home_assistant (keys mqtt_user / mqtt_pass) and renders a k8s
|
||||
# Secret `jbod-mqtt` with MQTT_USERNAME / MQTT_PASSWORD keys (consumed via
|
||||
# envFrom in the DaemonSet). VSO's `vso-read` policy can read secret/data/*,
|
||||
# so unlike the claude-read token it CAN read home_assistant — no creds ever
|
||||
# touch these manifests.
|
||||
#
|
||||
# Verify the transformation syntax against the installed VSO version.
|
||||
apiVersion: secrets.hashicorp.com/v1beta1
|
||||
kind: VaultStaticSecret
|
||||
metadata:
|
||||
name: jbod-mqtt
|
||||
namespace: jbod-monitor
|
||||
spec:
|
||||
vaultAuthRef: vault-secrets-operator-system/openbao-kubernetes
|
||||
mount: secret
|
||||
type: kv-v2
|
||||
path: home_assistant
|
||||
refreshAfter: 1h
|
||||
destination:
|
||||
name: jbod-mqtt
|
||||
create: true
|
||||
transformation:
|
||||
excludeRaw: true
|
||||
# Drop ALL passthrough source keys (api_key, vantage_*, etc.) — keep only
|
||||
# the templated MQTT_USERNAME/MQTT_PASSWORD below. Without this, envFrom
|
||||
# would inject the entire home_assistant secret into the web container.
|
||||
excludes:
|
||||
- ".*"
|
||||
templates:
|
||||
MQTT_USERNAME:
|
||||
text: '{{ .Secrets.mqtt_user }}'
|
||||
MQTT_PASSWORD:
|
||||
text: '{{ .Secrets.mqtt_pass }}'
|
||||
18
k8s/registry-cred-vaultstaticsecret.yaml
Normal file
18
k8s/registry-cred-vaultstaticsecret.yaml
Normal file
@@ -0,0 +1,18 @@
|
||||
# Image pull secret for docker.adamksmith.xyz, synced by VSO from OpenBao.
|
||||
# Mirrors the cluster's existing *-registry-cred pattern (secret/registry/nexus
|
||||
# -> dockerconfigjson). Referenced as imagePullSecrets in the DaemonSet.
|
||||
apiVersion: secrets.hashicorp.com/v1beta1
|
||||
kind: VaultStaticSecret
|
||||
metadata:
|
||||
name: docker-registry-cred
|
||||
namespace: jbod-monitor
|
||||
spec:
|
||||
vaultAuthRef: vault-secrets-operator-system/openbao-kubernetes
|
||||
mount: secret
|
||||
path: registry/nexus
|
||||
type: kv-v2
|
||||
refreshAfter: 1h
|
||||
destination:
|
||||
name: docker-registry-cred
|
||||
create: true
|
||||
type: kubernetes.io/dockerconfigjson
|
||||
@@ -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()})
|
||||
|
||||
147
services/host_sensors.py
Normal file
147
services/host_sensors.py
Normal file
@@ -0,0 +1,147 @@
|
||||
"""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 os
|
||||
import re
|
||||
from collections import Counter
|
||||
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]
|
||||
# Skip relative/derived sensors (e.g. "P1 Therm Margin") — they read in
|
||||
# "degrees C" but are a distance-to-throttle, not an absolute temp.
|
||||
if "margin" in name.lower():
|
||||
continue
|
||||
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_hwmon_env() -> list[dict]:
|
||||
"""Read labelled temps from extra hwmon chips as env sensors.
|
||||
|
||||
For boxes without IPMI (e.g. Dell workstations), the BIOS SMM interface
|
||||
(`dell_smm`) exposes CPU/ambient/SODIMM/board temps. Chips are configurable
|
||||
via HOST_HWMON_CHIPS (comma list; default "dell_smm"); set to "" to disable.
|
||||
Names are "<chip> <label>", with the temp index appended when a label
|
||||
repeats within a chip (e.g. dell_smm's multiple "Other" sensors).
|
||||
"""
|
||||
chips = [c.strip() for c in os.environ.get("HOST_HWMON_CHIPS", "dell_smm").split(",") if c.strip()]
|
||||
if not chips or not HWMON.is_dir():
|
||||
return []
|
||||
out: list[dict] = []
|
||||
for hw in sorted(HWMON.iterdir()):
|
||||
try:
|
||||
chip = (hw / "name").read_text().strip()
|
||||
except OSError:
|
||||
continue
|
||||
if chip not in chips:
|
||||
continue
|
||||
entries: list[tuple[str, str, int]] = [] # (idx, label, temp)
|
||||
for label_f in sorted(hw.glob("temp*_label")):
|
||||
try:
|
||||
label = label_f.read_text().strip()
|
||||
except OSError:
|
||||
continue
|
||||
input_f = label_f.with_name(label_f.name.replace("_label", "_input"))
|
||||
try:
|
||||
temp = round(int(input_f.read_text().strip()) / 1000)
|
||||
except (OSError, ValueError):
|
||||
continue
|
||||
idx = label_f.name[len("temp"):-len("_label")]
|
||||
entries.append((idx, label, temp))
|
||||
dup = {lbl for lbl, n in Counter(l for _, l, _ in entries).items() if n > 1}
|
||||
for idx, label, temp in entries:
|
||||
name = f"{chip} {label}" + (f" {idx}" if label in dup else "")
|
||||
out.append({"name": name, "temperature_c": temp, "status": "ok", "kind": "env"})
|
||||
return out
|
||||
|
||||
|
||||
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
|
||||
packages: list[dict] = []
|
||||
cores: list[int] = []
|
||||
for label_f in sorted(hw.glob("temp*_label")):
|
||||
try:
|
||||
label = label_f.read_text().strip()
|
||||
except OSError:
|
||||
continue
|
||||
input_f = label_f.with_name(label_f.name.replace("_label", "_input"))
|
||||
try:
|
||||
temp = round(int(input_f.read_text().strip()) / 1000)
|
||||
except (OSError, ValueError):
|
||||
continue
|
||||
low = label.lower()
|
||||
if low.startswith("package id"):
|
||||
pkg = label.split("id")[-1].strip()
|
||||
packages.append({"name": f"CPU {pkg}", "temperature_c": temp, "kind": "cpu"})
|
||||
elif low.startswith("core"):
|
||||
cores.append(temp)
|
||||
# Prefer the per-package sensor; older CPUs (no package sensor) fall back
|
||||
# to the hottest core as the CPU temp.
|
||||
if packages:
|
||||
cpus.extend(packages)
|
||||
elif cores:
|
||||
cpus.append({"name": "CPU", "temperature_c": max(cores), "kind": "cpu"})
|
||||
return cpus
|
||||
@@ -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
|
||||
|
||||
@@ -23,9 +23,13 @@ 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_ENV_META = "jbod:host_poll:env_meta"
|
||||
K_HOST_DRIVE_META = "jbod:host_poll:drive_meta"
|
||||
K_LED_QUEUE = "jbod:led:queue"
|
||||
K_LOCK = "jbod:poll:lock"
|
||||
K_HOST_LOCK = "jbod:host_poll:lock"
|
||||
|
||||
|
||||
# ── writes (poller) ─────────────────────────────────────────────────────
|
||||
@@ -49,9 +53,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 +83,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 +138,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:
|
||||
|
||||
Reference in New Issue
Block a user