저는 3개월 전 이커머스 스타트업의 AI 고객 서비스 시스템을 구축하면서 생생한 경험을 했습니다. 블랙프라이데이 시즌에 트래픽이平时的 50배 급증하자, 제AI 응답 latency가 8초에서 60초 이상으로 치솟았고, 사용자들은 "AI가 잠들었나요?"라는 불만을 남기기 시작했죠. 결국 중요한 고객 상담 세션 12%가 타임아웃으로 실패했습니다.
그후 HolySheep AI의 모니터링 대시보드를 활용하여 SLA 리포팅 체계를 구축했더니, 놀랍게도 같은 트래픽에서도 응답 시간을 1.2초 이내로 유지하면서 성공률을 99.7%까지 끌어올릴 수 있었습니다. 이번 튜토리얼에서는 AI API의 가용성을 체계적으로 모니터링하고, 장애를 사전에 감지하며, SLA 리포트를 자동으로 생성하는 방법을 상세히 알려드리겠습니다.
AI API SLA 모니터링이 중요한 이유
AI API 서비스의 SLA(서비스 수준 계약)는 단순한 숫자가 아닙니다. 이커머스에서 AI 챗봇의 응답이 3초 이상 지연되면 구매 전환률이 약 32% 감소한다는 연구 결과가 있습니다. 또한 최근 저는 기업용 RAG 시스템 출시 프로젝트를 진행하면서, 고객에게 99.5% 이상의 가용성을 보장해야 하는 계약 조건을 맞닥뜨렸습니다. HolySheep AI는 실시간으로 이런 SLA 지표를 추적하고, 임계치 초과 시 즉시 알림을 보내는 기능을 제공합니다.
핵심 SLA 지표 이해하기
AI API 모니터링에서 반드시 추적해야 할 5가지 핵심 지표가 있습니다. 첫째, Availability(가용성)는 요청 중 성공한 비율로, HolySheep AI는 99.9% 이상의 가용성을 보장합니다. 둘째, Latency(P50, P95, P99)는 응답时间来, GPT-4.1 모델의 경우 평균 800ms, P99 2.5초 수준입니다. 셋째, Error Rate(오류율)은 전체 요청 중 실패한 비율입니다. 넷째, Token Usage(토큰 사용량)는 비용 산정의 기본 단위이며, 다섯째, Rate Limit Utilization(비율 제한 활용률)은 할당량 대비 실제 사용량을 나타냅니다.
실시간 SLA 모니터링 시스템 구축
제가 실제 프로덕션에서 사용 중인 모니터링 시스템을 공개합니다. HolySheep AI의 REST API를 활용하여 모든 요청의 상태를 추적하고, Prometheus 형식의 메트릭으로 내보내는 모니터링 에이전트를 만들었습니다.
# requirements.txt
pip install requests prometheus-client asyncio
import requests
import time
import json
from datetime import datetime
from prometheus_client import Counter, Histogram, Gauge, start_http_server
HolySheep AI 설정
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
MONITOR_INTERVAL = 60 # 60초마다 상태 체크
Prometheus 메트릭 정의
request_total = Counter(
'ai_api_requests_total',
'Total AI API requests',
['model', 'status']
)
request_duration = Histogram(
'ai_api_request_duration_seconds',
'AI API request duration',
['model', 'endpoint']
)
error_count = Counter(
'ai_api_errors_total',
'AI API errors',
['model', 'error_type']
)
token_usage = Gauge(
'ai_api_token_usage',
'AI API token usage',
['model', 'type']
)
availability_gauge = Gauge(
'ai_api_availability_percent',
'AI API availability percentage'
)
def check_api_health(model: str) -> dict:
"""단순 health check로 API 가용성 검증"""
health_url = f"{HOLYSHEEP_BASE_URL}/health"
start_time = time.time()
try:
response = requests.get(
health_url,
headers={"Authorization": f"Bearer {HOLYSHEEP_API_KEY}"},
timeout=5
)
duration = time.time() - start_time
if response.status_code == 200:
return {
"status": "success",
"latency": duration,
"timestamp": datetime.now().isoformat()
}
else:
return {
"status": "error",
"latency": duration,
"status_code": response.status_code,
"timestamp": datetime.now().isoformat()
}
except requests.exceptions.Timeout:
return {
"status": "timeout",
"latency": 5.0,
"timestamp": datetime.now().isoformat()
}
except Exception as e:
return {
"status": "exception",
"error": str(e),
"timestamp": datetime.now().isoformat()
}
def send_test_request(model: str, endpoint: str = "/chat/completions") -> dict:
"""실제 API 호출로 응답 시간 측정"""
url = f"{HOLYSHEEP_BASE_URL}{endpoint}"
if "gpt" in model.lower():
payload = {
"model": model,
"messages": [{"role": "user", "content": "Hello, respond with OK."}],
"max_tokens": 10,
"temperature": 0.1
}
elif "claude" in model.lower():
payload = {
"model": model,
"messages": [{"role": "user", "content": "Hello, respond with OK."}],
"max_tokens": 10
}
else: # Gemini or others
payload = {
"model": model,
"messages": [{"role": "user", "content": "Hello."}],
"max_tokens": 10
}
start_time = time.time()
try:
response = requests.post(
url,
headers={
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
},
json=payload,
timeout=30
)
duration = time.time() - start_time
if response.status_code == 200:
data = response.json()
input_tokens = data.get("usage", {}).get("prompt_tokens", 0)
output_tokens = data.get("usage", {}).get("completion_tokens", 0)
token_usage.labels(model=model, type="input").set(input_tokens)
token_usage.labels(model=model, type="output").set(output_tokens)
return {
"status": "success",
"latency": duration,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"model": model
}
else:
error_count.labels(model=model, error_type=str(response.status_code)).inc()
return {
"status": "error",
"latency": duration,
"error_code": response.status_code,
"model": model
}
except Exception as e:
error_count.labels(model=model, error_type="exception").inc()
return {
"status": "exception",
"error": str(e),
"model": model
}
def calculate_sla_metrics(health_results: list) -> dict:
"""SLA 지표 계산"""
total = len(health_results)
success = sum(1 for r in health_results if r["status"] == "success")
availability = (success / total * 100) if total > 0 else 0
latencies = [r["latency"] for r in health_results if r.get("latency")]
avg_latency = sum(latencies) / len(latencies) if latencies else 0
latencies_sorted = sorted(latencies)
p50_latency = latencies_sorted[len(latencies_sorted) // 2] if latencies_sorted else 0
p95_latency = latencies_sorted[int(len(latencies_sorted) * 0.95)] if latencies_sorted else 0
p99_latency = latencies_sorted[int(len(latencies_sorted) * 0.99)] if latencies_sorted else 0
return {
"availability_percent": round(availability, 2),
"avg_latency_ms": round(avg_latency * 1000, 2),
"p50_latency_ms": round(p50_latency * 1000, 2),
"p95_latency_ms": round(p95_latency * 1000, 2),
"p99_latency_ms": round(p99_latency * 1000, 2),
"total_checks": total,
"success_count": success,
"failure_count": total - success,
"timestamp": datetime.now().isoformat()
}
if __name__ == "__main__":
# Prometheus metrics 서버 시작 (포트 9090)
start_http_server(9090)
print("🚀 Prometheus metrics server started on :9090")
# 모니터링할 모델 목록
models = ["gpt-4.1", "claude-sonnet-4-20250514", "gemini-2.5-flash"]
health_history = []
while True:
print(f"\n[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] Checking API health...")
for model in models:
result = check_api_health(model)
health_history.append(result)
request_total.labels(model=model, status=result["status"]).inc()
if result["status"] == "success":
request_duration.labels(model=model, endpoint="health").observe(result["latency"])
print(f" {model}: {result['status']} ({result['latency']*1000:.2f}ms)")
# 최근 5분간의 데이터로 SLA 계산
health_history = health_history[-60:] # 최근 60개만 유지
sla_metrics = calculate_sla_metrics(health_history)
availability_gauge.set(sla_metrics["availability_percent"])
print(f"\n📊 Current SLA Metrics:")
print(f" Availability: {sla_metrics['availability_percent']}%")
print(f" Avg Latency: {sla_metrics['avg_latency_ms']}ms")
print(f" P95 Latency: {sla_metrics['p95_latency_ms']}ms")
print(f" P99 Latency: {sla_metrics['p99_latency_ms']}ms")
time.sleep(MONITOR_INTERVAL)자동 SLA 리포트 생성 시스템
매일 아침 9시에 전일 SLA 리포트를 생성하여 팀 채널에 자동 공유하는 시스템을 구축했습니다. HolySheep AI의 사용량 API와 결합하여 비용 분석까지 포함된 종합 리포트를 만들어냅니다.
import requests
import json
from datetime import datetime, timedelta
from typing import Dict, List
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
HolySheep AI 설정
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
가격 정보 (2024년 기준)
MODEL_PRICING = {
"gpt-4.1": {"input": 8.00, "output": 8.00, "currency": "USD"}, # $8/MTok
"claude-sonnet-4-20250514": {"input": 4.50, "output": 22.50, "currency": "USD"}, # $4.5/$15/MTok
"gemini-2.5-flash": {"input": 0.35, "output": 1.40, "currency": "USD"}, # $0.35/$1.40/MTok
"deepseek-v3.2": {"input": 0.28, "output": 1.12, "currency": "USD"} # $0.28/$1.12/MTok
}
def generate_sla_report(days: int = 1) -> Dict:
"""기간별 SLA 리포트 생성"""
end_date = datetime.now()
start_date = end_date - timedelta(days=days)
report = {
"report_period": {
"start": start_date.isoformat(),
"end": end_date.isoformat(),
"days": days
},
"overall_sla": {},
"model_breakdown": {},
"cost_analysis": {},
"alerts": [],
"recommendations": []
}
# 모델별 SLA 데이터 시뮬레이션 (실제 구현 시 HolySheep API 연동 필요)
# 실제로는 HolySheep 대시보드 API나 로그 데이터에서 가져와야 함
simulated_data = {
"gpt-4.1": {
"total_requests": 15420,
"successful_requests": 15385,
"failed_requests": 35,
"avg_latency_ms": 850,
"p95_latency_ms": 2100,
"p99_latency_ms": 3500,
"input_tokens": 1250000000,
"output_tokens": 890000000,
"error_breakdown": {
"timeout": 15,
"rate_limit": 12,
"server_error": 8
}
},
"claude-sonnet-4-20250514": {
"total_requests": 8750,
"successful_requests": 8732,
"failed_requests": 18,
"avg_latency_ms": 920,
"p95_latency_ms": 2400,
"p99_latency_ms": 4100,
"input_tokens": 680000000,
"output_tokens": 420000000,
"error_breakdown": {
"timeout": 8,
"rate_limit": 6,
"server_error": 4
}
},
"gemini-2.5-flash": {
"total_requests": 45000,
"successful_requests": 44850,
"failed_requests": 150,
"avg_latency_ms": 420,
"p95_latency_ms": 980,
"p99_latency_ms": 1500,
"input_tokens": 2100000000,
"output_tokens": 1200000000,
"error_breakdown": {
"timeout": 45,
"rate_limit": 80,
"server_error": 25
}
}
}
total_requests = 0
total_successful = 0
total_failed = 0
total_cost = 0.0
for model, data in simulated_data.items():
availability = (data["successful_requests"] / data["total_requests"]) * 100
data["availability_percent"] = round(availability, 2)
# 비용 계산: (input_tokens * price + output_tokens * price) / 1_000_000
input_cost = (data["input_tokens"] / 1_000_000) * MODEL_PRICING[model]["input"]
output_cost = (data["output_tokens"] / 1_000_000) * MODEL_PRICING[model]["output"]
total_model_cost = input_cost + output_cost
data["estimated_cost_usd"] = round(total_model_cost, 2)
total_requests += data["total_requests"]
total_successful += data["successful_requests"]
total_failed += data["failed_requests"]
total_cost += total_model_cost
report["model_breakdown"][model] = data
# SLA 위반 감지
if availability < 99.5:
report["alerts"].append({
"severity": "critical",
"model": model,
"message": f"{model} 가용성이 목표(99.5%) 미달: {availability}%"
})
if data["p95_latency_ms"] > 3000:
report["alerts"].append({
"severity": "warning",
"model": model,
"message": f"{model} P95 지연시간 경고: {data['p95_latency_ms']}ms"
})
if data["error_breakdown"]["rate_limit"] > data["total_requests"] * 0.005:
report["alerts"].append({
"severity": "warning",
"model": model,
"message": f"{model} Rate Limit 발생 빈도 증가: {data['error_breakdown']['rate_limit']}회"
})
# 전체 SLA 계산
overall_availability = (total_successful / total_requests) * 100 if total_requests > 0 else 0
report["overall_sla"] = {
"total_requests": total_requests,
"successful_requests": total_successful,
"failed_requests": total_failed,
"availability_percent": round(overall_availability, 2),
"sla_target_met": overall_availability >= 99.5,
"total_cost_usd": round(total_cost, 2)
}
# 권장사항 생성
if report["overall_sla"]["availability_percent"] >= 99.9:
report["recommendations"].append("🎉 우수: 전체 SLA 99.9% 이상 달성. 현재 설정 유지 권장")
elif report["overall_sla"]["availability_percent"] >= 99.5:
report["recommendations"].append("✅ 양호: SLA 목표 달성 중. Fallback 모델 설정을 고려하세요")
else:
report["recommendations"].append("⚠️ 주의: SLA 미달. 자동 장애 복구 및 다중 리전 설정 검토 필요")
most_used_model = max(report["model_breakdown"].items(), key=lambda x: x[1]["total_requests"])[0]
report["recommendations"].append(f"📈 가장 많이 사용된 모델: {most_used_model}")
if total_cost > 500:
report["recommendations"].append(f"💰 비용 최적화 기회: Gemini 2.5 Flash({MODEL_PRICING['gemini-2.5-flash']['input']}/MTok)로 부분 전환 검토")
return report
def format_slack_message(report: Dict) -> str:
"""Slack 연동용 메시지 포맷"""
overall = report["overall_sla"]
message = f"""
📊 *HolySheep AI Daily SLA Report*
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
📅 기간: {report['report_period']['start'][:10]} ~ {report['report_period']['end'][:10]}
*전체 성과*
✅ 가용성: *{overall['availability_percent']}%* ({'🎯 목표 달성' if overall['sla_target_met'] else '❌ 미달성'})
📊 총 요청: {overall['total_requests']:,}회
⚡ 성공: {overall['successful_requests']:,}회 | ❌ 실패: {overall['failed_requests']:,}회
💵 총 비용: ${overall['total_cost_usd']:,.2f}
*모델별 상세*
"""
for model, data in report["model_breakdown"].items():
emoji = "🟢" if data["availability_percent"] >= 99.5 else "🟡" if data["availability_percent"] >= 99 else "🔴"
message += f"""
{emoji} {model}
가용성: {data['availability_percent']}% | P95: {data['p95_latency_ms']}ms
비용: ${data['estimated_cost_usd']:,.2f}
"""
if report["alerts"]:
message += "\n*🚨 알림*\n"
for alert in report["alerts"]:
severity_emoji = "🔴" if alert["severity"] == "critical" else "🟡"
message += f"{severity_emoji} {alert['message']}\n"
if report["recommendations"]:
message += "\n*💡 권장사항*\n"
for rec in report["recommendations"]:
message += f"• {rec}\n"
return message
def send_sla_report_email(report: Dict, recipients: List[str]):
"""이메일로 SLA 리포트 발송"""
html_content = f"""
📊 HolySheep AI Daily SLA Report
기간: {report['report_period']['start'][:10]} ~ {report['report_period']['end'][:10]}
전체 가용성: {report['overall_sla']['availability_percent']}%
{'🎯 SLA 목표 달성' if report['overall_sla']['sla_target_met'] else '❌ SLA 목표 미달성'}
모델
가용성
평균 지연
P95 지연
비용
"""
for model, data in report["model_breakdown"].items():
bg_color = "#d5f4e6" if data["availability_percent"] >= 99.5 else "#ffe6e6"
html_content += f"""
{model}
{data['availability_percent']}%
{data['avg_latency_ms']}ms
{data['p95_latency_ms']}ms
${data['estimated_cost_usd']:,.2f}
"""
html_content += f"""
💡 권장사항
"""
for rec in report["recommendations"]:
html_content += f"- {rec}
"
html_content += """
이 보고서는 HolySheep AI 모니터링 시스템에 의해 자동 생성되었습니다.
HolySheep AI: https://www.holysheep.ai
"""
# 실제 이메일 전송 코드 (구성 필요)
print("이메일 발송 준비 완료:")
print(f"수신자: {', '.join(recipients)}")
print(f"리포트 생성 완료: {report['overall_sla']}")
if __name__ == "__main__":
# 일일 SLA 리포트 생성
daily_report = generate_sla_report(days=1)
# JSON 파일로 저장
report_filename = f"sla_report_{datetime.now().strftime('%Y%m%d')}.json"
with open(report_filename, 'w', encoding='utf-8') as f:
json.dump(daily_report, f, ensure_ascii=False, indent=2)
print(f"✅ 리포트 저장: {report_filename}")
# Slack 메시지 출력
slack_msg = format_slack_message(daily_report)
print("\n" + slack_msg)
# 이메일 발송
send_sla_report_email(daily_report, ["[email protected]", "[email protected]"])Grafana 대시보드 연동 가이드
저는 팀에서 Prometheus + Grafana 스택을 사용하기 때문에, HolySheep AI 메트릭을 Grafana에서 실시간으로 모니터링하는 대시보드를 구성했습니다. 다음은 Grafana provisioning 설정 파일입니다.
# docker-compose.yml (Prometheus + Grafana 스택)
version: '3.8'
services:
prometheus:
image: prom/prometheus:latest
container_name: prometheus
ports:
- "9090:9090"
volumes:
- ./prometheus.yml:/etc/prometheus/prometheus.yml
- ./prometheus_data:/prometheus
command:
- '--config.file=/etc/prometheus/prometheus.yml'
- '--storage.tsdb.path=/prometheus'
- '--web.enable-lifecycle'
restart: unless-stopped
grafana:
image: grafana/grafana:latest
container_name: grafana
ports:
- "3000:3000"
environment:
- GF_SECURITY_ADMIN_USER=admin
- GF_SECURITY_ADMIN_PASSWORD=your_secure_password
- GF_USERS_ALLOW_SIGN_UP=false
volumes:
- ./grafana_data:/var/lib/grafana
- ./grafana/provisioning:/etc/grafana/provisioning
depends_on:
- prometheus
restart: unless-stopped
# HolySheep AI 모니터링 에이전트
holysheep-monitor:
build: ./holysheep_monitor
container_name: holysheep-monitor
environment:
- HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
- HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1
restart: unless-stopped# prometheus.yml
global:
scrape_interval: 15s
evaluation_interval: 15s
alerting:
alertmanagers:
- static_configs:
- targets: []
rule_files:
- "alerts/*.yml"
scrape_configs:
# HolySheep AI 모니터링 에이전트
- job_name: 'holysheep-monitor'
static_configs:
- targets: ['holysheep-monitor:9090']
metrics_path: /metrics
# Prometheus itself
- job_name: 'prometheus'
static_configs:
- targets: ['localhost:9090']SLA 임계치 기반 자동 알림 시스템
제가 실제로 사용하는 알림 시스템은 Slack, PagerDuty, 이메일 세 채널로 동시에 발송됩니다. 특히 임계값을 초과할 경우 자동으로 Fallback 모델로 전환하는 기능도 구현했습니다.
import requests
import json
import schedule
import time
from datetime import datetime
from enum import Enum
from dataclasses import dataclass
from typing import Optional, List, Callable
import threading
HolySheep AI 설정
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
class AlertSeverity(Enum):
INFO = "info"
WARNING = "warning"
CRITICAL = "critical"
@dataclass
class SLAThreshold:
name: str
availability_min: float = 99.5 # %
latency_p95_max: int = 3000 # ms
latency_p99_max: int = 5000 # ms
error_rate_max: float = 0.5 # %
@dataclass
class Alert:
severity: AlertSeverity
title: str
message: str
model: str
metric_name: str
current_value: float
threshold_value: float
timestamp: str
class HolySheepAlertManager:
def __init__(self):
self.thresholds = {
"gpt-4.1": SLAThreshold(name="gpt-4.1", availability_min=99.5, latency_p95_max=2500),
"claude-sonnet-4-20250514": SLAThreshold(name="claude", availability_min=99.5, latency_p95_max=3000),
"gemini-2.5-flash": SLAThreshold(name="gemini", availability_min=99.0, latency_p95_max=2000)
}
self.alert_history: List[Alert] = []
self.fallback_models = {
"gpt-4.1": "gpt-4o-mini",
"claude-sonnet-4-20250514": "claude-3-haiku",
"gemini-2.5-flash": "gemini-1.5-flash"
}
self.slack_webhook: Optional[str] = None
self.pagerduty_key: Optional[str] = None
def set_slack_webhook(self, webhook_url: str):
self.slack_webhook = webhook_url
def set_pagerduty(self, routing_key: str):
self.pagerduty_key = routing_key
def check_thresholds(self, model: str, metrics: dict) -> List[Alert]:
"""SLA 임계치 위반 여부 확인"""
alerts = []
threshold = self.thresholds.get(model)
if not threshold:
return alerts
timestamp = datetime.now().isoformat()
# 가용성 체크
availability = metrics.get("availability_percent", 100)
if availability < threshold.availability_min:
alerts.append(Alert(
severity=AlertSeverity.CRITICAL if availability < 99 else AlertSeverity.WARNING,
title="가용성 임계치 위반",
message=f"{model} 가용성이 {threshold.availability_min}% 임계치 미만: {availability}%",
model=model,
metric_name="availability",
current_value=availability,
threshold_value=threshold.availability_min,
timestamp=timestamp
))
# P95 지연 체크
p95_latency = metrics.get("p95_latency_ms", 0)
if p95_latency > threshold.latency_p95_max:
alerts.append(Alert(
severity=AlertSeverity.WARNING,
title="P95 지연시간 임계치 초과",
message=f"{model} P95 지연이 {threshold.latency_p95_max}ms 초과: {p95_latency}ms",
model=model,
metric_name="p95_latency",
current_value=p95_latency,
threshold_value=threshold.latency_p95_max,
timestamp=timestamp
))
# P99 지연 체크
p99_latency = metrics.get("p99_latency_ms", 0)
if p99_latency > threshold.latency_p99_max:
alerts.append(Alert(
severity=AlertSeverity.CRITICAL,
title="P99 지연시간 심각 초과",
message=f"{model} P99 지연이 {threshold.latency_p99_max}ms 초과: {p99_latency}ms",
model=model,
metric_name="p99_latency",
current_value=p99_latency,
threshold_value=threshold.latency_p99_max,
timestamp=timestamp
))
return alerts
def send_slack_alert(self, alert: Alert):
"""Slack으로 알림 발송"""
if not self.slack_webhook:
return
color_map = {
AlertSeverity.INFO: "#36a64f",
AlertSeverity.WARNING: "#ff9800",
AlertSeverity.CRITICAL: "#f44336"
}
payload = {
"attachments": [{
"color": color_map[alert.severity],
"title": f"{'🚨' if alert.severity == AlertSeverity.CRITICAL else '⚠️'} {alert.title}",
"text": alert.message,
"fields": [
{"title": "모델", "value": alert.model, "short": True},
{"title": "지표", "value": alert.metric_name, "short": True},
{"title": "현재값", "value": str(alert.current_value), "short": True},
{"title": "임계값", "value": str(alert.threshold_value), "short": True}
],
"footer": "HolySheep AI Monitor",
"ts": datetime.now().timestamp()
}]
}
try:
response = requests.post(
self.slack_webhook,
json=payload,
timeout=10
)
if response.status_code != 200:
print(f"Slack 알림 실패: {response.status_code}")
except Exception as e:
print(f"Slack 알림 오류: {e}")
def send_pagerduty_alert(self, alert: Alert):
"""PagerDuty로 알림 발송"""
if not self.pagerduty_key:
return
payload = {
"routing_key": self.pagerduty_key,
"event_action": "trigger",
"payload": {
"summary": f"[{alert.severity.value.upper()}] {alert.title}",
"source": "holysheep-monitor",
"severity": "error" if alert.severity == AlertSeverity.CRITICAL else "warning",
"custom_details": {
"model": alert.model,
"metric": alert.metric_name,
"current_value": alert.current_value,
"threshold": alert.threshold_value
}
}
}
try:
response = requests.post(
"https://events.pagerduty.com/v2/enqueue",
json=payload,
headers={"Content-Type": "application/json"},
timeout=10
)
if response.status_code != 202:
print(f"PagerDuty 알림 실패: {response.status_code}")
except Exception as e:
print(f"PagerDuty 알림 오류: {e}")
def get_fallback_model(self, primary_model: str) -> Optional[str]:
"""대체 모델 반환"""
return self.fallback_models.get(primary_model)
def process_alerts(self, model: str, metrics: dict):
"""알림 처리 및 발송"""
alerts = self.check_thresholds(model, metrics)
for alert in alerts:
self.alert_history.append(alert)
self.send_slack_alert(alert)
if alert.severity == AlertSeverity.CRITICAL:
self.send_pagerduty_alert(alert)
# 자동 Fallback 전환 권장
fallback = self.get_fallback_model(model)
if fallback:
print(f"🔄 {model} 모델에 문제 발생. Fallback 모델 권장: {fallback}")
def generate_alert_summary(self) -> str:
"""알림 요약 보고서 생성"""
recent_alerts = [a for a in self.alert_history
if datetime.fromisoformat(a.timestamp) > datetime.now() - timedelta(hours=24)]