107 lines
3.5 KiB
Python
107 lines
3.5 KiB
Python
"""
|
|
Failure Autopsy — post-hoc root cause analysis for a failed action.
|
|
|
|
Given a journal (TimelineStore) and an action name, produces:
|
|
- the failure event itself
|
|
- the backward dependency trace (recursive walk)
|
|
- the last scheduler decision for that action
|
|
- whether the action is on the critical path
|
|
"""
|
|
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
|
|
class FailureAutopsy:
|
|
def __init__(self, journal, critical_path: Optional[Dict[str, Any]] = None):
|
|
"""
|
|
journal: TimelineStore (or anything with `trace_action` + `find_failures`)
|
|
critical_path: optional dict with at least `score_map` (set of action names)
|
|
"""
|
|
self.journal = journal
|
|
self.critical_path = critical_path or {}
|
|
|
|
# -----------------------------
|
|
# FIND FAILURE EVENTS
|
|
# -----------------------------
|
|
def find_failures(self) -> List[Dict[str, Any]]:
|
|
return [
|
|
e for e in self.journal.find_failures()
|
|
if (e.get("data", {}).get("state") == "failed" or e.get("state") == "failed")
|
|
]
|
|
|
|
# -----------------------------
|
|
# TRACE BACKWARD DEPENDENCY CHAIN
|
|
# -----------------------------
|
|
def trace_dependencies(self, action_name: str) -> List[Dict[str, Any]]:
|
|
trace: List[Dict[str, Any]] = []
|
|
visited: set = set()
|
|
|
|
def walk(name: str):
|
|
if name in visited:
|
|
return
|
|
visited.add(name)
|
|
|
|
events = self.journal.trace_action(name)
|
|
trace.append({"action": name, "events": events})
|
|
|
|
for e in events:
|
|
d = e.get("data", e)
|
|
deps = d.get("deps", [])
|
|
for dep in deps:
|
|
walk(dep)
|
|
|
|
walk(action_name)
|
|
return trace
|
|
|
|
# -----------------------------
|
|
# GET LAST SCHEDULER DECISION
|
|
# -----------------------------
|
|
def last_decision(self, action_name: str) -> Optional[Dict[str, Any]]:
|
|
events = self.journal.trace_action(action_name)
|
|
for e in reversed(events):
|
|
if e.get("type") == "schedule_decision":
|
|
return e.get("data", e)
|
|
# Fallback: return last task_update with a node assignment
|
|
for e in reversed(events):
|
|
d = e.get("data", e)
|
|
if d.get("node"):
|
|
return {
|
|
"node": d.get("node"),
|
|
"score": d.get("score"),
|
|
"reason": "scheduler_assignment",
|
|
}
|
|
return None
|
|
|
|
# -----------------------------
|
|
# FULL AUTOPSY REPORT
|
|
# -----------------------------
|
|
def report(self, action_name: str) -> Dict[str, Any]:
|
|
failures = self.find_failures()
|
|
|
|
target_failure: Optional[Dict[str, Any]] = None
|
|
for f in failures:
|
|
d = f.get("data", f)
|
|
if d.get("action") == action_name:
|
|
target_failure = f
|
|
break
|
|
|
|
if not target_failure:
|
|
return {
|
|
"status": "no_failure_found",
|
|
"action": action_name,
|
|
}
|
|
|
|
deps_trace = self.trace_dependencies(action_name)
|
|
last_sched = self.last_decision(action_name)
|
|
|
|
score_map = self.critical_path.get("score_map", {}) if isinstance(self.critical_path, dict) else {}
|
|
|
|
return {
|
|
"status": "failure_detected",
|
|
"action": action_name,
|
|
"failure_event": target_failure,
|
|
"last_scheduler_decision": last_sched,
|
|
"dependency_trace": deps_trace,
|
|
"on_critical_path": action_name in score_map,
|
|
}
|