Skip to main content Skills Marketplace Descubra e explore skills de IA criadas pela comunidade.
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ê.
Mostrar detalhes do prompt Um comando direto ignora o prompt de revisão. Verifique a origem antes de executá-lo.
npx skills add https://github.com/aws-samples/sample-oh-my-aidlcops --skill predictive-scalingO comando permanece em uma só linha. Role horizontalmente para revisá-lo antes de copiar.
Prefere uma cópia local? Baixe os arquivos disponíveis atualmente no SkillsMP.
Baixar Zip Baixando... Ocupações relacionadas SOC
Baseado na classificação ocupacional SOC
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":[]}
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:.2 f}
- **Monthly Projected**: ${daily_cost * 30 :.2 f}
- **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
[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.
[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 크기 기반)
참고 자료
공식 문서
기술 블로그
관련 문서 (내부) 