HitonabiandClaude Opus 5.5
6af42bd42c
Statt „Arcane (Docker) · 11 Images“ zeigt das Homelab-Cockpit je Stack jeden Container mit Zustand, Healthcheck und Update-Stand („Update verfügbar“, „lokal gebaut“ für NerdQuiz und Rippy, „aktuell“). Der Wächter meldet gelb, wenn ein Container ungesund ist oder ständig neu startet; beendete Container sind kein Befund. Neue Stacks erscheinen unter „Neu entdeckt“ (eigenes Register, nur „Übernehmen“), Lucy meldet sie einmal. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
92 lines
4.8 KiB
Python
92 lines
4.8 KiB
Python
"""Docker-Dienste unter Arcane einzeln (Ausbauplan Welle 2, Punkt 8, seit 25.09.2026).
|
|
|
|
Bis zum 25.09. stand im Orchestrator nur „Arcane (Docker) · 11 Images“. Jetzt fragt er Arcane nach jedem Container
|
|
(GET /environments/{id}/containers, mit dem Schlüssel aus den Einstellungen, höchstens alle zwei Minuten) und zeigt je
|
|
Stack (Compose-Projekt) Zustand, Gesundheit, Image und Update-Stand:
|
|
• „Update verfügbar“, wenn Arcane für das Image einen neueren Stand kennt,
|
|
• „lokal gebaut“, wenn das Image aus keiner Registry kommt (Arcane: updateType „local“, oder die Registry kennt es
|
|
nicht — NerdQuiz und Rippy baut der User selbst) — dort gibt es keinen Abgleich,
|
|
• sonst „aktuell“.
|
|
Der Wächter meldet gelb, wenn ein Container ungesund ist (Healthcheck) oder immer wieder neu startet; beendete
|
|
Container (etwa einmalige Aufgaben) sind kein Befund. Neue Stacks erscheinen unter „Neu entdeckt“ (entdecken.py).
|
|
"""
|
|
|
|
import logging
|
|
import re
|
|
|
|
from services.homelab import apps, arcane, einstellungen, kanal
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
_GESUND = re.compile(r"\((healthy|unhealthy|health: starting)\)")
|
|
_LOKAL = re.compile(r"distribution inspect|manifest unknown|pull access denied|repository does not exist", re.IGNORECASE)
|
|
|
|
|
|
def container_zeile(c: dict) -> dict:
|
|
"""Ein Container aus Arcanes Liste in der Form der Oberfläche."""
|
|
labels = c.get("labels") if isinstance(c.get("labels"), dict) else {}
|
|
info = c.get("updateInfo") if isinstance(c.get("updateInfo"), dict) else {}
|
|
gesund = _GESUND.search(str(c.get("status") or ""))
|
|
fehler = str(info.get("error") or "")
|
|
if info.get("hasUpdate"):
|
|
update = "neu"
|
|
elif info.get("updateType") == "local" or _LOKAL.search(fehler):
|
|
update = "lokal"
|
|
elif fehler:
|
|
update = "unbekannt"
|
|
else:
|
|
update = "aktuell"
|
|
name = str((c.get("names") or ["?"])[0]).lstrip("/")
|
|
return {"id": str(c.get("id") or name)[:12], "name": name, "stack": labels.get("com.docker.compose.project"),
|
|
"dienst": labels.get("com.docker.compose.service"), "image": c.get("image"),
|
|
"zustand": str(c.get("state") or "unbekannt"), "status": c.get("status"),
|
|
"gesund": {"healthy": True, "unhealthy": False}.get(gesund.group(1)) if gesund else None,
|
|
"update": update, "aktuell": info.get("currentVersion") or None, "neu": info.get("latestVersion") or None}
|
|
|
|
|
|
def lage() -> dict:
|
|
"""Für GET /api/homelab/docker: {"verfuegbar", "grund", "stacks": [{"name", "container": […]}]}."""
|
|
url = einstellungen.arcane_url()
|
|
if not url:
|
|
return {"verfuegbar": False, "grund": "Arcane steht nicht im Bericht des Ausführers.", "stacks": []}
|
|
if not arcane.schluessel():
|
|
return {"verfuegbar": False, "grund": "Es fehlt der Arcane-API-Schlüssel (Einstellungen → Homelab).", "stacks": []}
|
|
roh = arcane.container_liste(url)
|
|
if roh is None:
|
|
return {"verfuegbar": False, "grund": "Arcane antwortet gerade nicht.", "stacks": []}
|
|
stacks: dict[str, list[dict]] = {}
|
|
for c in roh:
|
|
zeile = container_zeile(c)
|
|
stacks.setdefault(zeile["stack"] or zeile["name"], []).append(zeile)
|
|
return {"verfuegbar": True, "grund": None,
|
|
"stacks": [{"name": n, "container": sorted(cs, key=lambda z: z["name"])} for n, cs in sorted(stacks.items())]}
|
|
|
|
|
|
def arcane_ziel() -> str:
|
|
"""Die Ziel-ID der Arcane-VM (für die Hinweise), aus dem Bericht; sonst „arcane“."""
|
|
for g in ((kanal.bericht() or {}).get("bericht") or {}).get("gaeste") or []:
|
|
app = apps.app_fuer((g.get("app") or {}).get("kennung"), g.get("name"))
|
|
if app and app.kennung == "arcane":
|
|
return f"{'ct' if g.get('art') == 'lxc' else 'vm'}-{g['vmid']}"
|
|
return "arcane"
|
|
|
|
|
|
def pruefe_docker() -> list:
|
|
"""Wächter: gelb, wenn ein Container ungesund ist oder ständig neu startet. Ohne Arcane oder Schlüssel: nichts."""
|
|
from services.waechter import Befund # der Wächter registriert diese Prüfung, also erst hier
|
|
befunde = []
|
|
quelle = None
|
|
for s in lage()["stacks"]:
|
|
for c in s["container"]:
|
|
if c["gesund"] is False:
|
|
titel, text = (f"Docker-Dienst {c['name']} ist ungesund",
|
|
f"Sein Healthcheck schlägt fehl ({c['status']}). In Arcane: Container → {c['name']} → Logs.")
|
|
elif c["zustand"] == "restarting":
|
|
titel, text = (f"Docker-Dienst {c['name']} startet immer wieder neu",
|
|
f"Er kommt nicht hoch ({c['status']}). In Arcane: Container → {c['name']} → Logs.")
|
|
else:
|
|
continue
|
|
quelle = quelle or arcane_ziel()
|
|
befunde.append(Befund(id=f"docker:{c['name']}", stufe="gelb", titel=titel, text=text, quelle=quelle))
|
|
return befunde
|