52 lines
1.7 KiB
Python
52 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
import time
|
|
from collections import defaultdict
|
|
from typing import Dict
|
|
|
|
|
|
class MetricsRecorder:
|
|
"""Lightweight in-process metrics aggregator."""
|
|
|
|
def __init__(self) -> None:
|
|
self.counters: Dict[str, int] = defaultdict(int)
|
|
self.timers: Dict[str, dict[str, float]] = defaultdict(lambda: {"count": 0, "total": 0.0})
|
|
self.last_reset: float = time.time()
|
|
|
|
def inc(self, key: str, amount: int = 1) -> None:
|
|
self.counters[key] += amount
|
|
|
|
def observe(self, key: str, value: float) -> None:
|
|
bucket = self.timers[key]
|
|
bucket["count"] += 1
|
|
bucket["total"] += value
|
|
|
|
def snapshot(self) -> dict:
|
|
snap = {
|
|
"counters": dict(self.counters),
|
|
"timers": {
|
|
name: {
|
|
"count": data["count"],
|
|
"avg": (data["total"] / data["count"]) if data["count"] else 0.0,
|
|
}
|
|
for name, data in self.timers.items()
|
|
},
|
|
"uptime": time.time() - self.last_reset,
|
|
}
|
|
return snap
|
|
|
|
def format_snapshot(self) -> str:
|
|
snap = self.snapshot()
|
|
counter_parts = [f"{k}={v}" for k, v in sorted(snap["counters"].items())]
|
|
timer_parts = [
|
|
f"{name}:count={data['count']},avg_ms={data['avg']:.1f}"
|
|
for name, data in sorted(snap["timers"].items())
|
|
]
|
|
parts = []
|
|
if counter_parts:
|
|
parts.append("counters[" + ", ".join(counter_parts) + "]")
|
|
if timer_parts:
|
|
parts.append("timers[" + ", ".join(timer_parts) + "]")
|
|
parts.append(f"uptime={snap['uptime']:.1f}s")
|
|
return " ".join(parts)
|