Mở Đầu: Khi Hệ Thống Báo "ConnectionError: timeout"
Tôi vẫn nhớ rõ ngày hôm đó - 3 giờ sáng, điện thoại reo liên tục. Khách hàng phản ánh hệ thống chatbot AI hoàn toàn chết. Mở logs ra, toàn bộ request đều trả về ConnectionError: timeout after 30s. Tỷ lệ lỗi 100%. Sau 2 tiếng debug căng thẳng, tôi mới phát hiện - đó là do không có hệ thống giám sát SLA proactive. Kể từ đó, tôi xây dựng một framework giám sát API hoàn chỉnh, và hôm nay sẽ chia sẻ toàn bộ cho các bạn.
Tại Sao SLA Monitoring Quan Trọng Với AI API
Khi tích hợp AI API vào production, nhiều developers chỉ tập trung vào việc gọi API cho đúng mà quên mất một yếu tố sống còn: giám sát và báo cáo SLA. Một API AI có uptime 99% không đồng nghĩa với việc bạn nhận được response time ổn định. Với HolySheheep AI, chúng tôi cam kết uptime 99.9% và latency trung bình dưới 50ms, nhưng điều quan trọng là bạn cần có công cụ để xác minh và phát hiện vấn đề trước khi khách hàng phàn nàn.
Xây Dựng Hệ Thống Monitor SLA Hoàn Chỉnh
1. Thiết Lập Health Check Endpoint
Đầu tiên, chúng ta cần một endpoint health check để theo dõi trạng thái API liên tục:
# health_check.py
import requests
import time
from datetime import datetime
from typing import Dict, List
import statistics
class APIMonitor:
def __init__(self, api_key: str):
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
self.results = []
def check_health(self, timeout: int = 5) -> Dict:
"""Kiểm tra health của API với đo latencies thực tế"""
start = time.time()
try:
# Test endpoint - gọi models list để kiểm tra kết nối
response = requests.get(
f"{self.base_url}/models",
headers=self.headers,
timeout=timeout
)
latency_ms = (time.time() - start) * 1000
return {
"timestamp": datetime.utcnow().isoformat(),
"status_code": response.status_code,
"latency_ms": round(latency_ms, 2),
"success": response.status_code == 200,
"error": None
}
except requests.exceptions.Timeout:
return {
"timestamp": datetime.utcnow().isoformat(),
"status_code": 0,
"latency_ms": timeout * 1000,
"success": False,
"error": "ConnectionError: timeout"
}
except requests.exceptions.ConnectionError as e:
return {
"timestamp": datetime.utcnow().isoformat(),
"status_code": 0,
"latency_ms": 0,
"success": False,
"error": f"ConnectionError: {str(e)}"
}
except Exception as e:
return {
"timestamp": datetime.utcnow().isoformat(),
"status_code": 0,
"latency_ms": 0,
"success": False,
"error": str(e)
}
def run_monitoring(self, interval: int = 60, duration: int = 3600):
"""Chạy monitoring liên tục trong khoảng thời gian"""
print(f"Bắt đầu giám sát API trong {duration}s...")
start_time = time.time()
while time.time() - start_time < duration:
result = self.check_health()
self.results.append(result)
status = "✓" if result["success"] else "✗"
print(f"{status} [{result['timestamp']}] "
f"Status: {result['status_code'] or 'ERR'} | "
f"Latency: {result['latency_ms']}ms")
time.sleep(interval)
return self.generate_report()
def generate_report(self) -> Dict:
"""Tạo báo cáo SLA tổng hợp"""
if not self.results:
return {"error": "Không có dữ liệu"}
successful = [r for r in self.results if r["success"]]
latencies = [r["latency_ms"] for r in successful if r["latency_ms"] > 0]
return {
"total_requests": len(self.results),
"successful_requests": len(successful),
"failed_requests": len(self.results) - len(successful),
"uptime_percentage": round(len(successful) / len(self.results) * 100, 2),
"avg_latency_ms": round(statistics.mean(latencies), 2) if latencies else 0,
"p50_latency_ms": round(statistics.median(latencies), 2) if latencies else 0,
"p95_latency_ms": round(statistics.quantiles(latencies, n=20)[18], 2) if len(latencies) > 20 else 0,
"p99_latency_ms": round(statistics.quantiles(latencies, n=100)[98], 2) if len(latencies) > 100 else 0,
"errors": [r["error"] for r in self.results if r["error"]]
}
Sử dụng
monitor = APIMonitor("YOUR_HOLYSHEEP_API_KEY")
report = monitor.run_monitoring(interval=30, duration=3600)
print("\n=== BÁO CÁO SLA ===")
for key, value in report.items():
print(f"{key}: {value}")2. Tích Hợp Alerting Với Slack/Discord
Điều quan trọng không kém là nhận cảnh báo khi có sự cố. Dưới đây là module alerting tự động:
# alerting.py
import requests
import json
from datetime import datetime
from typing import Optional
class AlertManager:
def __init__(self, slack_webhook: str = None, discord_webhook: str = None):
self.slack_webhook = slack_webhook
self.discord_webhook = discord_webhook
def send_slack_alert(self, title: str, message: str, severity: str = "warning"):
"""Gửi cảnh báo qua Slack"""
if not self.slack_webhook:
return
color_map = {
"critical": "#FF0000",
"warning": "#FFA500",
"info": "#36A64F"
}
payload = {
"attachments": [{
"color": color_map.get(severity, "#FFA500"),
"title": f"🚨 {title}",
"text": message,
"footer": "HolySheep AI Monitor",
"ts": datetime.utcnow().timestamp()
}]
}
try:
response = requests.post(self.slack_webhook, json=payload, timeout=10)
return response.status_code == 200
except Exception as e:
print(f"Lỗi gửi Slack: {e}")
return False
def send_discord_alert(self, title: str, message: str, severity: str = "warning"):
"""Gửi cảnh báo qua Discord"""
if not self.discord_webhook:
return
color_map = {
"critical": 15158332,
"warning": 15105570,
"info": 3447003
}
payload = {
"embeds": [{
"title": f"🚨 {title}",
"description": message,
"color": color_map.get(severity, 15105570),
"footer": {"text": "HolySheep AI Monitor"},
"timestamp": datetime.utcnow().isoformat()
}]
}
try:
response = requests.post(self.discord_webhook, json=payload, timeout=10)
return response.status_code == 204
except Exception as e:
print(f"Lỗi gửi Discord: {e}")
return False
class SLAAlertChecker:
def __init__(self, alert_manager: AlertManager):
self.alert_manager = alert_manager
# Ngưỡng cảnh báo
self.sla_thresholds = {
"uptime_min": 99.0, # Uptime tối thiểu 99%
"latency_p95_max": 500, # P95 latency tối đa 500ms
"error_rate_max": 1.0, # Tỷ lệ lỗi tối đa 1%
}
def check_and_alert(self, report: dict):
"""Kiểm tra SLA và gửi cảnh báo nếu vượt ngưỡng"""
alerts_sent = []
# Kiểm tra uptime
uptime = report.get("uptime_percentage", 0)
if uptime < self.sla_thresholds["uptime_min"]:
msg = (f"⚠️ **Cảnh Báo SLA!**\n\n"
f"Uptime: {uptime}% (Ngưỡng: {self.sla_thresholds['uptime_min']}%)\n"
f"Yêu cầu thất bại: {report.get('failed_requests', 0)}/{report.get('total_requests', 0)}\n"
f"API: HolySheep AI - https://api.holysheep.ai/v1")
self.alert_manager.send_slack_alert("SLA Downtime Alert", msg, "critical")
self.alert_manager.send_discord_alert("SLA Downtime Alert", msg, "critical")
alerts_sent.append("uptime")
# Kiểm tra latency P95
p95 = report.get("p95_latency_ms", 0)
if p95 > self.sla_thresholds["latency_p95_max"]:
msg = (f"⚠️ **Latency Cao!**\n\n"
f"P95 Latency: {p95}ms (Ngưỡng: {self.sla_thresholds['latency_p95_max']}ms)\n"
f"Trung bình: {report.get('avg_latency_ms', 0)}ms\n"
f"API: HolySheep AI")
self.alert_manager.send_slack_alert("High Latency Alert", msg, "warning")
alerts_sent.append("latency")
# Kiểm tra tỷ lệ lỗi
error_rate = (report.get("failed_requests", 0) / report.get("total_requests", 1)) * 100
if error_rate > self.sla_thresholds["error_rate_max"]:
msg = (f"🚨 **Tỷ Lệ Lỗi Cao!**\n\n"
f"Tỷ lệ lỗi: {error_rate:.2f}% (Ngưỡng: {self.sla_thresholds['error_rate_max']}%)\n"
f"Lỗi: {', '.join(set(report.get('errors', [])))}")
self.alert_manager.send_slack_alert("High Error Rate", msg, "critical")
alerts_sent.append("error_rate")
return alerts_sent
Sử dụng với Slack/Discord
alerts = AlertManager(
slack_webhook="YOUR_SLACK_WEBHOOK_URL",
discord_webhook="YOUR_DISCORD_WEBHOOK_URL"
)
checker = SLAAlertChecker(alerts)
triggered = checker.check_and_alert(report)
print(f"Cảnh báo đã gửi: {triggered}")3. Dashboard Grafana + Prometheus Integration
Để visualize dữ liệu SLA một cách chuyên nghiệp, hãy tích hợp với Prometheus metrics:
# prometheus_exporter.py
from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time
Định nghĩa metrics
REQUEST_COUNT = Counter(
'api_requests_total',
'Tổng số request API',
['status', 'endpoint']
)
REQUEST_LATENCY = Histogram(
'api_request_latency_seconds',
'Độ trễ request API (giây)',
['endpoint'],
buckets=[0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0]
)
API_UPTIME = Gauge(
'api_uptime_percentage',
'Phần trăm uptime API trong 5 phút gần nhất'
)
ERROR_RATE = Gauge(
'api_error_rate_percentage',
'Tỷ lệ lỗi API (%)'
)
class PrometheusMetricsExporter:
def __init__(self, api_key: str, port: int = 9090):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.port = port
self.recent_results = []
self.window_size = 300 # 5 phút = 300 giây
def record_request(self, success: bool, latency_ms: float, status_code: int = 200):
"""Ghi nhận một request vào metrics"""
endpoint = "chat/completions"
status = "success" if success else f"error_{status_code}"
REQUEST_COUNT.labels(status=status, endpoint=endpoint).inc()
REQUEST_LATENCY.labels(endpoint=endpoint).observe(latency_ms / 1000)
# Lưu vào window để tính uptime
self.recent_results.append({
"success": success,
"timestamp": time.time()
})
# Clean old data
cutoff = time.time() - self.window_size
self.recent_results = [r for r in self.recent_results if r["timestamp"] > cutoff]
# Update gauges
total = len(self.recent_results)
successful = sum(1 for r in self.recent_results if r["success"])
if total > 0:
API_UPTIME.set((successful / total) * 100)
ERROR_RATE.set(((total - successful) / total) * 100)
def start_exporter(self):
"""Khởi động Prometheus exporter"""
start_http_server(self.port)
print(f"Prometheus exporter đang chạy tại http://localhost:{self.port}/metrics")
print("Thêm vào prometheus.yml:")
print(f" - targets: ['localhost:{self.port}']")
Khởi động exporter
exporter = PrometheusMetricsExporter("YOUR_HOLYSHEEP_API_KEY", port=9090)
exporter.start_exporter()
Ví dụ: Ghi nhận request
exporter.record_request(success=True, latency_ms=45.23)
exporter.record_request(success=False, latency_ms=0, status_code=401)Tạo Báo Cáo SLA Tự Động Hàng Ngày
Báo cáo SLA chi tiết giúp bạn theo dõi xu hướng và đưa ra quyết định tối ưu hóa:
# sla_report_generator.py
import json
from datetime import datetime, timedelta
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
class SLAReportGenerator:
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.data_store = "sla_data.json" # Lưu trữ local
def load_historical_data(self) -> list:
"""Load dữ liệu lịch sử từ file"""
try:
with open(self.data_store, 'r') as f:
return json.load(f)
except FileNotFoundError:
return []
def save_data(self, data: list):
"""Lưu dữ liệu vào file"""
with open(self.data_store, 'w') as f:
json.dump(data, f, indent=2)
def generate_daily_report(self, date: datetime = None) -> str:
"""Tạo báo cáo SLA hàng ngày"""
if date is None:
date = datetime.utcnow()
data = self.load_historical_data()
# Filter data theo ngày
day_start = date.replace(hour=0, minute=0, second=0)
day_end = day_start + timedelta(days=1)
day_data = [
r for r in data
if day_start <= datetime.fromisoformat(r['timestamp']) < day_end
]
if not day_data:
return "Không có dữ liệu cho ngày này"
successful = [r for r in day_data if r['success']]
latencies = [r['latency_ms'] for r in successful if r['latency_ms'] > 0]
# Tính toán metrics
total_requests = len(day_data)
uptime = (len(successful) / total_requests * 100) if total_requests > 0 else 0
report = f"""
╔══════════════════════════════════════════════════════════════╗
║ BÁO CÁO SLA - HOLYSHEEP AI - {date.strftime('%Y-%m-%d')} ║
╠══════════════════════════════════════════════════════════════╣
║ TỔNG QUAN ║
║ ├─ Tổng requests: {total_requests:>6} ║
║ ├─ Thành công: {len(successful):>6} ({len(successful)/total_requests*100:.1f}%) ║
║ ├─ Thất bại: {total_requests - len(successful):>6} ({(total_requests-len(successful))/total_requests*100:.1f}%) ║
║ └─ Uptime: {uptime:.2f}% ║
╠══════════════════════════════════════════════════════════════╣
║ PERFORMANCE ║
║ ├─ Avg Latency: {sum(latencies)/len(latencies) if latencies else 0:>6.2f}ms ║
║ ├─ P50 Latency: {sorted(latencies)[len(latencies)//2] if latencies else 0:>6.2f}ms ║
║ ├─ P95 Latency: {sorted(latencies)[int(len(latencies)*0.95)] if latencies else 0:>6.2f}ms ║
║ └─ P99 Latency: {sorted(latencies)[int(len(latencies)*0.99)] if latencies else 0:>6.2f}ms ║
╠══════════════════════════════════════════════════════════════╣
║ CAM KẾT SLA: Uptime 99.9% | Latency <50ms trung bình ║
║ Trạng thái: {'✅ ĐẠT CHUẨN' if uptime >= 99 and (sum(latencies)/len(latencies) if latencies else 0) < 100 else '⚠️ CẦN CẢI THIỆN'} ║
╚══════════════════════════════════════════════════════════════╝
"""
return report
def send_email_report(self, report: str, recipient: str):
"""Gửi báo cáo qua email"""
msg = MIMEMultipart()
msg['From'] = '[email protected]'
msg['To'] = recipient
msg['Subject'] = f'Báo Cáo SLA HolySheep AI - {datetime.utcnow().strftime("%Y-%m-%d")}'
body = f"""
Báo Cáo SLA Hàng Ngày
{report}
API Endpoint: https://api.holysheep.ai/v1
Giá tham khảo: GPT-4.1 $8/MTok, Claude Sonnet 4.5 $15/MTok,
Gemini 2.5 Flash $2.50/MTok, DeepSeek V3.2 $0.42/MTok
"""
msg.attach(MIMEText(body, 'html'))
try:
# Cấu hình SMTP của bạn
with smtplib.SMTP('smtp.gmail.com', 587) as server:
server.starttls()
server.login('[email protected]', 'your_app_password')
server.send_message(msg)
print("Đã gửi báo cáo qua email!")
except Exception as e:
print(f"Lỗi gửi email: {e}")
Sử dụng
generator = SLAReportGenerator("YOUR_HOLYSHEEP_API_KEY")
report = generator.generate_daily_report()
print(report)
generator.send_email_report(report, "[email protected]")
Lỗi Thường Gặp và Cách Khắc Phục
1. Lỗi "401 Unauthorized" - Authentication Failed
Mô tả lỗi: Khi gọi API, nhận được response 401 Unauthorized với message "Invalid API key".
# Cách khắc phục 401 Unauthorized
import requests
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
def check_api_key():
"""Kiểm tra và xác thực API key"""
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
response = requests.get(f"{BASE_URL}/models", headers=headers)
if response.status_code == 401:
print("❌ Lỗi 401: API Key không hợp lệ")
print("Kiểm tra:")
print(" 1. API key có đúng format không?")
print(" 2. API key đã được kích hoạt chưa?")
print(" 3. Tài khoản có còn credits không?")
return False
elif response.status_code == 200:
print("✅ Xác thực thành công!")
return True
else:
print(f"⚠️ Lỗi khác: {response.status_code}")
return False
Chạy kiểm tra
check_api_key()Nguyên nhân và giải pháp:
- Nguyên nhân 1: API key bị sai hoặc chưa copy đúng → Copy lại key từ dashboard HolySheep
- Nguyên nhân 2: Key hết hạn hoặc bị revoke → Tạo API key mới tại dashboard HolySheep AI
- Nguyên nhân 3: Tài khoản hết credits → Nạp thêm credits hoặc sử dụng gói miễn phí ban đầu
2. Lỗi "ConnectionError: timeout" - Request Timeout
Mô tả lỗi: Request bị timeout sau 30 giây mà không nhận được response.
# Cách khắc phục Connection Timeout
import requests
import time
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def create_resilient_session():
"""Tạo session với retry logic và timeout thông minh"""
session = requests.Session()
# Retry strategy: 3 lần thử, backoff exponential
retry_strategy = Retry(
total=3,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504],
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("http://", adapter)
session.mount("https://", adapter)
return session
def call_api_with_timeout_handling():
"""Gọi API với xử lý timeout toàn diện"""
session = create_resilient_session()
headers = {
"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
}
payload = {
"model": "gpt-4.1",
"messages": [{"role": "user", "content": "Hello"}],
"max_tokens": 100
}
try:
# Timeout tăng dần cho các endpoint khác nhau
timeout = (5, 30) # (connect_timeout, read_timeout)
response = session.post(
"https://api.holysheep.ai/v1/chat/completions",
json=payload,
headers=headers,
timeout=timeout
)
print(f"✅ Response: {response.json()}")
return response.json()
except requests.exceptions.Timeout:
print("❌ Timeout: API mất quá lâu để phản hồi")
print("Giải pháp:")
print(" 1. Giảm max_tokens nếu request quá dài")
print(" 2. Kiểm tra kết nối mạng")
print(" 3. Thử lại sau vài giây (HolySheep cam kết <50ms latency)")
return None
except requests.exceptions.ConnectionError as e:
print(f"❌ Connection Error: {e}")
print("Giải pháp:")
print(" 1. Kiểm tra URL: https://api.holysheep.ai/v1")
print(" 2. Firewall có chặn port 443 không?")
print(" 3. DNS resolution có vấn đề không?")
return None
call_api_with_timeout_handling()Nguyên nhân và giải pháp:
- Nguyên nhân 1: Network connectivity issues → Kiểm tra firewall, proxy, VPN
- Nguyên nhân 2: Server quá tải → Implement exponential backoff retry
- Nguyên nhân 3: Request quá phức tạp → Giảm max_tokens, chia nhỏ request
3. Lỗi "429 Too Many Requests" - Rate Limit Exceeded
Mô tả lỗi: Nhận được HTTP 429 khi gọi API liên tục.
# Cách khắc phục 429 Rate Limit
import time
import requests
from datetime import datetime, timedelta
class RateLimitHandler:
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
self.request_timestamps = []
self.max_requests_per_minute = 60 # Điều chỉnh theo tier của bạn
def wait_if_needed(self):
"""Chờ nếu cần để tránh rate limit"""
now = datetime.utcnow()
# Clean timestamps cũ hơn 1 phút
self.request_timestamps = [
ts for ts in self.request_timestamps
if now - ts < timedelta(minutes=1)
]
if len(self.request_timestamps) >= self.max_requests_per_minute:
# Tính thời gian chờ
oldest = min(self.request_timestamps)
wait_seconds = 60 - (now - oldest).total_seconds()
if wait_seconds > 0:
print(f"⏳ Rate limit sắp chạm. Chờ {wait_seconds:.1f}s...")
time.sleep(wait_seconds)
self.request_timestamps.append(now)
def call_with_rate_limit(self, payload: dict, max_retries: int = 3):
"""Gọi API với xử lý rate limit tự động"""
for attempt in range(max_retries):
self.wait_if_needed()
try:
response = requests.post(
f"{self.base_url}/chat/completions",
json=payload,
headers=self.headers,
timeout=30
)
if response.status_code == 429:
# Parse Retry-After header
retry_after = int(response.headers.get('Retry-After', 60))
print(f"⚠️ Rate limit hit. Chờ {retry_after}s...")
time.sleep(retry_after)
continue
elif response.status_code == 200:
return response.json()
else:
print(f"❌ Lỗi {response.status_code}: {response.text}")
return None
except Exception as e:
print(f"❌ Exception: {e}")
time.sleep(2 ** attempt) # Exponential backoff
print("❌ Đã hết số lần thử")
return None
Sử dụng
handler = RateLimitHandler("YOUR_HOLYSHEEP_API_KEY")
payload = {
"model": "gpt-4.1",
"messages": [{"role": "user", "content": "Hello!"}]
}
result = handler.call_with_rate_limit(payload)Nguyên nhân và giải pháp:
- Nguyên nhân 1: Gọi API quá nhanh → Implement request queue với rate limiting
- Nguyên nhân 2: Batch requests quá lớn → Chia nhỏ batch, xử lý tuần tự
- Nguyên nhân 3: Nhiều workers cùng lúc → Sử dụng semaphore để giới hạn concurrent requests
4. Lỗi "500 Internal Server Error" - Server Error
Mô tả lỗi: API trả về 500 Internal Server Error không đoán trước được.
# Cách khắc phục 500 Server Error
import requests
import time
def call_with_circuit_breaker():
"""Implement Circuit Breaker pattern để xử lý server errors"""
base_url = "https://api.holysheep.ai/v1"
headers = {
"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
}
# Circuit breaker state
failure_count = 0
last_failure_time = None
circuit_open = False
circuit_open_time = None
# Thresholds
FAILURE_THRESHOLD = 5
RECOVERY_TIMEOUT = 60 # 60 giây
HALF_OPEN_MAX_CALLS = 3
def call_api():
nonlocal failure_count, circuit_open, circuit_open_time
# Check if circuit should be reset
if circuit_open and time.time() - circuit_open_time > RECOVERY_TIMEOUT:
circuit_open = False
print("🔄 Circuit breaker: THỬ LẠI (Half-Open)")
failure_count = 0
if circuit_open:
print("⛔ Circuit breaker: OPEN - Từ chối request")
return None
try:
response = requests.post(
f"{base_url}/chat/completions",
json={"model": "gpt-4.1", "messages": [{"role": "user", "content": "test"}]},
headers=headers,
timeout=30
)
if response.status_code >= 500:
failure_count += 1
last_failure_time = time.time()
if failure_count >= FAILURE_THRESHOLD:
circuit_open = True
circuit_open_time = time.time()
print(f"⛔ Circuit breaker: OPEN (quá nhiều