#[derive(Default, Debug)] pub struct VhostMetrics { pub requests: AtomicU64, pub completed: AtomicU64, pub active: AtomicU64, pub rejected: AtomicU64, pub errors: AtomicU64, pub bytes_in: AtomicU64, pub bytes_out: AtomicU64, pub duration_us: AtomicU64, pub ewma_duration_us: CacheAlignedAtomicU64, pub status: [AtomicU64; 5], pub buckets: [AtomicU64; 11], } #[derive(Debug, Clone, Serialize)] pub struct VhostMetricsView { pub requests: u64, pub completed: u64, pub active: u64, pub rejected: u64, pub errors: u64, pub bytes_in: u64, pub bytes_out: u64, pub average_latency_ms: f64, pub ewma_latency_ms: f64, } impl VhostMetrics { pub fn complete( &self, status: u16, failed: bool, duration: std::time::Duration, bytes_in: u64, bytes_out: u64, ) { self.active.fetch_sub(1, Ordering::Relaxed); self.completed.fetch_add(1, Ordering::Relaxed); if failed || status >= 500 { self.errors.fetch_add(1, Ordering::Relaxed); } self.bytes_in.fetch_add(bytes_in, Ordering::Relaxed); self.bytes_out.fetch_add(bytes_out, Ordering::Relaxed); self.status[(status as usize / 100).clamp(1, 5) - 1].fetch_add(1, Ordering::Relaxed); let micros = duration.as_micros().min(u64::MAX as u128) as u64; self.duration_us.fetch_add(micros, Ordering::Relaxed); let mut current = self.ewma_duration_us.load(Ordering::Relaxed); loop { let next = if current == 0 { micros } else { current.saturating_mul(7).saturating_add(micros) / 8 }; match self.ewma_duration_us.compare_exchange_weak( current, next, Ordering::Relaxed, Ordering::Relaxed, ) { Ok(_) => break, Err(observed) => current = observed, } } // Write exactly one exclusive histogram bin per exchange. Prometheus // cumulative le=... buckets are reconstructed only when scraped. // This removes up to 11 contended atomics for short 1 KiB responses. let bucket = DURATION_BUCKETS_US.partition_point(|threshold| micros > *threshold); self.buckets[bucket].fetch_add(1, Ordering::Relaxed); } /// A scrape-consistent cumulative representation of exclusive bins. The /// sum of atomic snapshots is monotonic and the final count includes +Inf. pub fn duration_histogram(&self) -> ([u64; DURATION_BUCKETS_US.len()], u64) { let mut cumulative = [0u64; DURATION_BUCKETS_US.len()]; let mut count = 0u64; for (index, bin) in self.buckets.iter().enumerate() { count = count.saturating_add(bin.load(Ordering::Relaxed)); if index < DURATION_BUCKETS_US.len() { cumulative[index] = count; } } (cumulative, count) } pub fn view(&self) -> VhostMetricsView { let completed = n(&self.completed); let duration_us = n(&self.duration_us); VhostMetricsView { requests: n(&self.requests), completed, active: n(&self.active), rejected: n(&self.rejected), errors: n(&self.errors), bytes_in: n(&self.bytes_in), bytes_out: n(&self.bytes_out), average_latency_ms: if completed == 0 { 0.0 } else { duration_us as f64 / completed as f64 / 1000.0 }, ewma_latency_ms: n(&self.ewma_duration_us) as f64 / 1000.0, } } } #[cfg(test)] include!("vhost/tests.rs");