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ê.
Um comando direto ignora o prompt de revisão. Verifique a origem antes de executá-lo.
@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),
}
