Ba tháng trước, tôi nhận được cuộc gọi lúc 2 giờ sáng từ khách hàng — hệ thống RAG doanh nghiệp của họ phục vụ 50,000 người dùng đột nhiên trả về toàn response rỗng. Root cause? Một exception không được handle đã khiến toàn bộ service pool bị deadlock. Kể từ đó, tôi xây dựng một hệ thống Exception Pattern AI Service Monitoring hoàn chỉnh, và hôm nay sẽ chia sẻ toàn bộ kiến thức thực chiến với các bạn.
Tại Sao Monitoring AI Service Khác Với Monitoring Thông Thường?
AI API calls có đặc thù riêng: latency không deterministic (từ 45ms đến 30 giây), response có thể rất dài, và các lỗi không phải lúc nào cũng throw exception rõ ràng. Một response trả về empty string có thể là rate_limit, context_length_exceeded, hoặc đơn giản là model không hiểu prompt.
Kiến Trúc Exception Pattern Monitoring
Trước khi đi vào code, các bạn cần hiểu architecture tổng thể:
┌─────────────────────────────────────────────────────────────┐
│ API Gateway Layer │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────┐ │
│ │ Rate Limit │ │ Auth Check │ │ Request Validation │ │
│ │ Handler │ │ Middleware│ │ Middleware │ │
│ └─────────────┘ └─────────────┘ └─────────────────────┘ │
└────────────────────────────┬────────────────────────────────┘
│
┌────────────────────────────▼────────────────────────────────┐
│ AI Service Abstraction │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ HolySheep AI (base_url: https://api.holysheep.ai) │ │
│ │ - DeepSeek V3.2: $0.42/MTok │ │
│ │ - GPT-4.1: $8/MTok │ │
│ │ - Gemini 2.5 Flash: $2.50/MTok │ │
│ └─────────────────────────────────────────────────────┘ │
└────────────────────────────┬────────────────────────────────┘
│
┌────────────────────────────▼────────────────────────────────┐
│ Exception Pattern Layer │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────┐ │
│ │TimeoutErr│ │RateLimit │ │AuthError │ │ ContextLen │ │
│ │ Pattern │ │ Pattern │ │ Pattern │ │ Pattern │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────────┘ │
└─────────────────────────────────────────────────────────────┘
│
┌────────────────────────────▼────────────────────────────────┐
│ Monitoring & Alerting │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────┐ │
│ │ Prometheus │ │ Grafana │ │ Slack/PagerDuty │ │
│ │ Metrics │ │ Dashboard │ │ Alert Channel │ │
│ └──────────────┘ └──────────────┘ └──────────────────┘ │
└─────────────────────────────────────────────────────────────┘
1. Triển Khai Exception Pattern Classes
Đầu tiên, tôi xây dựng một hệ thống phân loại exception chi tiết. Đây là cách tôi tổ chức trong dự án thực tế:
# exceptions.py - HolySheep AI Service Exception Patterns
from enum import Enum
from dataclasses import dataclass, field
from typing import Optional, Dict, Any
from datetime import datetime
import json
import traceback
class ExceptionSeverity(Enum):
"""Mức độ nghiêm trọng của exception - critical cho production"""
LOW = "low" # Retry thành công, không cần alert
MEDIUM = "medium" # Retry thất bại, cần investigate
HIGH = "high" # Service degraded, cần attention
CRITICAL = "critical" # System down, alert immediately
@dataclass
class AIExceptionContext:
"""Context đầy đủ cho việc debug và monitoring"""
exception_type: str
message: str
severity: ExceptionSeverity
timestamp: datetime = field(default_factory=datetime.utcnow)
provider: str = "holysheep"
model: Optional[str] = None
latency_ms: Optional[float] = None
request_id: Optional[str] = None
user_id: Optional[str] = None
retry_count: int = 0
raw_response: Optional[Dict[str, Any]] = None
stack_trace: Optional[str] = None
def to_prometheus_labels(self) -> Dict[str, str]:
"""Convert sang Prometheus labels format"""
return {
"exception_type": self.exception_type,
"severity": self.severity.value,
"provider": self.provider,
"model": self.model or "unknown",
"retry_count": str(self.retry_count)
}
class BaseAIException(Exception):
"""Base exception class cho tất cả AI service errors"""
def __init__(
self,
message: str,
severity: ExceptionSeverity = ExceptionSeverity.MEDIUM,
context: Optional[AIExceptionContext] = None,
retryable: bool = True
):
super().__init__(message)
self.message = message
self.severity = severity
self.context = context or AIExceptionContext(
exception_type=self.__class__.__name__,
message=message,
severity=severity
)
self.retryable = retryable
def to_dict(self) -> Dict[str, Any]:
return {
"exception_class": self.__class__.__name__,
"message": self.message,
"severity": self.severity.value,
"retryable": self.retryable,
"timestamp": self.context.timestamp.isoformat(),
"provider": self.context.provider,
"context": {
"model": self.context.model,
"latency_ms": self.context.latency_ms,
"retry_count": self.context.retry_count
}
}
Specific Exception Patterns
class TimeoutException(BaseAIException):
"""Timeout - có thể retry với exponential backoff"""
def __init__(self, message: str, timeout_ms: int = 30000, **kwargs):
super().__init__(
message=message,
severity=ExceptionSeverity.MEDIUM,
retryable=True,
**kwargs
)
self.timeout_ms = timeout_ms
class RateLimitException(BaseAIException):
"""Rate limit - cần backoff theo retry-after header"""
def __init__(self, message: str, retry_after: int = 60, **kwargs):
super().__init__(
message=message,
severity=ExceptionSeverity.MEDIUM,
retryable=True,
**kwargs
)
self.retry_after = retry_after
class AuthenticationException(BaseAIException):
"""Auth error - KHÔNG retry, cần fix config ngay"""
def __init__(self, message: str, **kwargs):
super().__init__(
message=message,
severity=ExceptionSeverity.CRITICAL,
retryable=False,
**kwargs
)
class ContextLengthExceededException(BaseAIException):
"""Context quá dài - cần truncate hoặc chunk documents"""
def __init__(
self,
message: str,
context_limit: int = 128000,
current_tokens: int = 0,
**kwargs
):
super().__init__(
message=message,
severity=ExceptionSeverity.MEDIUM,
retryable=False, # Retry sẽ thất bại nếu không truncate
**kwargs
)
self.context_limit = context_limit
self.current_tokens = current_tokens
class InvalidResponseException(BaseAIException):
"""Response không hợp lệ - có thể là model bug hoặc prompt issue"""
def __init__(self, message: str, raw_response: Optional[Dict] = None, **kwargs):
super().__init__(
message=message,
severity=ExceptionSeverity.HIGH,
retryable=True,
**kwargs
)
self.raw_response = raw_response
class ServiceUnavailableException(BaseAIException):
"""Service down - retry với circuit breaker pattern"""
def __init__(self, message: str, is_retryable: bool = True, **kwargs):
super().__init__(
message=message,
severity=ExceptionSeverity.CRITICAL if not is_retryable else ExceptionSeverity.HIGH,
retryable=is_retryable,
**kwargs
)
print("✅ Exception patterns loaded - Ready for production monitoring")
2. HolySheep AI Client Với Built-in Monitoring
Đây là client hoàn chỉnh tích hợp monitoring. Lưu ý: base_url phải là https://api.holysheep.ai/v1 — không dùng OpenAI endpoint:
# holy_sheep_client.py - Production-ready AI client với monitoring
import requests
import time
import asyncio
from typing import Optional, Dict, Any, List, Callable
from dataclasses import dataclass
import logging
from datetime import datetime
Import exception patterns
from exceptions import (
BaseAIException,
TimeoutException,
RateLimitException,
AuthenticationException,
ContextLengthExceededException,
InvalidResponseException,
ServiceUnavailableException,
ExceptionSeverity,
AIExceptionContext
)
logger = logging.getLogger(__name__)
@dataclass
class MonitoringMetrics:
"""Metrics cho Prometheus/Grafana"""
total_requests: int = 0
successful_requests: int = 0
failed_requests: int = 0
total_latency_ms: float = 0.0
timeout_count: int = 0
rate_limit_count: int = 0
auth_error_count: int = 0
context_length_count: int = 0
other_error_count: int = 0
def record_request(self, latency_ms: float, success: bool, exception_type: Optional[str] = None):
self.total_requests += 1
self.total_latency_ms += latency_ms
if success:
self.successful_requests += 1
else:
self.failed_requests += 1
if exception_type:
count_attr = f"{exception_type.lower().replace('exception', '')}_count"
if hasattr(self, count_attr):
setattr(self, count_attr, getattr(self, count_attr) + 1)
@property
def success_rate(self) -> float:
if self.total_requests == 0:
return 0.0
return (self.successful_requests / self.total_requests) * 100
@property
def avg_latency_ms(self) -> float:
if self.total_requests == 0:
return 0.0
return self.total_latency_ms / self.total_requests
class HolySheepAIClient:
"""
Production-ready client cho HolySheep AI API
base_url: https://api.holysheep.ai/v1
Pricing (2026):
- DeepSeek V3.2: $0.42/MTok (tiết kiệm 85%+ so với GPT-4)
- GPT-4.1: $8/MTok
- Gemini 2.5 Flash: $2.50/MTok
"""
def __init__(
self,
api_key: str = "YOUR_HOLYSHEEP_API_KEY", # Thay bằng key thực
base_url: str = "https://api.holysheep.ai/v1",
default_model: str = "deepseek-v3.2",
timeout_seconds: int = 60,
max_retries: int = 3,
retry_delay: float = 1.0,
on_exception: Optional[Callable[[BaseAIException], None]] = None
):
self.api_key = api_key
self.base_url = base_url.rstrip('/')
self.default_model = default_model
self.timeout_seconds = timeout_seconds
self.max_retries = max_retries
self.retry_delay = retry_delay
self.on_exception = on_exception
# Monitoring
self.metrics = MonitoringMetrics()
# Supported models với pricing
self.model_pricing = {
"deepseek-v3.2": {"input": 0.42, "output": 1.68, "currency": "USD"},
"gpt-4.1": {"input": 8.0, "output": 24.0, "currency": "USD"},
"gpt-4.1-mini": {"input": 2.0, "output": 8.0, "currency": "USD"},
"gemini-2.5-flash": {"input": 2.50, "output": 10.0, "currency": "USD"},
"claude-sonnet-4.5": {"input": 15.0, "output": 75.0, "currency": "USD"}
}
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
})
# Circuit breaker state
self._circuit_open = False
self._circuit_opened_at: Optional[datetime] = None
self._circuit_timeout_seconds = 300 # 5 phút
def _parse_error_response(self, status_code: int, response_data: Dict) -> BaseAIException:
"""Parse error response từ HolySheep API"""
error_type = response_data.get("error", {}).get("type", "unknown")
error_message = response_data.get("error", {}).get("message", "Unknown error")
error_code = response_data.get("error", {}).get("code")
# Map error types sang exception classes
error_mapping = {
"authentication_error": AuthenticationException,
"rate_limit_error": RateLimitException,
"context_length_exceeded": ContextLengthExceededException,
"server_error": ServiceUnavailableException,
"timeout": TimeoutException,
}
exception_class = error_mapping.get(error_type, ServiceUnavailableException)
# Extract retry_after từ headers nếu có
retry_after = int(response_data.get("error", {}).get("retry_after", 60))
if exception_class == RateLimitException:
return exception_class(
message=error_message,
retry_after=retry_after,
context=AIExceptionContext(
exception_type=exception_class.__name__,
message=error_message,
severity=ExceptionSeverity.MEDIUM,
provider="holysheep",
raw_response=response_data
)
)
elif exception_class == ContextLengthExceededException:
return exception_class(
message=error_message,
context_limit=response_data.get("error", {}).get("limit", 128000),
current_tokens=response_data.get("error", {}).get("current_tokens", 0),
context=AIExceptionContext(
exception_type=exception_class.__name__,
message=error_message,
severity=ExceptionSeverity.MEDIUM,
provider="holysheep"
)
)
elif exception_class == AuthenticationException:
return exception_class(
message=f"Authentication failed: {error_message}",
context=AIExceptionContext(
exception_type="AuthenticationException",
message=error_message,
severity=ExceptionSeverity.CRITICAL,
provider="holysheep"
)
)
else:
return exception_class(
message=error_message,
context=AIExceptionContext(
exception_type=exception_class.__name__,
message=error_message,
severity=ExceptionSeverity.HIGH if status_code >= 500 else ExceptionSeverity.MEDIUM,
provider="holysheep"
)
)
def _check_circuit_breaker(self) -> bool:
"""Kiểm tra circuit breaker - mở sau 5 failures liên tiếp"""
if self._circuit_open:
if self._circuit_opened_at:
elapsed = (datetime.utcnow() - self._circuit_opened_at).total_seconds()
if elapsed >= self._circuit_timeout_seconds:
logger.info("🔄 Circuit breaker closing - attempting recovery")
self._circuit_open = False
return True
return False
return True
def chat_completion(
self,
messages: List[Dict[str, str]],
model: Optional[str] = None,
temperature: float = 0.7,
max_tokens: int = 4096,
user_id: Optional[str] = None
) -> Dict[str, Any]:
"""
Gọi HolySheep Chat Completion API
Args:
messages: List of message objects [{"role": "user", "content": "..."}]
model: Model name (default: deepseek-v3.2)
temperature: Sampling temperature (0-2)
max_tokens: Maximum tokens trong response
user_id: User identifier cho tracking
Returns:
Response dict với usage information
"""
model = model or self.default_model
start_time = time.time()
request_id = f"req_{int(start_time * 1000)}"
# Check circuit breaker
if not self._check_circuit_breaker():
raise ServiceUnavailableException(
message="Circuit breaker is OPEN - service temporarily unavailable",
is_retryable=True,
context=AIExceptionContext(
exception_type="CircuitBreakerOpen",
message="Service unavailable due to circuit breaker",
severity=ExceptionSeverity.CRITICAL,
latency_ms=(time.time() - start_time) * 1000
)
)
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
last_exception: Optional[BaseAIException] = None
for attempt in range(self.max_retries):
try:
response = self.session.post(
f"{self.base_url}/chat/completions",
json=payload,
timeout=self.timeout_seconds
)
latency_ms = (time.time() - start_time) * 1000
if response.status_code == 200:
data = response.json()
# Build response với monitoring data
result = {
"id": data.get("id", request_id),
"model": data.get("model", model),
"content": data["choices"][0]["message"]["content"],
"usage": data.get("usage", {}),
"latency_ms": round(latency_ms, 2),
"finish_reason": data["choices"][0].get("finish_reason"),
"_raw_response": data # Debug purposes
}
# Calculate cost
if "usage" in data:
input_tokens = data["usage"].get("prompt_tokens", 0)
output_tokens = data["usage"].get("completion_tokens", 0)
pricing = self.model_pricing.get(model, {"input": 0, "output": 0})
result["cost_usd"] = round(
(input_tokens / 1_000_000) * pricing["input"] +
(output_tokens / 1_000_000) * pricing["output"],
6 # Precision đến 6 chữ số thập phân
)
# Record success metrics
self.metrics.record_request(latency_ms, success=True)
logger.info(
f"✅ Request {request_id} completed: "
f"latency={latency_ms:.2f}ms, "
f"tokens={data.get('usage', {}).get('total_tokens', 'N/A')}"
)
return result
elif response.status_code == 429:
# Rate limit - extract retry-after
retry_after = int(response.headers.get("Retry-After", 60))
error_data = response.json() if response.text else {}
exception = self._parse_error_response(429, error_data)
exception.context.latency_ms = latency_ms
exception.context.retry_count = attempt
exception.context.request_id = request_id
exception.context.user_id = user_id
if self.on_exception:
self.on_exception(exception)
if attempt < self.max_retries - 1:
logger.warning(
f"⚠️ Rate limit hit, retrying in {retry_after}s "
f"(attempt {attempt + 1}/{self.max_retries})"
)
time.sleep(retry_after)
continue
self.metrics.record_request(latency_ms, success=False, exception_type="RateLimit")
self._circuit_open = True
self._circuit_opened_at = datetime.utcnow()
raise exception
elif response.status_code == 401:
error_data = response.json() if response.text else {}
exception = self._parse_error_response(401, error_data)
exception.context.latency_ms = latency_ms
exception.context.request_id = request_id
self.metrics.record_request(latency_ms, success=False, exception_type="Auth")
raise exception # Không retry auth errors
elif response.status_code >= 500:
# Server error - có thể retry
error_data = response.json() if response.text else {}
exception = self._parse_error_response(response.status_code, error_data)
exception.context.latency_ms = latency_ms
exception.context.retry_count = attempt
if self.on_exception:
self.on_exception(exception)
if attempt < self.max_retries - 1:
delay = self.retry_delay * (2 ** attempt) # Exponential backoff
logger.warning(
f"⚠️ Server error {response.status_code}, retrying in {delay}s "
f"(attempt {attempt + 1}/{self.max_retries})"
)
time.sleep(delay)
continue
self.metrics.record_request(latency_ms, success=False, exception_type="Server")
self._circuit_open = True
self._circuit_opened_at = datetime.utcnow()
raise exception
else:
# Client error
error_data = response.json() if response.text else {}
exception = self._parse_error_response(response.status_code, error_data)
exception.context.latency_ms = latency_ms
raise exception
except requests.exceptions.Timeout:
latency_ms = (time.time() - start_time) * 1000
exception = TimeoutException(
message=f"Request timeout after {self.timeout_seconds}s",
timeout_ms=self.timeout_seconds * 1000,
context=AIExceptionContext(
exception_type="TimeoutException",
message=f"Timeout after {self.timeout_seconds}s",
severity=ExceptionSeverity.MEDIUM,
latency_ms=latency_ms,
request_id=request_id,
retry_count=attempt
)
)
if self.on_exception:
self.on_exception(exception)
if attempt < self.max_retries - 1:
delay = self.retry_delay * (2 ** attempt)
logger.warning(f"⏰ Timeout, retrying in {delay}s (attempt {attempt + 1})")
time.sleep(delay)
continue
self.metrics.record_request(latency_ms, success=False, exception_type="Timeout")
raise exception
except requests.exceptions.ConnectionError as e:
latency_ms = (time.time() - start_time) * 1000
exception = ServiceUnavailableException(
message=f"Connection error: {str(e)}",
is_retryable=True,
context=AIExceptionContext(
exception_type="ConnectionError",
message=str(e),
severity=ExceptionSeverity.HIGH,
latency_ms=latency_ms,
retry_count=attempt
)
)
if self.on_exception:
self.on_exception(exception)
if attempt < self.max_retries - 1:
delay = self.retry_delay * (2 ** attempt)
logger.warning(f"🔌 Connection error, retrying in {delay}s")
time.sleep(delay)
continue
self.metrics.record_request(latency_ms, success=False, exception_type="Connection")
self._circuit_open = True
self._circuit_opened_at = datetime.utcnow()
raise exception
# Should not reach here
raise ServiceUnavailableException("Max retries exceeded")
def get_metrics(self) -> MonitoringMetrics:
"""Get current monitoring metrics"""
return self.metrics
def reset_circuit_breaker(self):
"""Manually reset circuit breaker"""
self._circuit_open = False
self._circuit_opened_at = None
logger.info("🔧 Circuit breaker manually reset")
Usage Example
if __name__ == "__main__":
# Initialize client
client = HolySheepAIClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
default_model="deepseek-v3.2" # $0.42/MTok - best value
)
# Exception handler
def handle_exception(exc: BaseAIException):
print(f"🚨 Exception caught: {exc.severity.value} - {exc.message}")
# Send to monitoring system
client.on_exception = handle_exception
# Make request
try:
response = client.chat_completion(
messages=[
{"role": "system", "content": "Bạn là trợ lý AI hữu ích."},
{"role": "user", "content": "Giải thích exception pattern trong monitoring"}
],
model="deepseek-v3.2"
)
print(f"Response: {response['content']}")
print(f"Latency: {response['latency_ms']}ms")
print(f"Cost: ${response.get('cost_usd', 0):.6f}")
except BaseAIException as e:
print(f"Exception: {e.to_dict()}")
3. Prometheus Metrics Integration
Để đưa metrics lên Prometheus/Grafana, tôi sử dụng prometheus_client:
# prometheus_metrics.py - Export metrics to Prometheus
from prometheus_client import Counter, Histogram, Gauge, CollectorRegistry, push_to_gateway
from holy_sheep_client import HolySheepAIClient, BaseAIException
Create custom registry
registry = CollectorRegistry()
Define metrics
REQUEST_COUNT = Counter(
'ai_service_requests_total',
'Total AI service requests',
['model', 'status', 'exception_type'],
registry=registry
)
REQUEST_LATENCY = Histogram(
'ai_service_request_latency_seconds',
'Request latency in seconds',
['model', 'status'],
buckets=[0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0],
registry=registry
)
TOKEN_USAGE = Counter(
'ai_service_tokens_total',
'Total tokens processed',
['model', 'token_type'], # token_type: prompt/completion
registry=registry
)
COST_USD = Counter(
'ai_service_cost_usd_total',
'Total cost in USD',
['model'],
registry=registry
)
ACTIVE_CIRCUIT_BREAKER = Gauge(
'ai_service_circuit_breaker_open',
'Circuit breaker status (1=open, 0=closed)',
['provider'],
registry=registry
)
EXCEPTION_RATE = Counter(
'ai_service_exceptions_total',
'Total exceptions by type',
['exception_class', 'severity', 'retryable'],
registry=registry
)
class PrometheusMonitoredClient(HolySheepAIClient):
"""HolySheep client với Prometheus metrics export"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._last_circuit_state = False
def _record_metrics(self, response: dict, success: bool, exception: BaseAIException = None):
"""Record metrics to Prometheus"""
model = response.get("model", "unknown")
status = "success" if success else "error"
exception_type = exception.__class__.__name__ if exception else "none"
# Request count
REQUEST_COUNT.labels(
model=model,
status=status,
exception_type=exception_type
).inc()
# Request latency
latency_seconds = response.get("latency_ms", 0) / 1000
REQUEST_LATENCY.labels(model=model, status=status).observe(latency_seconds)
# Token usage
if "usage" in response:
usage = response["usage"]
TOKEN_USAGE.labels(model=model, token_type="prompt").inc(
usage.get("prompt_tokens", 0)
)
TOKEN_USAGE.labels(model=model, token_type="completion").inc(
usage.get("completion_tokens", 0)
)
# Cost calculation
if "cost_usd" in response:
COST_USD.labels(model=model).inc(response["cost_usd"])
# Exception metrics
if exception:
EXCEPTION_RATE.labels(
exception_class=exception.__class__.__name__,
severity=exception.severity.value,
retryable=str(exception.retryable)
).inc()
# Circuit breaker status
circuit_open = 1 if self._circuit_open else 0
if circuit_open != self._last_circuit_state:
ACTIVE_CIRCUIT_BREAKER.labels(provider="holysheep").set(circuit_open)
self._last_circuit_state = bool(circuit_open)
def chat_completion(self, *args, **kwargs):
"""Override để add Prometheus metrics"""
try:
response = super().chat_completion(*args, **kwargs)
self._record_metrics(response, success=True)
return response
except BaseAIException as e:
# Build fake response for metrics
fake_response = {
"model": kwargs.get("model", self.default_model),
"latency_ms": 0,
"exception": e
}
self._record_metrics(fake_response, success=False, exception=e)
raise
Grafana Dashboard JSON (example queries)
GRAFANA_QUERIES = '''
Success Rate
sum(rate(ai_service_requests_total{status="success"}[5m])) /
sum(rate(ai_service_requests_total[5m])) * 100
P95 Latency
histogram_quantile(0.95,
sum(rate(ai_service_request_latency_seconds_bucket[5m])) by (le)
)
Error Rate by Type
sum by (exception_type) (rate(ai_service_exceptions_total{severity="critical"}[5m]))
Cost per Hour
sum(rate(ai_service_cost_usd_total[1h]))
Token Usage Breakdown
sum by (model, token_type) (rate(ai_service_tokens_total[1h]))
Circuit Breaker Status
ai_service_circuit_breaker_open{provider="holysheep"}
'''
if __name__ == "__main__":
# Initialize monitored client
client = PrometheusMonitoredClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
default_model="deepseek-v3.2"
)
# Test metrics
print("📊 Prometheus metrics configuration:")
print(f" REQUEST_COUNT: {REQUEST_COUNT}")
print(f" REQUEST_LATENCY: {REQUEST_LATENCY}")
print(f" COST_USD: {COST_USD}")
print()
print("📈 Grafana queries:")
print(GRAFANA_QUERIES)
4. Alerting System Với Slack Integration
Monitoring không hoàn chỉnh nếu không có alerting. Dưới đây là system tôi dùng cho các dự án thực tế:
# alerting.py - Intelligent alerting system
import json
import requests
from typing import List, Dict, Any, Optional
from datetime import datetime, timedelta
from dataclasses import dataclass
from enum import Enum
class AlertLevel(Enum):
INFO = "info"
WARNING = "warning"
CRITICAL = "critical"
@dataclass
class Alert:
level: AlertLevel
title: str
message: str
metrics: Dict[str, Any]
timestamp: datetime = None
def __post_init__(self):
if self.timestamp is None:
self.timestamp = datetime.utcnow()
def to_slack_block(self) -> Dict:
"""Convert to Slack Block Kit format"""
color_map = {
AlertLevel.INFO: "#36a64f",
AlertLevel.WARNING: "#ff9900",
AlertLevel.CRITICAL: "#ff0000"
}
return {
"color": color_map[self.level],
"blocks": [
{
"type": "header",
"text": {
"type": "plain_text",
"text": f"🚨 {self.title}",
"emoji": True
}
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": self.message
}
},
{
"type": "section",
"fields": [
{
"type": "mrkdwn",
"text": f"*Level:*\n{self.level.value.upper()}"
},
{
"type": "mrkdwn",
"text": f"*Time:*\n{self.timestamp.strftime('%Y-%m-%d %H:%M:%S')} UTC"
}
]
}
]
}
class AlertManager:
"""
Intelligent Alert Manager
Chỉ alert khi thực sự cần thiết - tránh alert fatigue
"""
def __init__(
self,
slack_webhook_url: str,
alert_cooldown_minutes: int = 15,
pagerduty_key: Optional[str] = None
):
self.slack_webhook = slack_webhook_url
self.pagerduty_key = pagerduty_key
self.alert_cooldown = timedelta(minutes=alert_cooldown_minutes)
# Track last alert time per alert type
self._last_alerts: Dict[str, datetime] = {}
# Alert thresholds
self.thresholds = {
"error_rate_pct": 5.0, # Alert if error rate > 5%
"latency_p95_ms": 5000, # Alert if P95 latency > 5s
"cost_per_hour_usd": 100.0, # Alert if cost > $100/hour
"circuit_breaker_open": True, # Alert immediately
"consecutive_errors": 3 # Alert