platform/
metrics.rs

1//! Prometheus 监控指标模块
2//!
3//! 提供全局指标收集和导出功能
4
5use lazy_static::lazy_static;
6use prometheus::{
7    HistogramOpts, HistogramVec, IntCounterVec, IntGauge, IntGaugeVec, Opts, Registry,
8};
9use std::collections::HashMap;
10use std::sync::{Once, RwLock};
11use std::time::Instant;
12
13static METRICS_INIT: Once = Once::new();
14
15lazy_static! {
16    /// 全局 Prometheus Registry
17    pub static ref REGISTRY: Registry = Registry::new();
18
19    // ========== 业务指标 ==========
20
21    /// 当前注册的 Actor 总数（按 realm 分组）
22    pub static ref ACTORS_TOTAL: IntGaugeVec = IntGaugeVec::new(
23        Opts::new("actrix_actors_total", "Total number of registered actors")
24            .namespace("actrix"),
25        &["realm_id", "service"]
26    ).unwrap();
27
28    /// 当前注册的服务总数
29    pub static ref SERVICES_TOTAL: IntGaugeVec = IntGaugeVec::new(
30        Opts::new("actrix_services_total", "Total number of registered services")
31            .namespace("actrix"),
32        &["service_name"]
33    ).unwrap();
34
35    /// WebSocket 连接数
36    pub static ref WEBSOCKET_CONNECTIONS: IntGauge = IntGauge::new(
37        "actrix_websocket_connections",
38        "Number of active WebSocket connections"
39    ).unwrap();
40
41    /// Token 颁发次数（AIS）
42    pub static ref TOKENS_ISSUED: IntCounterVec = IntCounterVec::new(
43        Opts::new("actrix_tokens_issued_total", "Total number of tokens issued")
44            .namespace("actrix"),
45        &["realm_id", "status"]
46    ).unwrap();
47
48    /// Token 验证次数（Signaling/TURN）
49    pub static ref TOKENS_VALIDATED: IntCounterVec = IntCounterVec::new(
50        Opts::new("actrix_tokens_validated_total", "Total number of tokens validated")
51            .namespace("actrix"),
52        &["service", "status"]
53    ).unwrap();
54
55    // ========== 性能指标 ==========
56
57    /// HTTP 请求延迟（秒）
58    pub static ref REQUEST_DURATION: HistogramVec = HistogramVec::new(
59        HistogramOpts::new("actrix_request_duration_seconds", "HTTP request duration in seconds")
60            .namespace("actrix")
61            .buckets(vec![0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1.0, 5.0]),
62        &["service", "method", "path", "status"]
63    ).unwrap();
64
65    /// HTTP 请求总数
66    pub static ref REQUESTS_TOTAL: IntCounterVec = IntCounterVec::new(
67        Opts::new("actrix_requests_total", "Total number of HTTP requests")
68            .namespace("actrix"),
69        &["service", "method", "path", "status"]
70    ).unwrap();
71
72    /// 错误次数
73    pub static ref ERRORS_TOTAL: IntCounterVec = IntCounterVec::new(
74        Opts::new("actrix_errors_total", "Total number of errors")
75            .namespace("actrix"),
76        &["service", "error_type"]
77    ).unwrap();
78
79    // ========== 系统指标 ==========
80
81    /// 缓存命中次数
82    pub static ref CACHE_HITS: IntCounterVec = IntCounterVec::new(
83        Opts::new("actrix_cache_hits_total", "Total number of cache hits")
84            .namespace("actrix"),
85        &["cache_type"]
86    ).unwrap();
87
88    /// 缓存未命中次数
89    pub static ref CACHE_MISSES: IntCounterVec = IntCounterVec::new(
90        Opts::new("actrix_cache_misses_total", "Total number of cache misses")
91            .namespace("actrix"),
92        &["cache_type"]
93    ).unwrap();
94
95    /// 数据库连接池状态
96    pub static ref DB_CONNECTIONS: IntGaugeVec = IntGaugeVec::new(
97        Opts::new("actrix_db_connections", "Number of database connections")
98            .namespace("actrix"),
99        &["pool", "state"]
100    ).unwrap();
101
102    // ========== 安全指标 ==========
103
104    /// 速率限制触发次数
105    pub static ref RATE_LIMIT_EXCEEDED: IntCounterVec = IntCounterVec::new(
106        Opts::new("actrix_rate_limit_exceeded_total", "Total number of rate limit violations")
107            .namespace("actrix"),
108        &["service", "limiter_type"]
109    ).unwrap();
110
111    /// 认证失败次数
112    pub static ref AUTH_FAILURES: IntCounterVec = IntCounterVec::new(
113        Opts::new("actrix_auth_failures_total", "Total number of authentication failures")
114            .namespace("actrix"),
115        &["service", "reason"]
116    ).unwrap();
117
118    /// 非法请求次数
119    pub static ref INVALID_REQUESTS: IntCounterVec = IntCounterVec::new(
120        Opts::new("actrix_invalid_requests_total", "Total number of invalid requests")
121            .namespace("actrix"),
122        &["service", "reason"]
123    ).unwrap();
124
125    // ========== KS 特定指标 ==========
126
127    /// 密钥生成次数
128    pub static ref KEYS_GENERATED: IntCounterVec = IntCounterVec::new(
129        Opts::new("actrix_keys_generated_total", "Total number of keys generated")
130            .namespace("actrix"),
131        &["key_type"]
132    ).unwrap();
133
134    /// 密钥轮转次数
135    pub static ref KEY_ROTATIONS: IntCounterVec = IntCounterVec::new(
136        Opts::new("actrix_key_rotations_total", "Total number of key rotations")
137            .namespace("actrix"),
138        &["reason"]
139    ).unwrap();
140
141    // ========== TURN 特定指标 ==========
142
143    /// TURN 分配请求
144    pub static ref TURN_ALLOCATIONS: IntCounterVec = IntCounterVec::new(
145        Opts::new("actrix_turn_allocations_total", "Total number of TURN allocations")
146            .namespace("actrix"),
147        &["status"]
148    ).unwrap();
149
150    /// TURN 活跃会话数
151    pub static ref TURN_ACTIVE_SESSIONS: IntGauge = IntGauge::new(
152        "actrix_turn_active_sessions",
153        "Number of active TURN sessions"
154    ).unwrap();
155
156    /// TURN 流量统计（字节）
157    pub static ref TURN_BYTES_RELAYED: IntCounterVec = IntCounterVec::new(
158        Opts::new("actrix_turn_bytes_relayed_total", "Total bytes relayed by TURN")
159            .namespace("actrix"),
160        &["direction"]
161    ).unwrap();
162}
163
164/// Per-service Prometheus registries for isolated metrics export.
165pub struct ServiceMetricsRegistry {
166    registries: RwLock<HashMap<String, Registry>>,
167}
168
169impl Default for ServiceMetricsRegistry {
170    fn default() -> Self {
171        Self {
172            registries: RwLock::new(HashMap::new()),
173        }
174    }
175}
176
177impl ServiceMetricsRegistry {
178    pub fn new() -> Self {
179        Self {
180            registries: RwLock::new(HashMap::new()),
181        }
182    }
183
184    /// Get an existing registry or create a new one for the given service.
185    pub fn get_or_create(&self, service: &str) -> Registry {
186        {
187            let read = self.registries.read().unwrap();
188            if let Some(r) = read.get(service) {
189                return r.clone();
190            }
191        }
192        let mut write = self.registries.write().unwrap();
193        write.entry(service.to_string()).or_default().clone()
194    }
195
196    /// Export Prometheus text for a single service registry.
197    pub fn export_for(&self, service: &str) -> String {
198        use prometheus::Encoder;
199        let read = self.registries.read().unwrap();
200        let Some(registry) = read.get(service) else {
201            return String::new();
202        };
203        let encoder = prometheus::TextEncoder::new();
204        let families = registry.gather();
205        let mut buf = Vec::new();
206        encoder.encode(&families, &mut buf).unwrap();
207        String::from_utf8(buf).unwrap()
208    }
209
210    /// Export all per-service registries merged with the global REGISTRY.
211    pub fn export_all(&self) -> String {
212        use prometheus::Encoder;
213        let encoder = prometheus::TextEncoder::new();
214
215        // Gather from global registry
216        let mut families = REGISTRY.gather();
217
218        // Gather from all per-service registries
219        let read = self.registries.read().unwrap();
220        for registry in read.values() {
221            families.extend(registry.gather());
222        }
223
224        let mut buf = Vec::new();
225        encoder.encode(&families, &mut buf).unwrap();
226        String::from_utf8(buf).unwrap()
227    }
228}
229
230lazy_static! {
231    /// Per-service Prometheus registries
232    pub static ref SERVICE_REGISTRIES: ServiceMetricsRegistry = ServiceMetricsRegistry::new();
233}
234
235/// 注册所有指标到全局 Registry
236///
237/// This function is idempotent - calling it multiple times is safe.
238/// Only the first call will actually register the metrics.
239pub fn register_metrics() -> Result<(), prometheus::Error> {
240    let mut result = Ok(());
241
242    METRICS_INIT.call_once(|| {
243        let register_result = (|| {
244            // 业务指标
245            REGISTRY.register(Box::new(ACTORS_TOTAL.clone()))?;
246            REGISTRY.register(Box::new(SERVICES_TOTAL.clone()))?;
247            REGISTRY.register(Box::new(WEBSOCKET_CONNECTIONS.clone()))?;
248            REGISTRY.register(Box::new(TOKENS_ISSUED.clone()))?;
249            REGISTRY.register(Box::new(TOKENS_VALIDATED.clone()))?;
250
251            // 性能指标
252            REGISTRY.register(Box::new(REQUEST_DURATION.clone()))?;
253            REGISTRY.register(Box::new(REQUESTS_TOTAL.clone()))?;
254            REGISTRY.register(Box::new(ERRORS_TOTAL.clone()))?;
255
256            // 系统指标
257            REGISTRY.register(Box::new(CACHE_HITS.clone()))?;
258            REGISTRY.register(Box::new(CACHE_MISSES.clone()))?;
259            REGISTRY.register(Box::new(DB_CONNECTIONS.clone()))?;
260
261            // 安全指标
262            REGISTRY.register(Box::new(RATE_LIMIT_EXCEEDED.clone()))?;
263            REGISTRY.register(Box::new(AUTH_FAILURES.clone()))?;
264            REGISTRY.register(Box::new(INVALID_REQUESTS.clone()))?;
265
266            // KS 特定指标
267            REGISTRY.register(Box::new(KEYS_GENERATED.clone()))?;
268            REGISTRY.register(Box::new(KEY_ROTATIONS.clone()))?;
269
270            // TURN 特定指标
271            REGISTRY.register(Box::new(TURN_ALLOCATIONS.clone()))?;
272            REGISTRY.register(Box::new(TURN_ACTIVE_SESSIONS.clone()))?;
273            REGISTRY.register(Box::new(TURN_BYTES_RELAYED.clone()))?;
274
275            Ok::<(), prometheus::Error>(())
276        })();
277
278        if let Err(e) = register_result {
279            result = Err(e);
280        }
281    });
282
283    result
284}
285
286/// HTTP 请求计时器
287pub struct RequestTimer {
288    start: Instant,
289    service: String,
290    method: String,
291    path: String,
292}
293
294impl RequestTimer {
295    /// 创建计时器
296    pub fn new(service: &str, method: &str, path: &str) -> Self {
297        Self {
298            start: Instant::now(),
299            service: service.to_string(),
300            method: method.to_string(),
301            path: path.to_string(),
302        }
303    }
304
305    /// 完成计时并记录指标
306    pub fn observe(self, status: u16) {
307        let duration = self.start.elapsed().as_secs_f64();
308        let status_str = status.to_string();
309
310        REQUEST_DURATION
311            .with_label_values(&[&self.service, &self.method, &self.path, &status_str])
312            .observe(duration);
313
314        REQUESTS_TOTAL
315            .with_label_values(&[&self.service, &self.method, &self.path, &status_str])
316            .inc();
317    }
318}
319
320/// 导出 Prometheus 格式的指标
321pub fn export_metrics() -> String {
322    use prometheus::Encoder;
323    let encoder = prometheus::TextEncoder::new();
324    let metric_families = REGISTRY.gather();
325
326    let mut buffer = Vec::new();
327    encoder.encode(&metric_families, &mut buffer).unwrap();
328
329    String::from_utf8(buffer).unwrap()
330}
331
332#[cfg(test)]
333mod tests {
334    use super::*;
335
336    #[test]
337    fn test_register_metrics() {
338        // 注册应该成功（或者已经注册过了）
339        let result = register_metrics();
340        // 如果已注册，会返回错误，但不影响功能
341        assert!(result.is_ok() || result.is_err());
342    }
343
344    #[test]
345    fn test_request_timer() {
346        let _ = register_metrics();
347
348        let timer = RequestTimer::new("test-service", "GET", "/test");
349        std::thread::sleep(std::time::Duration::from_millis(10));
350        timer.observe(200);
351
352        // 验证计数器增加
353        let before = REQUESTS_TOTAL
354            .with_label_values(&["test-service", "GET", "/test", "200"])
355            .get();
356
357        let timer2 = RequestTimer::new("test-service", "GET", "/test");
358        timer2.observe(200);
359
360        let after = REQUESTS_TOTAL
361            .with_label_values(&["test-service", "GET", "/test", "200"])
362            .get();
363
364        assert!(after > before);
365    }
366
367    #[test]
368    fn test_export_metrics() {
369        let _ = register_metrics();
370
371        WEBSOCKET_CONNECTIONS.set(42);
372
373        let output = export_metrics();
374        // Check for the metric name (may appear with or without namespace prefix)
375        assert!(
376            output.contains("actrix_websocket_connections")
377                || output.contains("websocket_connections"),
378            "Output should contain websocket_connections metric. Output: {output}"
379        );
380        assert!(
381            output.contains("42"),
382            "Output should contain value 42. Output: {output}"
383        );
384    }
385}
platform/metrics.rs

platform/
metrics.rs
