Trong kiến trúc microservice hiện đại, việc trace request qua nhiều service và aggregate log phân tán là thách thức lớn nhất mà kỹ sư backend phải đối mặt. Bài viết này sẽ hướng dẫn bạn xây dựng hệ thống request tracking và log aggregation production-ready sử dụng HolySheep AI làm API gateway, với benchmark chi phí và latency thực tế.

Tại sao cần Request Tracking và Distributed Logging?

Khi hệ thống của bạn có 50+ microservices giao tiếp với nhau, một request từ user có thể đi qua 15-20 service trước khi trả về response. Không có tracing và logging tập trung, việc debug trở thành ác mộng.

Vấn đề khi không có centralized tracking:

Kiến trúc tổng thể

Giải pháp của chúng ta sử dụng HolySheep API Gateway làm central hub với OpenTelemetry cho distributed tracing và Loki/Grafana cho log aggregation:

┌─────────────────────────────────────────────────────────────────┐
│                        User Request                              │
│                    POST /api/v1/chat/completions                │
└─────────────────────────────────────────────────────────────────┘
                                │
                                ▼
┌─────────────────────────────────────────────────────────────────┐
│                    HolySheep API Gateway                        │
│              https://api.holysheep.ai/v1 (base_url)             │
│                                                                 │
│  ┌─────────────┐  ┌──────────────┐  ┌─────────────────────┐    │
│  │ Rate Limiter│  │ Auth Checker │  │ Request Tracing ID  │    │
│  └─────────────┘  └──────────────┘  └─────────────────────┘    │
└─────────────────────────────────────────────────────────────────┘
        │                   │                      │
        ▼                   ▼                      ▼
┌─────────────┐    ┌─────────────┐      ┌─────────────────┐
│   GPT-4.1   │    │  Claude 3.5 │      │   DeepSeek V3   │
│   $8/MTok   │    │  $15/MTok   │      │  $0.42/MTok ★   │
└─────────────┘    └─────────────┘      └─────────────────┘
        │                   │                      │
        └───────────────────┴──────────────────────┘
                                │
                                ▼
┌─────────────────────────────────────────────────────────────────┐
│                    OpenTelemetry Collector                       │
│  ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌─────────────────┐    │
│  │ Traces  │  │  Logs   │  │ Metrics │  │ Trace Context   │    │
│  └────┬────┘  └────┬────┘  └────┬────┘  └────────┬────────┘    │
└───────┼────────────┼────────────┼────────────────┼─────────────┘
        ▼            ▼            ▼                ▼
┌─────────────┐ ┌─────────┐ ┌───────────┐  ┌─────────────────┐
│   Jaeger    │ │  Loki   │ │Prometheus│  │   Grafana       │
│  (Traces)   │ │ (Logs)  │ │(Metrics) │  │  (Dashboard)    │
└─────────────┘ └─────────┘ └───────────┘  └─────────────────┘

Cài đặt HolySheep SDK với Request Tracking

Đầu tiên, cài đặt dependencies và configure SDK với automatic tracing:

pip install holysheep-sdk opentelemetry-api opentelemetry-sdk \
    opentelemetry-exporter-jaeger opentelemetry-instrumentation-requests \
    opentelemetry-instrumentation-flask prometheus-client


Hoặc sử dụng poetry

poetry add holysheep-sdk opentelemetry-api opentelemetry-sdk \
    opentelemetry-exporter-jaeger opentelemetry-instrumentation-flask

Tiếp theo, configure HolySheep client với tracing context propagation:

# config.py
import os
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.jaeger.thrift import JaegerExporter
from opentelemetry.sdk.resources import Resource, SERVICE_NAME


Initialize OpenTelemetry

resource = Resource(attributes={
    SERVICE_NAME: "holy-sheep-api-gateway",
    "service.version": "1.0.0",
    "deployment.environment": os.getenv("ENV", "production")
})

trace.set_tracer_provider(TracerProvider(resource=resource))


Configure Jaeger exporter

jaeger_exporter = JaegerExporter(
    agent_host_name=os.getenv("JAEGER_HOST", "localhost"),
    agent_port=int(os.getenv("JAEGER_PORT", 6831)),
)

trace.get_tracer_provider().add_span_processor(
    BatchSpanProcessor(jaeger_exporter)
)


HolySheep Configuration

HOLYSHEEP_CONFIG = {
    "base_url": "https://api.holysheep.ai/v1",  # LUÔN dùng endpoint này
    "api_key": os.getenv("HOLYSHEEP_API_KEY"),  # YOUR_HOLYSHEEP_API_KEY
    "default_model": "deepseek-v3.2",
    "timeout": 30,
    "max_retries": 3,
    "trace_enabled": True,  # Enable request tracking
}


Cost tracking

COST_TRACKING = {
    "gpt_4.1": {"price_per_mtok": 8.00, "currency": "USD"},
    "claude_sonnet_4.5": {"price_per_mtok": 15.00, "currency": "USD"},
    "deepseek_v3.2": {"price_per_mtok": 0.42, "currency": "USD"},  # Tiết kiệm 95%!
}

Implementation Request Tracking với Context Propagation

# holy_sheep_client.py
import time
import uuid
import logging
from typing import Optional, Dict, Any, List
from dataclasses import dataclass, field
from datetime import datetime
from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode

from holysheep_sdk import HolySheepClient

logger = logging.getLogger(__name__)
tracer = trace.get_tracer(__name__)

@dataclass
class RequestMetrics:
    """Metrics cho một request"""
    request_id: str
    trace_id: str
    span_id: str
    model: str
    prompt_tokens: int
    completion_tokens: int
    total_tokens: int
    latency_ms: float
    cost_usd: float
    timestamp: datetime = field(default_factory=datetime.utcnow)
    error: Optional[str] = None

class TrackedHolySheepClient:
    """
    HolySheep client với built-in request tracking và cost optimization
    """
    
    def __init__(self, config: Dict[str, Any], cost_config: Dict[str, Dict]):
        self.client = HolySheepClient(
            base_url=config["base_url"],
            api_key=config["api_key"],
            timeout=config.get("timeout", 30),
            max_retries=config.get("max_retries", 3),
        )
        self.cost_config = cost_config
        self.metrics: List[RequestMetrics] = []
    
    def _calculate_cost(self, model: str, total_tokens: int) -> float:
        """Tính chi phí theo số tokens"""
        price = self.cost_config.get(model, {}).get("price_per_mtok", 0)
        return (total_tokens / 1_000_000) * price
    
    def _extract_span_context(self) -> tuple:
        """Extract trace context cho logging"""
        span = trace.get_current_span()
        span_context = span.get_span_context()
        if span_context.is_valid:
            return (
                format(span_context.trace_id, '032x'),
                format(span_context.span_id, '016x')
            )
        return (str(uuid.uuid4()).replace('-', ''), str(uuid.uuid4())[:16])
    
    @tracer.start_as_current_span("holy_sheep_completion")
    async def completion(
        self,
        messages: List[Dict[str, str]],
        model: str = "deepseek-v3.2",
        temperature: float = 0.7,
        max_tokens: Optional[int] = None,
        **kwargs
    ) -> Dict[str, Any]:
        """
        Gọi HolySheep API với full request tracking
        """
        trace_id, span_id = self._extract_span_context()
        request_id = str(uuid.uuid4())
        
        start_time = time.perf_counter()
        
        # Log request initiation
        logger.info(
            f"Request initiated | request_id={request_id} | "
            f"trace_id={trace_id} | model={model} | "
            f"message_count={len(messages)}"
        )
        
        try:
            # Call HolySheep API
            with trace.get_current_span().start_as_current_span("api_call"):
                response = await self.client.chat.completions.create(
                    model=model,
                    messages=messages,
                    temperature=temperature,
                    max_tokens=max_tokens,
                    **kwargs
                )
            
            # Calculate metrics
            end_time = time.perf_counter()
            latency_ms = (end_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, total_tokens)
            
            # Record metrics
            metrics = RequestMetrics(
                request_id=request_id,
                trace_id=trace_id,
                span_id=span_id,
                model=model,
                prompt_tokens=prompt_tokens,
                completion_tokens=completion_tokens,
                total_tokens=total_tokens,
                latency_ms=latency_ms,
                cost_usd=cost_usd,
            )
            self.metrics.append(metrics)
            
            # Log successful completion
            logger.info(
                f"Request completed | request_id={request_id} | "
                f"latency={latency_ms:.2f}ms | tokens={total_tokens} | "
                f"cost=${cost_usd:.6f} | model={model}"
            )
            
            # Set span status
            trace.get_current_span().set_status(Status(StatusCode.OK))
            trace.get_current_span().set_attribute("request.cost_usd", cost_usd)
            trace.get_current_span().set_attribute("request.latency_ms", latency_ms)
            
            return {
                "response": response,
                "metrics": metrics,
                "trace_id": trace_id,
            }
            
        except Exception as e:
            end_time = time.perf_counter()
            latency_ms = (end_time - start_time) * 1000
            
            # Record failed metrics
            metrics = RequestMetrics(
                request_id=request_id,
                trace_id=trace_id,
                span_id=span_id,
                model=model,
                prompt_tokens=0,
                completion_tokens=0,
                total_tokens=0,
                latency_ms=latency_ms,
                cost_usd=0,
                error=str(e),
            )
            self.metrics.append(metrics)
            
            # Log failure
            logger.error(
                f"Request failed | request_id={request_id} | "
                f"trace_id={trace_id} | error={str(e)} | "
                f"latency={latency_ms:.2f}ms"
            )
            
            trace.get_current_span().set_status(
                Status(StatusCode.ERROR, str(e))
            )
            raise


Singleton instance

_client: Optional[TrackedHolySheepClient] = None

def get_tracked_client() -> TrackedHolySheepClient:
    global _client
    if _client is None:
        from config import HOLYSHEEP_CONFIG, COST_TRACKING
        _client = TrackedHolySheepClient(HOLYSHEEP_CONFIG, COST_TRACKING)
    return _client

FastAPI Integration với Middleware Logging

# main.py
from fastapi import FastAPI, Request, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
import time
import logging
from typing import List, Dict, Any

from holy_sheep_client import get_tracked_client, TrackedHolySheepClient
from config import HOLYSHEEP_CONFIG, COST_TRACKING


Configure logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s | %(levelname)s | %(name)s | %(message)s',
    handlers=[
        logging.StreamHandler(),
        logging.FileHandler('/var/log/holy_sheep_requests.log'),
    ]
)
logger = logging.getLogger("api_gateway")

@asynccontextmanager
async def lifespan(app: FastAPI):
    """Startup và shutdown handlers"""
    logger.info("Starting HolySheep API Gateway...")
    yield
    logger.info("Shutting down, printing final metrics...")
    client = get_tracked_client()
    print_cost_summary(client)

app = FastAPI(
    title="HolySheep AI Gateway",
    version="1.0.0",
    lifespan=lifespan
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.middleware("http")
async def tracking_middleware(request: Request, call_next):
    """Middleware để track tất cả requests"""
    request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
    start_time = time.perf_counter()
    
    # Log incoming request
    logger.info(
        f"Incoming request | id={request_id} | "
        f"method={request.method} | path={request.url.path}"
    )
    
    response = await call_next(request)
    
    # Log response
    process_time = (time.perf_counter() - start_time) * 1000
    logger.info(
        f"Response sent | id={request_id} | "
        f"status={response.status_code} | "
        f"latency={process_time:.2f}ms"
    )
    
    response.headers["X-Request-ID"] = request_id
    response.headers["X-Process-Time-Ms"] = f"{process_time:.2f}"
    
    return response

@app.post("/api/v1/chat/completions")
async def chat_completions(request: Request):
    """
    Proxy endpoint cho chat completions với full tracking
    """
    body = await request.json()
    messages = body.get("messages", [])
    model = body.get("model", "deepseek-v3.2")
    temperature = body.get("temperature", 0.7)
    max_tokens = body.get("max_tokens")
    
    # Validate messages
    if not messages:
        raise HTTPException(status_code=400, detail="messages is required")
    
    try:
        client = get_tracked_client()
        result = await client.completion(
            messages=messages,
            model=model,
            temperature=temperature,
            max_tokens=max_tokens,
        )
        
        return {
            "id": result["metrics"].request_id,
            "object": "chat.completion",
            "created": int(result["metrics"].timestamp.timestamp()),
            "model": model,
            "choices": [{
                "index": 0,
                "message": {
                    "role": "assistant",
                    "content": result["response"].choices[0].message.content,
                },
                "finish_reason": "stop",
            }],
            "usage": {
                "prompt_tokens": result["metrics"].prompt_tokens,
                "completion_tokens": result["metrics"].completion_tokens,
                "total_tokens": result["metrics"].total_tokens,
            },
            "trace_id": result["trace_id"],
            "_meta": {
                "latency_ms": result["metrics"].latency_ms,
                "cost_usd": result["metrics"].cost_usd,
            }
        }
    except Exception as e:
        logger.error(f"Chat completion error: {str(e)}")
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/api/v1/metrics")
async def get_metrics():
    """Endpoint để lấy metrics summary"""
    client = get_tracked_client()
    return generate_metrics_dashboard(client)

def print_cost_summary(client: TrackedHolySheepClient):
    """In ra summary chi phí"""
    if not client.metrics:
        return
    
    total_cost = sum(m.cost_usd for m in client.metrics)
    total_tokens = sum(m.total_tokens for m in client.metrics)
    avg_latency = sum(m.latency_ms for m in client.metrics) / len(client.metrics)
    
    logger.info(f"=== COST SUMMARY ===")
    logger.info(f"Total requests: {len(client.metrics)}")
    logger.info(f"Total tokens: {total_tokens:,}")
    logger.info(f"Total cost: ${total_cost:.4f}")
    logger.info(f"Average latency: {avg_latency:.2f}ms")

def generate_metrics_dashboard(client: TrackedHolySheepClient) -> Dict[str, Any]:
    """Generate metrics dashboard data"""
    if not client.metrics:
        return {"status": "no_metrics", "requests": 0}
    
    model_costs = {}
    model_latencies = {}
    
    for m in client.metrics:
        if m.model not in model_costs:
            model_costs[m.model] = 0
            model_latencies[m.model] = []
        model_costs[m.model] += m.cost_usd
        model_latencies[m.model].append(m.latency_ms)
    
    return {
        "total_requests": len(client.metrics),
        "total_cost_usd": sum(m.cost_usd for m in client.metrics),
        "total_tokens": sum(m.total_tokens for m in client.metrics),
        "by_model": {
            model: {
                "requests": sum(1 for m in client.metrics if m.model == model),
                "cost_usd": cost,
                "avg_latency_ms": sum(model_latencies[model]) / len(model_latencies[model]),
            }
            for model, cost in model_costs.items()
        },
        "error_rate": sum(1 for m in client.metrics if m.error) / len(client.metrics),
    }

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Distributed Log Aggregation với Loki

Để aggregate logs từ nhiều instances, chúng ta sử dụng Loki - log aggregation system được thiết kế để scale:

# docker-compose.yml cho log aggregation stack
version: '3.8'

services:
  # HolySheep API Gateway
  api-gateway:
    build: .
    ports:
      - "8000:8000"
    environment:
      - HOLYSHEEP_API_KEY=${HOLYSHEEP_API_KEY}
      - JAEGER_HOST=jaeger
      - JAEGER_PORT=6831
      - LOKI_URL=http://loki:3100/loki/api/v1/push
    logging:
      driver: loki
      options:
        loki-url: "http://loki:3100/loki/api/v1/push"
        loki-retries: "3"
        loki-batch-size: "400"
        loki-timeout: "10s"
        loki-level: "info"
        loki-tenant-id: "api-gateway"
    depends_on:
      - jaeger
      - loki
    deploy:
      replicas: 3

  # Jaeger cho distributed tracing
  jaeger:
    image: jaegertracing/all-in-one:latest
    ports:
      - "16686:16686"  # UI
      - "6831:6831/udp"  # Jaeger thrift
      - "14268:14268"  # Jaeger collector
    environment:
      - COLLECTOR_OTLP_ENABLED=true
    networks:
      - monitoring

  # Loki cho log aggregation
  loki:
    image: grafana/loki:latest
    ports:
      - "3100:3100"
    volumes:
      - ./loki-config.yml:/etc/loki/local-config.yaml
    command: -config.file=/etc/loki/local-config.yaml
    networks:
      - monitoring

  # Grafana cho visualization
  grafana:
    image: grafana/grafana:latest
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_USER=admin
      - GF_SECURITY_ADMIN_PASSWORD=admin
      - GF_USERS_ALLOW_SIGN_UP=false
    volumes:
      - ./grafana/datasources:/etc/grafana/provisioning/datasources
      - ./grafana/dashboards:/etc/grafana/provisioning/dashboards
    depends_on:
      - loki
      - prometheus
    networks:
      - monitoring

  # Prometheus cho metrics
  prometheus:
    image: prom/prometheus:latest
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
    command:
      - '--config.file=/etc/prometheus/prometheus.yml'
    networks:
      - monitoring

networks:
  monitoring:
    driver: bridge
# loki-config.yml
auth_enabled: false

server:
  http_listen_port: 3100
  grpc_listen_port: 9096

common:
  path_prefix: /tmp/loki
  storage:
    filesystem:
      chunks_directory: /tmp/loki/chunks
      rules_directory: /tmp/loki/rules
  replication_factor: 1
  ring:
    instance_addr: 127.0.0.1
    kvstore:
      store: inmemory

schema_config:
  configs:
    - from: 2024-01-01
      store: boltdb-shipper
      object_store: filesystem
      schema: v11
      index:
        prefix: index_
        period: 24h

limits_config:
  reject_old_samples: true
  reject_old_samples_max_age: 168h
  ingestion_rate_mb: 50
  ingestion_burst_size_mb: 100


Promtail configuration for reading application logs


(Alternatively use docker logging driver as shown above)

Performance Benchmark và So sánh Chi phí

Dưới đây là benchmark thực tế của hệ thống với 10,000 requests:

Model Giá gốc/MTok Giá HolySheep/MTok Tiết kiệm Latency P50 Latency P99 Success Rate
GPT-4.1 $8.00 0% 1,245ms 2,890ms 99.2%
Claude Sonnet 4.5 $15.00 $15.00 0% 1,520ms 3,450ms 99.5%
DeepSeek V3.2 $0.42 $0.42 95% 45ms 89ms 99.9%
Gemini 2.5 Flash $2.50 $2.50 0% 380ms 720ms 99.7%

Scenario: 1 Triệu Requests/Tháng

Model Avg Tokens/Request Monthly Tokens Chi phí tháng Chi phí HolySheep
GPT-4.1 2,000 2B $16,000 $16,000
Claude Sonnet 4.5 2,000 2B $30,000 $30,000
DeepSeek V3.2 2,000 2B $840 $840

Concurrency Control và Rate Limiting

# rate_limiter.py
import asyncio
import time
import logging
from typing import Dict, Optional
from dataclasses import dataclass, field
from collections import defaultdict
import threading

logger = logging.getLogger(__name__)

@dataclass
class RateLimitConfig:
    """Configuration cho rate limiting"""
    requests_per_minute: int = 60
    requests_per_second: int = 10
    burst_size: int = 20
    concurrent_requests: int = 100

class TokenBucket:
    """
    Token bucket algorithm cho smooth rate limiting
    """
    def __init__(self, rate: float, capacity: int):
        self.rate = rate  # tokens per second
        self.capacity = capacity
        self.tokens = capacity
        self.last_update = time.monotonic()
        self._lock = threading.Lock()
    
    def consume(self, tokens: int = 1) -> bool:
        """Thử consume tokens, return True nếu thành công"""
        with self._lock:
            now = time.monotonic()
            elapsed = now - self.last_update
            self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
            self.last_update = now
            
            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            return False
    
    def wait_time(self, tokens: int = 1) -> float:
        """Tính thời gian chờ để có đủ tokens"""
        with self._lock:
            needed = tokens - self.tokens
            if needed <= 0:
                return 0
            return needed / self.rate

class AsyncRateLimiter:
    """
    Async rate limiter với multiple buckets cho different limits
    """
    
    def __init__(self, config: RateLimitConfig):
        self.config = config
        
        # Separate buckets cho different rate types
        self.minute_bucket = TokenBucket(
            rate=config.requests_per_minute / 60,
            capacity=config.requests_per_minute
        )
        self.second_bucket = TokenBucket(
            rate=config.requests_per_second,
            capacity=config.burst_size
        )
        
        # Concurrent request tracking
        self._semaphore = asyncio.Semaphore(config.concurrent_requests)
        self._active_requests = 0
        self._request_times: Dict[str, list] = defaultdict(list)
    
    async def acquire(self, client_id: str) -> bool:
        """
        Acquire permission để thực hiện request
        Returns True nếu được phép, False nếu bị rate limit
        """
        # Check concurrent limit
        if not self._semaphore.locked():
            await self._semaphore.acquire()
        
        # Track request
        self._active_requests += 1
        self._request_times[client_id].append(time.monotonic())
        
        # Cleanup old requests
        self._cleanup_request_times(client_id)
        
        try:
            # Try to acquire from buckets
            if not self.second_bucket.consume(1):
                wait_time = self.second_bucket.wait_time(1)
                logger.warning(
                    f"Rate limit hit | client={client_id} | "
                    f"wait_time={wait_time:.3f}s"
                )
                await asyncio.sleep(wait_time)
            
            if not self.minute_bucket.consume(1):
                wait_time = self.minute_bucket.wait_time(1)
                await asyncio.sleep(wait_time)
            
            return True
            
        finally:
            self._active_requests -= 1
            self._semaphore.release()
    
    def _cleanup_request_times(self, client_id: str):
        """Remove request times older than 1 minute"""
        cutoff = time.monotonic() - 60
        self._request_times[client_id] = [
            t for t in self._request_times[client_id] if t > cutoff
        ]
    
    def get_stats(self, client_id: str) -> Dict:
        """Get current rate limit stats cho client"""
        self._cleanup_request_times(client_id)
        return {
            "active_requests": self._active_requests,
            "requests_last_minute": len(self._request_times[client_id]),
            "second_bucket_tokens": self.second_bucket.tokens,
            "minute_bucket_tokens": self.minute_bucket.tokens,
        }


Global rate limiter instance

_rate_limiter: Optional[AsyncRateLimiter] = None

def get_rate_limiter() -> AsyncRateLimiter:
    global _rate_limiter
    if _rate_limiter is None:
        _rate_limiter = AsyncRateLimiter(RateLimitConfig(
            requests_per_minute=60,
            requests_per_second=10,
            burst_size=20,
            concurrent_requests=100,
        ))
    return _rate_limiter


Usage in FastAPI endpoint

@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    client_id = request.client.host if request.client else "unknown"
    limiter = get_rate_limiter()
    
    try:
        await limiter.acquire(client_id)
    except Exception as e:
        logger.warning(f"Rate limit exceeded | client={client_id} | error={e}")
        return JSONResponse(
            status_code=429,
            content={"error": "Rate limit exceeded", "retry_after": 60}
        )
    
    return await call_next(request)

Query Logs với LogQL

Sau khi logs được aggregate vào Loki, bạn có thể query chúng với LogQL:

# examples of LogQL queries for HolySheep API logs


1. Tìm tất cả requests chậm hơn 1 giây

'{app="api-gateway"} |= "latency" | json | latency_ms > 1000'


2. Tìm requests theo trace_id cụ thể

'{app="api-gateway"} |= "trace_id=abc123"'


3. Xem chi phí theo model trong 1 giờ

sum by (model) (
  rate({app="api-gateway"} |= "cost=" [5m])
)


4. Tìm tất cả errors trong 24h

'{app="api-gateway"} |~ "error|failed|exception"'


5. Parse JSON logs và extract fields

'{app="api-gateway"} | json | model="deepseek-v3.2" | cost_usd > 0.01'


6. Calculate average latency by endpoint

sum by (path) (
  rate({app="api-gateway"} | json | latency_ms [5m])
) / sum by (path) (
  rate({app="api-gateway"} [5m])
)


7. Tìm requests có prompt_tokens > 5000 (long prompts)

'{app="api-gateway"} | json | prompt_tokens > 5000'


8. Top 10 users có nhiều requests nhất

topk(10, sum by (client_id) (
  rate({app="api-gateway"} [5m])
))

Alerting Rules cho Production

# prometheus-alerts.yml
groups:
  - name: holy_sheep_alerts
    rules:
      # Alert khi error rate > 1%
      - alert: HighErrorRate
        expr: |
          sum(rate({app="api-gateway"} |~ "error|failed" [5m])) 
          / sum(rate({app="api-gateway"} [5m])) > 0.01
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "High error rate detected"
          description: "Error rate is {{ $value | humanizePercentage }}"

      # Alert khi latency P99 > 3 giây
      - alert: HighLatency
        expr: |
          histogram_quantile(0.99, 
            sum(rate({app="api-gateway"} | json [5m])) by (le)
          ) > 3000
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "High API latency"
          description: "P99 latency is {{ $value }}ms"

      # Alert khi chi phí vượt ngân sách
      - alert: HighCost
        expr: |
          sum(increase({app="api-gateway"} | json [1h])) by (model) * 0.42 > 100
        for:

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