メニュー
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コマンドは1行のまま表示されます。コピー前に横へスクロールして全体を確認してください。
ローカルで確認しますか？SkillsMP が現在取得できるファイルをダウンロードできます。
알려진 장애 패턴에 대해 사전 정의된 Runbook 기반 자동 복구를 실행한다. RCA 결과 또는 incident-response의 가설을 입력으로 받아 매칭되는 remediation playbook을 선택하고, 복구 전/후 상태를 검증하며, 실패 시 에스컬레이션한다. SEV2/3만 자동 복구 대상이며 SEV1은 사람 전용이다.
Agent 또는 skill의 프로덕션 배포를 canary 1% → 10% → 50% → 100% 4단계로 자동 진행한다. 각 단계는 continuous-eval 통과와 SLO 준수를 gate로 하며 circuit breaker가 SLO 위반을 감지하면 즉시 이전 단계로 자동 롤백한다. 100% 승격은 사람 승인이 필수다.
Ragas 기반 품질·안전 평가를 매 배포 직후와 매 1시간 cron으로 실행하여 faithfulness, answer relevance, context precision, toxicity, PII leakage 지표를 측정한다. Golden dataset 관리와 regression gate(baseline 대비 5%p 하락 시 차단)를 수행하며 autopilot-deploy의 canary gate에 사용된다.
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 트리거 |