HitonabiandClaude Opus 5.5
0df593c05a
Der erste Bericht nach dem Neustart vom 25.09. kam 34 s nach dem Hochfahren – Proxmox kannte die Gäste noch nicht („unknown“), und zehn Minuten lang zeigte die Oberfläche jeden Gast als unbekannt. Enthält ein Bericht solche Gäste, folgt der nächste nach 60 s. Der Grund steht jetzt auf Deutsch da. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
334 lines
17 KiB
Python
334 lines
17 KiB
Python
"""Vom Bericht des Ausführers zu Zielen im gemeinsamen Modell (kern/ziele.py) — Phase 3, 24.09.2026.
|
|
|
|
Je Gerät ein Ziel: der Proxmox-Host (Pakete, Neustart) und jeder freigegebene Gast (Etikett
|
|
community-script oder watcher, nicht watcher-aus) mit seinen Bausteinen „App“ (Community-Script,
|
|
Vergleich mit der neuesten Veröffentlichung auf GitHub) und „Pakete“ (Betriebssystem im Gast).
|
|
Dazu prüft dieser Teil selbst, ob die Weboberfläche jedes Gasts antwortet.
|
|
|
|
Wie ehrlich der Stand ist, steht dabei: Sind die Paketlisten im Gast älter als zwei Wochen, heißt es
|
|
„unbekannt“ (mit Knopf zum Suchen) statt „aktuell“; ist der Bericht älter als eine halbe Stunde, sagt
|
|
die Übersicht, dass der Ausführer schweigt. Ist das Update-Skript einer App frisch geändert, bleibt sie
|
|
„neu“, aber ohne Knopf (karenz.py). Rückweg und Rückfrage sagen, was vor dem Update geschieht: Snapshot,
|
|
wo keiner geht eine Sicherung, sonst keiner. Ohne Freigabe (Etikett) legt der Ausführer weder das eine noch
|
|
das andere an — das betrifft Arcane, dessen Docker-Updates auch ohne Etikett über Arcane selbst laufen.
|
|
"""
|
|
|
|
import re
|
|
import threading
|
|
import time
|
|
from datetime import datetime
|
|
|
|
import httpx
|
|
from kern.github import neueste_version
|
|
from kern.zeit import LOCAL_TZ
|
|
from kern.ziele import Aktion, BausteinStand, ZielStand
|
|
|
|
from services.homelab import apps, arcane, kanal, karenz
|
|
|
|
BERICHT_ALT_S = 30 * 60
|
|
LISTEN_ALT_S = 14 * 24 * 3600
|
|
EIGENES_ETIKETT = "mc2"
|
|
ERREICHBAR_CACHE_S = 60
|
|
OHNE_FREIGABE = "Das Etikett „watcher“ fehlt, der Ausführer darf hier keinen Snapshot anlegen"
|
|
|
|
_erreichbar_cache: dict[str, tuple[float, bool]] = {}
|
|
_erreichbar_lock = threading.Lock()
|
|
|
|
|
|
def erreichbar(url: str) -> bool:
|
|
"""Antwortet die Weboberfläche? Alles unter 500 zählt (eine Anmeldeseite ist eine Antwort)."""
|
|
jetzt = time.time()
|
|
with _erreichbar_lock:
|
|
if (eintrag := _erreichbar_cache.get(url)) and jetzt - eintrag[0] < ERREICHBAR_CACHE_S:
|
|
return eintrag[1]
|
|
try:
|
|
# Selbstsignierte Zertifikate sind im Heimnetz üblich (PBS, NPMplus); geprüft wird nur, ob jemand antwortet.
|
|
ok = httpx.get(url, timeout=5.0, verify=False, follow_redirects=False).status_code < 500
|
|
except httpx.HTTPError:
|
|
ok = False
|
|
with _erreichbar_lock:
|
|
_erreichbar_cache[url] = (jetzt, ok)
|
|
return ok
|
|
|
|
|
|
def erreichbar_frisch(url: str) -> bool:
|
|
"""Wie erreichbar(), aber ohne den Zwischenspeicher — für die Prüfung direkt nach einem Update."""
|
|
with _erreichbar_lock:
|
|
_erreichbar_cache.pop(url, None)
|
|
return erreichbar(url)
|
|
|
|
|
|
def _teile(version: str) -> tuple[int, ...]:
|
|
return tuple(int(x) for x in re.findall(r"\d+", version.split("-")[0])[:4])
|
|
|
|
|
|
def neuer(verfuegbar: str, installiert: str) -> bool | None:
|
|
"""Ist verfuegbar neuer als installiert? None, wenn sich das nicht sagen lässt."""
|
|
a, b = _teile(verfuegbar), _teile(installiert)
|
|
if not a or not b:
|
|
return None
|
|
return a > b
|
|
|
|
|
|
def _datum(ts: float | None) -> str:
|
|
return datetime.fromtimestamp(ts, LOCAL_TZ).strftime("%d.%m.%Y") if ts else "?"
|
|
|
|
|
|
def _rueckweg(gast: dict) -> str:
|
|
"""Die Zeile „Rückweg“ der Karte: was vor jedem Update geschieht."""
|
|
if not gast.get("erlaubt"):
|
|
return f"Kein automatischer Rückweg: {OHNE_FREIGABE}."
|
|
if gast.get("snapshot_moeglich"):
|
|
return "Snapshot vor jedem Update"
|
|
grund = gast.get("snapshot_grund") or "kein Snapshot möglich"
|
|
if gast.get("sicherung_moeglich"):
|
|
return f"Sicherung auf „{gast.get('sicherung_speicher') or 'dem Host'}“ vor jedem Update ({grund})"
|
|
if gast.get("sicherung_grund"):
|
|
return f"Kein automatischer Rückweg: {grund}. Auch keine Sicherung: {gast['sicherung_grund']}"
|
|
return grund
|
|
|
|
|
|
def _rueckweg_art(gast: dict) -> str:
|
|
"""Dasselbe in einem Wort — die Oberfläche fasst den Rückweg in der Update-Liste kurz."""
|
|
if not gast.get("erlaubt"):
|
|
return "keiner"
|
|
if gast.get("snapshot_moeglich"):
|
|
return "snapshot"
|
|
return "sicherung" if gast.get("sicherung_moeglich") else "keiner"
|
|
|
|
|
|
def _rueckweg_frage(gast: dict) -> str:
|
|
"""Der Satz zum Rückweg in der Rückfrage vor „Jetzt updaten“."""
|
|
if not gast.get("erlaubt"):
|
|
return f"{OHNE_FREIGABE} — scheitert das Update, gibt es keinen automatischen Rückweg."
|
|
if gast.get("snapshot_moeglich"):
|
|
return "Vorher wird ein Snapshot angelegt; ist die Prüfung danach rot, geht es von selbst zurück."
|
|
if gast.get("sicherung_moeglich"):
|
|
return "Vorher wird eine Sicherung angelegt; ist die Prüfung danach rot, wird sie zurückgespielt."
|
|
satz = "Ein Snapshot ist hier nicht möglich — scheitert das Update, gibt es keinen automatischen Rückweg."
|
|
if gast.get("sicherung_grund"):
|
|
satz += f" Eine Sicherung geht auch nicht: {gast['sicherung_grund']}"
|
|
return satz
|
|
|
|
|
|
def _app_baustein(gast: dict, app: apps.App, zid: str) -> BausteinStand:
|
|
installiert = (gast.get("app") or {}).get("version")
|
|
stand = BausteinStand(id="app", name=app.name, zustand="aktuell", installiert=installiert)
|
|
if not app.quelle:
|
|
return stand
|
|
try:
|
|
neu = neueste_version(app.quelle)
|
|
except Exception:
|
|
neu = None
|
|
if not neu:
|
|
stand.zustand, stand.grund = "unbekannt", "Die neueste Version ist gerade nicht abrufbar (GitHub)."
|
|
return stand
|
|
stand.verfuegbar = neu["version"]
|
|
if installiert and installiert.startswith("untagged-"):
|
|
# Manche Projekte veröffentlichen ohne Versions-Tag; die App kennt dann nur dieses Tag (PVE Scripts Local).
|
|
if installiert == neu.get("tag_roh"):
|
|
stand.installiert = neu["version"]
|
|
return stand
|
|
stand.zustand, stand.grund = "unbekannt", "Die installierte Fassung lässt sich keiner Version zuordnen."
|
|
return stand
|
|
if app.vergleich == "datum":
|
|
# Docker-Image ohne Versionsnummer: neuer, wenn die Veröffentlichung jünger ist als das Image.
|
|
bild = re.search(r"\d{4}-\d{2}-\d{2}", installiert or "")
|
|
ist_neuer = (neu["datum"] or "") > bild.group(0) if bild else None
|
|
else:
|
|
ist_neuer = neuer(neu["version"], installiert) if installiert else None
|
|
if ist_neuer is None:
|
|
stand.zustand, stand.grund = "unbekannt", "Die installierte Version ließ sich nicht lesen."
|
|
elif ist_neuer and app.weg != "skript":
|
|
stand.zustand, stand.kurz = "neu", f"{installiert} → {neu['version']}"
|
|
stand.grund = f"{app.name} aktualisiert sich über die eigene Oberfläche; das Community-Script tut es nicht."
|
|
elif ist_neuer:
|
|
stand.zustand, stand.kurz = "neu", f"{installiert} → {neu['version']}"
|
|
# Frisch geänderte Update-Skripte erst nach der Wartezeit (karenz.py): „neu“, aber ohne Knopf.
|
|
wartezeit = karenz.pruefen(app.kennung)
|
|
if wartezeit["gesperrt"]:
|
|
stand.grund = wartezeit["gesperrt"]
|
|
return stand
|
|
frage = f"{app.name} jetzt aktualisieren? {_rueckweg_frage(gast)}"
|
|
if not wartezeit["geprueft"]:
|
|
frage += f" {karenz.NICHT_GEPRUEFT}"
|
|
stand.aktion = Aktion("POST", f"/api/homelab/ziele/{zid}/update", frage)
|
|
return stand
|
|
|
|
|
|
def _os_baustein(gast: dict, zid: str, jetzt: float) -> BausteinStand:
|
|
os_stand = gast.get("os_updates") or {}
|
|
stand = BausteinStand(id="os", name="Pakete", zustand="aktuell")
|
|
anzahl, listen = os_stand.get("anzahl"), os_stand.get("listen_stand")
|
|
suchen = Aktion("POST", f"/api/homelab/ziele/{zid}/suchen", "Die Paketlisten im Gast jetzt erneuern?",
|
|
label="Nach Updates suchen")
|
|
if anzahl is None:
|
|
stand.zustand, stand.grund = "unbekannt", "Die Pakete ließen sich nicht lesen."
|
|
elif listen and jetzt - listen > LISTEN_ALT_S:
|
|
stand.zustand, stand.aktion = "unbekannt", suchen
|
|
stand.grund = f"Die Paketlisten sind vom {_datum(listen)}; was fehlt, weiß erst eine neue Suche."
|
|
elif anzahl > 0:
|
|
stand.zustand, stand.kurz = "neu", f"{anzahl} Pakete"
|
|
stand.aktion = Aktion("POST", f"/api/homelab/ziele/{zid}/os-update",
|
|
f"{anzahl} Pakete im Gast jetzt einspielen? {_rueckweg_frage(gast)}")
|
|
return stand
|
|
|
|
|
|
def _arcane_bausteine(basis: str | None, zid: str, gast: dict) -> list[BausteinStand]:
|
|
"""Arcane selbst (öffentliche Version) und die Docker-Images (mit API-Schlüssel)."""
|
|
app = arcane.app_version(basis) if basis else None
|
|
selbst = BausteinStand(id="arcane", name="Arcane", zustand="unbekannt",
|
|
installiert=app["installiert"] if app else None, verfuegbar=app["neu"] if app else None)
|
|
if not app:
|
|
selbst.grund = "Arcane antwortet nicht."
|
|
elif app["update"]:
|
|
selbst.zustand, selbst.kurz = "neu", f"{app['installiert']} → {app['neu']}"
|
|
selbst.grund = "Arcane spielt sein eigenes Update über die eigene Oberfläche ein (Einstellungen)."
|
|
else:
|
|
selbst.zustand = "aktuell"
|
|
docker = BausteinStand(id="docker", name="Docker-Images", zustand="unbekannt")
|
|
if not arcane.schluessel():
|
|
docker.grund = ("Es fehlt ein Arcane-API-Schlüssel: Ohne ihn sieht der Orchestrator die Images nicht. "
|
|
"Eintragen unter Einstellungen (Arcane-API-Schlüssel).")
|
|
elif basis:
|
|
try:
|
|
images = arcane.images_mit_update(basis)
|
|
except Exception:
|
|
docker.grund = "Arcane verweigert die Auskunft oder antwortet nicht (Schlüssel prüfen)."
|
|
else:
|
|
# Nur Images, die ein Container benutzt: Arcanes Updater tauscht Container aus. Ein veraltetes Image ohne
|
|
# Container (Basis-Images für Builds, CI) bliebe sonst für immer „neu“ (25.09.2026: 9 von 11 nach dem Lauf).
|
|
ohne = [i for i in images if not i.get("benutzt")]
|
|
images = [i for i in images if i.get("benutzt")]
|
|
if ohne:
|
|
docker.grund = (f"{len(ohne)} ältere Images ohne Container ({', '.join(i['name'] for i in ohne[:4])}"
|
|
+ (" …" if len(ohne) > 4 else "") + ") betreffen nichts; aufräumen in Arcane unter Images.")
|
|
if images:
|
|
docker.zustand, docker.kurz = "neu", f"{len(images)} Images"
|
|
docker.grund = ", ".join(i["name"] for i in images[:6]) + (" …" if len(images) > 6 else "")
|
|
probe = not arcane.echt()
|
|
docker.aktion = Aktion(
|
|
"POST", f"/api/homelab/ziele/{zid}/docker",
|
|
("Arcane prüft im Probelauf, welche Container es aktualisieren würde — geändert wird noch nichts."
|
|
if probe else "Arcane aktualisiert jetzt die Container mit neuen Images; Daten-Volumes bleiben. "
|
|
+ _rueckweg_frage(gast)),
|
|
label="Probelauf" if probe else "Jetzt updaten")
|
|
else:
|
|
docker.zustand = "aktuell"
|
|
return [selbst, docker]
|
|
|
|
|
|
def _status_text(status: object) -> str:
|
|
"""Warum ein Gast keine Bausteine hat — auf Deutsch. „unknown“ kommt kurz nach dem Hochfahren des Hosts, bevor
|
|
Proxmox den Zustand kennt (der Ausführer berichtet dann nach einer Minute erneut)."""
|
|
if status == "stopped":
|
|
return "Der Gast ist gestoppt."
|
|
if status == "paused":
|
|
return "Der Gast ist angehalten."
|
|
return "Proxmox kennt seinen Zustand gerade nicht (etwa kurz nach dem Hochfahren des Hosts)."
|
|
|
|
|
|
def _gast_ziel(gast: dict, jetzt: float) -> ZielStand:
|
|
art = "container" if gast["art"] == "lxc" else "vm"
|
|
zid = f"{'ct' if art == 'container' else 'vm'}-{gast['vmid']}"
|
|
app = apps.app_fuer((gast.get("app") or {}).get("kennung"), gast.get("name"))
|
|
url = apps.adresse(app, gast.get("ip"))
|
|
laeuft = gast.get("status") == "running"
|
|
ziel = ZielStand(id=zid, name=app.name if app else str(gast.get("name") or zid), art=art, instanz="homelab",
|
|
erreichbar=(erreichbar(url) if laeuft else False) if url else (laeuft or None),
|
|
rueckweg=_rueckweg(gast), rueckweg_art=_rueckweg_art(gast), platte=gast.get("platte"))
|
|
if not laeuft:
|
|
ziel.bausteine.append(BausteinStand(id="status", name="Status", zustand="unbekannt",
|
|
grund=_status_text(gast.get("status"))))
|
|
return ziel
|
|
if app and app.kennung == "arcane":
|
|
# Arcane läuft über seine eigene Schnittstelle, nicht über den Ausführer: kein Etikett nötig — nur für den
|
|
# Snapshot vorher (Rückweg), den legt der Ausführer an.
|
|
ziel.bausteine += _arcane_bausteine(url.rstrip("/") if url else None, zid, gast)
|
|
return ziel
|
|
if not gast.get("erlaubt"):
|
|
ziel.bausteine.append(BausteinStand(
|
|
id="freigabe", name="Freigabe", zustand="unbekannt",
|
|
grund="Nicht freigegeben: Das Etikett „watcher“ fehlt (oder „watcher-aus“ ist gesetzt)."))
|
|
return ziel
|
|
if app and app.kennung in apps.KATALOG:
|
|
ziel.bausteine.append(_app_baustein(gast, app, zid))
|
|
if gast["art"] == "lxc":
|
|
ziel.bausteine.append(_os_baustein(gast, zid, jetzt))
|
|
return ziel
|
|
|
|
|
|
def _host_ziel(host: dict) -> ZielStand:
|
|
ziel = ZielStand(id="pve", name="Proxmox-Host", art="proxmox-host", instanz="homelab", erreichbar=True,
|
|
rueckweg="Kein Snapshot möglich; der Host bleibt beim Update in Betrieb, neu gestartet wird getrennt.",
|
|
rueckweg_art="keiner")
|
|
updates = host.get("updates")
|
|
pakete = BausteinStand(id="pakete", name="Proxmox VE und Debian", zustand="aktuell",
|
|
installiert=host.get("version"))
|
|
if updates is None:
|
|
pakete.zustand, pakete.grund = "unbekannt", host.get("fehler") or "Die Update-Liste fehlt im Bericht."
|
|
elif updates:
|
|
pakete.zustand, pakete.kurz = "neu", f"{len(updates)} Pakete"
|
|
pakete.aktion = Aktion("POST", "/api/homelab/ziele/pve/update",
|
|
f"{len(updates)} Pakete auf dem Proxmox-Host einspielen? Alle Gäste laufen weiter. "
|
|
"Einen Neustart löst das nicht aus — der ist ein eigener Knopf.")
|
|
ziel.bausteine.append(pakete)
|
|
if host.get("neustart_noetig"):
|
|
ziel.bausteine.append(BausteinStand(
|
|
id="neustart", name="Neustart", zustand="neu", kurz="steht an",
|
|
installiert=host.get("kernel"), verfuegbar=host.get("kernel_neu"),
|
|
grund=(f"Neuer Kernel {host['kernel_neu']} ist installiert, es läuft noch {host.get('kernel')}."
|
|
if host.get("kernel_neu") else "Ein Update verlangt einen Neustart."),
|
|
aktion=Aktion("POST", "/api/homelab/ziele/pve/neustart",
|
|
"Den Proxmox-Host jetzt neu starten? ALLE Container und die Arcane-VM sind dann ein paar "
|
|
"Minuten weg. Die KI-Box prüft danach von außen, ob alles wiederkommt.",
|
|
label="Jetzt neu starten")))
|
|
return ziel
|
|
|
|
|
|
def ziele() -> dict:
|
|
"""Alle Ziele des Homelabs aus dem letzten Bericht, dazu wie frisch er ist."""
|
|
eingang = kanal.bericht()
|
|
jetzt = time.time()
|
|
zuletzt = kanal.zuletzt()
|
|
ausfuehrer = {"verbunden": bool(zuletzt and jetzt - zuletzt < BERICHT_ALT_S), "zuletzt": zuletzt}
|
|
if not eingang:
|
|
return {"ziele": [], "bericht": None, "ausfuehrer": ausfuehrer}
|
|
b = eingang["bericht"]
|
|
liste = [_host_ziel(b.get("host") or {})]
|
|
# Der Container des Orchestrators selbst (Etikett „mc2“) ist kein Gerät, das er pflegt.
|
|
liste += [_gast_ziel(g, jetzt) for g in b.get("gaeste") or [] if EIGENES_ETIKETT not in (g.get("etiketten") or [])]
|
|
for z in liste:
|
|
z.geprueft = eingang["empfangen"]
|
|
return {"ziele": [z.als_dict() for z in liste],
|
|
"bericht": {"empfangen": eingang["empfangen"], "veraltet": jetzt - eingang["empfangen"] > BERICHT_ALT_S},
|
|
"ausfuehrer": ausfuehrer}
|
|
|
|
|
|
def erreichbarkeit() -> list[dict]:
|
|
"""Für den Wächter: jede Weboberfläche laufender, freigegebener Gäste — ohne GitHub-Abfragen. Gäste mitten in
|
|
einem Update-Lauf bleiben außen vor: Sicherung, Update und Zurückspielen legen sie kurz still, und was dabei
|
|
herauskommt, meldet der Lauf selbst (sonst käme ein zweiter Alarm samt „wieder ok“)."""
|
|
from services.homelab import updates # updates importiert dieses Modul
|
|
eingang = kanal.bericht()
|
|
im_update = updates.laufende_ziele()
|
|
ergebnis = []
|
|
for g in (eingang or {}).get("bericht", {}).get("gaeste") or []:
|
|
app = apps.app_fuer((g.get("app") or {}).get("kennung"), g.get("name"))
|
|
url = apps.adresse(app, g.get("ip"))
|
|
zid = f"{'ct' if g['art'] == 'lxc' else 'vm'}-{g['vmid']}"
|
|
if url and g.get("erlaubt") and zid not in im_update:
|
|
ergebnis.append({"id": zid, "name": app.name if app else g.get("name"), "url": url,
|
|
"laeuft": g.get("status") == "running",
|
|
"ok": g.get("status") == "running" and erreichbar(url)})
|
|
return ergebnis
|
|
|
|
|
|
def gast(zid: str) -> dict | None:
|
|
"""Der Gast zu einer Ziel-ID aus dem letzten Bericht (ct-104 → vmid 104)."""
|
|
m = re.fullmatch(r"(ct|vm)-(\d+)", zid)
|
|
eingang = kanal.bericht()
|
|
if not m or not eingang:
|
|
return None
|
|
return next((g for g in eingang["bericht"].get("gaeste") or [] if g.get("vmid") == int(m.group(2))), None)
|