mission-control-v2/backend/services/homelab/messwerte.py

"""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-<vmid>“, „vm-<vmid>“ 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-<vmid> bzw. vm-<vmid> 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