"""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 .audio import wav_duration 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 == "tts": speech = getattr(self.chat, "speech", None) if speech is None: return ProbeResult(slot.name, True, True, 0.0, "kein Speech-Client") ok = wav_duration(speech(slot.spec.litellm_model, "Test.")) >= 0.2 elif 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