diff --git a/backend/app.py b/backend/app.py index 14dff2f..fd77f14 100644 --- a/backend/app.py +++ b/backend/app.py @@ -128,6 +128,7 @@ def _box_router(app: FastAPI) -> None: gateway_proxy, hermes_ui, maintenance, + messwerte, models, radar, routing, @@ -140,6 +141,7 @@ def _box_router(app: FastAPI) -> None: app.include_router(boxwart.router) # Cockpit: Start, Hinweise, Modell-Nutzung, Zeitplan app.include_router(einstellungen.router) # Update-Zeitfenster, Radar-Schalter, Telegram-Test app.include_router(radar.router) # Modell-Radar: Kandidaten, Nachttests, Übernehmen/Verwerfen + app.include_router(messwerte.router) # Messwerte mit Verlauf (1h, 24h, 7d; schreibt der Steward) app.include_router(models.router) app.include_router(routing.router) app.include_router(system.router) diff --git a/backend/kern/messreihen.py b/backend/kern/messreihen.py new file mode 100644 index 0000000..c166c98 --- /dev/null +++ b/backend/kern/messreihen.py @@ -0,0 +1,301 @@ +"""Messreihen: Minutenwerte mit Verlauf für beide Bereiche (Monitoring, seit 24.09.2026). + +Ablage: je Quelle und Tag eine Datei, darin eine JSON-Zeile je Minute: + /mc2-messwerte/-JJJJ-MM-TT.jsonl (Tag nach Berliner Zeit, kern/zeit.py) + {"t":1790000000,"cpu":12.5,"ram":41.2,…} (t = Unix-Sekunden der Messung, null = keine Messung) +Quellen: „box“ (die KI-Box; schreibt ihr Steward) sowie „pve“, „ct-“ und „vm-“ (das Homelab; schreibt der +Homelab-Teil, was der Ausführer jede Minute schickt). Dateien, deren Tag mehr als AUFBEWAHREN_TAGE zurückliegt, löscht +das erste Schreiben eines neuen Tages. + +Schreiben: eine Zeile anhängen, unter einer Thread-Sperre und (Linux) flock auf der Datei. Fehlt am Dateiende der +Zeilenumbruch (Absturz mitten im Schreiben), beginnt die neue Zeile auf einer eigenen statt an die kaputte anzuwachsen. + +Lesen (lesen()): 1h in 60-s-Schritten, 24h als 5-min-Mittel, 7d als 30-min-Mittel. Die Schritte liegen auf vollen +Vielfachen ihrer Länge, der letzte ist der laufende. Ein Schritt ohne Messung bleibt null — Lücken werden nicht +aufgefüllt. Kaputte Zeilen werden übersprungen. Die Summen je Tagesdatei werden für 5- und 30-min-Schritte gemerkt, +solange sich die Datei nicht ändert (vergangene Tage liest so nur die erste Anfrage). +""" + +import json +import logging +import math +import os +import re +import threading +import time +from collections.abc import Iterable, Iterator +from datetime import datetime, timedelta +from pathlib import Path + +from kern.einstellungen import einstellungen +from kern.zeit import LOCAL_TZ + +try: + import fcntl # Linux: Sperre über Prozessgrenzen +except ImportError: # Windows (Entwicklung): nur die Thread-Sperre + fcntl = None + +log = logging.getLogger(__name__) + +ORDNER_NAME = "mc2-messwerte" +AUFBEWAHREN_TAGE = 8 +# Zeitraum → (Länge, Schritt) in Sekunden. +ZEITRAEUME: dict[str, tuple[int, int]] = {"1h": (3600, 60), "24h": (24 * 3600, 300), "7d": (7 * 24 * 3600, 1800)} +MERKEN_AB_S = 300 # Summen je Datei merken; bei 60-s-Schritten (1440 je Tag) lohnt es den Speicher nicht +MERKEN_MAX = 512 +LUECKE_MAX_S = 300 # Zählerstände, die weiter auseinanderliegen, ergeben keine Rate mehr +RATE_MAX = 5e9 # mehr als 5 GB/s ist kein Verkehr, sondern ein Zählersprung + +_QUELLE = re.compile(r"[a-z0-9][a-z0-9-]{0,62}") +_DATEI = re.compile(r"(?P[a-z0-9][a-z0-9-]*)-(?P\d{4}-\d{2}-\d{2})\.jsonl") + +_lock = threading.Lock() +_aufgeraeumt: dict[str, str] = {} # Ordner → Tag des letzten Aufräumens (je Prozess) +_merk_lock = threading.Lock() +_merk: dict[tuple[str, int], tuple[tuple[int, int], dict[int, dict[str, list[float]]]]] = {} + + +def ordner() -> Path: + return einstellungen().daten_dir / ORDNER_NAME + + +def ist_zahl(wert: object) -> bool: + """Eine endliche Zahl. bool zählt nicht, NaN und unendlich auch nicht.""" + return isinstance(wert, (int, float)) and not isinstance(wert, bool) and math.isfinite(wert) + + +def zeitraum_pruefen(zeitraum: str) -> tuple[int, int]: + """(Länge, Schritt) in Sekunden; ValueError bei einem unbekannten Zeitraum.""" + if zeitraum not in ZEITRAEUME: + raise ValueError(f"Den Zeitraum „{zeitraum}“ gibt es nicht; möglich sind {', '.join(ZEITRAEUME)}.") + return ZEITRAEUME[zeitraum] + + +def _datum(t: float): + return datetime.fromtimestamp(t, LOCAL_TZ).date() + + +def _tage(von: float, bis: float) -> list[str]: + """Alle Tage (Berliner Zeit), deren Dateien Messungen aus [von, bis) enthalten können.""" + tag, letzter = _datum(von), _datum(max(von, bis - 1)) + tage = [] + while tag <= letzter: + tage.append(tag.isoformat()) + tag += timedelta(days=1) + return tage + + +def _datei(quelle: str, tag: str) -> Path: + if not _QUELLE.fullmatch(quelle): + raise ValueError(f"Ungültiger Name einer Messquelle: {quelle!r}") + return ordner() / f"{quelle}-{tag}.jsonl" + + +# --- Schreiben ----------------------------------------------------------------------------------- + +def _wert(wert: object) -> int | float | None: + if not ist_zahl(wert): + return None + return wert if isinstance(wert, int) else round(float(wert), 3) + + +def schreiben(quelle: str, werte: dict[str, object], t: float | None = None) -> None: + """Einen Messpunkt anhängen. werte: Name → Zahl oder None (nicht gemessen).""" + t = time.time() if t is None else float(t) + punkt: dict[str, object] = {"t": int(t)} + punkt.update({name: _wert(wert) for name, wert in werte.items() if name != "t"}) + zeile = (json.dumps(punkt, separators=(",", ":")) + "\n").encode() + pfad = _datei(quelle, _datum(t).isoformat()) + with _lock: + pfad.parent.mkdir(parents=True, exist_ok=True) + with open(pfad, "a+b") as f: + if fcntl is not None: + fcntl.flock(f, fcntl.LOCK_EX) + try: + f.seek(0, os.SEEK_END) + if f.tell() > 0: + f.seek(-1, os.SEEK_END) + if f.read(1) != b"\n": + zeile = b"\n" + zeile # die letzte Zeile brach ab: nicht an sie anhängen + f.write(zeile) + f.flush() + finally: + if fcntl is not None: + fcntl.flock(f, fcntl.LOCK_UN) + _einmal_am_tag_aufraeumen(t) + + +def _einmal_am_tag_aufraeumen(t: float) -> None: + schluessel, tag = str(ordner()), _datum(t).isoformat() + if _aufgeraeumt.get(schluessel) == tag: + return + _aufgeraeumt[schluessel] = tag + try: + aufraeumen(t) + except OSError: + log.warning("messreihen: Aufräumen in %s ging nicht", schluessel, exc_info=True) + + +def aufraeumen(jetzt: float | None = None) -> int: + """Tagesdateien löschen, deren Tag mehr als AUFBEWAHREN_TAGE zurückliegt. Andere Dateien bleiben. Rückgabe: wie + viele gelöscht wurden.""" + grenze = (_datum(time.time() if jetzt is None else jetzt) - timedelta(days=AUFBEWAHREN_TAGE)).isoformat() + try: + namen = os.listdir(ordner()) + except FileNotFoundError: + return 0 + geloescht = 0 + for name in namen: + treffer = _DATEI.fullmatch(name) + if treffer and treffer["tag"] < grenze: + try: + (ordner() / name).unlink() + geloescht += 1 + except FileNotFoundError: + pass # ein anderer Prozess war schneller + return geloescht + + +# --- Lesen --------------------------------------------------------------------------------------- + +def _punkt(zeile: bytes) -> dict | None: + """Eine Zeile als Messpunkt; None bei allem, was keiner ist (kaputt, leer, ohne Zeit).""" + zeile = zeile.strip() + if not zeile: + return None + try: + punkt = json.loads(zeile) + except ValueError: # auch UnicodeDecodeError + return None + return punkt if isinstance(punkt, dict) and ist_zahl(punkt.get("t")) else None + + +def _punkte_der_datei(pfad: Path) -> Iterator[dict]: + try: + roh = pfad.read_bytes() + except OSError: + return + for zeile in roh.split(b"\n"): + if (punkt := _punkt(zeile)) is not None: + yield punkt + + +def _eimer(pfad: Path, schritt: int, ab: float) -> dict[int, dict[str, list[float]]]: + """Summe und Anzahl je Schritt und Wert für eine Tagesdatei. Gemerkt (ab 5-min-Schritten) wird die ganze Datei; + sonst zählen erst Messungen ab `ab`. Das Ergebnis nicht verändern: es kann gemerkt sein.""" + try: + stand = pfad.stat() + except OSError: + return {} + merken = schritt >= MERKEN_AB_S + schluessel, kennung = (str(pfad), schritt), (stand.st_mtime_ns, stand.st_size) + if merken: + with _merk_lock: + gemerkt = _merk.get(schluessel) + if gemerkt and gemerkt[0] == kennung: + return gemerkt[1] + eimer: dict[int, dict[str, list[float]]] = {} + for punkt in _punkte_der_datei(pfad): + if not merken and punkt["t"] < ab: + continue + ziel = eimer.setdefault(int(punkt["t"]) // schritt * schritt, {}) + for name, wert in punkt.items(): + if name != "t" and ist_zahl(wert): + summe = ziel.setdefault(name, [0.0, 0]) + summe[0] += wert + summe[1] += 1 + if merken: + with _merk_lock: + if len(_merk) >= MERKEN_MAX: + _merk.clear() + _merk[schluessel] = (kennung, eimer) + return eimer + + +def _mittel(summe: list[float] | None) -> int | float | None: + """Mittel eines Schritts: ab 100 ganzzahlig, darunter eine Nachkommastelle; ohne Messung None.""" + if not summe or not summe[1]: + return None + wert = summe[0] / summe[1] + return round(wert) if abs(wert) >= 100 else round(wert, 1) + + +def lesen(quelle: str, zeitraum: str, namen: Iterable[str], jetzt: float | None = None) -> dict: + """Die Reihen einer Quelle: {"zeitraum", "schritt_s", "reihen": {name: [[t, wert], …]}}. t = Beginn des Schritts + (Unix-Sekunden), wert = Mittel der Messungen darin oder None. Jede Reihe hat Länge/Schritt Einträge (60, 288, 336).""" + dauer, schritt = zeitraum_pruefen(zeitraum) + jetzt = time.time() if jetzt is None else jetzt + ende = (int(jetzt) // schritt + 1) * schritt # Ende des laufenden Schritts + beginn = ende - dauer + summen: dict[int, dict[str, list[float]]] = {} + for tag in _tage(beginn, ende): + for t0, werte in _eimer(_datei(quelle, tag), schritt, beginn).items(): + if not beginn <= t0 < ende: + continue + ziel = summen.setdefault(t0, {}) + for name, (summe, anzahl) in werte.items(): + gesamt = ziel.setdefault(name, [0.0, 0]) + gesamt[0] += summe + gesamt[1] += anzahl + raster = range(beginn, ende, schritt) + return {"zeitraum": zeitraum, "schritt_s": schritt, + "reihen": {name: [[t0, _mittel(summen.get(t0, {}).get(name))] for t0 in raster] for name in namen}} + + +def _ende_der_datei(pfad: Path, groesse: int = 16384) -> list[bytes]: + """Die letzten Zeilen einer Datei, ohne sie ganz zu lesen.""" + try: + with open(pfad, "rb") as f: + f.seek(0, os.SEEK_END) + laenge = f.tell() + f.seek(max(0, laenge - groesse)) + roh = f.read() + except OSError: + return [] + zeilen = roh.split(b"\n") + return zeilen[1:] if laenge > groesse else zeilen # die erste Zeile ist dann angeschnitten + + +def letzter_punkt(quelle: str, jetzt: float | None = None, max_alter_s: float | None = None) -> dict | None: + """Die jüngste Messung einer Quelle (heute, sonst gestern). None, wenn es keine gibt oder sie älter ist als + max_alter_s — ein alter Wert ist kein aktueller.""" + jetzt = time.time() if jetzt is None else jetzt + heute = _datum(jetzt) + for tag in (heute, heute - timedelta(days=1)): + for zeile in reversed(_ende_der_datei(_datei(quelle, tag.isoformat()))): + if (punkt := _punkt(zeile)) is not None: + if max_alter_s is not None and jetzt - punkt["t"] > max_alter_s: + return None + return punkt + return None + + +def punkte(quelle: str, von: float, bis: float) -> list[dict]: + """Alle Messungen einer Quelle mit von ≤ t < bis, nach Zeit geordnet (für Prüfungen des Wächters).""" + gefunden = [p for tag in _tage(von, bis) for p in _punkte_der_datei(_datei(quelle, tag)) if von <= p["t"] < bis] + return sorted(gefunden, key=lambda p: p["t"]) + + +def quellen(von: float, bis: float) -> list[str]: + """Die Quellen, von denen es für [von, bis) eine Tagesdatei gibt.""" + tage = set(_tage(von, bis)) + try: + namen = os.listdir(ordner()) + except OSError: + return [] + return sorted({t["quelle"] for name in namen if (t := _DATEI.fullmatch(name)) and t["tag"] in tage}) + + +# --- Zähler → Raten -------------------------------------------------------------------------------- + +def rate(alt: object, neu: object, dt: float) -> float | None: + """Zuwachs je Sekunde zwischen zwei Ständen eines kumulativen Zählers (Bytes, Tokens). None, wenn ein Stand fehlt, + die Stände mehr als LUECKE_MAX_S auseinanderliegen, der Zähler kleiner wurde (Neustart, Überlauf) oder der Sprung + unmöglich groß ist — eine erfundene Rate wäre schlimmer als eine Lücke.""" + if not (ist_zahl(alt) and ist_zahl(neu)) or not 0 < dt <= LUECKE_MAX_S: + return None + zuwachs = float(neu) - float(alt) + if zuwachs < 0: + return None + wert = zuwachs / dt + return None if wert > RATE_MAX else wert diff --git a/backend/routers/homelab.py b/backend/routers/homelab.py index 4224c45..ab5ecb6 100644 --- a/backend/routers/homelab.py +++ b/backend/routers/homelab.py @@ -155,3 +155,22 @@ async def _strom(request: Request, takt_s: float = 2.0, lebenszeichen_takte: int async def stream(request: Request) -> StreamingResponse: return StreamingResponse(_strom(request), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}) + + +# --- Messwerte mit Verlauf (Monitoring, seit 24.09.2026) ------------------------------------------------------ +# POST /api/homelab/ausfuehrer/messwerte jede Minute vom Ausführer (Kopfzeile X-MC2-Ausfuehrer) +# GET /api/homelab/messwerte?zeitraum=1h|24h|7d Proxmox-Host und Gäste: Reihen und letzter Punkt +# Die Logik liegt in services/homelab/messwerte.py, die Ablage in kern/messreihen.py. + +@router.post("/ausfuehrer/messwerte", dependencies=[Depends(_nur_ausfuehrer)]) +def messwerte_annehmen(daten: dict) -> dict: + from services.homelab import messwerte as homelab_messwerte + return {"ok": True, "geraete": homelab_messwerte.annehmen(daten)} + + +@router.get("/messwerte") +def messwerte_lesen(zeitraum: str = "1h") -> dict: + from services.homelab import messwerte as homelab_messwerte + if zeitraum not in homelab_messwerte.ZEITRAEUME: + raise HTTPException(status_code=400, detail="Den Zeitraum gibt es nicht; möglich sind 1h, 24h und 7d.") + return homelab_messwerte.abfrage(zeitraum) diff --git a/backend/routers/messwerte.py b/backend/routers/messwerte.py new file mode 100644 index 0000000..217fa53 --- /dev/null +++ b/backend/routers/messwerte.py @@ -0,0 +1,18 @@ +"""Messwerte der KI-Box mit Verlauf (Rolle box, seit 24.09.2026). + + GET /api/messwerte?zeitraum=1h|24h|7d Minutenwerte des Stewards, verdichtet auf 60 s, 5 min bzw. 30 min + +Dünn: die Logik liegt in services/messwerte.py, die Ablage in kern/messreihen.py. +""" + +from fastapi import APIRouter, HTTPException +from services import messwerte + +router = APIRouter(prefix="/api", tags=["messwerte"]) + + +@router.get("/messwerte") +def messwerte_lesen(zeitraum: str = "1h") -> dict: + if zeitraum not in messwerte.ZEITRAEUME: + raise HTTPException(status_code=400, detail="Den Zeitraum gibt es nicht; möglich sind 1h, 24h und 7d.") + return messwerte.abfrage(zeitraum) diff --git a/backend/services/aufraeumen.py b/backend/services/aufraeumen.py index 31f3f58..b6795c7 100644 --- a/backend/services/aufraeumen.py +++ b/backend/services/aufraeumen.py @@ -30,7 +30,7 @@ from config import MODELS_DIR from services import llamaswap AKTIVE_ROLLEN = {"hermes", "fast", "coder", "heavy", "vision", "coder-bild", "embed", "reranker"} -SYSTEM_ORDNER = {"mc2-backups", "radar", "pruefstand-cache", "lost+found"} +SYSTEM_ORDNER = {"mc2-backups", "mc2-messwerte", "radar", "pruefstand-cache", "lost+found"} AUF_DER_BOX = os.name == "posix" # auf dem Windows-PC (Entwicklung) wird nie gelöscht # Was der Nutzer über bekannte Ordner wissen sollte, bevor er löscht (Stand 24.09.2026). diff --git a/backend/services/homelab/messwerte.py b/backend/services/homelab/messwerte.py new file mode 100644 index 0000000..714aa93 --- /dev/null +++ b/backend/services/homelab/messwerte.py @@ -0,0 +1,247 @@ +"""Messwerte des Homelabs mit Verlauf (Monitoring, seit 24.09.2026). + +Der Ausführer auf dem Proxmox-Host schickt jede Minute POST /api/homelab/ausfuehrer/messwerte (eigener Faden, unabhängig +vom 10-min-Bericht und von Aufträgen): + {"zeit": Unix-Sekunden der Messung, + "host": {"cpu": 0–1 (Mittel der Minute), "speicher": {"belegt", "gesamt"}, "rootfs": {"belegt", "gesamt"}, "load1", + "uptime", "temp_cpu" (°C oder null), "netz": {"rx", "tx"} (kumulativ, Bytes)}, + "gaeste": [{"vmid", "art" (lxc|qemu), "name", "etiketten", "status", "cpu" (0–1 der eigenen Kerne), "maxcpu", "mem", + "maxmem", "disk", "maxdisk", "netin", "netout" (kumulativ, Bytes), "uptime"}]} +Hier wird daraus je Gerät ein Punkt über kern/messreihen.py (Quellen „pve“, „ct-“, „vm-“ wie im Inventar): +CPU, RAM und Platte in %, Netz in Bytes/s aus der Differenz zum vorigen Zählerstand. Ist ein Zähler kleiner geworden +oder die Laufzeit (Neustart des Gasts oder Hosts), fehlt der vorige Stand (Neustart dieses Teils) oder liegt er mehr als +fünf Minuten zurück, bleibt die Rate dieser Minute null. Gestoppte Gäste haben keine Messwerte (null), keine Nullen. +Der eigene Container (Etikett „mc2“) wird wie im Inventar nicht erfasst. Die Platte von VMs sieht Proxmox nicht (null). + +GET /api/homelab/messwerte liefert je Gerät die Reihen (1h, 24h, 7d) und den letzten Punkt; pruefe_host() meldet dem +Wächter, wenn der Host zehn Minuten lang über 95 % RAM oder 90 °C liegt. +""" + +import os +import re +import threading +import time + +from kern import messreihen + +from services.homelab import apps, inventar, kanal + +HOST = "pve" +_GAST = re.compile(r"(ct|vm)-\d+") +ZEITRAEUME = messreihen.ZEITRAEUME +HOST_REIHEN = ("cpu", "ram", "platte", "netz_rx", "netz_tx", "temp_cpu") +GAST_REIHEN = ("cpu", "ram", "platte", "netz_rx", "netz_tx") +AKTUELL = ("cpu", "ram", "ram_used", "ram_total", "platte", "netz_rx", "netz_tx", "temp_cpu", "load1", "uptime_s") +AKTUELL_MAX_S = 180 # älter ist der letzte Punkt nicht „aktuell“ (der Ausführer schickt gerade nichts) +ZEIT_TOLERANZ_S = 120 # die Messzeit des Ausführers gilt, wenn sie so nah an der eigenen Uhr liegt + +# Wächter: der Host zehn Minuten lang über diesen Schwellen → gelber Hinweis. +WACHE_S = 600 +WACHE_MIN_PUNKTE = 8 # so viele Minutenpunkte müssen in den zehn Minuten liegen (ein, zwei dürfen fehlen) +RAM_GELB = float(os.environ.get("MC_WAECHTER_HOST_RAM", "95")) +TEMP_GELB = float(os.environ.get("MC_WAECHTER_HOST_TEMP", "90")) + +_lock = threading.Lock() +# Vorige Zählerstände je Gerät (für die Raten) und die Geräte der letzten Meldung (für Namen und die Geräteliste). +# Nur im Speicher: Nach einem Neustart dieses Teils fehlt für eine Minute die Netz-Rate, sonst nichts. +_zustand: dict = {"zaehler": {}, "geraete": {}} + + +def _zahl(wert: object) -> float | None: + return float(wert) if messreihen.ist_zahl(wert) else None + + +def _ganz(wert: object) -> int | None: + return int(wert) if messreihen.ist_zahl(wert) else None + + +def _prozent(anteil: object) -> float | None: + """0–1 → %, eine Nachkommastelle, auf 0–100 begrenzt.""" + return round(min(100.0, max(0.0, float(anteil) * 100)), 1) if messreihen.ist_zahl(anteil) else None + + +def _anteil(teil: object, ganz: object) -> float | None: + if not (messreihen.ist_zahl(teil) and messreihen.ist_zahl(ganz)) or float(ganz) <= 0: + return None + return _prozent(float(teil) / float(ganz)) + + +def _dict(wert: object) -> dict: + return wert if isinstance(wert, dict) else {} + + +def gast_id(gast: dict) -> str | None: + """ct- bzw. vm- wie im Inventar; None bei allem, was kein Gast ist.""" + vmid, art = gast.get("vmid"), gast.get("art") + if not isinstance(vmid, int) or isinstance(vmid, bool) or not 100 <= vmid <= 999_999_999: + return None + return {"lxc": f"ct-{vmid}", "qemu": f"vm-{vmid}"}.get(art) + + +def _art(geraet_id: str) -> str: + return "proxmox-host" if geraet_id == HOST else "container" if geraet_id.startswith("ct-") else "vm" + + +# --- Zähler → Bytes/s ------------------------------------------------------------------------------ + +def raten(vorher: dict | None, jetzt: dict) -> tuple[int | None, int | None]: + """Empfangen und gesendet in Bytes/s zwischen zwei Zählerständen {"t", "rx", "tx", "uptime", "laeuft"}. None, wenn + es keinen vorigen Stand gibt, einer der beiden nicht lief, die Laufzeit kleiner wurde (neu gestartet: die Zähler + begannen von vorn) oder messreihen.rate() den Zuwachs verwirft (Zähler kleiner, Lücke, Sprung).""" + if not vorher or not vorher.get("laeuft") or not jetzt.get("laeuft"): + return None, None + alt_lauf, neu_lauf = vorher.get("uptime"), jetzt.get("uptime") + if messreihen.ist_zahl(alt_lauf) and messreihen.ist_zahl(neu_lauf) and neu_lauf < alt_lauf: + return None, None + dt = jetzt["t"] - vorher["t"] + rx, tx = messreihen.rate(vorher.get("rx"), jetzt.get("rx"), dt), messreihen.rate(vorher.get("tx"), jetzt.get("tx"), dt) + return (None if rx is None else round(rx)), (None if tx is None else round(tx)) + + +def _zaehlen(geraet_id: str, t: float, rx: object, tx: object, uptime: object, laeuft: bool) -> tuple: + """Den neuen Zählerstand merken und die Raten gegen den vorigen liefern (unter _lock aufrufen).""" + stand = {"t": t, "rx": rx, "tx": tx, "uptime": uptime, "laeuft": laeuft} + ergebnis = raten(_zustand["zaehler"].get(geraet_id), stand) + _zustand["zaehler"][geraet_id] = stand + return ergebnis + + +def _host_punkt(host: dict, t: float) -> dict: + speicher, rootfs, netz = _dict(host.get("speicher")), _dict(host.get("rootfs")), _dict(host.get("netz")) + rx, tx = _zaehlen(HOST, t, netz.get("rx"), netz.get("tx"), host.get("uptime"), True) + temp, last = _zahl(host.get("temp_cpu")), _zahl(host.get("load1")) + return {"cpu": _prozent(host.get("cpu")), + "ram": _anteil(speicher.get("belegt"), speicher.get("gesamt")), + "ram_used": _ganz(speicher.get("belegt")), "ram_total": _ganz(speicher.get("gesamt")), + "platte": _anteil(rootfs.get("belegt"), rootfs.get("gesamt")), + "netz_rx": rx, "netz_tx": tx, + "temp_cpu": None if temp is None else round(temp, 1), + "load1": None if last is None else round(last, 2), + "uptime_s": _ganz(host.get("uptime"))} + + +def _gast_punkt(geraet_id: str, gast: dict, t: float) -> dict: + laeuft = gast.get("status") == "running" + rx, tx = _zaehlen(geraet_id, t, gast.get("netin"), gast.get("netout"), gast.get("uptime"), laeuft) + if not laeuft: + return {"cpu": None, "ram": None, "ram_used": None, "ram_total": _ganz(gast.get("maxmem")), "platte": None, + "netz_rx": None, "netz_tx": None, "uptime_s": _ganz(gast.get("uptime"))} + # Proxmox rechnet die CPU eines Gasts schon auf seine eigenen Kerne (1,0 = alle voll). Die Belegung der Platte + # kennt es nur bei Containern; bei VMs steht dort immer 0. + disk = gast.get("disk") + platte = _anteil(disk, gast.get("maxdisk")) if gast.get("art") == "lxc" and messreihen.ist_zahl(disk) and disk else None + return {"cpu": _prozent(gast.get("cpu")), + "ram": _anteil(gast.get("mem"), gast.get("maxmem")), + "ram_used": _ganz(gast.get("mem")), "ram_total": _ganz(gast.get("maxmem")), + "platte": platte, "netz_rx": rx, "netz_tx": tx, "uptime_s": _ganz(gast.get("uptime"))} + + +def annehmen(daten: dict, jetzt: float | None = None) -> int: + """Einen Minutenpunkt des Ausführers speichern. Rückgabe: für wie viele Geräte. Kaputte Einträge fallen still weg.""" + jetzt = time.time() if jetzt is None else jetzt + zeit = daten.get("zeit") + t = float(zeit) if messreihen.ist_zahl(zeit) and abs(float(zeit) - jetzt) <= ZEIT_TOLERANZ_S else jetzt + punkte: list[tuple[str, dict]] = [] + with _lock: + if isinstance(daten.get("host"), dict): + punkte.append((HOST, _host_punkt(daten["host"], t))) + gaeste = daten.get("gaeste") if isinstance(daten.get("gaeste"), list) else [] + geraete = {} + for gast in gaeste: + if not isinstance(gast, dict) or inventar.EIGENES_ETIKETT in (gast.get("etiketten") or []): + continue + if not (gid := gast_id(gast)): + continue + punkte.append((gid, _gast_punkt(gid, gast, t))) + geraete[gid] = str(gast.get("name") or gid) + if gaeste: + _zustand["geraete"] = geraete + for quelle, werte in punkte: + messreihen.schreiben(quelle, werte, t) + return len(punkte) + + +# --- Lesen ------------------------------------------------------------------------------------------ + +def _geraete(von: float, bis: float) -> list[dict]: + """Die Geräte wie im Inventar: der Host und die Gäste aus dem letzten Bericht (Namen aus apps.py), dazu Gäste, die + erst die Messwerte kennen. Ohne beides: alle Quellen mit Dateien im Zeitraum.""" + eingang = kanal.bericht() + geraete: dict[str, dict] = {} + eigene: set[str] = set() + if eingang: + geraete[HOST] = {"id": HOST, "name": "Proxmox-Host", "art": "proxmox-host"} + for gast in (eingang or {}).get("bericht", {}).get("gaeste") or []: + if not (gid := gast_id(gast)): + continue + if inventar.EIGENES_ETIKETT in (gast.get("etiketten") or []): + eigene.add(gid) + continue + app = apps.app_fuer((gast.get("app") or {}).get("kennung"), gast.get("name")) + geraete[gid] = {"id": gid, "name": app.name if app else str(gast.get("name") or gid), "art": _art(gid)} + with _lock: + gemeldet = dict(_zustand["geraete"]) + host_gemeldet = HOST in _zustand["zaehler"] + if host_gemeldet: + geraete.setdefault(HOST, {"id": HOST, "name": "Proxmox-Host", "art": "proxmox-host"}) + for gid, name in gemeldet.items(): + if gid not in geraete and gid not in eigene: + app = apps.app_fuer(None, name) + geraete[gid] = {"id": gid, "name": app.name if app else name, "art": _art(gid)} + if not geraete: + for quelle in messreihen.quellen(von, bis): + if quelle == HOST or _GAST.fullmatch(quelle): + geraete[quelle] = {"id": quelle, "name": "Proxmox-Host" if quelle == HOST else quelle, + "art": _art(quelle)} + # Reihenfolge wie im Inventar: der Host, dann die Gäste nach Nummer. + return sorted(geraete.values(), key=lambda g: (g["id"] != HOST, int(g["id"].split("-")[1]) if g["id"] != HOST else 0)) + + +def abfrage(zeitraum: str, jetzt: float | None = None) -> dict: + """Für GET /api/homelab/messwerte: {"zeitraum", "schritt_s", "geraete": [{"id", "name", "art", "reihen", + "aktuell"}]}. temp_cpu gibt es als Reihe nur beim Host; in „aktuell“ stehen temp_cpu und load1 bei Gästen auf null.""" + dauer, schritt = messreihen.zeitraum_pruefen(zeitraum) + jetzt = time.time() if jetzt is None else jetzt + geraete = [] + for geraet in _geraete(jetzt - dauer, jetzt): + host = geraet["id"] == HOST + daten = messreihen.lesen(geraet["id"], zeitraum, HOST_REIHEN if host else GAST_REIHEN, jetzt) + letzter = messreihen.letzter_punkt(geraet["id"], jetzt, AKTUELL_MAX_S) or {} + aktuell = {name: letzter.get(name) for name in AKTUELL} + if not host: + aktuell["temp_cpu"] = aktuell["load1"] = None + geraete.append({**geraet, "reihen": daten["reihen"], "aktuell": aktuell}) + return {"zeitraum": zeitraum, "schritt_s": schritt, "geraete": geraete} + + +# --- Wächter (Rolle homelab, registriert in services/waechter.py) ------------------------------------------- + +def _gb(anzahl: object) -> str: + return f"{float(anzahl) / 1e9:.1f}".replace(".", ",") if messreihen.ist_zahl(anzahl) else "?" + + +def pruefe_host(jetzt: float | None = None) -> list: + """Gelb, wenn der Proxmox-Host zehn Minuten lang über RAM_GELB % Arbeitsspeicher oder TEMP_GELB °C CPU-Temperatur + liegt. Ohne genug Messungen (Ausführer schweigt, gerade erst gestartet) kein Befund — das Schweigen meldet + pruefe_ausfuehrer().""" + from services.waechter import Befund # der Wächter registriert diese Prüfung, also erst hier + jetzt = time.time() if jetzt is None else jetzt + punkte = messreihen.punkte(HOST, jetzt - WACHE_S, jetzt + 1) + befunde = [] + ram = [p["ram"] for p in punkte if messreihen.ist_zahl(p.get("ram"))] + if len(ram) >= WACHE_MIN_PUNKTE and min(ram) > RAM_GELB: + letzter = next(p for p in reversed(punkte) if messreihen.ist_zahl(p.get("ram"))) + befunde.append(Befund( + id="host-ram", stufe="gelb", titel=f"Proxmox-Host: Arbeitsspeicher seit 10 Minuten über {RAM_GELB:.0f} %", + text=(f"Zuletzt {ram[-1]:.0f} % belegt ({_gb(letzter.get('ram_used'))} von {_gb(letzter.get('ram_total'))} " + "GB). Wird es noch enger, beendet der Kernel Prozesse, auch in den Gästen. Die Messwerte zeigen, " + "welcher Gast wie viel braucht; einem davon weniger Speicher geben oder ihn stoppen."), + quelle=HOST, sofort=True)) + temp = [p["temp_cpu"] for p in punkte if messreihen.ist_zahl(p.get("temp_cpu"))] + if len(temp) >= WACHE_MIN_PUNKTE and min(temp) > TEMP_GELB: + befunde.append(Befund( + id="host-temp", stufe="gelb", titel=f"Proxmox-Host: CPU seit 10 Minuten über {TEMP_GELB:.0f} °C", + text=(f"Zuletzt {temp[-1]:.0f} °C. So heiß drosselt sich die CPU, und auf Dauer leidet die Hardware. " + "Lüfter und Luftwege prüfen (Staub) und in den Messwerten nachsehen, welcher Gast die Last macht."), + quelle=HOST, sofort=True)) + return befunde diff --git a/backend/services/messwerte.py b/backend/services/messwerte.py new file mode 100644 index 0000000..f44376b --- /dev/null +++ b/backend/services/messwerte.py @@ -0,0 +1,184 @@ +"""Messwerte der KI-Box mit Verlauf (Monitoring, seit 24.09.2026). + +Der Ereignisstrom (/api/stream) zeigt nur den Augenblick. Für den Verlauf schreibt der Steward (Rolle box, eigener +Prozess, übersteht MC2-Neustarts) jede Minute zur halben Minute einen Punkt über kern/messreihen.py (Quelle „box“): + + cpu Mittel der Minute in % (Rechnung wie psutil.cpu_percent(interval=None), aber aus eigenen + cpu_times-Ständen: psutil merkt sich den letzten Aufruf je Thread, und die Messung läuft in wechselnden + Threads des Event-Loops) + ram, platte Belegung in % beim Messen (platte = das Modell-Laufwerk, wie im Cockpit) + gpu Mittel der Proben alle PROBE_S Sekunden in %: gpu_busy_percent zeigt nur den Augenblick, eine Probe je + Minute träfe die kurzen Antworten der Modelle kaum + temp_cpu/_gpu Mittel derselben Proben in °C + netz_rx/_tx Bytes/s über die Minute, alle Schnittstellen außer lo + tokens Prompt- plus Antwort-Tokens pro Minute (Differenz der Gesamtzähler aus services.token_stats) + +Ein Zähler, der kleiner wird (Neustart des Gateways, neue Token-Datei), ergibt für diese Minute keine Rate (null). +GET /api/messwerte (routers/messwerte.py) liefert daraus die Reihen für 1h, 24h und 7d und den letzten Punkt. +""" + +import asyncio +import logging +import os +import time +from collections.abc import Callable + +import psutil +from config import MODELS_DIR +from kern import messreihen + +from services import system, token_stats + +log = logging.getLogger(__name__) + +QUELLE = "box" +REIHEN = ("cpu", "ram", "gpu", "temp_cpu", "temp_gpu", "platte", "netz_rx", "netz_tx", "tokens") +# Ein Trocken- oder Probelauf neben dem echten Steward schreibt nicht mit (sonst stünden zwei Punkte je Minute da). +ENABLED = (os.environ.get("MC_MESSWERTE_ENABLED", "1") != "0" and os.environ.get("MC_WAECHTER_TROCKEN", "") != "1" + and os.environ.get("MC_PROBELAUF", "") != "1") +PROBE_S = float(os.environ.get("MC_MESSWERTE_PROBE_S", "5")) +PHASE_S = 30 # gemessen wird zur halben Minute: jeder 60-s-Schritt bekommt genau einen Punkt +AKTUELL_MAX_S = 180 # älter ist der letzte Punkt nicht „aktuell“ (der Steward schreibt gerade nicht) + + +# --- Rohwerte ------------------------------------------------------------------------------------ + +def _cpu_zeiten() -> tuple[float, float] | None: + """(beschäftigt, gesamt) in Sekunden seit dem Start — gezählt wie psutil.cpu_percent (guest steckt in user).""" + try: + z = psutil.cpu_times() + except Exception: + return None + gesamt = sum(z) - getattr(z, "guest", 0.0) - getattr(z, "guest_nice", 0.0) + return gesamt - z.idle - getattr(z, "iowait", 0.0), gesamt + + +def cpu_prozent(vorher: tuple[float, float] | None, jetzt: tuple[float, float] | None) -> float | None: + if not vorher or not jetzt or jetzt[1] - vorher[1] <= 0: + return None + anteil = (jetzt[0] - vorher[0]) / (jetzt[1] - vorher[1]) + return round(min(100.0, max(0.0, anteil * 100)), 1) + + +def _netz() -> tuple[int, int] | None: + """Empfangene und gesendete Bytes aller Schnittstellen außer lo (Gesamtzähler).""" + try: + zaehler = psutil.net_io_counters(pernic=True) + except Exception: + return None + return (sum(z.bytes_recv for name, z in zaehler.items() if name != "lo"), + sum(z.bytes_sent for name, z in zaehler.items() if name != "lo")) + + +def _tokens() -> int | None: + """Prompt- plus Antwort-Tokens seit Beginn der Zählung (schreibt der Gateway-Prozess).""" + try: + stand = token_stats.get_stats() + return int(stand.get("prompt_tokens") or 0) + int(stand.get("completion_tokens") or 0) + except Exception: + return None + + +def _platte() -> float | None: + try: + return round(psutil.disk_usage(str(MODELS_DIR) if MODELS_DIR.exists() else os.getcwd()).percent, 1) + except Exception: + return None + + +def _mittel(werte: list[float]) -> float | None: + return round(sum(werte) / len(werte), 1) if werte else None + + +def _runden(wert: float | None, stellen: int | None = None) -> float | int | None: + return None if wert is None else round(wert, stellen) + + +# --- Die Minute -------------------------------------------------------------------------------------- + +class Messer: + """Was zwischen zwei Minutenpunkten mitläuft: die Stände der Zähler und die Proben.""" + + def __init__(self, uhr: Callable[[], float] = time.monotonic) -> None: + self._uhr = uhr + self._seit = uhr() + self._cpu = _cpu_zeiten() + self._netz = _netz() + self._tokens = _tokens() + self._proben: dict[str, list[float]] = {"gpu": [], "temp_cpu": [], "temp_gpu": []} + + def probe(self) -> None: + """GPU-Last und Temperaturen einmal ablesen (sysfs, Bruchteile einer Millisekunde).""" + gpu = system._gpu_sysfs() or {} + temp = system._temps() or {} + for name, wert in (("gpu", gpu.get("busy_percent")), ("temp_cpu", temp.get("cpu")), + ("temp_gpu", temp.get("gpu"))): + if messreihen.ist_zahl(wert): + self._proben[name].append(float(wert)) + + def punkt(self) -> dict[str, float | int | None]: + """Der Punkt dieser Minute; danach beginnt die nächste.""" + self.probe() + jetzt = self._uhr() + dt = jetzt - self._seit + cpu, netz, tokens = _cpu_zeiten(), _netz(), _tokens() + try: + ram = round(psutil.virtual_memory().percent, 1) + except Exception: + ram = None + tok_s = messreihen.rate(self._tokens, tokens, dt) + werte = { + "cpu": cpu_prozent(self._cpu, cpu), + "ram": ram, + "gpu": _mittel(self._proben["gpu"]), + "temp_cpu": _mittel(self._proben["temp_cpu"]), + "temp_gpu": _mittel(self._proben["temp_gpu"]), + "platte": _platte(), + "netz_rx": _runden(messreihen.rate(self._netz[0], netz[0], dt)) if self._netz and netz else None, + "netz_tx": _runden(messreihen.rate(self._netz[1], netz[1], dt)) if self._netz and netz else None, + "tokens": None if tok_s is None else round(tok_s * 60, 1), + } + self._seit, self._cpu, self._netz, self._tokens = jetzt, cpu, netz, tokens + for proben in self._proben.values(): + proben.clear() + return werte + + +def naechste_messung(jetzt: float) -> float: + """Die nächste halbe Minute, mindestens eine Sekunde entfernt.""" + ziel = jetzt - jetzt % 60 + PHASE_S + return ziel if ziel > jetzt + 1 else ziel + 60 + + +async def messwerte_loop() -> None: + """Im mc2-steward (Rolle box): Proben alle PROBE_S Sekunden, ein Punkt je Minute. Fehler kosten einen Punkt, nie + die Schleife.""" + messer = await asyncio.to_thread(Messer) + naechste = naechste_messung(time.time()) + log.info("messwerte: aktiv (ein Punkt je Minute nach %s, Proben alle %ss)", messreihen.ordner(), PROBE_S) + while True: + await asyncio.sleep(max(0.2, min(PROBE_S, naechste - time.time()))) + try: + if time.time() >= naechste: + werte = await asyncio.to_thread(messer.punkt) + await asyncio.to_thread(messreihen.schreiben, QUELLE, werte) + naechste = naechste_messung(time.time()) + else: + await asyncio.to_thread(messer.probe) + except Exception: + log.warning("messwerte: Messung fehlgeschlagen", exc_info=True) + naechste = naechste_messung(time.time()) + + +# --- Lesen (MC2-Prozess) ------------------------------------------------------------------------------- + +ZEITRAEUME = messreihen.ZEITRAEUME + + +def abfrage(zeitraum: str, jetzt: float | None = None) -> dict: + """Für GET /api/messwerte: {"zeitraum", "schritt_s", "reihen": {…}, "aktuell": {…}}. aktuell ist der letzte + Minutenpunkt, solange er höchstens AKTUELL_MAX_S alt ist, sonst lauter null.""" + jetzt = time.time() if jetzt is None else jetzt + daten = messreihen.lesen(QUELLE, zeitraum, REIHEN, jetzt) + letzter = messreihen.letzter_punkt(QUELLE, jetzt, AKTUELL_MAX_S) or {} + return {**daten, "aktuell": {name: letzter.get(name) for name in REIHEN}} diff --git a/backend/services/waechter.py b/backend/services/waechter.py index 9a5de49..b85264c 100644 --- a/backend/services/waechter.py +++ b/backend/services/waechter.py @@ -597,6 +597,8 @@ PRUEFUNGEN = { "homelab": (pruefe_platte, pruefe_partner, pruefe_ausfuehrer, pruefe_gaeste, pruefe_gast_platten), }[ROLLE] PRUEFUNGEN += (__import__("services.homelab.pflege").homelab.pflege.pruefe_paketlisten,) if ROLLE == "homelab" else () +# Messwerte des Proxmox-Hosts (seit 24.09.2026): zehn Minuten über 95 % RAM oder 90 °C = gelb. +PRUEFUNGEN += (__import__("services.homelab.messwerte").homelab.messwerte.pruefe_host,) if ROLLE == "homelab" else () BETREFF_PROBLEM, BETREFF_OK = {"box": ("[Box-Problem]", "[Box wieder ok]"), "homelab": ("[Homelab-Problem]", "[Homelab wieder ok]")}[ROLLE] diff --git a/backend/steward.py b/backend/steward.py index 5e31767..14968c3 100644 --- a/backend/steward.py +++ b/backend/steward.py @@ -12,6 +12,8 @@ Mini-Dienst (mc2-steward.service, Restart=always): • waechter.waechter_loop — Box-Wart-Wächter (seit 09/2026, löst sentry ab): Dienste, Timer, Hermes-Jobs, Kern, Platte → Hinweise + Selbstreparatur; beobachtet im Steward-Modus AUCH MC2 selbst + mc2-gateway + • messwerte.messwerte_loop — Messwerte der KI-Box mit Verlauf (seit 24.09.2026): ein Punkt + je Minute nach /mc2-messwerte/ (nur Rolle box) Das dritte Loop (Gedächtnis-Dubletten gegen den Sidecar auf :8765) ist mit dessen Ablösung am 07.08.2026 entfallen — Hermes führt sein Gedächtnis selbst (docs/wissen/VERDIKTE.md). @@ -32,7 +34,7 @@ import os from config import CONFIG_PATH from kern.einstellungen import einstellungen -from services import waechter, warmer +from services import messwerte, waechter, warmer logging.basicConfig( level=os.environ.get("MC_LOG_LEVEL", "INFO").upper(), @@ -70,6 +72,9 @@ async def main() -> None: warmer.INTERVAL, CONFIG_WATCH_S) if waechter.ENABLED: tasks.append(asyncio.create_task(waechter.waechter_loop())) + # Messwerte mit Verlauf (Logik in services/messwerte.py); das Homelab bekommt seine vom Ausführer. + if einstellungen().rolle == "box" and messwerte.ENABLED: + tasks.append(asyncio.create_task(messwerte.messwerte_loop())) if not tasks: log.warning("steward: alle Loops per Env deaktiviert — nichts zu tun, Ende.") return diff --git a/backend/tests/test_messwerte.py b/backend/tests/test_messwerte.py new file mode 100644 index 0000000..5bca3aa --- /dev/null +++ b/backend/tests/test_messwerte.py @@ -0,0 +1,394 @@ +"""Messwerte mit Verlauf (Monitoring, 24.09.2026): der gemeinsame Speicher (kern/messreihen.py), der Minutenpunkt der +KI-Box, die Raten und Endpunkte des Homelabs, die reine Auswertung des Ausführers (mit echten Zeilen vom Proxmox-PC, +gelesen am 24.09.) und die Wächter-Prüfung des Hosts.""" + +import importlib.util +import json +import time +from datetime import datetime +from pathlib import Path +from types import SimpleNamespace + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient +from kern import einstellungen as einstellungen_mod +from kern import messreihen +from kern.zeit import LOCAL_TZ + +BERICHT = json.loads((Path(__file__).parent / "fixtures" / "pve-bericht.json").read_text(encoding="utf-8")) +AUSFUEHRER = Path(__file__).resolve().parents[2] / "deploy" / "homelab" / "ausfuehrer.py" +MITTAG = datetime(2026, 9, 24, 12, 0, 10, tzinfo=LOCAL_TZ).timestamp() # Do 24.09.2026, 12:00:10 Berlin + + +@pytest.fixture +def daten(tmp_path, monkeypatch): + monkeypatch.setenv("MC_DATEN_DIR", str(tmp_path)) + monkeypatch.delenv("MC_AUSFUEHRER_TOKEN", raising=False) + einstellungen_mod.einstellungen.cache_clear() + from services.homelab import kanal + from services.homelab import messwerte as homelab_messwerte + monkeypatch.setattr(kanal, "_kontakt", {"zuletzt": None}) + monkeypatch.setattr(homelab_messwerte, "_zustand", {"zaehler": {}, "geraete": {}}) + yield tmp_path + einstellungen_mod.einstellungen.cache_clear() + + +def _werte(reihe: list) -> list: + return [v for _, v in reihe if v is not None] + + +# --- Der gemeinsame Speicher ----------------------------------------------------------------------- + +def test_stunde_im_minutenraster_mit_luecke(daten): + for minute, cpu in ((-5, 10), (-4, 20.0), (-2, 40), (-1, 50)): # 11:57 fehlt + messreihen.schreiben("box", {"cpu": cpu, "ram": 30}, t=MITTAG - 10 + minute * 60 + 30) + stunde = messreihen.lesen("box", "1h", ("cpu", "ram"), jetzt=MITTAG) + cpu = stunde["reihen"]["cpu"] + assert stunde["schritt_s"] == 60 and len(cpu) == 60 and len(stunde["reihen"]["ram"]) == 60 + assert all(t % 60 == 0 for t, _ in cpu) and cpu[-1][0] == MITTAG - 10 # der laufende Schritt ist der letzte + assert [v for _, v in cpu[-6:]] == [10, 20.0, None, 40, 50, None] # Lücke bleibt Lücke, nichts erfunden + assert _werte(stunde["reihen"]["ram"]) == [30, 30, 30, 30] + + +def test_verdichtung_auf_fuenf_und_dreissig_minuten(daten): + beginn = MITTAG - 10 - 3600 # 11:00:00 + for minute in range(60): + messreihen.schreiben("box", {"cpu": minute}, t=beginn + minute * 60 + 30) + tag = messreihen.lesen("box", "24h", ("cpu",), jetzt=MITTAG) + assert tag["schritt_s"] == 300 and len(tag["reihen"]["cpu"]) == 288 + assert _werte(tag["reihen"]["cpu"]) == [2, 7, 12, 17, 22, 27, 32, 37, 42, 47, 52, 57] # Mittel je 5 Minuten + woche = messreihen.lesen("box", "7d", ("cpu",), jetzt=MITTAG) + assert woche["schritt_s"] == 1800 and len(woche["reihen"]["cpu"]) == 336 + assert _werte(woche["reihen"]["cpu"]) == [14.5, 44.5] + # Ein zweites Lesen kommt aus den gemerkten Summen und ist gleich; ein neuer Punkt ändert die Datei und zählt mit. + assert messreihen.lesen("box", "7d", ("cpu",), jetzt=MITTAG) == woche + messreihen.schreiben("box", {"cpu": 100}, t=MITTAG) + assert _werte(messreihen.lesen("box", "7d", ("cpu",), jetzt=MITTAG)["reihen"]["cpu"]) == [14.5, 44.5, 100] + with pytest.raises(ValueError): + messreihen.lesen("box", "2d", ("cpu",), jetzt=MITTAG) + + +def test_kaputte_zeilen_werden_uebersprungen(daten): + messreihen.schreiben("pve", {"cpu": 1.0}, t=MITTAG - 200) + datei = messreihen.ordner() / "pve-2026-09-24.jsonl" + with open(datei, "ab") as f: + f.write(b'kein json\n[1, 2]\n{"cpu": 5}\n{"t": "gestern", "cpu": 5}\n\x00\x00\n') + f.write(b'{"t": %d, "cpu": NaN, "ram": true}\n' % int(MITTAG - 140)) + f.write(b'{"t": %d, "cpu": 9' % int(MITTAG - 100)) # Absturz mitten in der Zeile + messreihen.schreiben("pve", {"cpu": 3.0, "ram": True, "platte": float("inf")}, t=MITTAG - 40) + punkte = messreihen.punkte("pve", MITTAG - 300, MITTAG) + assert [p["t"] for p in punkte] == [int(MITTAG - 200), int(MITTAG - 140), int(MITTAG - 40)] + assert punkte[-1] == {"t": int(MITTAG - 40), "cpu": 3.0, "ram": None, "platte": None} + assert _werte(messreihen.lesen("pve", "1h", ("cpu", "ram"), jetzt=MITTAG)["reihen"]["cpu"]) == [1.0, 3.0] + assert messreihen.letzter_punkt("pve", jetzt=MITTAG)["cpu"] == 3.0 + + +def test_aufraeumen_nach_acht_tagen(daten): + ordner = messreihen.ordner() + ordner.mkdir(parents=True) + for name in ("box-2026-09-14.jsonl", "box-2026-09-15.jsonl", "box-2026-09-16.jsonl", "ct-100-2026-09-15.jsonl", + "ct-100-2026-09-23.jsonl", "notiz.txt", "zustand.json"): + (ordner / name).write_text("", encoding="utf-8") + messreihen.schreiben("box", {"cpu": 1}, t=MITTAG) # das erste Schreiben des Tages räumt auf + assert sorted(p.name for p in ordner.iterdir()) == [ + "box-2026-09-16.jsonl", "box-2026-09-24.jsonl", "ct-100-2026-09-23.jsonl", "notiz.txt", "zustand.json"] + assert messreihen.aufraeumen(MITTAG) == 0 + + +def test_letzter_punkt_auch_von_gestern_und_nie_zu_alt(daten): + mitternacht = datetime(2026, 9, 24, 0, 0, 20, tzinfo=LOCAL_TZ).timestamp() + messreihen.schreiben("box", {"cpu": 7}, t=mitternacht - 50) # 23.09., 23:59:30 + assert messreihen.letzter_punkt("box", jetzt=mitternacht)["cpu"] == 7 + assert messreihen.letzter_punkt("box", jetzt=mitternacht, max_alter_s=30) is None + assert messreihen.letzter_punkt("ct-999", jetzt=mitternacht) is None + assert messreihen.quellen(mitternacht - 3600, mitternacht) == ["box"] + + +def test_quellennamen_werden_geprueft(daten): + for boese in ("../box", "Box", "", "box/x"): + with pytest.raises(ValueError): + messreihen.schreiben(boese, {"cpu": 1}, t=MITTAG) + + +def test_rate_aus_zaehlerstaenden(): + assert messreihen.rate(1000, 7000, 60) == 100.0 + assert messreihen.rate(7000, 1000, 60) is None # Zähler kleiner: Neustart, keine negative Rate + assert messreihen.rate(1000, 7000, 301) is None # Lücke: keine Rate über fünf Minuten + assert messreihen.rate(None, 7000, 60) is None and messreihen.rate(1000, 7000, 0) is None + assert messreihen.rate(0, 1e12, 60) is None # 16 GB/s: ein Sprung, kein Verkehr + + +# --- KI-Box -------------------------------------------------------------------------------------- + +def test_minutenpunkt_der_box(monkeypatch): + from services import messwerte, system + folge = {"cpu": [(1000.0, 10000.0), (1300.0, 10600.0), (1400.0, 10700.0)], + "netz": [(1_000_000, 500_000), (1_600_000, 800_000), (100, 50)], + "tokens": [1000, 4000, 10], "gpu": [10, 30, 50, 0], "uhr": [0.0, 60.0, 120.0]} + monkeypatch.setattr(messwerte, "_cpu_zeiten", lambda: folge["cpu"].pop(0)) + monkeypatch.setattr(messwerte, "_netz", lambda: folge["netz"].pop(0)) + monkeypatch.setattr(messwerte, "_tokens", lambda: folge["tokens"].pop(0)) + monkeypatch.setattr(messwerte, "_platte", lambda: 15.2) + monkeypatch.setattr(system, "_gpu_sysfs", lambda: {"busy_percent": folge["gpu"].pop(0)}) + monkeypatch.setattr(system, "_temps", lambda: {"cpu": 40.0, "gpu": 35.5}) + monkeypatch.setattr(messwerte.psutil, "virtual_memory", lambda: SimpleNamespace(percent=41.26)) + messer = messwerte.Messer(uhr=lambda: folge["uhr"].pop(0)) + messer.probe() + messer.probe() + # CPU: 300 von 600 Ticks beschäftigt; GPU: Mittel aus drei Proben; Netz und Tokens aus der Differenz über 60 s. + assert messer.punkt() == {"cpu": 50.0, "ram": 41.3, "gpu": 30.0, "temp_cpu": 40.0, "temp_gpu": 35.5, + "platte": 15.2, "netz_rx": 10000, "netz_tx": 5000, "tokens": 3000.0} + # Neustart des Gateways (neue Token-Zählung) und neue Netz-Zähler: diese Minute keine Rate statt einer negativen. + zweiter = messer.punkt() + assert (zweiter["netz_rx"], zweiter["netz_tx"], zweiter["tokens"], zweiter["gpu"]) == (None, None, None, 0.0) + assert zweiter["cpu"] == 100.0 + + +def test_gemessen_wird_zur_halben_minute(): + from services import messwerte + assert messwerte.naechste_messung(MITTAG) == MITTAG + 20 # 12:00:10 → 12:00:30 + assert messwerte.naechste_messung(MITTAG + 20.5) == MITTAG + 80 # eben gemessen → 12:01:30 + assert messwerte.naechste_messung(MITTAG + 25) == MITTAG + 80 # 12:00:35 → 12:01:30 + + +def test_endpunkt_der_box(daten): + from routers import messwerte as messwerte_router + from services import messwerte + app = FastAPI() + app.include_router(messwerte_router.router) + client = TestClient(app) + jetzt = time.time() + punkt = {"cpu": 12.5, "ram": 40.0, "gpu": 3.0, "temp_cpu": 45.0, "temp_gpu": 40.0, "platte": 15.0, + "netz_rx": 1200, "netz_tx": 300, "tokens": 900.0} + messreihen.schreiben("box", punkt, t=jetzt - 30) + antwort = client.get("/api/messwerte", params={"zeitraum": "24h"}).json() + assert (antwort["zeitraum"], antwort["schritt_s"]) == ("24h", 300) + assert list(antwort["reihen"]) == list(messwerte.REIHEN) + assert all(len(reihe) == 288 for reihe in antwort["reihen"].values()) + assert _werte(antwort["reihen"]["tokens"]) == [900] and _werte(antwort["reihen"]["cpu"]) == [12.5] + assert antwort["aktuell"] == punkt + stunde = client.get("/api/messwerte").json() # ohne Angabe: die letzte Stunde + assert stunde["schritt_s"] == 60 and len(stunde["reihen"]["cpu"]) == 60 + assert client.get("/api/messwerte", params={"zeitraum": "1w"}).status_code == 400 + # Schreibt der Steward nicht mehr, ist nichts „aktuell“. + assert set(messwerte.abfrage("1h", jetzt=jetzt + 600)["aktuell"].values()) == {None} + + +def test_aufraeumen_bietet_den_messwerte_ordner_nie_an(tmp_path, monkeypatch): + from services import aufraeumen, llamaswap + (tmp_path / "mc2-messwerte").mkdir() + (tmp_path / "Alt-GGUF").mkdir() + monkeypatch.setattr(aufraeumen, "MODELS_DIR", tmp_path) + monkeypatch.setattr(llamaswap, "read_config", lambda: {"models": {}}) + assert [k["name"] for k in aufraeumen.kandidaten()] == ["Alt-GGUF"] + + +# --- Homelab: Raten, Annahme, Endpunkte --------------------------------------------------------------- + +def test_netz_raten_mit_zaehlersprung_und_neustart(): + from services.homelab import messwerte as hm + a = {"t": 0, "rx": 1000, "tx": 2000, "uptime": 500, "laeuft": True} + b = {"t": 60, "rx": 61000, "tx": 2600, "uptime": 560, "laeuft": True} + assert hm.raten(a, b) == (1000, 10) + assert hm.raten(None, b) == (None, None) # kein voriger Stand (dieser Teil neu gestartet) + c = {"t": 120, "rx": 500, "tx": 100, "uptime": 30, "laeuft": True} + assert hm.raten(b, c) == (None, None) # Gast neu gestartet: Laufzeit kleiner + d = {"t": 180, "rx": 400, "tx": 99999, "uptime": 90, "laeuft": True} + assert hm.raten(c, d) == (None, 1665) # nur rx kleiner geworden: nur dort keine Rate + assert hm.raten(d, {**d, "t": 600, "rx": 1000}) == (None, None) # Lücke über fünf Minuten + aus = {"t": 240, "rx": 0, "tx": 0, "uptime": 0, "laeuft": False} + assert hm.raten(d, aus) == (None, None) and hm.raten(aus, {**d, "t": 300}) == (None, None) + + +def _gast(vmid, art, name, etiketten, status="running", **werte): + gast = {"vmid": vmid, "art": art, "name": name, "etiketten": etiketten, "status": status, "cpu": 0.0, "maxcpu": 1, + "mem": 0, "maxmem": 536870912, "disk": 0, "maxdisk": 4143677440, "netin": 0, "netout": 0, "uptime": 0} + return {**gast, **werte} + + +def _minute(zeit: float, host_rx: int, gitea_netin: int) -> dict: + return {"zeit": zeit, + "host": {"cpu": 0.125, "speicher": {"belegt": 12_000_000_000, "gesamt": 32_000_000_000}, + "rootfs": {"belegt": 48e9, "gesamt": 100e9}, "load1": 0.72, "uptime": 3379388, "temp_cpu": 50.5, + "netz": {"rx": host_rx, "tx": 1000, "schnittstellen": ["eno1"]}}, + "gaeste": [ + _gast(104, "lxc", "gitea", ["community-script", "git"], cpu=0.0183, mem=340934656, + maxmem=1073741824, disk=4914618368, maxdisk=8350298112, netin=gitea_netin, netout=5000, + uptime=1000), + _gast(106, "qemu", "arcane", [], cpu=0.25, maxcpu=4, mem=5325914112, maxmem=8589934592, + maxdisk=161061273600, netin=10, netout=10, uptime=99), + _gast(107, "lxc", "homelab-orchestrator", ["mc2"], cpu=0.5, mem=1, maxmem=2), + _gast(102, "lxc", "netbird", ["community-script"], status="stopped"), + ]} + + +def _homelab_client() -> TestClient: + from routers import homelab + app = FastAPI() + app.include_router(homelab.router) + return TestClient(app) + + +def test_homelab_endpunkte(daten): + from services.homelab import kanal + client = _homelab_client() + jetzt = time.time() + assert client.post("/api/homelab/ausfuehrer/messwerte", json=_minute(jetzt, 0, 0)).status_code == 403 + kopf = {"X-MC2-Ausfuehrer": kanal.token()} + erste = client.post("/api/homelab/ausfuehrer/messwerte", json=_minute(jetzt - 60, 5_000_000, 1_000_000), + headers=kopf) + assert erste.json() == {"ok": True, "geraete": 4} # Host und drei Gäste; der eigene Container fehlt + client.post("/api/homelab/ausfuehrer/messwerte", json=_minute(jetzt, 5_600_000, 1_600_000), headers=kopf) + assert not list(messreihen.ordner().glob("ct-107-*")) + + # Noch ohne Bericht: die Geräte der letzten Meldung, Namen soweit bekannt. + ohne = client.get("/api/homelab/messwerte").json() + assert [(g["id"], g["name"], g["art"]) for g in ohne["geraete"]] == [ + ("pve", "Proxmox-Host", "proxmox-host"), ("ct-102", "netbird", "container"), ("ct-104", "gitea", "container"), + ("vm-106", "Arcane (Docker)", "vm")] + + kanal.bericht_speichern(BERICHT) # mit Bericht: Geräte und Namen wie im Inventar + antwort = client.get("/api/homelab/messwerte", params={"zeitraum": "1h"}).json() + assert (antwort["zeitraum"], antwort["schritt_s"]) == ("1h", 60) + geraete = {g["id"]: g for g in antwort["geraete"]} + assert list(geraete) == ["pve", "ct-100", "ct-101", "ct-102", "ct-103", "ct-104", "ct-105", "vm-106"] + assert (geraete["ct-104"]["name"], geraete["ct-100"]["name"], geraete["vm-106"]["name"]) == ( + "Gitea", "AdGuard Home", "Arcane (Docker)") + + host = geraete["pve"] + assert list(host["reihen"]) == ["cpu", "ram", "platte", "netz_rx", "netz_tx", "temp_cpu"] + assert all(len(reihe) == 60 for reihe in host["reihen"].values()) + assert _werte(host["reihen"]["netz_rx"]) == [10000] # erste Minute ohne Vorgänger: keine Rate + assert host["aktuell"] == {"cpu": 12.5, "ram": 37.5, "ram_used": 12_000_000_000, "ram_total": 32_000_000_000, + "platte": 48.0, "netz_rx": 10000, "netz_tx": 0, "temp_cpu": 50.5, "load1": 0.72, + "uptime_s": 3379388} + gitea = geraete["ct-104"] + assert list(gitea["reihen"]) == ["cpu", "ram", "platte", "netz_rx", "netz_tx"] + assert gitea["aktuell"] == {"cpu": 1.8, "ram": 31.8, "ram_used": 340934656, "ram_total": 1073741824, + "platte": 58.9, "netz_rx": 10000, "netz_tx": 0, "temp_cpu": None, "load1": None, + "uptime_s": 1000} + arcane = geraete["vm-106"]["aktuell"] + assert (arcane["cpu"], arcane["ram"], arcane["platte"]) == (25.0, 62.0, None) # die Platte einer VM sieht Proxmox nicht + netbird = geraete["ct-102"]["aktuell"] # gestoppt: keine Messwerte, keine Nullen + assert (netbird["cpu"], netbird["ram"], netbird["netz_rx"], netbird["ram_total"]) == (None, None, None, 536870912) + assert set(geraete["ct-100"]["aktuell"].values()) == {None} # nichts gemessen + assert client.get("/api/homelab/messwerte", params={"zeitraum": "7d"}).json()["schritt_s"] == 1800 + assert client.get("/api/homelab/messwerte", params={"zeitraum": "30d"}).status_code == 400 + + +# --- Der Ausführer (läuft auf dem Proxmox-Host) ----------------------------------------------------------- + +NETDEV = """Inter-| Receive | Transmit + face |bytes packets errs drop fifo frame compressed multicast|bytes packets errs drop fifo colls carrier compressed + lo: 12220307 51502 0 0 0 0 0 0 12220307 51502 0 0 0 0 0 0 + eno1: 848364497291 629881217 0 114788 0 0 0 3360103 211535843861 102116072 0 4 0 0 0 0 +enp2s0: 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 + vmbr0: 736642320585 66694945 0 3399405 0 0 0 2462406 212959884634 52971342 0 5 0 0 0 0 +veth100i0: 963620053 5314692 0 0 0 0 0 0 5535391141 14176389 0 0 0 0 0 0 +veth104i0: 205263214515 35385337 0 0 0 0 0 0 6152585548 42736082 0 0 0 0 0 0 +tap106i0: 10110528132 6394361 0 0 0 0 0 0 35706903774 10586880 0 0 0 0 0 0 +fwpr106p0: 10110528132 6394361 0 0 0 0 0 0 35706903840 10586881 0 0 0 0 0 0 +fwln106i0: 35706903840 10586881 0 0 0 0 0 0 10110528132 6394361 0 0 0 0 0 0 +""" +MEMINFO = "MemTotal: 32225256 kB\nMemFree: 13537532 kB\nMemAvailable: 20092076 kB\nBuffers: 1 kB\n" +RESSOURCEN = [ + {"id": "lxc/100", "type": "lxc", "vmid": 100, "name": "adguard", "node": "pve", "status": "running", + "tags": "adblock;community-script", "template": 0, "cpu": 0.000218199761345051, "maxcpu": 1, "mem": 247558144, + "maxmem": 536870912, "disk": 1361436672, "maxdisk": 2040373248, "netin": 5535402482, "netout": 963621401, + "uptime": 3379433}, + {"id": "qemu/106", "type": "qemu", "vmid": 106, "name": "arcane", "node": "pve", "status": "running", "tags": None, + "template": 0, "cpu": 0.0194595849334057, "maxcpu": 4, "mem": 5325914112, "maxmem": 8589934592, "disk": 0, + "maxdisk": 161061273600, "netin": 35706938260, "netout": 10111961580, "uptime": 1726847}, + {"id": "lxc/102", "type": "lxc", "vmid": 102, "name": "netbird", "node": "pve", "status": "stopped", + "tags": "community-script", "template": 0, "cpu": 0, "maxcpu": 1, "mem": 0, "maxmem": 536870912, "disk": 0, + "maxdisk": 4143677440, "netin": 0, "netout": 0, "uptime": 0}, + {"id": "lxc/900", "type": "lxc", "vmid": 900, "name": "vorlage", "node": "pve", "template": 1}, + {"id": "lxc/300", "type": "lxc", "vmid": 300, "name": "fremd", "node": "pve2", "status": "running"}, +] +FUEHLER = [("r8169_0_100:00", "", 40.0), ("nvme", "Composite", 40.85), ("k10temp", "Tctl", 50.5), + ("amdgpu", "edge", 38.0)] + + +def _ausfuehrer_modul(): + spec = importlib.util.spec_from_file_location("ausfuehrer_messwerte", AUSFUEHRER) + modul = importlib.util.module_from_spec(spec) + spec.loader.exec_module(modul) + return modul + + +def _roh(zeit: float, netdev: str = NETDEV) -> dict: + return {"zeit": zeit, "stat": "cpu 1000 0 500 8000 500 0 0 0 0 0\ncpu0 500 0 250 4000 250 0 0 0 0 0\n", + "meminfo": MEMINFO, "loadavg": "0.72 0.50 0.49 3/665 2945458\n", "uptime": "3379388.63 39689887.90\n", + "netdev": netdev, "rootfs": (100861726720, 52717445120), "physisch": {"eno1", "enp2s0", "wlp3s0"}, + "fuehler": FUEHLER, "ressourcen": RESSOURCEN} + + +def test_ausfuehrer_baut_den_messpunkt_aus_echten_zeilen(): + a = _ausfuehrer_modul() + vorher = a.cpu_zeiten("cpu 900 0 400 7500 480 0 0 0 0 0\n") + punkt = a.messpunkt(_roh(1790000000.0), vorher, node="pve") + # Speicher und rootfs wie pvesh /nodes/pve/status am 24.09.2026 (used 12424376320 bzw. 48144281600). + assert punkt["host"] == {"cpu": 0.2778, "speicher": {"belegt": 12424376320, "gesamt": 32998662144}, + "rootfs": {"belegt": 48144281600, "gesamt": 100861726720}, "load1": 0.72, + "uptime": 3379388, "temp_cpu": 50.5, + "netz": {"rx": 848364497291, "tx": 211535843861, "schnittstellen": ["eno1", "enp2s0"]}} + gaeste = {g["vmid"]: g for g in punkt["gaeste"]} + assert list(gaeste) == [100, 102, 106] # keine Vorlage, kein fremder Knoten + # Netz aus veth/tap, zeitgleich mit dem Host gelesen (Richtung aus Sicht des Gasts), nicht aus fwpr/fwln. + assert (gaeste[100]["netin"], gaeste[100]["netout"]) == (5535391141, 963620053) + assert (gaeste[106]["netin"], gaeste[106]["netout"]) == (35706903774, 10110528132) + assert (gaeste[102]["netin"], gaeste[102]["status"]) == (0, "stopped") # ohne Schnittstelle: /cluster/resources + assert gaeste[100]["etiketten"] == ["adblock", "community-script"] and gaeste[106]["art"] == "qemu" + assert (gaeste[106]["cpu"], gaeste[106]["maxcpu"], gaeste[106]["maxdisk"]) == (0.0194595849334057, 4, 161061273600) + # Ohne vorigen Stand keine CPU, ohne pvesh keine Gäste — der Host kommt trotzdem. + leer = a.messpunkt({**_roh(1.0), "ressourcen": None, "rootfs": None, "fuehler": []}, None, node="pve") + assert leer["gaeste"] == [] and leer["host"]["cpu"] is None and leer["host"]["temp_cpu"] is None + assert leer["host"]["rootfs"] is None and leer["host"]["load1"] == 0.72 + + +def test_ausfuehrer_cpu_temperatur_und_netz_ohne_physische_schnittstelle(): + a = _ausfuehrer_modul() + assert a.cpu_temperatur([("k10temp", "Tctl", 60.0), ("k10temp", "Tdie", 55.0)]) == 55.0 + assert a.cpu_temperatur([("coretemp", "Core 0", 50.0), ("coretemp", "Package id 0", 58.3)]) == 58.3 + assert a.cpu_temperatur([("nvme", "Composite", 40.0), ("amdgpu", "edge", 38.0)]) is None + zaehler = a.netz_zaehler(NETDEV) + assert a.host_netz(zaehler, set())["schnittstellen"] == ["vmbr0"] + assert a.host_netz({"lo": (1, 1)}, set()) is None + assert a.cpu_anteil((100, 1000), (100, 1000)) is None and a.speicher_aus("") is None + + +def test_ausfuehrer_und_homelab_teil_verstehen_sich(daten): + """Der Messpunkt des Ausführers geht so, wie er ist, durch annehmen(): Namen der Felder und Richtung der Zähler.""" + from services.homelab import messwerte as hm + a = _ausfuehrer_modul() + t0 = MITTAG + 20 + erster = a.messpunkt(_roh(t0 - 60), None, node="pve") + netdev = NETDEV.replace("848364497291", "848370497291").replace("5535391141", "5535991141") + zweiter = a.messpunkt(_roh(t0, netdev), None, node="pve") + assert hm.annehmen(erster, jetzt=t0) == 4 and hm.annehmen(zweiter, jetzt=t0) == 4 + host = messreihen.letzter_punkt("pve", jetzt=t0) + # 6 MB mehr auf eno1 in 60 s; RAM 12,42 von 33,0 GB wie im Proxmox-Knoten. + assert (host["netz_rx"], host["netz_tx"], host["ram"], host["platte"], host["temp_cpu"]) == (100000, 0, 37.7, 47.7, + 50.5) + adguard = messreihen.letzter_punkt("ct-100", jetzt=t0) + assert (adguard["netz_rx"], adguard["netz_tx"], adguard["cpu"], adguard["platte"]) == (10000, 0, 0.0, 66.7) + assert messreihen.letzter_punkt("vm-106", jetzt=t0)["platte"] is None + + +# --- Wächter (Rolle homelab) --------------------------------------------------------------------------- + +def test_waechter_meldet_vollen_oder_heissen_host(daten): + from services.homelab import messwerte as hm + for i in range(10): + messreihen.schreiben("pve", {"ram": 96.0 + i / 10, "ram_used": 31e9, "ram_total": 32e9, "temp_cpu": 91.0}, + t=MITTAG - 590 + i * 60) + befunde = {b.id: b for b in hm.pruefe_host(MITTAG)} + assert set(befunde) == {"host-ram", "host-temp"} and {b.stufe for b in befunde.values()} == {"gelb"} + assert befunde["host-ram"].titel == "Proxmox-Host: Arbeitsspeicher seit 10 Minuten über 95 %" + assert "Zuletzt 97 % belegt (31,0 von 32,0 GB)" in befunde["host-ram"].text + assert befunde["host-temp"].titel == "Proxmox-Host: CPU seit 10 Minuten über 90 °C" + assert hm.pruefe_host(MITTAG + 400) == [] # nur drei Minuten im Fenster: zu wenig + messreihen.schreiben("pve", {"ram": 90.0, "temp_cpu": 80.0}, t=MITTAG - 30) + assert hm.pruefe_host(MITTAG) == [] # eine Minute darunter: nicht „10 Minuten lang“ + # Registriert ist die Prüfung nur in der Rolle homelab: test_partner.py prüft die Liste in einem eigenen Prozess. diff --git a/backend/tests/test_partner.py b/backend/tests/test_partner.py index aea126c..fa70a61 100644 --- a/backend/tests/test_partner.py +++ b/backend/tests/test_partner.py @@ -220,7 +220,7 @@ from services import waechter print(json.dumps({"pruefungen": [p.__name__ for p in waechter.PRUEFUNGEN], "daten": str(waechter.DATEN_DIR)})) """) assert ergebnis["pruefungen"] == ["pruefe_platte", "pruefe_partner", "pruefe_ausfuehrer", "pruefe_gaeste", - "pruefe_gast_platten", "pruefe_paketlisten"] + "pruefe_gast_platten", "pruefe_paketlisten", "pruefe_host"] def test_homelab_rolle_hat_die_homelab_schnittstellen(tmp_path): diff --git a/deploy/homelab/ausfuehrer.py b/deploy/homelab/ausfuehrer.py index 4b9f8a4..8046342 100644 --- a/deploy/homelab/ausfuehrer.py +++ b/deploy/homelab/ausfuehrer.py @@ -16,6 +16,9 @@ Container des Homelab-Teils braucht keinen Proxmox-Schlüssel. nie auf einen Speicher der Art pbs, sonst sicherte sich der PBS selbst. Nur eigene Sicherungen (Notiz „mc2-sicherung“) werden zurückgespielt oder gelöscht. An der Speicher-Konfiguration ändert der Ausführer nichts. + • Messwerte jede Minute (seit 24.09.2026) in einem eigenen Faden, unabhängig vom Bericht und von + Aufträgen: Host aus /proc und /sys, Gäste aus einem einzigen pvesh-Aufruf. Nur lesend; + Fehler bleiben still. Nur die Standardbibliothek (Debian-Python des Hosts). Konfiguration: /etc/mc2-ausfuehrer.json {"server": "http://192.168.178.x:9001", "token": "", "node": "pve", "nur_lesen": false} @@ -25,9 +28,11 @@ der Sicherungen annimmt). ausfuehrer-einrichten.sh schreibt die Datei neu; den S Aufruf: ausfuehrer.py Dauerbetrieb (systemd) ausfuehrer.py --bericht Bericht einmal auf die Konsole (nur lesend, zum Prüfen) + ausfuehrer.py --messwerte einen Messpunkt auf die Konsole (nur lesend, CPU über eine Sekunde) """ import argparse +import glob import json import logging import os @@ -35,6 +40,7 @@ import platform import re import subprocess import sys +import threading import time import urllib.error import urllib.request @@ -602,6 +608,214 @@ def ausfuehren(auftrag: dict) -> dict: return {"code": code, "text": text[-20000:]} +# --- Messwerte: jede Minute, eigener Faden (reine Auswertungen ohne Host testbar) ---------------- + +MESSWERTE_ALLE_S = 60 +MESS_PHASE_S = 30 # gemessen wird zur halben Minute (wie auf der KI-Box): genau ein Punkt je Minutenschritt +# CPU-Fühler in hwmon, in dieser Reihenfolge. Auf dem Proxmox-PC (Ryzen 5 5500U) gibt es nur k10temp mit „Tctl“; +# /sys/class/thermal und lm-sensors fehlen dort (nachgesehen am 24.09.2026). +CPU_FUEHLER = ("k10temp", "zenpower", "coretemp", "cpu_thermal") +CPU_BESCHRIFTUNG = ("Tdie", "Tctl", "Package id 0") +_GAST_SCHNITTSTELLE = re.compile(r"(?:veth|tap)(\d+)i\d+") + + +def _erste_zahl(text: str | None) -> float | None: + try: + return float(str(text or "").split()[0]) + except (IndexError, ValueError): + return None + + +def cpu_zeiten(stat: str) -> tuple[int, int] | None: + """(beschäftigt, gesamt) in Ticks aus der Zeile „cpu“ von /proc/stat: user nice system idle iowait irq softirq + steal (guest steckt schon in user); beschäftigt ist alles außer idle und iowait.""" + for zeile in stat.splitlines(): + if zeile.startswith("cpu "): + try: + werte = [int(x) for x in zeile.split()[1:9]] + except ValueError: + return None + werte += [0] * (8 - len(werte)) + gesamt = sum(werte) + return gesamt - werte[3] - werte[4], gesamt + return None + + +def cpu_anteil(vorher: tuple[int, int] | None, jetzt: tuple[int, int] | None) -> float | None: + """Wie viel die CPU zwischen zwei Ständen von /proc/stat arbeitete (0–1) — das Mittel der Minute. pvesh kann das + nicht: /nodes//status zeigt in einem frischen pvesh-Prozess immer cpu 0 (24.09.2026 nachgesehen).""" + if not vorher or not jetzt or jetzt[1] <= vorher[1]: + return None + return round(min(1.0, max(0.0, (jetzt[0] - vorher[0]) / (jetzt[1] - vorher[1]))), 4) + + +def speicher_aus(meminfo: str) -> dict | None: + """Arbeitsspeicher in Bytes; belegt = MemTotal − MemAvailable, wie Proxmox rechnet.""" + werte = {} + for zeile in meminfo.splitlines(): + name, _, rest = zeile.partition(":") + teile = rest.split() + if teile and teile[0].isdigit(): + werte[name.strip()] = int(teile[0]) * 1024 + gesamt, verfuegbar = werte.get("MemTotal"), werte.get("MemAvailable") + if not gesamt or verfuegbar is None: + return None + return {"belegt": gesamt - verfuegbar, "gesamt": gesamt} + + +def netz_zaehler(netdev: str) -> dict[str, tuple[int, int]]: + """Schnittstelle → (empfangen, gesendet) in Bytes aus /proc/net/dev.""" + zaehler = {} + for zeile in netdev.splitlines(): + name, trenner, rest = zeile.partition(":") + teile = rest.split() + if trenner and len(teile) >= 9: + try: + zaehler[name.strip()] = (int(teile[0]), int(teile[8])) + except ValueError: + continue + return zaehler + + +def host_netz(zaehler: dict[str, tuple[int, int]], physisch: set[str]) -> dict | None: + """Verkehr des Hosts: die Summe seiner physischen Schnittstellen, so zeigt ihn auch Proxmox für den Knoten. Die + Brücke vmbr0 zählt nur, was der Host selbst empfängt und sendet, nicht den Verkehr der Gäste nach draußen. Ohne + erkennbare physische Schnittstelle doch vmbr0.""" + namen = sorted(n for n in zaehler if n in physisch) or [n for n in ("vmbr0",) if n in zaehler] + if not namen: + return None + return {"rx": sum(zaehler[n][0] for n in namen), "tx": sum(zaehler[n][1] for n in namen), "schnittstellen": namen} + + +def gast_netz(zaehler: dict[str, tuple[int, int]]) -> dict[int, tuple[int, int]]: + """vmid → (netin, netout) aus den Schnittstellen der Gäste (vethiN, tapiN). Auf der Host-Seite ist die + Richtung vertauscht: Was der Host dorthin sendet, empfängt der Gast (so rechnet auch pvestatd). Zeitgleich mit + dem Host gelesen — /cluster/resources hat Stände, die bis zu 10 s alt sind, das verzerrte die Minutenrate.""" + gaeste: dict[int, tuple[int, int]] = {} + for name, (rx, tx) in zaehler.items(): + if treffer := _GAST_SCHNITTSTELLE.fullmatch(name): + netin, netout = gaeste.get(int(treffer.group(1)), (0, 0)) + gaeste[int(treffer.group(1))] = (netin + tx, netout + rx) + return gaeste + + +def cpu_temperatur(fuehler: list[tuple[str, str, float]]) -> float | None: + """CPU-Temperatur in °C aus den hwmon-Fühlern (Chip, Beschriftung, °C): AMD k10temp/zenpower (Tdie vor Tctl), Intel + coretemp (Package id 0), sonst der erste Wert des Chips. Ohne CPU-Fühler None.""" + for chip in CPU_FUEHLER: + werte = [(beschriftung, grad) for name, beschriftung, grad in fuehler if name == chip] + if werte: + bevorzugt = [grad for gesucht in CPU_BESCHRIFTUNG for beschriftung, grad in werte if beschriftung == gesucht] + return round((bevorzugt or [werte[0][1]])[0], 1) + return None + + +def _gast_messwerte(eintrag: dict, netz: dict[int, tuple[int, int]]) -> dict: + vmid = int(eintrag["vmid"]) + netin, netout = netz.get(vmid, (eintrag.get("netin"), eintrag.get("netout"))) + werte = {"vmid": vmid, "art": eintrag.get("type"), "name": eintrag.get("name"), + "etiketten": _etiketten(eintrag.get("tags")), "status": eintrag.get("status")} + werte.update({k: eintrag.get(k) for k in ("cpu", "maxcpu", "mem", "maxmem", "disk", "maxdisk", "uptime")}) + werte.update(netin=netin, netout=netout) + return werte + + +def messpunkt(roh: dict, cpu_vorher: tuple[int, int] | None, node: str | None = None) -> dict: + """Der Minutenpunkt für den Homelab-Teil aus Rohdaten (reine Auswertung): + roh = {"zeit", "stat", "meminfo", "loadavg", "uptime", "netdev" (Dateiinhalte), "rootfs": (gesamt, frei) oder None, + "physisch": {Schnittstellen}, "fuehler": [(Chip, Beschriftung, °C)], "ressourcen": /cluster/resources oder None} + Netz-Zähler sind kumulativ (Bytes); die Raten rechnet der Homelab-Teil.""" + node = node or NODE + zaehler = netz_zaehler(roh.get("netdev") or "") + rootfs = roh.get("rootfs") + uptime = _erste_zahl(roh.get("uptime")) + host = {"cpu": cpu_anteil(cpu_vorher, cpu_zeiten(roh.get("stat") or "")), + "speicher": speicher_aus(roh.get("meminfo") or ""), + # belegt wie bei Proxmox (/nodes//status): gesamt minus frei, samt der für root reservierten Blöcke + "rootfs": {"belegt": rootfs[0] - rootfs[1], "gesamt": rootfs[0]} if rootfs else None, + "load1": _erste_zahl(roh.get("loadavg")), + "uptime": None if uptime is None else int(uptime), + "temp_cpu": cpu_temperatur(roh.get("fuehler") or []), + "netz": host_netz(zaehler, roh.get("physisch") or set())} + gnetz = gast_netz(zaehler) + gaeste = [_gast_messwerte(e, gnetz) for e in roh.get("ressourcen") or [] + if isinstance(e, dict) and e.get("type") in ("lxc", "qemu") and e.get("node") == node + and not e.get("template") and e.get("vmid") is not None] + return {"zeit": roh.get("zeit"), "host": host, "gaeste": sorted(gaeste, key=lambda g: g["vmid"])} + + +def _lesen(pfad: str) -> str: + try: + with open(pfad, encoding="utf-8", errors="replace") as f: + return f.read() + except OSError: + return "" + + +def _physische_schnittstellen() -> set[str]: + """Netzschnittstellen mit einem Gerät dahinter (/sys/class/net//device): keine Brücken, veth oder tap.""" + try: + return {n for n in os.listdir("/sys/class/net") if os.path.exists(f"/sys/class/net/{n}/device")} + except OSError: + return set() + + +def _hwmon_fuehler() -> list[tuple[str, str, float]]: + fuehler = [] + for ordner in sorted(glob.glob("/sys/class/hwmon/hwmon*")): + chip = _lesen(f"{ordner}/name").strip() + for eingang in sorted(glob.glob(f"{ordner}/temp*_input")): + try: + grad = int(_lesen(eingang).strip()) / 1000 + except ValueError: + continue + fuehler.append((chip, _lesen(eingang[: -len("_input")] + "_label").strip(), grad)) + return fuehler + + +def _messwerte_roh() -> dict: + """Alles für einen Minutenpunkt: Dateien aus /proc und /sys (billig) und ein einziger pvesh-Aufruf (Gäste).""" + roh = {"zeit": time.time(), "stat": _lesen("/proc/stat"), "netdev": _lesen("/proc/net/dev"), + "meminfo": _lesen("/proc/meminfo"), "loadavg": _lesen("/proc/loadavg"), "uptime": _lesen("/proc/uptime"), + "physisch": _physische_schnittstellen(), "fuehler": _hwmon_fuehler(), "rootfs": None, "ressourcen": None} + try: + stand = os.statvfs("/") + roh["rootfs"] = (stand.f_blocks * stand.f_frsize, stand.f_bfree * stand.f_frsize) + except (AttributeError, OSError): + pass + try: + roh["ressourcen"] = _pvesh("/cluster/resources", "--type", "vm") + except Exception: + pass # dann fehlen in diesem Punkt die Gäste, der Host nicht + return roh + + +def _bis_zur_messung(jetzt: float) -> float: + """Sekunden bis zur nächsten halben Minute, mindestens eine.""" + warten = (MESS_PHASE_S - jetzt) % MESSWERTE_ALLE_S + return warten if warten >= 1 else warten + MESSWERTE_ALLE_S + + +def messwerte_schleife() -> None: + """Jede Minute ein Messpunkt an den Homelab-Teil — im eigenen Faden, damit weder der Bericht noch ein langer Auftrag + (Sicherung, Host-Update: bis zu einer Stunde) ihn aufhält. Fehler bleiben still: derselbe steht nur einmal im + Journal, bis wieder ein Punkt durchgeht.""" + vorher = cpu_zeiten(_lesen("/proc/stat")) + zuletzt_fehler = None + while True: + time.sleep(_bis_zur_messung(time.time())) + try: + roh = _messwerte_roh() + punkt = messpunkt(roh, vorher) + vorher = cpu_zeiten(roh["stat"]) or vorher + _anfrage("POST", "/api/homelab/ausfuehrer/messwerte", punkt) + zuletzt_fehler = None + except Exception as exc: + if str(exc) != zuletzt_fehler: + log.warning("Messwerte nicht gesendet: %s", exc) + zuletzt_fehler = str(exc) + + # --- Verbindung zum Homelab-Teil ------------------------------------------------------------- def _anfrage(methode: str, pfad: str, daten: object = None) -> object: @@ -614,6 +828,8 @@ def _anfrage(methode: str, pfad: str, daten: object = None) -> object: def dauerbetrieb() -> None: + # Messwerte im eigenen Faden: Ein Auftrag darf hier eine Stunde laufen, die Messung jede Minute nicht warten. + threading.Thread(target=messwerte_schleife, name="messwerte", daemon=True).start() naechster_bericht = 0.0 while True: try: @@ -655,11 +871,18 @@ def main() -> int: logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") teile = argparse.ArgumentParser(description=__doc__.splitlines()[0]) teile.add_argument("--bericht", action="store_true", help="Bericht einmal ausgeben (nur lesend)") + teile.add_argument("--messwerte", action="store_true", help="einen Messpunkt ausgeben (nur lesend)") args = teile.parse_args() if args.bericht: json.dump(bericht(), sys.stdout, ensure_ascii=False, indent=2) print() return 0 + if args.messwerte: + vorher = cpu_zeiten(_lesen("/proc/stat")) + time.sleep(1) # die CPU über eine Sekunde statt über die Minute + json.dump(messpunkt(_messwerte_roh(), vorher), sys.stdout, ensure_ascii=False, indent=2) + print() + return 0 if not SERVER or not TOKEN: log.error("Konfiguration %s fehlt oder ist unvollständig (server, token).", KONFIG) return 1