File size: 4,085 Bytes
945aae0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 | import time
import uuid
import statistics
from typing import Any
TRACE_TTL = 3600
class TraceEvent:
def __init__(self, event_type: str, data: dict):
self.id: str = str(uuid.uuid4())
self.type: str = event_type
self.data: dict = data
self.start_time: float = time.time()
self.end_time: float | None = None
self.steps: list[dict] = []
self.success: bool = True
class ObservabilityBrain:
def __init__(self):
self._traces: dict[str, TraceEvent] = {}
self._metrics: dict[str, list[float]] = {
"request_latency": [],
"policy_latency": [],
"memory_latency": [],
"service_latency": [],
}
self._error_count: int = 0
self._request_count: int = 0
def _cleanup_old_traces(self) -> None:
cutoff = time.time() - TRACE_TTL
expired = [tid for tid, t in self._traces.items() if (t.end_time or time.time()) < cutoff]
for tid in expired:
del self._traces[tid]
def start_trace(self, event: dict) -> str:
if len(self._traces) > 10000:
self._cleanup_old_traces()
trace = TraceEvent(event.get("type", "unknown"), event)
self._traces[trace.id] = trace
self._request_count += 1
return trace.id
def add_step(self, trace_id: str, step_name: str, data: Any = None) -> None:
trace = self._traces.get(trace_id)
if trace is None:
return
trace.steps.append({
"step": step_name,
"data": data,
"timestamp": time.time(),
})
def record_latency(self, metric_name: str, value: float) -> None:
if metric_name in self._metrics:
self._metrics[metric_name].append(value)
if len(self._metrics[metric_name]) > 1000:
self._metrics[metric_name] = self._metrics[metric_name][-1000:]
def end_trace(self, trace_id: str, success: bool = True) -> None:
trace = self._traces.get(trace_id)
if trace is None:
return
trace.end_time = time.time()
trace.success = success
if not success:
self._error_count += 1
latency = trace.end_time - trace.start_time
self.record_latency("request_latency", latency)
def get_trace(self, trace_id: str) -> dict | None:
trace = self._traces.get(trace_id)
if trace is None:
return None
return {
"id": trace.id,
"type": trace.type,
"duration": (trace.end_time or time.time()) - trace.start_time,
"success": trace.success,
"steps": trace.steps,
}
def replay(self, trace_id: str) -> dict | None:
return self.get_trace(trace_id)
def get_metrics(self) -> dict:
def _avg(values: list[float]) -> float:
return round(statistics.mean(values), 3) if values else 0.0
def _p95(values: list[float]) -> float:
if not values:
return 0.0
sorted_vals = sorted(values)
idx = int(len(sorted_vals) * 0.95)
return round(sorted_vals[min(idx, len(sorted_vals) - 1)], 3)
return {
"requests": self._request_count,
"errors": self._error_count,
"error_rate": round(self._error_count / max(self._request_count, 1), 4),
"avg_latency": _avg(self._metrics["request_latency"]),
"p95_latency": _p95(self._metrics["request_latency"]),
"avg_policy_latency": _avg(self._metrics["policy_latency"]),
"avg_memory_latency": _avg(self._metrics["memory_latency"]),
"avg_service_latency": _avg(self._metrics["service_latency"]),
"active_traces": sum(
1 for t in self._traces.values() if t.end_time is None
),
}
def reset(self) -> None:
self._traces.clear()
for k in self._metrics:
self._metrics[k].clear()
self._error_count = 0
self._request_count = 0
|