#![allow(non_snake_case)]
use serde::{Deserialize, Serialize};
use std::collections::VecDeque;
use std::time::{Duration, Instant};
use sysinfo::{CpuExt, DiskExt, NetworkExt, System, SystemExt};
#[allow(dead_code)]
#[derive(Debug, Clone)]
pub struct RiSlidingWindow<T> {
_window_size: Duration,
bucket_size: Duration,
buckets: VecDeque<WindowBucket<T>>,
current_time: Instant,
}
#[allow(dead_code)]
#[derive(Debug, Clone)]
struct WindowBucket<T> {
_start_time: Instant,
end_time: Instant,
data_points: Vec<T>,
}
#[allow(dead_code)]
impl<T> RiSlidingWindow<T> {
pub fn new(window_size: Duration, bucket_size: Duration) -> Self {
let bucket_count = (window_size.as_millis() / bucket_size.as_millis()).max(1) as usize;
let mut buckets = VecDeque::with_capacity(bucket_count);
let now = Instant::now();
for i in 0..bucket_count {
let bucket_start = now - window_size
+ Duration::from_millis(i as u64 * bucket_size.as_millis() as u64);
buckets.push_back(WindowBucket {
_start_time: bucket_start,
end_time: bucket_start + bucket_size,
data_points: Vec::new(),
});
}
Self {
_window_size: window_size,
bucket_size,
buckets,
current_time: now,
}
}
pub fn add(&mut self, value: T) {
self.advance_window();
if let Some(current_bucket) = self.buckets.back_mut() {
current_bucket.data_points.push(value);
}
}
pub fn get_data_points(&self) -> Vec<&T> {
self.buckets
.iter()
.flat_map(|bucket| &bucket.data_points)
.collect()
}
pub fn count(&self) -> usize {
self.buckets
.iter()
.map(|bucket| bucket.data_points.len())
.sum()
}
fn advance_window(&mut self) {
let now = Instant::now();
let elapsed = now.duration_since(self.current_time);
if elapsed >= self.bucket_size {
let buckets_to_advance = (elapsed.as_millis() / self.bucket_size.as_millis()) as usize;
for _ in 0..buckets_to_advance.min(self.buckets.len()) {
self.buckets.pop_front();
let new_bucket_start = match self.buckets.back() {
Some(bucket) => bucket.end_time,
None => {
continue;
}
};
self.buckets.push_back(WindowBucket {
_start_time: new_bucket_start,
end_time: new_bucket_start + self.bucket_size,
data_points: Vec::new(),
});
}
self.current_time = now;
}
}
}
#[allow(dead_code)]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RiQuantileCalculator {
sorted_data: Vec<f64>,
}
#[allow(dead_code)]
impl Default for RiQuantileCalculator {
fn default() -> Self {
Self::new()
}
}
#[allow(dead_code)]
impl RiQuantileCalculator {
pub fn new() -> Self {
Self {
sorted_data: Vec::new(),
}
}
pub fn add(&mut self, value: f64) {
self.sorted_data.push(value);
}
pub fn quantile(&mut self, q: f64) -> Option<f64> {
if self.sorted_data.is_empty() || !(0.0..=1.0).contains(&q) {
return None;
}
self.sorted_data.sort_by(|a, b| {
a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal)
});
let n = self.sorted_data.len();
let index = q * (n - 1) as f64;
let lower = index.floor() as usize;
let upper = index.ceil() as usize;
if lower == upper {
Some(self.sorted_data[lower])
} else {
let weight = index - lower as f64;
let lower_val = self.sorted_data[lower];
let upper_val = self.sorted_data[upper];
Some(lower_val + weight * (upper_val - lower_val))
}
}
pub fn quantiles(&mut self, quantiles: &[f64]) -> Vec<Option<f64>> {
quantiles.iter().map(|&q| self.quantile(q)).collect()
}
pub fn min(&self) -> Option<f64> {
self.sorted_data.iter().fold(None, |min, &val| match min {
None => Some(val),
Some(m) => Some(m.min(val)),
})
}
pub fn max(&self) -> Option<f64> {
self.sorted_data.iter().fold(None, |max, &val| match max {
None => Some(val),
Some(m) => Some(m.max(val)),
})
}
pub fn mean(&self) -> Option<f64> {
if self.sorted_data.is_empty() {
return None;
}
let sum: f64 = self.sorted_data.iter().sum();
Some(sum / self.sorted_data.len() as f64)
}
pub fn std_dev(&self) -> Option<f64> {
let mean = self.mean()?;
if self.sorted_data.len() <= 1 {
return Some(0.0);
}
let variance: f64 = self
.sorted_data
.iter()
.map(|&x| (x - mean).powi(2))
.sum::<f64>()
/ (self.sorted_data.len() - 1) as f64;
Some(variance.sqrt())
}
pub fn clear(&mut self) {
self.sorted_data.clear();
}
}
#[allow(dead_code)]
pub struct RiPerformanceCollector {
latency_window: RiSlidingWindow<f64>,
throughput_window: RiSlidingWindow<u64>,
error_rate_window: RiSlidingWindow<bool>,
quantile_calculator: RiQuantileCalculator,
}
#[allow(dead_code)]
impl RiPerformanceCollector {
pub fn new(window_size: Duration, bucket_size: Duration) -> Self {
Self {
latency_window: RiSlidingWindow::new(window_size, bucket_size),
throughput_window: RiSlidingWindow::new(window_size, bucket_size),
error_rate_window: RiSlidingWindow::new(window_size, bucket_size),
quantile_calculator: RiQuantileCalculator::new(),
}
}
pub fn record_request(&mut self, latency_ms: f64, is_error: bool) {
self.latency_window.add(latency_ms);
self.throughput_window.add(1);
self.error_rate_window.add(is_error);
}
pub fn get_metrics(&mut self) -> RiPerformanceMetrics {
let latencies: Vec<f64> = self
.latency_window
.get_data_points()
.iter()
.map(|&&x| x)
.collect();
self.quantile_calculator.clear();
for &latency in &latencies {
self.quantile_calculator.add(latency);
}
let p50 = self.quantile_calculator.quantile(0.5).unwrap_or(0.0);
let p95 = self.quantile_calculator.quantile(0.95).unwrap_or(0.0);
let p99 = self.quantile_calculator.quantile(0.99).unwrap_or(0.0);
let errors: Vec<bool> = self
.error_rate_window
.get_data_points()
.iter()
.map(|&&x| x)
.collect();
let error_count = errors.iter().filter(|&&x| x).count();
let total_requests = errors.len();
let error_rate = if total_requests > 0 {
error_count as f64 / total_requests as f64
} else {
0.0
};
RiPerformanceMetrics {
p50_latency_ms: p50,
p95_latency_ms: p95,
p99_latency_ms: p99,
mean_latency_ms: self.quantile_calculator.mean().unwrap_or(0.0),
min_latency_ms: self.quantile_calculator.min().unwrap_or(0.0),
max_latency_ms: self.quantile_calculator.max().unwrap_or(0.0),
throughput_rps: self.throughput_window.count() as f64,
error_rate,
total_requests: total_requests as u64,
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RiPerformanceMetrics {
pub p50_latency_ms: f64,
pub p95_latency_ms: f64,
pub p99_latency_ms: f64,
pub mean_latency_ms: f64,
pub min_latency_ms: f64,
pub max_latency_ms: f64,
pub throughput_rps: f64,
pub error_rate: f64,
pub total_requests: u64,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[cfg_attr(feature = "pyo3", pyo3::prelude::pyclass(get_all))]
pub struct RiCPUMetrics {
pub total_usage_percent: f64,
pub per_core_usage: Vec<f64>,
pub context_switches: u64,
pub interrupts: u64,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[cfg_attr(feature = "pyo3", pyo3::prelude::pyclass(get_all))]
pub struct RiMemoryMetrics {
pub total_bytes: u64,
pub used_bytes: u64,
pub free_bytes: u64,
pub usage_percent: f64,
pub swap_total_bytes: u64,
pub swap_used_bytes: u64,
pub swap_free_bytes: u64,
pub swap_usage_percent: f64,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[cfg_attr(feature = "pyo3", pyo3::prelude::pyclass(get_all))]
pub struct RiDiskMetrics {
pub total_bytes: u64,
pub used_bytes: u64,
pub free_bytes: u64,
pub usage_percent: f64,
pub read_bytes: u64,
pub write_bytes: u64,
pub read_count: u64,
pub write_count: u64,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[cfg_attr(feature = "pyo3", pyo3::prelude::pyclass(get_all))]
pub struct RiNetworkMetrics {
pub total_received_bytes: u64,
pub total_transmitted_bytes: u64,
pub received_bytes_per_sec: u64,
pub transmitted_bytes_per_sec: u64,
pub total_received_packets: u64,
pub total_transmitted_packets: u64,
pub received_packets_per_sec: u64,
pub transmitted_packets_per_sec: u64,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[cfg_attr(feature = "pyo3", pyo3::prelude::pyclass)]
pub struct RiSystemMetrics {
pub cpu: RiCPUMetrics,
pub memory: RiMemoryMetrics,
pub disk: RiDiskMetrics,
pub network: RiNetworkMetrics,
pub timestamp: u64,
}
#[allow(dead_code)]
#[cfg_attr(feature = "pyo3", pyo3::prelude::pyclass)]
pub struct RiSystemMetricsCollector {
system: System,
last_network_received: u64,
last_network_transmitted: u64,
last_network_received_packets: u64,
last_network_transmitted_packets: u64,
last_network_time: Instant,
}
#[allow(dead_code)]
impl Default for RiSystemMetricsCollector {
fn default() -> Self {
Self::new()
}
}
#[allow(dead_code)]
impl RiSystemMetricsCollector {
pub fn new() -> Self {
let mut system = System::new_all();
system.refresh_all();
let (received_bytes, transmitted_bytes, received_packets, transmitted_packets) =
Self::get_network_total(&system);
Self {
system,
last_network_received: received_bytes,
last_network_transmitted: transmitted_bytes,
last_network_received_packets: received_packets,
last_network_transmitted_packets: transmitted_packets,
last_network_time: Instant::now(),
}
}
pub fn collect(&mut self) -> RiSystemMetrics {
self.system.refresh_all();
let cpu = self.get_cpu_metrics();
let memory = self.get_memory_metrics();
let disk = self.get_disk_metrics();
let network = self.get_network_metrics();
let timestamp = chrono::Utc::now().timestamp_millis() as u64;
RiSystemMetrics {
cpu,
memory,
disk,
network,
timestamp,
}
}
fn get_cpu_metrics(&self) -> RiCPUMetrics {
let total_usage = self.system.global_cpu_info().cpu_usage();
let per_core_usage: Vec<f64> = self
.system
.cpus()
.iter()
.map(|cpu| cpu.cpu_usage() as f64)
.collect();
RiCPUMetrics {
total_usage_percent: total_usage as f64,
per_core_usage,
context_switches: 0,
interrupts: 0,
}
}
fn get_memory_metrics(&self) -> RiMemoryMetrics {
let total = self.system.total_memory();
let used = self.system.used_memory();
let free = self.system.free_memory();
let usage_percent = (used as f64 / total as f64) * 100.0;
let swap_total = self.system.total_swap();
let swap_used = self.system.used_swap();
let swap_free = self.system.free_swap();
let swap_usage_percent = if swap_total > 0 {
(swap_used as f64 / swap_total as f64) * 100.0
} else {
0.0
};
RiMemoryMetrics {
total_bytes: total,
used_bytes: used,
free_bytes: free,
usage_percent,
swap_total_bytes: swap_total,
swap_used_bytes: swap_used,
swap_free_bytes: swap_free,
swap_usage_percent,
}
}
fn get_disk_metrics(&self) -> RiDiskMetrics {
if let Some(disk) = self.system.disks().first() {
let total = disk.total_space();
let available = disk.available_space();
let used = total - available;
let usage_percent = (used as f64 / total as f64) * 100.0;
RiDiskMetrics {
total_bytes: total,
used_bytes: used,
free_bytes: available,
usage_percent,
read_bytes: 0,
write_bytes: 0,
read_count: 0,
write_count: 0,
}
} else {
RiDiskMetrics {
total_bytes: 0,
used_bytes: 0,
free_bytes: 0,
usage_percent: 0.0,
read_bytes: 0,
write_bytes: 0,
read_count: 0,
write_count: 0,
}
}
}
fn get_network_metrics(&mut self) -> RiNetworkMetrics {
let (received_bytes, transmitted_bytes, received_packets, transmitted_packets) =
Self::get_network_total(&self.system);
let now = Instant::now();
let elapsed = now
.duration_since(self.last_network_time)
.as_secs_f64()
.max(1.0);
let received_bytes_per_sec =
((received_bytes - self.last_network_received) as f64 / elapsed) as u64;
let transmitted_bytes_per_sec =
((transmitted_bytes - self.last_network_transmitted) as f64 / elapsed) as u64;
let received_packets_per_sec =
((received_packets - self.last_network_received_packets) as f64 / elapsed) as u64;
let transmitted_packets_per_sec =
((transmitted_packets - self.last_network_transmitted_packets) as f64 / elapsed) as u64;
self.last_network_received = received_bytes;
self.last_network_transmitted = transmitted_bytes;
self.last_network_received_packets = received_packets;
self.last_network_transmitted_packets = transmitted_packets;
self.last_network_time = now;
RiNetworkMetrics {
total_received_bytes: received_bytes,
total_transmitted_bytes: transmitted_bytes,
received_bytes_per_sec,
transmitted_bytes_per_sec,
total_received_packets: received_packets,
total_transmitted_packets: transmitted_packets,
received_packets_per_sec,
transmitted_packets_per_sec,
}
}
fn get_network_total(system: &System) -> (u64, u64, u64, u64) {
let mut received_bytes = 0;
let mut transmitted_bytes = 0;
let mut received_packets = 0;
let mut transmitted_packets = 0;
for (_, data) in system.networks() {
received_bytes += data.received();
transmitted_bytes += data.transmitted();
received_packets += data.packets_received();
transmitted_packets += data.packets_transmitted();
}
(
received_bytes,
transmitted_bytes,
received_packets,
transmitted_packets,
)
}
}
#[cfg(feature = "pyo3")]
#[pyo3::prelude::pymethods]
impl RiSystemMetricsCollector {
#[new]
fn py_new() -> Self {
Self::new()
}
#[pyo3(name = "collect")]
fn py_collect(&mut self) -> RiSystemMetrics {
self.collect()
}
#[pyo3(name = "refresh")]
fn py_refresh(&mut self) {
self.system.refresh_all();
}
}
#[cfg(feature = "pyo3")]
#[pyo3::prelude::pymethods]
impl RiSystemMetrics {
#[pyo3(name = "cpu")]
fn py_cpu(&self) -> RiCPUMetrics {
self.cpu.clone()
}
#[pyo3(name = "memory")]
fn py_memory(&self) -> RiMemoryMetrics {
self.memory.clone()
}
#[pyo3(name = "disk")]
fn py_disk(&self) -> RiDiskMetrics {
self.disk.clone()
}
#[pyo3(name = "network")]
fn py_network(&self) -> RiNetworkMetrics {
self.network.clone()
}
#[pyo3(name = "timestamp")]
fn py_timestamp(&self) -> u64 {
self.timestamp
}
}