deploy: Dockerfile, Compose pro Rolle, LiteLLM, Prometheus/Alertmanager, Canary-Prober, n8n-Workflows

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

core/src/knaster/api.py

@@ -53,10 +53,14 @@ class GraduateIn(BaseModel):
     min_variants: int = 10
 
 
-def create_app(system: System | None = None) -> FastAPI:
+def create_app(system: System | None = None, start_prober: bool | None = None) -> FastAPI:
+    if start_prober is None:
+        start_prober = system is None  # nur im echten Betrieb, nicht in Tests mit injiziertem System
     system = system or build_system(Settings.from_env())
     app = FastAPI(title="knaster", version="0.1.0")
     app.state.system = system
+    if start_prober and system.settings.canary_interval > 0:
+        system.prober.start(system.settings.canary_interval)
 
     def auth(authorization: str | None = Header(default=None)) -> None:
         token = system.settings.api_token

core/src/knaster/canary.py

@@ -0,0 +1,89 @@
+"""Canary-Prober: prüft Slots regelmäßig mit einer Mini-Anfrage und liefert Metriken für die Healer-Alerts."""
+
+from __future__ import annotations
+
+import threading
+import time
+from dataclasses import dataclass
+
+from prometheus_client import Gauge
+
+from .embeddings import Embedder
+from .llm import ChatClient, LLMError
+from .registry import Registry, Slot
+
+OPS_LABELS = ("docker_host", "container", "pve_node", "vmid")
+PROMPT = [{"role": "system", "content": "Antworte nur mit dem Wort OK."}, {"role": "user", "content": "Test"}]
+
+
+@dataclass
+class ProbeResult:
+    slot: str
+    up: bool       # Endpunkt antwortet
+    ok: bool       # Antwort inhaltlich korrekt
+    latency: float
+    error: str = ""
+
+
+class CanaryProber:
+    def __init__(self, registry: Registry, chat: ChatClient, embedder_for: dict[str, Embedder] | None = None,
+                 metrics_registry=None, include_tier0: bool = False):  # noqa: ANN001
+        self.registry = registry
+        self.chat = chat
+        self.embedder_for = embedder_for or {}
+        self.include_tier0 = include_tier0
+        labels = ["slot", "node", *OPS_LABELS]
+        kw = {"registry": metrics_registry} if metrics_registry is not None else {}
+        self.g_up = Gauge("knaster_slot_endpoint_up", "1 = Endpunkt erreichbar", labels, **kw)
+        self.g_ok = Gauge("knaster_slot_canary_ok", "1 = Canary-Antwort korrekt", labels, **kw)
+        self.g_latency = Gauge("knaster_slot_canary_latency_seconds", "Latenz der Canary-Anfrage", labels, **kw)
+
+    def _targets(self) -> list[Slot]:
+        out = []
+        for slot in self.registry.slots():
+            if not slot.spec.canary or slot.spec.kind == "rerank":
+                continue
+            if slot.spec.tier == 0 and not self.include_tier0:
+                continue  # kostet Geld; Erreichbarkeit überwacht LiteLLM
+            out.append(slot)
+        return out
+
+    def probe(self, slot: Slot) -> ProbeResult:
+        start = time.monotonic()
+        try:
+            if slot.spec.kind == "embedding":
+                embedder = self.embedder_for.get(slot.name)
+                if embedder is None:
+                    return ProbeResult(slot.name, True, True, 0.0, "kein Embedder konfiguriert")
+                vec = embedder.embed(["Test"])[0]
+                ok = len(vec) > 0
+            else:
+                res = self.chat.chat(slot.spec.litellm_model, PROMPT, max_tokens=5, temperature=0)
+                ok = "ok" in res.content.lower()
+            return ProbeResult(slot.name, True, ok, time.monotonic() - start)
+        except LLMError as exc:
+            return ProbeResult(slot.name, False, False, time.monotonic() - start, str(exc))
+        except Exception as exc:  # noqa: BLE001 – Netzwerkfehler des Embedders etc.
+            return ProbeResult(slot.name, False, False, time.monotonic() - start, f"{type(exc).__name__}: {exc}")
+
+    def run_once(self) -> list[ProbeResult]:
+        results = []
+        for slot in self._targets():
+            r = self.probe(slot)
+            labels = {"slot": slot.name, "node": slot.spec.node or "",
+                      **{k: slot.spec.ops.get(k, "") for k in OPS_LABELS}}
+            self.g_up.labels(**labels).set(1 if r.up else 0)
+            self.g_ok.labels(**labels).set(1 if r.ok else 0)
+            self.g_latency.labels(**labels).set(r.latency)
+            results.append(r)
+        return results
+
+    def start(self, interval: float) -> threading.Thread:
+        def loop() -> None:
+            while True:
+                self.run_once()
+                time.sleep(interval)
+
+        t = threading.Thread(target=loop, name="canary-prober", daemon=True)
+        t.start()
+        return t

core/src/knaster/healer/drivers.py

@@ -91,12 +91,14 @@ class ProxmoxDriver:
 
 
 class WebhookNotifier:
-    def __init__(self, url: str, client: httpx.Client | None = None):
+    def __init__(self, url: str, auth: str = "", client: httpx.Client | None = None):
         self.url = url
+        self.headers = {"Authorization": auth} if auth else {}
         self.client = client or httpx.Client(timeout=15.0)
 
     def notify(self, text: str, severity: str = "info", data: dict[str, Any] | None = None) -> str:
-        resp = self.client.post(self.url, json={"text": text, "severity": severity, "data": data or {}})
+        resp = self.client.post(self.url, json={"text": text, "severity": severity, "data": data or {}},
+                                headers=self.headers)
         resp.raise_for_status()
         return "benachrichtigt"
 

core/src/knaster/lifecycle.py

@@ -52,14 +52,15 @@ class LoggingJobRunner:
 
 
 class WebhookJobRunner:
-    def __init__(self, url: str, client: Any | None = None):
+    def __init__(self, url: str, auth: str = "", client: Any | None = None):
         import httpx
 
         self.url = url
+        self.headers = {"Authorization": auth} if auth else {}
         self.client = client or httpx.Client(timeout=15.0)
 
     def submit(self, job: str, candidate: dict[str, Any]) -> str:
-        resp = self.client.post(self.url, json={"job": job, "candidate": candidate})
+        resp = self.client.post(self.url, json={"job": job, "candidate": candidate}, headers=self.headers)
         resp.raise_for_status()
         return f"Job {job} übergeben"
 

core/src/knaster/registry.py

@@ -26,6 +26,8 @@ class SlotSpec(BaseModel):
     supports_lora: bool = False
     fallbacks: list[str] = Field(default_factory=list)
     kind: Literal["chat", "embedding", "rerank"] = "chat"
+    canary: bool = True            # vom Prober geprüft (für tier 0 standardmäßig aus, siehe canary.py)
+    ops: dict[str, str] = Field(default_factory=dict)  # docker_host, container, pve_node, vmid → Healer-Labels
 
 
 class AdapterSpec(BaseModel):

core/src/knaster/settings.py

@@ -35,7 +35,9 @@ class Settings:
     proxmox_token_secret: str = ""
     notify_webhook_url: str = ""
     job_webhook_url: str = ""
+    webhook_auth: str = ""  # Wert des Authorization-Headers für ausgehende Webhooks (n8n Header Auth)
     request_timeout: float = 120.0
+    canary_interval: float = 60.0  # 0 = Prober aus
 
     @classmethod
     def from_env(cls) -> Settings:
@@ -60,5 +62,7 @@ class Settings:
             proxmox_token_secret=env.get("PROXMOX_TOKEN_SECRET", ""),
             notify_webhook_url=env.get("NOTIFY_WEBHOOK_URL", ""),
             job_webhook_url=env.get("KNASTER_JOB_WEBHOOK_URL", ""),
+            webhook_auth=env.get("KNASTER_WEBHOOK_AUTH", ""),
             request_timeout=float(env.get("KNASTER_REQUEST_TIMEOUT", base.request_timeout)),
+            canary_interval=float(env.get("KNASTER_CANARY_INTERVAL", base.canary_interval)),
         )

core/src/knaster/system.py

@@ -7,6 +7,7 @@ from dataclasses import dataclass
 from datetime import datetime
 from pathlib import Path
 
+from .canary import CanaryProber
 from .classifier import EmbeddingClassifier
 from .embeddings import Embedder, LlamaCppReranker, OpenAIEmbedder, Reranker
 from .executor import Executor
@@ -46,6 +47,7 @@ class System:
     maintenance: MaintenanceAgent
     healer: Healer
     metrics: Metrics
+    prober: CanaryProber
 
 
 def _load_classifiers(directory: Path) -> dict[str, EmbeddingClassifier]:
@@ -95,8 +97,8 @@ def build_system(
     runner = Runner(router, executor, verifier, library, store, metrics)
 
     if jobs is None:
-        jobs = WebhookJobRunner(settings.job_webhook_url) if settings.job_webhook_url and not settings.dry_run \
-            else LoggingJobRunner()
+        use_webhook = settings.job_webhook_url and not settings.dry_run
+        jobs = WebhookJobRunner(settings.job_webhook_url, settings.webhook_auth) if use_webhook else LoggingJobRunner()
     lifecycle = Lifecycle(store, registry, jobs, life_cfg)
 
     if infra is None and not settings.dry_run:
@@ -106,7 +108,7 @@ def build_system(
             if settings.proxmox_url and settings.proxmox_token_id else None
         infra = CompositeInfra(portainer, proxmox)
     if notifier is None and settings.notify_webhook_url:
-        notifier = WebhookNotifier(settings.notify_webhook_url)
+        notifier = WebhookNotifier(settings.notify_webhook_url, settings.webhook_auth)
 
     actions = ActionExecutor(store=store, registry=registry, library=library, lifecycle=lifecycle, infra=infra,
                              notifier=notifier, dry_run=settings.dry_run,
@@ -117,5 +119,7 @@ def build_system(
                                    policy=policy, actions=actions, lifecycle=lifecycle, config=maint_cfg,
                                    intent_mapper=intent_mapper, clock=clock, wallclock=wallclock)
     healer = Healer(healer_cfg, store, actions, policy, propose=maintenance.propose, clock=clock)
+    prober = CanaryProber(registry, chat, embedder_for={settings.embed_model: embedder},
+                          metrics_registry=metrics.registry)
     return System(settings, store, registry, library, embedder, chat, router, executor, verifier, runner, policy,
-                  lifecycle, actions, maintenance, healer, metrics)
+                  lifecycle, actions, maintenance, healer, metrics, prober)

core/tests/test_artifacts.py

@@ -0,0 +1,155 @@
+"""Strukturprüfungen der Deployment-Artefakte (ohne laufende Infrastruktur)."""
+
+import json
+
+import httpx
+import pytest
+import yaml
+
+from knaster.healer.engine import HealerConfig
+from knaster.settings import REPO_ROOT
+
+DEPLOY = REPO_ROOT / "deploy"
+
+
+def test_alert_names_match_healer_rules():
+    alerts = yaml.safe_load((DEPLOY / "prometheus/alerts.yml").read_text())
+    names = {r["alert"] for g in alerts["groups"] for r in g["rules"]}
+    rules = HealerConfig.from_file(REPO_ROOT / "config/healer-rules.yaml").rules
+    matched = {r.match["alertname"] for r in rules}
+    assert names == matched
+
+
+def test_healer_labels_available_on_alerts():
+    """Labels, die Healer-Regeln brauchen, müssen von Metrik oder Scrape-Target kommen."""
+    prom = yaml.safe_load((DEPLOY / "prometheus/prometheus.yml").read_text())
+    jobs = {j["job_name"]: j for j in prom["scrape_configs"]}
+    for job in ("node", "gpu", "vllm"):
+        for sc in jobs[job]["static_configs"]:
+            assert {"slot", "pve_node", "vmid"} <= set(sc["labels"]), (job, sc)
+    from knaster.canary import OPS_LABELS
+    assert set(OPS_LABELS) == {"docker_host", "container", "pve_node", "vmid"}
+
+
+def test_alertmanager_targets_router_endpoint():
+    am = yaml.safe_load((DEPLOY / "alertmanager/alertmanager.yml").read_text())
+    url = am["receivers"][0]["webhook_configs"][0]["url"]
+    assert url.endswith("/v1/healer/alertmanager")
+    assert am["receivers"][0]["webhook_configs"][0]["send_resolved"] is True
+
+
+@pytest.mark.parametrize("role", ["infra", "gpu-vllm", "gpu-llamacpp"])
+def test_compose_files_parse(role):
+    data = yaml.safe_load((DEPLOY / f"compose/{role}/docker-compose.yml").read_text())
+    assert data["services"]
+    assert (DEPLOY / f"compose/{role}/.env.example").exists()
+
+
+def test_ops_labels_consistent_between_registry_and_prometheus():
+    reg = yaml.safe_load((REPO_ROOT / "sites/knaster/registry.yaml").read_text())
+    prom = yaml.safe_load((DEPLOY / "prometheus/prometheus.yml").read_text())
+    vllm = {sc["labels"]["slot"]: sc["labels"] for sc in
+            next(j for j in prom["scrape_configs"] if j["job_name"] == "vllm")["static_configs"]}
+    for slot, labels in vllm.items():
+        ops = reg["slots"][slot]["ops"]
+        assert ops["vmid"] == labels["vmid"] and ops["container"] == labels["container"]
+
+
+WORKFLOWS = sorted((REPO_ROOT / "n8n/workflows").glob("*.json"))
+
+
+@pytest.mark.parametrize("path", WORKFLOWS, ids=[p.name for p in WORKFLOWS])
+def test_n8n_workflow_structure(path):
+    wf = json.loads(path.read_text())
+    assert wf["active"] is False
+    names = [n["name"] for n in wf["nodes"]]
+    assert len(names) == len(set(names))
+    ids = [n["id"] for n in wf["nodes"]]
+    assert len(ids) == len(set(ids))
+    for src, outs in wf["connections"].items():
+        assert src in names
+        for branch in outs["main"]:
+            for target in branch:
+                assert target["node"] in names
+    # jeder Knoten außer dem Trigger ist erreichbar
+    targets = {t["node"] for o in wf["connections"].values() for b in o["main"] for t in b}
+    triggers = [n for n in wf["nodes"] if "trigger" in n["type"].lower() or n["type"].endswith(".webhook")]
+    assert len(triggers) == 1
+    assert set(names) - targets == {triggers[0]["name"]}
+
+
+def test_n8n_router_paths_exist(make_system):
+    from knaster.api import create_app
+
+    paths = {r.path for r in create_app(make_system()).routes}
+    for path in WORKFLOWS:
+        for n in json.loads(path.read_text())["nodes"]:
+            url = n["parameters"].get("url", "")
+            if "router_url" in url:
+                assert url.split("}}")[-1] in paths
+
+
+def test_webhook_notifier_sends_auth_header():
+    from knaster.healer.drivers import WebhookNotifier
+
+    seen = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen["auth"] = request.headers.get("authorization")
+        seen["body"] = json.loads(request.content)
+        return httpx.Response(200)
+
+    n = WebhookNotifier("http://n8n/webhook/knaster/notify", "Bearer t",
+                        client=httpx.Client(transport=httpx.MockTransport(handler)))
+    n.notify("hallo", "high")
+    assert seen == {"auth": "Bearer t", "body": {"text": "hallo", "severity": "high", "data": {}}}
+
+
+def test_drivers_build_correct_requests():
+    from knaster.healer.drivers import PortainerDriver, ProxmoxDriver
+
+    calls = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        calls.append((request.method, request.url.path, dict(request.headers)))
+        if request.url.path == "/api/endpoints":
+            return httpx.Response(200, json=[{"Id": 3, "Name": "gpu-main"}])
+        return httpx.Response(200, json={"data": "UPID:1"})
+
+    client = httpx.Client(transport=httpx.MockTransport(handler))
+    PortainerDriver("https://portainer", "k", client=client).restart_container("gpu-main", "vllm")
+    assert calls[-1][:2] == ("POST", "/api/endpoints/3/docker/containers/vllm/restart")
+    assert calls[-1][2]["x-api-key"] == "k"
+    ProxmoxDriver("https://pve:8006", "knaster@pve!healer", "s", client=client).restart_vm("pve-ada", "201")
+    assert calls[-1][:2] == ("POST", "/api2/json/nodes/pve-ada/qemu/201/status/reboot")
+    assert calls[-1][2]["authorization"] == "PVEAPIToken=knaster@pve!healer=s"
+
+
+def test_llm_and_embedding_clients():
+    from knaster.embeddings import LlamaCppReranker, OpenAIEmbedder
+    from knaster.llm import ContextOverflowError, EndpointError, OpenAIChatClient
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        body = json.loads(request.content)
+        if request.url.path == "/v1/embeddings":
+            return httpx.Response(200, json={"data": [{"index": 1, "embedding": [0.0, 1.0]},
+                                                      {"index": 0, "embedding": [1.0, 0.0]}]})
+        if request.url.path == "/v1/rerank":
+            return httpx.Response(200, json={"results": [{"index": 1, "relevance_score": 0.9},
+                                                         {"index": 0, "relevance_score": 0.1}]})
+        if body["model"] == "overflow":
+            return httpx.Response(400, text="This model's maximum context length is 8192 tokens")
+        if body["model"] == "down":
+            return httpx.Response(503, text="unavailable")
+        return httpx.Response(200, json={"model": body["model"], "usage": {"total_tokens": 3},
+                                         "choices": [{"message": {"content": "hi"}, "finish_reason": "stop"}]})
+
+    client = httpx.Client(transport=httpx.MockTransport(handler))
+    assert OpenAIEmbedder("http://x", "e", client=client).embed(["a", "b"]) == [[1.0, 0.0], [0.0, 1.0]]
+    assert LlamaCppReranker("http://x", client=client).rerank("q", ["a", "b"]) == [0.1, 0.9]
+    chat = OpenAIChatClient("http://x", "k", client=client)
+    assert chat.chat("m", [{"role": "user", "content": "x"}]).content == "hi"
+    with pytest.raises(ContextOverflowError):
+        chat.chat("overflow", [])
+    with pytest.raises(EndpointError):
+        chat.chat("down", [])

core/tests/test_inventory_validate.py

@@ -88,3 +88,26 @@ def test_cli(capsys):
                  "--catalog", str(REPO_ROOT / "catalog")]) == 0
     out = capsys.readouterr().out
     assert "gpu-main" in out
+
+
+def test_canary_prober(registry, fake_chat):
+    from prometheus_client import CollectorRegistry, generate_latest
+
+    from knaster.canary import CanaryProber
+    from knaster.embeddings import HashingEmbedder
+    from knaster.llm import EndpointError
+
+    def down(m):
+        raise EndpointError("weg")
+
+    fake_chat.handlers.update({"tier1-general": lambda m: "OK", "tier1-flex": down,
+                               "verifier-small": lambda m: "Hallo"})
+    reg = CollectorRegistry()
+    prober = CanaryProber(registry, fake_chat, {"embed-default": HashingEmbedder()}, metrics_registry=reg)
+    results = {r.slot: r for r in prober.run_once()}
+    assert "tier0-strong" not in results
+    assert results["tier1-general"].ok and not results["tier1-flex"].up
+    assert results["verifier-small"].up and not results["verifier-small"].ok
+    assert results["embed-default"].ok
+    text = generate_latest(reg).decode()
+    assert 'knaster_slot_endpoint_up{container="",docker_host="",node="gpu-flex"' in text