SLI 메트릭을 자동 수집하여 SLO 대비 추적하고, Error Budget 소진율에 따라 배포 게이트를 제어한다. 번다운 차트 생성, 예측 기반 SLO 위반 사전 경고, Error Budget 정책(freeze/slow-down/normal) 자동 적용을 수행하며 continuous-eval의 품질 게이트를 보완한다.
Install with Codex or Claude Copy this prompt, paste it into Codex, Claude, or another assistant, and let it review the skill page and install it for you.
A direct command skips the review prompt. Inspect the source before running it.
SLI 메트릭을 자동 수집하여 SLO 대비 추적하고, Error Budget 소진율에 따라 배포 게이트를 제어한다. 번다운 차트 생성, 예측 기반 SLO 위반 사전 경고, Error Budget 정책(freeze/slow-down/normal) 자동 적용을 수행하며 continuous-eval의 품질 게이트를 보완한다.
# 경계 규칙: min은 inclusive(이상), max는 exclusive(미만). 최상위 밴드만 max=100 inclusive.
-
remaining_pct_min:
50
remaining_pct_max:
100
# inclusive (최상위)
mode:
normal
deploy_allowed:
true
deploy_frequency:
"unlimited"
description:
"정상 운영. 배포 제한 없음."
-
remaining_pct_min:
25
remaining_pct_max:
50
# exclusive (50% 미만)
mode:
slow-down
deploy_allowed:
true
deploy_frequency:
"max 1/day"
description:
"주의 구간. 하루 1회 배포로 제한."
-
remaining_pct_min:
10
remaining_pct_max:
25
# exclusive (25% 미만)
mode:
caution
deploy_allowed:
true
deploy_frequency:
"max 1/week"
description:
"경고 구간. 주 1회 배포로 제한. 안정성 우선."
-
remaining_pct_min:
0
remaining_pct_max:
10
# exclusive (10% 미만)
mode:
freeze
deploy_allowed:
false
exception:
"security-patch-only"
description:
"동결. 보안 패치 외 배포 금지."
alerts:
-
at_remaining_pct:
50
action:
notify
channel:
slack
-
at_remaining_pct:
25
action:
notify_and_slow
channel:
slack
+
pagerduty
-
at_remaining_pct:
10
action:
freeze_and_escalate
channel:
pagerduty
Error Budget 계산
import json
import os
import yaml
from datetime import datetime, timedelta
from dataclasses import dataclass
@dataclassclassErrorBudget:
slo_name: str
target: float
window_days: int
total_events: int
good_events: int
bad_events: int
allowed_bad_events: float
consumed_bad_events: int
remaining_budget: float# 남은 bad event 허용량
remaining_pct: float# 잔여 비율 (0~100)
burn_rate: float# 현재 소진 속도 (1.0 = 정상)
projected_exhaustion_days: float | None# 소진 예상 일수defcalculate_error_budget(slo_config: dict,
good_events: int,
total_events: int,
window_days: int) -> ErrorBudget:
"""Error Budget 잔량 계산.
Error Budget = (1 - SLO target) × total events
Remaining = Error Budget - actual bad events
"""
target = slo_config["target"]
bad_events = total_events - good_events
# 허용된 bad event 수
allowed_bad = (1 - target) * total_events
# 잔여 budget
remaining = max(0, allowed_bad - bad_events)
remaining_pct = (remaining / allowed_bad * 100) if allowed_bad > 0else100.0# Burn rate: 현재 소진 속도 / 지속 가능 소진 속도# 지속 가능 = allowed_bad / window_days (하루 평균 허용량)# 롤링 윈도우이므로 경과 일수 = window_days (전체 윈도우 기간에 걸쳐 관측)
elapsed_days = window_days
sustainable_daily_burn = allowed_bad / window_days
actual_daily_burn = bad_events / elapsed_days if elapsed_days > 0else0
burn_rate = actual_daily_burn / sustainable_daily_burn if sustainable_daily_burn > 0else0# 소진 예상 일수if actual_daily_burn > 0and remaining > 0:
projected_exhaustion = remaining / actual_daily_burn
else:
projected_exhaustion = None# 소진 안 됨 또는 이미 소진return ErrorBudget(
slo_name=slo_config["name"],
target=target,
window_days=window_days,
total_events=total_events,
good_events=good_events,
bad_events=bad_events,
allowed_bad_events=allowed_bad,
consumed_bad_events=bad_events,
remaining_budget=remaining,
remaining_pct=remaining_pct,
burn_rate=burn_rate,
projected_exhaustion_days=projected_exhaustion,
)
실행 흐름
Step 1: SLI 수집
각 SLO 정의에 따라 해당 메트릭을 조회합니다.
defcollect_sli(slo_config: dict) -> tuple[int, int]:
"""SLI 데이터 수집. (good_events, total_events) 반환.
SLI 타입에 따라 수집 방식이 다름:
- ratio: good/total 이벤트 수 직접 조회
- threshold: 임계값 이하 이벤트 수 / 전체 이벤트 수
- gauge: 현재 값이 target 이상인 시간 비율
"""
sli = slo_config["sli"]
window_days = slo_config.get("window", "30d")
window_seconds = int(window_days.replace("d", "")) * 86400if sli["type"] == "ratio":
# Prometheus range query# good = mcp__prometheus__query(sli["query_good"])# total = mcp__prometheus__query(sli["query_total"])
good = query_prometheus_scalar(sli["query_good"], window_seconds)
total = query_prometheus_scalar(sli["query_total"], window_seconds)
returnint(good), int(total)
elif sli["type"] == "threshold":
good = query_prometheus_scalar(sli["query_good"], window_seconds)
total = query_prometheus_scalar(sli["query_total"], window_seconds)
returnint(good), int(total)
elif sli["type"] == "gauge":
# Gauge: 시간 기반 — target 이상인 시간 비율 (query_good/query_total 사용)
good = query_prometheus_scalar(sli["query_good"], window_seconds)
total = query_prometheus_scalar(sli["query_total"], window_seconds)
returnint(good), int(total)
return0, 0defquery_prometheus_scalar(query: str, window_seconds: int) -> float:
"""Prometheus instant query 실행."""# mcp__prometheus__query(query=f'sum(increase({query}[{window_seconds}s]))')return0.0# placeholderdefquery_prometheus_range(query: str, window_seconds: int, step: int) -> list[float]:
"""Prometheus range query 실행."""# mcp__prometheus__query_range(query=query, start=..., end=..., step=step)return [] # placeholder
Step 2: Error Budget 계산 및 정책 적용
defevaluate_budget_policy(service: str, budgets: list[ErrorBudget],
policy_config: list[dict]) -> dict:
"""Error Budget 잔량에 따라 정책 결정.
가장 낮은 remaining_pct를 기준으로 정책을 적용합니다.
(하나의 SLO라도 위험하면 전체 서비스에 제한 적용)
"""# 가장 낮은 budget을 기준으로 정책 결정
min_budget = min(budgets, key=lambda b: b.remaining_pct)
remaining = min_budget.remaining_pct
# 정책 매칭 (min inclusive, max exclusive; 최상위 밴드만 max inclusive)
active_policy = Nonefor policy insorted(policy_config, key=lambda p: p["remaining_pct_min"]):
pmin = policy["remaining_pct_min"]
pmax = policy["remaining_pct_max"]
if pmax == 100:
in_band = pmin <= remaining <= pmax
else:
in_band = pmin <= remaining < pmax
if in_band:
active_policy = policy
breakifnot active_policy:
active_policy = {"mode": "normal", "deploy_allowed": True}
return {
"service": service,
"limiting_slo": min_budget.slo_name,
"remaining_pct": remaining,
"mode": active_policy["mode"],
"deploy_allowed": active_policy["deploy_allowed"],
"deploy_frequency": active_policy.get("deploy_frequency", "unlimited"),
"description": active_policy.get("description", ""),
}
Step 3: 배포 게이트 판정
Error Budget 잔량에 따라 autopilot-deploy에 배포 허용/차단 신호를 전달합니다.
defdeploy_gate_decision(service: str) -> dict:
"""Error Budget 기반 배포 게이트 판정.
autopilot-deploy의 pre-flight gate에서 호출됩니다.
"""# SLO 정의 로드
slo_file = f".omao/plans/slo/definitions/{service}.yaml"withopen(slo_file) as f:
slo_def = yaml.safe_load(f)
# 각 SLO별 budget 계산
budgets = []
for slo in slo_def["slos"]:
window_days = int(slo.get("window", "30d").replace("d", ""))
good, total = collect_sli(slo)
budget = calculate_error_budget(slo, good, total, window_days)
budgets.append(budget)
# 정책 평가
policy = evaluate_budget_policy(service, budgets, slo_def["error_budget_policy"])
# 게이트 결과 저장
gate_file = f".omao/state/slo/{service}/deploy-gate.json"
os.makedirs(os.path.dirname(gate_file), exist_ok=True)
withopen(gate_file, "w") as f:
json.dump(policy, f, indent=2)
return policy
Step 4: Burn Rate Alert & 예측
Error Budget 소진 속도를 기반으로 사전 경고를 발행합니다.
defcheck_burn_rate_alerts(service: str, budgets: list[ErrorBudget],
alert_config: list[dict]) -> list[dict]:
"""Burn rate 기반 알림 발행."""
alerts_fired = []
for budget in budgets:
for alert in alert_config:
if budget.remaining_pct <= alert["at_remaining_pct"]:
alert_event = {
"service": service,
"slo_name": budget.slo_name,
"remaining_pct": budget.remaining_pct,
"burn_rate": budget.burn_rate,
"projected_exhaustion_days": budget.projected_exhaustion_days,
"action": alert["action"],
"channel": alert["channel"],
"timestamp": datetime.utcnow().isoformat() + "Z",
}
alerts_fired.append(alert_event)
# 심각도에 따라 후속 조치if alert["action"] == "freeze_and_escalate":
trigger_deploy_freeze(service, budget)
trigger_incident_response(service, budget, severity="SEV3")
return alerts_fired
defpredict_budget_exhaustion(budget: ErrorBudget) -> dict:
"""Error Budget 소진 예측.
현재 burn rate가 유지될 경우 소진 시점을 예측합니다.
"""if budget.burn_rate <= 1.0:
return {
"status": "sustainable",
"message": f"Current burn rate ({budget.burn_rate:.2f}x) is sustainable",
}
if budget.projected_exhaustion_days isnotNone:
if budget.projected_exhaustion_days < 3:
urgency = "critical"elif budget.projected_exhaustion_days < 7:
urgency = "warning"else:
urgency = "info"return {
"status": "at_risk",
"urgency": urgency,
"projected_exhaustion_days": budget.projected_exhaustion_days,
"burn_rate": budget.burn_rate,
"message": f"Budget will exhaust in {budget.projected_exhaustion_days:.1f} days at current rate",
}
return {"status": "exhausted", "message": "Error budget already exhausted"}
