adam
0f829e0380
Rearchitect so a single poller process is the sole owner of all SAS/SMART/ SES hardware I/O, serialized behind one gate and paced, writing results to Redis. The API/web and MQTT publisher become pure Redis readers — they no longer issue any subprocess and can restart freely without touching the bus. This addresses backplane/expander stress from concurrent + restart-triggered SMART/SES storms (the prior model re-ran a hardware sweep on every container start, and polled sg_ses 0x02+0x07 every 60s; 0x07 errored on the IOM6/ Xyratex expanders). - poller.py: paced sweep (inventory, SMART per-drive w/ gap, SES 0x02, ZFS, host/MegaRAID), startup jitter, single-instance Redis lock, LED-queue worker - services/hwgate.py: global serialization semaphore (POLL_CONCURRENCY=1) - services/store.py: Redis as the only producer<->consumer interface + LED queue - services/health.py: shared drive-health classifier (fixes overview double-count) - gate smartctl/sg_ses/zpool/ledctl; drop sg_ses 0x07 from the hot path - routers + temps read-only from the store; main.py drops the in-process poller - compose: 3 services (privileged poller + unprivileged app + redis) w/ pacing knobs
135 lines
4.3 KiB
Python
135 lines
4.3 KiB
Python
"""Redis as the sole interface between the poller (producer) and the API/
|
|
MQTT consumers.
|
|
|
|
The poller writes hardware-derived state here; the API and MQTT publisher
|
|
read it and never touch hardware. Also carries the locate-LED command queue
|
|
(API enqueues, poller drains) and the poller single-instance lock.
|
|
"""
|
|
import json
|
|
import logging
|
|
import os
|
|
import time
|
|
|
|
from services.cache import cache_get, cache_set, get_client
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Data TTL: must comfortably exceed the sweep interval so values persist
|
|
# between sweeps (last-good is served stale rather than blanked).
|
|
DATA_TTL = int(os.environ.get("POLL_DATA_TTL", "900"))
|
|
|
|
K_SMART = "jbod:smart:{}"
|
|
K_SES = "jbod:ses:{}"
|
|
K_ZFS = "jbod:zfs_map"
|
|
K_INVENTORY = "jbod:inventory"
|
|
K_HOST_DRIVES = "jbod:host_drives"
|
|
K_META = "jbod:poll:meta"
|
|
K_LED_QUEUE = "jbod:led:queue"
|
|
K_LOCK = "jbod:poll:lock"
|
|
|
|
|
|
# ── writes (poller) ─────────────────────────────────────────────────────
|
|
async def set_smart(device: str, data: dict) -> None:
|
|
await cache_set(K_SMART.format(device), data, DATA_TTL)
|
|
|
|
|
|
async def set_ses(enc_id: str, data: dict) -> None:
|
|
await cache_set(K_SES.format(enc_id), data, DATA_TTL)
|
|
|
|
|
|
async def set_zfs_map(pool_map: dict) -> None:
|
|
await cache_set(K_ZFS, pool_map, DATA_TTL)
|
|
|
|
|
|
async def set_inventory(inventory: dict) -> None:
|
|
await cache_set(K_INVENTORY, inventory, DATA_TTL)
|
|
|
|
|
|
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:
|
|
# Meta has no TTL — it's the heartbeat; staleness is judged from its ts.
|
|
await cache_set(K_META, meta, 0)
|
|
|
|
|
|
# ── reads (consumers) ───────────────────────────────────────────────────
|
|
async def get_smart(device: str) -> dict | None:
|
|
return await cache_get(K_SMART.format(device))
|
|
|
|
|
|
async def get_ses(enc_id: str) -> dict | None:
|
|
return await cache_get(K_SES.format(enc_id))
|
|
|
|
|
|
async def get_zfs_map() -> dict:
|
|
return await cache_get(K_ZFS) or {}
|
|
|
|
|
|
async def get_inventory() -> dict:
|
|
return await cache_get(K_INVENTORY) or {"enclosures": []}
|
|
|
|
|
|
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)
|
|
|
|
|
|
# ── locate-LED command queue ────────────────────────────────────────────
|
|
async def enqueue_led(device: str, state: str) -> bool:
|
|
"""API side: push a LED request for the poller to execute. Returns False
|
|
if Redis is unavailable (so the API can surface 503)."""
|
|
client = get_client()
|
|
if client is None:
|
|
return False
|
|
await client.rpush(K_LED_QUEUE, json.dumps({"device": device, "state": state}))
|
|
return True
|
|
|
|
|
|
async def pop_led(timeout: int = 5) -> dict | None:
|
|
"""Poller side: block up to `timeout`s for the next LED request."""
|
|
client = get_client()
|
|
if client is None:
|
|
return None
|
|
item = await client.blpop(K_LED_QUEUE, timeout=timeout)
|
|
if not item:
|
|
return None
|
|
_key, raw = item
|
|
try:
|
|
return json.loads(raw)
|
|
except (ValueError, TypeError):
|
|
return None
|
|
|
|
|
|
# ── poller single-instance lock ─────────────────────────────────────────
|
|
async def acquire_lock(token: str, ttl: int = 30) -> bool:
|
|
"""Try to claim the poller lock. Returns True if acquired/owned."""
|
|
client = get_client()
|
|
if client is None:
|
|
return True # no Redis → no coordination possible; run anyway
|
|
got = await client.set(K_LOCK, token, nx=True, ex=ttl)
|
|
if got:
|
|
return True
|
|
current = await client.get(K_LOCK)
|
|
return current == token
|
|
|
|
|
|
async def refresh_lock(token: str, ttl: int = 30) -> bool:
|
|
"""Extend the lock if we still own it."""
|
|
client = get_client()
|
|
if client is None:
|
|
return True
|
|
current = await client.get(K_LOCK)
|
|
if current == token:
|
|
await client.expire(K_LOCK, ttl)
|
|
return True
|
|
return False
|
|
|
|
|
|
def now() -> float:
|
|
return time.time()
|