菜单
인시던트 발생 시 관련 메트릭·로그·이벤트·변경 이력을 자동 수집하고 인과 관계를 추론하여 근본 원인을 식별한다. 타임라인 기반 이벤트 상관관계 분석, 변경-장애 매핑, 의존성 그래프 탐색을 수행하며 RCA 보고서를 자동 생성한다.
默认使用会先检查来源的 Prompt；你也可以切换为直接命令，或下载本地副本。
决定是否安装前，请先阅读 SKILL.md，以及 SkillsMP 当前展示的配套文件。
用 Codex 或 Claude 帮你安装 复制这段 Prompt，粘贴到 Codex、Claude 或其他助手里，让它检查 Skill 页面并帮你完成安装。
直接命令不会经过审查 Prompt；运行前请先检查来源。
npx skills add https://github.com/aws-samples/sample-oh-my-aidlcops --skill root-cause-analysis命令会保持在同一行。复制前请横向滚动并检查完整内容。
想先保存到本地？可下载 SkillsMP 当前能够提供的文件。
正在显示 SKILL.md
基于 SOC 职业分类
| name | root-cause-analysis |
| description | 인시던트 발생 시 관련 메트릭·로그·이벤트·변경 이력을 자동 수집하고 인과 관계를 추론하여 근본 원인을 식별한다. 타임라인 기반 이벤트 상관관계 분석, 변경-장애 매핑, 의존성 그래프 탐색을 수행하며 RCA 보고서를 자동 생성한다. |
| argument-hint | [incident-id or symptom description] |
| user-invocable | true |
| model | claude-opus-4-7 |
| allowed-tools | Read,Grep,Bash,mcp__cloudwatch,mcp__prometheus,mcp__eks |
| ontology | {"consumes":["Incident"],"references":["Deployment","Agent"]} |
| loop-interface | {"accepts-from":["anomaly-detection.anomaly"],"emits-to":["automated-remediation.rca"],"emits-signals":[]} |
incident-response가 SEV1~3 인시던트를 분류한 직후, 근본 원인 분석이 필요할 때anomaly-detection이 Critical anomaly를 탐지하고 원인 추적이 필요할 때사용 제외:
incident-response의 Step 3~4 (Hypothesis + Diagnostic)로 이미 원인이 확정된 경우@latest 금지, PyPI 버전 pin 필수).incident-response가 생성한 인시던트 상태 파일 (.omao/state/incident/)..omao/plans/observability/dependency-map.yaml..omao/plans/observability/rca-patterns.yaml (학습 결과 축적).# .omao/plans/observability/dependency-map.yaml
services:
rag-qa-agent:
type: agent
namespace: production
dependencies:
- name: milvus
type: vector-db
protocol: grpc
port: 19530
health_check: "milvus_health_status"
- name: bedrock
type: llm-provider
protocol: https
endpoint:
인시던트 시점 ±30분의 데이터를 다차원으로 자동 수집합니다.
import json
import os
from datetime import datetime, timedelta
from dataclasses import dataclass, field
@dataclass
class EvidenceWindow:
incident_time: datetime
start: datetime
end: datetime
@classmethod
def from_incident(cls, incident_time: str, buffer_minutes: int = 30):
t = datetime.fromisoformat(incident_time.replace("Z", "+00:00"))
return cls(
incident_time=t,
start=t - timedelta(minutes=buffer_minutes),
end=t + timedelta(minutes=buffer_minutes),
)
@dataclass
class Evidence:
source: str
timestamp: str
description: str
severity: str = "info"
raw_data: dict = field(default_factory=dict)
def collect_evidence(incident_id: str, window: EvidenceWindow) -> dict[str, list[Evidence]]:
"""인시던트 관련 증거를 다차원으로 수집.
수집 소스:
1. CloudWatch Metrics — 관련 메트릭 시계열
2. CloudWatch Logs — 에러 로그
3. EKS Events — Pod/Node 이벤트
4. Git/ArgoCD — 변경 이력
5. CloudTrail — AWS API 호출 이력
"""
evidence = {
"metrics": collect_metrics_evidence(window),
"logs": collect_log_evidence(window),
"k8s_events": collect_k8s_events(window),
"changes": collect_change_history(window),
"cloudtrail": collect_cloudtrail_events(window),
}
# 증거 저장
evidence_dir = f".omao/state/incident/{incident_id}/evidence"
os.makedirs(evidence_dir, exist_ok=True)
for source, items in evidence.items():
with open(os.path.join(evidence_dir, f"{source}.json"), "w") as f:
json.dump([vars(e) for e in items], f, indent=2, default=str)
return evidence
def collect_metrics_evidence(window: EvidenceWindow) -> list[Evidence]:
"""CloudWatch + Prometheus 메트릭에서 이상 구간 추출."""
evidence = []
# Prometheus: 에러율 급증 구간
# mcp__prometheus__query_range(
# query='rate(agent_request_errors_total[5m])',
# start=window.start.timestamp(),
# end=window.end.timestamp(),
# step=60
# )
# CloudWatch: 관련 알람 상태 변경
# mcp__cloudwatch__describe_alarm_history(
# alarm_name=...,
# start_date=window.start.isoformat(),
# end_date=window.end.isoformat()
# )
return evidence
def collect_log_evidence(window: EvidenceWindow) -> list[Evidence]:
"""CloudWatch Logs에서 에러/경고 로그 추출."""
evidence = []
# mcp__cloudwatch__filter_log_events(
# log_group='/aws/eks/production/containers',
# start_time=int(window.start.timestamp() * 1000),
# end_time=int(window.end.timestamp() * 1000),
# filter_pattern='?ERROR ?FATAL ?Exception ?Traceback'
# )
return evidence
def collect_k8s_events(window: EvidenceWindow) -> list[Evidence]:
"""EKS 클러스터 이벤트 수집 (Pod restart, OOM, scheduling 실패 등)."""
evidence = []
# mcp__eks__get_k8s_events(
# cluster_name='production',
# kind='Pod',
# namespace='production'
# )
return evidence
def collect_change_history(window: EvidenceWindow) -> list[Evidence]:
"""변경 이력 수집 (Git, ArgoCD, CloudTrail)."""
evidence = []
# Git: 최근 커밋/배포
# git log --after=window.start --before=window.end --oneline
# ArgoCD: sync history
# kubectl get applications -n argocd -o json
return evidence
def collect_cloudtrail_events(window: EvidenceWindow) -> list[Evidence]:
"""CloudTrail에서 인프라 변경 이벤트 수집."""
evidence = []
# aws cloudtrail lookup-events \
# --start-time window.start \
# --end-time window.end \
# --lookup-attributes AttributeKey=EventSource,AttributeValue=eks.amazonaws.com
return evidence
수집된 이벤트를 시간순으로 정렬하여 인과 관계 후보를 식별합니다.
@dataclass
class TimelineEvent:
timestamp: datetime
source: str
description: str
severity: str
is_change: bool = False
is_symptom: bool = False
def build_timeline(evidence: dict[str, list[Evidence]],
incident_time: datetime) -> list[TimelineEvent]:
"""모든 증거를 단일 타임라인으로 병합하고 인과 관계 후보를 마킹."""
events = []
for source, items in evidence.items():
for item in items:
ts = datetime.fromisoformat(item.timestamp.replace("Z", "+00:00"))
event = TimelineEvent(
timestamp=ts,
source=source,
description=item.description,
severity=item.severity,
is_change=(source in ("changes", "cloudtrail")),
is_symptom=(item.severity in ("critical", "warning") and ts >= incident_time),
)
events.append(event)
# 시간순 정렬
events.sort(key=lambda e: e.timestamp)
return events
def identify_causal_candidates(timeline: list[TimelineEvent],
incident_time: datetime) -> list[dict]:
"""타임라인에서 인과 관계 후보 식별.
규칙:
1. 장애 시점 이전의 변경 이벤트 → 원인 후보
2. 시간적 근접도가 높을수록 상관 점수 높음
3. 동일 소스의 연속 이벤트는 그룹핑
"""
candidates = []
for event in timeline:
if not event.is_change:
continue
if event.timestamp >= incident_time:
continue
time_delta_sec = (incident_time - event.timestamp).total_seconds()
# 시간 근접도 점수 (1시간 이내 = 높음, 24시간 이내 = 중간)
if time_delta_sec < 3600:
proximity_score = 0.9
elif time_delta_sec < 14400: # 4시간
proximity_score = 0.6
elif time_delta_sec < 86400: # 24시간
proximity_score = 0.3
else:
proximity_score = 0.1
candidates.append({
"event": event,
"time_before_incident_sec": time_delta_sec,
"proximity_score": proximity_score,
"source": event.source,
})
return sorted(candidates, key=lambda x: -x["proximity_score"])
장애 시점 이전 변경 이력과의 상관관계를 심층 분석합니다.
@dataclass
class CorrelationResult:
change_event: TimelineEvent
correlation_score: float # 0.0 ~ 1.0
evidence_chain: list[str]
confidence: str # high, medium, low
def correlate_changes(candidates: list[dict],
symptoms: list[TimelineEvent],
dependency_map: dict) -> list[CorrelationResult]:
"""변경 이벤트와 증상 간 상관관계 분석.
상관 점수 계산:
- 시간 근접도 (40%)
- 영향 범위 일치 (30%) — 변경 대상과 장애 서비스의 의존성 관계
- 유사 패턴 이력 (30%) — 과거 RCA에서 동일 변경→장애 패턴 존재 여부
"""
results = []
for candidate in candidates:
event = candidate["event"]
# 영향 범위 일치 점수
scope_score = calculate_scope_match(event, symptoms, dependency_map)
# 유사 패턴 이력 점수
pattern_score = check_historical_patterns(event, symptoms)
# 종합 점수
total_score = (
candidate["proximity_score"] * 0.4 +
scope_score * 0.3 +
pattern_score * 0.3
)
# 증거 체인 구성
evidence_chain = build_evidence_chain(event, symptoms)
# 신뢰도 분류
if total_score >= 0.7:
confidence = "high"
elif total_score >= 0.4:
confidence = "medium"
else:
confidence = "low"
results.append(CorrelationResult(
change_event=event,
correlation_score=total_score,
evidence_chain=evidence_chain,
confidence=confidence,
))
return sorted(results, key=lambda r: -r.correlation_score)
def calculate_scope_match(change: TimelineEvent, symptoms: list[TimelineEvent],
dependency_map: dict) -> float:
"""변경 대상과 장애 서비스 간 의존성 관계 점수."""
# 직접 의존: 1.0, 간접 의존(1-hop): 0.6, 무관: 0.1
return 0.5 # placeholder
def check_historical_patterns(change: TimelineEvent,
symptoms: list[TimelineEvent]) -> float:
"""과거 RCA 패턴 DB에서 유사 패턴 검색."""
patterns_file = ".omao/plans/observability/rca-patterns.yaml"
if not os.path.exists(patterns_file):
return 0.0
# 패턴 매칭 로직
return 0.0 # 초기에는 패턴 없음
def build_evidence_chain(change: TimelineEvent,
symptoms: list[TimelineEvent]) -> list[str]:
"""변경 → 증상 간 인과 관계 체인 구성."""
chain = [
f"[{change.timestamp.isoformat()}Z] Change: {change.description}",
]
for symptom in symptoms[:3]: # 상위 3개 증상
chain.append(f"[{symptom.timestamp.isoformat()}Z] Symptom: {symptom.description}")
return chain
장애 서비스의 upstream/downstream 의존성을 탐색하여 전파 경로를 추적합니다.
import yaml
def traverse_dependencies(affected_service: str,
dependency_map_file: str,
direction: str = "both") -> dict:
"""의존성 그래프를 탐색하여 장애 전파 경로 추적.
Args:
affected_service: 장애가 발생한 서비스명
dependency_map_file: 의존성 맵 파일 경로
direction: upstream, downstream, both
Returns:
전파 경로 및 영향 범위
"""
with open(dependency_map_file) as f:
dep_map = yaml.safe_load(f)
services = dep_map.get("services", {})
# Upstream: 이 서비스가 의존하는 서비스들
upstream = []
if direction in ("upstream", "both"):
svc = services.get(affected_service, {})
for dep in svc.get("dependencies", []):
upstream.append({
"name": dep["name"],
"type": dep["type"],
"health_check": dep.get("health_check"),
})
# Downstream: 이 서비스에 의존하는 서비스들
downstream = []
if direction in ("downstream", "both"):
for svc_name, svc_config in services.items():
if svc_name == affected_service:
continue
for dep in svc_config.get("dependencies", []):
if dep["name"] == affected_service:
downstream.append({
"name": svc_name,
"type": svc_config.get("type"),
})
return {
"affected_service": affected_service,
"upstream_dependencies": upstream,
"downstream_impact": downstream,
"total_blast_radius": 1 + len(downstream),
}
분석 결과를 구조화된 보고서로 생성합니다.
@dataclass
class RCAReport:
incident_id: str
root_cause: str
confidence: str
category: str
timeline: list[TimelineEvent]
evidence_chain: list[str]
contributing_factors: list[str]
recommendations: list[dict]
blast_radius: dict
def generate_rca_report(incident_id: str,
correlations: list[CorrelationResult],
timeline: list[TimelineEvent],
dependency_result: dict) -> str:
"""RCA 보고서를 마크다운으로 생성."""
top_cause = correlations[0] if correlations else None
report = f"""# RCA Report — {incident_id}
## Summary
- **Root Cause**: {top_cause.change_event.description if top_cause else "Undetermined"}
- **Confidence**: {top_cause.confidence if top_cause else "low"}
- **Category**: {categorize_root_cause(top_cause)}
- **Correlation Score**: {top_cause.correlation_score:.2f if top_cause else "N/A"}
## Timeline
| Timestamp | Source | Event | Type |
|-----------|--------|-------|------|
"""
for event in timeline[:20]: # 상위 20개 이벤트
event_type = "🔄 Change" if event.is_change else ("⚠️ Symptom" if event.is_symptom else "ℹ️ Info")
report += f"| {event.timestamp.strftime('%Y-%m-%d %H:%M:%S')}Z | {event.source} | {event.description} | {event_type} |\n"
report += f"""
## Evidence Chain
"""
if top_cause:
for i, step in enumerate(top_cause.evidence_chain, 1):
report += f"{i}. {step}\n"
report += f"""
## Blast Radius
- **Affected Service**: {dependency_result['affected_service']}
- **Upstream Dependencies**: {len(dependency_result['upstream_dependencies'])}
- **Downstream Impact**: {len(dependency_result['downstream_impact'])} services
"""
for ds in dependency_result["downstream_impact"]:
report += f" - `{ds['name']}` ({ds['type']})\n"
report += f"""
## Contributing Factors
"""
for corr in correlations[1:4]: # 2~4위 후보
report += f"- [{corr.confidence}] {corr.change_event.description} (score: {corr.correlation_score:.2f})\n"
report += f"""
## Recommendations
- [ ] **단기 조치**: {"Rollback " + top_cause.change_event.description if top_cause else "Manual investigation required"}
- [ ] **중기 조치**: Add monitoring for similar change patterns
- [ ] **장기 조치**: Update dependency-map and add pre-deploy validation
"""
# 보고서 저장
report_path = f".omao/state/incident/{incident_id}/rca-report.md"
os.makedirs(os.path.dirname(report_path), exist_ok=True)
with open(report_path, "w") as f:
f.write(report)
return report
def categorize_root_cause(correlation: CorrelationResult | None) -> str:
"""근본 원인 카테고리 분류."""
if not correlation:
return "unknown"
source = correlation.change_event.source
desc = correlation.change_event.description.lower()
if source == "changes" and ("deploy" in desc or "release" in desc):
return "deployment"
elif source == "cloudtrail" and ("security" in desc or "iam" in desc):
return "configuration"
elif source == "k8s_events" and ("node" in desc or "resource" in desc):
return "infrastructure"
elif "dependency" in desc or "upstream" in desc:
return "dependency"
else:
return "other"
RCA 결과를 패턴 DB에 축적하여 향후 분석 정확도를 높입니다.
def update_rca_patterns(incident_id: str, rca_report: RCAReport):
"""RCA 결과를 패턴 DB에 축적."""
patterns_file = ".omao/plans/observability/rca-patterns.yaml"
if os.path.exists(patterns_file):
with open(patterns_file) as f:
patterns = yaml.safe_load(f) or {"patterns": []}
else:
patterns = {"patterns": []}
new_pattern = {
"id": f"pattern-{len(patterns['patterns']) + 1}",
"incident_id": incident_id,
"root_cause_category": rca_report.category,
"change_signature": rca_report.root_cause,
"symptoms": [e.description for e in rca_report.timeline if e.is_symptom][:5],
"confidence": rca_report.confidence,
"created_at": datetime.utcnow().isoformat() + "Z",
}
patterns["patterns"].append(new_pattern)
with open(patterns_file, "w") as f:
yaml.dump(patterns, f, default_flow_style=False, allow_unicode=True)
.omao/state/incident/${id}/rca-report.md — RCA 보고서.omao/state/incident/${id}/evidence/ — 수집된 증거 raw 데이터
metrics.json — 메트릭 증거logs.json — 로그 증거k8s_events.json — K8s 이벤트changes.json — 변경 이력cloudtrail.json — CloudTrail 이벤트.omao/state/incident/${id}/timeline.jsonl — 재구성된 타임라인.omao/plans/observability/rca-patterns.yaml — 반복 패턴 학습 결과.omao/plans/observability/dependency-map.yaml — 서비스 의존성 맵Input: /root-cause-analysis sev2-20260508-1405
Output:
[14:06:00Z] Starting RCA for incident: sev2-20260508-1405
[14:06:00Z] Incident time: 2026-05-08T14:05:00Z
[14:06:00Z] Evidence window: 13:35:00Z — 14:35:00Z (±30min)
[14:06:01Z] Phase 1: Evidence Collection
Metrics : 4 time series collected (Error Rate, P99 Latency, Bedrock Latency, Token Usage)
Logs : 23 error entries found
K8s Events : 5 events (2 Warning, 3 Normal)
Changes : 3 events (1 deploy, 1 ConfigMap update, 1 HPA change)
CloudTrail : 2 events (ModifyTargetGroup, UpdateService)
[14:06:03Z] Phase 2: Timeline Reconstruction
Total events: 37
Change events (pre-incident): 3
Symptom events: 8
[14:06:04Z] Phase 3: Change-Incident Correlation
Candidate 1: [13:52Z] Deploy rag-qa-agent:v2.3.1 → score=0.82 (HIGH)
- Proximity: 13min before incident (0.9)
- Scope: direct match — same service (1.0)
- Pattern: similar deploy→latency pattern seen in pattern-3 (0.6)
Candidate 2: [13:45Z] ConfigMap update: prompt-templates → score=0.54 (MEDIUM)
Candidate 3: [12:30Z] HPA maxReplicas 10→8 → score=0.31 (LOW)
[14:06:05Z] Phase 4: Dependency Graph Traversal
Affected: rag-qa-agent
Upstream: milvus (healthy), bedrock (healthy), redis (healthy)
Downstream: 2 services impacted (api-gateway, slack-bot)
[14:06:06Z] Phase 5: RCA Report Generated
Root Cause: Deploy rag-qa-agent:v2.3.1 (confidence: HIGH)
Category: deployment
Blast Radius: 3 services
[14:06:06Z] Phase 6: Pattern DB Updated (pattern-7)
[14:06:06Z] Report saved: .omao/state/incident/sev2-20260508-1405/rca-report.md
[14:06:06Z] Next action: automated-remediation (runbook match available: rollback-canary)
Input: /root-cause-analysis sev3-20260508-0300 --postmortem
Output (undetermined):
[03:15:00Z] Starting RCA for incident: sev3-20260508-0300
[03:15:05Z] Phase 3: Change-Incident Correlation
No change events found in ±30min window.
Expanding window to ±2 hours...
Candidate 1: [01:15Z] AWS maintenance event (ap-northeast-2a) → score=0.45 (MEDIUM)
[03:15:07Z] Root Cause: [UNDETERMINED — medium confidence]
Best candidate: AWS maintenance event (score=0.45)
Recommendation: Check AWS Health Dashboard, expand evidence window to ±6h
[03:15:07Z] Report saved with status: NEEDS_HUMAN_REVIEW
| 연동 대상 | 방향 | 설명 |
|---|---|---|
incident-response | ← 입력 | 인시던트 분류 결과 및 가설을 입력으로 수신 |
anomaly-detection | ← 입력 | Critical anomaly 이벤트를 RCA 트리거로 수신 |
automated-remediation | → 출력 | RCA 완료 + 알려진 패턴 → remediation 트리거 |
self-improving-loop | → 출력 | RCA 결과가 "prompt 품질"이면 개선 루프 트리거 |
audit-trail | → 출력 | RCA 과정 전체를 감사 로그에 기록 |
slo-management | ← 입력 | SLO 위반 인시던트의 RCA 요청 수신 |