Coverage for merco/observability/metrics.py: 78%

1"""指标收集""" 

2 

3import time 

4from collections import defaultdict 

5 

6 

7class MetricsCollector: 

8 """收集与报告系统指标""" 

9 

10 def __init__(self): 

11 self._counters: dict[str, int] = defaultdict(int) 

12 self._timings: dict[str, list[float]] = defaultdict(list) 

13 self._events: list[dict] = [] 

14 

15 def increment(self, name: str, value: int = 1): 

16 """增加计数器""" 

17 self._counters[name] += value 

18 

19 def record_timing(self, name: str, duration: float): 

20 """记录耗时""" 

21 self._timings[name].append(duration) 

22 

23 def record_event(self, event_type: str, **kwargs): 

24 """记录事件""" 

25 self._events.append({ 

26 "type": event_type, 

27 "timestamp": time.time(), 

28 **kwargs, 

29 }) 

30 

31 def get_counter(self, name: str) -> int: 

32 """获取计数器值""" 

33 return self._counters.get(name, 0) 

34 

35 def get_avg_timing(self, name: str) -> float: 

36 """获取平均耗时""" 

37 timings = self._timings.get(name, []) 

38 return sum(timings) / len(timings) if timings else 0.0 

39 

40 def get_events(self, event_type: str = None, limit: int = 100) -> list[dict]: 

41 """获取事件列表""" 

42 events = self._events 

43 if event_type: 

44 events = [e for e in events if e["type"] == event_type] 

45 return events[-limit:] 

46 

47 def get_counters(self) -> dict[str, int]: 

48 """获取所有计数器（只读）""" 

49 return dict(self._counters) 

50 

51 def get_summary(self) -> dict: 

52 """获取指标摘要""" 

53 return { 

54 "counters": dict(self._counters), 

55 "avg_timings": { 

56 name: sum(vals) / len(vals) 

57 for name, vals in self._timings.items() 

58 if vals 

59 }, 

60 "total_events": len(self._events), 

61 }