How to Monitor Multi-Model API Health ด้วย Prometheus Metrics

ในโลกของ AI application ที่ต้องใช้งานหลายโมเดลพร้อมกัน การ monitor health ของ API เป็นสิ่งสำคัญมาก วันนี้ผมจะเล่าประสบการณ์ตรงที่เจอปัญหา ConnectionError: timeout และ 401 Unauthorized ใน production และวิธีแก้ไขด้วย Prometheus

ทำไมต้อง Monitor Multi-Model API

เมื่อใช้งานหลาย provider เช่น GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash หรือ DeepSeek V3.2 ผ่าน HolySheep AI ที่ราคาเริ่มต้นเพียง $0.42/MTok (DeepSeek V3.2) และ latency ต่ำกว่า 50ms เราต้องมั่นใจว่า API ทุกตัวทำงานได้ปกติ

การตั้งค่า Prometheus Client Library

# ติดตั้ง Prometheus client
pip install prometheus-client

โครงสร้างโปรเจกต์
api_monitor/
├── main.py
├── metrics.py
└── config.py

สร้าง Prometheus Metrics Exporter

import requests
from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time

กำหนด metrics
REQUEST_COUNT = Counter(
    'api_requests_total',
    'Total API requests',
    ['model', 'status']
)

REQUEST_LATENCY = Histogram(
    'api_request_duration_seconds',
    'API request latency',
    ['model']
)

API_HEALTH = Gauge(
    'api_health_status',
    'API health status (1=healthy, 0=unhealthy)',
    ['model']
)

ฟังก์ชันตรวจสอบ health
def check_api_health(model: str, api_key: str) -> bool:
    """ตรวจสอบสถานะ API ของแต่ละ model"""
    base_url = "https://api.holysheep.ai/v1"
    
    endpoints = {
        'gpt-4.1': f"{base_url}/models/gpt-4.1",
        'claude-sonnet': f"{base_url}/models/claude-sonnet-4.5",
        'gemini-flash': f"{base_url}/models/gemini-2.5-flash",
        'deepseek': f"{base_url}/models/deepseek-v3.2"
    }
    
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    
    try:
        start = time.time()
        response = requests.get(
            endpoints.get(model, endpoints['deepseek']),
            headers=headers,
            timeout=5
        )
        duration = time.time() - start
        
        REQUEST_LATENCY.labels(model=model).observe(duration)
        
        if response.status_code == 200:
            API_HEALTH.labels(model=model).set(1)
            REQUEST_COUNT.labels(model=model, status='success').inc()
            return True
        else:
            API_HEALTH.labels(model=model).set(0)
            REQUEST_COUNT.labels(model=model, status=f'error_{response.status_code}').inc()
            return False
            
    except requests.exceptions.Timeout:
        API_HEALTH.labels(model=model).set(0)
        REQUEST_COUNT.labels(model=model, status='timeout').inc()
        return False
    except requests.exceptions.ConnectionError:
        API_HEALTH.labels(model=model).set(0)
        REQUEST_COUNT.labels(model=model, status='connection_error').inc()
        return False
    except Exception as e:
        API_HEALTH.labels(model=model).set(0)
        REQUEST_COUNT.labels(model=model, status='exception').inc()
        return False

ฟังก์ชันทดสอบ API ด้วยการเรียกจริง
def test_model_inference(model: str, api_key: str) -> dict:
    """ทดสอบ inference จริงเพื่อวัด performance"""
    base_url = "https://api.holysheep.ai/v1"
    
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    
    payload = {
        "model": model,
        "messages": [{"role": "user", "content": "Hello"}],
        "max_tokens": 10
    }
    
    start = time.time()
    try:
        response = requests.post(
            f"{base_url}/chat/completions",
            headers=headers,
            json=payload,
            timeout=10
        )
        duration = time.time() - start
        
        if response.status_code == 200:
            return {
                'success': True,
                'latency_ms': round(duration * 1000, 2),
                'response': response.json()
            }
        else:
            return {
                'success': False,
                'latency_ms': round(duration * 1000, 2),
                'error': f"HTTP {response.status_code}",
                'detail': response.text
            }
    except Exception as e:
        return {
            'success': False,
            'latency_ms': round((time.time() - start) * 1000, 2),
            'error': str(e)
        }

if __name__ == "__main__":
    # เริ่ม HTTP server สำหรับ Prometheus scrape
    start_http_server(8000)
    
    api_key = "YOUR_HOLYSHEEP_API_KEY"
    models = ['deepseek', 'gpt-4.1', 'claude-sonnet', 'gemini-flash']
    
    # วน loop ตรวจสอบทุก 30 วินาที
    while True:
        for model in models:
            check_api_health(model, api_key)
            result = test_model_inference(model, api_key)
            print(f"{model}: {result}")
        time.sleep(30)

Alerting Rules สำหรับ Prometheus

# prometheus_alerts.yml
groups:
  - name: api_health_alerts
    rules:
      # Alert เมื่อ API down
      - alert: APIModelDown
        expr: api_health_status == 0
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "API Model {{ $labels.model }} is down"
          description: "API {{ $labels.model }} ไม่ตอบสนองเกิน 1 นาที"
      
      # Alert เมื่อ latency สูง
      - alert: APIModelLatencyHigh
        expr: histogram_quantile(0.95, api_request_duration_seconds) > 2
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "API {{ $labels.model }} latency สูง"
          description: "P95 latency เกิน 2 วินาที: {{ $value }}s"
      
      # Alert เมื่อ error rate สูง
      - alert: APIErrorRateHigh
        expr: |
          sum(rate(api_requests_total{status!="success"}[5m])) 
          / sum(rate(api_requests_total[5m])) > 0.05
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "API error rate เกิน 5%"
          description: "Error rate: {{ $value | humanizePercentage }}"

Grafana Dashboard Query

# Query สำหรับแสดง API health status
api_health_status

Query สำหรับ latency distribution
histogram_quantile(0.50, rate(api_request_duration_seconds_bucket[5m]))
histogram_quantile(0.95, rate(api_request_duration_seconds_bucket[5m]))
histogram_quantile(0.99, rate(api_request_duration_seconds_bucket[5m]))

Query สำหรับ request rate ตาม model
sum(rate(api_requests_total[1m])) by (model)

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

1. ConnectionError: timeout

สาเหตุ: API server ไม่ตอบสนองภายในเวลาที่กำหนด หรือ network connectivity มีปัญหา

# วิธีแก้ไข: เพิ่ม retry logic และ timeout ที่เหมาะสม
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def call_api_with_retry(url: str, headers: dict, payload: dict) -> dict:
    """เรียก API พร้อม retry เมื่อ timeout"""
    try:
        response = requests.post(
            url,
            headers=headers,
            json=payload,
            timeout=(5, 30)  # (connect_timeout, read_timeout)
        )
        response.raise_for_status()
        return response.json()
    except requests.exceptions.Timeout:
        print("Request timeout - will retry")
        raise
    except requests.exceptions.ConnectionError as e:
        print(f"Connection error: {e}")
        raise

2. 401 Unauthorized

สาเหตุ: API key ไม่ถูกต้อง หมดอายุ หรือ format ของ header ผิดพลาด

# วิธีแก้ไข: ตรวจสอบ format ของ API key
def validate_api_key(api_key: str) -> bool:
    """ตรวจสอบความถูกต้องของ API key"""
    if not api_key:
        raise ValueError("API key is required")
    
    if not api_key.startswith("sk-"):
        raise ValueError("Invalid API key format - must start with 'sk-'")
    
    if len(api_key) < 20:
        raise ValueError("API key too short")
    
    return True

ตรวจสอบ health ก่อนเรียก API
def check_auth_before_request(api_key: str) -> bool:
    """ทดสอบ authentication ก่อนใช้งานจริง"""
    base_url = "https://api.holysheep.ai/v1"
    headers = {"Authorization": f"Bearer {api_key}"}
    
    try:
        response = requests.get(
            f"{base_url}/models",
            headers=headers,
            timeout=5
        )
        if response.status_code == 401:
            print("Authentication failed - check your API key")
            return False
        return response.status_code == 200
    except Exception as e:
        print(f"Auth check failed: {e}")
        return False

3. Rate Limit Exceeded (429)

สาเหตุ: เรียก API บ่อยเกินไปเกิน quota ที่กำหนด

# วิธีแก้ไข: ใช้ rate limiter และ exponential backoff
import threading
import time
from collections import defaultdict

class RateLimiter:
    """Token bucket rate limiter"""
    def __init__(self, calls_per_second: float = 10):
        self.calls_per_second = calls_per_second
        self.last_call = defaultdict(float)
        self.lock = threading.Lock()
    
    def wait_if_needed(self, model: str):
        """รอถ้าจำเป็นต้อง throttle"""
        with self.lock:
            min_interval = 1.0 / self.calls_per_second
            elapsed = time.time() - self.last_call[model]
            
            if elapsed < min_interval:
                sleep_time = min_interval - elapsed
                time.sleep(sleep_time)
            
            self.last_call[model] = time.time()
    
    def handle_rate_limit_error(self, response: requests.Response, model: str):
        """จัดการเมื่อเจอ 429 error"""
        if response.status_code == 429:
            retry_after = int(response.headers.get('Retry-After', 60))
            print(f"Rate limited for {model}, waiting {retry_after}s")
            time.sleep(retry_after)
            return True
        return False

ใช้งาน
limiter = RateLimiter(calls_per_second=10)

def safe_api_call(model: str, api_key: str, payload: dict) -> dict:
    """เรียก API อย่างปลอดภัยด้วย rate limiting"""
    base_url = "https://api.holysheep.ai/v1"
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    
    limiter.wait_if_needed(model)
    
    max_retries = 3
    for attempt in range(max_retries):
        try:
            response = requests.post(
                f"{base_url}/chat/completions",
                headers=headers,
                json={**payload, "model": model},
                timeout=30
            )
            
            if response.status_code == 429:
                if limiter.handle_rate_limit_error(response, model):
                    continue
            
            response.raise_for_status()
            return response.json()
            
        except requests.exceptions.RequestException as e:
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)
    
    return {}

สรุป

การ monitor multi-model API health ด้วย Prometheus เป็นสิ่งจำเป็นสำหรับ production system ที่ต้องการความเสถียร ด้วย HolySheep AI ที่รองรับ GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash และ DeepSeek V3.2 ในราคาที่ประหยัด (เริ่มต้น $0.42/MTok) พร้อม latency ต่ำกว่า 50ms คุณสามารถตั้งค่า health monitoring ได้ตามที่อธิบายข้างต้น

ข้อดีของ HolySheep AI:

• ราคาประหยัด สูงสุด 85%+ เมื่อเทียบกับ provider อื่น
• รองรับหลายโมเดลใน API เดียว
• Latency ต่ำกว่า 50ms
• รองรับ WeChat และ Alipay
• เครดิตฟรีเมื่อลงทะเบียน

ด้วยโค้ดและ config ที่แชร์ในบทความนี้ คุณสามารถวัด latency, error rate และ availability ของทุก model ได้อย่างมีประสิทธิภาพ พร้อม alerting เมื่อมีปัญหาเกิดขึ้น

👉 สมัคร HolySheep AI — รับเครดิตฟรีเมื่อลงทะเบียน