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    #[allow(dead_code)]
321    fn record(&self, value: f64, labels: Vec<(String, String)>) -> RiResult<()> {
322        let sample = RiMetricSample {
323            timestamp: Self::current_timestamp(),
324            value,
325            labels,
326        };
327        
328        {
329            let mut window = self.sliding_window.write_safe("sliding window")?;
330            window.add_sample(sample);
331        }
332        
333        {
334            let mut count = self.total_count.write_safe("total count")?;
335            *count += 1;
336        }
337        
338        {
339            let mut sum = self.total_sum.write_safe("total sum")?;
340            *sum += value;
341        }
342        
343        Ok(())
344    }
345    
346    #[allow(dead_code)]
347    fn get_stats(&self) -> RiWindowStats {
348        match self.sliding_window.read_safe("sliding window stats") {
349            Ok(window) => window.get_window_stats(),
350            Err(_) => RiWindowStats::default(),
351        }
352    }
353    
354    #[allow(dead_code)]
355    fn get_total_count(&self) -> u64 {
356        match self.total_count.read_safe("total count") {
357            Ok(count) => *count,
358            Err(_) => 0,
359        }
360    }
361    
362    #[allow(dead_code)]
363    fn get_total_sum(&self) -> f64 {
364        match self.total_sum.read_safe("total sum") {
365            Ok(sum) => *sum,
366            Err(_) => 0.0,
367        }
368    }
369    
370    fn get_config(&self) -> &RiMetricConfig {
371        &self.config
372    }
373
374    pub fn get_value(&self) -> f64 {
375        match self.total_count.read_safe("total count value") {
376            Ok(count) => *count as f64,
377            Err(_) => 0.0,
378        }
379    }
380
381    #[allow(dead_code)]
382    fn current_timestamp() -> u64 {
383        SystemTime::now()
384            .duration_since(UNIX_EPOCH)
385            .unwrap_or_default()
386            .as_secs()
387    }
388}
389
390/// Metrics registry to manage multiple metrics
391#[cfg_attr(feature = "pyo3", pyo3::prelude::pyclass)]
392#[derive(Clone)]
393pub struct RiMetricsRegistry {
394    metrics: Arc<RwLock<HashMap<String, Arc<RiMetric>>>>,
395}
396
397impl RiMetricsRegistry {
398    /// Maximum number of metrics that can be registered
399    const MAX_METRICS: usize = 10000;
400    
401    /// Maximum metric name length
402    const MAX_NAME_LENGTH: usize = 256;
403    
404    pub fn new() -> Self {
405        Self {
406            metrics: Arc::new(RwLock::new(HashMap::new())),
407        }
408    }
409    
410    /// Validates a metric name to prevent injection attacks.
411    ///
412    /// # Security
413    ///
414    /// Metric names must:
415    /// - Be 1-256 characters long
416    /// - Contain only alphanumeric characters, underscores, and colons
417    /// - Start with a letter
418    fn validate_metric_name(name: &str) -> RiResult<()> {
419        if name.is_empty() || name.len() > Self::MAX_NAME_LENGTH {
420            return Err(crate::core::RiError::Other(format!(
421                "Metric name must be 1-{} characters",
422                Self::MAX_NAME_LENGTH
423            )));
424        }
425        
426        let chars: Vec<char> = name.chars().collect();
427        
428        // First character must be a letter
429        if !chars[0].is_ascii_alphabetic() {
430            return Err(crate::core::RiError::Other(
431                "Metric name must start with a letter".to_string()
432            ));
433        }
434        
435        for c in &chars {
436            if !c.is_ascii_alphanumeric() && *c != '_' && *c != ':' {
437                return Err(crate::core::RiError::Other(
438                    "Metric name can only contain alphanumeric characters, underscores, and colons".to_string()
439                ));
440            }
441        }
442        
443        Ok(())
444    }
445    
446    pub fn register(&self, metric: Arc<RiMetric>) -> RiResult<()> {
447        let name = metric.get_config().name.clone();
448        
449        // Security: Validate metric name
450        Self::validate_metric_name(&name)?;
451        
452        let mut metrics = self.metrics.write_safe("metrics registry")?;
453        
454        // Security: Check maximum metrics limit
455        if metrics.len() >= Self::MAX_METRICS && !metrics.contains_key(&name) {
456            return Err(crate::core::RiError::Other(format!(
457                "Maximum metrics limit reached: {} metrics",
458                Self::MAX_METRICS
459            )));
460        }
461        
462        metrics.insert(name, metric);
463        Ok(())
464    }
465    
466    pub fn get_metric(&self, name: &str) -> Option<Arc<RiMetric>> {
467        match self.metrics.read_safe("metrics registry") {
468            Ok(metrics) => metrics.get(name).cloned(),
469            Err(_) => None,
470        }
471    }
472    
473    pub fn get_all_metrics(&self) -> HashMap<String, Arc<RiMetric>> {
474        match self.metrics.read_safe("metrics registry") {
475            Ok(metrics) => metrics.clone(),
476            Err(_) => HashMap::new(),
477        }
478    }
479    
480    /// Export metrics in Prometheus format
481    #[cfg(feature = "observability")]
482    pub fn export_prometheus(&self) -> String {
483        let mut output = String::new();
484        let metrics = match self.metrics.read_safe("metrics registry for export") {
485            Ok(m) => m,
486            Err(_) => return "# Error: Failed to acquire metrics registry lock".to_string(),
487        };
488        
489        for (name, metric) in metrics.iter() {
490            let config = metric.get_config();
491            
492            // Write help and type
493            output.push_str(&format!("# HELP {} {}\n", name, config.help));
494            output.push_str(&format!("# TYPE {} {:?}\n", name, config.metric_type));
495            
496            // Write metric value
497            let stats = metric.get_stats();
498            match config.metric_type {
499                RiMetricType::Counter => {
500                    output.push_str(&format!("{} {}\n", name, metric.get_total_count()));
501                }
502                RiMetricType::Gauge => {
503                    output.push_str(&format!("{} {}\n", name, stats.mean));
504                }
505                RiMetricType::Histogram => {
506                    output.push_str(&format!("{}_count {}\n", name, stats.count));
507                    output.push_str(&format!("{}_sum {}\n", name, stats.sum));
508                    output.push_str(&format!("{}_min {}\n", name, stats.min));
509                    output.push_str(&format!("{}_max {}\n", name, stats.max));
510                    output.push_str(&format!("{}_avg {}\n", name, stats.mean));
511                    output.push_str(&format!("{}_p50 {}\n", name, stats.p50));
512                    output.push_str(&format!("{}_p90 {}\n", name, stats.p90));
513                    output.push_str(&format!("{}_p95 {}\n", name, stats.p95));
514                    output.push_str(&format!("{}_p99 {}\n", name, stats.p99));
515                }
516                RiMetricType::Summary => {
517                    output.push_str(&format!("{} {}\n", name, stats.mean));
518                }
519            }
520            
521            output.push('\n');
522        }
523        
524        output
525    }
526}
527
528#[cfg(feature = "pyo3")]
529/// Python methods for RiMetricsRegistry
530#[pyo3::prelude::pymethods]
531impl RiMetricsRegistry {
532    /// Create a new metrics registry from Python
533    #[new]
534    fn py_new() -> Self {
535        Self::new()
536    }
537    
538    /// Register a metric from Python
539    #[pyo3(name = "register")]
540    fn register_py(&self, metric: &RiMetric) -> PyResult<()> {
541        let name = metric.config.name.clone();
542        let mut metrics = self.metrics.write_safe("metrics registry")
543            .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))?;
544        metrics.insert(name, Arc::new(RiMetric::new(metric.config.clone())));
545        Ok(())
546    }
547    
548    /// Get a metric's current value by name from Python
549    #[pyo3(name = "get_metric_value")]
550    fn get_metric_value_impl(&self, name: &str) -> Option<f64> {
551        self.get_metric(name).map(|m| m.get_value())
552    }
553
554    /// Get all metric names from Python
555    #[pyo3(name = "get_all_metric_names")]
556    fn get_all_metric_names_impl(&self) -> Vec<String> {
557        let metrics = match self.metrics.read_safe("metrics registry for names") {
558            Ok(m) => m,
559            Err(_) => return Vec::new(),
560        };
561        metrics.keys().cloned().collect()
562    }
563    
564    /// Export metrics in Prometheus format from Python
565    #[pyo3(name = "export_prometheus")]
566    fn export_prometheus_impl(&self) -> String {
567        #[cfg(feature = "observability")]
568        {
569            self.export_prometheus()
570        }
571        #[cfg(not(feature = "observability"))]
572        {
573            "# Observability feature not enabled".to_string()
574        }
575    }
576    
577    /// Get metric count from Python
578    #[pyo3(name = "get_metric_count")]
579    fn get_metric_count_impl(&self) -> usize {
580        let metrics = match self.metrics.read_safe("metrics registry for count") {
581            Ok(m) => m,
582            Err(_) => return 0,
583        };
584        metrics.len()
585    }
586}
587
588#[cfg(feature = "pyo3")]
589#[pyo3::prelude::pymethods]
590impl RiMetricConfig {
591    #[new]
592    #[pyo3(signature = (name, metric_type, help="", buckets=None, quantiles=None))]
593    fn py_new(
594        name: String,
595        metric_type: RiMetricType,
596        help: &str,
597        buckets: Option<Vec<f64>>,
598        quantiles: Option<Vec<f64>>,
599    ) -> Self {
600        Self {
601            name,
602            metric_type,
603            help: help.to_string(),
604            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]),
605            quantiles: quantiles.unwrap_or_else(|| vec![0.5, 0.9, 0.95, 0.99]),
606            max_age: Duration::from_secs(600),
607            age_buckets: 5,
608        }
609    }
610}
611
612#[cfg(feature = "pyo3")]
613#[pyo3::prelude::pymethods]
614impl RiMetric {
615    #[new]
616    fn py_new(config: RiMetricConfig) -> Self {
617        Self::new(config)
618    }
619    
620    #[pyo3(name = "record")]
621    fn record_py(&self, value: f64) -> PyResult<()> {
622        self.record(value, vec![])
623            .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))
624    }
625    
626    #[pyo3(name = "get_value")]
627    fn get_value_py(&self) -> f64 {
628        self.get_value()
629    }
630    
631    #[pyo3(name = "get_total_count")]
632    fn get_total_count_py(&self) -> u64 {
633        self.get_total_count()
634    }
635}
ri/observability/metrics.rs

ri/observability/
metrics.rs
