메뉴
과거 트래픽 패턴과 시계열 예측을 기반으로 리소스 수요를 사전 예측하고 스케일링을 권고한다. 시간대별/요일별 계절성 분석, 이벤트 기반 수요 급증 예측, 비용 대비 성능 최적 구성 제안을 수행하며 cost-governance와 연동하여 예산 범위 내 스케일링을 보장한다.
기본적으로 소스를 먼저 확인하는 Prompt가 선택됩니다. 직접 명령으로 전환하거나 로컬 사본을 다운로드할 수도 있습니다.
설치 여부를 결정하기 전에 SKILL.md와 SkillsMP에 표시된 보조 파일을 읽어 보세요.
Codex 또는 Claude로 설치 이 Prompt를 복사해 Codex, Claude 또는 다른 어시스턴트에 붙여 넣으면 Skill 페이지를 검토하고 설치를 진행할 수 있습니다.
직접 명령은 검토 Prompt를 거치지 않습니다. 실행하기 전에 소스를 확인하세요.
npx skills add https://github.com/aws-samples/sample-oh-my-aidlcops --skill predictive-scaling명령은 한 줄로 유지됩니다. 복사하기 전에 가로로 스크롤해 전체 내용을 확인하세요.
로컬 사본을 원하시나요? SkillsMP에서 현재 제공할 수 있는 파일을 다운로드하세요.
SOC 직업 분류 기준
SKILL.md 표시 중
| 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"]} |
| loop-interface | {"accepts-from":["cost-governance.budget-context","anomaly-detection.anomaly"],"emits-to":["slo-management.scaling-event"],"emits-signals":[]} |
cost-governance와 연동하여 비용 효율적 스케일링slo-management에서 SLO 위반 위험 감지 시 스케일업 트리거autopilot-deploy의 canary 단계에서 트래픽 증가 예측 시사용 제외:
@latest 금지, PyPI 버전 pin 필수). 승인된 스케일링 스케줄 적용(CronHPA 업데이트)은 EKS MCP가 아닌 kubectl Bash 명령으로 실행하며, EKS MCP는 read-only 기본값을 유지합니다..omao/plans/scaling/targets/${service}.yaml.cost-governance 예산 파일 접근 — 비용 상한 내 스케일링 보장.# .omao/plans/scaling/targets/rag-qa-agent.yaml
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:
[, , ]
시계열을 트렌드, 계절성, 잔차로 분해합니다.
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")
# 1. 트렌드 추출 (7일 이동 평균)
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)
# 2. 시간대별 계절성 (24시간 패턴)
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)
# 3. 요일별 보정 계수
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
# 4. 잔차 계산 및 신뢰구간
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-score for 95% confidence interval (hardcoded to avoid scipy dependency)
z = 1.96
# 5. 예측 생성
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
알려진 이벤트를 반영하여 예측을 보정합니다.
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:
# 간단한 cron 매칭 (시간/요일)
parts = event["cron"].split()
cron_min, cron_hour = int(parts[0]), int(parts[1])
cron_dow = parts[4]
# 요일 매칭 (cron: 0=Sun..6=Sat → Python isoweekday: 1=Mon..7=Sun)
dow_match = (cron_dow == "*" or int(cron_dow) == ts.isoweekday() % 7)
# 시간 매칭 (duration 고려)
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
비용 제약 내에서 최적 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 per replica (spot 혼합)
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:
# Upper CI 기준 필요 replica 수
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
# 30일 시계열 데이터 조회 (시간 단위)
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
# CloudWatch에서 CPU 사용률 조회
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)"
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",
}
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
# cost-governance 예산 로드
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
def generate_scaling_recommendation(service: str, schedule: list[dict],
cost_check: dict) -> str:
"""스케일링 권고 마크다운 생성."""
now = datetime.utcnow().strftime("%Y-%m-%d")
# Peak/Trough 식별
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
사람 승인 후 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",
}
# CronHPA manifest 생성
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"]
# EKS에 적용
# mcp__eks__apply_yaml(yaml_path=..., cluster_name=..., namespace=...)
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 실제 비교 이력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 크기 기반) |