Standardmäßig ist der Prompt ausgewählt, der zuerst die Quelle prüft. Sie können zu einem direkten Befehl wechseln oder eine lokale Kopie herunterladen.
Quelldateien prüfen
Lesen Sie SKILL.md und alle von SkillsMP angezeigten Begleitdateien, bevor Sie sich für eine Installation entscheiden.
Mit Codex oder Claude installieren Kopieren Sie diesen Prompt, fügen Sie ihn in Codex, Claude oder einen anderen Assistant ein und lassen Sie die Skill-Seite prüfen und installieren.
Ein direkter Befehl überspringt den Prüf-Prompt. Prüfen Sie die Quelle, bevor Sie ihn ausführen.
@dataclassclassTimelineEvent:
timestamp: datetime
source: str
description: str
severity: str
is_change: bool = False
is_symptom: bool = Falsedefbuild_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
defidentify_causal_candidates(timeline: list[TimelineEvent],
incident_time: datetime) -> list[dict]:
"""타임라인에서 인과 관계 후보 식별.
규칙:
1. 장애 시점 이전의 변경 이벤트 → 원인 후보
2. 시간적 근접도가 높을수록 상관 점수 높음
3. 동일 소스의 연속 이벤트는 그룹핑
"""
candidates = []
for event in timeline:
ifnot event.is_change:
continueif event.timestamp >= incident_time:
continue
time_delta_sec = (incident_time - event.timestamp).total_seconds()
# 시간 근접도 점수 (1시간 이내 = 높음, 24시간 이내 = 중간)if time_delta_sec < 3600:
proximity_score = 0.9elif time_delta_sec < 14400: # 4시간
proximity_score = 0.6elif time_delta_sec < 86400: # 24시간
proximity_score = 0.3else:
proximity_score = 0.1
candidates.append({
"event": event,
"time_before_incident_sec": time_delta_sec,
"proximity_score": proximity_score,
"source": event.source,
})
returnsorted(candidates, key=lambda x: -x["proximity_score"])
Phase 3: Change-Incident Correlation
장애 시점 이전 변경 이력과의 상관관계를 심층 분석합니다.
@dataclassclassCorrelationResult:
change_event: TimelineEvent
correlation_score: float# 0.0 ~ 1.0
evidence_chain: list[str]
confidence: str# high, medium, lowdefcorrelate_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,
))
returnsorted(results, key=lambda r: -r.correlation_score)
defcalculate_scope_match(change: TimelineEvent, symptoms: list[TimelineEvent],
dependency_map: dict) -> float:
"""변경 대상과 장애 서비스 간 의존성 관계 점수."""# 직접 의존: 1.0, 간접 의존(1-hop): 0.6, 무관: 0.1return0.5# placeholderdefcheck_historical_patterns(change: TimelineEvent,
symptoms: list[TimelineEvent]) -> float:
"""과거 RCA 패턴 DB에서 유사 패턴 검색."""
patterns_file = ".omao/plans/observability/rca-patterns.yaml"ifnot os.path.exists(patterns_file):
return0.0# 패턴 매칭 로직return0.0# 초기에는 패턴 없음defbuild_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
Phase 4: Dependency Graph Traversal
장애 서비스의 upstream/downstream 의존성을 탐색하여 전파 경로를 추적합니다.
import yaml
deftraverse_dependencies(affected_service: str,
dependency_map_file: str,
direction: str = "both") -> dict:
"""의존성 그래프를 탐색하여 장애 전파 경로 추적.
Args:
affected_service: 장애가 발생한 서비스명
dependency_map_file: 의존성 맵 파일 경로
direction: upstream, downstream, both
Returns:
전파 경로 및 영향 범위
"""withopen(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:
continuefor 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),
}
