Trong bài viết này, tôi sẽ hướng dẫn bạn cách xây dựng hệ thống phân tích log cho API Gateway AI từ con số 0. Qua 3 năm vận hành các hệ thống AI production, tôi nhận ra rằng việc debug và monitor API không chỉ giúp phát hiện lỗi nhanh hơn mà còn tiết kiệm đến 60% chi phí vận hành khi biết được pattern sử dụng của users.

1. Tại sao cần phân tích Log cho API Gateway?

Khi bạn xây dựng ứng dụng AI, có đến 80% thời gian debug là để trả lời câu hỏi: "Tại sao request này chậm?" hoặc "Tại sao chi phí tăng đột biến?". Không có log system, bạn sẽ như lái xe mà không có đồng hồ đo tốc độ.

Lợi ích cụ thể:

2. Kiến trúc tổng quan

Hệ thống gồm 4 thành phần chính:

+------------------+     +-------------------+     +---------------+
|  Your App/Client | --> |  API Gateway      | --> | HolySheep AI  |
+------------------+     +-------------------+     +---------------+
                               |                            |
                               v                            v
                    +--------------------+         +----------------+
                    |  OpenTelemetry SDK | -->     |  ClickHouse    |
                    +--------------------+         +----------------+

3. Cài đặt môi trường

Trước tiên, hãy chuẩn bị môi trường với Docker (để chạy ClickHouse) và Python 3.10+:

# Cài đặt Docker (Ubuntu)
sudo apt update
sudo apt install docker.io docker-compose -y
sudo systemctl start docker


Tạo file docker-compose.yml cho ClickHouse

cat > docker-compose.yml << 'EOF'
version: '3.8'
services:
  clickhouse:
    image: clickhouse/clickhouse-server:24.3
    container_name: holysheep-clickhouse
    ports:
      - "8123:8123"
      - "9000:9000"
    environment:
      CLICKHOUSE_DB: apilogs
      CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT: 1
    volumes:
      - clickhouse_data:/var/lib/clickhouse
    command: ['--max_execution_time', '60']

volumes:
  clickhouse_data:
EOF


Khởi động ClickHouse

docker-compose up -d


Verify ClickHouse đang chạy

docker logs holysheep-clickhouse 2>&1 | grep -i "ready for connections"

4. Cài đặt OpenTelemetry Collector

OpenTelemetry sẽ thu thập và xử lý telemetry data từ ứng dụng của bạn:

# Tạo file cấu hình OpenTelemetry
cat > otel-collector-config.yaml << 'EOF'
receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318

processors:
  batch:
    timeout: 1s
    send_batch_size: 1024
  memory_limiter:
    check_interval: 1s
    limit_mib: 512

exporters:
  clickhouse:
    dsn: clickhouse://default:@clickhouse:9000/apilogs?timeout=60s&compress=lz4
    traces_table: traces
    metrics_table: metrics
    logs_table: logs

service:
  pipelines:
    logs:
      receivers: [otlp]
      processors: [memory_limiter, batch]
      exporters: [clickhouse]
EOF


Chạy OpenTelemetry Collector

docker run -d \
  --name otel-collector \
  --network host \
  -v $(pwd)/otel-collector-config.yaml:/etc/otelcol-contrib/config.yaml \
  otel/opentelemetry-collector-contrib:0.98.0

5. Tạo bảng Log trong ClickHouse

Bây giờ chúng ta sẽ tạo schema để lưu trữ log. ClickHouse có performance cực nhanh cho việc query time-series data:

-- Kết nối ClickHouse qua HTTP interface
curl -X POST "http://localhost:8123/query" \
  --data-urlencode "CREATE DATABASE IF NOT EXISTS apilogs"

-- Tạo bảng logs
curl -X POST "http://localhost:8123/query" \
  --data-urlencode "
CREATE TABLE IF NOT EXISTS apilogs.api_requests (
    timestamp DateTime DEFAULT now(),
    trace_id String,
    span_id String,
    request_id String,
    method String,
    endpoint String,
    model String,
    prompt_tokens UInt32,
    completion_tokens UInt32,
    total_tokens UInt32,
    latency_ms Float64,
    status_code UInt16,
    error_message String,
    cost_usd Float64,
    user_id String
) ENGINE = MergeTree()
ORDER BY (timestamp, trace_id)
PARTITION BY toYYYYMM(timestamp)
TTL timestamp + INTERVAL 90 DAY;
"

6. Integration với HolySheep AI API

Đây là phần quan trọng nhất — tích hợp HolySheep AI với OpenTelemetry logging. Với HolySheep, bạn được hưởng tỷ giá ¥1=$1 (tiết kiệm 85%+ so với các provider khác), hỗ trợ WeChat/Alipay thanh toán, và latency trung bình chỉ <50ms.

Bảng giá tham khảo (2026/MTok):

# Cài đặt dependencies
pip install opentelemetry-api opentelemetry-sdk \
  opentelemetry-exporter-otlp-proto-grpc \
  openai httpx aiofiles


File: holysheep_client.py

import os
import time
import json
from typing import Dict, Any, Optional
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.trace import Status, StatusCode


Khởi tạo OpenTelemetry

trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer(__name__)


OTLP Exporter -> gửi sang collector

otlp_exporter = OTLPSpanExporter(
    endpoint="http://localhost:4317",
    insecure=True
)
trace.get_tracer_provider().add_span_processor(
    BatchSpanProcessor(otlp_exporter)
)


Import thư viện sau khi cài

try:
    from openai import OpenAI
except ImportError:
    import subprocess
    subprocess.run(["pip", "install", "openai"], check=True)
    from openai import OpenAI

class HolySheepAIClient:
    """Client cho HolySheep AI với OpenTelemetry logging"""
    
    BASE_URL = "https://api.holysheep.ai/v1"
    
    # Pricing (USD per 1M tokens) - 2026
    PRICING = {
        "gpt-4.1": {"input": 8.00, "output": 8.00},
        "claude-sonnet-4.5": {"input": 15.00, "output": 15.00},
        "gemini-2.5-flash": {"input": 2.50, "output": 2.50},
        "deepseek-v3.2": {"input": 0.42, "output": 0.42}
    }
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.client = OpenAI(
            api_key=api_key,
            base_url=self.BASE_URL,
            timeout=120.0
        )
    
    def calculate_cost(
        self, 
        model: str, 
        prompt_tokens: int, 
        completion_tokens: int
    ) -> float:
        """Tính chi phí theo số token"""
        pricing = self.PRICING.get(model, {"input": 0, "output": 0})
        cost = (
            (prompt_tokens / 1_000_000) * pricing["input"] +
            (completion_tokens / 1_000_000) * pricing["output"]
        )
        return round(cost, 6)
    
    @tracer.start_as_current_span("chat_completion")
    def chat_completion(
        self,
        model: str,
        messages: list,
        temperature: float = 0.7,
        max_tokens: int = 2048
    ) -> Dict[str, Any]:
        """Gọi API với automatic logging"""
        
        span = trace.get_current_span()
        start_time = time.time()
        
        # Ghi thông tin request
        span.set_attribute("request.model", model)
        span.set_attribute("request.temperature", temperature)
        span.set_attribute("request.max_tokens", max_tokens)
        span.set_attribute("request.message_count", len(messages))
        
        try:
            response = self.client.chat.completions.create(
                model=model,
                messages=messages,
                temperature=temperature,
                max_tokens=max_tokens
            )
            
            # Calculate metrics
            latency_ms = (time.time() - start_time) * 1000
            prompt_tokens = response.usage.prompt_tokens
            completion_tokens = response.usage.completion_tokens
            total_tokens = response.usage.total_tokens
            cost_usd = self.calculate_cost(
                model, prompt_tokens, completion_tokens
            )
            
            # Ghi metrics vào span
            span.set_attribute("response.latency_ms", latency_ms)
            span.set_attribute("response.prompt_tokens", prompt_tokens)
            span.set_attribute("response.completion_tokens", completion_tokens)
            span.set_attribute("response.total_tokens", total_tokens)
            span.set_attribute("response.cost_usd", cost_usd)
            span.set_attribute("response.id", response.id)
            span.set_status(Status(StatusCode.OK))
            
            # Log ra console để verify
            print(f"✅ [HolySheep] {model} | "
                  f"Latency: {latency_ms:.2f}ms | "
                  f"Tokens: {total_tokens} | "
                  f"Cost: ${cost_usd:.6f}")
            
            return {
                "response": response,
                "metrics": {
                    "latency_ms": latency_ms,
                    "prompt_tokens": prompt_tokens,
                    "completion_tokens": completion_tokens,
                    "total_tokens": total_tokens,
                    "cost_usd": cost_usd
                }
            }
            
        except Exception as e:
            latency_ms = (time.time() - start_time) * 1000
            span.set_attribute("error.latency_ms", latency_ms)
            span.set_attribute("error.type", type(e).__name__)
            span.set_attribute("error.message", str(e))
            span.set_status(Status(StatusCode.ERROR, str(e)))
            
            print(f"❌ [HolySheep] Error: {e}")
            raise



Sử dụng client

if __name__ == "__main__":
    client = HolySheepAIClient(api_key="YOUR_HOLYSHEEP_API_KEY")
    
    # Test với DeepSeek V3.2 (giá rẻ nhất: $0.42/MTok)
    result = client.chat_completion(
        model="deepseek-v3.2",
        messages=[
            {"role": "system", "content": "Bạn là trợ lý AI hữu ích"},
            {"role": "user", "content": "Giải thích OpenTelemetry là gì?"}
        ],
        temperature=0.7,
        max_tokens=500
    )
    
    print(f"\n📊 Metrics: {result['metrics']}")

7. Query và Phân tích Log trong ClickHouse

Sau khi có data, bạn có thể query để phân tích performance và chi phí:

# Query 1: Top 10 request chậm nhất
curl -X POST "http://localhost:8123/query" \
  --data-urlencode "
SELECT 
    timestamp,
    model,
    latency_ms,
    total_tokens,
    cost_usd,
    status_code
FROM apilogs.api_requests
ORDER BY latency_ms DESC
LIMIT 10
FORMAT PrettyCompact
"


Query 2: Chi phí theo model (ngày)

curl -X POST "http://localhost:8123/query" \
  --data-urlencode "
SELECT 
    model,
    sum(total_tokens) as total_tokens,
    round(sum(cost_usd), 4) as total_cost_usd,
    count() as request_count,
    round(avg(latency_ms), 2) as avg_latency_ms
FROM apilogs.api_requests
WHERE timestamp >= now() - INTERVAL 7 DAY
GROUP BY model
ORDER BY total_cost_usd DESC
FORMAT PrettyCompact
"


Query 3: P50, P95, P99 latency

curl -X POST "http://localhost:8123/query" \
  --data-urlencode "
SELECT 
    model,
    quantile(0.50)(latency_ms) as p50_ms,
    quantile(0.95)(latency_ms) as p95_ms,
    quantile(0.99)(latency_ms) as p99_ms
FROM apilogs.api_requests
WHERE timestamp >= now() - INTERVAL 1 DAY
GROUP BY model
FORMAT PrettyCompact
"


Query 4: Error rate by endpoint

curl -X POST "http://localhost:8123/query" \
  --data-urlencode "
SELECT 
    endpoint,
    status_code,
    count() as error_count,
    round(count() * 100.0 / sum(count()) OVER (PARTITION BY endpoint), 2) as error_rate
FROM apilogs.api_requests
WHERE status_code >= 400
GROUP BY endpoint, status_code
HAVING error_count > 10
ORDER BY error_count DESC
FORMAT PrettyCompact
"

8. Dashboard Visualization với Grafana

Để visualize dữ liệu, bạn có thể kết nối Grafana với ClickHouse:

# Thêm ClickHouse datasource vào Grafana

URL: http://clickhouse:8123


Database: apilogs


Username: default


Password: (để trống)



Query cho Dashboard - Request Rate

SELECT 
    toStartOfInterval(timestamp, INTERVAL 5 MINUTE) as time,
    count() as request_count,
    round(sum(cost_usd), 4) as cost_usd
FROM apilogs.api_requests
WHERE $__timeFilter(timestamp)
GROUP BY time
ORDER BY time


Query cho Dashboard - Latency Distribution

SELECT 
    model,
    histogram(10)(latency_ms) as latency_hist
FROM apilogs.api_requests
WHERE $__timeFilter(timestamp)
GROUP BY model

9. Tối ưu Chi phí với Smart Routing

Từ kinh nghiệm thực chiến, tôi nhận ra rằng 70% chi phí có thể tiết kiệm bằng cách route request thông minh:

# File: smart_router.py
import asyncio
from typing import List, Dict, Any
from holysheep_client import HolySheepAIClient

class SmartRouter:
    """Router thông minh để tối ưu chi phí và performance"""
    
    # Routing rules
    ROUTING_RULES = {
        "simple_question": {
            "models": ["deepseek-v3.2", "gemini-2.5-flash"],
            "max_tokens": 256,
            "criteria": "complexity < 0.3"
        },
        "code_generation": {
            "models": ["gpt-4.1", "deepseek-v3.2"],
            "max_tokens": 2048,
            "criteria": "contains_code"
        },
        "complex_analysis": {
            "models": ["gpt-4.1", "claude-sonnet-4.5"],
            "max_tokens": 4096,
            "criteria": "complexity > 0.7"
        }
    }
    
    def __init__(self, api_key: str):
        self.client = HolySheepAIClient(api_key)
        self.cost_savings = {
            "total_requests": 0,
            "total_cost": 0.0,
            "baseline_cost": 0.0
        }
    
    def analyze_complexity(self, messages: List[Dict]) -> float:
        """Phân tích độ phức tạp của request"""
        total_chars = sum(len(m.get("content", "")) for m in messages)
        code_indicators = ["```", "def ", "function", "class ", "import "]
        
        content = " ".join(m.get("content", "") for m in messages)
        has_code = any(indicator in content for indicator in code_indicators)
        
        if has_code:
            return 0.6
        return min(total_chars / 5000, 1.0)
    
    def select_model(self, messages: List[Dict]) -> str:
        """Chọn model phù hợp dựa trên complexity"""
        complexity = self.analyze_complexity(messages)
        
        if complexity < 0.3:
            # Simple task -> dùng model rẻ
            return "deepseek-v3.2"
        elif complexity < 0.7:
            # Medium task -> balance cost/quality
            return "gemini-2.5-flash"
        else:
            # Complex task -> dùng model mạnh
            return "gpt-4.1"
    
    async def route_request(
        self, 
        messages: List[Dict],
        user_preference: str = None
    ) -> Dict[str, Any]:
        """Route request với auto-scaling"""
        
        # Chọn model
        model = user_preference or self.select_model(messages)
        
        # Calculate baseline cost (giả sử dùng Claude Sonnet)
        baseline_cost = 15.00 / 1_000_000 * 1000  # ~1000 tokens
        
        # Call API
        result = self.client.chat_completion(
            model=model,
            messages=messages,
            max_tokens=512
        )
        
        # Track savings
        actual_cost = result["metrics"]["cost_usd"]
        self.cost_savings["total_requests"] += 1
        self.cost_savings["total_cost"] += actual_cost
        self.cost_savings["baseline_cost"] += baseline_cost
        
        return {
            **result,
            "routing": {
                "selected_model": model,
                "savings_percent": round(
                    (baseline_cost - actual_cost) / baseline_cost * 100, 1
                )
            }
        }


async def main():
    router = SmartRouter("YOUR_HOLYSHEEP_API_KEY")
    
    test_requests = [
        # Simple questions

Tài nguyên liên quan
Bài viết liên quan