Dedicated hardware poller + read-only consumers #4
@@ -52,7 +52,8 @@ services:
|
||||
- REDIS_HOST=127.0.0.1
|
||||
- REDIS_PORT=6379
|
||||
- REDIS_DB=0
|
||||
- HOST_POLL_INTERVAL=60
|
||||
- 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
|
||||
|
||||
121
host_poller.py
121
host_poller.py
@@ -1,12 +1,16 @@
|
||||
"""Host-poller — server chassis temps (IPMI/iDRAC + CPU) and on-metal drives.
|
||||
"""Host-poller — server chassis temps, split into two independent loops.
|
||||
|
||||
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.
|
||||
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:
|
||||
|
||||
Host-drive ZFS annotation reuses jbod:zfs_map written by the SAS poller.
|
||||
- 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
|
||||
@@ -26,25 +30,24 @@ logging.basicConfig(
|
||||
)
|
||||
logger = logging.getLogger("host-poller")
|
||||
|
||||
SWEEP_INTERVAL = int(os.environ.get("HOST_POLL_INTERVAL", "120"))
|
||||
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() -> None:
|
||||
async def sweep_env() -> None:
|
||||
"""IPMI/iDRAC environmental + CPU temps (the responsive loop)."""
|
||||
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({
|
||||
@@ -53,30 +56,57 @@ async def sweep() -> None:
|
||||
"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,
|
||||
"last_sweep_duration": round(time.monotonic() - t0, 1),
|
||||
"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),
|
||||
)
|
||||
"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:
|
||||
@@ -117,30 +147,17 @@ async def main() -> None:
|
||||
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)
|
||||
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:
|
||||
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))
|
||||
await asyncio.gather(env_task, drive_task)
|
||||
finally:
|
||||
refresher.cancel()
|
||||
for t in (refresher, env_task, drive_task):
|
||||
t.cancel()
|
||||
await close_cache()
|
||||
|
||||
|
||||
|
||||
@@ -25,7 +25,8 @@ 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_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"
|
||||
|
||||
Reference in New Issue
Block a user