저는 2년 전 이커머스 스타트업에서 AI 고객 서비스 봇을 개발할 때, 트래픽 급증에 대응하지 못해 시스템이 다운되는 경험을 했습니다.深夜 세일 이벤트 때 每秒 500건의 AI 쿼리가 발생했는데, 기존 단일 서버架构에서는 赤字가 발생했고 고객 이탈률이 급증했죠. Kubernetes와 HolySheep AI를 활용한 AI API 게이트웨이 클러스터를 구축한 뒤, 같은 트래픽을 안정적으로 처리하면서 월간 AI 비용을 40% 절감했습니다.이번 포스트에서는 개발자들이 실제 Production 환경에서 바로 적용할 수 있는 AI API Kubernetes 클러스터 배포 방법을 상세히 설명드리겠습니다.
왜 Kubernetes에 AI API 게이트웨이가 필요한가?
AI API를 단일 서버에서 호출하면 여러 문제점이 발생합니다.첫째, API 응답 지연이 발생할 때 단일 장애점(Single Point of Failure)이 생깁니다.둘째, 동시 요청 수가 증가하면 Rate Limit에 도달하거나 타임아웃이 빈번하게 발생합니다.셋째, 비용 최적화가 불가능합니다—모든 요청을 동일한 모델로 처리하면 저렴한 모델로 처리해도 되는 쿼리에도 비싼 모델을 사용하게 됩니다.
HolySheep AI를 Kubernetes 클러스터에 통합하면 이러한 문제들이 해결됩니다.단일 API 키로 GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash, DeepSeek V3.2 등 모든 주요 모델에 접근하면서, 请求 라우팅, 자동 재시도, 비용 모니터링, 응답 캐싱을 자동으로 관리해줍니다.
AI API Kubernetes 아키텍처 설계
실제 Production 환경에서 검증된 아키텍처는 다음과 같습니다:
┌─────────────────────────────────────────────────────────────────┐
│ Kubernetes Cluster │
├─────────────────────────────────────────────────────────────────┤
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ Ingress │ │ Ingress │ │ Ingress │ │
│ │ Controller │ │ Controller │ │ Controller │ │
│ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │
│ │ │ │ │
│ └────────────────┼────────────────┘ │
│ ▼ │
│ ┌───────────────────────┐ │
│ │ AI Gateway Pod │ │
│ │ (Nginx/Envoy) │ │
│ │ + Rate Limiter │ │
│ └───────────┬───────────┘ │
│ │ │
│ ┌────────────────┼────────────────┐ │
│ ▼ ▼ ▼ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ FastAPI │ │ FastAPI │ │ FastAPI │ │
│ │ Worker 1 │ │ Worker 2 │ │ Worker N │ │
│ │ (HPA) │ │ (HPA) │ │ (HPA) │ │
│ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │
│ │ │ │ │
│ └────────────────┼────────────────┘ │
│ ▼ │
│ ┌───────────────────────┐ │
│ │ Redis Cache │ │
│ │ (Response Cache) │ │
│ └───────────┬───────────┘ │
│ │ │
│ ▼ │
│ ┌───────────────────────┐ │
│ │ HolySheep AI │ │
│ │ Gateway Endpoint │ │
│ │ api.holysheep.ai │ │
│ └───────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘이 아키텍처의 핵심은 3-tier 설계입니다. Ingress Layer에서 트래픽을 분산시키고, Application Layer에서 AI 요청을 처리하며, Cache Layer에서 반복 요청을 최적화합니다.
Kubernetes 클러스터 구축 및 설정
먼저 Kubernetes 클러스터를 구축합니다. 저는 AWS EKS를 사용하지만, GKE나 로컬 Minikube도 동일한 원칙이 적용됩니다.
# EKS 클러스터 생성 (AWS CLI 필요)
eksctl create cluster \
--name holysheep-ai-cluster \
--region ap-northeast-2 \
--node-type t3.medium \
--nodes 3 \
--nodes-min 2 \
--nodes-max 10 \
--managed
클러스터 접속 확인
kubectl get nodes
출력 예시:
NAME STATUS ROLES AGE VERSION
ip-10-0-1-100.ap-northeast-2 Ready <none> 5m v1.28.0
ip-10-0-2-101.ap-northeast-2 Ready <none> 5m v1.28.0
ip-10-0-3-102.ap-northeast-2 Ready <none> 5m v1.28.0
이제 HolySheep AI API를 호출하는 Python FastAPI 어플리케이션을 만들어 Kubernetes에 배포하겠습니다. 먼저 프로젝트 구조를 생성합니다.
# 프로젝트 디렉토리 생성
mkdir -p ai-api-gateway/{app,k8s,tests}
cd ai-api-gateway
Python 의존성 파일 생성
cat > requirements.txt << 'EOF'
fastapi==0.109.0
uvicorn[standard]==0.27.0
httpx==0.26.0
redis==5.0.1
pydantic==2.5.3
python-dotenv==1.0.0
kubernetes==29.0.0
EOF
메인 어플리케이션 코드
cat > app/main.py << 'EOF'
"""
AI API Gateway - HolySheep AI 통합
Production-ready Kubernetes 배포용
"""
import os
import json
import hashlib
from datetime import datetime
from typing import Optional, List
from contextlib import asynccontextmanager
import httpx
import redis.asyncio as redis
from fastapi import FastAPI, HTTPException, Request, Depends
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
from dotenv import load_dotenv
load_dotenv()
HolySheep AI 설정
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
Redis 캐시 설정
REDIS_URL = os.getenv("REDIS_URL", "redis://redis-service:6379")
모델별 가격 (USD per 1M tokens)
MODEL_PRICING = {
"gpt-4.1": {"input": 8.0, "output": 8.0},
"claude-sonnet-4.5": {"input": 15.0, "output": 15.0},
"gemini-2.5-flash": {"input": 2.50, "output": 10.0},
"deepseek-v3.2": {"input": 0.42, "output": 1.68},
}
class ChatMessage(BaseModel):
role: str = Field(..., pattern="^(system|user|assistant)$")
content: str
class ChatRequest(BaseModel):
model: str = "deepseek-v3.2" # 기본값: 가장 저렴한 모델
messages: List[ChatMessage]
temperature: float = Field(default=0.7, ge=0, le=2)
max_tokens: Optional[int] = Field(default=1000, ge=1, le=32000)
stream: bool = False
class UsageStats(BaseModel):
prompt_tokens: int = 0
completion_tokens: int = 0
total_tokens: int = 0
estimated_cost_usd: float = 0.0
전역 Redis 클라이언트
redis_client: Optional[redis.Redis] = None
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup
global redis_client
try:
redis_client = redis.from_url(
REDIS_URL,
encoding="utf-8",
decode_responses=True,
socket_connect_timeout=5,
)
await redis_client.ping()
print("✅ Redis 연결 성공")
except Exception as e:
print(f"⚠️ Redis 연결 실패: {e}")
redis_client = None
yield
# Shutdown
if redis_client:
await redis_client.close()
app = FastAPI(
title="AI API Gateway",
description="HolySheep AI 통합 Kubernetes-ready API Gateway",
version="1.0.0",
lifespan=lifespan,
)
def get_cache_key(request: ChatRequest) -> str:
"""요청 기반 캐시 키 생성"""
content = json.dumps({
"model": request.model,
"messages": [m.model_dump() for m in request.messages],
"temperature": request.temperature,
"max_tokens": request.max_tokens,
}, sort_keys=True)
return f"ai:cache:{hashlib.sha256(content.encode()).hexdigest()}"
async def call_holysheep_api(request: ChatRequest) -> dict:
"""HolySheep AI API 호출"""
headers = {
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json",
}
payload = {
"model": request.model,
"messages": [m.model_dump() for m in request.messages],
"temperature": request.temperature,
"max_tokens": request.max_tokens,
}
async with httpx.AsyncClient(timeout=60.0) as client:
response = await client.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers=headers,
json=payload,
)
if response.status_code != 200:
raise HTTPException(
status_code=response.status_code,
detail=f"HolySheep API 오류: {response.text}",
)
return response.json()
def calculate_cost(usage: dict, model: str) -> float:
"""토큰 사용량 기반 비용 계산"""
if model not in MODEL_PRICING:
return 0.0
pricing = MODEL_PRICING[model]
input_cost = (usage.get("prompt_tokens", 0) / 1_000_000) * pricing["input"]
output_cost = (usage.get("completion_tokens", 0) / 1_000_000) * pricing["output"]
return round(input_cost + output_cost, 6)
@app.post("/v1/chat/completions")
async def chat_completions(request: ChatRequest) -> JSONResponse:
"""메인 채팅 완료 엔드포인트"""
# 캐시 확인
if redis_client:
cache_key = get_cache_key(request)
cached = await redis_client.get(cache_key)
if cached:
print(f"📦 캐시 히트: {cache_key[:16]}...")
cached_data = json.loads(cached)
cached_data["cached"] = True
return JSONResponse(content=cached_data)
# HolySheep API 호출
try:
response = await call_holysheep_api(request)
# 비용 계산 및 로깅
if "usage" in response:
cost = calculate_cost(response["usage"], request.model)
response["usage"]["estimated_cost_usd"] = cost
print(f"💰 {request.model} - 비용: ${cost:.6f}")
# 응답 캐싱 (TTL: 1시간)
if redis_client and "id" in response:
cache_key = get_cache_key(request)
await redis_client.setex(
cache_key,
3600,
json.dumps(response),
)
print(f"💾 캐시 저장: {cache_key[:16]}...")
response["cached"] = False
return JSONResponse(content=response)
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/health")
async def health_check():
"""헬스체크 엔드포인트"""
redis_status = "connected" if redis_client else "disconnected"
try:
if redis_client:
await redis_client.ping()
except:
redis_status = "error"
return {
"status": "healthy",
"timestamp": datetime.utcnow().isoformat(),
"redis": redis_status,
"holy_sheep_api": HOLYSHEEP_BASE_URL,
}
@app.get("/models")
async def list_models():
"""사용 가능한 모델 목록 반환"""
return {
"models": [
{
"id": "gpt-4.1",
"name": "GPT-4.1",
"pricing": MODEL_PRICING["gpt-4.1"],
"description": "가장 강력한 모델, 복잡한 작업에 적합",
},
{
"id": "claude-sonnet-4.5",
"name": "Claude Sonnet 4.5",
"pricing": MODEL_PRICING["claude-sonnet-4.5"],
"description": "긴 컨텍스트 처리, 코딩에 최적",
},
{
"id": "gemini-2.5-flash",
"name": "Gemini 2.5 Flash",
"pricing": MODEL_PRICING["gemini-2.5-flash"],
"description": "빠른 응답, 대량 처리용",
},
{
"id": "deepseek-v3.2",
"name": "DeepSeek V3.2",
"pricing": MODEL_PRICING["deepseek-v3.2"],
"description": "최저가 모델, 일반적인 작업에 적합",
},
]
}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
EOF
echo "✅ Python 어플리케이션 생성 완료"이제 Kubernetes 매니페스트 파일들을 생성하겠습니다.
# ConfigMap - 환경변수 설정
cat > k8s/configmap.yaml << 'EOF'
apiVersion: v1
kind: ConfigMap
metadata:
name: ai-gateway-config
namespace: default
data:
HOLYSHEEP_BASE_URL: "https://api.holysheep.ai/v1"
REDIS_URL: "redis://redis-service:6379"
LOG_LEVEL: "INFO"
EOF
Secret - API 키 (실제 환경에서는 sealed-secrets 또는 external-secrets 사용 권장)
cat > k8s/secret.yaml << 'EOF'
apiVersion: v1
kind: Secret
metadata:
name: holy-sheep-api-key
namespace: default
type: Opaque
stringData:
HOLYSHEEP_API_KEY: "YOUR_HOLYSHEEP_API_KEY" # 실제 키로 교체
EOF
Deployment - AI Gateway API
cat > k8s/deployment.yaml << 'EOF'
apiVersion: apps/v1
kind: Deployment
metadata:
name: ai-gateway
labels:
app: ai-gateway
version: v1
spec:
replicas: 3
selector:
matchLabels:
app: ai-gateway
template:
metadata:
labels:
app: ai-gateway
version: v1
annotations:
prometheus.io/scrape: "true"
prometheus.io/port: "8000"
prometheus.io/path: "/metrics"
spec:
containers:
- name: ai-gateway
image: python:3.11-slim
imagePullPolicy: IfNotPresent
command: ["bash", "-c"]
args:
- |
pip install fastapi uvicorn httpx redis pydantic python-dotenv &&
cd /app && python -m uvicorn app.main:app --host 0.0.0.0 --port 8000
ports:
- containerPort: 8000
name: http
envFrom:
- configMapRef:
name: ai-gateway-config
env:
- name: HOLYSHEEP_API_KEY
valueFrom:
secretKeyRef:
name: holy-sheep-api-key
key: HOLYSHEEP_API_KEY
resources:
requests:
memory: "256Mi"
cpu: "250m"
limits:
memory: "512Mi"
cpu: "500m"
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 10
periodSeconds: 30
readinessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 5
periodSeconds: 10
affinity:
podAntiAffinity:
preferredDuringSchedulingIgnoredDuringExecution:
- weight: 100
podAffinityTerm:
labelSelector:
matchExpressions:
- key: app
operator: In
values:
- ai-gateway
topologyKey: kubernetes.io/hostname
EOF
Service - 내부용 ClusterIP
cat > k8s/service.yaml << 'EOF'
apiVersion: v1
kind: Service
metadata:
name: ai-gateway-service
labels:
app: ai-gateway
spec:
type: ClusterIP
ports:
- port: 80
targetPort: 8000
protocol: TCP
name: http
selector:
app: ai-gateway
EOF
Horizontal Pod Autoscaler
cat > k8s/hpa.yaml << 'EOF'
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: ai-gateway-hpa
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: ai-gateway
minReplicas: 2
maxReplicas: 20
metrics:
- type: Resource
resource:
name: cpu
target:
type: Utilization
averageUtilization: 70
- type: Resource
resource:
name: memory
target:
type: Utilization
averageUtilization: 80
behavior:
scaleDown:
stabilizationWindowSeconds: 300
policies:
- type: Percent
value: 10
periodSeconds: 60
scaleUp:
stabilizationWindowSeconds: 0
policies:
- type: Percent
value: 100
periodSeconds: 15
- type: Pods
value: 4
periodSeconds: 15
selectPolicy: Max
EOF
Ingress - 외부 접근용
cat > k8s/ingress.yaml << 'EOF'
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
name: ai-gateway-ingress
annotations:
kubernetes.io/ingress.class: nginx
cert-manager.io/cluster-issuer: "letsencrypt-prod"
nginx.ingress.kubernetes.io/rate-limit: "100"
nginx.ingress.kubernetes.io/rate-limit-window: "1m"
nginx.ingress.kubernetes.io/proxy-body-size: "10m"
nginx.ingress.kubernetes.io/proxy-read-timeout: "60"
nginx.ingress.kubernetes.io/proxy-send-timeout: "60"
spec:
tls:
- hosts:
- api.yourdomain.com
secretName: ai-gateway-tls
rules:
- host: api.yourdomain.com
http:
paths:
- path: /
pathType: Prefix
backend:
service:
name: ai-gateway-service
port:
number: 80
EOF
Redis Deployment (Response Cache용)
cat > k8s/redis.yaml << 'EOF'
apiVersion: apps/v1
kind: Deployment
metadata:
name: redis
spec:
replicas: 1
selector:
matchLabels:
app: redis
template:
metadata:
labels:
app: redis
spec:
containers:
- name: redis
image: redis:7-alpine
ports:
- containerPort: 6379
command: ["redis-server", "--maxmemory", "256mb", "--maxmemory-policy", "allkeys-lru"]
resources:
requests:
memory: "128Mi"
cpu: "100m"
limits:
memory: "256Mi"
cpu: "200m"
livenessProbe:
tcpSocket:
port: 6379
initialDelaySeconds: 5
readinessProbe:
exec:
command: ["redis-cli", "ping"]
initialDelaySeconds: 5
---
apiVersion: v1
kind: Service
metadata:
name: redis-service
spec:
type: ClusterIP
ports:
- port: 6379
targetPort: 6379
selector:
app: redis
EOF
echo "✅ Kubernetes 매니페스트 생성 완료"이제 모든 리소스를 Kubernetes 클러스터에 배포하겠습니다.
# Kubernetes 리소스 배포
cd ai-api-gateway
1. ConfigMap과 Secret 배포
kubectl apply -f k8s/configmap.yaml
kubectl apply -f k8s/secret.yaml
2. Redis 배포
kubectl apply -f k8s/redis.yaml
3. AI Gateway Deployment 배포
kubectl apply -f k8s/deployment.yaml
4. Service 배포
kubectl apply -f k8s/service.yaml
5. HPA 배포 (metrics-server 필요)
kubectl apply -f k8s/hpa.yaml
6. Ingress 배포 (선택사항)
kubectl apply -f k8s/ingress.yaml
배포 상태 확인
echo "=== 배포 상태 확인 ==="
kubectl get all
echo ""
echo "=== Pod 상세 정보 ==="
kubectl get pods -o wide
echo ""
echo "=== HPA 상태 ==="
kubectl get hpa
echo ""
echo "=== 서비스 상태 ==="
kubectl get svc
로그 확인
echo ""
echo "=== AI Gateway 로그 ==="
kubectl logs -l app=ai-gateway --tail=50
포트 포워드로 로컬 테스트
echo ""
echo "=== 로컬 테스트 시작 ==="
kubectl port-forward svc/ai-gateway-service 8080:80 &
sleep 3
헬스체크 테스트
curl -s http://localhost:8080/health | python3 -m json.tool
모델 목록 확인
echo ""
echo "=== 사용 가능한 모델 목록 ==="
curl -s http://localhost:8080/models | python3 -m json.tool이제 실제 HolySheep AI API를 호출하는 테스트를 실행하겠습니다.
# HolySheep AI API 실제 호출 테스트
cat > test_api.py << 'EOF'
"""
HolySheep AI API 통합 테스트
실제 Production 환경 검증 스크립트
"""
import httpx
import time
import asyncio
from datetime import datetime
HolySheep AI 설정
BASE_URL = "http://localhost:8080" # 로컬 테스트용
Production에서는: "https://api.yourdomain.com"
async def test_health_check():
"""헬스체크 테스트"""
print("=" * 60)
print("1. 헬스체크 테스트")
print("=" * 60)
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.get(f"{BASE_URL}/health")
print(f"상태 코드: {response.status_code}")
print(f"응답: {response.json()}")
if response.status_code == 200:
print("✅ 헬스체크 성공")
else:
print("❌ 헬스체크 실패")
print()
async def test_model_list():
"""모델 목록 조회 테스트"""
print("=" * 60)
print("2. 모델 목록 조회 테스트")
print("=" * 60)
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.get(f"{BASE_URL}/models")
data = response.json()
for model in data["models"]:
print(f"\n모델: {model['id']}")
print(f" 이름: {model['name']}")
print(f" 입력 비용: ${model['pricing']['input']}/MTok")
print(f" 출력 비용: ${model['pricing']['output']}/MTok")
print(f" 설명: {model['description']}")
print("\n✅ 모델 목록 조회 성공")
async def test_ai_chat(model: str, prompt: str, description: str):
"""AI 채팅 테스트"""
print("=" * 60)
print(f"3. {description}")
print("=" * 60)
print(f"모델: {model}")
print(f"프롬프트: {prompt[:100]}...")
start_time = time.time()
async with httpx.AsyncClient(timeout=60.0) as client:
response = await client.post(
f"{BASE_URL}/v1/chat/completions",
json={
"model": model,
"messages": [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": prompt}
],
"temperature": 0.7,
"max_tokens": 500
}
)
elapsed = time.time() - start_time
print(f"상태 코드: {response.status_code}")
if response.status_code == 200:
data = response.json()
print(f"\n응답 시간: {elapsed*1000:.0f}ms")
print(f"캐시 여부: {data.get('cached', False)}")
if "usage" in data:
usage = data["usage"]
print(f"\n토큰 사용량:")
print(f" 입력 토큰: {usage.get('prompt_tokens', 0)}")
print(f" 출력 토큰: {usage.get('completion_tokens', 0)}")
print(f" 총 토큰: {usage.get('total_tokens', 0)}")
print(f" 예상 비용: ${usage.get('estimated_cost_usd', 0):.6f}")
print(f"\nAI 응답:")
if data.get("choices"):
content = data["choices"][0]["message"]["content"]
print(f" {content[:300]}...")
print("✅ API 호출 성공")
else:
print(f"❌ API 호출 실패: {response.text}")
print()
async def test_concurrent_requests():
"""동시 요청 테스트"""
print("=" * 60)
print("4. 동시 요청 테스트 (5개 동시 요청)")
print("=" * 60)
prompts = [
"한국의首都は谁知道ですか?",
"Explain quantum computing in simple terms",
"Pythonでリストの内包表記を教えてください",
"What is the meaning of life?",
"Write a haiku about programming",
]
tasks = [
test_ai_chat("deepseek-v3.2", prompt, f"동시 요청 {i+1}")
for i, prompt in enumerate(prompts)
]
start_time = time.time()
await asyncio.gather(*tasks)
total_time = time.time() - start_time
print(f"\n총 소요 시간: {total_time:.2f}초")
print(f"평균 응답 시간: {total_time/len(prompts):.2f}초")
print("✅ 동시 요청 테스트 완료")
async def test_cache_effectiveness():
"""캐시 효과 테스트"""
print("=" * 60)
print("5. 캐시 효과 테스트 (동일 요청 3회)")
print("=" * 60)
prompt = "What is Kubernetes? Explain briefly."
for i in range(3):
print(f"\n--- 요청 {i+1} ---")
await test_ai_chat("deepseek-v3.2", prompt, f"캐시 테스트 {i+1}")
await asyncio.sleep(1)
async def main():
"""메인 테스트 실행"""
print("\n" + "=" * 60)
print("🚀 HolySheep AI Kubernetes Gateway 통합 테스트")
print("=" * 60)
print(f"테스트 시간: {datetime.now().isoformat()}")
print(f"API 엔드포인트: {BASE_URL}")
print()
try:
# 순차 테스트
await test_health_check()
await test_model_list()
# 모델별 테스트
await test_ai_chat(
"deepseek-v3.2",
"한국의 기술 스타트업 생태계에 대해 설명해주세요.",
"DeepSeek V3.2 테스트 (저렴한 모델)"
)
await test_ai_chat(
"gemini-2.5-flash",
"Explain the benefits of microservices architecture.",
"Gemini 2.5 Flash 테스트 (빠른 응답)"
)
# 고급 테스트
await test_concurrent_requests()
await test_cache_effectiveness()
print("\n" + "=" * 60)
print("🎉 모든 테스트 완료!")
print("=" * 60)
except httpx.ConnectError:
print("❌ 연결 실패: Kubernetes 서비스가 실행 중인지 확인하세요.")
print(" kubectl port-forward svc/ai-gateway-service 8080:80")
except Exception as e:
print(f"❌ 테스트 중 오류 발생: {e}")
if __name__ == "__main__":
asyncio.run(main())
EOF
테스트 실행
python3 test_api.py모니터링 및 로깅 설정
Production 환경에서는 Prometheus와 Grafana를 활용한 모니터링이 필수입니다. 실제 운영에서는 平均 응답 시간, 토큰 사용량, 비용 추적 등을 실시간으로监控해야 합니다.
# Prometheus 모니터링 설정
cat > k8s/prometheus-config.yaml << 'EOF'
apiVersion: v1
kind: ConfigMap
metadata:
name: prometheus-config
data:
prometheus.yml: |
global:
scrape_interval: 15s
evaluation_interval: 15s
scrape_configs:
- job_name: 'ai-gateway'
kubernetes_sd_configs:
- role: pod
relabel_configs:
- source_labels: [__meta_kubernetes_pod_annotation_prometheus_io_scrape]
action: keep
regex: true
- source_labels: [__meta_kubernetes_pod_annotation_prometheus_io_path]
action: replace
target_label: __metrics_path__
regex: (.+)
- source_labels: [__address__, __meta_kubernetes_pod_annotation_prometheus_io_port]
action: replace
regex: ([^:]+)(?::\d+)?;(\d+)
replacement: $1:$2
target_label: __address__
- action: labelmap
regex: __meta_kubernetes_pod_label_(.+)
EOF
Grafana Dashboard 설정 (비용 모니터링)
cat > k8s/grafana-dashboard.yaml << 'EOF'
apiVersion: v1
kind: ConfigMap
metadata:
name: grafana-dashboard
labels:
grafana_dashboard: "1"
data:
ai-gateway-dashboard.json: |
{
"dashboard": {
"title": "AI Gateway Cost Monitoring",
"panels": [
{
"title": "Total API Cost (USD)",
"type": "stat",
"gridPos": {"h": 8, "w": 6, "x": 0, "y": 0},
"targets": [
{
"expr": "sum(increase(ai_request_cost_total[24h]))",
"legendFormat": "24h Cost"
}
],
"fieldConfig": {
"defaults": {
"unit": "currencyUSD",
"thresholds": {
"mode": "absolute",
"steps": [
{"color": "green", "value": null},
{"color": "yellow", "value": 100},
{"color": "red", "value": 500}
]
}
}
}
},
{
"title": "Requests by Model",
"type": "piechart",
"gridPos": {"h": 8, "w": 9, "x": 6, "y": 0},
"targets": [
{
"expr": "sum by (model) (increase(ai_requests_total[24h]))",
"legendFormat": "{{model}}"
}
]
},
{
"title": "Response Time (P95)",
"type": "timeseries",
"gridPos": {"h": 8, "w": 9, "x": 15, "y": 0},
"targets": [
{
"expr": "histogram_quantile(0.95, rate(ai_request_duration_seconds_bucket[5m]))",
"legendFormat": "P95 Latency"
}
]
},
{
"title": "Token Usage by Model",
"type": "timeseries",
"gridPos": {"h": 8, "w": 12, "x": 0, "y": 8},
"targets": [
{
"expr": "sum by (model) (increase(ai_tokens_total[24h]))",
"legendFormat": "{{model}}"
}
],
"fieldConfig": {
"defaults": {
"unit": "short",
"custom": {
"fillOpacity": 50,
"lineWidth": 2
}
}
}
},
{
"title": "Cache Hit Rate",
"type": "gauge",
"gridPos": {"h": 8, "w": 6, "x": 12, "y": 8},
"targets": [
{
"expr": "sum(rate(ai_cache_hits_total[5m])) / sum(rate(ai_requests_total[5m])) * 100"
}
],
"fieldConfig": {
"defaults": {
"unit": "percent",
"min": 0,
"max": 100,
"thresholds": {
"mode":