40 lines
1.9 KiB
Python
40 lines
1.9 KiB
Python
"""Prometheus-Metriken des Router-Service (eigene Registry pro Instanz)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING
|
|
|
|
from prometheus_client import CollectorRegistry, Counter, Gauge, Histogram, generate_latest
|
|
|
|
if TYPE_CHECKING:
|
|
from .runner import RunResult
|
|
|
|
|
|
class Metrics:
|
|
def __init__(self) -> None:
|
|
self.registry = CollectorRegistry()
|
|
self.runs = Counter("knaster_runs_total", "Läufe nach Ergebnis und Endstufe", ["status", "stage"],
|
|
registry=self.registry)
|
|
self.escalations = Counter("knaster_escalations_total", "Eskalationen nach Signal", ["signal"],
|
|
registry=self.registry)
|
|
self.verifier = Counter("knaster_verifier_total", "Verifier-Ergebnisse", ["skill", "stage", "passed"],
|
|
registry=self.registry)
|
|
self.attempts = Histogram("knaster_run_attempts", "Versuche pro Lauf", buckets=(1, 2, 3, 4),
|
|
registry=self.registry)
|
|
self.slot_status = Gauge("knaster_slot_available", "1 = Slot verfügbar", ["slot"], registry=self.registry)
|
|
self.open_proposals = Gauge("knaster_open_proposals", "Offene Pflege-Vorschläge", registry=self.registry)
|
|
|
|
def observe_run(self, result: RunResult) -> None:
|
|
self.runs.labels(result.status, str(result.stage)).inc()
|
|
self.attempts.observe(len(result.attempts))
|
|
for attempt in result.attempts[1:]:
|
|
for s in attempt.decision.signals[:1]:
|
|
self.escalations.labels(s.value).inc()
|
|
for attempt in result.attempts:
|
|
if attempt.verification is not None:
|
|
self.verifier.labels(attempt.decision.skill or "-", str(attempt.decision.stage),
|
|
str(attempt.verification.passed).lower()).inc()
|
|
|
|
def render(self) -> bytes:
|
|
return generate_latest(self.registry)
|