feat(fleet): Auslastungs-Monitor pro Box — live nvidia-smi + Graphen (Stage E)
Pro Box ein "Auslastung"-Button in der Compute-Flotte → Modal mit live nvidia-smi (1s), Graphen (GPU-Auslastung + Tokens/Intervall) und Besen-Reset. Historie liegt auf der Box, Diagnostic holt sie via RVS. - node_stats.py (identisch in allen 4 Worker-Build-Contexts): Sampler alle 15s (nvidia-smi + Token-Delta → Ringpuffer ~500 Punkte, persistent als JSON auf der Box), Live-Stream (node_stats, 1s, Auto-Stop 300s), History-Request, Reset. nvidia-smi via async subprocess, fail-safe ohne GPU. - Worker-Wiring (f5tts/whisper/voxtral/llm-adapter): Import, Sampler-Task, _stats.handle() nach dem targetInstance-Filter. llm-adapter zaehlt Tokens (usage.total_tokens) → Token-Graph nur bei LLM-Boxen. Dockerfiles kopieren node_stats.py. - compose: llm-adapter bekommt runtime:nvidia + NVIDIA_VISIBLE_DEVICES=all + DRIVER_CAPABILITIES=utility (nur nvidia-smi, KEIN VRAM/Compute). - diagnostic/server.js: relay node_stats_* (Browser→Box) + forward (Box→Browser). - diagnostic/index.html: Auslastung-Button pro Node (Ziel bevorzugt llm-Instanz), Modal mit live nvidia-smi + Inline-SVG-Sparklines, Besen-Reset. Reporter-Wahl bevorzugt die llm-Instanz (sieht alle GPUs + Tokens); GPU-Worker sehen ihre gepinnte Karte. Gitignored Historie stoert git-Baum der Box nicht. Deploy: diagnostic + GPU-Boxen neu bauen. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.8
@@ -0,0 +1,167 @@
|
||||
"""
|
||||
ARIA Node-Stats — Auslastungs-Monitor pro Box (GPU + optional Tokens).
|
||||
|
||||
Identische Kopie in jedem Worker-Build-Context (f5tts/whisper/voxtral/llm-adapter),
|
||||
weil jeder Worker ein eigener Docker-Build-Context ist.
|
||||
|
||||
Aufgaben:
|
||||
- Sampler-Loop (alle SAMPLE_SEC): nvidia-smi-Auslastung + Token-Delta → Ringpuffer
|
||||
(persistent als JSON auf der Box). Laeuft unabhaengig vom Modal.
|
||||
- Live-Stream: bei node_stats_stream_start jede Sekunde rohes nvidia-smi + Werte
|
||||
senden (bis stop / Auto-Timeout).
|
||||
- History-Request + Reset (Besen).
|
||||
|
||||
Reicht `handle(ws, mtype, payload)` in die Worker-Message-Loop ein; gibt True
|
||||
zurueck, wenn die Nachricht eine node_stats_*-Nachricht war.
|
||||
"""
|
||||
import asyncio
|
||||
import json
|
||||
import os
|
||||
import time
|
||||
|
||||
SAMPLE_SEC = int(os.getenv("STATS_SAMPLE_SEC", "15"))
|
||||
HISTORY_CAP = int(os.getenv("STATS_HISTORY_CAP", "500")) # ~2h bei 15s
|
||||
STREAM_MAX_SEC = int(os.getenv("STATS_STREAM_MAX_SEC", "300"))
|
||||
|
||||
|
||||
async def _run_cmd(*args, timeout=8) -> str:
|
||||
"""Fuehrt ein Kommando aus, gibt stdout (str) zurueck; '' bei Fehler."""
|
||||
try:
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
*args,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.DEVNULL,
|
||||
)
|
||||
out, _ = await asyncio.wait_for(proc.communicate(), timeout=timeout)
|
||||
return (out or b"").decode("utf-8", "replace")
|
||||
except Exception:
|
||||
return ""
|
||||
|
||||
|
||||
class NodeStats:
|
||||
def __init__(self, instance_id: str, node_name: str, history_path: str,
|
||||
token_getter=None, logger=None):
|
||||
self.instance_id = instance_id
|
||||
self.node_name = node_name
|
||||
self.history_path = history_path
|
||||
self.token_getter = token_getter # callable -> kumulative Token-Zahl (oder None)
|
||||
self.log = logger
|
||||
self.samples = self._load()
|
||||
self._last_tokens = self._tokens_now()
|
||||
self._stream_task = None
|
||||
|
||||
# ── Persistenz ──────────────────────────────────────────
|
||||
def _load(self) -> list:
|
||||
try:
|
||||
with open(self.history_path) as f:
|
||||
data = json.load(f)
|
||||
return data if isinstance(data, list) else []
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
def _persist(self) -> None:
|
||||
try:
|
||||
os.makedirs(os.path.dirname(self.history_path) or ".", exist_ok=True)
|
||||
tmp = self.history_path + ".tmp"
|
||||
with open(tmp, "w") as f:
|
||||
json.dump(self.samples[-HISTORY_CAP:], f)
|
||||
os.replace(tmp, self.history_path)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _tokens_now(self) -> int:
|
||||
try:
|
||||
return int(self.token_getter()) if self.token_getter else 0
|
||||
except Exception:
|
||||
return 0
|
||||
|
||||
# ── nvidia-smi ──────────────────────────────────────────
|
||||
async def _query_gpu(self) -> dict:
|
||||
"""Aggregierte GPU-Werte ueber alle sichtbaren Karten."""
|
||||
out = await _run_cmd(
|
||||
"nvidia-smi",
|
||||
"--query-gpu=utilization.gpu,memory.used,memory.total",
|
||||
"--format=csv,noheader,nounits")
|
||||
utils, used, total = [], 0, 0
|
||||
for line in out.strip().splitlines():
|
||||
parts = [p.strip() for p in line.split(",")]
|
||||
if len(parts) < 3:
|
||||
continue
|
||||
try:
|
||||
utils.append(float(parts[0]))
|
||||
used += float(parts[1])
|
||||
total += float(parts[2])
|
||||
except ValueError:
|
||||
continue
|
||||
gpu = round(sum(utils) / len(utils), 1) if utils else 0.0
|
||||
return {"gpu": gpu, "memUsed": int(used), "memTotal": int(total)}
|
||||
|
||||
async def _nvidia_smi_text(self) -> str:
|
||||
txt = await _run_cmd("nvidia-smi")
|
||||
return txt or "nvidia-smi nicht verfuegbar"
|
||||
|
||||
# ── Sampler (Verlauf) ───────────────────────────────────
|
||||
async def run_sampler(self) -> None:
|
||||
while True:
|
||||
try:
|
||||
g = await self._query_gpu()
|
||||
now_tok = self._tokens_now()
|
||||
dtok = max(0, now_tok - self._last_tokens)
|
||||
self._last_tokens = now_tok
|
||||
self.samples.append({
|
||||
"ts": int(time.time()),
|
||||
"gpu": g["gpu"], "memUsed": g["memUsed"],
|
||||
"memTotal": g["memTotal"], "tokens": dtok,
|
||||
})
|
||||
if len(self.samples) > HISTORY_CAP:
|
||||
self.samples = self.samples[-HISTORY_CAP:]
|
||||
self._persist()
|
||||
except Exception as e:
|
||||
if self.log:
|
||||
self.log.debug("node_stats sample fehlgeschlagen: %s", e)
|
||||
await asyncio.sleep(SAMPLE_SEC)
|
||||
|
||||
# ── Live-Stream ─────────────────────────────────────────
|
||||
async def _stream(self, ws, send) -> None:
|
||||
t0 = time.time()
|
||||
try:
|
||||
while time.time() - t0 < STREAM_MAX_SEC:
|
||||
g = await self._query_gpu()
|
||||
smi = await self._nvidia_smi_text()
|
||||
await send(ws, "node_stats", {
|
||||
"instanceId": self.instance_id, "node": self.node_name,
|
||||
"nvidiaSmi": smi, **g,
|
||||
})
|
||||
await asyncio.sleep(1)
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception:
|
||||
return
|
||||
|
||||
# ── Dispatch ────────────────────────────────────────────
|
||||
async def handle(self, ws, mtype: str, payload: dict, send) -> bool:
|
||||
if mtype == "node_stats_stream_start":
|
||||
if self._stream_task and not self._stream_task.done():
|
||||
self._stream_task.cancel()
|
||||
self._stream_task = asyncio.create_task(self._stream(ws, send))
|
||||
return True
|
||||
if mtype == "node_stats_stream_stop":
|
||||
if self._stream_task:
|
||||
self._stream_task.cancel()
|
||||
self._stream_task = None
|
||||
return True
|
||||
if mtype == "node_stats_history_request":
|
||||
await send(ws, "node_stats_history", {
|
||||
"instanceId": self.instance_id, "node": self.node_name,
|
||||
"samples": self.samples[-HISTORY_CAP:],
|
||||
"tokenCapable": self.token_getter is not None,
|
||||
"sampleSec": SAMPLE_SEC,
|
||||
})
|
||||
return True
|
||||
if mtype == "node_stats_reset":
|
||||
self.samples = []
|
||||
self._persist()
|
||||
await send(ws, "node_stats_reset_done",
|
||||
{"instanceId": self.instance_id, "node": self.node_name})
|
||||
return True
|
||||
return False
|
||||
Reference in New Issue
Block a user