SLI 메트릭을 자동 수집하여 SLO 대비 추적하고, Error Budget 소진율에 따라 배포 게이트를 제어한다. 번다운 차트 생성, 예측 기반 SLO 위반 사전 경고, Error Budget 정책(freeze/slow-down/normal) 자동 적용을 수행하며 continuous-eval의 품질 게이트를 보완한다.
Instalar com Codex ou Claude Copie este prompt, cole no Codex, Claude ou outro assistente e deixe que ele revise a página da skill e instale para você.
Um comando direto ignora o prompt de revisão. Verifique a origem antes de executá-lo.
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"}
