ri/observability/
metrics.rs

1//! Copyright © 2025-2026 Wenze Wei. All Rights Reserved.
2//!
3//! This file is part of Ri.
4//! The Ri project belongs to the Dunimd Team.
5//!
6//! Licensed under the Apache License, Version 2.0 (the "License");
7//! You may not use this file except in compliance with the License.
8//! You may obtain a copy of the License at
9//!
10//!     http://www.apache.org/licenses/LICENSE-2.0
11//!
12//! Unless required by applicable law or agreed to in writing, software
13//! distributed under the License is distributed on an "AS IS" BASIS,
14//! WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15//! See the License for the specific language governing permissions and
16//! limitations under the License.
17
18#![allow(non_snake_case)]
19
20//! # Metrics Collection and Aggregation Module
21//! 
22//! This module provides a comprehensive metrics collection and aggregation system for Ri.
23//! It supports various metric types, sliding window aggregation, and Prometheus-compatible export.
24//! 
25//! ## Key Components
26//! 
27//! - **RiMetricType**: Enum defining supported metric types (Counter, Gauge, Histogram, Summary)
28//! - **RiMetricSample**: Represents a single metric sample with timestamp, value, and labels
29//! - **RiMetricConfig**: Configuration for creating metrics
30//! - **RiSlidingWindow**: Internal sliding time window for metric aggregation
31//! - **RiWindowStats**: Aggregated statistics from the sliding window
32//! - **RiMetric**: Individual metric with sliding window aggregation
33//! - **RiMetricsRegistry**: Registry for managing multiple metrics
34//! 
35//! ## Design Principles
36//! 
37//! 1. **Multiple Metric Types**: Supports Counter, Gauge, Histogram, and Summary metrics
38//! 2. **Sliding Window Aggregation**: Efficiently aggregates metrics over configurable time windows
39//! 3. **Thread Safety**: Uses Arc and RwLock for safe concurrent access
40//! 4. **Prometheus Compatible**: Exports metrics in Prometheus format
41//! 5. **Label Support**: Allows adding custom labels to metric samples
42//! 6. **Configurable**: Supports custom window sizes, bucket sizes, and other parameters
43//! 7. **Type Safety**: Strongly typed metrics with compile-time checks
44//! 8. **Efficient Memory Usage**: Automatically rotates and prunes old metric data
45//! 
46//! ## Usage
47//! 
48//! ```rust
49//! use ri::prelude::*;
50//! use std::time::Duration;
51//! 
52//! fn example() -> RiResult<()> {
53//!     // Create a metrics registry
54//!     let registry = RiMetricsRegistry::new();
55//!     
56//!     // Configure a counter metric
57//!     let counter_config = RiMetricConfig {
58//!         metric_type: RiMetricType::Counter,
59//!         name: "http_requests_total".to_string(),
60//!         help: "Total number of HTTP requests".to_string(),
61//!         buckets: Vec::new(),
62//!         quantiles: Vec::new(),
63//!         max_age: Duration::from_secs(300),
64//!         age_buckets: 5,
65//!     };
66//!     
67//!     // Create and register the metric
68//!     let counter = Arc::new(RiMetric::new(counter_config));
69//!     registry.register(counter.clone())?;
70//!     
71//!     // Record some metrics
72//!     counter.record(1.0, vec![("method".to_string(), "GET".to_string())])?;
73//!     counter.record(1.0, vec![("method".to_string(), "POST".to_string())])?;
74//!     
75//!     // Export metrics in Prometheus format
76//!     let prometheus_output = registry.export_prometheus();
77//!     println!("{}", prometheus_output);
78//!     
79//!     Ok(())
80//! }
81//! ```
82
83use std::collections::{VecDeque, HashMap};
84use std::sync::{Arc, RwLock};
85use std::time::{SystemTime, UNIX_EPOCH, Duration};
86use serde::{Serialize, Deserialize};
87
88#[cfg(feature = "pyo3")]
89use pyo3::prelude::*;
90
91use crate::core::RiResult;
92use crate::core::lock::RwLockExtensions;
93
94/// Metric types supported
95#[derive(Debug, Clone, Serialize, Deserialize)]
96#[cfg_attr(feature = "pyo3", pyo3::prelude::pyclass)]
97pub enum RiMetricType {
98    Counter,
99    Gauge,
100    Histogram,
101    Summary,
102}
103
104/// A single metric sample
105#[derive(Debug, Clone, Serialize, Deserialize)]
106#[cfg_attr(feature = "pyo3", pyo3::prelude::pyclass)]
107pub struct RiMetricSample {
108    pub timestamp: u64, // seconds since epoch
109    pub value: f64,
110    pub labels: Vec<(String, String)>,
111}
112
113/// Metric configuration
114#[derive(Debug, Clone)]
115#[cfg_attr(feature = "pyo3", pyo3::prelude::pyclass)]
116pub struct RiMetricConfig {
117    pub metric_type: RiMetricType,
118    pub name: String,
119    pub help: String,
120    pub buckets: Vec<f64>, // for histogram
121    pub quantiles: Vec<f64>, // for summary
122    pub max_age: Duration, // for summary
123    pub age_buckets: usize, // for summary
124}
125
126/// Sliding time window for metric aggregation
127#[allow(dead_code)]
128struct RiSlidingWindow {
129    #[allow(dead_code)]
130    window_size: Duration,
131    #[allow(dead_code)]
132    bucket_size: Duration,
133    buckets: VecDeque<Vec<RiMetricSample>>,
134    current_bucket: Vec<RiMetricSample>,
135    #[allow(dead_code)]
136    last_rotation: u64,
137}
138
139impl RiSlidingWindow {
140    fn new(window_size: Duration, bucket_size: Duration) -> Self {
141        let bucket_count = window_size.as_secs().div_ceil(bucket_size.as_secs());
142        
143        Self {
144            window_size,
145            bucket_size,
146            buckets: VecDeque::with_capacity(bucket_count as usize),
147            current_bucket: Vec::new(),
148            last_rotation: Self::current_timestamp(),
149        }
150    }
151    
152    #[allow(dead_code)]
153    fn current_timestamp() -> u64 {
154        SystemTime::now()
155            .duration_since(UNIX_EPOCH)
156            .unwrap_or(Duration::from_secs(0))
157            .as_secs()
158    }
159    
160    #[allow(dead_code)]
161    fn rotate_if_needed(&mut self) {
162        let now = Self::current_timestamp();
163        let elapsed = now.saturating_sub(self.last_rotation);
164        
165        if elapsed >= self.bucket_size.as_secs() {
166            let rotations = elapsed / self.bucket_size.as_secs();
167            
168            for _ in 0..rotations {
169                self.buckets.push_back(std::mem::take(&mut self.current_bucket));
170                
171                // Remove old buckets outside window
172                let max_buckets = self.window_size.as_secs().div_ceil(self.bucket_size.as_secs());
173                while self.buckets.len() > max_buckets as usize {
174                    self.buckets.pop_front();
175                }
176            }
177            
178            self.last_rotation = now;
179        }
180    }
181    
182    #[allow(dead_code)]
183    fn add_sample(&mut self, sample: RiMetricSample) {
184        self.rotate_if_needed();
185        self.current_bucket.push(sample);
186    }
187    
188    #[allow(dead_code)]
189    fn get_samples(&self) -> Vec<RiMetricSample> {
190        let mut all_samples = Vec::with_capacity(8);
191        
192        for bucket in &self.buckets {
193            all_samples.extend(bucket.iter().cloned());
194        }
195        all_samples.extend(self.current_bucket.iter().cloned());
196        
197        all_samples
198    }
199    
200    #[allow(dead_code)]
201    fn get_window_stats(&self) -> RiWindowStats {
202        let samples = self.get_samples();
203        
204        if samples.is_empty() {
205            return RiWindowStats::default();
206        }
207        
208        let mut sorted_values: Vec<f64> = samples.iter().map(|s| s.value).collect();
209        sorted_values.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
210        
211        let count = sorted_values.len();
212        let sum: f64 = sorted_values.iter().sum();
213        let min = sorted_values[0];
214        let max = sorted_values[count - 1];
215        let mean = sum / count as f64;
216        
217        // Calculate variance and standard deviation
218        let variance: f64 = sorted_values
219            .iter()
220            .map(|x| (x - mean).powi(2))
221            .sum::<f64>() / count as f64;
222        let stddev = variance.sqrt();
223        
224        // Calculate quantiles
225        let p50 = Self::quantile(&sorted_values, 0.50);
226        let p90 = Self::quantile(&sorted_values, 0.90);
227        let p95 = Self::quantile(&sorted_values, 0.95);
228        let p99 = Self::quantile(&sorted_values, 0.99);
229        
230        RiWindowStats {
231            count: count as u64,
232            sum,
233            min,
234            max,
235            mean,
236            stddev,
237            p50,
238            p90,
239            p95,
240            p99,
241        }
242    }
243    
244    #[allow(dead_code)]
245    fn quantile(sorted_values: &[f64], q: f64) -> f64 {
246        if sorted_values.is_empty() {
247            return 0.0;
248        }
249        
250        let index = q * (sorted_values.len() - 1) as f64;
251        let lower = index.floor() as usize;
252        let upper = index.ceil() as usize;
253        
254        if lower == upper {
255            sorted_values[lower]
256        } else {
257            let weight = index - lower as f64;
258            sorted_values[lower] * (1.0 - weight) + sorted_values[upper] * weight
259        }
260    }
261}
262
263/// Window statistics
264#[derive(Debug, Clone, Serialize, Deserialize)]
265pub struct RiWindowStats {
266    pub count: u64,
267    pub sum: f64,
268    pub min: f64,
269    pub max: f64,
270    pub mean: f64,
271    pub stddev: f64,
272    pub p50: f64,
273    pub p90: f64,
274    pub p95: f64,
275    pub p99: f64,
276}
277
278impl Default for RiWindowStats {
279    fn default() -> Self {
280        Self {
281            count: 0,
282            sum: 0.0,
283            min: 0.0,
284            max: 0.0,
285            mean: 0.0,
286            stddev: 0.0,
287            p50: 0.0,
288            p90: 0.0,
289            p95: 0.0,
290            p99: 0.0,
291        }
292    }
293}
294
295/// A single metric with sliding window aggregation
296#[cfg_attr(feature = "pyo3", pyo3::prelude::pyclass)]
297pub struct RiMetric {
298    config: RiMetricConfig,
299    sliding_window: RwLock<RiSlidingWindow>,
300    total_count: RwLock<u64>,
301    #[allow(dead_code)]
302    total_sum: RwLock<f64>,
303}
304
305impl RiMetric {
306    pub fn new(config: RiMetricConfig) -> Self {
307        let sliding_window = RiSlidingWindow::new(
308            Duration::from_secs(300), // 5 minute window
309            Duration::from_secs(10),  // 10 second buckets
310        );
311        
312        Self {
313            config,
314            sliding_window: RwLock::new(sliding_window),
315            total_count: RwLock::new(0),
316            total_sum: RwLock::new(0.0),
317        }
318    }
319    
320    pub fn record(&self, value: f64, labels: Vec<(String, String)>) -> RiResult<()> {
321        let sample = RiMetricSample {
322            timestamp: Self::current_timestamp(),
323            value,
324            labels,
325        };
326        
327        {
328            let mut window = self.sliding_window.write_safe("sliding window")?;
329            window.add_sample(sample);
330        }
331        
332        {
333            let mut count = self.total_count.write_safe("total count")?;
334            *count += 1;
335        }
336        
337        {
338            let mut sum = self.total_sum.write_safe("total sum")?;
339            *sum += value;
340        }
341        
342        Ok(())
343    }
344    
345    #[allow(dead_code)]
346    fn get_stats(&self) -> RiWindowStats {
347        match self.sliding_window.read_safe("sliding window stats") {
348            Ok(window) => window.get_window_stats(),
349            Err(_) => RiWindowStats::default(),
350        }
351    }
352    
353    pub fn get_total_count(&self) -> u64 {
354        match self.total_count.read_safe("total count") {
355            Ok(count) => *count,
356            Err(_) => 0,
357        }
358    }
359    
360    #[allow(dead_code)]
361    fn get_total_sum(&self) -> f64 {
362        match self.total_sum.read_safe("total sum") {
363            Ok(sum) => *sum,
364            Err(_) => 0.0,
365        }
366    }
367    
368    pub fn get_config(&self) -> &RiMetricConfig {
369        &self.config
370    }
371
372    pub fn get_value(&self) -> f64 {
373        match self.total_count.read_safe("total count value") {
374            Ok(count) => *count as f64,
375            Err(_) => 0.0,
376        }
377    }
378
379    #[allow(dead_code)]
380    fn current_timestamp() -> u64 {
381        SystemTime::now()
382            .duration_since(UNIX_EPOCH)
383            .unwrap_or_default()
384            .as_secs()
385    }
386}
387
388/// Metrics registry to manage multiple metrics
389#[cfg_attr(feature = "pyo3", pyo3::prelude::pyclass)]
390#[derive(Clone)]
391pub struct RiMetricsRegistry {
392    metrics: Arc<RwLock<HashMap<String, Arc<RiMetric>>>>,
393}
394
395impl RiMetricsRegistry {
396    /// Maximum number of metrics that can be registered
397    const MAX_METRICS: usize = 10000;
398    
399    /// Maximum metric name length
400    const MAX_NAME_LENGTH: usize = 256;
401    
402    pub fn new() -> Self {
403        Self {
404            metrics: Arc::new(RwLock::new(HashMap::new())),
405        }
406    }
407    
408    /// Validates a metric name to prevent injection attacks.
409    ///
410    /// # Security
411    ///
412    /// Metric names must:
413    /// - Be 1-256 characters long
414    /// - Contain only alphanumeric characters, underscores, and colons
415    /// - Start with a letter
416    fn validate_metric_name(name: &str) -> RiResult<()> {
417        if name.is_empty() || name.len() > Self::MAX_NAME_LENGTH {
418            return Err(crate::core::RiError::Other(format!(
419                "Metric name must be 1-{} characters",
420                Self::MAX_NAME_LENGTH
421            )));
422        }
423        
424        let chars: Vec<char> = name.chars().collect();
425        
426        // First character must be a letter
427        if !chars[0].is_ascii_alphabetic() {
428            return Err(crate::core::RiError::Other(
429                "Metric name must start with a letter".to_string()
430            ));
431        }
432        
433        for c in &chars {
434            if !c.is_ascii_alphanumeric() && *c != '_' && *c != ':' {
435                return Err(crate::core::RiError::Other(
436                    "Metric name can only contain alphanumeric characters, underscores, and colons".to_string()
437                ));
438            }
439        }
440        
441        Ok(())
442    }
443    
444    pub fn register(&self, metric: Arc<RiMetric>) -> RiResult<()> {
445        let name = metric.get_config().name.clone();
446        
447        // Security: Validate metric name
448        Self::validate_metric_name(&name)?;
449        
450        let mut metrics = self.metrics.write_safe("metrics registry")?;
451        
452        // Security: Check maximum metrics limit
453        if metrics.len() >= Self::MAX_METRICS && !metrics.contains_key(&name) {
454            return Err(crate::core::RiError::Other(format!(
455                "Maximum metrics limit reached: {} metrics",
456                Self::MAX_METRICS
457            )));
458        }
459        
460        metrics.insert(name, metric);
461        Ok(())
462    }
463    
464    pub fn get_metric(&self, name: &str) -> Option<Arc<RiMetric>> {
465        match self.metrics.read_safe("metrics registry") {
466            Ok(metrics) => metrics.get(name).cloned(),
467            Err(_) => None,
468        }
469    }
470    
471    pub fn get_all_metrics(&self) -> HashMap<String, Arc<RiMetric>> {
472        match self.metrics.read_safe("metrics registry") {
473            Ok(metrics) => metrics.clone(),
474            Err(_) => HashMap::new(),
475        }
476    }
477    
478    /// Export metrics in Prometheus format
479    #[cfg(feature = "observability")]
480    pub fn export_prometheus(&self) -> String {
481        let mut output = String::new();
482        let metrics = match self.metrics.read_safe("metrics registry for export") {
483            Ok(m) => m,
484            Err(_) => return "# Error: Failed to acquire metrics registry lock".to_string(),
485        };
486        
487        for (name, metric) in metrics.iter() {
488            let config = metric.get_config();
489            
490            // Write help and type
491            output.push_str(&format!("# HELP {} {}\n", name, config.help));
492            output.push_str(&format!("# TYPE {} {:?}\n", name, config.metric_type));
493            
494            // Write metric value
495            let stats = metric.get_stats();
496            match config.metric_type {
497                RiMetricType::Counter => {
498                    output.push_str(&format!("{} {}\n", name, metric.get_total_count()));
499                }
500                RiMetricType::Gauge => {
501                    output.push_str(&format!("{} {}\n", name, stats.mean));
502                }
503                RiMetricType::Histogram => {
504                    output.push_str(&format!("{}_count {}\n", name, stats.count));
505                    output.push_str(&format!("{}_sum {}\n", name, stats.sum));
506                    output.push_str(&format!("{}_min {}\n", name, stats.min));
507                    output.push_str(&format!("{}_max {}\n", name, stats.max));
508                    output.push_str(&format!("{}_avg {}\n", name, stats.mean));
509                    output.push_str(&format!("{}_p50 {}\n", name, stats.p50));
510                    output.push_str(&format!("{}_p90 {}\n", name, stats.p90));
511                    output.push_str(&format!("{}_p95 {}\n", name, stats.p95));
512                    output.push_str(&format!("{}_p99 {}\n", name, stats.p99));
513                }
514                RiMetricType::Summary => {
515                    output.push_str(&format!("{} {}\n", name, stats.mean));
516                }
517            }
518            
519            output.push('\n');
520        }
521        
522        output
523    }
524}
525
526#[cfg(feature = "pyo3")]
527/// Python methods for RiMetricsRegistry
528#[pyo3::prelude::pymethods]
529impl RiMetricsRegistry {
530    /// Create a new metrics registry from Python
531    #[new]
532    fn py_new() -> Self {
533        Self::new()
534    }
535    
536    /// Register a metric from Python
537    #[pyo3(name = "register")]
538    fn register_py(&self, metric: &RiMetric) -> PyResult<()> {
539        let name = metric.config.name.clone();
540        let mut metrics = self.metrics.write_safe("metrics registry")
541            .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))?;
542        metrics.insert(name, Arc::new(RiMetric::new(metric.config.clone())));
543        Ok(())
544    }
545    
546    /// Get a metric's current value by name from Python
547    #[pyo3(name = "get_metric_value")]
548    fn get_metric_value_impl(&self, name: &str) -> Option<f64> {
549        self.get_metric(name).map(|m| m.get_value())
550    }
551
552    /// Get all metric names from Python
553    #[pyo3(name = "get_all_metric_names")]
554    fn get_all_metric_names_impl(&self) -> Vec<String> {
555        let metrics = match self.metrics.read_safe("metrics registry for names") {
556            Ok(m) => m,
557            Err(_) => return Vec::new(),
558        };
559        metrics.keys().cloned().collect()
560    }
561    
562    /// Export metrics in Prometheus format from Python
563    #[pyo3(name = "export_prometheus")]
564    fn export_prometheus_impl(&self) -> String {
565        #[cfg(feature = "observability")]
566        {
567            self.export_prometheus()
568        }
569        #[cfg(not(feature = "observability"))]
570        {
571            "# Observability feature not enabled".to_string()
572        }
573    }
574    
575    /// Get metric count from Python
576    #[pyo3(name = "get_metric_count")]
577    fn get_metric_count_impl(&self) -> usize {
578        let metrics = match self.metrics.read_safe("metrics registry for count") {
579            Ok(m) => m,
580            Err(_) => return 0,
581        };
582        metrics.len()
583    }
584}
585
586#[cfg(feature = "pyo3")]
587#[pyo3::prelude::pymethods]
588impl RiMetricConfig {
589    #[new]
590    #[pyo3(signature = (name, metric_type, help="", buckets=None, quantiles=None))]
591    fn py_new(
592        name: String,
593        metric_type: RiMetricType,
594        help: &str,
595        buckets: Option<Vec<f64>>,
596        quantiles: Option<Vec<f64>>,
597    ) -> Self {
598        Self {
599            name,
600            metric_type,
601            help: help.to_string(),
602            buckets: buckets.unwrap_or_else(|| vec![0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0]),
603            quantiles: quantiles.unwrap_or_else(|| vec![0.5, 0.9, 0.95, 0.99]),
604            max_age: Duration::from_secs(600),
605            age_buckets: 5,
606        }
607    }
608}
609
610#[cfg(feature = "pyo3")]
611#[pyo3::prelude::pymethods]
612impl RiMetric {
613    #[new]
614    fn py_new(config: RiMetricConfig) -> Self {
615        Self::new(config)
616    }
617    
618    #[pyo3(name = "record")]
619    fn record_py(&self, value: f64) -> PyResult<()> {
620        self.record(value, vec![])
621            .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))
622    }
623    
624    #[pyo3(name = "get_value")]
625    fn get_value_py(&self) -> f64 {
626        self.get_value()
627    }
628    
629    #[pyo3(name = "get_total_count")]
630    fn get_total_count_py(&self) -> u64 {
631        self.get_total_count()
632    }
633}
ri/observability/metrics.rs

ri/observability/
metrics.rs
