화이트라벨 API Key 발급, 서브테넌트用量 격리, 청구서 분할과 초과熔断 설정 베스트 프랙티스
2026-05-13 | HolySheep AI 기술 블로그 | 엔지니어를 위한 프로덕션 가이드
개요
여러분이 SaaS 플랫폼을 운영하면서 고객에게 AI 기능을 제공해야 하는 상황이라면, 가장 중요한 질문은 이것입니다: "내 고객들에게 어떻게 안전하고 비용 효율적으로 AI API를開放하면서, 동시에 내 마진도 지키고 싶을까?"
저는 HolySheep AI의 플랫폼 기능을 활용하여 화이트라벨 AI 게이트웨이를 구축한 경험이 있습니다. 이번 포스트에서는 지금 가입하고 시작할 수 있는 HolySheep의 멀티테넌트 AI API 솔루션에 대해 깊이 있게 다루겠습니다.
왜 멀티테넌트 AI 게이트웨이가 필요한가
- 비용 격리: 각 고객이 자신의 사용량만 결제
- 보안 격리: 고객 A의 API Key로 고객 B의 리소스 접근 불가
- 마진 확보: HolySheep 가격에 서비스 마진을叠加하여 고객에게 판매
- 브랜딩: 자체 도메인과 UI로 화이트라벨 제공
- 통계 & 감사: 각 고객별 사용량, 비용, 에러율 모니터링
핵심 아키텍처: 서브테넌트 격리 모델
+-------------------+ +------------------------+ +------------------+
| HolySheep API | | Your SaaS Platform | | End Customers |
| | | | | (Sub-tenants) |
| +---------------+ | | +------------------+ | | +--------------+ |
| | Your Master | | | | API Gateway | | | | Customer A | |
| | API Key | |---->| | (Multi-tenant) | |---->| | API Key: sa- | |
| +---------------+ | | | | | | | a_xxx | |
| | | | - Rate Limiting | | | +--------------+ |
| | Sub-tenant | | | | - Quota Tracking | | | +--------------+ |
| | Management | | | | - Cost Splitting | | | | Customer B | |
| +---------------+ | | | - Circuit Break | | | | API Key: sa- | |
+-------------------+ | +------------------+ | | | b_xxx | |
+------------------------+ | +--------------+ |
+------------------+
1단계: 화이트라벨 API Key 발급 시스템
HolySheep는 단일 마스터 API Key로 모든 모델에 접근하지만, 여러분의 고객에게는 격리된 서브 API Key를 발급해야 합니다. 저는 이 시스템을 구현하면서 다음과 같은 패턴을 사용했습니다.
#!/usr/bin/env python3
"""
HolySheep Multi-tenant API Key Manager
서브테넌트용 API Key 발급 및 관리 시스템
"""
import hashlib
import hmac
import time
import requests
from typing import Optional, Dict, Any
from dataclasses import dataclass
from enum import Enum
class PlanTier(Enum):
FREE = "free"
STARTER = "starter"
PRO = "pro"
ENTERPRISE = "enterprise"
@dataclass
class TenantConfig:
tenant_id: str
plan: PlanTier
monthly_limit_usd: float
rate_limit_rpm: int # requests per minute
rate_limit_tpm: int # tokens per minute
models: list[str]
circuit_breaker_threshold: float # USD per hour
@dataclass
class IssuedKey:
key_id: str
key_hash: str
tenant_id: str
created_at: int
plan: PlanTier
class HolySheepMultiTenantManager:
"""HolySheep AI 멀티테넌트 게이트웨이 매니저"""
BASE_URL = "https://api.holysheep.ai/v1"
# 플랜별 기본 설정
PLAN_CONFIGS = {
PlanTier.FREE: {
"monthly_limit": 0,
"rate_limit_rpm": 10,
"rate_limit_tpm": 10000,
"models": ["gpt-4.1-mini", "claude-3-5-haiku"],
"circuit_threshold": 1.0
},
PlanTier.STARTER: {
"monthly_limit": 50,
"rate_limit_rpm": 60,
"rate_limit_tpm": 100000,
"models": ["gpt-4.1-mini", "gpt-4.1", "claude-3-5-haiku"],
"circuit_threshold": 5.0
},
PlanTier.PRO: {
"monthly_limit": 500,
"rate_limit_rpm": 300,
"rate_limit_tpm": 500000,
"models": ["gpt-4.1", "claude-sonnet-4-5", "gemini-2.5-flash"],
"circuit_threshold": 20.0
},
PlanTier.ENTERPRISE: {
"monthly_limit": -1, # 무제한
"rate_limit_rpm": 1000,
"rate_limit_tpm": 2000000,
"models": ["gpt-4.1", "claude-opus-4", "gemini-2.5-pro", "deepseek-v3"],
"circuit_threshold": 100.0
}
}
def __init__(self, master_api_key: str, markup_percentage: float = 20.0):
"""
Args:
master_api_key: HolySheep 마스터 API Key
markup_percentage: 고객에게 적용할 마진률 (%)
"""
self.master_key = master_api_key
self.markup = 1.0 + (markup_percentage / 100.0)
self._tenant_cache: Dict[str, TenantConfig] = {}
self._issued_keys: Dict[str, IssuedKey] = {}
self._usage_tracker: Dict[str, Dict[str, Any]] = {}
def generate_sub_tenant_key(self, tenant_id: str, plan: PlanTier) -> str:
"""
서브테넌트용 API Key 생성
실제 HolySheep API 호출에는 마스터 Key를 사용하지만,
SaaS 플랫폼 내부에서는 격리된 키 ID로 고객을 구분
"""
timestamp = int(time.time())
key_id = f"sa_{tenant_id[:8]}_{timestamp}"
# 키 해시 생성 (보안 검증용)
raw_key = f"{self.master_key}:{tenant_id}:{timestamp}"
key_hash = hashlib.sha256(raw_key.encode()).hexdigest()[:16]
# 발급된 키 정보 저장
issued = IssuedKey(
key_id=key_id,
key_hash=key_hash,
tenant_id=tenant_id,
created_at=timestamp,
plan=plan
)
self._issued_keys[key_id] = issued
# 테넌트 설정 캐시
config = self._get_tenant_config(tenant_id, plan)
self._tenant_cache[tenant_id] = config
# 사용량 추적 초기화
self._usage_tracker[tenant_id] = {
"monthly_spend": 0.0,
"monthly_start": timestamp,
"request_count": 0,
"error_count": 0,
"last_request": timestamp
}
# 실제 SaaS 플랫폼에서는 이 key_id를 고객에게 노출
return key_id
def _get_tenant_config(self, tenant_id: str, plan: PlanTier) -> TenantConfig:
"""테넌트별 설정 반환"""
plan_cfg = self.PLAN_CONFIGS[plan]
return TenantConfig(
tenant_id=tenant_id,
plan=plan,
monthly_limit_usd=plan_cfg["monthly_limit"],
rate_limit_rpm=plan_cfg["rate_limit_rpm"],
rate_limit_tpm=plan_cfg["rate_limit_tpm"],
models=plan_cfg["models"],
circuit_breaker_threshold=plan_cfg["circuit_threshold"]
)
def check_rate_limit(self, tenant_id: str, requested_tokens: int) -> bool:
"""레이트 리밋 확인"""
if tenant_id not in self._tenant_cache:
return False
config = self._tenant_cache[tenant_id]
usage = self._usage_tracker.get(tenant_id, {})
# 분당 RPM 체크 (간단한 구현)
current_time = int(time.time())
minute_key = current_time // 60
rpm_key = f"rpm_{minute_key}"
current_rpm = usage.get(rpm_key, 0)
if current_rpm >= config.rate_limit_rpm:
return False
# TPM 체크
tpm_key = f"tpm_{minute_key}"
current_tpm = usage.get(tpm_key, 0)
if current_tpm + requested_tokens > config.rate_limit_tpm:
return False
return True
def check_circuit_breaker(self, tenant_id: str) -> Dict[str, Any]:
"""
초과熔断 (Circuit Breaker) 체크
시간당 비용이 임계치를 초과하면 요청 차단
"""
if tenant_id not in self._tenant_cache:
return {"allowed": False, "reason": "unknown_tenant"}
config = self._tenant_cache[tenant_id]
usage = self._usage_tracker.get(tenant_id, {})
# 시간당 비용 계산
current_hour = int(time.time()) // 3600
hour_spend = usage.get(f"hourly_spend_{current_hour}", 0.0)
# 월간 한도 체크
if config.monthly_limit_usd > 0:
if usage.get("monthly_spend", 0.0) >= config.monthly_limit_usd:
return {
"allowed": False,
"reason": "monthly_limit_exceeded",
"limit": config.monthly_limit_usd,
"current": usage.get("monthly_spend", 0.0)
}
# 시간당 초과熔断 체크
if hour_spend >= config.circuit_breaker_threshold:
return {
"allowed": False,
"reason": "hourly_circuit_breaker",
"threshold": config.circuit_breaker_threshold,
"current_hour_spend": hour_spend
}
return {"allowed": True}
def ai_request(
self,
tenant_id: str,
model: str,
messages: list,
max_tokens: int = 1000
) -> Dict[str, Any]:
"""
HolySheep AI API 호출 (마스터 Key 사용)
모든 요청은 이 메서드를 통해 라우팅되어 사용량 추적
"""
# 1단계: Circuit Breaker 체크
breaker_status = self.check_circuit_breaker(tenant_id)
if not breaker_status["allowed"]:
return {
"error": True,
"code": "CIRCUIT_BREAKER_OPEN",
"message": f"요청이 차단되었습니다: {breaker_status['reason']}",
"details": breaker_status
}
# 2단계: 레이트 리밋 체크
if not self.check_rate_limit(tenant_id, max_tokens):
return {
"error": True,
"code": "RATE_LIMIT_EXCEEDED",
"message": "레이트 리밋 초과. 잠시 후 다시 시도하세요."
}
# 3단계: 모델 권한 체크
config = self._tenant_cache.get(tenant_id)
if config and model not in config.models:
return {
"error": True,
"code": "MODEL_NOT_ALLOWED",
"message": f"현재 플랜에서 {model} 사용 불가",
"allowed_models": config.models
}
# 4단계: HolySheep API 호출
headers = {
"Authorization": f"Bearer {self.master_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"max_tokens": max_tokens
}
try:
response = requests.post(
f"{self.BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=60
)
response.raise_for_status()
result = response.json()
# 5단계: 사용량 추적 및 비용 계산
self._track_usage(tenant_id, model, result, max_tokens)
return {
"error": False,
"data": result,
"tenant_id": tenant_id
}
except requests.exceptions.RequestException as e:
self._usage_tracker[tenant_id]["error_count"] += 1
return {
"error": True,
"code": "API_ERROR",
"message": str(e)
}
def _track_usage(self, tenant_id: str, model: str, response: Dict, requested_tokens: int):
"""사용량 추적 및 비용 계산"""
usage = self._usage_tracker[tenant_id]
current_time = int(time.time())
current_hour = current_time // 3600
current_minute = current_time // 60
# 기본 사용량 카운트
usage["request_count"] += 1
usage["last_request"] = current_time
# 분당 카운트 업데이트
usage[f"rpm_{current_minute}"] = usage.get(f"rpm_{current_minute}", 0) + 1
# 토큰 사용량 (응답에서 실제 사용량 가져오기)
if "usage" in response:
prompt_tokens = response["usage"].get("prompt_tokens", 0)
completion_tokens = response["usage"].get("completion_tokens", 0)
total_tokens = response["usage"].get("total_tokens", requested_tokens)
usage[f"tpm_{current_minute}"] = usage.get(f"tpm_{current_minute}", 0) + total_tokens
# HolySheep 가격표 (마크업 적용)
cost_per_mtok = self._calculate_cost_with_markup(model)
cost = (total_tokens / 1_000_000) * cost_per_mtok
usage["monthly_spend"] = usage.get("monthly_spend", 0.0) + cost
usage[f"hourly_spend_{current_hour}"] = usage.get(f"hourly_spend_{current_hour}", 0.0) + cost
def _calculate_cost_with_markup(self, model: str) -> float:
"""마크업 적용된 비용 계산"""
# HolySheep 기본 가격 (USD per million tokens)
base_prices = {
"gpt-4.1": 8.0,
"gpt-4.1-mini": 1.0,
"gpt-4.1-flash": 0.4,
"claude-sonnet-4-5": 15.0,
"claude-opus-4": 75.0,
"claude-3-5-haiku": 0.8,
"gemini-2.5-flash": 2.5,
"gemini-2.5-pro": 7.5,
"deepseek-v3": 0.42
}
base_cost = base_prices.get(model, 5.0)
return base_cost * self.markup
def get_tenant_usage_report(self, tenant_id: str) -> Dict[str, Any]:
"""테넌트별 사용량 리포트 생성"""
if tenant_id not in self._usage_tracker:
return {"error": "Tenant not found"}
usage = self._usage_tracker[tenant_id]
config = self._tenant_cache[tenant_id]
return {
"tenant_id": tenant_id,
"plan": config.plan.value,
"monthly_spend_usd": round(usage.get("monthly_spend", 0.0), 4),
"monthly_limit_usd": config.monthly_limit_usd,
"usage_percentage": round(
(usage.get("monthly_spend", 0.0) / config.monthly_limit_usd * 100)
if config.monthly_limit_usd > 0 else 0, 2
),
"total_requests": usage.get("request_count", 0),
"error_count": usage.get("error_count", 0),
"error_rate": round(
(usage.get("error_count", 0) / max(usage.get("request_count", 1), 1)) * 100, 2
),
"base_cost_usd": round(usage.get("monthly_spend", 0.0) / self.markup, 4),
"margin_earned_usd": round(
usage.get("monthly_spend", 0.0) * (1 - 1/self.markup), 4
)
}
========================================
사용 예시
========================================
if __name__ == "__main__":
# HolySheep 마스터 API Key로 초기화
# 마진 20% 설정 (고객에게 HolySheep 가격의 120%로 과금)
manager = HolySheepMultiTenantManager(
master_api_key="YOUR_HOLYSHEEP_API_KEY",
markup_percentage=20.0
)
# 고객 A: PRO 플랜 (월 $500 한도)
customer_a_key = manager.generate_sub_tenant_key(
tenant_id="customer_a_001",
plan=PlanTier.PRO
)
print(f"Customer A API Key: {customer_a_key}")
# 고객 B: STARTER 플랜 (월 $50 한도)
customer_b_key = manager.generate_sub_tenant_key(
tenant_id="customer_b_002",
plan=PlanTier.STARTER
)
print(f"Customer B API Key: {customer_b_key}")
# Customer A의 AI 요청
response = manager.ai_request(
tenant_id="customer_a_001",
model="gpt-4.1-mini",
messages=[
{"role": "system", "content": "당신은 도움이 되는 어시스턴트입니다."},
{"role": "user", "content": "안녕하세요!"}
],
max_tokens=500
)
if not response.get("error"):
print(f"SUCCESS: {response['data']['choices'][0]['message']['content'][:100]}...")
# 사용량 리포트 확인
report = manager.get_tenant_usage_report("customer_a_001")
print(f"\n사용량 리포트:")
print(f" 월간 지출: ${report['monthly_spend_usd']}")
print(f" 기본 비용: ${report['base_cost_usd']}")
print(f" 순 마진: ${report['margin_earned_usd']}")
print(f" 사용률: {report['usage_percentage']}%")
2단계: 청구서 분할 및 정산 시스템
멀티테넌트 환경에서 각 고객별 청구를 분리하는 것은 필수입니다. 저는 HolySheep의 토큰 기반 과금을 활용하여 자동화된 청구서 분할 시스템을 구현했습니다.
#!/usr/bin/env python3
"""
HolySheep Multi-tenant Billing System
자동 청구서 생성 및 정산 시스템
"""
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from decimal import Decimal, ROUND_HALF_UP
import json
@dataclass
class BillingPeriod:
start_date: datetime
end_date: datetime
def days_in_period(self) -> int:
return (self.end_date - self.start_date).days + 1
@dataclass
class UsageRecord:
timestamp: datetime
model: str
tokens_used: int
cost_base: Decimal
cost_markup: Decimal
request_id: str
@dataclass
class TenantInvoice:
tenant_id: str
period: BillingPeriod
line_items: List[Dict] = field(default_factory=list)
subtotal: Decimal = Decimal("0")
discount: Decimal = Decimal("0")
tax: Decimal = Decimal("0")
total: Decimal = Decimal("0")
usage_by_model: Dict[str, int] = field(default_factory=dict)
usage_by_day: Dict[str, int] = field(default_factory=dict)
class BillingSplitter:
"""청구서 분할 및 정산 매니저"""
# HolySheep 기본 가격표 (USD per million tokens)
HOLYSHEEP_PRICES = {
"gpt-4.1": Decimal("8.00"),
"gpt-4.1-mini": Decimal("1.00"),
"gpt-4.1-flash": Decimal("0.40"),
"claude-sonnet-4-5": Decimal("15.00"),
"claude-opus-4": Decimal("75.00"),
"claude-3-5-haiku": Decimal("0.80"),
"gemini-2.5-flash": Decimal("2.50"),
"gemini-2.5-pro": Decimal("7.50"),
"deepseek-v3": Decimal("0.42")
}
def __init__(
self,
platform_margin: Decimal = Decimal("0.20"), # 20% 마진
tax_rate: Decimal = Decimal("0.10"), # 10% 세금
min_invoice_amount: Decimal = Decimal("1.00") # 최소 청구 금액
):
self.margin = platform_margin
self.tax_rate = tax_rate
self.min_invoice = min_invoice_amount
self._usage_storage: Dict[str, List[UsageRecord]] = {}
def record_usage(
self,
tenant_id: str,
model: str,
tokens_used: int,
request_id: str
) -> UsageRecord:
"""사용량 기록"""
record = UsageRecord(
timestamp=datetime.utcnow(),
model=model,
tokens_used=tokens_used,
cost_base=self._calculate_base_cost(model, tokens_used),
cost_markup=Decimal("0"),
request_id=request_id
)
record.cost_markup = record.cost_base * (1 + self.margin)
if tenant_id not in self._usage_storage:
self._usage_storage[tenant_id] = []
self._usage_storage[tenant_id].append(record)
return record
def _calculate_base_cost(self, model: str, tokens: int) -> Decimal:
"""기본 비용 계산 (HolySheep 가격)"""
price_per_mtok = self.HOLYSHEEP_PRICES.get(model, Decimal("5.00"))
return (Decimal(tokens) / Decimal(1_000_000)) * price_per_mtok
def generate_invoice(
self,
tenant_id: str,
billing_period: BillingPeriod,
discount_code: Optional[str] = None,
custom_discount: Optional[Decimal] = None
) -> TenantInvoice:
"""청구서 생성"""
invoice = TenantInvoice(
tenant_id=tenant_id,
period=billing_period
)
# 해당 기간의 사용량 필터링
usage_records = [
r for r in self._usage_storage.get(tenant_id, [])
if billing_period.start_date <= r.timestamp <= billing_period.end_date
]
# 모델별 사용량 집계
model_usage: Dict[str, int] = {}
day_usage: Dict[str, int] = {}
for record in usage_records:
# 모델별
if record.model not in model_usage:
model_usage[record.model] = 0
model_usage[record.model] += record.tokens_used
# 일별
day_key = record.timestamp.strftime("%Y-%m-%d")
if day_key not in day_usage:
day_usage[day_key] = 0
day_usage[day_key] += record.tokens_used
# 라인 아이템
invoice.line_items.append({
"date": record.timestamp.isoformat(),
"model": record.model,
"tokens": record.tokens_used,
"base_cost": float(record.cost_base.quantize(Decimal("0.0001"))),
"markup_cost": float(record.cost_markup.quantize(Decimal("0.0001"))),
"request_id": record.request_id
})
invoice.subtotal += record.cost_markup
invoice.usage_by_model = model_usage
invoice.usage_by_day = day_usage
# 할인 적용
discount_amount = Decimal("0")
if discount_code == "NEWYEAR20":
discount_amount = invoice.subtotal * Decimal("0.20")
elif custom_discount:
discount_amount = custom_discount
invoice.discount = discount_amount.quantize(Decimal("0.01"))
# 세금 계산 (할인 후)
taxable_amount = invoice.subtotal - invoice.discount
invoice.tax = (taxable_amount * self.tax_rate).quantize(Decimal("0.01"))
# 총액
invoice.total = (taxable_amount + invoice.tax).quantize(Decimal("0.01"))
return invoice
def generate_invoice_report(self, invoice: TenantInvoice) -> str:
"""청구서 리포트 생성 (HTML/텍스트)"""
report = []
report.append("=" * 60)
report.append(f"HolySheep AI-powered Invoice")
report.append(f"플랫폼 명: HolySheep Gateway SaaS")
report.append("=" * 60)
report.append(f"테넌트 ID: {invoice.tenant_id}")
report.append(f"청구 기간: {invoice.period.start_date.strftime('%Y-%m-%d')} ~ "
f"{invoice.period.end_date.strftime('%Y-%m-%d')}")
report.append("")
report.append("-" * 60)
report.append("모델별 사용량 (Tokens)")
report.append("-" * 60)
for model, tokens in sorted(invoice.usage_by_model.items()):
cost = self._calculate_base_cost(model, tokens) * (1 + self.margin)
report.append(f" {model:<25} {tokens:>12,} ${float(cost):>8.2f}")
report.append("-" * 60)
report.append(f"{'소계 (마진 포함):':<40} ${float(invoice.subtotal):>8.2f}")
if invoice.discount > 0:
report.append(f"{'할인:':<40} -${float(invoice.discount):>8.2f}")
report.append(f"{'세금 (10%):':<40} ${float(invoice.tax):>8.2f}")
report.append("-" * 60)
report.append(f"{'총 청구 금액:':<40} ${float(invoice.total):>8.2f}")
report.append("=" * 60)
# 마진 분석
base_total = invoice.subtotal / (1 + self.margin)
margin_earned = invoice.subtotal - base_total
report.append("")
report.append("마진 분석:")
report.append(f" HolySheep 기본 비용: ${float(base_total.quantize(Decimal('0.01'))):.2f}")
report.append(f" 플랫폼 마진 (20%): ${float(margin_earned.quantize(Decimal('0.01'))):.2f}")
report.append(f" 고객 청구 금액: ${float(invoice.subtotal.quantize(Decimal('0.01'))):.2f}")
return "\n".join(report)
def export_to_json(self, invoice: TenantInvoice) -> str:
"""청구서를 JSON으로 내보내기 (웹 대시보드용)"""
return json.dumps({
"tenant_id": invoice.tenant_id,
"period": {
"start": invoice.period.start_date.isoformat(),
"end": invoice.period.end_date.isoformat()
},
"usage_by_model": invoice.usage_by_model,
"usage_by_day": invoice.usage_by_day,
"line_items": invoice.line_items,
"financials": {
"subtotal": float(invoice.subtotal),
"discount": float(invoice.discount),
"tax": float(invoice.tax),
"total": float(invoice.total)
},
"margin_breakdown": {
"base_cost": float(invoice.subtotal / (1 + self.margin)),
"platform_margin": float(
(invoice.subtotal / (1 + self.margin)) * self.margin
),
"margin_rate": float(self.margin * 100)
}
}, indent=2, ensure_ascii=False)
========================================
사용 예시
========================================
if __name__ == "__main__":
# 빌링 시스템 초기화 (20% 마진, 10% 세금)
billing = BillingSplitter(
platform_margin=Decimal("0.20"),
tax_rate=Decimal("0.10")
)
# 시뮬레이션: Customer A의 한 달 사용량
tenant_id = "customer_a_001"
# 각 모델별 사용량 시뮬레이션
usage_scenarios = [
("gpt-4.1-mini", 500000), # 500K tokens
("gpt-4.1", 100000), # 100K tokens
("claude-3-5-haiku", 300000), # 300K tokens
("gemini-2.5-flash", 200000),# 200K tokens
]
for model, tokens in usage_scenarios:
billing.record_usage(
tenant_id=tenant_id,
model=model,
tokens_used=tokens,
request_id=f"req_{model}_{tokens}"
)
# 청구 기간 (월간)
now = datetime.utcnow()
billing_period = BillingPeriod(
start_date=datetime(now.year, now.month, 1),
end_date=now
)
# 청구서 생성 (할인 코드 적용)
invoice = billing.generate_invoice(
tenant_id=tenant_id,
billing_period=billing_period,
discount_code=None # "NEWYEAR20"으로 변경하여 할인 적용 가능
)
# 리포트 출력
print(billing.generate_invoice_report(invoice))
print("\n" + "=" * 60)
print("JSON 내보내기:")
print("=" * 60)
print(billing.export_to_json(invoice))
3단계: 초과熔断 (Circuit Breaker) 상세 설정
예측 불가능한 사용량 급증이나 비정상적인 요청으로부터 플랫폼을 보호하는超额熔断 설정은 프로덕션에서 필수입니다. HolySheep의 유연한 레이트 리밋과 조합하여 다층 방어 체계를 구축했습니다.
#!/usr/bin/env python3
"""
HolySheep Circuit Breaker & Protection System
초과熔断 및 보호 메커니즘 구현
"""
import time
import threading
from dataclasses import dataclass, field
from enum import Enum
from typing import Dict, Callable, Optional, Any
from collections import defaultdict
from functools import wraps
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class BreakerState(Enum):
CLOSED = "closed" # 정상 - 요청 허용
OPEN = "open" # 차단 - 요청 거부
HALF_OPEN = "half_open" # 반개방 - 테스트 요청 허용
@dataclass
class CircuitBreakerConfig:
"""회로 차단기 설정"""
# 비용 기반
hourly_cost_threshold_usd: float = 10.0
daily_cost_threshold_usd: float = 100.0
monthly_cost_threshold_usd: float = 500.0
# 요청 기반
max_requests_per_minute: int = 100
max_requests_per_hour: int = 1000
max_tokens_per_minute: int = 100000
# 시간 기반
open_duration_seconds: int = 300 # 5분간 차단
half_open_max_requests: int = 5 # 반개방 시 허용 요청 수
# 점진적 복구
recovery_factor: float = 0.5 # 비용 임계치의 50%까지 감소 시 복구 시도
@dataclass
class TenantMetrics:
"""테넌트별 메트릭"""
tenant_id: str
# 비용 추적
hourly_cost: float = 0.0
daily_cost: float = 0.0
monthly_cost: float = 0.0
total_cost: float = 0.0
# 요청 추적
current_minute_requests: int = 0
current_hour_requests: int = 0
total_requests: int = 0
# 토큰 추적
current_minute_tokens: int = 0
# 에러 추적
error_count: int = 0
error_rate: float = 0.0
# 회로 차단기 상태
breaker_state: BreakerState = BreakerState.CLOSED
last_failure_time: float = 0
half_open_requests: int = 0
# 타임스탬프
hour_start: int = 0
day_start: int = 0
month_start: int = 0
class CircuitBreaker:
"""개별 회로 차단기"""
def __init__(self, tenant_id: str, config: CircuitBreakerConfig):
self.tenant_id = tenant_id
self.config = config
self.state = BreakerState.CLOS