"""Snapshots und Sicherungen verwalten — die Seite Homelab → Snapshots (seit 25.09.2026, User-Wunsch). Liste (liste): je Gerät aus dem Bericht des Ausführers die Snapshots mit Zeitpunkt, Beschreibung und Herkunft, dazu die Sicherungen des Orchestrators (vzdump-Archive, heute nur beim PBS, der keinen Snapshot kann) und ob für das Gerät gerade ein Lauf läuft. Ein Ausführer vor dem 25.09.2026 liefert nur die Namen (snapshots, sicherungen): Dann kommt der Zeitpunkt aus dem Namen (mc2-JJJJMMTT-HHMMSS bzw. …-JJJJ_MM_TT-HH_MM_SS, Ortszeit des Hosts), von Hand angelegte Snapshots stehen ohne Zeit da. Wie viel ein Snapshot belegt, sagt Proxmox bei LVM-Thin nicht — hier steht deshalb keine Zahl, die Seite sagt es so. Herkunft eines Snapshots: update Rückweg vor einem Update (Name mc2-…) knopf „Snapshot anlegen“ auf dieser Seite (Name mc2-…; der Ausführer schreibt „Auf Knopfdruck …“ in die Beschreibung, und der Lauf hier merkt sich den Namen). Solche räumt kein grünes Update weg (updates._aufraeumen fragt knopf_namen) — die löscht nur der User. proxmox von Hand in Proxmox angelegt: nur zur Ansicht, nie gelöscht oder zurückgesetzt (der Ausführer lehnt das ohnehin ab, er fasst nur mc2-… an). Die Aktionen laufen als Läufe wie „Jetzt updaten“ (Mechanik aus updates.py, bewusst mitbenutzt): in homelab-laeufe.json, je Gerät höchstens einer, nie während „Alle aktualisieren“ (SAMMELLAUF_SPERRE) oder eines Updates irgendwo (_sperre: der Ausführer arbeitet einen Auftrag nach dem anderen ab) und nur, wenn der Ausführer verbunden ist. So stehen sie im Protokoll, der Wächter lässt ein Gerät mitten im Zurücksetzen in Ruhe, und nichts kommt einem Update in die Quere. Neue Befehle braucht der Ausführer nicht: snapshot-anlegen snapshot (Anlass „knopf“) → frischer Bericht. Eine Meldung nur, wenn es scheitert. snapshot-loeschen snapshot_loeschen (nur mc2-…) → frischer Bericht. Eine Meldung nur, wenn es scheitert. snapshot-zurueck zurueck (nur mc2-…; der Ausführer stoppt, setzt zurück, startet) → 20 s warten → frischer Bericht → Prüfung wie nach einem Update (läuft, Weboberfläche antwortet). Immer mit Meldung, dringend, wenn es scheitert oder die Prüfung rot ist. sicherung-loeschen sicherung_loeschen (nur Sicherungen des Orchestrators) → frischer Bericht. Meldung nur bei Fehler. Im Update-Verlauf stehen sie nicht: Das sind keine Updates. Seit dem 25.09.2026 außerdem (User-Klärung): je Gerät „Rückweg testen“ (updates.starten(…, "probe"): Snapshot bzw. Sicherung, absichtlich rot, zurück, Prüfung) und oben „Alle Update-Rückwege löschen“ (rueckwege_loeschen): löscht die Snapshots mit Herkunft „update“ und die Sicherungen des Orchestrators nacheinander über dieselben Läufe wie die einzelnen Knöpfe. Per Knopf und von Hand angelegte Snapshots bleiben. Endgültiges Löschen geschieht nur auf den Klick des Users hin — nie von selbst. """ import logging import re import threading import time from collections.abc import Callable from datetime import datetime from kern.zeit import LOCAL_TZ from services import announce from services.homelab import apps, inventar, kanal, sammellauf, updates log = logging.getLogger(__name__) # Wie SNAPSHOT_NAME im Ausführer: Nur solche Snapshots fasst er an. Der Zeitpunkt im Namen ist die Ortszeit des # Proxmox-Hosts (Europe/Berlin, wie LOCAL_TZ). SNAPSHOT_NAME = re.compile(r"^mc2-(\d{4})(\d{2})(\d{2})-(\d{2})(\d{2})(\d{2})$") ARCHIV_ZEIT = re.compile(r"-(\d{4})_(\d{2})_(\d{2})-(\d{2})_(\d{2})_(\d{2})\.") KNOPF_BESCHREIBUNG = "Auf Knopfdruck" # so beginnt die Beschreibung, die der Ausführer beim Anlass „knopf“ setzt # Die Läufe dieser Seite: was gerade geschieht (für „Läuft: …“) und wie die Aktion heißt. LAEUFT = {"snapshot-anlegen": "Snapshot wird angelegt", "snapshot-loeschen": "Snapshot wird gelöscht", "snapshot-zurueck": "Wird zurückgesetzt", "sicherung-loeschen": "Sicherung wird gelöscht"} WAS = {"snapshot-anlegen": "Snapshot anlegen", "snapshot-loeschen": "Snapshot löschen", "snapshot-zurueck": "Zurücksetzen", "sicherung-loeschen": "Sicherung löschen"} GUT = {"angelegt", "geloescht", "zurueckgesetzt"} ZULETZT_S = 6 * 3600 # so lange steht das Ergebnis der letzten Aktion beim Gerät BETREFF = "[Homelab-Snapshot]" BETREFF_ALARM = "[Alarm] Homelab-Snapshot" # --- Kleine Helfer ------------------------------------------------------------------------------------------------- def _zid(g: dict) -> str: return f"{'ct' if g.get('art') == 'lxc' else 'vm'}-{g['vmid']}" def _wo(g: dict) -> str: return f"{'Container' if g.get('art') == 'lxc' else 'VM'} {g['vmid']}" def _name(g: dict) -> str: """Der Name wie in der Update-Liste (Gitea, Arcane (Docker) …).""" app = apps.app_fuer((g.get("app") or {}).get("kennung"), g.get("name")) return app.name if app else str(g.get("name") or _zid(g)) def _ortszeit(m: re.Match | None) -> float | None: if not m: return None try: return datetime(*(int(x) for x in m.groups()), tzinfo=LOCAL_TZ).timestamp() except ValueError: return None def _wann(ts: float | None) -> str: """„vom 25.09.2026 11:26“ — leer, wo der Zeitpunkt fehlt.""" return f"vom {datetime.fromtimestamp(ts, LOCAL_TZ):%d.%m.%Y %H:%M}" if ts else "" def _groesse(n: int | None) -> str | None: if not n: return None return f"{n / 1e6:.0f} MB" if n < 1e9 else f"{n / 1e9:.1f} GB".replace(".", ",") def _aktion(pfad: str, frage: str, label: str) -> dict: """Ein Knopf wie im Ziel-Modell (kern/ziele.py: Aktion): Methode, Pfad, Rückfrage, Beschriftung.""" return {"methode": "POST", "pfad": pfad, "frage": frage, "label": label} def _grund(exc: Exception) -> str: return str(exc).rstrip(".") # --- Snapshots und Sicherungen eines Geräts --------------------------------------------------------------------------- def _knopf(laeufe: list[dict]) -> set[tuple[str, str]]: """(Gerät, Snapshot) aus den Läufen „Snapshot anlegen“ — so erkennt die Seite ihre Snapshots auch mit einem Ausführer, der die Beschreibung „Auf Knopfdruck …“ noch nicht schreibt.""" return {(str(x.get("ziel")), str(x["snapshot"])) for x in laeufe if x.get("baustein") == "snapshot-anlegen" and x.get("snapshot")} def knopf_namen(ziel_id: str, gast: dict) -> set[str]: """Die Snapshots eines Gasts, die jemand auf der Seite Snapshots angelegt hat. Kein grünes Update räumt sie weg (updates._aufraeumen) — die löscht nur der User.""" namen = {name for ziel, name in _knopf(updates.laeufe(100)) if ziel == ziel_id} for d in gast.get("snapshot_details") or []: if isinstance(d, dict) and str(d.get("beschreibung") or "").startswith(KNOPF_BESCHREIBUNG): namen.add(str(d.get("name"))) return namen def _snapshots(g: dict, zid: str, knopf: set[tuple[str, str]]) -> list[dict]: """Die Snapshots eines Gasts, ältester zuerst: Name, Zeitpunkt, Beschreibung, Herkunft, ob vom Orchestrator.""" details = g.get("snapshot_details") roh = ([d for d in details if isinstance(d, dict) and d.get("name")] if isinstance(details, list) else [{"name": n} for n in g.get("snapshots") or [] if n]) liste = [] for d in roh: name = str(d["name"]) m = SNAPSHOT_NAME.match(name) zeit = d.get("zeit") if isinstance(d.get("zeit"), (int, float)) else _ortszeit(m) beschreibung = str(d.get("beschreibung") or "").strip() or None if not m: herkunft = "proxmox" elif (zid, name) in knopf or (beschreibung or "").startswith(KNOPF_BESCHREIBUNG): herkunft = "knopf" else: herkunft = "update" liste.append({"name": name, "zeit": zeit, "beschreibung": beschreibung, "herkunft": herkunft, "eigen": bool(m)}) return sorted(liste, key=lambda s: (s["zeit"] or 0, s["name"])) def _sicherungen(g: dict) -> list[dict]: """Die Sicherungen des Orchestrators für diesen Gast (der Ausführer nennt nur eigene), älteste zuerst.""" details = g.get("sicherung_details") roh = ([d for d in details if isinstance(d, dict) and d.get("volid")] if isinstance(details, list) else [{"volid": v} for v in g.get("sicherungen") or [] if v]) liste = [] for d in roh: volid = str(d["volid"]) datei = volid.rsplit("/", 1)[-1] zeit = d.get("zeit") if isinstance(d.get("zeit"), (int, float)) else _ortszeit(ARCHIV_ZEIT.search(datei)) groesse = d.get("groesse") if isinstance(d.get("groesse"), int) else None liste.append({"datei": datei, "speicher": volid.split(":", 1)[0], "zeit": zeit, "groesse": groesse}) return sorted(liste, key=lambda e: (e["zeit"] or 0, e["datei"])) def _anlegen_grund(g: dict) -> str | None: """Warum es für diesen Gast keinen Knopf „Snapshot anlegen“ gibt — oder None.""" if not g.get("erlaubt"): return ("Nicht freigegeben: Das Etikett „watcher“ fehlt (oder „watcher-aus“ ist gesetzt). Der Orchestrator fasst " "dieses Gerät nicht an; Snapshots gehen hier nur von Hand in Proxmox.") if "snapshot_moeglich" not in g: return "Ob ein Snapshot geht, ließ sich nicht lesen" + (f": {g['fehler']}" if g.get("fehler") else ".") if not g["snapshot_moeglich"]: satz = f"Kein Snapshot möglich: {str(g.get('snapshot_grund') or 'Der Speicher kann keine').rstrip('.')}." if g.get("sicherung_moeglich"): satz += " Vor einem Update legt der Orchestrator stattdessen eine Sicherung an." return satz return None # --- Rückfragen (die Oberfläche zeigt sie wörtlich) --------------------------------------------------------------- def _frage_anlegen(g: dict, name: str) -> str: vm = g.get("art") != "lxc" saetze = [f"Jetzt einen Snapshot von {name} ({_wo(g)}) anlegen?"] if g.get("status") == "running": saetze.append(f"{'Die VM' if vm else 'Der Container'} läuft dabei weiter.") saetze.append("Diesen Snapshot räumt kein Update weg — du löschst ihn hier, wenn du ihn nicht mehr brauchst.") saetze.append("Die VM schreibt laufend; ihr Snapshot wächst im Speicherpool schnell mit, also nicht lange stehen " "lassen." if vm else "Im Speicherpool belegt er so viel, wie sich seitdem im Container ändert.") return " ".join(saetze) def _frage_loeschen(name: str, s: dict) -> str: wie = "Rückweg eines Updates" if s["herkunft"] == "update" else "auf Knopfdruck angelegt" return (f"Den Snapshot {s['name']} von {name} ({', '.join(t for t in (_wann(s['zeit']), wie) if t)}) endgültig " "löschen? Am Gerät selbst ändert sich nichts; nur der Weg zurück auf diesen Stand ist danach weg.") def _frage_zurueck(g: dict, name: str, s: dict) -> str: vm = g.get("art") != "lxc" wann = _wann(s["zeit"]) stand = f"den Snapshot {wann} ({s['name']})" if wann else f"den Snapshot {s['name']}" saetze = [f"{name} ({_wo(g)}) auf {stand} zurücksetzen?", (f"Alles, was sich seitdem {'in der VM' if vm else 'im Container'} geändert hat — Daten, Einstellungen, " "Updates —, ist danach verloren. Rückgängig machen lässt sich das nicht."), (f"{'Die VM' if vm else 'Der Container'} wird dafür gestoppt und wieder gestartet; " f"{name} ist ein paar Minuten nicht erreichbar.")] if g.get("status") != "running": saetze.append(f"Gerade {'ist die VM' if vm else 'ist der Container'} gestoppt — danach läuft " f"{'sie' if vm else 'er'}.") saetze.append(f"Danach prüft der Orchestrator, ob {name} wieder antwortet, und meldet das Ergebnis. Willst du den " "jetzigen Stand behalten, lege vorher einen Snapshot an.") return " ".join(saetze) def _frage_probe(g: dict, name: str) -> str: vm = g.get("art") != "lxc" snapshot = bool(g.get("snapshot_moeglich")) return (f"Den Rückweg von {name} ({_wo(g)}) echt testen? Der Orchestrator legt " f"{'einen Snapshot' if snapshot else 'eine Sicherung'} an, wertet die Prüfung absichtlich als rot und setzt " f"{'die VM' if vm else 'den Container'} darauf zurück — ein Update gibt es dabei nicht. {name} ist ein paar " "Minuten nicht erreichbar; danach prüft er, ob es wieder läuft und antwortet, und meldet das Ergebnis. " f"{'Der Snapshot' if snapshot else 'Die Sicherung'} bleibt danach liegen — wegräumen kannst du hier.") def _frage_sicherung(name: str, e: dict) -> str: teile = [t for t in (_wann(e["zeit"]), _groesse(e["groesse"])) if t] + [f"auf „{e['speicher']}“"] return (f"Die Sicherung {e['datei']} von {name} ({', '.join(teile)}) endgültig löschen? Der Orchestrator hat sie " "vor einem Update als Rückweg angelegt. Am Gerät selbst ändert sich nichts, und die nächtlichen Sicherungen " "(Proxmox → Backup) bleiben, wie sie sind.") # --- Liste für die Oberfläche ------------------------------------------------------------------------------------------ def _laeuft_text(x: dict) -> str: """Was für ein Gerät gerade läuft, in Worten — auch ein Update, das die Knöpfe hier sperrt.""" baustein = str(x.get("baustein")) if baustein == "snapshot-zurueck" and "zurueck: ok" in (x.get("schritte") or []): return "Zurückgesetzt, wird geprüft" return LAEUFT.get(baustein, "Update läuft") def _zuletzt(x: dict | None, jetzt: float) -> dict | None: """Das Ergebnis der letzten Aktion dieser Seite für ein Gerät, solange es frisch ist (ZULETZT_S).""" ende = (x or {}).get("ende") or (x or {}).get("start") if not isinstance(ende, (int, float)) or jetzt - ende > ZULETZT_S: return None if x.get("status") == "unterbrochen": return {"ok": False, "ende": ende, "text": f"{WAS.get(str(x.get('baustein')), 'Die Aktion')} unterbrochen: Der Homelab-Teil wurde " "währenddessen neu gestartet. Was am Gerät geschah, zeigt die Liste."} return {"ok": x.get("ergebnis") in GUT, "ende": ende, "text": str(x.get("text") or "")} def _geraet(g: dict, laufend: dict[str, dict], fertig: dict[str, dict], knopf: set, jetzt: float) -> dict: zid, name = _zid(g), _name(g) grund = _anlegen_grund(g) snapshots = _snapshots(g, zid, knopf) for s in snapshots: darf = s["eigen"] and bool(g.get("erlaubt")) pfad = f"/api/homelab/snapshots/{zid}/{s['name']}" s["loeschen"] = _aktion(f"{pfad}/loeschen", _frage_loeschen(name, s), "Snapshot löschen") if darf else None s["zuruecksetzen"] = (_aktion(f"{pfad}/zuruecksetzen", _frage_zurueck(g, name, s), "Zurücksetzen") if darf else None) sicherungen = _sicherungen(g) for e in sicherungen: e["loeschen"] = (_aktion(f"/api/homelab/sicherungen/{zid}/{e['datei']}/loeschen", _frage_sicherung(name, e), "Sicherung löschen") if g.get("erlaubt") else None) lauf = laufend.get(zid) probe_geht = bool(g.get("erlaubt")) and bool(g.get("snapshot_moeglich") or g.get("sicherung_moeglich")) return {"id": zid, "vmid": g["vmid"], "art": "container" if g.get("art") == "lxc" else "vm", "name": name, "wo": _wo(g), "status": g.get("status"), "erlaubt": bool(g.get("erlaubt")), "anlegen": None if grund else _aktion(f"/api/homelab/snapshots/{zid}/anlegen", _frage_anlegen(g, name), "Snapshot anlegen"), "probe": (_aktion(f"/api/homelab/ziele/{zid}/rueckweg-probe", _frage_probe(g, name), "Rückweg testen") if probe_geht else None), "grund": grund, "laeuft": _laeuft_text(lauf) if lauf else None, "zuletzt": _zuletzt(fertig.get(zid), jetzt), "snapshots": snapshots, "sicherungen": sicherungen} def liste(jetzt: float | None = None) -> dict: """GET /api/homelab/snapshots: je Gerät (ohne den Container des Orchestrators selbst) Snapshots und Sicherungen.""" jetzt = time.time() if jetzt is None else jetzt eingang = kanal.bericht() bericht = (eingang or {}).get("bericht") or {} zuletzt = kanal.zuletzt() gaeste = [g for g in bericht.get("gaeste") or [] if isinstance(g, dict) and isinstance(g.get("vmid"), int) and inventar.EIGENES_ETIKETT not in (g.get("etiketten") or [])] laeufe = updates.laeufe(100) # neueste zuerst laufend: dict[str, dict] = {} fertig: dict[str, dict] = {} for x in laeufe: if x.get("status") == "laeuft": laufend.setdefault(str(x.get("ziel")), x) elif x.get("baustein") in LAEUFT: fertig.setdefault(str(x.get("ziel")), x) knopf = _knopf(laeufe) geraete = [_geraet(g, laufend, fertig, knopf, jetzt) for g in sorted(gaeste, key=lambda g: g["vmid"])] alle = [s for g in geraete for s in g["snapshots"]] sicherungen = [e for g in geraete for e in g["sicherungen"]] empfangen = (eingang or {}).get("empfangen") rueckwege = _rueckwege(geraete) return { "jetzt": jetzt, "bericht": {"empfangen": empfangen, "veraltet": jetzt - empfangen > inventar.BERICHT_ALT_S} if eingang else None, "ausfuehrer": {"verbunden": bool(zuletzt and jetzt - zuletzt < inventar.BERICHT_ALT_S), "zuletzt": zuletzt, "nur_lesen": bool((bericht.get("host") or {}).get("nur_lesen"))}, # Warum gerade keine Aktion geht („Alle aktualisieren“, ein Update, der Ausführer) — wie die Knöpfe es sagten. "sperre": _sperre(), # Ausführer vor dem 25.09.2026: nur Namen, Zeitpunkte aus dem Namen, keine Beschreibung. "details": any("snapshot_details" in g for g in gaeste), "geraete": geraete, "summe": {"snapshots": len(alle), "eigene": sum(1 for s in alle if s["eigen"]), "sicherungen": len(sicherungen), "sicherungen_groesse": sum(e["groesse"] or 0 for e in sicherungen) or None}, # „Alle Update-Rückwege löschen“: nur, wenn es welche gibt und das Aufräumen nicht schon läuft. "aufraeumen": (_aktion("/api/homelab/snapshots/rueckwege-loeschen", _frage_rueckwege(rueckwege), f"Alle {len(rueckwege)} Update-Rückwege löschen") if rueckwege and not _AUFRAEUMEN.is_set() else None), "aufraeumen_laeuft": _AUFRAEUMEN.is_set(), } # --- Alle Update-Rückwege auf einmal (User-Klick) --------------------------------------------------------------------- _AUFRAEUMEN = threading.Event() # gesetzt, solange rueckwege_loeschen seine Liste abarbeitet AUFRAEUMEN_ZEITLIMIT_S = 10 * 60 # je Löschung; der Ausführer braucht Sekunden def _rueckwege(geraete: list[dict]) -> list[tuple[str, str, str, str]]: """(Gerät, Name, Art, Eintrag) aller Rückwege von Updates, die sich löschen lassen: Snapshots mit Herkunft „update“ und die Sicherungen des Orchestrators. Per Knopf oder von Hand angelegte gehören nicht dazu.""" liste = [] for g in geraete: liste += [(g["id"], g["name"], "snapshot", s["name"]) for s in g["snapshots"] if s["herkunft"] == "update" and s.get("loeschen")] liste += [(g["id"], g["name"], "sicherung", e["datei"]) for e in g["sicherungen"] if e.get("loeschen")] return liste def _frage_rueckwege(rueckwege: list[tuple[str, str, str, str]]) -> str: snaps = sum(1 for r in rueckwege if r[2] == "snapshot") sich = len(rueckwege) - snaps teile = [] if snaps: teile.append(f"{snaps} {'Snapshot' if snaps == 1 else 'Snapshots'}") if sich: teile.append(f"{sich} {'Sicherung' if sich == 1 else 'Sicherungen'}") geraete = ", ".join(dict.fromkeys(r[1] for r in rueckwege)) return (f"{' und '.join(teile)} endgültig löschen ({geraete})? Das sind die Rückwege, die der Orchestrator vor den " "letzten Updates angelegt hat. An den Geräten ändert sich nichts; nur der Weg zurück auf die Stände vor " "diesen Updates ist danach weg. Snapshots, die du hier per Knopf oder von Hand in Proxmox angelegt hast, " "bleiben. Gelöscht wird nacheinander; am Ende kommt eine Meldung.") def _warten_auf(lauf_id: str) -> dict | None: ende = time.monotonic() + AUFRAEUMEN_ZEITLIMIT_S while time.monotonic() < ende: lauf = updates.lauf_lesen(lauf_id) if lauf and lauf.get("status") != "laeuft": return lauf time.sleep(ABFRAGE_TAKT_S) return None ABFRAGE_TAKT_S = 1.0 def _rueckwege_abarbeiten(rueckwege: list[tuple[str, str, str, str]]) -> None: geloescht, fehler = 0, [] try: for zid, name, art, eintrag in rueckwege: antwort = loeschen(zid, eintrag) if art == "snapshot" else sicherung_loeschen(zid, eintrag) if not antwort.get("ok"): fehler.append(f"{name} {eintrag}: {antwort.get('detail')}") continue lauf = _warten_auf(antwort["lauf"]) if lauf and lauf.get("ergebnis") == "geloescht": geloescht += 1 else: fehler.append(f"{name} {eintrag}: {(lauf or {}).get('text') or 'keine Antwort'}") finally: _AUFRAEUMEN.clear() text = f"Update-Rückwege aufgeräumt: {geloescht} von {len(rueckwege)} gelöscht." if fehler: text += " Nicht gelöscht: " + "; ".join(fehler[:5]) try: announce.notify_telegram(BETREFF_ALARM if fehler else BETREFF, text) except Exception: log.warning("homelab: Meldung zum Aufräumen ging nicht raus", exc_info=True) def rueckwege_loeschen() -> dict: """Knopf „Alle Update-Rückwege löschen“: startet das Abarbeiten im eigenen Faden und kehrt sofort zurück.""" if _AUFRAEUMEN.is_set(): return {"ok": False, "detail": "Das Aufräumen läuft schon."} if grund := _sperre(): return {"ok": False, "detail": grund} rueckwege = _rueckwege(liste()["geraete"]) if not rueckwege: return {"ok": False, "detail": "Es gibt keine Rückwege von Updates zu löschen."} _AUFRAEUMEN.set() threading.Thread(target=_rueckwege_abarbeiten, args=(rueckwege,), name="homelab-rueckwege", daemon=True).start() return {"ok": True, "anzahl": len(rueckwege)} # --- Aktionen ------------------------------------------------------------------------------------------------------ def _sperre() -> str | None: """Warum gerade gar keine Aktion dieser Seite geht — oder None. Derselbe Satz steht auf der Seite über der Liste. Läuft irgendwo ein Update, warten alle: Der Ausführer arbeitet einen Auftrag nach dem anderen ab. Hinter einem Update (bis zu einer Stunde) liefe eine Aktion in ihr Zeitlimit — und käme danach doch noch dran, ohne dass ein Lauf hinsieht; beim Zurücksetzen wäre das gefährlich. Vor Aktionen dieser Seite stehen so nur kurze Aufträge. Umgekehrt darf ein Update neben ihnen starten: Es wartet höchstens ein paar Sekunden.""" if sammellauf.laeuft(): return f"{updates.SAMMELLAUF_SPERRE} Bis es fertig ist, ruhen die Knöpfe hier." im_update = [str(x.get("name") or x.get("ziel")) for x in updates.laeufe(100) if x.get("status") == "laeuft" and x.get("baustein") not in LAEUFT] if im_update: return (f"Gerade läuft ein Update ({', '.join(dict.fromkeys(im_update))}). Anlegen, löschen und zurücksetzen " "geht, sobald es fertig ist — der Ausführer arbeitet einen Auftrag nach dem anderen ab.") zuletzt = kanal.zuletzt() if not (zuletzt and time.time() - zuletzt < inventar.BERICHT_ALT_S): return ("Der Ausführer auf dem Proxmox-Host meldet sich nicht. Die Liste zeigt den letzten bekannten Stand; " "anlegen, löschen und zurücksetzen geht erst, wenn er wieder da ist.") if (((kanal.bericht() or {}).get("bericht") or {}).get("host") or {}).get("nur_lesen"): return "Der Ausführer läuft im Nur-Lesen-Modus: Er zeigt alles, ändert aber nichts." return None def _vorab(ziel_id: str) -> tuple[dict | None, str | None]: """Was vor jeder Aktion gelten muss (unter updates.START_SPERRE geprüft): (Gast, None) oder (None, warum nicht).""" if grund := _sperre(): return None, grund if updates.laeuft(ziel_id): return None, ("Für dieses Gerät läuft gerade schon etwas (ein Update oder eine Snapshot-Aktion). Das geht " "erst, wenn es fertig ist.") gast = inventar.gast(ziel_id) if not gast or inventar.EIGENES_ETIKETT in (gast.get("etiketten") or []): return None, "Dieses Gerät steht nicht im Bericht des Ausführers." if not gast.get("erlaubt"): return None, "Dieses Gerät ist nicht freigegeben (Etikett „watcher“ fehlt); der Ausführer fasst es nicht an." return gast, None def _starten(ziel_id: str, baustein: str, vorbereiten: Callable[[dict], tuple[str | None, dict]], schritte: Callable[[dict, dict], None]) -> dict: """Prüfen und den Lauf anlegen am Stück (wie updates.starten), dann im eigenen Faden ablaufen lassen. Rückgabe {"ok": True, "lauf": id} oder {"ok": False, "detail": …}.""" with updates.START_SPERRE: gast, grund = _vorab(ziel_id) mehr: dict = {} if gast is not None: grund, mehr = vorbereiten(gast) if grund or gast is None: return {"ok": False, "detail": grund or "Das geht gerade nicht."} lauf = updates._neuer_lauf(ziel_id, baustein, _name(gast), **mehr) threading.Thread(target=_ablauf, args=(schritte, lauf, gast), name=f"homelab-{ziel_id}", daemon=True).start() return {"ok": True, "lauf": lauf["id"]} def _eigener_snapshot(gast: dict, name: str) -> tuple[str | None, dict]: if not SNAPSHOT_NAME.match(name): return ("Nur Snapshots des Orchestrators (mc2-…) lassen sich hier löschen oder zurücksetzen; von Hand " "angelegte verwaltest du in Proxmox."), {} s = next((s for s in _snapshots(gast, "", set()) if s["name"] == name), None) if s is None: return f"Den Snapshot {name} gibt es laut dem letzten Bericht nicht (mehr).", {} return None, {"snapshot": name, "snapshot_zeit": s["zeit"]} def anlegen(ziel_id: str) -> dict: """Knopf „Snapshot anlegen“: nur freigegebene Gäste, auf deren Speicher ein Snapshot geht.""" return _starten(ziel_id, "snapshot-anlegen", lambda g: (_anlegen_grund(g), {}), _anlegen) def loeschen(ziel_id: str, name: str) -> dict: """Knopf „Löschen“ an einem Snapshot des Orchestrators.""" return _starten(ziel_id, "snapshot-loeschen", lambda g: _eigener_snapshot(g, name), _loeschen) def zuruecksetzen(ziel_id: str, name: str) -> dict: """Knopf „Zurücksetzen“ an einem Snapshot des Orchestrators.""" return _starten(ziel_id, "snapshot-zurueck", lambda g: _eigener_snapshot(g, name), _zurueck) def sicherung_loeschen(ziel_id: str, datei: str) -> dict: """Knopf „Löschen“ an einer Sicherung des Orchestrators (nur, was der Bericht als eigene nennt).""" def vorbereiten(gast: dict) -> tuple[str | None, dict]: volid = next((str(v) for v in gast.get("sicherungen") or [] if str(v).rsplit("/", 1)[-1] == datei), None) if volid is None: return f"Die Sicherung {datei} gibt es laut dem letzten Bericht nicht (mehr).", {} return None, {"sicherung": volid} return _starten(ziel_id, "sicherung-loeschen", vorbereiten, _sicherung_loeschen) # --- Abläufe (im eigenen Faden) -------------------------------------------------------------------------------------- def _ende(lauf: dict, ergebnis: str, text: str, melden: bool = False, dringend: bool = False) -> None: betreff = (BETREFF_ALARM if dringend else BETREFF) if melden else None lauf.update(status="fertig", ergebnis=ergebnis, text=text, ende=time.time(), dringend=dringend, gemeldet=melden, betreff=betreff) updates._merken(lauf) if betreff: try: announce.notify_telegram(betreff, text) except Exception: log.warning("homelab: Meldung zu %s ging nicht raus", lauf.get("baustein"), exc_info=True) def _ablauf(schritte: Callable[[dict, dict], None], lauf: dict, gast: dict) -> None: try: schritte(lauf, gast) except Exception as exc: # Sonst hieße der Lauf bis zum nächsten Neustart „läuft“, und das Gerät wäre für alles gesperrt. log.exception("homelab: %s für %s ist abgestürzt", lauf.get("baustein"), lauf.get("ziel")) _ende(lauf, "fehler", f"{lauf['name']}: {WAS.get(lauf['baustein'], 'Die Aktion')} ist mit einem internen " f"Fehler stehen geblieben ({exc.__class__.__name__}).", melden=True) def _anlegen(lauf: dict, gast: dict) -> None: vmid, name = gast["vmid"], lauf["name"] try: a = updates._auftrag(lauf, "snapshot", {"vmid": vmid, "anlass": "knopf"}, updates.ZEITLIMIT_KURZ_S) except RuntimeError as exc: _ende(lauf, "fehler", f"{name}: Snapshot nicht angelegt — {_grund(exc)}.", melden=True) return m = re.search(r"snapshot=(mc2-\d{8}-\d{6})", a.get("text") or "") lauf["snapshot"] = m.group(1) if m else None nachher = updates._frischer_gast(lauf, vmid) if lauf["snapshot"] and nachher is not None and lauf["snapshot"] not in (nachher.get("snapshots") or []): _ende(lauf, "fehler", f"{name}: Der Ausführer meldete den Snapshot {lauf['snapshot']}, im neuen Bericht fehlt " "er aber.", melden=True) return _ende(lauf, "angelegt", f"{name}: Snapshot {lauf['snapshot']} angelegt." if lauf["snapshot"] else f"{name}: Snapshot angelegt.") def _loeschen(lauf: dict, gast: dict) -> None: vmid, name, snapshot = gast["vmid"], lauf["name"], lauf["snapshot"] try: updates._auftrag(lauf, "snapshot_loeschen", {"vmid": vmid, "snapshot": snapshot}, updates.ZEITLIMIT_KURZ_S) except RuntimeError as exc: _ende(lauf, "fehler", f"{name}: Snapshot {snapshot} ließ sich nicht löschen — {_grund(exc)}.", melden=True) return updates._frischer_gast(lauf, vmid) _ende(lauf, "geloescht", f"{name}: Snapshot {snapshot} gelöscht.") def _lage(name: str, gast: dict | None) -> str: """Wie es dem Gerät nach einem gescheiterten Zurücksetzen geht, laut frischem Bericht.""" if gast is None: return f"Ob {name} läuft, ließ sich nicht prüfen — bitte in Proxmox nachsehen." if gast.get("status") == "running": return f"{name} läuft." return f"{name} läuft nicht ({gast.get('status')}) — bitte in Proxmox nachsehen." def _zurueck(lauf: dict, gast: dict) -> None: vmid, name, snapshot = gast["vmid"], lauf["name"], lauf["snapshot"] wann = _wann(lauf.get("snapshot_zeit")) stand = f"den Snapshot {wann} ({snapshot})" if wann else f"den Snapshot {snapshot}" try: updates._auftrag(lauf, "zurueck", {"vmid": vmid, "snapshot": snapshot}, updates.ZEITLIMIT_KURZ_S) except RuntimeError as exc: nachher = updates._frischer_gast(lauf, vmid) _ende(lauf, "fehler", f"{name}: Zurücksetzen auf {stand} gescheitert — {_grund(exc)}. {_lage(name, nachher)}", melden=True, dringend=True) return lauf["zurueckgesetzt"] = True time.sleep(updates.WARTEN_NACH_UPDATE_S) nachher = updates._frischer_gast(lauf, vmid) fehler = updates.pruefen(gast, nachher, "zurueck") if fehler: _ende(lauf, "fehler", f"{name}: auf {stand} zurückgesetzt, aber die Prüfung ist rot — {' '.join(fehler)}", melden=True, dringend=True) return _ende(lauf, "zurueckgesetzt", f"{name}: auf {stand} zurückgesetzt, Prüfung grün.", melden=True) def _sicherung_loeschen(lauf: dict, gast: dict) -> None: vmid, name, volid = gast["vmid"], lauf["name"], lauf["sicherung"] datei = volid.rsplit("/", 1)[-1] try: updates._auftrag(lauf, "sicherung_loeschen", {"vmid": vmid, "sicherung": volid}, updates.ZEITLIMIT_KURZ_S) except RuntimeError as exc: _ende(lauf, "fehler", f"{name}: Sicherung {datei} ließ sich nicht löschen — {_grund(exc)}.", melden=True) return updates._frischer_gast(lauf, vmid) _ende(lauf, "geloescht", f"{name}: Sicherung {datei} gelöscht.")