"""Speicher und Sicherungen des Homelabs (seit 25.09.2026, User-Wunsch). Alles aus dem Bericht des Ausführers (alle 10 Minuten), dazu eine eigene Prüfung, ob das NAS antwortet: • Speicherpool (Thin-LVM, heute local-lvm): Läuft er voll, stehen alle Gäste darauf still. Gelb ab 80 %, rot ab 90 % — also bei 10 % frei (User-Vorgabe) —, für Daten wie Metadaten. Dazu Empfehlungen, was Platz bringt, der größte Gewinn zuerst: Platz, den ein Gast längst freigegeben hat (TRIM), ungenutzte und verwaiste Platten, Snapshots (seit 25.09.2026 in einem Vorschlag, verwaltet auf der Seite Homelab → Snapshots). • NAS: jedes Netzlaufwerk des Proxmox-Hosts (heute das QNAP per NFS, Ziel der nächtlichen Sicherungen): antwortet, ist eingebunden, hängt nicht, hat Platz (80/90 % wie oben). • Sicherungen: Für jeden Gast, den ein Sicherungsauftrag erfasst, liegt auf dessen Ziel eine frische Sicherung (täglicher Auftrag: höchstens 26 Stunden alt, sonst 8 Tage), und das Ziel ist erreichbar. • Systemplatte des Proxmox-Hosts (Stufen wie oben): Dort liegen die Kernel, die Sicherungen vor Updates (local) und alles vom Host selbst. Alte Kernel lassen sich per Knopf entfernen (Ausführer: kernel_aufraeumen). • Tempo (seit 25.09.2026, kern/prognose.py): Läuft etwas in weniger als 14 Tagen voll, ist es gelb, unter 3 Tagen rot — auch wenn der Füllstand selbst noch unauffällig ist. Den Verlauf dafür schreibt jeder Bericht in die Messreihe VERLAUF (verlauf_merken); Host und Container haben ihn schon in den Minutenwerten. Hinweise gehen über den Wächter (pruefe_speicher), die Oberfläche zeigt alles als Karte „Speicher und Sicherungen“ (lage). """ import logging import re import socket import threading import time from datetime import datetime from kern import messreihen, prognose from kern.zeit import LOCAL_TZ from services.homelab import apps, kanal from services.homelab.inventar import EIGENES_ETIKETT log = logging.getLogger(__name__) GELB, ROT = 0.80, 0.90 EMPFEHLUNG_AB = 1_000_000_000 # kleinere Gewinne sind keine eigene Empfehlung wert TRIM_AB = 250_000_000 # ein Container zählt beim gemeinsamen TRIM erst ab hier mit TAEGLICH_S, SONST_S = 26 * 3600, 8 * 86400 PORTS = {"nfs": 2049, "nfs4": 2049, "cifs": 445, "smb3": 445} ANTWORT_CACHE_S = 30 RANG = {"ok": 0, "gelb": 1, "rot": 2} _antwort_cache: dict[tuple[str, int], tuple[float, bool]] = {} _antwort_lock = threading.Lock() def _stufe(anteil: float | None) -> str: if anteil is None: return "ok" return "rot" if anteil >= ROT else "gelb" if anteil >= GELB else "ok" def _schlimmer(*stufen: str) -> str: return max(stufen, key=lambda s: RANG.get(s, 0)) def _gb(n: float | None) -> str: return "–" if n is None else f"{n / 1e9:.1f}".replace(".", ",") def _groesse(n: float | None) -> str: """„125 GB“, „2,3 TB“ — für Texte.""" if n is None: return "–" if n >= 1e12: return f"{n / 1e12:.1f}".replace(".", ",") + " TB" return f"{n / 1e9:.0f} GB" if n >= 10e9 else f"{_gb(n)} GB" def _name(g: dict) -> str: app = apps.app_fuer((g.get("app") or {}).get("kennung"), g.get("name")) return app.name if app else str(g.get("name") or g["vmid"]) def _wo(g: dict) -> str: return f"{'Container' if g.get('art') == 'lxc' else 'VM'} {g['vmid']}" def _zid(g: dict) -> str: return f"{'ct' if g.get('art') == 'lxc' else 'vm'}-{g['vmid']}" def _datum(ts: float) -> str: return datetime.fromtimestamp(ts, LOCAL_TZ).strftime("%d.%m. %H:%M") def _bericht() -> dict: return (kanal.bericht() or {}).get("bericht") or {} # --- Speicherpool ---------------------------------------------------------------------------------------------- def pools(bericht: dict) -> list[dict]: ergebnis = [] for p in (bericht.get("host") or {}).get("pools") or []: gesamt, belegt = int(p.get("gesamt") or 0), int(p.get("belegt") or 0) if not gesamt: continue anteil, meta = belegt / gesamt, p.get("meta_anteil") ergebnis.append({"speicher": p.get("speicher"), "belegt": belegt, "gesamt": gesamt, "frei": gesamt - belegt, "anteil": anteil, "meta_anteil": meta, "aktiv": p.get("aktiv", True), "stufe": _schlimmer(_stufe(anteil), _stufe(meta))}) return ergebnis def empfehlungen(bericht: dict) -> list[dict]: """Was im Speicherpool Platz bringt, der größte Gewinn zuerst; Snapshots (Größe unbekannt) danach. Je Eintrag: art, titel, text (was zu tun ist, mit Klickweg), kurz (für Meldungen), platz (Bytes oder None).""" gaeste = bericht.get("gaeste") or [] liste: list[dict] = [] container: list[tuple[dict, int]] = [] for g in gaeste: platte, im_pool = g.get("platte") or {}, g.get("pool_belegt") if im_pool and platte.get("belegt") is not None: gewinn = int(im_pool) - int(platte["belegt"]) if g.get("art") == "lxc": if gewinn >= TRIM_AB: container.append((g, gewinn)) elif gewinn >= EMPFEHLUNG_AB: liste.append(_trim_vm(g, int(im_pool), int(platte["belegt"]), gewinn)) for u in g.get("ungenutzt") or []: liste.append({ "art": "ungenutzt", "platz": u.get("belegt"), "titel": f"{_name(g)}: ungenutzte Platte {u['volume']}", "kurz": f"ungenutzte Platte bei {_name(g)}", "ort": "pool", "text": (f"Hängt an keinem Laufwerk mehr. Wird sie nicht mehr gebraucht: in Proxmox {_wo(g)} → Hardware " "→ „Unbenutzte Festplatte“ → Entfernen (löscht sie endgültig).")}) if (summe := sum(x for _, x in container)) >= EMPFEHLUNG_AB: container.sort(key=lambda t: -t[1]) einzeln = ", ".join(f"{_name(g)} {_gb(x)} GB" for g, x in container) befehle = ", ".join(f"`pct fstrim {g['vmid']} --ignore-mountpoints`" for g, _ in container) liste.append({ "art": "trim", "platz": summe, "titel": "Container: freigegebenen Platz an den Pool zurückgeben", "kurz": f"{_groesse(summe)} per TRIM in den Containern", "ort": "pool", "text": (f"Die Container belegen im Pool {_gb(summe)} GB mehr, als sie nutzen ({einzeln}). In Proxmox: pve → " f"Shell → {befehle} — im laufenden Betrieb und ohne Verlust.")}) for p in (bericht.get("host") or {}).get("pools") or []: for v in p.get("verwaist") or []: liste.append({ "art": "verwaist", "platz": v.get("belegt"), "titel": f"Verwaiste Platte {v['volume']}", "kurz": f"verwaiste Platte {v['volume']}", "ort": "pool", "text": (f"Gehört zu keinem Gast mehr. Wird sie nicht mehr gebraucht: in Proxmox pve → {p.get('speicher')} " "→ Datenträger → Entfernen (löscht sie endgültig).")}) if kernel := _kernel_vorschlag(bericht.get("host") or {}): liste.append(kernel) liste.sort(key=lambda e: -(e["platz"] or 0)) if snapshots := _snapshot_vorschlag(gaeste): liste.append(snapshots) return liste def _geraete_text(namen: list[str]) -> str: """„Gitea (2), NetBird und AdGuard Home“ — je Gerät einmal, mit der Zahl, wo es mehrere sind.""" teile = [f"{n} ({namen.count(n)})" if namen.count(n) > 1 else n for n in dict.fromkeys(namen)] return teile[0] if len(teile) == 1 else ", ".join(teile[:-1]) + " und " + teile[-1] def _snapshot_vorschlag(gaeste: list[dict]) -> dict | None: """Alle Snapshots in EINEM Vorschlag (seit 25.09.2026; vorher einer je Snapshot): Verwalten lassen sie sich auf der Seite Homelab → Snapshots, mit Zeitpunkt und Knöpfen. Der Container des Orchestrators selbst fehlt dort wie hier. Wie viel sie belegen, sagt Proxmox nicht — deshalb ohne Größe und zuletzt.""" eigene: list[str] = [] von_hand: list[str] = [] for g in gaeste: if EIGENES_ETIKETT in (g.get("etiketten") or []): continue for s in g.get("snapshots") or []: (eigene if re.fullmatch(r"mc2-\d{8}-\d{6}", str(s)) else von_hand).append(_name(g)) anzahl = len(eigene) + len(von_hand) if not anzahl: return None if eigene: titel = (f"{len(eigene)} Snapshot{'s' if len(eigene) != 1 else ''} des Orchestrators — verwalten unter " "Homelab → Snapshots") text = (f"Bei {_geraete_text(eigene)}. Läuft ein Gerät seit seinem Snapshot gut, kann der weg — auf der Seite " "Snapshots per Knopf, dort auch mit Zeitpunkt und dem Weg zurück.") else: titel = (f"{len(von_hand)} von Hand angelegte{'r' if len(von_hand) == 1 else ''} " f"Snapshot{'s' if len(von_hand) != 1 else ''} — ansehen unter Homelab → Snapshots") text = f"Bei {_geraete_text(von_hand)}." if eigene and von_hand: text += f" Dazu {len(von_hand)} von Hand in Proxmox angelegt ({_geraete_text(von_hand)})." if von_hand: text += " Von Hand angelegte löscht der Orchestrator nicht; das geht in Proxmox." text += (" Wie viel ein Snapshot belegt, sagt Proxmox nicht — bei Containern meist wenig, er wächst mit jeder " "Änderung seitdem.") return {"art": "snapshot", "platz": None, "ort": "pool", "titel": titel, "text": text, "kurz": f"{anzahl} Snapshot{'s' if anzahl != 1 else ''} (Homelab → Snapshots)"} def _kernel_vorschlag(host: dict) -> dict | None: """Alte Kernel auf der Systemplatte des Hosts — mit Knopf. Die Liste stammt vom Ausführer (was apt für entbehrlich hält, nie der laufende oder der nächste); beim Klick bestimmt er sie noch einmal selbst.""" alt = host.get("kernel_alt") or [] if not alt: return None platz = sum(int(k.get("groesse") or 0) for k in alt) bleiben = ", ".join(str(v).removesuffix("-pve") for v in host.get("kernel_bleiben") or []) or "der laufende" return { "art": "kernel", "ort": "systemplatte", "platz": platz, "titel": f"Proxmox-Host: {len(alt)} alte Kernel entfernen", "kurz": f"{_groesse(platz)} alte Kernel auf dem Proxmox-Host", "text": (f"Liegen auf der Systemplatte des Hosts, nicht im Speicherpool. Entfernt wird nur, was Proxmox selbst für " f"entbehrlich hält; es bleiben {bleiben} — darunter der laufende und der für den nächsten Start. Die " "Gäste laufen weiter, ein Neustart ist dafür nicht nötig."), "aktion": {"methode": "POST", "pfad": "/api/homelab/ziele/pve/kernel-aufraeumen", "label": "Alte Kernel entfernen", "frage": (f"{len(alt)} alte Kernel vom Proxmox-Host entfernen (etwa {_groesse(platz)})? Es bleiben " f"{bleiben}. Die Gäste laufen weiter, ein Neustart ist nicht nötig.")}, } def systemplatte(bericht: dict) -> dict | None: p = (bericht.get("host") or {}).get("systemplatte") or {} gesamt, belegt = p.get("gesamt"), p.get("belegt") if not gesamt or belegt is None: return None anteil = belegt / gesamt return {"belegt": belegt, "gesamt": gesamt, "frei": gesamt - belegt, "anteil": anteil, "stufe": _stufe(anteil)} def _trim_vm(g: dict, im_pool: int, genutzt: int, gewinn: int) -> dict: kopf = f"Im Pool belegt die VM {_gb(im_pool)} GB, genutzt sind darin {_gb(genutzt)} GB." if g.get("discard"): text = (f"{kopf} In der VM `sudo fstrim -av` ausführen; Linux-Gäste tun das meist ohnehin jede Woche " "(fstrim.timer).") else: text = (f"{kopf} Discard ist aus, deshalb kommt Platz, den die VM freigibt, nie im Pool an. In Proxmox: " f"VM {g['vmid']} → Hardware → Festplatte → Bearbeiten → „Discard“ anhaken, dann die VM einmal neu " "starten (die Dienste darin sind kurz weg). Danach gibt sie freien Platz jede Woche von selbst zurück; " "sofort mit `sudo fstrim -av` in der VM.") return {"art": "trim-vm", "platz": gewinn, "titel": f"{_name(g)}: {_groesse(gewinn)} an den Pool zurückgeben", "kurz": f"{_groesse(gewinn)} bei {_name(g)} per TRIM" + ("" if g.get("discard") else " (Discard einschalten)"), "ort": "pool", "text": text} # --- NAS -------------------------------------------------------------------------------------------------------- def antwortet(server: str, port: int) -> bool: """Nimmt das NAS auf dem Port seines Dienstes Verbindungen an? (30 s gemerkt: Wächter und Oberfläche fragen oft.)""" jetzt = time.monotonic() with _antwort_lock: if (e := _antwort_cache.get((server, port))) and jetzt - e[0] < ANTWORT_CACHE_S: return e[1] try: with socket.create_connection((server, port), timeout=3): ok = True except OSError: ok = False with _antwort_lock: _antwort_cache[(server, port)] = (jetzt, ok) return ok def nas(bericht: dict, pruefen=None) -> list[dict]: pruefen = pruefen or antwortet ergebnis = [] for m in (bericht.get("host") or {}).get("netzlaufwerke") or []: server, port = m.get("server"), PORTS.get(str(m.get("art"))) erreichbar = pruefen(server, port) if server and port else None platz = m.get("platz") or {} gesamt, belegt = platz.get("gesamt"), platz.get("belegt") anteil = belegt / gesamt if gesamt and belegt is not None else None eintrag = {"server": server, "ziel": m.get("ziel"), "quelle": m.get("quelle"), "art": m.get("art"), "eingebunden": bool(m.get("eingebunden")), "zustand": m.get("zustand"), "erreichbar": erreichbar, "belegt": belegt, "gesamt": gesamt, "anteil": anteil, "stufe": "ok", "problem": None} if erreichbar is False: eintrag.update(stufe="rot", problem=f"Das NAS ({server}) antwortet nicht.") elif not m.get("eingebunden"): eintrag.update(stufe="rot", problem=f"{m.get('ziel')} ist auf dem Proxmox-Host nicht eingebunden.") elif m.get("zustand") == "haengt": eintrag.update(stufe="rot", problem=f"Die Einbindung {m.get('ziel')} hängt: Der Proxmox-Host bekommt keine " "Antwort vom NAS.") elif m.get("zustand") == "fehler": eintrag.update(stufe="gelb", problem=f"Die Belegung von {m.get('ziel')} ließ sich nicht lesen.") else: eintrag["stufe"] = _stufe(anteil) ergebnis.append(eintrag) return ergebnis # --- Sicherungen ------------------------------------------------------------------------------------------------ def max_alter_s(zeitplan: object) -> int: """Wie alt die jüngste Sicherung höchstens sein darf: täglich (nur Uhrzeit, etwa „02:30“) 26 Stunden, bei Wochentagen oder allem Ausgefalleneren 8 Tage.""" return TAEGLICH_S if re.fullmatch(r"\s*(\d{1,2}|\*)(:\d{2})?\s*", str(zeitplan or "")) else SONST_S def _erfasst(job: dict, vmid: int) -> bool: """Sichert dieser Auftrag den Gast? (alle außer den ausgenommenen, oder die aufgezählten)""" if vmid in (job.get("ausgenommen") or []): return False return bool(job.get("alle")) or vmid in (job.get("vmids") or []) def sicherungen(bericht: dict, jetzt: float) -> dict: host = bericht.get("host") or {} jobs = [j for j in host.get("sicherung_jobs") or [] if j.get("an")] ziele = host.get("sicherungsziele") or {} probleme: list[str] = [] for name in sorted({j.get("speicher") for j in jobs if j.get("speicher")}): z = ziele.get(name) or {} if not z.get("aktiv"): probleme.append(f"Das Sicherungsziel {name} ist nicht erreichbar.") elif z.get("fehler"): probleme.append(f"Die Sicherungen auf {name} ließen sich nicht lesen.") erfasst, fehlend, zuletzt = 0, [], None for g in bericht.get("gaeste") or []: eigene = [j for j in jobs if _erfasst(j, g["vmid"])] if not eigene: continue erfasst += 1 stand = [((ziele.get(j.get("speicher")) or {}).get("neueste") or {}).get(str(g["vmid"])) for j in eigene] juengste = max((t for t in stand if t), default=None) if juengste: zuletzt = max(zuletzt or 0, juengste) if not any(t and jetzt - t <= max_alter_s(j.get("zeitplan")) for t, j in zip(stand, eigene, strict=True)): fehlend.append({"ziel": _zid(g), "name": _name(g), "zuletzt": juengste}) stufe = "keine" if not jobs else "ok" if fehlend: stufe = "rot" if len(fehlend) == erfasst else "gelb" if probleme: stufe = "rot" return {"jobs": [{"zeitplan": j.get("zeitplan"), "speicher": j.get("speicher"), "alle": j.get("alle")} for j in jobs], "ziele": [{"speicher": n, "art": z.get("art"), "aktiv": z.get("aktiv"), "belegt": z.get("belegt"), "gesamt": z.get("gesamt")} for n, z in sorted(ziele.items())], "erfasst": erfasst, "fehlend": fehlend, "zuletzt": zuletzt, "probleme": probleme, "stufe": stufe} # --- Vorhersage „voll in etwa N Tagen“ (seit 25.09.2026) -------------------------------------------------------- VERLAUF = "speicher" # Quelle in kern/messreihen.py: die Belegungen aus jedem Bericht (alle zehn Minuten) def _prozent(belegt: object, gesamt: object) -> float | None: if not (messreihen.ist_zahl(belegt) and messreihen.ist_zahl(gesamt)) or gesamt <= 0: return None return round(belegt / gesamt * 100, 2) def verlauf_merken(bericht: dict, jetzt: float | None = None) -> None: """Die Belegungen eines Berichts als ein Punkt der Messreihe VERLAUF (in %): je Speicherpool Daten („pool:“) und Metadaten („pool-meta:“), je Netzlaufwerk („nas:“), je Sicherungsziel („ziel:“) und die Platte jeder VM („vm-“ — die Minutenwerte kennen sie nicht). Ein kaputter Bericht kostet nur diesen Punkt.""" try: host = bericht.get("host") if isinstance(bericht.get("host"), dict) else {} werte: dict[str, float | None] = {} for p in pools(bericht): werte[f"pool:{p['speicher']}"] = round(p["anteil"] * 100, 2) if messreihen.ist_zahl(p["meta_anteil"]): werte[f"pool-meta:{p['speicher']}"] = round(p["meta_anteil"] * 100, 2) for m in host.get("netzlaufwerke") or []: platz = m.get("platz") or {} werte[f"nas:{m.get('ziel')}"] = _prozent(platz.get("belegt"), platz.get("gesamt")) for name, z in (host.get("sicherungsziele") or {}).items(): werte[f"ziel:{name}"] = _prozent(z.get("belegt"), z.get("gesamt")) for g in bericht.get("gaeste") or []: if g.get("art") == "qemu": platte = g.get("platte") or {} werte[f"vm-{g['vmid']}"] = _prozent(platte.get("belegt"), platte.get("gesamt")) werte = {name: wert for name, wert in werte.items() if wert is not None} if werte: messreihen.schreiben(VERLAUF, werte, jetzt) except Exception: log.warning("speicher: Verlauf der Belegungen nicht geschrieben", exc_info=True) def platten_reihe(geraet_id: str) -> tuple[str, str, float] | None: """Wo der Verlauf der Platte eines Geräts liegt: (Quelle, Name, voll bei %). Host und Container in den Minutenwerten; VMs im VERLAUF, denn ihre Platte kennt nur der Bericht (über den Gast-Agenten).""" if geraet_id == "pve": return "pve", "platte", prognose.VOLL_EXT4 if re.fullmatch(r"ct-\d+", geraet_id): return geraet_id, "platte", prognose.VOLL_EXT4 if re.fullmatch(r"vm-\d+", geraet_id): return VERLAUF, geraet_id, prognose.VOLL_EXT4 return None def platten_vorhersage(geraet_id: str, jetzt: float | None = None) -> dict | None: reihe = platten_reihe(geraet_id) return prognose.aus_messreihe(*reihe, jetzt=jetzt) if reihe else None def _vorhersage(name: str, jetzt: float | None) -> dict: return prognose.aus_messreihe(VERLAUF, name, jetzt=jetzt) def pool_vorhersage(speicher: str, jetzt: float | None = None) -> tuple[dict, bool]: """Die Vorhersage eines Pools — Daten oder Metadaten, was früher voll ist — und ob es die Metadaten sind.""" daten, meta = _vorhersage(f"pool:{speicher}", jetzt), _vorhersage(f"pool-meta:{speicher}", jetzt) return (meta, True) if prognose.bis_voll(meta) < prognose.bis_voll(daten) else (daten, False) # --- Für Oberfläche und Wächter ---------------------------------------------------------------------------------- def lage(jetzt: float | None = None) -> dict: """Für GET /api/homelab/speicher. Pools, Systemplatte, NAS und Sicherungsziele tragen seit 25.09.2026 je eine „vorhersage“ (kern/prognose.vorhersage). jetzt nur für Tests: Dann wird die Vorhersage nicht gemerkt.""" b = _bericht() zeit = time.time() if jetzt is None else jetzt ergebnis = {"bericht": bool(b), "pools": pools(b), "systemplatte": systemplatte(b), "nas": nas(b), "sicherungen": sicherungen(b, zeit), "empfehlungen": empfehlungen(b)} for p in ergebnis["pools"]: p["vorhersage"] = pool_vorhersage(p["speicher"], jetzt)[0] if ergebnis["systemplatte"]: ergebnis["systemplatte"]["vorhersage"] = platten_vorhersage("pve", jetzt) for n in ergebnis["nas"]: n["vorhersage"] = None if n["problem"] else _vorhersage(f"nas:{n['ziel']}", jetzt) for z in ergebnis["sicherungen"]["ziele"]: z["vorhersage"] = _vorhersage(f"ziel:{z['speicher']}", jetzt) return ergebnis def pruefe_speicher(jetzt: float | None = None) -> list: """Befunde für den Wächter. Ohne Bericht keine — das Schweigen des Ausführers meldet pruefe_ausfuehrer(). Pool, Systemplatte und NAS melden sich auch nach dem Tempo (siehe oben); bestimmt die Vorhersage die Stufe, steht „in etwa N Tagen voll“ im Titel, sonst hängt der Text sie an.""" from services.waechter import Befund # der Wächter registriert diese Prüfung, also erst hier b = _bericht() if not b: return [] zeit = time.time() if jetzt is None else jetzt befunde = [] tipps = empfehlungen(b) platz_tipp = "; ".join([e["kurz"] for e in tipps if e.get("ort") == "pool"][:3]) for p in pools(b): v, meta = pool_vorhersage(p["speicher"], jetzt) stufe, kommend = prognose.schlimmer(p["stufe"], v) if stufe == "ok": continue rest = ("Läuft er voll, stehen alle Gäste darauf still. " + (f"Platz bringt: {platz_tipp}. " if platz_tipp else "") + "Alles dazu im Homelab-Cockpit unter „Speicher und Sicherungen“.") if kommend: was = "Metadaten des Speicherpools" if meta else "Speicherpool" titel = f"{was} {p['speicher']} {prognose.in_tagen(v['tage'])} voll" text = f"{prognose.tempo(v)} Noch {_groesse(p['frei'])} frei. {rest}" else: anteil = max(p["anteil"], p["meta_anteil"] or 0) was = "Metadaten des Speicherpools" if (p["meta_anteil"] or 0) > p["anteil"] else "Speicherpool" titel = f"{was} {p['speicher']} zu {min(100, round(anteil * 100))} % voll" text = f"Noch {_groesse(p['frei'])} frei.{prognose.dazu(v)} {rest}" befunde.append(Befund(id=f"pool:{p['speicher']}", stufe=stufe, titel=titel, text=text, quelle="pve")) sp = systemplatte(b) v = platten_vorhersage("pve", jetzt) if sp else None stufe, kommend = prognose.schlimmer(sp["stufe"], v) if sp else ("ok", False) if sp and stufe != "ok": kernel = next((e for e in tipps if e["art"] == "kernel"), None) befunde.append(Befund( id="host-platte", stufe=stufe, titel=(f"Systemplatte des Proxmox-Hosts {prognose.in_tagen(v['tage'])} voll" if kommend else f"Systemplatte des Proxmox-Hosts zu {min(100, round(sp['anteil'] * 100))} % voll"), text=((f"{prognose.tempo(v)} " if kommend else "") + f"Noch {_groesse(sp['frei'])} frei." + ("" if kommend else prognose.dazu(v)) + " Läuft sie voll, scheitern Updates und die Sicherungen vor Updates. " + (f"Platz bringt: {kernel['kurz']} (Knopf im Homelab-Cockpit unter „Speicher und Sicherungen“)." if kernel else "Alte Sicherungen unter /var/lib/vz/dump und ISO-Abbilder prüfen.")), quelle="pve")) for n in nas(b): v = None if n["problem"] else _vorhersage(f"nas:{n['ziel']}", jetzt) stufe, kommend = prognose.schlimmer(n["stufe"], v) if stufe == "ok": continue frei = _groesse((n["gesamt"] or 0) - (n["belegt"] or 0)) platz = ("Die Sicherungen brauchen jede Nacht Platz: im PBS die Aufbewahrung kürzen (Datastore → Prune) oder " "alte Dateien auf dem NAS aufräumen.") if n["problem"]: titel, text = n["problem"], ("Die nächtlichen Sicherungen landen dort; bis das behoben ist, scheitern sie. " "Strom und Netz des NAS prüfen, dann in dessen Oberfläche nachsehen.") elif kommend: titel = f"NAS ({n['server']}) {prognose.in_tagen(v['tage'])} voll" text = f"{prognose.tempo(v)} Noch {frei} frei. {platz}" else: titel = f"NAS ({n['server']}) zu {min(100, round((n['anteil'] or 0) * 100))} % voll" text = f"Noch {frei} frei.{prognose.dazu(v)} {platz}" befunde.append(Befund(id=f"nas:{n['ziel']}", stufe=stufe, titel=titel, text=text, quelle="nas")) s = sicherungen(b, zeit) if s["stufe"] in ("gelb", "rot"): teile = list(s["probleme"]) if s["fehlend"]: namen = ", ".join(f"{f['name']} ({'zuletzt ' + _datum(f['zuletzt']) if f['zuletzt'] else 'noch nie'})" for f in s["fehlend"][:6]) teile.append(f"Keine frische Sicherung: {namen}.") befunde.append(Befund( id="sicherungen", stufe=s["stufe"], titel=("Sicherungen scheitern" if s["stufe"] == "rot" else f"Sicherung fehlt bei {len(s['fehlend'])} Gerät" + ("en" if len(s["fehlend"]) != 1 else "")), text=" ".join(teile) + " In Proxmox: Rechenzentrum → Backup und die Aufgaben-Liste unten zeigen, woran es lag.", quelle="pve")) return befunde