메뉴
CloudWatch 메트릭과 Prometheus 시계열 데이터에서 통계적 이상 징후를 자동 탐지하여 incident-response의 입력 소스로 제공한다. 베이스라인 학습(7일 이동 평균 + 3σ), 다변량 상관 분석, 계절성 보정을 수행하며 탐지된 anomaly를 severity 분류하여 알람을 생성한다.
기본적으로 소스를 먼저 확인하는 Prompt가 선택됩니다. 직접 명령으로 전환하거나 로컬 사본을 다운로드할 수도 있습니다.
설치 여부를 결정하기 전에 SKILL.md와 SkillsMP에 표시된 보조 파일을 읽어 보세요.
Codex 또는 Claude로 설치 이 Prompt를 복사해 Codex, Claude 또는 다른 어시스턴트에 붙여 넣으면 Skill 페이지를 검토하고 설치를 진행할 수 있습니다.
직접 명령은 검토 Prompt를 거치지 않습니다. 실행하기 전에 소스를 확인하세요.
npx skills add https://github.com/aws-samples/sample-oh-my-aidlcops --skill anomaly-detection명령은 한 줄로 유지됩니다. 복사하기 전에 가로로 스크롤해 전체 내용을 확인하세요.
로컬 사본을 원하시나요? SkillsMP에서 현재 제공할 수 있는 파일을 다운로드하세요.
SOC 직업 분류 기준
SKILL.md 표시 중
| name | anomaly-detection |
| description | CloudWatch 메트릭과 Prometheus 시계열 데이터에서 통계적 이상 징후를 자동 탐지하여 incident-response의 입력 소스로 제공한다. 베이스라인 학습(7일 이동 평균 + 3σ), 다변량 상관 분석, 계절성 보정을 수행하며 탐지된 anomaly를 severity 분류하여 알람을 생성한다. |
| argument-hint | [target-metric or agent-name] |
| user-invocable | true |
| model | claude-sonnet-4-6 |
| allowed-tools | Read,Grep,Bash,mcp__cloudwatch,mcp__prometheus |
| ontology | {"produces":["Incident"],"references":["Deployment","Agent"]} |
| loop-interface | {"accepts-from":[],"emits-to":["root-cause-analysis.anomaly","cost-governance.anomaly"],"emits-signals":["anomaly-detection.anomaly"]} |
incident-response의 알람 소스로 — CloudWatch/Prometheus 정적 임계값으로 잡히지 않는 이상 징후 탐지사용 제외:
@latest 금지, PyPI 버전 pin 필수)..omao/plans/observability/baselines/ 디렉토리 (본 skill이 자동 생성).incident-response skill 연동 — anomaly 탐지 시 자동 알람 생성.autopilot-deploy 상태 파일 접근 — 배포 직후 감도 조정 (배포 후 15분간 sensitivity 하향)..omao/plans/observability/monitoring-targets.yaml.# .omao/plans/observability/monitoring-targets.yaml
targets:
- name: rag-qa-agent
metrics:
- source: prometheus
query: 'rate(agent_request_errors_total{service="rag-qa-agent"}[5m])'
display_name: "Error Rate"
baseline_window_days: 7
sensitivity: 3.0 # σ multiplier
- source: prometheus
query: 'histogram_quantile(0.99, rate(agent_request_duration_seconds_bucket{service="rag-qa-agent"}[5m]))'
display_name: "P99 Latency"
baseline_window_days: 7
7일 이동 평균과 표준편차를 기반으로 정상 범위를 정의합니다. 베이스라인은 6시간마다 자동 갱신됩니다.
import numpy as np
from datetime import datetime, timedelta
from dataclasses import dataclass, field
import json
import yaml
@dataclass
class Baseline:
metric_name: str
mean: float
std: float
upper_bound: float
lower_bound: float
sensitivity: float
window_days: int
computed_at: str
seasonal_profile: dict = field(default_factory=dict)
def compute_baseline(metric_data: list[float], timestamps: list[str],
sensitivity: float = 3.0, window_days: int = 7) -> Baseline:
"""7일 이동 평균 + Nσ 기반 정상 범위 계산.
Args:
metric_data: 시계열 데이터 포인트
timestamps: ISO 8601 타임스탬프 목록
sensitivity: σ 배수 (기본 3.0)
window_days: 베이스라인 윈도우 (기본 7일)
Returns:
Baseline 객체 (정상 범위 포함)
"""
data = np.array(metric_data)
# Outlier 제거 (IQR 방식) — 베이스라인 오염 방지
q1, q3 = np.percentile(data, [25, 75])
iqr = q3 - q1
mask = (data >= q1 - 1.5 * iqr) & (data <= q3 + 1.5 * iqr)
clean_data = data[mask]
mean = float(np.mean(clean_data))
std = float(np.std(clean_data))
# 계절성 프로파일 생성 (시간대별 평균)
seasonal = {}
for ts, val in zip(timestamps, metric_data):
hour = datetime.fromisoformat(ts.replace("Z", "+00:00")).hour
dow = datetime.fromisoformat(ts.replace("Z", "+00:00")).weekday()
key = f"{dow}_{hour}"
seasonal.setdefault(key, []).append(val)
seasonal_profile = {k: float(np.mean(v)) for k, v in seasonal.items()}
return Baseline(
metric_name="",
mean=mean,
std=std,
upper_bound=mean + sensitivity * std,
lower_bound=max(0, mean - sensitivity * std),
sensitivity=sensitivity,
window_days=window_days,
computed_at=datetime.utcnow().isoformat() + "Z",
seasonal_profile=seasonal_profile,
)
시간대별/요일별 패턴을 보정하여 false positive를 줄입니다.
def seasonal_adjust(value: float, timestamp: str,
seasonal_profile: dict, global_mean: float) -> tuple[float, float]:
"""계절성 프로파일 기반 보정.
편차 기반 보정: value - seasonal_mean + global_mean
(비율 기반 value/factor는 저트래픽 시간대에서 false critical을 유발하므로 사용 금지)
Returns:
(adjusted_value, seasonal_mean)
"""
dt = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
key = f"{dt.weekday()}_{dt.hour}"
seasonal_mean = seasonal_profile.get(key)
if seasonal_mean is None:
return value, global_mean
# 편차 기반 보정: 현재 값에서 시간대 평균을 빼고 전역 평균을 더함
adjusted = value - seasonal_mean + global_mean
return adjusted, seasonal_mean
def is_in_exclusion_window(timestamp: str, exclusions: list[dict]) -> bool:
"""현재 시점이 제외 윈도우(배치 작업 등)에 해당하는지 확인."""
dt = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
for exc in exclusions:
if exc["type"] == "scheduled":
# 간단한 cron 매칭 (시간/분만)
parts = exc["cron"].split()
cron_min, cron_hour = int(parts[0]), int(parts[1])
if dt.hour == cron_hour and dt.minute >= cron_min:
if (dt.minute - cron_min) < exc["duration_min"]:
return True
return False
단일 메트릭 이상이 아닌, 관련 메트릭 간 상관관계 붕괴를 탐지합니다.
def correlation_check(metrics: dict[str, list[float]],
expected_correlations: list[dict],
window_size: int = 60) -> list[dict]:
"""메트릭 간 상관관계 이탈 탐지.
Args:
metrics: {metric_name: [values]} 딕셔너리
expected_correlations: 기대 상관관계 정의 목록
window_size: 상관관계 계산 윈도우 (데이터 포인트 수)
Returns:
탐지된 상관관계 이상 목록
"""
anomalies = []
for corr_def in expected_correlations:
m1_name, m2_name = corr_def["pair"].split(":")
expected_r = corr_def["expected_r"]
m1_data = metrics.get(m1_name, [])
m2_data = metrics.get(m2_name, [])
if len(m1_data) < window_size or len(m2_data) < window_size:
continue
# 최근 window_size 포인트로 상관관계 계산
recent_m1 = np.array(m1_data[-window_size:])
recent_m2 = np.array(m2_data[-window_size:])
# 분산이 0이면 상관관계 계산 불가
if np.std(recent_m1) == 0 or np.std(recent_m2) == 0:
continue
actual_r = float(np.corrcoef(recent_m1, recent_m2)[0, 1])
deviation = abs(actual_r - expected_r)
if deviation > 0.3: # 상관관계 0.3 이상 이탈
anomalies.append({
"type": "correlation_breakdown",
"pair": corr_def["pair"],
"description": corr_def.get("description", ""),
"expected_r": expected_r,
"actual_r": actual_r,
"deviation": deviation,
"severity": "critical" if deviation > 0.5 else "warning",
"detected_at": datetime.utcnow().isoformat() + "Z",
})
return anomalies
@dataclass
class AnomalyEvent:
metric_name: str
timestamp: str
value: float
baseline_mean: float
baseline_std: float
sigma_deviation: float
severity: str # critical, warning, info
type: str # point_anomaly, correlation_breakdown, trend_shift
context: dict = field(default_factory=dict)
def classify_anomaly(value: float, baseline: Baseline,
timestamp: str) -> AnomalyEvent | None:
"""단일 데이터 포인트의 이상 여부 판정 및 severity 분류."""
# 계절성 보정
adjusted_value, seasonal_mean = seasonal_adjust(value, timestamp, baseline.seasonal_profile, baseline.mean)
adjusted_mean = baseline.mean
if baseline.std == 0:
return None
sigma = abs(adjusted_value - adjusted_mean) / baseline.std
if sigma < 2.0:
return None # 정상 범위
# Severity 분류
if sigma >= 5.0:
severity = "critical"
elif sigma >= baseline.sensitivity: # 기본 3.0
severity = "warning"
else:
severity = "info"
return AnomalyEvent(
metric_name=baseline.metric_name,
timestamp=timestamp,
value=value,
baseline_mean=baseline.mean,
baseline_std=baseline.std,
sigma_deviation=sigma,
severity=severity,
type="point_anomaly",
context={
"seasonal_factor": factor,
"adjusted_value": adjusted_value,
"upper_bound": baseline.upper_bound,
"lower_bound": baseline.lower_bound,
},
)
| Severity | 기준 | 후속 조치 |
|---|---|---|
| Critical | 5σ 이상 이탈 또는 다변량 상관 붕괴 3건 이상 동시 | incident-response SEV2 자동 호출 |
| Warning | 3σ~5σ 이탈 또는 계절성 보정 후에도 이상 | incident-response SEV3 큐에 적재 |
| Info | 2σ~3σ 이탈 (계절성 보정 전) | 로그 기록만, 주간 리뷰 큐 |
# 모니터링 대상 로드
TARGETS_FILE=".omao/plans/observability/monitoring-targets.yaml"
# Prometheus 메트릭 조회 (최근 1시간, 1분 간격)
mcp__prometheus__query_range \
--query 'rate(agent_request_errors_total{service="rag-qa-agent"}[5m])' \
--start "$(date -u -d '-1 hour' +%s)" \
--end "$(date -u +%s)" \
--step 60
# CloudWatch 메트릭 조회
mcp__cloudwatch__get_metric_data \
--metric-name "InvocationLatency" \
--namespace "AWS/Bedrock" \
--dimensions '[{"Name":"ModelId","Value":"anthropic.claude-sonnet-4-6-v1:0"}]' \
--period 60 \
--start-time "$(date -u -d '-1 hour' +%Y-%m-%dT%H:%M:%SZ)" \
--end-time "$(date -u +%Y-%m-%dT%H:%M:%SZ)"
import os
import yaml
BASELINE_DIR = ".omao/plans/observability/baselines"
def load_or_refresh_baseline(metric_name: str, metric_config: dict,
force_refresh: bool = False) -> Baseline:
"""베이스라인 로드. 6시간 이상 경과 시 자동 갱신."""
baseline_file = os.path.join(BASELINE_DIR, f"{metric_name}.yaml")
if os.path.exists(baseline_file) and not force_refresh:
with open(baseline_file) as f:
data = yaml.safe_load(f)
computed_at = datetime.fromisoformat(data["computed_at"].replace("Z", "+00:00"))
age_hours = (datetime.utcnow() - computed_at.replace(tzinfo=None)).total_seconds() / 3600
if age_hours < 6: # 6시간 미만이면 기존 베이스라인 사용
return Baseline(**data)
# 베이스라인 갱신: 7일 데이터 조회
window_days = metric_config.get("baseline_window_days", 7)
historical_data, timestamps = fetch_historical_data(
metric_config, days=window_days
)
baseline = compute_baseline(
historical_data, timestamps,
sensitivity=metric_config.get("sensitivity", 3.0),
window_days=window_days,
)
baseline.metric_name = metric_name
# 저장
os.makedirs(BASELINE_DIR, exist_ok=True)
with open(baseline_file, "w") as f:
yaml.dump(vars(baseline), f, default_flow_style=False)
return baseline
def run_detection_cycle(targets_file: str) -> list[AnomalyEvent]:
"""단일 탐지 사이클 실행."""
with open(targets_file) as f:
config = yaml.safe_load(f)
all_anomalies = []
for target in config["targets"]:
target_metrics = {}
for metric_config in target["metrics"]:
metric_name = metric_config["display_name"]
# 제외 윈도우 확인
now = datetime.utcnow().isoformat() + "Z"
if is_in_exclusion_window(now, target.get("exclusions", [])):
continue
# 현재 값 조회
current_values, timestamps = fetch_recent_data(metric_config, minutes=5)
if not current_values:
continue
target_metrics[metric_name] = current_values
# 베이스라인 로드
baseline = load_or_refresh_baseline(metric_name, metric_config)
# 각 데이터 포인트 검사
for val, ts in zip(current_values, timestamps):
anomaly = classify_anomaly(val, baseline, ts)
if anomaly and anomaly.severity != "info":
all_anomalies.append(anomaly)
# 상관관계 검사
if target.get("correlations") and len(target_metrics) >= 2:
corr_anomalies = correlation_check(
target_metrics, target["correlations"]
)
for ca in corr_anomalies:
all_anomalies.append(AnomalyEvent(
metric_name=ca["pair"],
timestamp=ca["detected_at"],
value=ca["actual_r"],
baseline_mean=ca["expected_r"],
baseline_std=0,
sigma_deviation=0,
severity=ca["severity"],
type="correlation_breakdown",
context=ca,
))
return all_anomalies
import json
import os
ANOMALY_STATE_DIR = ".omao/state/anomaly"
def route_anomalies(anomalies: list[AnomalyEvent]) -> dict:
"""탐지된 anomaly를 severity에 따라 라우팅."""
os.makedirs(ANOMALY_STATE_DIR, exist_ok=True)
routed = {"critical": [], "warning": [], "info": []}
for anomaly in anomalies:
# 상태 파일 기록
event_file = os.path.join(
ANOMALY_STATE_DIR,
f"{anomaly.timestamp.replace(':', '-')}-{anomaly.metric_name.replace(' ', '_')}.json"
)
with open(event_file, "w") as f:
json.dump(vars(anomaly), f, indent=2, default=str)
routed[anomaly.severity].append(anomaly)
# Critical → incident-response SEV2 즉시 호출
if routed["critical"]:
trigger_incident_response(
severity="SEV2",
source="anomaly-detection",
anomalies=routed["critical"],
)
# Warning → incident-response SEV3 큐 적재
if routed["warning"]:
enqueue_incident(
severity="SEV3",
source="anomaly-detection",
anomalies=routed["warning"],
)
return {
"total_detected": len(anomalies),
"critical": len(routed["critical"]),
"warning": len(routed["warning"]),
"info": len(routed["info"]),
}
autopilot-deploy의 최근 배포 이벤트와 anomaly 시점을 매핑하여 change-correlated anomaly를 식별합니다.
def check_deploy_context(anomaly: AnomalyEvent) -> dict | None:
"""anomaly 시점에 최근 배포가 있었는지 확인."""
deploy_state_dir = ".omao/state/autopilot-deploy"
if not os.path.exists(deploy_state_dir):
return None
# 최근 배포 상태 확인
for state_file in sorted(os.listdir(deploy_state_dir), reverse=True):
if not state_file.endswith(".json"):
continue
with open(os.path.join(deploy_state_dir, state_file)) as f:
deploy = json.load(f)
deploy_time = datetime.fromisoformat(deploy.get("started_at", "").replace("Z", "+00:00"))
anomaly_time = datetime.fromisoformat(anomaly.timestamp.replace("Z", "+00:00"))
# 배포 후 15분 이내 anomaly → change-correlated
delta_min = (anomaly_time - deploy_time).total_seconds() / 60
if 0 < delta_min < 15:
return {
"correlated_deploy": deploy.get("target"),
"deploy_version": deploy.get("version"),
"time_after_deploy_min": delta_min,
"recommendation": "Consider rollback if anomaly persists",
}
return None
.omao/plans/observability/monitoring-targets.yaml — 모니터링 대상 정의.omao/plans/observability/baselines/${metric}.yaml — 메트릭별 베이스라인 (6시간마다 갱신).omao/state/anomaly/${timestamp}-${metric}.json — 탐지된 anomaly 이벤트.omao/state/anomaly/correlation-matrix.json — 메트릭 간 상관관계 매트릭스 (일간 갱신).omao/state/anomaly/summary.json — 최근 24시간 탐지 요약Input: /anomaly-detection rag-qa-agent
Output (anomaly detected):
[14:05:00Z] Loading monitoring targets for: rag-qa-agent
[14:05:01Z] Metrics to check: 4 (Error Rate, P99 Latency, Bedrock Latency, Token Usage)
[14:05:01Z] Exclusion check: not in exclusion window
[14:05:02Z] Fetching recent data (5min window)...
[14:05:03Z] Baseline status:
Error Rate : loaded (age: 2h, mean=0.012, σ=0.004)
P99 Latency : loaded (age: 2h, mean=380ms, σ=45ms)
Bedrock Latency : loaded (age: 5h, mean=220ms, σ=30ms)
Token Usage : refreshing (age: 7h)...done
[14:05:05Z] Detection results:
Error Rate : 0.031 (4.8σ) → WARNING
P99 Latency : 890ms (11.3σ) → CRITICAL
Bedrock Latency : 245ms (0.8σ) → normal
Token Usage : 1250 (1.2σ) → normal
[14:05:05Z] Correlation check:
Error Rate:P99 Latency — expected r=0.7, actual r=0.92 (Δ=0.22) → normal
Token Usage:P99 Latency — expected r=0.5, actual r=-0.1 (Δ=0.60) → CRITICAL
[14:05:06Z] Deploy context: canary v2.3.1 deployed 8min ago → CHANGE-CORRELATED
[14:05:06Z] Summary: 2 critical, 1 warning, 0 info
[14:05:06Z] Action: Triggering incident-response SEV2
Source: anomaly-detection
Evidence: P99 Latency 11.3σ + correlation breakdown (Token:Latency)
Context: Correlated with deploy rag-qa-agent:v2.3.1 (8min ago)
[14:05:06Z] Events saved to .omao/state/anomaly/
Output (all normal):
[14:05:00Z] Loading monitoring targets for: rag-qa-agent
[14:05:03Z] Detection results:
Error Rate : 0.013 (0.3σ) → normal
P99 Latency : 395ms (0.3σ) → normal
Bedrock Latency : 218ms (0.1σ) → normal
Token Usage : 980 (0.5σ) → normal
[14:05:04Z] Correlation check: all within expected range
[14:05:04Z] Summary: 0 critical, 0 warning, 0 info
[14:05:04Z] No anomalies detected. Next check in 5 minutes.
Output (exclusion window):
[03:05:00Z] Loading monitoring targets for: rag-qa-agent
[03:05:01Z] Exclusion check: IN EXCLUSION WINDOW
Reason: "Daily reindex batch job" (03:00-03:30 UTC)
[03:05:01Z] Skipping detection cycle. Next check at 03:35 UTC.
| 연동 대상 | 방향 | 설명 |
|---|---|---|
incident-response | → 출력 | Critical/Warning anomaly를 알람으로 전달 |
autopilot-deploy | ← 입력 | 배포 이벤트 시점을 anomaly 컨텍스트로 수신 |
continuous-eval | ← 입력 | 품질 메트릭 regression을 anomaly 소스로 활용 |
cost-governance | → 출력 | 비용 메트릭 이상(Token Usage 급증)을 cost alert로 전달 |
root-cause-analysis | → 출력 | Critical anomaly 발생 시 RCA 트리거 |