| name | predictive-scaling |
| description | 과거 트래픽 패턴과 시계열 예측을 기반으로 리소스 수요를 사전 예측하고 스케일링을 권고한다. 시간대별/요일별 계절성 분석, 이벤트 기반 수요 급증 예측, 비용 대비 성능 최적 구성 제안을 수행하며 cost-governance와 연동하여 예산 범위 내 스케일링을 보장한다. |
| argument-hint | [service-name or scaling-target] |
| user-invocable | true |
| model | claude-sonnet-4-6 |
| allowed-tools | Read,Grep,Bash,mcp__cloudwatch,mcp__prometheus,mcp__eks |
| ontology | {"references":["Deployment","Budget"]} |
When to Use
- Sub-Phase 5 (Govern) 확장 —
cost-governance와 연동하여 비용 효율적 스케일링
- 매일 cron으로 다음 24시간 수요 예측 및 스케일링 계획 생성
- 알려진 이벤트(마케팅 캠페인, 정기 배치 등) 전 사전 스케일링 준비
slo-management에서 SLO 위반 위험 감지 시 스케일업 트리거autopilot-deploy의 canary 단계에서 트래픽 증가 예측 시
사용 제외:
- 트래픽 패턴이 완전히 불규칙한 서비스 (reactive HPA로 충분)
- 과거 데이터가 14일 미만인 신규 서비스 (최소 2주 학습 필요)
- 이미 max_replicas에 도달한 서비스 (클러스터 용량 확장 필요)
Prerequisites
- awslabs.cloudwatch-mcp-server==0.0.25 — 과거 메트릭 조회.
- awslabs.prometheus-mcp-server==0.2.15 — 시계열 데이터.
- awslabs.eks-mcp-server==0.1.28 — HPA/Deployment 현재 설정 조회 전용 (
@latest 금지, PyPI 버전 pin 필수). 승인된 스케일링 스케줄 적용(CronHPA 업데이트)은 EKS MCP가 아닌 kubectl Bash 명령으로 실행하며, EKS MCP는 read-only 기본값을 유지합니다.
- 과거 트래픽 데이터: 최소 14일, 권장 30일 이상.
- 스케일링 정의:
.omao/plans/scaling/targets/${service}.yaml.
cost-governance 예산 파일 접근 — 비용 상한 내 스케일링 보장.
스케일링 대상 정의
service: rag-qa-agent
namespace: production
owner: ml-platform-team
scaling_targets:
- resource: deployment/rag-qa-agent
type: horizontal
current_hpa:
min_replicas: 2
max_replicas: 20
target_cpu_pct: 70
metrics:
- name: cpu_utilization
target_pct: 70
query: 'avg(rate(container_cpu_usage_seconds_total{pod=~"rag-qa-agent.*"}[5m])) / avg(kube_pod_container_resource_requests{resource="cpu",pod=~"rag-qa-agent.*"})'
- name: requests_per_second
target: 100
query: 'sum(rate(agent_request_total{service="rag-qa-agent"}[5m]))'
- name: memory_utilization
target_pct: 80
query: 'avg(container_memory_working_set_bytes{pod=~"rag-qa-agent.*"}) / avg(kube_pod_container_resource_requests{resource="memory",pod=~"rag-qa-agent.*"})'
prediction_config:
lookback_days: 30
forecast_hours: 24
granularity_minutes: 60
seasonality: [hourly, daily, weekly]
confidence_interval: 0.95
cost_constraints:
max_hourly_cost_usd: 50
cost_per_replica_hour_usd: 2.50
prefer_spot: true
spot_ratio: 0.7
spot_discount: 0.6
events_calendar:
- name: "Daily reindex"
cron: "0 3 * * *"
duration_hours: 1
traffic_multiplier: 0.3
- name: "Weekly report generation"
cron: "0 9 * * 1"
duration_hours: 2
traffic_multiplier: 2.5
예측 알고리즘
1. Seasonal Decomposition + Trend
시계열을 트렌드, 계절성, 잔차로 분해합니다.
import os
import numpy as np
import yaml
from datetime import datetime, timedelta
def seasonal_forecast(historical_data: list[float],
timestamps: list[str],
forecast_hours: int = 24,
confidence: float = 0.95) -> list[dict]:
"""계절성 분해 기반 수요 예측.
방법:
1. 7일 이동 평균으로 트렌드 추출
2. 시간대별(24h) + 요일별(7d) 계절성 패턴 추출
3. 잔차의 표준편차로 신뢰구간 계산
4. 트렌드 + 계절성으로 예측값 생성
"""
data = np.array(historical_data)
periods_per_day = 24
if len(data) < periods_per_day * 7:
raise ValueError("Minimum 7 days of hourly data required")
window = periods_per_day * 7
if len(data) >= window:
trend = np.convolve(data, np.ones(window)/window, mode='valid')
trend_slope = (trend[-1] - trend[0]) / len(trend) if len(trend) > 1 else 0
trend_last = trend[-1]
else:
trend_slope = 0
trend_last = np.mean(data)
hourly_pattern = np.zeros(periods_per_day)
hourly_counts = np.zeros(periods_per_day)
for i, ts in enumerate(timestamps):
hour = datetime.fromisoformat(ts.replace("Z", "+00:00")).hour
hourly_pattern[hour] += data[i]
hourly_counts[hour] += 1
hourly_pattern = hourly_pattern / np.maximum(hourly_counts, 1)
daily_factors = np.ones(7)
daily_counts = np.zeros(7)
daily_sums = np.zeros(7)
for i, ts in enumerate(timestamps):
dow = datetime.fromisoformat(ts.replace("Z", "+00:00")).weekday()
daily_sums[dow] += data[i]
daily_counts[dow] += 1
daily_avgs = daily_sums / np.maximum(daily_counts, 1)
overall_avg = np.mean(data)
if overall_avg > 0:
daily_factors = daily_avgs / overall_avg
residuals = []
for i, ts in enumerate(timestamps):
hour = datetime.fromisoformat(ts.replace("Z", "+00:00")).hour
expected = hourly_pattern[hour]
residuals.append(data[i] - expected)
residual_std = np.std(residuals)
z = 1.96
now = datetime.utcnow()
predictions = []
for h in range(forecast_hours):
future_time = now + timedelta(hours=h)
hour_of_day = future_time.hour
day_of_week = future_time.weekday()
if overall_avg > 0:
predicted = hourly_pattern[hour_of_day] * daily_factors[day_of_week]
predicted += trend_slope * h
else:
predicted = trend_last + trend_slope * h
upper = predicted + z * residual_std
lower = max(0, predicted - z * residual_std)
predictions.append({
"hour": h,
"timestamp": future_time.isoformat() + "Z",
"predicted_rps": round(float(predicted), 2),
"upper_ci": round(float(upper), 2),
"lower_ci": round(float(lower), 2),
})
return predictions
2. Event-Aware Adjustment
알려진 이벤트를 반영하여 예측을 보정합니다.
def apply_event_adjustments(predictions: list[dict],
events_calendar: list[dict]) -> list[dict]:
"""알려진 이벤트 기반 예측 보정.
이벤트 캘린더의 traffic_multiplier를 적용합니다.
"""
adjusted = []
for pred in predictions:
ts = datetime.fromisoformat(pred["timestamp"].replace("Z", "+00:00"))
multiplier = 1.0
event_name = None
for event in events_calendar:
parts = event["cron"].split()
cron_min, cron_hour = int(parts[0]), int(parts[1])
cron_dow = parts[4]
dow_match = (cron_dow == "*" or int(cron_dow) == ts.isoweekday() % 7)
hour_match = (cron_hour <= ts.hour < cron_hour + event["duration_hours"])
if dow_match and hour_match:
multiplier = event["traffic_multiplier"]
event_name = event["name"]
break
adj_pred = pred.copy()
adj_pred["predicted_rps"] = round(pred["predicted_rps"] * multiplier, 2)
adj_pred["upper_ci"] = round(pred["upper_ci"] * multiplier, 2)
adj_pred["lower_ci"] = round(max(0, pred["lower_ci"] * multiplier), 2)
if event_name:
adj_pred["event"] = event_name
adj_pred["multiplier"] = multiplier
adjusted.append(adj_pred)
return adjusted
3. Cost-Optimized Replica Calculation
비용 제약 내에서 최적 replica 수를 계산합니다.
def compute_scaling_schedule(predictions: list[dict],
scaling_config: dict,
cost_constraints: dict) -> list[dict]:
"""비용 제약 내 최적 스케일링 스케줄 계산.
원칙:
- upper_ci 기준으로 replica 수 결정 (안전 마진)
- 비용 상한 초과 시 target_utilization을 높여 replica 감소
- spot 인스턴스 비율 적용
"""
target_rps_per_replica = next(
(m["target"] for m in scaling_config.get("metrics", [])
if m.get("name") == "requests_per_second"),
100,
)
min_replicas = scaling_config.get("current_hpa", {}).get("min_replicas", 2)
max_replicas = scaling_config.get("current_hpa", {}).get("max_replicas", 20)
cost_per_hour = cost_constraints.get("cost_per_replica_hour_usd", 2.5)
max_hourly_cost = cost_constraints.get("max_hourly_cost_usd", 50)
spot_ratio = cost_constraints.get("spot_ratio", 0.7)
spot_discount = cost_constraints.get("spot_discount", 0.6)
effective_cost = cost_per_hour * (spot_ratio * spot_discount + (1 - spot_ratio) * 1.0)
max_replicas_by_cost = int(max_hourly_cost / effective_cost)
schedule = []
for pred in predictions:
needed_rps = pred["upper_ci"]
ideal_replicas = int(np.ceil(needed_rps / target_rps_per_replica))
replicas = max(min_replicas, min(ideal_replicas, max_replicas, max_replicas_by_cost))
hourly_cost = replicas * effective_cost
schedule.append({
"timestamp": pred["timestamp"],
"hour": pred["hour"],
"predicted_rps": pred["predicted_rps"],
"upper_ci_rps": pred["upper_ci"],
"replicas": replicas,
"hourly_cost_usd": round(hourly_cost, 2),
"event": pred.get("event"),
})
return schedule
실행 흐름
Step 1: 과거 데이터 수집
mcp__prometheus__query_range \
--query 'sum(rate(agent_request_total{service="rag-qa-agent"}[5m]))' \
--start "$(date -u -d '-30 days' +%s)" \
--end "$(date -u +%s)" \
--step 3600
mcp__cloudwatch__get_metric_data \
--metric-name "CPUUtilization" \
--namespace "ContainerInsights" \
--dimensions '[{"Name":"PodName","Value":"rag-qa-agent"}]' \
--period 3600 \
--start-time "$(date -u -d '-30 days' +%Y-%m-%dT%H:%M:%SZ)"
Step 2: 예측 실행
def run_forecast(service: str) -> dict:
"""전체 예측 파이프라인 실행."""
config_file = f".omao/plans/scaling/targets/{service}.yaml"
with open(config_file) as f:
config = yaml.safe_load(f)
pred_config = config["prediction_config"]
historical_data, timestamps = fetch_historical_traffic(
service, days=pred_config["lookback_days"]
)
predictions = seasonal_forecast(
historical_data, timestamps,
forecast_hours=pred_config["forecast_hours"],
confidence=pred_config["confidence_interval"],
)
if config.get("events_calendar"):
predictions = apply_event_adjustments(predictions, config["events_calendar"])
schedule = compute_scaling_schedule(
predictions,
config["scaling_targets"][0],
config["cost_constraints"],
)
return {
"service": service,
"predictions": predictions,
"schedule": schedule,
"generated_at": datetime.utcnow().isoformat() + "Z",
}
Step 3: Cost Governance 검증
def verify_cost_budget(schedule: list[dict], service: str) -> dict:
"""예측된 스케일링 비용이 예산 범위 내인지 검증.
cost-governance의 per_agent 예산과 비교합니다.
"""
daily_cost = sum(s["hourly_cost_usd"] for s in schedule)
monthly_projected = daily_cost * 30
budget_file = ".omao/plans/cost/budget.yaml"
if os.path.exists(budget_file):
with open(budget_file) as f:
budget = yaml.safe_load(f)
agent_budget = budget.get("monthly_ceiling_usd", {}).get("per_agent", {}).get(service)
else:
agent_budget = None
result = {
"daily_cost_usd": round(daily_cost, 2),
"monthly_projected_usd": round(monthly_projected, 2),
"agent_budget_usd": agent_budget,
"within_budget": True,
}
if agent_budget and monthly_projected > agent_budget:
result["within_budget"] = False
result["overage_usd"] = round(monthly_projected - agent_budget, 2)
result["recommendation"] = "Reduce max_replicas or increase spot_ratio"
return result
Step 4: 스케일링 권고 생성 및 저장
def generate_scaling_recommendation(service: str, schedule: list[dict],
cost_check: dict) -> str:
"""스케일링 권고 마크다운 생성."""
now = datetime.utcnow().strftime("%Y-%m-%d")
peak = max(schedule, key=lambda s: s["replicas"])
trough = min(schedule, key=lambda s: s["replicas"])
daily_cost = sum(s["hourly_cost_usd"] for s in schedule)
report = f"""## Scaling Recommendation — {service} ({now})
### Predicted Traffic Pattern
- **Peak**: {peak['timestamp'][:16]} UTC (예상 RPS: {peak['upper_ci_rps']}, replicas: {peak['replicas']})
- **Trough**: {trough['timestamp'][:16]} UTC (예상 RPS: {trough['upper_ci_rps']}, replicas: {trough['replicas']})
### Recommended Schedule
| Time (UTC) | Predicted RPS | Upper CI | Replicas | Cost/hr | Event |
|------------|--------------|----------|----------|---------|-------|
"""
for s in schedule:
event = s.get("event", "-")
report += f"| {s['timestamp'][11:16]} | {s['predicted_rps']} | {s['upper_ci_rps']} | {s['replicas']} | ${s['hourly_cost_usd']} | {event} |\n"
report += f"""
### Cost Summary
- **Daily Estimated Cost**: ${daily_cost:.2f}
- **Monthly Projected**: ${daily_cost * 30:.2f}
- **Budget Status**: {'✅ Within budget' if cost_check['within_budget'] else '⚠️ Over budget by $' + str(cost_check.get('overage_usd', 0))}
"""
if not cost_check["within_budget"]:
report += f"\n### ⚠️ Cost Warning\n{cost_check.get('recommendation', '')}\n"
forecast_dir = f".omao/plans/scaling/forecasts"
os.makedirs(forecast_dir, exist_ok=True)
with open(f"{forecast_dir}/{service}-{now}.md", "w") as f:
f.write(report)
return report
Step 5: 자동 적용 (승인 후)
사람 승인 후 EKS CronHPA 또는 scheduled scaling을 업데이트합니다.
def apply_scaling_schedule(service: str, schedule: list[dict],
approved: bool = False) -> dict:
"""승인된 스케일링 스케줄을 EKS에 적용.
⚠️ 사람 승인 필수 — approved=True일 때만 실행.
"""
if not approved:
return {
"status": "pending_approval",
"message": "Scaling schedule requires human approval before application.",
"approval_path": f".omao/plans/scaling/schedules/{service}-approved.yaml",
}
cron_entries = []
prev_replicas = None
for s in schedule:
if s["replicas"] != prev_replicas:
ts = datetime.fromisoformat(s["timestamp"].replace("Z", "+00:00"))
cron_entries.append({
"schedule": f"{ts.minute} {ts.hour} * * *",
"minReplicas": s["replicas"],
"maxReplicas": max(s["replicas"] + 2, s["replicas"]),
})
prev_replicas = s["replicas"]
return {
"status": "applied",
"cron_entries": len(cron_entries),
"message": f"Applied {len(cron_entries)} scaling schedule entries",
}
상태 관리
.omao/plans/scaling/targets/${service}.yaml — 스케일링 대상 정의.omao/plans/scaling/forecasts/${service}-${date}.md — 일간 예측 및 권고.omao/plans/scaling/schedules/${service}-approved.yaml — 승인된 스케줄.omao/state/scaling/${service}/current.json — 현재 스케일링 상태.omao/state/scaling/${service}/history.jsonl — 예측 vs 실제 비교 이력
Example Inputs/Outputs
Input: /predictive-scaling rag-qa-agent
Output (normal):
[00:05:00Z] Predictive scaling for: rag-qa-agent
[00:05:01Z] Loading config: .omao/plans/scaling/targets/rag-qa-agent.yaml
[00:05:02Z] Fetching 30-day historical data...
Data points: 720 (hourly, 30 days)
Average RPS: 450, Peak: 1200, Trough: 80
[00:05:04Z] Forecast generated (next 24 hours):
Peak: 14:00 UTC — predicted 980 RPS (upper CI: 1150)
Trough: 04:00 UTC — predicted 95 RPS (upper CI: 140)
[00:05:04Z] Event adjustments applied:
03:00-04:00 UTC: "Daily reindex" (×0.3) → RPS reduced
[00:05:05Z] Scaling schedule:
| Time | Replicas | Cost/hr |
| 00:00 | 3 | $4.50 |
| 06:00 | 6 | $9.00 |
| 09:00 | 10 | $15.00 |
| 14:00 | 12 | $18.00 |
| 18:00 | 8 | $12.00 |
| 22:00 | 4 | $6.00 |
[00:05:05Z] Cost verification:
Daily: $258.00
Monthly projected: $7,740
Budget (per_agent): $8,000
Status: ✅ Within budget (96.8%)
[00:05:06Z] Recommendation saved: .omao/plans/scaling/forecasts/rag-qa-agent-2026-05-08.md
[00:05:06Z] AWAITING HUMAN APPROVAL for schedule application.
Output (over budget):
[00:05:05Z] Cost verification:
Daily: $312.00
Monthly projected: $9,360
Budget (per_agent): $8,000
Status: ⚠️ Over budget by $1,360/month
[00:05:05Z] Optimization suggestions:
1. Increase spot_ratio 0.7 → 0.85 (saves ~$1,100/mo)
2. Reduce peak replicas 12 → 10 (saves ~$720/mo, risk: SLO pressure)
3. Request budget increase from cost-governance
[00:05:06Z] Sending cost verification request to cost-governance...
Output (SLO-triggered scale-up):
[10:15:00Z] SLO-triggered scaling for: rag-qa-agent
[10:15:00Z] Trigger: slo-management reports latency_p99 burn rate 2.3x
[10:15:01Z] Current state: 8 replicas, CPU 82%, RPS 750
[10:15:01Z] Immediate recommendation: scale to 12 replicas
Reason: reduce per-pod load to meet P99 < 500ms SLO
Cost impact: +$6.00/hr (+$144/day)
Budget check: within limits
[10:15:02Z] AWAITING HUMAN APPROVAL for immediate scale-up.
기존 스킬 연동
| 연동 대상 | 방향 | 설명 |
|---|
cost-governance | ↔ 양방향 | 예산 제약 수신 / 예상 비용 전달 |
autopilot-deploy | → 출력 | 배포 시 트래픽 증가 예측 정보 제공 |
anomaly-detection | ← 입력 | 예측 대비 실제 트래픽 이상 탐지 결과 수신 |
slo-management | ← 입력 | SLO 위반 위험 시 스케일업 트리거 |
continuous-eval | ← 입력 | 평가 부하 예측 (golden dataset 크기 기반) |
참고 자료
공식 문서
기술 블로그
관련 문서 (내부)
