distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""StatManager — central statistics repository.
2
3Ported from net.i2p.stat.StatManager.
4"""
5
6import threading
7from typing import Dict, List, Optional, Set
8
9from i2p_stat.rate_stat import RateStat
10from i2p_stat.frequency_stat import FrequencyStat
11
12
13class StatManager:
14 """Central registry for all RateStats and FrequencyStats.
15
16 By default, only "required" stats are created. Set stat_full=True
17 to allow all stats (used in router context for full monitoring).
18 """
19
20 _FREQ_COALESCE_RATE = 9
21
22 def __init__(self, stat_full: bool = False) -> None:
23 self._stat_full = stat_full
24 self._lock = threading.Lock()
25 self._rate_stats: Dict[str, RateStat] = {}
26 self._frequency_stats: Dict[str, FrequencyStat] = {}
27 self._coalesce_counter = 0
28
29 def shutdown(self) -> None:
30 with self._lock:
31 self._rate_stats.clear()
32 self._frequency_stats.clear()
33
34 # --- Rate Stats ---
35
36 def create_rate_stat(
37 self,
38 name: str,
39 description: str,
40 group: str,
41 periods: List[int],
42 ) -> None:
43 """Create a rate stat if stat_full is True."""
44 if not self._stat_full:
45 return
46 self.create_required_rate_stat(name, description, group, periods)
47
48 def create_required_rate_stat(
49 self,
50 name: str,
51 description: str,
52 group: str,
53 periods: List[int],
54 ) -> None:
55 """Always create the rate stat."""
56 if name not in self._rate_stats:
57 self._rate_stats[name] = RateStat(name, description, group, periods)
58
59 def remove_rate_stat(self, name: str) -> None:
60 self._rate_stats.pop(name, None)
61
62 def add_rate_data(
63 self, name: str, data: int, event_duration: int = 0
64 ) -> None:
65 stat = self._rate_stats.get(name)
66 if stat is not None:
67 stat.add_data(data, event_duration)
68
69 def get_rate(self, name: str) -> Optional[RateStat]:
70 return self._rate_stats.get(name)
71
72 def get_rate_names(self) -> Set[str]:
73 return set(self._rate_stats.keys())
74
75 def is_rate(self, name: str) -> bool:
76 return name in self._rate_stats
77
78 # --- Frequency Stats ---
79
80 def create_frequency_stat(
81 self,
82 name: str,
83 description: str,
84 group: str,
85 periods: List[int],
86 ) -> None:
87 if not self._stat_full:
88 return
89 self.create_required_frequency_stat(name, description, group, periods)
90
91 def create_required_frequency_stat(
92 self,
93 name: str,
94 description: str,
95 group: str,
96 periods: List[int],
97 ) -> None:
98 if name not in self._frequency_stats:
99 self._frequency_stats[name] = FrequencyStat(
100 name, description, group, periods
101 )
102
103 def update_frequency(self, name: str) -> None:
104 stat = self._frequency_stats.get(name)
105 if stat is not None:
106 stat.event_occurred()
107
108 def get_frequency(self, name: str) -> Optional[FrequencyStat]:
109 return self._frequency_stats.get(name)
110
111 def get_frequency_names(self) -> Set[str]:
112 return set(self._frequency_stats.keys())
113
114 def is_frequency(self, name: str) -> bool:
115 return name in self._frequency_stats
116
117 # --- Coalesce ---
118
119 def coalesce_stats(self) -> None:
120 """Coalesce all rate stats, and frequency stats every Nth call."""
121 with self._lock:
122 self._coalesce_counter += 1
123 do_freq = self._coalesce_counter % self._FREQ_COALESCE_RATE == 0
124
125 for rs in list(self._rate_stats.values()):
126 rs.coalesce_stats()
127
128 if do_freq:
129 for fs in list(self._frequency_stats.values()):
130 fs.coalesce_stats()
131
132 # --- Grouping ---
133
134 def get_stats_by_group(self) -> Dict[str, Set[str]]:
135 result: Dict[str, Set[str]] = {}
136 for name, rs in self._rate_stats.items():
137 result.setdefault(rs.group_name, set()).add(name)
138 for name, fs in self._frequency_stats.items():
139 result.setdefault(fs.group_name, set()).add(name)
140 return result