use once_cell::sync::Lazy;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
pub static GLOBAL_METRICS: Lazy<Arc<Metrics>> = Lazy::new(|| Arc::new(Metrics::new()));
pub fn global_metrics() -> Arc<Metrics> {
Arc::clone(&GLOBAL_METRICS)
}
#[derive(Debug)]
pub struct Metrics {
blocks_created: AtomicUsize,
cids_generated: AtomicUsize,
blocks_verified: AtomicUsize,
chunks_created: AtomicUsize,
total_bytes_processed: AtomicU64,
errors_total: AtomicUsize,
serialization_errors: AtomicUsize,
validation_errors: AtomicUsize,
network_errors: AtomicUsize,
timings: Mutex<TimingStats>,
memory_allocations: AtomicU64,
pool_hits: AtomicUsize,
pool_misses: AtomicUsize,
start_time: Instant,
}
#[derive(Debug, Clone)]
struct TimingStats {
cid_generation_samples: Vec<u64>, block_creation_samples: Vec<u64>, chunking_samples: Vec<u64>, verification_samples: Vec<u64>, max_samples: usize, }
impl Default for TimingStats {
fn default() -> Self {
Self {
cid_generation_samples: Vec::with_capacity(10000),
block_creation_samples: Vec::with_capacity(10000),
chunking_samples: Vec::with_capacity(1000),
verification_samples: Vec::with_capacity(10000),
max_samples: 10000,
}
}
}
impl TimingStats {
fn add_sample(samples: &mut Vec<u64>, value: u64, max_samples: usize) {
if samples.len() >= max_samples {
samples.drain(0..max_samples / 4);
}
samples.push(value);
}
fn percentile(sorted_samples: &[u64], p: f64) -> u64 {
if sorted_samples.is_empty() {
return 0;
}
let idx = ((sorted_samples.len() as f64 - 1.0) * p) as usize;
sorted_samples[idx]
}
fn get_percentiles(samples: &[u64]) -> PercentileStats {
if samples.is_empty() {
return PercentileStats::default();
}
let mut sorted = samples.to_vec();
sorted.sort_unstable();
PercentileStats {
p50: Self::percentile(&sorted, 0.50),
p90: Self::percentile(&sorted, 0.90),
p95: Self::percentile(&sorted, 0.95),
p99: Self::percentile(&sorted, 0.99),
min: sorted[0],
max: sorted[sorted.len() - 1],
}
}
}
#[derive(Debug, Clone, Default)]
pub struct PercentileStats {
pub p50: u64,
pub p90: u64,
pub p95: u64,
pub p99: u64,
pub min: u64,
pub max: u64,
}
#[derive(Debug, Clone)]
pub struct MetricsSnapshot {
pub blocks_created: usize,
pub cids_generated: usize,
pub blocks_verified: usize,
pub chunks_created: usize,
pub total_bytes_processed: u64,
pub errors_total: usize,
pub serialization_errors: usize,
pub validation_errors: usize,
pub network_errors: usize,
pub cid_generation: PercentileStats,
pub block_creation: PercentileStats,
pub chunking: PercentileStats,
pub verification: PercentileStats,
pub avg_cid_generation_us: f64,
pub avg_block_size_bytes: f64,
pub throughput_bytes_per_sec: f64,
pub memory_allocations: u64,
pub pool_hit_rate: f64,
pub uptime_seconds: u64,
}
impl Metrics {
pub fn new() -> Self {
Self {
blocks_created: AtomicUsize::new(0),
cids_generated: AtomicUsize::new(0),
blocks_verified: AtomicUsize::new(0),
chunks_created: AtomicUsize::new(0),
total_bytes_processed: AtomicU64::new(0),
errors_total: AtomicUsize::new(0),
serialization_errors: AtomicUsize::new(0),
validation_errors: AtomicUsize::new(0),
network_errors: AtomicUsize::new(0),
timings: Mutex::new(TimingStats::default()),
memory_allocations: AtomicU64::new(0),
pool_hits: AtomicUsize::new(0),
pool_misses: AtomicUsize::new(0),
start_time: Instant::now(),
}
}
pub fn record_block_created(&self, size_bytes: u64) {
self.blocks_created.fetch_add(1, Ordering::Relaxed);
self.total_bytes_processed
.fetch_add(size_bytes, Ordering::Relaxed);
}
pub fn record_block_created_timed(&self, size_bytes: u64, duration_us: u64) {
self.record_block_created(size_bytes);
if let Ok(mut timings) = self.timings.lock() {
let max_samples = timings.max_samples;
TimingStats::add_sample(
&mut timings.block_creation_samples,
duration_us,
max_samples,
);
}
}
pub fn record_cid_generated(&self, duration_us: u64) {
self.cids_generated.fetch_add(1, Ordering::Relaxed);
if let Ok(mut timings) = self.timings.lock() {
let max_samples = timings.max_samples;
TimingStats::add_sample(
&mut timings.cid_generation_samples,
duration_us,
max_samples,
);
}
}
pub fn record_block_verified(&self, duration_us: u64) {
self.blocks_verified.fetch_add(1, Ordering::Relaxed);
if let Ok(mut timings) = self.timings.lock() {
let max_samples = timings.max_samples;
TimingStats::add_sample(&mut timings.verification_samples, duration_us, max_samples);
}
}
pub fn record_chunking(&self, num_chunks: usize, duration_us: u64) {
self.chunks_created.fetch_add(num_chunks, Ordering::Relaxed);
if let Ok(mut timings) = self.timings.lock() {
let max_samples = timings.max_samples;
TimingStats::add_sample(&mut timings.chunking_samples, duration_us, max_samples);
}
}
pub fn record_serialization_error(&self) {
self.errors_total.fetch_add(1, Ordering::Relaxed);
self.serialization_errors.fetch_add(1, Ordering::Relaxed);
}
pub fn record_validation_error(&self) {
self.errors_total.fetch_add(1, Ordering::Relaxed);
self.validation_errors.fetch_add(1, Ordering::Relaxed);
}
pub fn record_network_error(&self) {
self.errors_total.fetch_add(1, Ordering::Relaxed);
self.network_errors.fetch_add(1, Ordering::Relaxed);
}
pub fn record_memory_allocation(&self, bytes: u64) {
self.memory_allocations.fetch_add(bytes, Ordering::Relaxed);
}
pub fn record_pool_hit(&self) {
self.pool_hits.fetch_add(1, Ordering::Relaxed);
}
pub fn record_pool_miss(&self) {
self.pool_misses.fetch_add(1, Ordering::Relaxed);
}
pub fn snapshot(&self) -> MetricsSnapshot {
let blocks_created = self.blocks_created.load(Ordering::Relaxed);
let cids_generated = self.cids_generated.load(Ordering::Relaxed);
let total_bytes = self.total_bytes_processed.load(Ordering::Relaxed);
let pool_hits = self.pool_hits.load(Ordering::Relaxed);
let pool_misses = self.pool_misses.load(Ordering::Relaxed);
let timings = self.timings.lock().unwrap_or_else(|e| e.into_inner());
let avg_block_size = if blocks_created > 0 {
total_bytes as f64 / blocks_created as f64
} else {
0.0
};
let uptime_seconds = self.start_time.elapsed().as_secs();
let throughput = if uptime_seconds > 0 {
total_bytes as f64 / uptime_seconds as f64
} else {
0.0
};
let avg_cid_gen = if !timings.cid_generation_samples.is_empty() {
timings.cid_generation_samples.iter().sum::<u64>() as f64
/ timings.cid_generation_samples.len() as f64
} else {
0.0
};
let pool_total = pool_hits + pool_misses;
let hit_rate = if pool_total > 0 {
pool_hits as f64 / pool_total as f64
} else {
0.0
};
MetricsSnapshot {
blocks_created,
cids_generated,
blocks_verified: self.blocks_verified.load(Ordering::Relaxed),
chunks_created: self.chunks_created.load(Ordering::Relaxed),
total_bytes_processed: total_bytes,
errors_total: self.errors_total.load(Ordering::Relaxed),
serialization_errors: self.serialization_errors.load(Ordering::Relaxed),
validation_errors: self.validation_errors.load(Ordering::Relaxed),
network_errors: self.network_errors.load(Ordering::Relaxed),
cid_generation: TimingStats::get_percentiles(&timings.cid_generation_samples),
block_creation: TimingStats::get_percentiles(&timings.block_creation_samples),
chunking: TimingStats::get_percentiles(&timings.chunking_samples),
verification: TimingStats::get_percentiles(&timings.verification_samples),
avg_cid_generation_us: avg_cid_gen,
avg_block_size_bytes: avg_block_size,
throughput_bytes_per_sec: throughput,
memory_allocations: self.memory_allocations.load(Ordering::Relaxed),
pool_hit_rate: hit_rate,
uptime_seconds,
}
}
pub fn reset(&self) {
self.blocks_created.store(0, Ordering::Relaxed);
self.cids_generated.store(0, Ordering::Relaxed);
self.blocks_verified.store(0, Ordering::Relaxed);
self.chunks_created.store(0, Ordering::Relaxed);
self.total_bytes_processed.store(0, Ordering::Relaxed);
self.errors_total.store(0, Ordering::Relaxed);
self.serialization_errors.store(0, Ordering::Relaxed);
self.validation_errors.store(0, Ordering::Relaxed);
self.network_errors.store(0, Ordering::Relaxed);
self.memory_allocations.store(0, Ordering::Relaxed);
self.pool_hits.store(0, Ordering::Relaxed);
self.pool_misses.store(0, Ordering::Relaxed);
if let Ok(mut timings) = self.timings.lock() {
timings.cid_generation_samples.clear();
timings.block_creation_samples.clear();
timings.chunking_samples.clear();
timings.verification_samples.clear();
}
}
pub fn is_healthy(&self) -> bool {
let snapshot = self.snapshot();
let total_ops = snapshot.blocks_created + snapshot.cids_generated;
if total_ops > 0 {
let error_rate = snapshot.errors_total as f64 / total_ops as f64;
if error_rate > 0.10 {
return false;
}
}
if snapshot.cid_generation.p99 > 10_000 {
return false;
}
true
}
}
impl Default for Metrics {
fn default() -> Self {
Self::new()
}
}
pub struct Timer {
start: Instant,
}
impl Timer {
pub fn start() -> Self {
Self {
start: Instant::now(),
}
}
pub fn elapsed_us(&self) -> u64 {
self.start.elapsed().as_micros() as u64
}
pub fn elapsed(&self) -> Duration {
self.start.elapsed()
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_metrics_basic() {
let metrics = Metrics::new();
metrics.record_block_created(1024);
metrics.record_cid_generated(100);
let snapshot = metrics.snapshot();
assert_eq!(snapshot.blocks_created, 1);
assert_eq!(snapshot.cids_generated, 1);
assert_eq!(snapshot.total_bytes_processed, 1024);
}
#[test]
fn test_metrics_timing() {
let metrics = Metrics::new();
metrics.record_cid_generated(100);
metrics.record_cid_generated(200);
metrics.record_cid_generated(300);
let snapshot = metrics.snapshot();
assert_eq!(snapshot.cids_generated, 3);
assert_eq!(snapshot.cid_generation.min, 100);
assert_eq!(snapshot.cid_generation.max, 300);
}
#[test]
fn test_metrics_errors() {
let metrics = Metrics::new();
metrics.record_serialization_error();
metrics.record_validation_error();
metrics.record_network_error();
let snapshot = metrics.snapshot();
assert_eq!(snapshot.errors_total, 3);
assert_eq!(snapshot.serialization_errors, 1);
assert_eq!(snapshot.validation_errors, 1);
assert_eq!(snapshot.network_errors, 1);
}
#[test]
fn test_metrics_pool_stats() {
let metrics = Metrics::new();
metrics.record_pool_hit();
metrics.record_pool_hit();
metrics.record_pool_miss();
let snapshot = metrics.snapshot();
assert_eq!(snapshot.pool_hit_rate, 2.0 / 3.0);
}
#[test]
fn test_metrics_reset() {
let metrics = Metrics::new();
metrics.record_block_created(1024);
metrics.record_cid_generated(100);
metrics.reset();
let snapshot = metrics.snapshot();
assert_eq!(snapshot.blocks_created, 0);
assert_eq!(snapshot.cids_generated, 0);
}
#[test]
fn test_percentile_calculation() {
let metrics = Metrics::new();
for i in 1..=100 {
metrics.record_cid_generated(i * 10);
}
let snapshot = metrics.snapshot();
assert!(snapshot.cid_generation.p50 > 0);
assert!(snapshot.cid_generation.p90 > snapshot.cid_generation.p50);
assert!(snapshot.cid_generation.p99 > snapshot.cid_generation.p90);
}
#[test]
fn test_timer() {
let timer = Timer::start();
std::thread::sleep(Duration::from_micros(100));
let elapsed = timer.elapsed_us();
assert!(elapsed >= 100);
}
#[test]
fn test_health_check() {
let metrics = Metrics::new();
for _ in 0..100 {
metrics.record_block_created(1024);
metrics.record_cid_generated(100);
}
assert!(metrics.is_healthy());
for _ in 0..50 {
metrics.record_validation_error();
}
assert!(!metrics.is_healthy());
}
#[test]
fn test_global_metrics() {
let metrics = global_metrics();
metrics.record_block_created(2048);
let snapshot = metrics.snapshot();
assert!(snapshot.blocks_created > 0);
}
#[test]
fn test_throughput_calculation() {
let metrics = Metrics::new();
std::thread::sleep(Duration::from_millis(10));
metrics.record_block_created(1_000_000);
std::thread::sleep(Duration::from_millis(100));
let snapshot = metrics.snapshot();
assert!(snapshot.uptime_seconds > 0 || snapshot.throughput_bytes_per_sec >= 0.0);
}
#[test]
fn test_avg_block_size() {
let metrics = Metrics::new();
metrics.record_block_created(1000);
metrics.record_block_created(2000);
metrics.record_block_created(3000);
let snapshot = metrics.snapshot();
assert_eq!(snapshot.avg_block_size_bytes, 2000.0);
}
}