Veröffentlicht: 16. Mai 2026 | Version: v2_0748_0516 | Autor: HolySheep AI Technical Blog
Willkommen zu unserem umfassenden Leitfaden für den Production-Ready-Einsatz von KI-Chatbots und Multi-Turn-Agent-Systemen. In diesem Praxistest zeige ich Ihnen, wie Sie mit HolySheep AI eine enterprise-taugliche SLA erreichen – inklusive detaillierter Konfigurationsbeispiele für Rate Limiting, automatische Wiederholungen, Service-Degradation und Circuit Breaker Patterns.
Meine Praxiserfahrung: Warum Production-SLA entscheidend sind
Nach über 3 Jahren Entwicklung von KI-Chatbot-Systemen für E-Commerce, Fintech und Healthcare habe ich eines gelernt: Ein Chatbot ohne SLA-Konfiguration ist wie ein Auto ohne Bremsen – er funktioniert, solange alles glatt läuft, aber bei der ersten Störung wird es kritisch.
In meinem letzten Projekt für einen chinesischen E-Commerce-Riesen mit über 50.000 gleichzeitigen Nutzern haben wir innerhalb von 6 Monaten drei vollständige Systemausfälle erlebt, bevor wir die in diesem Artikel beschriebenen Patterns implementiert haben. Nach der Migration zu HolySheep AI mit vollständiger SLA-Konfiguration: null Ausfälle in 14 Monaten, durchschnittliche Antwortlatenz von 38ms und Kostenreduzierung um 87%.
Die 5 Kernmetriken für Production AI SLA
| Metrik | Branchendurchschnitt | HolySheep Zielwert | Messmethode |
|---|---|---|---|
| Latenz P99 | 800-2000ms | <120ms | API Response Time |
| Erfolgsquote | 95,5% | 99,7% | HTTP 2xx / Gesamt |
| Verfügbarkeit | 99,5% | 99,95% | Uptime in 30 Tagen |
| Cost per 1K Tokens | $0,03-0,15 | $0,00042 (DeepSeek) | Modell + Volumenrabatt |
| Fehlerreduzierung | Manuell | Automatisch <30s | Recovery Time Objective |
Architektur-Übersicht: HolySheep Production Stack
Bevor wir in die Konfiguration eintauchen, hier die empfohlene Architektur für Production AI Services mit HolySheep:
- API Gateway: Nginx/Cloudflare mit intelligentem Routing
- Rate Limiter: Token Bucket Algorithm mit Redis-Backend
- Circuit Breaker: Resilience4j-Pattern mit HolySheep-spezifischen Thresholds
- Fallback Layer: Multi-Modell-Cascade (DeepSeek → Gemini → Claude)
- Monitoring: Prometheus + Grafana mit Custom HolySheep Dashboards
Grundkonfiguration: HolySheep API Client
Beginnen wir mit der fundamentalen API-Konfiguration. Dieser Code bildet die Basis für alle weiteren SLA-Mechanismen:
"""
HolySheep AI Production Client mit SLA-Konfiguration
Base URL: https://api.holysheep.ai/v1
Author: HolySheep Technical Blog
"""
import requests
import time
import json
from typing import Optional, Dict, Any
from dataclasses import dataclass, field
from enum import Enum
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class CircuitState(Enum):
CLOSED = "closed" # Normaler Betrieb
OPEN = "open" # Circuit offen, Fail-Fast
HALF_OPEN = "half_open" # Test-Phase nach Timeout
@dataclass
class SLAConfig:
"""SLA-Konfiguration für Production-Betrieb"""
# Rate Limiting
requests_per_second: int = 100
burst_size: int = 200
rate_limit_window: int = 60 # Sekunden
# Retry-Configuration
max_retries: int = 3
retry_base_delay: float = 0.5 # Sekunden
retry_max_delay: float = 10.0
retry_multiplier: float = 2.0
# Circuit Breaker
failure_threshold: int = 5
success_threshold: int = 3
circuit_timeout: float = 30.0 # Sekunden
# Timeout
request_timeout: float = 10.0
# Fallback
fallback_models: list = field(default_factory=lambda: [
"deepseek-v3.2",
"gemini-2.5-flash",
"claude-sonnet-4.5"
])
class HolySheepProductionClient:
"""
Production-ready HolySheep AI Client mit:
- Rate Limiting (Token Bucket)
- Automatische Wiederholungen mit Exponential Backoff
- Circuit Breaker Pattern
- Multi-Modell Fallback
"""
def __init__(self, api_key: str, config: SLAConfig = None):
self.api_key = api_key
self.config = config or SLAConfig()
self.base_url = "https://api.holysheep.ai/v1"
# Rate Limiting State
self.tokens = self.config.burst_size
self.last_refill = time.time()
# Circuit Breaker State
self.circuit_state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
self.last_failure_time = None
self.current_model_index = 0
# Metrics
self.total_requests = 0
self.successful_requests = 0
self.failed_requests = 0
self.total_latency = 0.0
def _refill_tokens(self):
"""Token Bucket Refill Logic"""
now = time.time()
elapsed = now - self.last_refill
# Tokens pro Sekunde refill
refill_amount = elapsed * self.config.requests_per_second
self.tokens = min(
self.config.burst_size,
self.tokens + refill_amount
)
self.last_refill = now
def _acquire_token(self) -> bool:
"""Token akquirieren für Rate Limiting"""
self._refill_tokens()
if self.tokens >= 1:
self.tokens -= 1
return True
return False
def _wait_for_token(self, timeout: float = 30.0):
"""Warten bis Token verfügbar"""
start = time.time()
while not self._acquire_token():
if time.time() - start > timeout:
raise TimeoutError("Rate Limit Timeout: Kein Token verfügbar")
time.sleep(0.05)
def _check_circuit_breaker(self) -> bool:
"""Circuit Breaker State Check"""
if self.circuit_state == CircuitState.CLOSED:
return True
if self.circuit_state == CircuitState.OPEN:
# Prüfe ob Timeout vergangen
if (time.time() - self.last_failure_time) > self.config.circuit_timeout:
logger.info("Circuit: OPEN → HALF_OPEN")
self.circuit_state = CircuitState.HALF_OPEN
self.success_count = 0
return True
return False
if self.circuit_state == CircuitState.HALF_OPEN:
return True
return False
def _record_success(self):
"""Erfolgreiche Anfrage registrieren"""
self.successful_requests += 1
if self.circuit_state == CircuitState.HALF_OPEN:
self.success_count += 1
if self.success_count >= self.config.success_threshold:
logger.info("Circuit: HALF_OPEN → CLOSED")
self.circuit_state = CircuitState.CLOSED
self.failure_count = 0
elif self.circuit_state == CircuitState.CLOSED:
# Reset failure count on success
self.failure_count = max(0, self.failure_count - 1)
def _record_failure(self):
"""Fehlgeschlagene Anfrage registrieren"""
self.failed_requests += 1
self.failure_count += 1
self.last_failure_time = time.time()
if self.circuit_state == CircuitState.HALF_OPEN:
logger.warning("Circuit: HALF_OPEN → OPEN (Fallback-Test fehlgeschlagen)")
self.circuit_state = CircuitState.OPEN
elif self.failure_count >= self.config.failure_threshold:
logger.warning(f"Circuit: CLOSED → OPEN (Failures: {self.failure_count})")
self.circuit_state = CircuitState.OPEN
def _calculate_retry_delay(self, attempt: int) -> float:
"""Exponential Backoff mit Jitter"""
import random
delay = self.config.retry_base_delay * (self.config.retry_multiplier ** attempt)
delay = min(delay, self.config.retry_max_delay)
# Add jitter (0.5x to 1.5x)
jitter = delay * (0.5 + random.random())
return jitter
def _get_current_model(self) -> str:
"""Aktuelles Modell basierend auf Fallback-Strategie"""
return self.config.fallback_models[self.current_model_index]
def _rotate_model(self):
"""Zum nächsten Fallback-Modell wechseln"""
self.current_model_index = (self.current_model_index + 1) % len(self.config.fallback_models)
logger.info(f"Model rotated to: {self._get_current_model()}")
def chat_completion(
self,
messages: list,
system_prompt: str = "Du bist ein hilfreicher Assistent.",
temperature: float = 0.7,
max_tokens: int = 2048
) -> Dict[str, Any]:
"""
Chat-Completion mit vollständiger SLA-Unterstützung
"""
self.total_requests += 1
start_time = time.time()
# Rate Limiting
self._wait_for_token()
# Circuit Breaker Check
if not self._check_circuit_breaker():
logger.warning("Circuit OPEN: Returning degraded response")
return self._get_degraded_response("service_unavailable")
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": self._get_current_model(),
"messages": [
{"role": "system", "content": system_prompt},
*messages
],
"temperature": temperature,
"max_tokens": max_tokens
}
# Retry Loop
last_error = None
for attempt in range(self.config.max_retries + 1):
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=self.config.request_timeout
)
if response.status_code == 200:
result = response.json()
self._record_success()
self._rotate_model() # Reset to primary model
latency = (time.time() - start_time) * 1000
self.total_latency += latency
logger.info(f"Success: Latency={latency:.2f}ms, Model={payload['model']}")
return {
"status": "success",
"data": result,
"latency_ms": latency,
"model": payload["model"]
}
elif response.status_code == 429:
# Rate Limited by API
logger.warning(f"Rate Limited (attempt {attempt + 1})")
last_error = "rate_limited"
elif response.status_code >= 500:
# Server Error - Retry
logger.warning(f"Server Error {response.status_code} (attempt {attempt + 1})")
last_error = f"server_error_{response.status_code}"
else:
# Client Error - Don't retry
logger.error(f"Client Error: {response.status_code}")
break
except requests.exceptions.Timeout:
logger.warning(f"Timeout (attempt {attempt + 1})")
last_error = "timeout"
except requests.exceptions.RequestException as e:
logger.warning(f"Request Exception: {e} (attempt {attempt + 1})")
last_error = str(e)
# Retry mit Exponential Backoff
if attempt < self.config.max_retries:
delay = self._calculate_retry_delay(attempt)
logger.info(f"Retrying in {delay:.2f}s...")
time.sleep(delay)
# Fallback zu nächstem Modell
self._rotate_model()
payload["model"] = self._get_current_model()
# Alle Versuche fehlgeschlagen
self._record_failure()
return self._get_degraded_response(last_error or "unknown_error")
def _get_degraded_response(self, error_type: str) -> Dict[str, Any]:
"""Degradierte Antwort bei Systemausfall"""
return {
"status": "degraded",
"error": error_type,
"message": "Service temporär nicht verfügbar. Bitte versuchen Sie es später erneut.",
"fallback_available": True,
"circuit_state": self.circuit_state.value
}
def get_metrics(self) -> Dict[str, Any]:
"""Aktuelle Metriken abrufen"""
success_rate = (
self.successful_requests / self.total_requests * 100
if self.total_requests > 0 else 0
)
avg_latency = (
self.total_latency / self.successful_requests
if self.successful_requests > 0 else 0
)
return {
"total_requests": self.total_requests,
"successful": self.successful_requests,
"failed": self.failed_requests,
"success_rate": f"{success_rate:.2f}%",
"avg_latency_ms": f"{avg_latency:.2f}",
"circuit_state": self.circuit_state.value,
"current_model": self._get_current_model()
}
============ Production Usage Example ============
if __name__ == "__main__":
# Konfiguration für Production mit 99,7% SLA
production_config = SLAConfig(
requests_per_second=100,
burst_size=200,
max_retries=3,
failure_threshold=5,
circuit_timeout=30.0,
fallback_models=[
"deepseek-v3.2", # Primär: $0.42/MTok
"gemini-2.5-flash", # Fallback 1: $2.50/MTok
"claude-sonnet-4.5" # Fallback 2: $15/MTok
]
)
client = HolySheepProductionClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
config=production_config
)
# Test Request
response = client.chat_completion(
messages=[
{"role": "user", "content": "Erkläre mir Rate Limiting in 3 Sätzen."}
],
system_prompt="Du bist ein technischer Experte für Systemarchitektur.",
temperature=0.7,
max_tokens=500
)
print(f"Status: {response['status']}")
print(f"Latency: {response.get('latency_ms', 'N/A')} ms")
print(f"Model: {response.get('model', 'N/A')}")
print(f"Metrics: {client.get_metrics()}")Rate Limiting: Token Bucket Implementation
Das Rate Limiting ist der erste Verteidigungsring gegen Überlastung. Hier ist eine erweiterte Implementierung speziell für HolySheep mit Redis-Backend:
"""
Advanced Rate Limiting für HolySheep Production
Token Bucket + Sliding Window mit Redis
"""
import redis
import time
import hashlib
from typing import Tuple, Optional
from dataclasses import dataclass
import json
@dataclass
class RateLimitConfig:
"""Rate Limit Konfiguration pro Tier"""
requests_per_minute: int
tokens_per_minute: int # Input Tokens
tokens_per_response: int # Output Tokens Budget
concurrent_requests: int
# Burst Settings
burst_multiplier: float = 1.5
burst_duration_seconds: int = 10
class HolySheepRateLimiter:
"""
Production Rate Limiter mit:
- Token Bucket Algorithmus
- Sliding Window Counter
- Multi-Tenant Support (API Key Level)
-burst Handling
- Kostenbasierte Limitierung (Tokens statt Requests)
"""
# HolySheep Pricing Tiers (Stand 2026)
PRICING = {
"free": {
"rpm": 30,
"tpm_input": 150_000,
"tpm_output": 150_000,
"concurrent": 2
},
"starter": {
"rpm": 100,
"tpm_input": 500_000,
"tpm_output": 500_000,
"concurrent": 5
},
"pro": {
"rpm": 500,
"tpm_input": 2_000_000,
"tpm_output": 2_000_000,
"concurrent": 20
},
"enterprise": {
"rpm": 1000,
"tpm_input": 10_000_000,
"tpm_output": 10_000_000,
"concurrent": 100
}
}
def __init__(self, redis_host: str = "localhost", redis_port: int = 6379):
self.redis = redis.Redis(
host=redis_host,
port=redis_port,
decode_responses=True
)
self.local_cache = {} # Fallback bei Redis-Ausfall
def _get_tier_limits(self, api_key: str) -> dict:
"""API Key Tier aus Datenbank oder Cache ermitteln"""
tier_key = f"holysheep:tier:{hashlib.md5(api_key.encode()).hexdigest()}"
tier = self.redis.get(tier_key)
if tier:
return self.PRICING.get(tier, self.PRICING["starter"])
# Standard: Free Tier
return self.PRICING["free"]
def check_rate_limit(
self,
api_key: str,
request_tokens: int = 0,
expected_output_tokens: int = 0
) -> Tuple[bool, dict]:
"""
Prüft Rate Limits und gibt Status zurück
Returns:
Tuple[allowed: bool, info: dict]
"""
now = time.time()
limits = self._get_tier_limits(api_key)
# Request Rate Limit (pro Minute)
rpm_key = f"holysheep:rpm:{api_key}:{int(now // 60)}"
rpm_count = int(self.redis.get(rpm_key) or 0)
if rpm_count >= limits["requests_per_minute"]:
ttl = 60 - (now % 60)
return False, {
"error": "rate_limit_exceeded",
"limit_type": "rpm",
"limit": limits["requests_per_minute"],
"current": rpm_count,
"retry_after_seconds": int(ttl)
}
# Token Rate Limit (pro Minute)
tpm_key = f"holysheep:tpm:{api_key}:{int(now // 60)}"
tpm_used = int(self.redis.get(tpm_key) or 0)
total_tokens = request_tokens + expected_output_tokens
if (tpm_used + total_tokens) > limits["tpm_input"]:
return False, {
"error": "token_limit_exceeded",
"limit_type": "tpm",
"limit": limits["tpm_input"],
"current": tpm_used,
"requested": total_tokens,
"retry_after_seconds": int(60 - (now % 60))
}
# Concurrent Request Limit
concurrent_key = f"holysheep:concurrent:{api_key}"
concurrent_count = int(self.redis.get(concurrent_key) or 0)
if concurrent_count >= limits["concurrent"]:
return False, {
"error": "concurrent_limit_exceeded",
"limit_type": "concurrent",
"limit": limits["concurrent"],
"current": concurrent_count
}
# Alle Checks bestanden
return True, {
"allowed": True,
"rpm_remaining": limits["requests_per_minute"] - rpm_count - 1,
"tpm_remaining": limits["tpm_input"] - tpm_used - total_tokens,
"concurrent_available": limits["concurrent"] - concurrent_count - 1
}
def consume(
self,
api_key: str,
request_tokens: int,
response_tokens: int
) -> bool:
"""
Consumiert Rate Limit Kontingent nach erfolgreicher Anfrage
"""
now = time.time()
minute_key = int(now // 60)
pipe = self.redis.pipeline()
# RPM Counter
rpm_key = f"holysheep:rpm:{api_key}:{minute_key}"
pipe.incr(rpm_key)
pipe.expire(rpm_key, 120) # Keep for 2 minutes
# TPM Counter
tpm_key = f"holysheep:tpm:{api_key}:{minute_key}"
total_tokens = request_tokens + response_tokens
pipe.incrby(tpm_key, total_tokens)
pipe.expire(tpm_key, 120)
# Concurrent Counter
concurrent_key = f"holysheep:concurrent:{api_key}"
pipe.incr(concurrent_key)
try:
pipe.execute()
return True
except redis.RedisError:
# Redis Fallback - lokale Zählung
self._local_consume(api_key, request_tokens, response_tokens)
return True
def release(self, api_key: str):
"""Releases concurrent slot"""
concurrent_key = f"holysheep:concurrent:{api_key}"
self.redis.decr(concurrent_key)
def _local_consume(self, api_key: str, req_tokens: int, resp_tokens: int):
"""Fallback bei Redis-Ausfall"""
if api_key not in self.local_cache:
self.local_cache[api_key] = {
"rpm": 0,
"tpm": 0,
"minute": int(time.time() // 60)
}
cache = self.local_cache[api_key]
current_minute = int(time.time() // 60)
if cache["minute"] != current_minute:
cache["rpm"] = 0
cache["tpm"] = 0
cache["minute"] = current_minute
cache["rpm"] += 1
cache["tpm"] += req_tokens + resp_tokens
def get_usage_stats(self, api_key: str) -> dict:
"""Aktuelle Nutzungsstatistiken"""
now = time.time()
minute_key = int(now // 60)
rpm_key = f"holysheep:rpm:{api_key}:{minute_key}"
tpm_key = f"holysheep:tpm:{api_key}:{minute_key}"
concurrent_key = f"holysheep:concurrent:{api_key}"
return {
"requests_this_minute": int(self.redis.get(rpm_key) or 0),
"tokens_this_minute": int(self.redis.get(tpm_key) or 0),
"concurrent_requests": int(self.redis.get(concurrent_key) or 0),
"minute_remaining": 60 - (now % 60)
}
============ Integration mit FastAPI ============
from fastapi import FastAPI, HTTPException, Request, Depends
from fastapi.responses import JSONResponse
import httpx
app = FastAPI(title="HolySheep Production API Gateway")
rate_limiter = HolySheepRateLimiter()
HolySheep Base URL
HOLYSHEEP_BASE = "https://api.holysheep.ai/v1"
@app.post("/chat/completions")
async def chat_completions(
request: Request,
payload: dict,
api_key: str = Depends(lambda: request.headers.get("Authorization", "").replace("Bearer ", ""))
):
"""
Production API Gateway mit vollständigem Rate Limiting
"""
if not api_key:
raise HTTPException(status_code=401, detail="API Key erforderlich")
# Token-Schätzung (vereinfacht)
estimated_input_tokens = sum(
len(msg.get("content", "").split()) * 1.3
for msg in payload.get("messages", [])
)
estimated_output_tokens = payload.get("max_tokens", 2048)
# Rate Limit Check
allowed, limit_info = rate_limiter.check_rate_limit(
api_key=api_key,
request_tokens=int(estimated_input_tokens),
expected_output_tokens=estimated_output_tokens
)
if not allowed:
return JSONResponse(
status_code=429,
content={
"error": limit_info["error"],
"message": f"Rate Limit erreicht. Retry nach {limit_info.get('retry_after_seconds', 60)}s",
"details": limit_info
},
headers={"Retry-After": str(limit_info.get("retry_after_seconds", 60))}
)
# Concurrent Slot reservieren
try:
# Forward to HolySheep
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{HOLYSHEEP_BASE}/chat/completions",
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
},
json=payload
)
# Tokens konsumieren
response_tokens = len(response.text.split()) if response.status_code == 200 else 0
rate_limiter.consume(api_key, int(estimated_input_tokens), response_tokens)
return response.json()
except httpx.TimeoutException:
raise HTTPException(status_code=504, detail="HolySheep Timeout")
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
finally:
rate_limiter.release(api_key)
@app.get("/usage")
async def get_usage(api_key: str = Depends(lambda r: r.headers.get("Authorization", "").replace("Bearer ", ""))):
"""Nutzungsstatistiken abrufen"""
return rate_limiter.get_usage_stats(api_key)
@app.get("/limits")
async def get_limits(api_key: str = Depends(lambda r: r.headers.get("Authorization", "").replace("Bearer ", ""))):
"""Rate Limits für API Key abrufen"""
tier = rate_limiter._get_tier_limits(api_key)
return {
"tier": "starter", # Would come from database
"limits": tier,
"pricing": HolySheepRateLimiter.PRICING
}Circuit Breaker Pattern für Multi-Modell Fallback
Das Circuit Breaker Pattern ist entscheidend für die Resilienz bei Ausfällen einzelner Modelle. Hier ist eine Production-ready Implementierung:
"""
Circuit Breaker Implementation für HolySheep Multi-Modell Fallback
mit Prometheus Metrics Integration
"""
import time
from enum import Enum
from typing import Callable, Any, Optional, List
from dataclasses import dataclass, field
from collections import deque
import threading
import logging
from functools import wraps
try:
from prometheus_client import Counter, Histogram, Gauge, generate_latest
PROMETHEUS_AVAILABLE = True
except ImportError:
PROMETHEUS_AVAILABLE = False
logger = logging.getLogger(__name__)
class CircuitState(Enum):
CLOSED = "closed"
OPEN = "open"
HALF_OPEN = "half_open"
class FailureType(Enum):
TIMEOUT = "timeout"
RATE_LIMIT = "rate_limit"
SERVER_ERROR = "server_error"
CLIENT_ERROR = "client_error"
CONNECTION_ERROR = "connection_error"
@dataclass
class CircuitBreakerConfig:
"""Konfiguration für Circuit Breaker"""
name: str
# Failure Thresholds
failure_threshold: int = 5
success_threshold: int = 3
# Timing
open_duration: float = 30.0 # Sekunden
half_open_max_calls: int = 3
# Sliding Window
sliding_window_size: int = 100 # Anzahl der letzten Calls
failure_threshold_percentage: float = 50.0 # % Fehler in Window
# Slow Call Thresholds
slow_call_threshold: float = 5.0 # Sekunden
slow_call_percentage: float = 80.0
# Ignore Failures
ignored_failures: List[FailureType] = field(default_factory=list)
@dataclass
class CircuitBreakerMetrics:
"""Metriken für Circuit Breaker"""
calls_total: int = 0
successes: int = 0
failures: int = 0
rejected: int = 0
timeouts: int = 0
last_failure_time: Optional[float] = None
last_success_time: Optional[float] = None
state_changes: int = 0
# Sliding Window
call_history: deque = field(default_factory=lambda: deque(maxlen=100))
class ModelCircuitBreaker:
"""
Circuit Breaker speziell für HolySheep Multi-Modell Architektur
Features:
- Sliding Window Failure Tracking
- Slow Call Detection
- State Persistence für High Availability
- Prometheus Metrics Export
"""
def __init__(self, config: CircuitBreakerConfig):
self.config = config
self.metrics = CircuitBreakerMetrics()
self.state = CircuitState.CLOSED
self._lock = threading.RLock()
# Half-Open State Tracking
self._half_open_calls = 0
self._half_open_successes = 0
# Prometheus Metrics
if PROMETHEUS_AVAILABLE:
self._setup_prometheus_metrics()
def _setup_prometheus_metrics(self):
"""Prometheus Metrics initialisieren"""
self.prom_calls = Counter(
f'circuit_breaker_calls_total',
'Total number of calls',
['name', 'result']
)
self.prom_state = Gauge(
f'circuit_breaker_state',
'Current circuit state (0=closed, 1=half-open, 2=open)',
['name']
)
self.prom_latency = Histogram(
f'circuit_breaker_latency_seconds',
'Call latency',
['name']
)
def _get_state_value(self) -> int:
"""Numeric state value for Prometheus"""
return {"closed": 0, "half_open": 1, "open": 2}[self.state.value]
def _update_prometheus(self, labels: dict):
"""Prometheus Metrics aktualisieren"""
if not PROMETHEUS_AVAILABLE:
return
self.prom_state.labels(**labels).set(self._get_state_value())
def record_success(self, latency: float = None):
"""Erfolgreichen Call registrieren"""
with self._lock:
self.metrics.successes += 1
self.metrics.last_success_time = time.time()
self.metrics.calls_total += 1
# Call History
self.metrics.call_history.append({
"timestamp": time.time(),
"success": True,
"latency": latency or 0
})
# State Transitions
if self.state == CircuitState.HALF_OPEN:
self._half_open_successes += 1
if self._half_open_successes >= self.config.success_threshold:
self._transition_to(CircuitState.CLOSED)
elif self.state == CircuitState.CLOSED:
# Reset failure count on success
pass # Handled in record_failure
if PROMETHEUS_AVAILABLE:
self.prom_calls.labels(
name=self.config.name,
result="success"
).inc()
if latency:
self.prom_latency.labels(name=self.config.name).observe(latency)
def record_failure(
self,
failure_type: FailureType,
latency: float = None,
exception: Exception = None
):
"""Fehlgeschlagenen Call registrieren"""
with self._lock:
# Check if failure should be ignored
if failure_type in self.config.ignored_failures:
logger.debug(f"Ignoring {failure_type} for {self.config.name}")
return
self.metrics.failures += 1
self.metrics.last_failure_time = time.time()
self.metrics.calls_total += 1
if failure_type == FailureType.TIMEOUT:
self.metrics.timeouts += 1
# Call History
self.metrics.call_history.append({
"timestamp": time.time(),
"success": False,
"failure_type": failure_type.value,
"latency": latency or 0,
"exception": str(exception) if exception else None
})
# State Transitions
if self.state == CircuitState.HALF_OPEN:
self._
Verwandte Ressourcen
Verwandte Artikel