Letzte Aktualisierung: Januar 2025 | Lesezeit: 12 Minuten
Ein Fehler, der mich eine ganze Nacht kostete
Es war 2:47 Uhr morgens, als mein Telefon zu vibrieren begann. Ein kritischer Geschäftspartner rief an: „Ihre KI-Chatbot-Anwendung antwortet nicht mehr." Ich öffnete hastig meinen Laptop und sah im Dashboard einen scrollenden Strom von Fehlermeldungen:
ConnectionError: timeout after 30000ms
ConnectionError: timeout after 30000ms
ConnectionError: timeout after 30000ms
HTTP 429: Too Many Requests
HTTP 503: Service Unavailable
HTTP 401: UnauthorizedWas war passiert? Mein Monitoring war nicht richtig konfiguriert. Ich hatte zwar Metriken gesammelt, aber keine rechtzeitigen Alerts eingerichtet. Die Folge: Ein Produktionsausfall von über 3 Stunden, währenddessen Tausende von Nutzern vergeblich auf Antworten warteten.
Dieser Vorfall war der Auslöser, warum ich heute einen umfassenden Leitfaden zum Thema AI API SLA Monitoring und Alerting schreibe – speziell für die Nutzung mit HolySheep AI.
Warum SLA Monitoring für AI APIs entscheidend ist
Bei HolySheep AI erhalten Sie Zugriff auf hochwertige Modelle wie GPT-4.1, Claude Sonnet 4.5 und DeepSeek V3.2 mit einer beeindruckenden Latenz von unter 50ms. Doch selbst bei diesem Niveau an Zuverlässigkeit gilt:
- Latenz-Spikes: Netzwerkbedingte Verzögerungen können auftreten
- Rate-Limits: Bei hohem Traffic können 429-Fehler erscheinen
- Authentifizierungsprobleme: Ungültige oder abgelaufene API-Keys führen zu 401-Fehlern
- Modell-Verfügbarkeit: Geplante Wartungen oder unvorhergesehene Ausfälle
Ein robustes Monitoring-System ermöglicht es Ihnen, Probleme zu erkennen, bevor sie zu kritischen Ausfällen werden.
Die Architektur: Überblick
Bevor wir in den Code eintauchen, hier die Gesamtarchitektur unseres Monitoring-Systems:
┌─────────────────────────────────────────────────────────────────┐
│ MONITORING ARCHITEKTUR │
├─────────────────────────────────────────────────────────────────┤
│ │
│ [Ihre Anwendung] ──► [HolySheep API] ──► [Response Data] │
│ │ │ │
│ ▼ ▼ │
│ [Prometheus Metrics] [Latenz-Logger] │
│ │ │ │
│ ▼ ▼ │
│ [Alertmanager] ──► [Paging (Slack/Email/PagerDuty)] │
│ │
│ SLA Ziele: │
│ • Verfügbarkeit: 99.9% (max. 8.7h Ausfall/Jahr) │
│ • Latenz P99: < 500ms │
│ • Fehlerrate: < 0.1% │
└─────────────────────────────────────────────────────────────────┘Grundlegendes Monitoring mit Python
Beginnen wir mit dem Kernstück: einem robusten API-Client, der automatisch Metriken sammelt. Der folgende Code ist vollständig einsatzbereit und verwendet die HolySheep AI API:
#!/usr/bin/env python3
"""
HolySheep AI API Client mit integriertem Monitoring
Version: 2.1.0
Kompatibel mit Python 3.8+
"""
import time
import logging
import httpx
from datetime import datetime, timedelta
from typing import Optional, Dict, Any, List
from dataclasses import dataclass, field
from collections import defaultdict
import threading
import statistics
Für Prometheus-Metriken
try:
from prometheus_client import Counter, Histogram, Gauge, CollectorRegistry, push_to_gateway
PROMETHEUS_AVAILABLE = True
except ImportError:
PROMETHEUS_AVAILABLE = False
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s | %(levelname)-8s | %(name)s | %(message)s'
)
logger = logging.getLogger("HolySheepMonitor")
@dataclass
class SLAMetrics:
"""Speichert SLA-relevante Metriken"""
total_requests: int = 0
successful_requests: int = 0
failed_requests: int = 0
timeout_count: int = 0
auth_error_count: int = 0
rate_limit_count: int = 0
latencies: List[float] = field(default_factory=list)
last_success_time: Optional[datetime] = None
last_error_time: Optional[datetime] = None
last_error_type: Optional[str] = None
@property
def success_rate(self) -> float:
if self.total_requests == 0:
return 100.0
return (self.successful_requests / self.total_requests) * 100
@property
def error_rate(self) -> float:
return 100.0 - self.success_rate
@property
def p50_latency(self) -> Optional[float]:
if not self.latencies:
return None
return statistics.median(self.latencies)
@property
def p95_latency(self) -> Optional[float]:
if len(self.latencies) < 20:
return None
sorted_latencies = sorted(self.latencies)
index = int(len(sorted_latencies) * 0.95)
return sorted_latencies[index]
@property
def p99_latency(self) -> Optional[float]:
if len(self.latencies) < 100:
return None
sorted_latencies = sorted(self.latencies)
index = int(len(sorted_latencies) * 0.99)
return sorted_latencies[index]
class HolySheepMonitor:
"""
Monitoring-Client für HolySheep AI API mit automatischer
Metrik-Sammlung und Alert-Funktionalität.
base_url: https://api.holysheep.ai/v1
"""
def __init__(
self,
api_key: str,
base_url: str = "https://api.holysheep.ai/v1",
timeout: float = 30.0,
enable_prometheus: bool = True
):
self.api_key = api_key
self.base_url = base_url.rstrip('/')
self.timeout = timeout
self.metrics = SLAMetrics()
self._lock = threading.Lock()
# HTTP-Client mit Retry-Logik
self._client = httpx.Client(
timeout=timeout,
limits=httpx.Limits(max_keepalive_connections=20, max_connections=100)
)
# Prometheus-Metriken initialisieren
if PROMETHEUS_AVAILABLE and enable_prometheus:
self._setup_prometheus_metrics()
# Alert-Thresholds
self.alert_thresholds = {
'latency_p99_ms': 500,
'error_rate_percent': 1.0,
'timeout_rate_percent': 5.0,
'consecutive_failures': 5
}
self._consecutive_failures = 0
logger.info(f"HolySheepMonitor initialisiert für {base_url}")
def _setup_prometheus_metrics(self):
"""Richtet Prometheus-Metriken ein"""
self.prom_requests_total = Counter(
'holysheep_requests_total',
'Total number of API requests',
['status', 'model']
)
self.prom_latency_seconds = Histogram(
'holysheep_request_latency_seconds',
'Request latency in seconds',
['model', 'endpoint']
)
self.prom_in_flight = Gauge(
'holysheep_requests_in_flight',
'Number of requests currently in flight'
)
logger.info("Prometheus-Metriken aktiviert")
def _make_request(
self,
method: str,
endpoint: str,
**kwargs
) -> httpx.Response:
"""Interner HTTP-Request mit automatischer Fehlerbehandlung"""
url = f"{self.base_url}/{endpoint.lstrip('/')}"
headers = kwargs.pop('headers', {})
headers['Authorization'] = f'Bearer {self.api_key}'
headers['Content-Type'] = 'application/json'
try:
response = self._client.request(
method=method,
url=url,
headers=headers,
**kwargs
)
return response
except httpx.TimeoutException as e:
logger.error(f"Timeout bei {endpoint}: {e}")
raise
except httpx.ConnectError as e:
logger.error(f"Verbindungsfehler bei {endpoint}: {e}")
raise
def call_chat_completion(
self,
model: str,
messages: List[Dict[str, str]],
temperature: float = 0.7,
max_tokens: int = 1000,
**kwargs
) -> Dict[str, Any]:
"""
Führt einen Chat-Completion-Aufruf durch und sammelt Metriken.
Verfügbare Modelle bei HolySheep AI:
- gpt-4.1: $8.00/MTok (Premium)
- claude-sonnet-4.5: $15.00/MTok (Premium)
- gemini-2.5-flash: $2.50/MTok (Kosteneffizient)
- deepseek-v3.2: $0.42/MTok (Budget-freundlich)
"""
start_time = time.perf_counter()
if PROMETHEUS_AVAILABLE:
self.prom_in_flight.inc()
try:
payload = {
'model': model,
'messages': messages,
'temperature': temperature,
'max_tokens': max_tokens,
**kwargs
}
response = self._make_request(
method='POST',
endpoint='/chat/completions',
json=payload
)
latency_ms = (time.perf_counter() - start_time) * 1000
with self._lock:
self.metrics.total_requests += 1
self.metrics.latencies.append(latency_ms)
# Nur die letzten 1000 Latenzen behalten
if len(self.metrics.latencies) > 1000:
self.metrics.latencies = self.metrics.latencies[-1000:]
if response.status_code == 200:
with self._lock:
self.metrics.successful_requests += 1
self.metrics.last_success_time = datetime.now()
self._consecutive_failures = 0
if PROMETHEUS_AVAILABLE:
self.prom_requests_total.labels(status='success', model=model).inc()
self.prom_latency_seconds.labels(model=model, endpoint='chat').observe(
latency_ms / 1000
)
# Latenz-Prüfung für Alerting
self._check_latency_alert(latency_ms)
return response.json()
elif response.status_code == 401:
with self._lock:
self.metrics.failed_requests += 1
self.metrics.auth_error_count += 1
self.metrics.last_error_time = datetime.now()
self.metrics.last_error_type = '401_UNAUTHORIZED'
logger.error("Authentifizierungsfehler! API-Key prüfen.")
if PROMETHEUS_AVAILABLE:
self.prom_requests_total.labels(status='auth_error', model=model).inc()
raise AuthenticationError("Ungültiger API-Key")
elif response.status_code == 429:
with self._lock:
self.metrics.failed_requests += 1
self.metrics.rate_limit_count += 1
self.metrics.last_error_time = datetime.now()
self.metrics.last_error_type = '429_RATE_LIMIT'
retry_after = int(response.headers.get('Retry-After', 60))
logger.warning(f"Rate-Limit erreicht. Retry nach {retry_after}s")
if PROMETHEUS_AVAILABLE:
self.prom_requests_total.labels(status='rate_limit', model=model).inc()
raise RateLimitError(f"Rate-Limit erreicht. Retry nach {retry_after}s")
else:
with self._lock:
self.metrics.failed_requests += 1
self.metrics.last_error_time = datetime.now()
self.metrics.last_error_type = f'HTTP_{response.status_code}'
logger.error(f"HTTP {response.status_code}: {response.text}")
if PROMETHEUS_AVAILABLE:
self.prom_requests_total.labels(status='error', model=model).inc()
raise APIError(f"API-Fehler: {response.status_code}")
except httpx.TimeoutException:
with self._lock:
self.metrics.failed_requests += 1
self.metrics.timeout_count += 1
self.metrics.last_error_time = datetime.now()
self.metrics.last_error_type = 'TIMEOUT'
self._consecutive_failures += 1
if PROMETHEUS_AVAILABLE:
self.prom_requests_total.labels(status='timeout', model=model).inc()
logger.error(f"Timeout nach {self.timeout}s bei Modell {model}")
raise TimeoutError(f"Anfrage an {model} timeout nach {self.timeout}s")
except Exception as e:
with self._lock:
self.metrics.failed_requests += 1
self.metrics.last_error_time = datetime.now()
self.metrics.last_error_type = 'CONNECTION_ERROR'
self._consecutive_failures += 1
if PROMETHEUS_AVAILABLE:
self.prom_requests_total.labels(status='exception', model=model).inc()
logger.error(f"Unerwarteter Fehler: {e}")
raise
finally:
if PROMETHEUS_AVAILABLE:
self.prom_in_flight.dec()
# Fehler-Rate prüfen
self._check_error_rate_alert()
def _check_latency_alert(self, latency_ms: float):
"""Prüft, ob Latenz-Alert ausgelöst werden soll"""
if latency_ms > self.alert_thresholds['latency_p99_ms']:
logger.warning(
f"⚠️ LATENZ-ALERT: {latency_ms:.0f}ms (Schwellwert: "
f"{self.alert_thresholds['latency_p99_ms']}ms)"
)
def _check_error_rate_alert(self):
"""Prüft, ob Fehler-Rate-Alert ausgelöst werden soll"""
if self._consecutive_failures >= self.alert_thresholds['consecutive_failures']:
logger.critical(
f"🚨 KRITISCHER ALERT: {self._consecutive_failures} aufeinanderfolgende "
f"Fehler! Fehlerrate: {self.metrics.error_rate:.2f}%"
)
def get_sla_report(self) -> Dict[str, Any]:
"""Generiert einen vollständigen SLA-Bericht"""
with self._lock:
return {
'timestamp': datetime.now().isoformat(),
'total_requests': self.metrics.total_requests,
'successful_requests': self.metrics.successful_requests,
'failed_requests': self.metrics.failed_requests,
'success_rate_percent': round(self.metrics.success_rate, 3),
'error_rate_percent': round(self.metrics.error_rate, 3),
'latency': {
'p50_ms': round(self.metrics.p50_latency, 2) if self.metrics.p50_latency else None,
'p95_ms': round(self.metrics.p95_latency, 2) if self.metrics.p95_latency else None,
'p99_ms': round(self.metrics.p99_latency, 2) if self.metrics.p99_latency else None,
},
'error_breakdown': {
'timeouts': self.metrics.timeout_count,
'auth_errors': self.metrics.auth_error_count,
'rate_limits': self.metrics.rate_limit_count,
},
'last_success': self.metrics.last_success_time.isoformat() if self.metrics.last_success_time else None,
'last_error': {
'time': self.metrics.last_error_time.isoformat() if self.metrics.last_error_time else None,
'type': self.metrics.last_error_type
},
'sla_compliance': {
'availability_target': 99.9,
'availability_actual': round(self.metrics.success_rate, 3),
'meets_sla': self.metrics.success_rate >= 99.9
}
}
def reset_metrics(self):
"""Setzt alle Metriken zurück"""
with self._lock:
self.metrics = SLAMetrics()
self._consecutive_failures = 0
logger.info("Metriken zurückgesetzt")
class AuthenticationError(Exception):
"""Authentifizierungsfehler bei der API"""
pass
class RateLimitError(Exception):
"""Rate-Limit überschritten"""
pass
class APIError(Exception):
"""Allgemeiner API-Fehler"""
pass
============================================
BENUTZUNGSBEISPIEL
============================================
if __name__ == "__main__":
# API-Key aus Umgebungsvariable oder direkt
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
# Monitoring-Client initialisieren
monitor = HolySheepMonitor(
api_key=API_KEY,
base_url="https://api.holysheep.ai/v1",
timeout=30.0
)
# Test-Anfrage
try:
response = monitor.call_chat_completion(
model="deepseek-v3.2", # $0.42/MTok - sehr kostengünstig
messages=[
{"role": "system", "content": "Du bist ein hilfreicher Assistent."},
{"role": "user", "content": "Erkläre SLA Monitoring in einem Satz."}
],
max_tokens=150
)
print(f"✅ Antwort: {response['choices'][0]['message']['content']}")
except Exception as e:
print(f"❌ Fehler: {e}")
# SLA-Bericht ausgeben
print("\n📊 SLA-BERIGHT:")
import json
print(json.dumps(monitor.get_sla_report(), indent=2))Alerting-System mit Alertmanager-Integration
Der folgende Code erweitert unser Monitoring um ein vollständiges Alerting-System mit Slack-, E-Mail- und PagerDuty-Integration:
#!/usr/bin/env python3
"""
Alerting-System für HolySheep AI API
Sendet Alerts bei SLA-Verletzungen über Multiple Channels
"""
import os
import json
import smtplib
import sqlite3
from datetime import datetime, timedelta
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from typing import List, Dict, Callable, Optional
from dataclasses import dataclass, field
from enum import Enum
import logging
import asyncio
import aiohttp
import time
import threading
from collections import deque
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("AlertManager")
class AlertSeverity(Enum):
"""Alert-Schweregrade nach ITIL-Standard"""
INFO = "info"
WARNING = "warning"
ERROR = "error"
CRITICAL = "critical"
class AlertChannel(Enum):
"""Verfügbare Alert-Kanäle"""
SLACK = "slack"
EMAIL = "email"
PAGERDUTY = "pagerduty"
WEBHOOK = "webhook"
LOG = "log"
@dataclass
class Alert:
"""Repräsentiert einen Alert"""
id: str
timestamp: datetime
severity: AlertSeverity
title: str
message: str
metric_name: str
metric_value: float
threshold: float
unit: str = ""
resolved: bool = False
resolved_at: Optional[datetime] = None
def to_dict(self) -> Dict:
return {
'id': self.id,
'timestamp': self.timestamp.isoformat(),
'severity': self.severity.value,
'title': self.title,
'message': self.message,
'metric_name': self.metric_name,
'metric_value': self.metric_value,
'threshold': self.threshold,
'unit': self.unit,
'resolved': self.resolved,
'resolved_at': self.resolved_at.isoformat() if self.resolved_at else None
}
@dataclass
class AlertRule:
"""Definiert eine Alert-Regel"""
name: str
metric_key: str
condition: str # 'gt', 'lt', 'eq', 'gte', 'lte'
threshold: float
severity: AlertSeverity
description: str
consecutive_violations_required: int = 1
cooldown_seconds: int = 300 # 5 Minuten Cooldown zwischen Alerts
class SlackNotifier:
"""Slack Webhook-Integration"""
def __init__(self, webhook_url: str, channel: str = "#alerts"):
self.webhook_url = webhook_url
self.channel = channel
def send(self, alert: Alert) -> bool:
"""Sendet Alert an Slack"""
severity_emojis = {
AlertSeverity.INFO: "ℹ️",
AlertSeverity.WARNING: "⚠️",
AlertSeverity.ERROR: "❌",
AlertSeverity.CRITICAL: "🚨"
}
color_map = {
AlertSeverity.INFO: "#36a64f",
AlertSeverity.WARNING: "#ff9900",
AlertSeverity.ERROR: "#dc3545",
AlertSeverity.CRITICAL: "#8b0000"
}
payload = {
"channel": self.channel,
"username": "HolySheep AI Alertbot",
"icon_emoji": ":zap:",
"attachments": [{
"color": color_map[alert.severity],
"blocks": [
{
"type": "header",
"text": {
"type": "plain_text",
"text": f"{severity_emojis[alert.severity]} {alert.title}"
}
},
{
"type": "section",
"fields": [
{
"type": "mrkdwn",
"text": f"*Schweregrad:*\n{alert.severity.value.upper()}"
},
{
"type": "mrkdwn",
"text": f"*Zeitstempel:*\n{alert.timestamp.strftime('%Y-%m-%d %H:%M:%S UTC')}"
}
]
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"*Metrik:* {alert.metric_name}\n*Wert:* {alert.metric_value:.2f}{alert.unit}\n*Schwellwert:* {alert.threshold}{alert.unit}"
}
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"*Beschreibung:*\n{alert.message}"
}
},
{
"type": "actions",
"elements": [
{
"type": "button",
"text": {"type": "plain_text", "text": "✅ Acknowledgen"},
"style": "primary",
"value": alert.id
},
{
"type": "button",
"text": {"type": "plain_text", "text": "📊 Dashboard"},
"value": "dashboard"
}
]
}
]
}]
}
try:
response = requests.post(
self.webhook_url,
json=payload,
headers={'Content-Type': 'application/json'},
timeout=10
)
return response.status_code == 200
except Exception as e:
logger.error(f"Slack-Notification fehlgeschlagen: {e}")
return False
class EmailNotifier:
"""E-Mail-Benachrichtigung via SMTP"""
def __init__(
self,
smtp_host: str,
smtp_port: int,
smtp_user: str,
smtp_password: str,
from_address: str,
to_addresses: List[str],
use_tls: bool = True
):
self.smtp_host = smtp_host
self.smtp_port = smtp_port
self.smtp_user = smtp_user
self.smtp_password = smtp_password
self.from_address = from_address
self.to_addresses = to_addresses
self.use_tls = use_tls
def send(self, alert: Alert) -> bool:
"""Sendet Alert per E-Mail"""
severity_prefix = {
AlertSeverity.INFO: "[INFO]",
AlertSeverity.WARNING: "[WARNUNG]",
AlertSeverity.ERROR: "[FEHLER]",
AlertSeverity.CRITICAL: "[KRITISCH]"
}
msg = MIMEMultipart('alternative')
msg['Subject'] = f"{severity_prefix[alert.severity]} HolySheep AI Alert: {alert.title}"
msg['From'] = self.from_address
msg['To'] = ', '.join(self.to_addresses)
html_body = f"""
{severity_prefix[alert.severity]} {alert.title}
{alert.message}
Metrik-Details
Metrik
Aktueller Wert
Schwellwert
Einheit
{alert.metric_name}
{alert.metric_value:.2f}
{alert.threshold}
{alert.unit}
Zeitstempel: {alert.timestamp.strftime('%Y-%m-%d %H:%M:%S UTC')}
Alert-ID: {alert.id}
Diese Alert wurde automatisch von HolySheep AI Monitoring generiert.
API-SLA: 99.9% Verfügbarkeit | Latenz: <50ms
"""
text_body = f"""
{severity_prefix[alert.severity]} HolySheep AI Alert: {alert.title}
{alert.message}
Metrik-Details:
- Metrik: {alert.metric_name}
- Aktueller Wert: {alert.metric_value:.2f}{alert.unit}
- Schwellwert: {alert.threshold}{alert.unit}
- Zeitstempel: {alert.timestamp.strftime('%Y-%m-%d %H:%M:%S UTC')}
- Alert-ID: {alert.id}
API-SLA: 99.9% Verfügbarkeit | Latenz: <50ms
"""
msg.attach(MIMEText(text_body, 'plain'))
msg.attach(MIMEText(html_body, 'html'))
try:
with smtplib.SMTP(self.smtp_host, self.smtp_port) as server:
if self.use_tls:
server.starttls()
server.login(self.smtp_user, self.smtp_password)
server.send_message(msg)
return True
except Exception as e:
logger.error(f"E-Mail-Notification fehlgeschlagen: {e}")
return False
class AlertManager:
"""
Zentraler Alert-Manager für HolySheep AI API Monitoring
"""
def __init__(self, db_path: str = "alerts.db"):
self.alerts: List[Alert] = []
self.rules: List[AlertRule] = []
self.notifiers: Dict[AlertChannel, Callable] = {}
self.violation_counts: Dict[str, int] = {}
self.last_alert_time: Dict[str, datetime] = {}
self.alert_history = deque(maxlen=1000) # Letzte 1000 Alerts behalten
self._lock = threading.Lock()
# SQLite-Datenbank für Alert-Historie
self._init_database(db_path)
# Standard-Regeln definieren
self._setup_default_rules()
logger.info("AlertManager initialisiert")
def _init_database(self, db_path: str):
"""Initialisiert SQLite-Datenbank"""
self.db_path = db_path
with sqlite3.connect(db_path) as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS alerts (
id TEXT PRIMARY KEY,
timestamp TEXT NOT NULL,
severity TEXT NOT NULL,
title TEXT NOT NULL,
message TEXT,
metric_name TEXT,
metric_value REAL,
threshold REAL,
unit TEXT,
resolved INTEGER DEFAULT 0,
resolved_at TEXT
)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_alerts_timestamp
ON alerts(timestamp DESC)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_alerts_severity
ON alerts(severity)
""")
def _setup_default_rules(self):
"""Definiert Standard-Alert-Regeln"""
self.add_rule(AlertRule(
name="Hohe Fehlerrate",
metric_key="error_rate_percent",
condition="gt",
threshold=1.0,
severity=AlertSeverity.ERROR,
description="Fehlerrate übersteigt 1%",
consecutive_violations_required=3,
cooldown_seconds=300
))
self.add_rule(AlertRule(
name="Kritische Fehlerrate",
metric_key="error_rate_percent",
condition="gt",
threshold=5.0,
severity=AlertSeverity.CRITICAL,
description="Fehlerrate übersteigt 5% - sofortige Untersuchung erforderlich",
consecutive_violations_required=1,
cooldown_seconds=60
))
self.add_rule(AlertRule(
name="Hohe Latenz P99",
metric_key="latency_p99_ms",
condition="gt",
threshold=500,
severity=AlertSeverity.WARNING,
description="P99-Latenz übersteigt 500ms",
consecutive_violations_required=5,
cooldown_seconds=300
))
self.add_rule(AlertRule(
name="Timeout-Storm",
metric_key="timeout_count",
condition="gt",
threshold=10,
severity=AlertSeverity.CRITICAL,
description="Mehr als 10 Timeouts in kurzer Zeit",
consecutive_violations_required=1,
cooldown_seconds=120
))
self.add_rule(AlertRule(
name="Authentifizierungsfehler",
metric_key="auth_error_count",
condition="gt",
threshold=1,
severity=AlertSeverity.CRITICAL,
description="Authentifizierungsfehler erkannt - API-Key prüfen",
consecutive_violations_required=1,
cooldown_seconds=0
))
self.add_rule(AlertRule(
name="Rate-Limit Überschreitung",
metric_key="rate_limit_count",
condition="gt",
threshold=5,
severity=AlertSeverity.WARNING,
description="Häufige Rate-Limit-Überschreitungen",
consecutive_violations_required=3,
cooldown_seconds=180
))
def add_rule(self, rule: AlertRule):
"""Fügt eine neue Alert-Regel hinzu"""
self.rules.append(rule)
self.violation_counts[rule.name] = 0
logger.info(f"Alert-Regel hinzugefügt: {rule.name}")
def register_notifier(self, channel: AlertChannel, notifier: Callable):
"""Registriert einen Notification-Channel"""
self.notifiers[channel] = notifier
logger.info(f"Notifier registriert: {channel.value}")
def check_metrics(self, metrics: Dict) -> List[Alert]:
"""
Prüft Metriken gegen alle Regeln und generiert ggf. Alerts.
"""
generated_alerts = []
with self._lock:
for rule in self.rules:
metric_value = metrics.get(rule.metric_key)
if metric_value is None:
continue
# Bedingung prüfen
violated = False
if rule.condition == "gt" and metric_value > rule.threshold:
violated = True
elif rule.condition == "gte" and metric_value >= rule.threshold:
violated = True
elif rule.condition == "lt" and