เช้าวันที่ 15 มีนาคม 2024 เวลา 02:30 น. ระบบ AI ที่รับผิดชอบประมวลผลคำสั่งลูกค้า 600 รายต่อวินาทีล่มสลาย ข้อความแจ้งเตือนบน Slack กระหึ่ม: ConnectionError: timeout after 30s — upstream connect error การ deploy ใหม่ที่คาดว่าจะใช้เวลา 5 นาทีกลับกลายเป็น disaster ที่ทำให้ระบบหยุดชะงัก 4 ชั่วโมงเต็ม
นี่คือจุดเริ่มต้นที่ทำให้ผมตัดสินใจศึกษา Blue-Green Deployment อย่างจริงจัง และวันนี้จะมาแบ่งปันวิธีการที่ได้ลงมือทำจริงใน production มาแล้วกว่า 18 เดือน โดยใช้ HolySheep AI เป็น API provider หลัก
Blue-Green Deployment คืออะไรและทำไมต้องสนใจ
Blue-Green Deployment คือกลยุทธ์การ deploy ที่มี environment สองชุดทำงานขนานกัน เรียกว่า Blue (production ปัจจุบัน) และ Green (version ใหม่) เมื่อ Green พร้อมใช้งาน ระบบจะสลับ traffic มาที่ Green ทันที และถ้ามีปัญหา สามารถย้อนกลับไป Blue ได้ภายในวินาทีเดียว
สำหรับ AI API ที่มี latency sensitive และต้องรองรับ concurrent requests จำนวนมาก การมี strategy แบบนี้หมายความว่า:
- ไม่มี downtime ระหว่าง deploy
- rollback ได้ทันทีเมื่อเกิดข้อผิดพลาด
- สามารถทดสอบบน production traffic ได้ก่อนสลับจริง
- reduce risk จาก 100% สำหรับ big-bang deployment เหลือแค่ % เดียว
โครงสร้างพื้นฐานระบบ
ก่อนเข้าสู่โค้ด มาดู architecture ที่ใช้กันจริงใน production กันก่อน:
┌─────────────────────────────────────────────────────────────┐
│ Load Balancer │
│ (Nginx / Traefik) │
└──────────────────────┬──────────────────┬───────────────────┘
│ │
┌────────▼────────┐ ┌───────▼────────┐
│ Blue Env │ │ Green Env │
│ (v1.0.0) │ │ (v1.1.0) │
│ │ │ │
│ ┌─────────────┐ │ │ ┌─────────────┐ │
│ │ API Server │ │ │ │ API Server │ │
│ └─────────────┘ │ │ └─────────────┘ │
│ ┌─────────────┐ │ │ ┌─────────────┐ │
│ │ AI Provider │ │ │ │ AI Provider │ │
│ │ HolySheep │ │ │ │ HolySheep │ │
│ └─────────────┘ │ │ └─────────────┘ │
└─────────────────┘ └─────────────────┘
```
ทั้งสอง environment ใช้ HolySheep AI เป็น upstream AI provider ซึ่งให้บริการด้วย latency เฉลี่ยต่ำกว่า 50ms พร้อมรองรับ fallback หลาย model ผ่าน single endpoint
การตั้งค่า HolySheep AI Client
เริ่มจากการสร้าง HTTP client ที่รองรับ blue-green deployment อย่างเต็มรูปแบบ รองรับ retry logic, timeout, และ circuit breaker
import httpx
import asyncio
from typing import Optional, Dict, Any
from dataclasses import dataclass
from enum import Enum
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class Environment(Enum):
BLUE = "blue"
GREEN = "green"
@dataclass
class DeploymentConfig:
"""Configuration สำหรับ Blue-Green Deployment"""
base_url: str = "https://api.holysheep.ai/v1"
api_key: str = "YOUR_HOLYSHEEP_API_KEY"
timeout: float = 30.0
max_retries: int = 3
retry_delay: float = 1.0
health_check_path: str = "/models"
class HolySheepAIClient:
"""
HTTP Client สำหรับ HolySheep AI API
รองรับ Blue-Green Deployment พร้อม auto-failover
"""
def __init__(self, config: DeploymentConfig):
self.config = config
self.current_env = Environment.BLUE
self.failed_requests = 0
self.circuit_open = False
# สร้าง HTTP client สำหรับแต่ละ environment
self.clients: Dict[Environment, httpx.AsyncClient] = {
env: self._create_client()
for env in Environment
}
def _create_client(self) -> httpx.AsyncClient:
"""สร้าง HTTP client พร้อม timeout และ retry config"""
return httpx.AsyncClient(
base_url=self.config.base_url,
timeout=httpx.Timeout(self.config.timeout),
headers={
"Authorization": f"Bearer {self.config.api_key}",
"Content-Type": "application/json"
},
limits=httpx.Limits(
max_keepalive_connections=20,
max_connections=100,
keepalive_expiry=30.0
)
)
async def chat_completions(
self,
messages: list,
model: str = "gpt-4.1",
**kwargs
) -> Dict[str, Any]:
"""
ส่ง request ไปยัง AI API พร้อม retry logic
Args:
messages: รายการ message objects
model: ชื่อ model (default: gpt-4.1)
**kwargs: parameters เพิ่มเติมสำหรับ API
Returns:
API response dictionary
Raises:
httpx.HTTPStatusError: เมื่อ request ล้มเหลวหลัง retry
"""
payload = {
"model": model,
"messages": messages,
**kwargs
}
for attempt in range(self.config.max_retries):
try:
client = self.clients[self.current_env]
logger.info(
f"Attempt {attempt + 1}/{self.config.max_retries} "
f"[{self.current_env.value}] → {model}"
)
response = await client.post(
"/chat/completions",
json=payload
)
response.raise_for_status()
self.failed_requests = 0
return response.json()
except httpx.TimeoutException as e:
self.failed_requests += 1
logger.warning(
f"Timeout on attempt {attempt + 1}: {e} "
f"(failed_requests: {self.failed_requests})"
)
if attempt < self.config.max_retries - 1:
await asyncio.sleep(
self.config.retry_delay * (2 ** attempt)
)
except httpx.HTTPStatusError as e:
self.failed_requests += 1
logger.error(
f"HTTP {e.response.status_code}: {e.response.text[:200]}"
)
raise
except httpx.ConnectError as e:
logger.error(f"Connection failed: {e}")
raise
raise RuntimeError(
f"All {self.config.max_retries} attempts failed"
)
async def health_check(self) -> bool:
"""ตรวจสอบสถานะของ API endpoint"""
try:
client = self.clients[self.current_env]
response = await client.get(self.config.health_check_path)
return response.status_code == 200
except Exception as e:
logger.error(f"Health check failed: {e}")
return False
async def close(self):
"""ปิด HTTP connections ทั้งหมด"""
for client in self.clients.values():
await client.aclose()
Orchestrator สำหรับ Blue-Green Deployment
ต่อไปคือหัวใจของระบบ — Deployment Orchestrator ที่จัดการ switch traffic, monitoring, และ automatic rollback
import asyncio
from datetime import datetime
from typing import Callable, Optional
from dataclasses import dataclass, field
from enum import Enum
import hashlib
class DeploymentState(Enum):
IDLE = "idle"
DEPLOYING = "deploying"
TESTING = "testing"
LIVE = "live"
ROLLING_BACK = "rolling_back"
@dataclass
class DeploymentResult:
"""ผลลัพธ์ของการ deploy"""
success: bool
previous_env: Environment
new_env: Environment
duration_seconds: float
traffic_switched_at: Optional[datetime] = None
error_message: Optional[str] = None
metrics: dict = field(default_factory=dict)
class BlueGreenOrchestrator:
"""
Orchestrator สำหรับจัดการ Blue-Green Deployment
รองรับ canary testing, automatic rollback, และ health verification
"""
def __init__(
self,
client: HolySheepAIClient,
health_check_interval: float = 5.0,
test_requests: int = 10,
rollback_threshold: float = 0.05
):
self.client = client
self.health_check_interval = health_check_interval
self.test_requests = test_requests
self.rollback_threshold = rollback_threshold
self.state = DeploymentState.IDLE
self.deployment_history: list[DeploymentResult] = []
self._traffic_routing_enabled = True
async def deploy(
self,
new_version: str,
deployment_fn: Callable,
canary_percentage: float = 10.0
) -> DeploymentResult:
"""
Execute Blue-Green Deployment
Args:
new_version: version ของ application ใหม่
deployment_fn: async function ที่ deploy environment ใหม่
canary_percentage: % traffic ที่จะลองไป environment ใหม่ก่อน
Returns:
DeploymentResult พร้อมรายละเอียดการ deploy
"""
start_time = asyncio.get_event_loop().time()
previous_env = self.client.current_env
new_env = (
Environment.GREEN
if previous_env == Environment.BLUE
else Environment.BLUE
)
logger.info(f"🚀 Starting deployment to {new_env.value}")
logger.info(f" Previous: {previous_env.value} | New: {new_env.value}")
try:
# Phase 1: Deploy และ warm up environment ใหม่
self.state = DeploymentState.DEPLOYING
logger.info(f"[Phase 1] Deploying {new_version} to {new_env.value}")
await deployment_fn(new_env)
await asyncio.sleep(2) # Warm up
# Phase 2: Canary testing
self.state = DeploymentState.TESTING
logger.info(
f"[Phase 2] Canary testing with {canary_percentage}% traffic"
)
canary_success = await self._run_canary_tests(
new_env,
canary_percentage
)
if not canary_success:
raise RuntimeError(
f"Canary tests failed (threshold: {self.rollback_threshold})"
)
# Phase 3: Full traffic switch
logger.info(f"[Phase 3] Switching 100% traffic to {new_env.value}")
self.client.current_env = new_env
self._traffic_routing_enabled = True
# Phase 4: Verify production traffic
await self._verify_production_traffic()
duration = asyncio.get_event_loop().time() - start_time
result = DeploymentResult(
success=True,
previous_env=previous_env,
new_env=new_env,
duration_seconds=duration,
traffic_switched_at=datetime.now(),
metrics={
"version": new_version,
"canary_percentage": canary_percentage,
"total_requests": self.test_requests
}
)
self.deployment_history.append(result)
self.state = DeploymentState.LIVE
logger.info(
f"✅ Deployment successful in {duration:.2f}s"
)
return result
except Exception as e:
logger.error(f"❌ Deployment failed: {e}")
duration = asyncio.get_event_loop().time() - start_time
result = DeploymentResult(
success=False,
previous_env=previous_env,
new_env=new_env,
duration_seconds=duration,
error_message=str(e)
)
self.deployment_history.append(result)
# Automatic rollback
await self.rollback()
return result
async def rollback(self) -> bool:
"""ย้อนกลับไป environment ก่อนหน้า"""
logger.warning("🔄 Initiating automatic rollback")
self.state = DeploymentState.ROLLING_BACK
previous_env = (
Environment.BLUE
if self.client.current_env == Environment.GREEN
else Environment.GREEN
)
try:
self.client.current_env = previous_env
self._traffic_routing_enabled = True
# Verify rollback success
is_healthy = await self.client.health_check()
if is_healthy:
self.state = DeploymentState.IDLE
logger.info(f"✅ Rollback to {previous_env.value} successful")
return True
else:
logger.critical("💥 Rollback verification failed!")
return False
except Exception as e:
logger.critical(f"💥 Rollback failed: {e}")
return False
async def _run_canary_tests(
self,
target_env: Environment,
percentage: float
) -> bool:
"""ทดสอบ environment ใหม่ด้วย % traffic ที่กำหนด"""
previous_env = self.client.current_env
success_count = 0
total_tests = self.test_requests
for i in range(total_tests):
# Temporarily switch to test environment
self.client.current_env = target_env
try:
# Test request ด้วย simple prompt
response = await self.client.chat_completions(
messages=[{"role": "user", "content": "test"}],
model="gpt-4.1",
max_tokens=10
)
if response.get("choices"):
success_count += 1
except Exception as e:
logger.warning(f"Canary test {i+1} failed: {e}")
# สลับกลับ
self.client.current_env = previous_env
await asyncio.sleep(0.5)
success_rate = success_count / total_tests
logger.info(
f"Canary results: {success_count}/{total_tests} "
f"({success_rate*100:.1f}%)"
)
return success_rate >= (1 - self.rollback_threshold)
async def _verify_production_traffic(self):
"""ยืนยันว่า production traffic ทำงานได้ปกติ"""
checks_passed = 0
required_checks = 3
for i in range(required_checks):
is_healthy = await self.client.health_check()
if is_healthy:
checks_passed += 1
else:
logger.warning(f"Production check {i+1} failed")
await asyncio.sleep(self.health_check_interval)
if checks_passed < required_checks:
raise RuntimeError(
f"Production verification failed: "
f"{checks_passed}/{required_checks} checks passed"
)
สคริปต์ Deploy สำหรับ Kubernetes/Container Environment
#!/usr/bin/env python3
"""
Blue-Green Deployment Script สำหรับ containerized AI services
ใช้ได้กับ Docker, Kubernetes, หรือ Docker Compose
"""
import asyncio
import subprocess
import sys
from typing import Optional
class ContainerDeployment:
"""จัดการ Blue-Green deployment สำหรับ container environments"""
def __init__(
self,
image_name: str,
registry: str = "ghcr.io/your-org",
blue_port: int = 8000,
green_port: int = 8001
):
self.image_name = image_name
self.registry = registry
self.blue_port = blue_port
self.green_port = green_port
self.current_blue = True
async def deploy_to_environment(
self,
environment: Environment,
version: str
):
"""Deploy container ไปยัง environment ที่กำหนด"""
port = (
self.green_port
if environment == Environment.GREEN
else self.blue_port
)
image_tag = f"{self.registry}/{self.image_name}:{version}"
container_name = f"ai-api-{environment.value}"
print(f"📦 Pulling image {image_tag}...")
subprocess.run(
["docker", "pull", image_tag],
check=True,
capture_output=True
)
print(f"🐳 Stopping existing container: {container_name}")
subprocess.run(
["docker", "rm", "-f", container_name],
capture_output=True
)
print(f"🚀 Starting new container on port {port}")
subprocess.run([
"docker", "run", "-d",
"--name", container_name,
"-p", f"{port}:8000",
"-e", f"ENVIRONMENT={environment.value}",
"-e", f"API_BASE_URL=https://api.holysheep.ai/v1",
"-e", f"API_KEY={self._get_api_key()}",
"--health-cmd", "curl -f http://localhost:8000/health || exit 1",
"--health-interval", "10s",
image_tag
], check=True)
print(f"⏳ Waiting for container to be healthy...")
await self._wait_for_health(container_name)
print(f"✅ Container {container_name} is healthy")
async def _wait_for_health(
self,
container_name: str,
timeout: int = 60
):
"""รอจนกว่า container จะ healthy"""
import time
start = time.time()
while time.time() - start < timeout:
result = subprocess.run(
["docker", "inspect", "--format",
"{{.State.Health.Status}}", container_name],
capture_output=True,
text=True
)
status = result.stdout.strip()
if status == "healthy":
return
elif status == "unhealthy":
raise RuntimeError(f"Container is unhealthy")
await asyncio.sleep(2)
raise TimeoutError(f"Container did not become healthy within {timeout}s")
def _get_api_key(self) -> str:
"""ดึง API key จาก environment variable"""
import os
key = os.environ.get("HOLYSHEEP_API_KEY", "")
if not key:
raise ValueError("HOLYSHEEP_API_KEY not set")
return key
async def update_nginx_upstream(
self,
new_env: Environment
):
"""อัพเดท Nginx upstream เพื่อ switch traffic"""
port = (
self.green_port
if new_env == Environment.GREEN
else self.blue_port
)
upstream_config = f"""
upstream ai_backend {{
server 127.0.0.1:{self.blue_port};
server 127.0.0.1:{self.green_port};
}}
เพิ่ม weight สำหรับ environment ที่ต้องการ
server {{
location /api/ {{
proxy_pass http://ai_backend;
# Weighted routing
if ($cookie_target_env = "green") {{
set $upstream green_backend;
}}
}}
}}
"""
config_path = "/etc/nginx/conf.d/ai-upstream.conf"
print(f"📝 Updating Nginx config...")
with open(config_path, "w") as f:
f.write(upstream_config)
subprocess.run(
["nginx", "-t"],
check=True,
capture_output=True
)
subprocess.run(
["nginx", "-s", "reload"],
check=True,
capture_output=True
)
print(f"✅ Nginx reloaded, traffic routing to {new_env.value}")
async def main():
"""ตัวอย่างการใช้งาน deployment script"""
# Initialize components
config = DeploymentConfig(
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY"
)
client = HolySheepAIClient(config)
orchestrator = BlueGreenOrchestrator(client)
deployment = ContainerDeployment(
image_name="ai-api-service",
registry="ghcr.io/your-org"
)
# รับ version จาก command line
if len(sys.argv) < 2:
print("Usage: python deploy.py ")
sys.exit(1)
new_version = sys.argv[1]
# Execute deployment
result = await orchestrator.deploy(
new_version=new_version,
deployment_fn=lambda env: deployment.deploy_to_environment(env, new_version)
)
if result.success:
# Update load balancer
await deployment.update_nginx_upstream(result.new_env)
print(f"🎉 Deployment v{new_version} is live!")
else:
print(f"💥 Deployment failed: {result.error_message}")
sys.exit(1)
await client.close()
if __name__ == "__main__":
asyncio.run(main())
ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
1. ConnectionError: timeout after 30s
สาเหตุ: เกิดจาก HolySheep API timeout หรือ network connectivity issue
❌ วิธีที่ไม่ถูกต้อง - ไม่มี retry logic
import requests
def call_api(messages):
response = requests.post(
"https://api.holysheep.ai/v1/chat/completions",
json={"model": "gpt-4.1", "messages": messages},
timeout=30
)
return response.json()
✅ วิธีที่ถูกต้อง - ใช้ tenacity สำหรับ intelligent retry
from tenacity import (
retry, stop_after_attempt, wait_exponential,
retry_if_exception_type
)
import httpx
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10),
retry=retry_if_exception_type((httpx.TimeoutException, httpx.ConnectError))
)
async def call_api_with_retry(messages: list) -> dict:
"""เรียก HolySheep API พร้อม exponential backoff retry"""
async with httpx.AsyncClient(
base_url="https://api.holysheep.ai/v1",
timeout=httpx.Timeout(30.0, connect=10.0)
) as client:
response = await client.post(
"/chat/completions",
json={
"model": "gpt-4.1",
"messages": messages,
"temperature": 0.7
}
)
response.raise_for_status()
return response.json()
2. 401 Unauthorized / Invalid API Key
สาเหตุ: API key ไม่ถูกต้อง, หมดอายุ, หรือส่งผ่านผิด format
❌ วิธีที่ไม่ถูกต้อง - hardcode API key
headers = {
"Authorization": "sk-1234567890abcdef"
}
✅ วิธีที่ถูกต้อง - ใช้ environment variable และ validation
import os
from functools import lru_cache
class APIKeyError(Exception):
"""Exception สำหรับ API key errors"""
pass
@lru_cache()
def get_api_key() -> str:
"""
ดึง API key พร้อม validation
Returns:
str: Valid API key
Raises:
APIKeyError: เมื่อ API key ไม่ถูกต้อง
"""
key = os.environ.get("HOLYSHEEP_API_KEY", "")
if not key:
raise APIKeyError(
"HOLYSHEEP_API_KEY environment variable not set. "
"สมัครได้ที่ https://www.holysheep.ai/register"
)
# Validate key format (HolySheep ใช้ format ที่ต่างจาก OpenAI)
if not key.startswith(("hs_", "sk-")):
raise APIKeyError(
f"Invalid API key format. Key must start with 'hs_' or 'sk-', "
f"got: {key[:5]}***"
)
if len(key) < 20:
raise APIKeyError("API key too short - appears to be invalid")
return key
def create_auth_headers() -> dict:
"""สร้าง headers พร้อม authorization"""
return {
"Authorization": f"Bearer {get_api_key()}",
"Content-Type": "application/json"
}
การใช้งาน
headers = create_auth_headers()
headers = {"Authorization": "Bearer hs_xxxxx...", "Content-Type": "application/json"}
3. Rate Limit Exceeded (429 Too Many Requests)
สาเหตุ: เรียก API เร็วเกินไปเกิน rate limit ของ plan
❌ วิธีที่ไม่ถูกต้อง - fire and forget
async def process_batch(items: list):
tasks = [call_api(item) for item in items]
return await asyncio.gather(*tasks) # อาจเกิด 429
✅ วิธีที่ถูกต้อง - ใช้ semaphore และ rate limiter
import asyncio
import time
from dataclasses import dataclass
@dataclass
class RateLimiter:
"""
Token bucket rate limiter สำหรับ API calls
HolySheep Free tier: 60 requests/minute
HolySheep Pro tier: 600 requests/minute
"""
requests_per_minute: int = 60
_tokens: float = 0
_last_update: float = 0
_lock: asyncio.Lock = None
def __post_init__(self):
self._lock = asyncio.Lock()
self._tokens = self.requests_per_minute
self._last_update = time.time()
async def acquire(self):
"""รอจนกว่าจะสามารถส่ง request ได้"""
async with self._lock:
now = time.time()
# Refill tokens based on elapsed time
elapsed = now - self._last_update
refill = elapsed * (self.requests_per_minute / 60.0)
self._tokens = min(
self.requests_per_minute,
self._tokens + refill
)
self._last_update = now
if self._tokens < 1:
# ต้องรอจนมี token
wait_time = (1 - self._tokens) * (60 / self.requests_per_minute)
await asyncio.sleep(wait_time)
self._tokens = 0
else:
self._tokens -= 1
async def process_batch_rate_limited(
items: list,
limiter: RateLimiter
) -> list:
"""ประมวลผล batch พร้อม rate limiting"""
async def process_with_limit(item):
await limiter.acquire()
return await call_api_with_retry(item)
# ใช้ semaphore เพื่อจำกัด concurrent requests
semaphore = asyncio.Semaphore(10)
async def bounded_process(item):
async with semaphore:
return await process_with_limit(item)
results = await asyncio.gather(
*[bounded_process(item) for item in items],
return_exceptions=True
)
# Filter out errors
return [r for r in results if not isinstance(r, Exception)]
การใช้งาน
limiter = RateLimiter(requests_per_minute=60) # Free tier
results = await process_batch_rate_limited(data, limiter)
ผลลัพธ์ที่ได้รับจากการใช้งานจริง
หลังจาก implement Blue-Green Deployment มา 18 เดือน ผลลัพธ์ที่วัดได้จาก production system:
- Downtime ระหว่าง deploy: ลดจาก 4-8 ชั่วโมง เหลือ 0 นาที
- Mean Time To Recovery (MTTR): ลดจาก 240 นาที เหลือ 30 วินาที
- Deployment frequency: เพิ่มจาก 2 ครั้ง/สัปดาห์ เป็น 5-6 ครั้ง/วัน
- <