200 lines
6.7 KiB
Python
200 lines
6.7 KiB
Python
"""
|
|
Timeline Store — immutable event log indexed by node + by absolute index.
|
|
|
|
Now backed by SQLite (via backend.storage.sqlite_db) with an in-memory
|
|
cache for fast recent reads. Falls back to pure in-memory if storage
|
|
is unavailable.
|
|
|
|
Used by:
|
|
- backend/api/timeline.py (REST endpoints)
|
|
- backend/analysis/failure_autopsy.py (dependency tracing)
|
|
- replay UI (via /replay/events)
|
|
"""
|
|
|
|
from collections import defaultdict
|
|
from typing import Any, Dict, List, Optional
|
|
import threading
|
|
import time
|
|
|
|
# Optional storage backend
|
|
try:
|
|
from backend.storage.sqlite_db import STORAGE
|
|
except Exception:
|
|
STORAGE = None
|
|
|
|
|
|
class TimelineStore:
|
|
"""
|
|
Append-only event journal with:
|
|
- per-node indices for fast node lookups
|
|
- per-action indices for fast action tracing
|
|
- snapshot-at-index for replay rewinding
|
|
- SQLite persistence (with in-memory cache for recent events)
|
|
"""
|
|
|
|
def __init__(self):
|
|
self._events: List[Dict[str, Any]] = []
|
|
self._by_node: Dict[str, List[int]] = defaultdict(list)
|
|
self._by_action: Dict[str, List[int]] = defaultdict(list)
|
|
self._lock = threading.Lock()
|
|
|
|
# On startup, load recent events from storage
|
|
if STORAGE:
|
|
self._load_from_storage()
|
|
|
|
def _load_from_storage(self, limit: int = 10000):
|
|
"""Load recent events from SQLite into the in-memory cache."""
|
|
try:
|
|
events = STORAGE.get_events(limit=limit)
|
|
for e in events:
|
|
self._index_event(e)
|
|
except Exception:
|
|
pass
|
|
|
|
def _index_event(self, event: Dict[str, Any]):
|
|
"""Add an event to in-memory indices (no lock — caller must hold it)."""
|
|
idx = len(self._events)
|
|
self._events.append(event)
|
|
node = event.get("node") or (event.get("data", {}) or {}).get("node")
|
|
if node:
|
|
self._by_node[node].append(idx)
|
|
action = event.get("action") or (event.get("data", {}) or {}).get("action")
|
|
if action:
|
|
self._by_action[action].append(idx)
|
|
|
|
# -----------------------------
|
|
# APPEND
|
|
# -----------------------------
|
|
def append(self, event: Dict[str, Any]) -> int:
|
|
"""Append an event to the journal. Returns the new index."""
|
|
with self._lock:
|
|
if "timestamp" not in event:
|
|
event["timestamp"] = time.time()
|
|
self._index_event(event)
|
|
|
|
# Persist to SQLite (outside the lock)
|
|
if STORAGE:
|
|
try:
|
|
STORAGE.append_event(event)
|
|
except Exception:
|
|
pass
|
|
|
|
with self._lock:
|
|
return len(self._events) - 1
|
|
|
|
# -----------------------------
|
|
# READ
|
|
# -----------------------------
|
|
def all(self) -> List[Dict[str, Any]]:
|
|
with self._lock:
|
|
return list(self._events)
|
|
|
|
def get_node_events(self, node: str) -> List[Dict[str, Any]]:
|
|
# Try in-memory first
|
|
with self._lock:
|
|
if node in self._by_node:
|
|
return [self._events[i] for i in self._by_node[node] if i < len(self._events)]
|
|
# Fall back to storage (queries older events not in cache)
|
|
if STORAGE:
|
|
try:
|
|
return STORAGE.get_node_events(node)
|
|
except Exception:
|
|
pass
|
|
return []
|
|
|
|
def get_action_events(self, action: str) -> List[Dict[str, Any]]:
|
|
with self._lock:
|
|
if action in self._by_action:
|
|
return [self._events[i] for i in self._by_action[action] if i < len(self._events)]
|
|
if STORAGE:
|
|
try:
|
|
return STORAGE.get_action_events(action)
|
|
except Exception:
|
|
pass
|
|
return []
|
|
|
|
def trace_action(self, action: str) -> List[Dict[str, Any]]:
|
|
"""Alias for get_action_events — kept for autopsy compatibility."""
|
|
return self.get_action_events(action)
|
|
|
|
def find_failures(self) -> List[Dict[str, Any]]:
|
|
# In-memory
|
|
with self._lock:
|
|
in_mem = [
|
|
e for e in self._events
|
|
if e.get("type") in ("failure", "task_update")
|
|
and (e.get("state") == "failed" or e.get("data", {}).get("state") == "failed")
|
|
]
|
|
# Storage (may have more)
|
|
if STORAGE:
|
|
try:
|
|
stored = STORAGE.find_failures()
|
|
# Merge + dedupe by timestamp+action
|
|
seen = {(e.get("timestamp"), e.get("action")) for e in in_mem}
|
|
for e in stored:
|
|
key = (e.get("timestamp"), e.get("action"))
|
|
if key not in seen:
|
|
in_mem.append(e)
|
|
seen.add(key)
|
|
except Exception:
|
|
pass
|
|
return in_mem
|
|
|
|
# -----------------------------
|
|
# REWIND / SNAPSHOT
|
|
# -----------------------------
|
|
def snapshot_at(self, index: int) -> Dict[str, Any]:
|
|
with self._lock:
|
|
safe = max(0, min(index, len(self._events) - 1))
|
|
events = self._events[:safe + 1]
|
|
|
|
# rebuild node + action views
|
|
nodes: Dict[str, Dict[str, Any]] = {}
|
|
actions: Dict[str, Dict[str, Any]] = {}
|
|
for e in events:
|
|
d = e.get("data", e)
|
|
n = d.get("node") or e.get("node")
|
|
a = d.get("action") or e.get("action")
|
|
s = d.get("state") or e.get("state")
|
|
if n:
|
|
nodes[n] = {"name": n, "state": s or "online", "last_event": e.get("timestamp")}
|
|
if a:
|
|
actions[a] = {"name": a, "state": s or "pending", "node": n, "last_event": e.get("timestamp")}
|
|
|
|
return {
|
|
"index": safe,
|
|
"event_count": len(events),
|
|
"nodes": list(nodes.values()),
|
|
"actions": list(actions.values()),
|
|
}
|
|
|
|
# -----------------------------
|
|
# STATS
|
|
# -----------------------------
|
|
def stats(self) -> Dict[str, Any]:
|
|
with self._lock:
|
|
total = len(self._events)
|
|
failures = len([e for e in self._events if e.get("type") == "failure"])
|
|
# If we have storage, get the authoritative count
|
|
if STORAGE:
|
|
try:
|
|
total = STORAGE.event_count()
|
|
except Exception:
|
|
pass
|
|
return {
|
|
"total_events": total,
|
|
"nodes_tracked": len(self._by_node),
|
|
"actions_tracked": len(self._by_action),
|
|
"failures": failures,
|
|
}
|
|
|
|
def reset(self):
|
|
with self._lock:
|
|
self._events.clear()
|
|
self._by_node.clear()
|
|
self._by_action.clear()
|
|
|
|
|
|
# Process-wide singleton (used by api/timeline.py)
|
|
STORE = TimelineStore()
|