作为在 HolySheep AI 平台管理过日均 50 万次 API 调用的技术负责人，我见过太多团队因为缺少成本监控而在月底收到天价账单。从最初的焦虑到现在的从容，我花了 6 个月时间构建了一套完整的成本异常告警系统。今天把这个经验完整分享给你。
Tại sao bạn cần hệ thống cảnh báo chi phí API
Khi sử dụng AI API cho production system, chi phí có thể tăng đột biến vì nhiều lý do: vòng lặp vô hạn trong code, prompt không được tối ưu, hoặc đơn giản là do traffic tăng đột biến. Hệ thống cảnh báo giúp bạn phát hiện vấn đề trong vài phút thay vì vài ngày.
Với HolySheep AI, tôi đã thiết lập các ngưỡng cảnh báo giúp tiết kiệm trung bình 40% chi phí hàng tháng. Hãy cùng tôi đi qua từng bước triển khai.
Kiến trúc hệ thống giám sát chi phí
Trước khi đi vào code, bạn cần hiểu luồng dữ liệu của hệ thống giám sát:
- Bộ đếm số lượng request và tokens cho mỗi mô hình
- Tính toán chi phí dựa trên bảng giá từng mô hình
- So sánh với ngưỡng định kỳ (hourly, daily, weekly)
- Kích hoạt webhook hoặc email khi vượt ngưỡng
- Tự động kill switch để ngăn chi phí tiếp tục tăng
Triển khai hệ thống cảnh báo với HolySheep API
Dưới đây là code Python đầy đủ để triển khai hệ thống giám sát chi phí. Tôi đã sử dụng production-quality code này trong 8 tháng qua.
1. Thiết lập client và cấu hình
# config.py
import os
from dataclasses import dataclass
@dataclass
class ModelPricing:
"""Bảng giá API từ HolySheep AI - Cập nhật 2026"""
# Giá mỗi triệu tokens (MTok)
GPT41_INPUT = 2.00
GPT41_OUTPUT = 8.00
CLAUDE_SONNET45_INPUT = 3.00
CLAUDE_SONNET45_OUTPUT = 15.00
GEMINI_FLASH_INPUT = 0.10
GEMINI_FLASH_OUTPUT = 0.40
DEEPSEEK_V32_INPUT = 0.08
DEEPSEEK_V32_OUTPUT = 0.42
@dataclass
class AlertThresholds:
"""Ngưỡng cảnh báo - tùy chỉnh theo ngân sách của bạn"""
HOURLY_BUDGET_USD = 50.00 # $50/giờ → Kill switch
DAILY_BUDGET_USD = 500.00 # $500/ngày → Cảnh báo qua email
WEEKLY_BUDGET_USD = 2500.00 # $2500/tuần → Slack notification
MONTHLY_BUDGET_USD = 8000.00 # $8000/tháng → Báo cáo
# Tỷ lệ tăng bất thường
ANOMALY_THRESHOLD_PCT = 200 # Tăng 200% so với baseline
HolySheep AI Configuration
HOLYSHEEP_CONFIG = {
"base_url": "https://api.holysheep.ai/v1",
"api_key": os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY"),
"webhook_url": os.getenv("ALERT_WEBHOOK_URL", ""),
"email_recipients": ["[email protected]"]
}
So sánh: OpenAI gốc $15/MTok cho GPT-4, HolySheep chỉ $8 → tiết kiệm 47%
Với WeChat/Alipay thanh toán, tỷ giá ¥1=$1 thực sự tiết kiệm
2. Class theo dõi chi phí với tính năng kill switch
# cost_monitor.py
import time
import requests
from datetime import datetime, timedelta
from collections import defaultdict
from threading import Lock
from typing import Dict, List, Optional
class CostMonitor:
"""
Hệ thống giám sát chi phí AI API thời gian thực
Tích hợp HolySheep AI - độ trễ trung bình <50ms
"""
def __init__(self, config: dict, pricing: object):
self.base_url = config["base_url"]
self.api_key = config["api_key"]
self.webhook_url = config.get("webhook_url", "")
self.pricing = pricing
self.pricing_lock = Lock()
# Bộ đếm tokens theo mô hình
self.tokens_used: Dict[str, Dict[str, int]] = defaultdict(
lambda: {"input": 0, "output": 0, "requests": 0}
)
# Lịch sử để phát hiện bất thường
self.hourly_history: List[Dict] = []
self.baseline_cost_per_hour = 0.0
# Trạng thái kill switch
self.kill_switch_active = False
self.emergency_stop_callback = None
def record_usage(self, model: str, input_tokens: int, output_tokens: int):
"""
Ghi nhận usage cho mỗi request
Call sau mỗi API call thành công
"""
with self.pricing_lock:
self.tokens_used[model]["input"] += input_tokens
self.tokens_used[model]["output"] += output_tokens
self.tokens_used[model]["requests"] += 1
def calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
"""Tính chi phí cho một request cụ thể"""
cost_per_mtok = self._get_pricing(model)
input_cost = (input_tokens / 1_000_000) * cost_per_mtok["input"]
output_cost = (output_tokens / 1_000_000) * cost_per_mtok["output"]
return input_cost + output_cost
def _get_pricing(self, model: str) -> Dict[str, float]:
"""Map model name sang bảng giá HolySheep AI"""
pricing_map = {
"gpt-4.1": {"input": self.pricing.GPT41_INPUT, "output": self.pricing.GPT41_OUTPUT},
"gpt-4.1-turbo": {"input": self.pricing.GPT41_INPUT, "output": self.pricing.GPT41_OUTPUT},
"claude-sonnet-4.5": {"input": self.pricing.CLAUDE_SONNET45_INPUT, "output": self.pricing.CLAUDE_SONNET45_OUTPUT},
"claude-opus-4": {"input": self.pricing.CLAUDE_SONNET45_INPUT * 1.5, "output": self.pricing.CLAUDE_SONNET45_OUTPUT * 1.5},
"gemini-2.5-flash": {"input": self.pricing.GEMINI_FLASH_INPUT, "output": self.pricing.GEMINI_FLASH_OUTPUT},
"deepseek-v3.2": {"input": self.pricing.DEEPSEEK_V32_INPUT, "output": self.pricing.DEEPSEEK_V32_OUTPUT},
}
return pricing_map.get(model.lower(), {"input": 0, "output": 0})
def get_current_hourly_cost(self) -> float:
"""Tính tổng chi phí giờ hiện tại"""
total = 0.0
with self.pricing_lock:
for model, usage in self.tokens_used.items():
cost = self._get_pricing(model)
total += (usage["input"] / 1_000_000) * cost["input"]
total += (usage["output"] / 1_000_000) * cost["output"]
return round(total, 4) # Chính xác đến cent
def check_and_alert(self, thresholds: object) -> Dict[str, any]:
"""
Kiểm tra tất cả ngưỡng và trigger alerts nếu cần
Trả về dict chứa các alerts được kích hoạt
"""
current_cost = self.get_current_hourly_cost()
alerts_triggered = []
# Check Hourly Budget - Kill Switch
if current_cost >= thresholds.HOURLY_BUDGET_USD:
self.kill_switch_active = True
alerts_triggered.append({
"level": "CRITICAL",
"type": "KILL_SWITCH",
"message": f"Vượt ngưỡng hourly budget: ${current_cost:.2f} >= ${thresholds.HOURLY_BUDGET_USD:.2f}",
"action": "Kill switch đã kích hoạt"
})
self._trigger_webhook("CRITICAL", alerts_triggered[-1])
# Check Anomaly Detection
if self.baseline_cost_per_hour > 0:
anomaly_ratio = current_cost / self.baseline_cost_per_hour
if anomaly_ratio * 100 >= thresholds.ANOMALY_THRESHOLD_PCT:
alerts_triggered.append({
"level": "WARNING",
"type": "ANOMALY",
"message": f"Chi phí tăng {anomaly_ratio*100:.0f}% so với baseline",
"ratio": round(anomaly_ratio, 2)
})
# Gửi cảnh báo
if alerts_triggered:
self._send_alerts(alerts_triggered)
return {
"current_cost": current_cost,
"kill_switch": self.kill_switch_active,
"alerts": alerts_triggered,
"timestamp": datetime.now().isoformat()
}
def _trigger_webhook(self, level: str, alert: Dict):
"""Gửi webhook khi kill switch kích hoạt"""
if not self.webhook_url:
return
payload = {
"level": level,
"alert_type": alert["type"],
"message": alert["message"],
"action": alert.get("action", ""),
"timestamp": datetime.now().isoformat(),
"current_cost": self.get_current_hourly_cost()
}
try:
response = requests.post(
self.webhook_url,
json=payload,
timeout=5
)
# HolySheep API response time <50ms, webhook nên <5s
except Exception as e:
print(f"Webhook failed: {e}")
def _send_alerts(self, alerts: List[Dict]):
"""Gửi thông báo qua email/Slack"""
for alert in alerts:
print(f"[{alert['level']}] {alert['message']}")
# Implement email/Slack integration ở đây
def reset_hourly_counters(self):
"""Reset bộ đếm mỗi giờ, lưu vào history"""
with self.pricing_lock:
record = {
"timestamp": datetime.now(),
"usage": dict(self.tokens_used),
"total_cost": self.get_current_hourly_cost()
}
self.hourly_history.append(record)
# Tính baseline từ 7 ngày gần nhất cùng giờ
self._calculate_baseline()
# Reset counters
self.tokens_used.clear()
def _calculate_baseline(self):
"""Tính baseline cost per hour từ lịch sử"""
if len(self.hourly_history) < 24:
return
# Lấy data cùng giờ trong 7 ngày qua
current_hour = datetime.now().hour
relevant_records = [
h for h in self.hourly_history[-168:] # 7 days * 24 hours
if h["timestamp"].hour == current_hour
]
if relevant_records:
self.baseline_cost_per_hour = sum(r["total_cost"] for r in relevant_records) / len(relevant_records)
def enable_emergency_stop(self, callback):
"""Đăng ký callback để emergency stop"""
self.emergency_stop_callback = callback
def is_kill_switch_active(self) -> bool:
"""Kiểm tra kill switch có đang active không"""
return self.kill_switch_active
def disable_kill_switch(self, authorized: bool = False):
"""Chỉ admin mới được tắt kill switch"""
if authorized:
self.kill_switch_active = False
print("Kill switch đã được vô hiệu hóa bởi admin")3. Wrapper cho HolySheep API với auto-monitoring
# holy_sheep_client.py
import requests
import time
from typing import Dict, List, Optional, Any
from cost_monitor import CostMonitor, HOLYSHEEP_CONFIG, AlertThresholds
class HolySheepAIClient:
"""
HolySheep AI Client với tích hợp giám sát chi phí tự động
base_url: https://api.holysheep.ai/v1
"""
def __init__(self, api_key: str, cost_monitor: CostMonitor):
self.base_url = "https://api.holysheep.ai/v1"
self.api_key = api_key
self.cost_monitor = cost_monitor
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
})
# Retry configuration
self.max_retries = 3
self.retry_delay = 1.0
def chat_completions(self, messages: List[Dict], model: str = "deepseek-v3.2",
**kwargs) -> Dict[str, Any]:
"""
Gọi Chat Completions API với auto-monitoring
Tự động ghi nhận usage và check kill switch
"""
# Check kill switch trước khi gọi
if self.cost_monitor.is_kill_switch_active():
raise RuntimeError("KILL_SWITCH_ACTIVE: API calls đã bị tạm dừng do vượt ngưỡng chi phí")
start_time = time.time()
for attempt in range(self.max_retries):
try:
response = self.session.post(
f"{self.base_url}/chat/completions",
json={
"model": model,
"messages": messages,
**kwargs
},
timeout=30
)
# HolySheep AI latency: <50ms trung bình
latency_ms = (time.time() - start_time) * 1000
if response.status_code == 200:
data = response.json()
# Ghi nhận usage vào cost monitor
usage = data.get("usage", {})
input_tokens = usage.get("prompt_tokens", 0)
output_tokens = usage.get("completion_tokens", 0)
self.cost_monitor.record_usage(model, input_tokens, output_tokens)
# Check alerts sau mỗi request
alert_result = self.cost_monitor.check_and_alert(AlertThresholds)
return {
"data": data,
"latency_ms": round(latency_ms, 2),
"cost_usd": round(self.cost_monitor.calculate_cost(
model, input_tokens, output_tokens
), 4),
"alerts": alert_result["alerts"],
"kill_switch": alert_result["kill_switch"]
}
elif response.status_code == 429:
# Rate limit - retry với exponential backoff
wait_time = self.retry_delay * (2 ** attempt)
time.sleep(wait_time)
continue
else:
response.raise_for_status()
except requests.exceptions.RequestException as e:
if attempt == self.max_retries - 1:
raise
time.sleep(self.retry_delay * (2 ** attempt))
raise RuntimeError(f"Failed after {self.max_retries} retries")
def embeddings(self, input_text: str, model: str = "text-embedding-3-small") -> Dict:
"""Gọi Embeddings API"""
if self.cost_monitor.is_kill_switch_active():
raise RuntimeError("KILL_SWITCH_ACTIVE: API calls đã bị tạm dừng")
response = self.session.post(
f"{self.base_url}/embeddings",
json={"model": model, "input": input_text},
timeout=10
)
if response.status_code == 200:
return response.json()
response.raise_for_status()
def get_cost_report(self) -> Dict:
"""Lấy báo cáo chi phí chi tiết"""
total_cost = self.cost_monitor.get_current_hourly_cost()
usage_summary = {}
with CostMonitor.__dict__.get("_CostMonitor__pricing_lock"):
for model, usage in self.cost_monitor.tokens_used.items():
pricing = self.cost_monitor._get_pricing(model)
cost = (usage["input"] / 1_000_000) * pricing["input"] + \
(usage["output"] / 1_000_000) * pricing["output"]
usage_summary[model] = {
"input_tokens": usage["input"],
"output_tokens": usage["output"],
"total_requests": usage["requests"],
"cost_usd": round(cost, 4)
}
return {
"total_cost_usd": round(total_cost, 4),
"kill_switch_active": self.cost_monitor.is_kill_switch_active(),
"usage_by_model": usage_summary,
"baseline_per_hour": round(self.cost_monitor.baseline_cost_per_hour, 4)
}
================== VÍ DỤ SỬ DỤNG ==================
if __name__ == "__main__":
# Khởi tạo cost monitor
monitor = CostMonitor(HOLYSHEEP_CONFIG, ModelPricing)
# Đăng ký emergency stop callback
def emergency_stop():
print("🚨 EMERGENCY STOP: Đang disable tất cả AI features...")
# Gọi feature flags để disable AI features
# Send notification đến team
monitor.enable_emergency_stop(emergency_stop)
# Khởi tạo client
client = HolySheepAIClient(
api_key=HOLYSHEEP_CONFIG["api_key"],
cost_monitor=monitor
)
# Ví dụ gọi API
try:
result = client.chat_completions(
messages=[{"role": "user", "content": "Hello, giải thích về cost monitoring"}],
model="deepseek-v3.2",
temperature=0.7
)
print(f"✅ Response nhận được:")
print(f" - Latency: {result['latency_ms']}ms")
print(f" - Cost: ${result['cost_usd']}")
print(f" - Response: {result['data']['choices'][0]['message']['content'][:100]}...")
except RuntimeError as e:
print(f"❌ Error: {e}")Cấu hình Cron Job cho giám sát định kỳ
Ngoài việc check sau mỗi request, bạn cần một cron job chạy định kỳ để phát hiện các vấn đề về chi phí tích lũy.
# cron_monitor.py
import schedule
import time
import threading
from datetime import datetime
from cost_monitor import CostMonitor, HOLYSHEEP_CONFIG, AlertThresholds
from holy_sheep_client import HolySheepAIClient
def hourly_cost_check():
"""Chạy mỗi giờ - kiểm tra chi phí và gửi report"""
print(f"\n{'='*50}")
print(f"⏰ Hourly Check - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
client = HolySheepAIClient(
api_key=HOLYSHEEP_CONFIG["api_key"],
cost_monitor=monitor
)
report = client.get_cost_report()
print(f"💰 Total Cost (current hour): ${report['total_cost_usd']}")
print(f"🚨 Kill Switch: {'ACTIVE' if report['kill_switch_active'] else 'Inactive'}")
print(f"📊 Usage by Model:")
for model, data in report["usage_by_model"].items():
print(f" - {model}: {data['total_requests']} requests, ${data['cost_usd']}")
# Trigger alert check
alert_result = monitor.check_and_alert(AlertThresholds)
if alert_result["alerts"]:
print(f"\n🚨 ALERTS TRIGGERED:")
for alert in alert_result["alerts"]:
print(f" [{alert['level']}] {alert['message']}")
# Reset counters cho giờ tiếp theo
monitor.reset_hourly_counters()
def daily_summary():
"""Chạy mỗi ngày lúc 00:00 - tổng kết và so sánh"""
print(f"\n{'='*50}")
print(f"📅 Daily Summary - {datetime.now().strftime('%Y-%m-%d')}")
# Tính tổng từ history
total_history_cost = sum(h["total_cost"] for h in monitor.hourly_history)
print(f"💰 Total 24h Cost: ${total_history_cost:.2f}")
print(f"📈 Avg per Hour: ${total_history_cost/24:.2f}")
print(f"🎯 Budget: ${AlertThresholds.DAILY_BUDGET_USD}")
if total_history_cost > AlertThresholds.DAILY_BUDGET_USD:
print(f"⚠️ WARNING: Vượt daily budget!")
send_email_alert("Daily Budget Exceeded", total_history_cost)
def start_scheduler():
"""Khởi động scheduler trong thread riêng"""
# Check mỗi 5 phút (hoặc 1 phút tùy nhu cầu)
schedule.every(5).minutes.do(hourly_cost_check)
# Check mỗi ngày lúc 00:05
schedule.every().day.at("00:05").do(daily_summary)
print("📅 Scheduler started - monitoring every 5 minutes")
while True:
schedule.run_pending()
time.sleep(60)
Khởi tạo monitor toàn cục
monitor = CostMonitor(HOLYSHEEP_CONFIG, ModelPricing)
if __name__ == "__main__":
# Chạy hourly check ngay lần đầu
hourly_cost_check()
# Start scheduler
start_scheduler()Dashboard giám sát trên Grafana
Để visualization tốt hơn, tôi recommend sử dụng Grafana với Prometheus. Dưới đây là Prometheus metrics exporter:
# prometheus_exporter.py
from prometheus_client import Counter, Gauge, Histogram, start_http_server
import time
Prometheus Metrics
API_REQUESTS = Counter(
'holysheep_api_requests_total',
'Total API requests',
['model', 'status']
)
API_LATENCY = Histogram(
'holysheep_api_latency_ms',
'API latency in milliseconds',
['model'],
buckets=[10, 25, 50, 100, 200, 500, 1000]
)
API_COST = Gauge(
'holysheep_api_cost_usd',
'Current API cost in USD',
['model']
)
TOTAL_COST_HOURLY = Gauge(
'holysheep_total_cost_hourly_usd',
'Total cost for current hour in USD'
)
KILL_SWITCH_STATUS = Gauge(
'holysheep_kill_switch_active',
'Kill switch status (1=active, 0=inactive)'
)
class PrometheusMetricsExporter:
"""Export metrics từ CostMonitor sang Prometheus"""
def __init__(self, cost_monitor: CostMonitor, port: int = 9090):
self.monitor = cost_monitor
self.port = port
self.last_export_time = time.time()
def update_metrics(self):
"""Update Prometheus metrics từ cost monitor"""
# Update total cost
TOTAL_COST_HOURLY.set(self.monitor.get_current_hourly_cost())
# Update kill switch status
KILL_SWITCH_STATUS.set(1 if self.monitor.is_kill_switch_active() else 0)
# Update per-model metrics
with self.monitor.pricing_lock:
for model, usage in self.monitor.tokens_used.items():
pricing = self.monitor._get_pricing(model)
cost = (usage["input"] / 1_000_000) * pricing["input"] + \
(usage["output"] / 1_000_000) * pricing["output"]
API_COST.labels(model=model).set(cost)
def record_request(self, model: str, status: str, latency_ms: float):
"""Record một request mới"""
API_REQUESTS.labels(model=model, status=status).inc()
API_LATENCY.labels(model=model).observe(latency_ms)
def start(self):
"""Start Prometheus HTTP server"""
start_http_server(self.port)
print(f"📊 Prometheus metrics available at http://localhost:{self.port}/metrics")
while True:
self.update_metrics()
time.sleep(15) # Update every 15 secondsSo sánh chi phí: HolySheep AI vs Other Providers
| Mô hình | HolySheep AI | OpenAI gốc | Tiết kiệm |
|---|---|---|---|
| GPT-4.1 (output) | $8.00/MTok | $15.00/MTok | 47% |
| Claude Sonnet 4.5 (output) | $15.00/MTok | $18.00/MTok | 17% |
| Gemini 2.5 Flash (input) | $2.50/MTok | $0.35/MTok | So sánh khác |
| DeepSeek V3.2 (output) | $0.42/MTok | N/A | Rẻ nhất |
Với tỷ giá ¥1=$1 và hỗ trợ WeChat/Alipay, HolySheep AI là lựa chọn tối ưu cho các team ở Trung Quốc muốn tiết kiệm chi phí API.
Lỗi thường gặp và cách khắc phục
1. Lỗi: "KILL_SWITCH_ACTIVE" - API calls bị block
Mã lỗi đầy đủ:
RuntimeError: KILL_SWITCH_ACTIVE: API calls đã bị tạm dừng do vượt ngưỡng chi phí
Nguyên nhân:
- Chi phí hourly vượt $50.00 (AlertThresholds.HOURLY_BUDGET_USD)
- Hoặc phát hiện anomaly (>200% so với baseline)
Khắc phục:
1. Kiểm tra nguyên nhân gốc rễ
monitor = CostMonitor(HOLYSHEEP_CONFIG, ModelPricing)
print(f"Kill switch active: {monitor.is_kill_switch_active()}")
print(f"Current cost: ${monitor.get_current_hourly_cost()}")
2. Xem lịch sử để phát hiện request bất thường
recent_requests = monitor.tokens_used
for model, usage in recent_requests.items():
print(f"{model}: {usage}")
3. Nếu xác định là false alarm, disable kill switch (chỉ admin)
monitor.disable_kill_switch(authorized=True)
4. Sau khi fix code, tăng ngưỡng nếu cần
AlertThresholds.HOURLY_BUDGET_USD = 100.00 # Tăng lên $100/giờ
5. Để tránh lặp lại, implement rate limiting ở application layer
def rate_limited_call():
global call_count
if call_count >= 100: # Max 100 calls per minute
raise Exception("Rate limit exceeded")
call_count += 12. Lỗi: Tính toán chi phí không chính xác
Mã lỗi đầy đủ:
# Vấn đề: Chi phí tính ra không đúng với bill thực tế
Ví dụ: Tính được $10 nhưng bill là $12
Nguyên nhân thường gặp:
1. Sử dụng model name không đúng format
2. Missing tokens trong response
3. Chưa tính cache tokens (nếu có)
Khắc phục:
from holy_sheep_client import HolySheepAIClient
client = HolySheepAIClient(HOLYSHEEP_CONFIG["api_key"], monitor)
Response thực tế từ HolySheep AI
result = client.chat_completions(
messages=[{"role": "user", "content": "Test"}],
model="deepseek-v3.2" # Luôn dùng lowercase
)
Kiểm tra usage object đầy đủ
usage = result["data"]["usage"]
print(f"Prompt tokens: {usage.get('prompt_tokens', 0)}")
print(f"Completion tokens: {usage.get('completion_tokens', 0)}")
print(f"Total tokens: {usage.get('total_tokens', 0)}")
Nếu có cache, kiểm tra thêm
if "prompt_cache_tokens" in usage:
print(f"Cache tokens: {usage['prompt_cache_tokens']}")
# Trừ cache tokens khi tính phí
Đối chiếu với billing dashboard của HolySheep
Truy cập: https://www.holysheep.ai/dashboard/billing
Fix: Update pricing config nếu có thay đổi
ModelPricing.DEEPSEEK_V32_OUTPUT = 0.45 # Cập nhật nếu giá thay đổi3. Lỗi: Webhook không gửi được hoặc timeout
Mã lỗi đầy đủ:
# Vấn đề: Webhook không trigger khi kill switch kích hoạt
Hoặc webhook bị timeout sau 5 giây
Nguyên nhân:
1. Webhook URL không đúng hoặc không accessible
2. Firewall block outbound requests
3. Target server quá chậm
Khắc phục:
import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
def test_webhook():
"""Test webhook trước khi deploy"""
webhook_url = "https://your-webhook-endpoint.com/alert"
test_payload = {
"level": "TEST",
"message": "Webhook test",
"timestamp": datetime.now().isoformat()
}
# Retry strategy
session = requests.Session()
retry = Retry(
total=3,
backoff_factor=0.5,
status_forcelist=[500, 502, 503, 504]
)
adapter = HTTPAdapter(max_retries=retry)
session.mount('http://', adapter)
session.mount('https://', adapter)
try:
response = session.post(
webhook_url,
json=test_payload,
timeout=10 # Tăng timeout lên 10s
)
print(f"Webhook OK: {response.status_code}")
except requests.exceptions.