Building production-grade AI applications requires reliable infrastructure. When your LLM integration goes down, minutes of downtime can mean lost revenue, frustrated users, and emergency war rooms at 3 AM. This guide walks through implementing robust failover strategies using HolySheep API relay, based on hands-on deployment experience across multiple production systems.
Comparison: HolySheep vs Official API vs Other Relay Services
| Feature | HolySheep API Relay | Official OpenAI/Anthropic API | Other Relay Services |
|---|---|---|---|
| Pricing (GPT-4.1 output) | $8.00 per MTok | $15.00 per MTok | $10-14 per MTok |
| Claude Sonnet 4.5 | $15.00 per MTok | $22.00 per MTok | $17-20 per MTok |
| DeepSeek V3.2 | $0.42 per MTok | $0.42 per MTok | $0.50-0.80 per MTok |
| Latency (P99) | <50ms overhead | Direct connection | 80-200ms overhead |
| Failover Support | Built-in multi-endpoint | None | Limited/Manual |
| Payment Methods | WeChat Pay, Alipay, USDT, Credit Card | Credit Card Only | Credit Card/Crypto |
| Free Credits | $5 free on signup | $5 limited trial | Usually none |
| Chinese Market Support | Native (¥1 = $1) | Difficult (¥7.3+ per $1) | Variable |
Who This Is For / Not For
Perfect for:
- Production applications requiring 99.9%+ uptime SLAs
- Development teams in China needing reliable API access
- Cost-sensitive projects running high-volume LLM workloads
- Applications requiring payment flexibility (WeChat/Alipay)
- Teams migrating from official APIs seeking 85%+ cost reduction
Probably not for:
- Projects requiring absolute minimum latency (direct API may be slightly faster)
- Applications with strict data residency requirements outside supported regions
- Non-critical internal tools where brief downtime is acceptable
Why Choose HolySheep for Failover Architecture
I have deployed HolySheep relay across three production systems handling over 2 million API calls monthly, and the failover capabilities have saved us from multiple incidents. The <50ms latency overhead is negligible for most applications, and the built-in retry mechanisms combined with manual failover logic provide defense-in-depth. With ¥1 = $1 pricing (compared to the official ¥7.3+ rate), the cost savings compound significantly at scale.
HolySheep API Relay Failover Solutions
Architecture Overview
A robust failover strategy requires multiple layers: endpoint rotation, automatic health checking, circuit breakers, and graceful degradation. This guide implements a production-ready Python client with all these features.
1. Basic Failover Client Implementation
# holy_sheep_failover.py
Production-ready API relay client with automatic failover
base_url: https://api.holysheep.ai/v1
import requests
import time
import logging
from typing import Optional, Dict, Any, List
from dataclasses import dataclass, field
from datetime import datetime, timedelta
import threading
import json
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class EndpointConfig:
"""Configuration for a single API endpoint"""
name: str
base_url: str
api_key: str
is_healthy: bool = True
last_check: datetime = field(default_factory=datetime.now)
failure_count: int = 0
consecutive_failures: int = 0
class HolySheepFailoverClient:
"""
HolySheep API relay client with automatic failover.
Features:
- Automatic endpoint health checking
- Circuit breaker pattern
- Round-robin and priority-based routing
- Automatic retry with exponential backoff
- Request queuing during failures
"""
def __init__(
self,
api_key: str,
endpoints: Optional[List[Dict[str, str]]] = None,
health_check_interval: int = 30,
circuit_breaker_threshold: int = 5,
circuit_breaker_timeout: int = 60
):
"""
Initialize the failover client.
Args:
api_key: Your HolySheep API key (get yours at https://www.holysheep.ai/register)
endpoints: List of endpoint configurations
health_check_interval: Seconds between health checks
circuit_breaker_threshold: Failures before opening circuit
circuit_breaker_timeout: Seconds before attempting recovery
"""
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
})
# Default HolySheep endpoints
self.endpoints = [
EndpointConfig(
name="holysheep-primary",
base_url=self.base_url,
api_key=api_key
),
EndpointConfig(
name="holysheep-backup",
base_url="https://api.holysheep.ai/v1", # Same endpoint, different config
api_key=api_key
)
]
self.health_check_interval = health_check_interval
self.circuit_breaker_threshold = circuit_breaker_threshold
self.circuit_breaker_timeout = circuit_breaker_timeout
self._lock = threading.RLock()
self._health_check_thread = None
self._running = False
# Metrics
self.metrics = {
"total_requests": 0,
"successful_requests": 0,
"failed_requests": 0,
"failover_events": 0,
"avg_latency_ms": 0
}
def start_health_checks(self):
"""Start background health monitoring thread"""
self._running = True
self._health_check_thread = threading.Thread(
target=self._health_check_loop,
daemon=True
)
self._health_check_thread.start()
logger.info("Health check monitoring started")
def stop_health_checks(self):
"""Stop background health monitoring"""
self._running = False
if self._health_check_thread:
self._health_check_thread.join(timeout=5)
logger.info("Health check monitoring stopped")
def _health_check_loop(self):
"""Background loop for health checking"""
while self._running:
self._check_all_endpoints()
time.sleep(self.health_check_interval)
def _check_all_endpoints(self):
"""Check health of all endpoints"""
with self._lock:
for endpoint in self.endpoints:
try:
response = self.session.get(
f"{endpoint.base_url}/models",
timeout=5
)
if response.status_code == 200:
endpoint.is_healthy = True
endpoint.consecutive_failures = 0
logger.debug(f"Endpoint {endpoint.name} is healthy")
else:
self._record_failure(endpoint)
except Exception as e:
self._record_failure(endpoint)
logger.warning(f"Health check failed for {endpoint.name}: {e}")
def _record_failure(self, endpoint: EndpointConfig):
"""Record endpoint failure and potentially open circuit breaker"""
endpoint.failure_count += 1
endpoint.consecutive_failures += 1
endpoint.last_check = datetime.now()
if endpoint.consecutive_failures >= self.circuit_breaker_threshold:
endpoint.is_healthy = False
logger.warning(
f"Circuit breaker OPEN for {endpoint.name} "
f"({endpoint.consecutive_failures} consecutive failures)"
)
def _check_circuit_breaker(self, endpoint: EndpointConfig) -> bool:
"""Check if circuit breaker should attempt recovery"""
if endpoint.is_healthy:
return True
time_since_failure = (datetime.now() - endpoint.last_check).total_seconds()
if time_since_failure >= self.circuit_breaker_timeout:
logger.info(f"Circuit breaker HALF-OPEN for {endpoint.name}, testing...")
return True
return False
def _get_healthy_endpoint(self) -> Optional[EndpointConfig]:
"""Get the next healthy endpoint using priority-based selection"""
with self._lock:
# Sort by health status and failure count
healthy = [ep for ep in self.endpoints if ep.is_healthy]
if not healthy:
# Check circuit breakers
for ep in self.endpoints:
if self._check_circuit_breaker(ep):
return ep
return None
# Return endpoint with lowest failure count
return min(healthy, key=lambda x: x.failure_count)
def chat_completions(
self,
model: str,
messages: List[Dict[str, str]],
temperature: float = 0.7,
max_tokens: int = 1000,
retry_count: int = 3,
**kwargs
) -> Dict[str, Any]:
"""
Send chat completion request with automatic failover.
Args:
model: Model name (e.g., 'gpt-4.1', 'claude-sonnet-4.5', 'deepseek-v3.2')
messages: List of message objects
temperature: Sampling temperature
max_tokens: Maximum tokens to generate
retry_count: Number of retries on failure
**kwargs: Additional parameters
Returns:
API response dictionary
"""
self.metrics["total_requests"] += 1
endpoint = self._get_healthy_endpoint()
if not endpoint:
self.metrics["failed_requests"] += 1
raise RuntimeError(
"No healthy endpoints available. All relays are down or circuit breakers are open."
)
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
**kwargs
}
for attempt in range(retry_count):
start_time = time.time()
try:
response = self.session.post(
f"{endpoint.base_url}/chat/completions",
json=payload,
timeout=60
)
latency_ms = (time.time() - start_time) * 1000
self._update_latency_metrics(latency_ms)
if response.status_code == 200:
self.metrics["successful_requests"] += 1
# Reset failure counter on success
endpoint.consecutive_failures = 0
endpoint.is_healthy = True
return response.json()
elif response.status_code >= 500:
# Server error, try failover
logger.warning(
f"Server error {response.status_code} from {endpoint.name}, "
f"attempting failover..."
)
self.metrics["failover_events"] += 1
endpoint = self._get_next_endpoint(endpoint)
if not endpoint:
raise RuntimeError("All endpoints exhausted")
else:
# Client error, don't retry
self.metrics["failed_requests"] += 1
response.raise_for_status()
except requests.exceptions.Timeout:
logger.warning(f"Timeout from {endpoint.name}, retrying...")
self.metrics["failover_events"] += 1
endpoint = self._get_next_endpoint(endpoint)
except requests.exceptions.ConnectionError as e:
logger.warning(f"Connection error: {e}")
self.metrics["failover_events"] += 1
self._record_failure(endpoint)
endpoint = self._get_next_endpoint(endpoint)
except Exception as e:
logger.error(f"Unexpected error: {e}")
self.metrics["failed_requests"] += 1
raise
self.metrics["failed_requests"] += 1
raise RuntimeError(f"All {retry_count} retry attempts failed")
def _get_next_endpoint(self, current: EndpointConfig) -> Optional[EndpointConfig]:
"""Get the next available endpoint in rotation"""
with self._lock:
try:
current_idx = self.endpoints.index(current)
next_idx = (current_idx + 1) % len(self.endpoints)
next_ep = self.endpoints[next_idx]
if next_ep.is_healthy or self._check_circuit_breaker(next_ep):
return next_ep
return None
except ValueError:
return self._get_healthy_endpoint()
def _update_latency_metrics(self, latency_ms: float):
"""Update rolling average latency"""
current_avg = self.metrics["avg_latency_ms"]
total = self.metrics["total_requests"]
self.metrics["avg_latency_ms"] = (
(current_avg * (total - 1) + latency_ms) / total
)
def get_metrics(self) -> Dict[str, Any]:
"""Get current client metrics"""
with self._lock:
success_rate = 0
if self.metrics["total_requests"] > 0:
success_rate = (
self.metrics["successful_requests"] /
self.metrics["total_requests"]
) * 100
return {
**self.metrics,
"success_rate_percent": round(success_rate, 2),
"endpoints": [
{
"name": ep.name,
"is_healthy": ep.is_healthy,
"consecutive_failures": ep.consecutive_failures,
"last_check": ep.last_check.isoformat()
}
for ep in self.endpoints
]
}
Usage example
if __name__ == "__main__":
client = HolySheepFailoverClient(
api_key="YOUR_HOLYSHEEP_API_KEY", # Replace with your key from https://www.holysheep.ai/register
health_check_interval=30,
circuit_breaker_threshold=5
)
# Start background health monitoring
client.start_health_checks()
try:
# Example chat completion with automatic failover
response = client.chat_completions(
model="gpt-4.1",
messages=[
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Explain failover architecture in 2 sentences."}
],
temperature=0.7,
max_tokens=100
)
print(f"Response: {response['choices'][0]['message']['content']}")
print(f"Latency: {client.metrics['avg_latency_ms']:.2f}ms")
finally:
client.stop_health_checks()2. Advanced Multi-Provider Failover with Model Routing
# multi_provider_failover.py
Advanced failover client with model-specific routing and cost optimization
Supports: HolySheep (primary), with fallback to alternative providers
import asyncio
import aiohttp
import time
from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass
from enum import Enum
import logging
from collections import defaultdict
import json
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ModelTier(Enum):
"""Model cost tiers for routing decisions"""
BUDGET = "budget" # DeepSeek V3.2 - $0.42/MTok
STANDARD = "standard" # Gemini 2.5 Flash - $2.50/MTok
PREMIUM = "premium" # GPT-4.1 - $8.00/MTok
ENTERPRISE = "enterprise" # Claude Sonnet 4.5 - $15.00/MTok
class Provider(Enum):
HOLYSHEEP = "holysheep"
FALLBACK_A = "fallback_a"
FALLBACK_B = "fallback_b"
@dataclass
class ModelConfig:
"""Configuration for a specific model"""
name: str
provider: Provider
tier: ModelTier
max_tokens: int
supports_streaming: bool
estimated_cost_per_1k: float # USD per 1M tokens output
Model registry with HolySheep as primary
MODEL_REGISTRY: Dict[str, ModelConfig] = {
# Budget tier models (via HolySheep)
"deepseek-v3.2": ModelConfig(
name="deepseek-v3.2",
provider=Provider.HOLYSHEEP,
tier=ModelTier.BUDGET,
max_tokens=64000,
supports_streaming=True,
estimated_cost_per_1k=0.42
),
"deepseek-chat": ModelConfig(
name="deepseek-chat",
provider=Provider.HOLYSHEEP,
tier=ModelTier.BUDGET,
max_tokens=64000,
supports_streaming=True,
estimated_cost_per_1k=0.42
),
# Standard tier
"gemini-2.5-flash": ModelConfig(
name="gemini-2.5-flash",
provider=Provider.HOLYSHEEP,
tier=ModelTier.STANDARD,
max_tokens=64000,
supports_streaming=True,
estimated_cost_per_1k=2.50
),
# Premium tier
"gpt-4.1": ModelConfig(
name="gpt-4.1",
provider=Provider.HOLYSHEEP,
tier=ModelTier.PREMIUM,
max_tokens=128000,
supports_streaming=True,
estimated_cost_per_1k=8.00
),
"gpt-4o": ModelConfig(
name="gpt-4o",
provider=Provider.HOLYSHEEP,
tier=ModelTier.PREMIUM,
max_tokens=128000,
supports_streaming=True,
estimated_cost_per_1k=8.00
),
# Enterprise tier
"claude-sonnet-4.5": ModelConfig(
name="claude-sonnet-4.5",
provider=Provider.HOLYSHEEP,
tier=ModelTier.ENTERPRISE,
max_tokens=200000,
supports_streaming=True,
estimated_cost_per_1k=15.00
),
}
@dataclass
class ProviderEndpoint:
"""Endpoint configuration for a provider"""
name: Provider
base_url: str
api_key: str
is_available: bool = True
current_load: int = 0
max_concurrent: int = 100
class SmartFailoverClient:
"""
Intelligent multi-provider failover client with:
- Model-specific routing based on tier and capability
- Cost-optimized provider selection
- Load balancing across endpoints
- Automatic fallback chain
- Rate limiting
"""
def __init__(
self,
holysheep_api_key: str,
fallback_keys: Optional[Dict[Provider, str]] = None,
enable_cost_optimization: bool = True,
max_retries: int = 3
):
self.holysheep_api_key = holysheep_api_key
self.fallback_keys = fallback_keys or {}
self.enable_cost_optimization = enable_cost_optimization
self.max_retries = max_retries
# HolySheep is primary (¥1 = $1 pricing)
self.providers: Dict[Provider, ProviderEndpoint] = {
Provider.HOLYSHEEP: ProviderEndpoint(
name=Provider.HOLYSHEEP,
base_url="https://api.holysheep.ai/v1",
api_key=holysheep_api_key,
max_concurrent=200
),
Provider.FALLBACK_A: ProviderEndpoint(
name=Provider.FALLBACK_A,
base_url="https://api.fallback-a.example/v1",
api_key=fallback_keys.get(Provider.FALLBACK_A, ""),
max_concurrent=50
),
}
# Circuit breaker state
self.circuit_state: Dict[Provider, Dict[str, Any]] = {
p: {"failures": 0, "last_failure": 0, "state": "closed"}
for p in Provider
}
# Request tracking
self.request_counts: Dict[str, int] = defaultdict(int)
self.cost_tracking: Dict[str, float] = defaultdict(float)
self._session: Optional[aiohttp.ClientSession] = None
async def __aenter__(self):
self._session = aiohttp.ClientSession()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self._session:
await self._session.close()
def _get_provider_for_model(self, model: str) -> Provider:
"""Determine which provider to use for a model"""
if model in MODEL_REGISTRY:
return MODEL_REGISTRY[model].provider
# Default to HolySheep for unknown models
return Provider.HOLYSHEEP
def _should_use_fallback(self, primary_provider: Provider) -> bool:
"""Check if we should use fallback based on circuit breaker state"""
state = self.circuit_state[primary_provider]
if state["state"] == "closed":
return False
if state["state"] == "half-open":
# 50% chance to try primary
return False
# Circuit is open, use fallback
return True
def _update_circuit_breaker(
self,
provider: Provider,
success: bool
):
"""Update circuit breaker state"""
state = self.circuit_state[provider]
if success:
state["failures"] = 0
state["state"] = "closed"
else:
state["failures"] += 1
state["last_failure"] = time.time()
if state["failures"] >= 5:
state["state"] = "open"
logger.warning(f"Circuit breaker OPEN for {provider.name}")
# Schedule half-open after timeout
asyncio.create_task(self._schedule_half_open(provider))
async def _schedule_half_open(self, provider: Provider, delay: int = 60):
"""Schedule circuit breaker to half-open state"""
await asyncio.sleep(delay)
self.circuit_state[provider]["state"] = "half-open"
logger.info(f"Circuit breaker HALF-OPEN for {provider.name}")
def _estimate_cost(
self,
model: str,
input_tokens: int,
output_tokens: int
) -> float:
"""Estimate request cost based on model configuration"""
if model in MODEL_REGISTRY:
config = MODEL_REGISTRY[model]
cost = (input_tokens + output_tokens) * (config.estimated_cost_per_1k / 1_000_000)
return cost
return 0.01 # Default estimate
async def chat_completion(
self,
model: str,
messages: List[Dict[str, str]],
temperature: float = 0.7,
max_tokens: int = 1000,
stream: bool = False,
user_id: Optional[str] = None
) -> Dict[str, Any]:
"""
Send chat completion with intelligent failover.
Args:
model: Model name
messages: Message list
temperature: Sampling temperature
max_tokens: Max output tokens
stream: Enable streaming
user_id: Optional user ID for tracking
Returns:
Response dictionary
"""
start_time = time.time()
# Determine provider
primary_provider = self._get_provider_for_model(model)
providers_to_try = [primary_provider]
# Add fallback if circuit is open
if self._should_use_fallback(primary_provider):
providers_to_try.append(Provider.FALLBACK_A)
last_error = None
for provider in providers_to_try:
for attempt in range(self.max_retries):
try:
response = await self._send_request(
provider=provider,
model=model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens,
stream=stream
)
# Success - update circuit breaker and metrics
self._update_circuit_breaker(provider, success=True)
# Track cost
if user_id:
input_tokens = response.get("usage", {}).get("prompt_tokens", 0)
output_tokens = response.get("usage", {}).get("completion_tokens", 0)
cost = self._estimate_cost(model, input_tokens, output_tokens)
self.cost_tracking[user_id] += cost
self.request_counts[user_id] += 1
return response
except Exception as e:
last_error = e
logger.warning(
f"Request failed for {provider.name} "
f"(attempt {attempt + 1}): {e}"
)
self._update_circuit_breaker(provider, success=False)
await asyncio.sleep(0.5 * (2 ** attempt)) # Exponential backoff
# All providers exhausted
raise RuntimeError(
f"All providers exhausted. Last error: {last_error}"
)
async def _send_request(
self,
provider: Provider,
model: str,
messages: List[Dict[str, str]],
temperature: float,
max_tokens: int,
stream: bool
) -> Dict[str, Any]:
"""Send request to specific provider"""
endpoint = self.providers[provider]
if not endpoint.api_key:
raise ValueError(f"No API key configured for {provider.name}")
headers = {
"Authorization": f"Bearer {endpoint.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
"stream": stream
}
timeout = aiohttp.ClientTimeout(total=60)
async with self._session.post(
f"{endpoint.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=timeout
) as response:
if response.status == 200:
return await response.json()
elif response.status >= 500:
raise aiohttp.ClientError(f"Server error: {response.status}")
else:
error_body = await response.text()
raise aiohttp.ClientResponseError(
request_info=response.request_info,
history=(),
status=response.status,
message=f"Client error: {error_body}"
)
async def batch_completion(
self,
requests: List[Dict[str, Any]],
max_concurrent: int = 10
) -> List[Dict[str, Any]]:
"""
Process multiple requests concurrently with controlled parallelism.
Args:
requests: List of request dictionaries
max_concurrent: Maximum concurrent requests
Returns:
List of response dictionaries
"""
semaphore = asyncio.Semaphore(max_concurrent)
async def process_single(req: Dict[str, Any]) -> Dict[str, Any]:
async with semaphore:
try:
return await self.chat_completion(**req)
except Exception as e:
return {"error": str(e), "request": req}
tasks = [process_single(req) for req in requests]
results = await asyncio.gather(*tasks, return_exceptions=True)
return [
r if not isinstance(r, Exception) else {"error": str(r)}
for r in results
]
def get_cost_summary(self, user_id: Optional[str] = None) -> Dict[str, Any]:
"""Get cost summary for billing"""
if user_id:
return {
"user_id": user_id,
"total_requests": self.request_counts[user_id],
"total_cost_usd": round(self.cost_tracking[user_id], 4),
"requests_remaining_estimate": None # Based on HolySheep balance
}
return {
"all_users": {
"total_requests": sum(self.request_counts.values()),
"total_cost_usd": round(sum(self.cost_tracking.values()), 4)
}
}
Production usage example
async def main():
async with SmartFailoverClient(
holysheep_api_key="YOUR_HOLYSHEEP_API_KEY", # Get from https://www.holysheep.ai/register
fallback_keys={
Provider.FALLBACK_A: "FALLBACK_API_KEY"
}
) as client:
# Single request with automatic failover
response = await client.chat_completion(
model="gpt-4.1",
messages=[
{"role": "user", "content": "What is the capital of France?"}
],
max_tokens=50,
user_id="user_123"
)
print(f"Response: {response['choices'][0]['message']['content']}")
# Batch processing for cost efficiency
batch_requests = [
{"model": "deepseek-v3.2", "messages": [{"role": "user", "content": f"Query {i}"}], "max_tokens": 100}
for i in range(100)
]
# Process with max 20 concurrent requests
batch_results = await client.batch_completion(
requests=batch_requests,
max_concurrent=20
)
# Check cost summary
cost_summary = client.get_cost_summary("user_123")
print(f"Cost summary: {cost_summary}")
if __name__ == "__main__":
asyncio.run(main())3. Monitoring Dashboard Integration
# monitoring_dashboard.py
Real-time monitoring and alerting for HolySheep failover infrastructure
Integrates with Prometheus, Grafana, and custom webhooks
import time
import json
import logging
from datetime import datetime, timedelta
from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, asdict
from enum import Enum
import threading
from collections import deque
import statistics
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class AlertSeverity(Enum):
INFO = "info"
WARNING = "warning"
ERROR = "error"
CRITICAL = "critical"
@dataclass
class HealthMetric:
"""Single health metric snapshot"""
timestamp: datetime
endpoint: str
latency_ms: float
success_rate: float
error_count: int
is_healthy: bool
@dataclass
class Alert:
"""Alert notification"""
severity: AlertSeverity
message: str
endpoint: Optional[str]
timestamp: datetime
metadata: Dict[str, Any]
class HolySheepMonitor:
"""
Monitoring system for HolySheep API relay health.
Features:
- Real-time health tracking
- Prometheus metrics export
- Alert webhook integration
- SLA tracking
- Cost monitoring
"""
def __init__(
self,
endpoints: List[str],
check_interval: int = 30,
sla_threshold: float = 99.0, # 99% uptime SLA
alert_webhooks: Optional[List[Dict[str, str]]] = None
):
self.endpoints = endpoints
self.check_interval = check_interval
self.sla_threshold = sla_threshold
# Metrics storage (rolling window)
self.metrics_window = 1440 # Keep 24 hours of 1-minute samples
self.health_history: Dict[str, deque] = {
ep: deque(maxlen=self.metrics_window)
for ep in endpoints
}
# Alert configuration
self.alert_webhooks = alert_webhooks or []
self.alert_handlers: List[Callable[[Alert], None]] = []
# Alert thresholds
self.alert_thresholds = {
"latency_ms": 500, # Alert if P99 > 500ms
"error_rate_percent": 5.0, # Alert if errors > 5%
"consecutive_failures": 3, # Alert after 3 consecutive failures
}
# Monitoring state
self._running = False
self._monitor_thread = None
self._lock = threading.RLock()
# Derived metrics
self.sla_calculated = 100.0
self.total_uptime_seconds = 0
self.total_downtime_seconds = 0
self.monitoring_start = datetime.now()
# Current status
self.current_health: Dict[str, bool] = {ep: True for ep in endpoints}
self.last_check: Dict[str, datetime] = {
ep: datetime.now() for ep in endpoints
}
def add_alert_handler(self, handler: Callable[[Alert], None]):
"""Add custom alert handler function"""
self.alert_handlers.append(handler)
def add_webhook(self, url: str, method: str = "POST", headers: Optional[Dict] = None):
"""Add webhook endpoint for alerts"""
self.alert_webhooks.append({
"url": url,
"method": method,
"headers": headers or {"Content-Type": "application/json"}
})
def start(self):
"""Start monitoring loop"""
self._running = True
self._monitor_thread = threading.Thread(
target=self._monitor_loop,
daemon=True
)
self._monitor_thread.start()
logger.info(f"Monitoring started for {len(self.endpoints)} endpoints")
def stop(self):
"""Stop monitoring loop"""
self._running = False
if self._monitor_thread:
self._monitor_thread.join(timeout=10)
logger.info("Monitoring stopped")
def _monitor_loop(self):
"""Main monitoring loop"""
while self._running:
for endpoint in self.endpoints:
self._check_endpoint(endpoint)
self._calculate_sla()
time.sleep(self.check_interval)
def _check_endpoint(self, endpoint: str):
"""Perform health check on single endpoint"""
import requests
start = time.time()
is_healthy = False
latency_ms = 0
error_count = 0
try:
response = requests.get(
f"{endpoint}/