use std::{ sync::atomic::{AtomicU32, Ordering}, time::Duration, }; use lazy_static::lazy_static; use prometheus::{Histogram, HistogramVec, IntCounterVec, IntGauge, IntGaugeVec}; use tici_common::metrics::{ OperationMetrics, OperationMetricsGuard, begin_operation_metrics1, register_histogram_with_buckets, register_histogram_with_labels_and_buckets, register_int_counter_with_labels, register_int_gauge, register_int_gauge_with_labels, }; use tici_system::{cpu_time::ProcessStat, get_global_memory_usage}; /// Global counter for requests processed in current heartbeat cycle static REQUESTS_IN_CYCLE: AtomicU32 = AtomicU32::new(0); /// 0.01s, 0.02s, 0.04s, ..., 81.92s. fn reader_warmup_duration_buckets() -> Vec { let mut buckets = Vec::with_capacity(14); let mut current = 0.01; for _ in 0..14 { buckets.push(current); current *= 2.0; } buckets } fn exponential_bucket_values(start: f64, factor: f64, count: usize) -> Vec { let mut buckets = Vec::with_capacity(count); let mut value = start; for _ in 0..count { buckets.push(value); value *= factor; } buckets } lazy_static! { pub static ref READER_WARMUP_DURATION_HISTOGRAM: HistogramVec = register_histogram_with_labels_and_buckets( "tici_reader_warmup_duration", "Duration of reader warmup stages in seconds", &["stage"], reader_warmup_duration_buckets(), ); pub static ref READER_SHARD_REFRESH_DURATION_HISTOGRAM: HistogramVec = register_histogram_with_labels_and_buckets( "tici_reader_shard_refresh_duration", "Duration of reader shard refresh work", &["type"], exponential_bucket_values(0.01, 2.0, 12), ); pub static ref READER_SHARD_REFRESH_COUNTER: IntCounterVec = register_int_counter_with_labels( "tici_reader_shard_refresh_count", "Count of reader shard refresh work", &["type", "result"], ); pub static ref READER_SHARD_REFRESH_INFLIGHT_GAUGE: IntGaugeVec = register_int_gauge_with_labels( "tici_reader_shard_refresh_inflight", "Inflight count of reader shard refresh work", &["type"], ); pub static ref READER_SHARD_REFRESH_QUEUE_DURATION_HISTOGRAM: Histogram = register_histogram_with_buckets( "tici_reader_shard_refresh_queue_duration", "Queued time before reader shard refresh starts", exponential_bucket_values(0.001, 2.0, 16), ); pub static ref READER_HEARTBEAT_QUEUE_BACKLOG_GAUGE: IntGauge = register_int_gauge( "tici_reader_heartbeat_queue_backlog", "Number of shard refresh tasks waiting in the reader heartbeat response queue", ); pub static ref READER_SHARD_REFRESH_EVENT_COUNTER: IntCounterVec = register_int_counter_with_labels( "tici_reader_shard_refresh_event_count", "Count of reader shard refresh events", &["type"], ); } pub fn observe_reader_warmup_duration(stage: &'static str, elapsed: Duration) { READER_WARMUP_DURATION_HISTOGRAM .with_label_values(&[stage]) .observe(elapsed.as_secs_f64()); } /// System metrics collector for reader nodes pub struct SystemMetrics { process_stat: Option, } impl SystemMetrics { pub fn new() -> Self { Self { process_stat: ProcessStat::cur_proc_stat().ok(), } } /// Get current CPU usage percentage using tici_system CPU monitoring pub fn get_cpu_usage(&mut self) -> f64 { if let Some(ref mut process_stat) = self.process_stat { match process_stat.cpu_usage() { Ok(usage) => usage * 100.0, Err(_) => { // Try to reinitialize on error self.process_stat = ProcessStat::cur_proc_stat().ok(); 0.0 } } } else { // Try to initialize if not available self.process_stat = ProcessStat::cur_proc_stat().ok(); 0.0 } } /// Get current memory usage in bytes using tici_system memory monitoring pub fn get_memory_usage(&self) -> u64 { get_global_memory_usage() } /// Get current cache hit rate (placeholder for now) pub fn get_cache_hit_rate(&self) -> f64 { // TODO: Calculate based on actual cache statistics // For now, return a constant value 85.0 } } impl Default for SystemMetrics { fn default() -> Self { Self::new() } } pub fn begin_shard_refresh_metrics(refresh_type: &'static str) -> OperationMetricsGuard { begin_operation_metrics1( OperationMetrics::new( &READER_SHARD_REFRESH_DURATION_HISTOGRAM, &READER_SHARD_REFRESH_COUNTER, &READER_SHARD_REFRESH_INFLIGHT_GAUGE, ), refresh_type, ) } pub fn observe_shard_refresh_queue_duration(duration: Duration) { READER_SHARD_REFRESH_QUEUE_DURATION_HISTOGRAM.observe(duration.as_secs_f64()); } pub fn increment_heartbeat_queue_backlog() { READER_HEARTBEAT_QUEUE_BACKLOG_GAUGE.inc(); } pub fn decrement_heartbeat_queue_backlog() { READER_HEARTBEAT_QUEUE_BACKLOG_GAUGE.dec(); } pub fn reset_heartbeat_queue_backlog() { READER_HEARTBEAT_QUEUE_BACKLOG_GAUGE.set(0); } pub fn inc_shard_refresh_event(event_type: &'static str) { READER_SHARD_REFRESH_EVENT_COUNTER .with_label_values(&[event_type]) .inc(); } /// Increment the requests processed in current heartbeat cycle pub fn increment_requests_in_cycle() -> u32 { REQUESTS_IN_CYCLE.fetch_add(1, Ordering::Relaxed) + 1 } /// Get the current number of requests processed in this heartbeat cycle pub fn get_requests_in_cycle() -> u32 { REQUESTS_IN_CYCLE.load(Ordering::Relaxed) } /// Reset the requests counter (called after sending heartbeat) pub fn reset_requests_in_cycle() -> u32 { REQUESTS_IN_CYCLE.swap(0, Ordering::Relaxed) } #[cfg(test)] mod tests { use super::*; #[test] fn test_metrics_functionality() { // Reset counter for clean test REQUESTS_IN_CYCLE.store(0, Ordering::Relaxed); // Test requests counter assert_eq!(get_requests_in_cycle(), 0); let count1 = increment_requests_in_cycle(); assert_eq!(count1, 1); assert_eq!(get_requests_in_cycle(), 1); let count2 = increment_requests_in_cycle(); assert_eq!(count2, 2); assert_eq!(get_requests_in_cycle(), 2); // Test reset functionality let previous_count = reset_requests_in_cycle(); assert_eq!(previous_count, 2); assert_eq!(get_requests_in_cycle(), 0); // Test system metrics basic functionality let mut metrics = SystemMetrics::new(); let cpu_usage = metrics.get_cpu_usage(); let memory_usage = metrics.get_memory_usage(); let cache_hit_rate = metrics.get_cache_hit_rate(); assert!(cpu_usage >= 0.0); // memory_usage is u64, so it's always >= 0, just check it's a valid value let _ = memory_usage; assert_eq!(cache_hit_rate, 85.0); } }