//! Aggregate metrics: no attacker-controlled labels or unbounded series. use serde::Serialize; use std::sync::atomic::{AtomicU64, Ordering}; #[derive(Default)] pub struct Metrics { pub admin_requests: AtomicU64, pub admin_response_headers: AtomicU64, pub admin_header_duration_us: AtomicU64, pub admin_status: [AtomicU64; 5], pub compression_gzip: AtomicU64, pub compression_br: AtomicU64, pub compression_zstd: AtomicU64, pub compression_deflate: AtomicU64, pub status_1xx: AtomicU64, pub http1_requests: AtomicU64, pub http3_requests: AtomicU64, pub http3_connections: AtomicU64, pub http3_active_connections: AtomicU64, pub http3_rejected_connections: AtomicU64, pub http2_requests: AtomicU64, pub http2_connections: AtomicU64, pub http2_active_connections: AtomicU64, pub tls_connections: AtomicU64, pub tls_handshake_errors: AtomicU64, pub websocket_total: AtomicU64, pub websocket_active: AtomicU64, pub websocket_h1_total: AtomicU64, pub websocket_h1_active: AtomicU64, pub websocket_h2_total: AtomicU64, pub websocket_h2_active: AtomicU64, pub websocket_h3_total: AtomicU64, pub websocket_h3_active: AtomicU64, pub websocket_closed: AtomicU64, pub websocket_errors: AtomicU64, pub websocket_handshake_errors: AtomicU64, pub websocket_bytes_in: AtomicU64, pub websocket_bytes_out: AtomicU64, pub websocket_duration_us: AtomicU64, pub requests: AtomicU64, pub completed: AtomicU64, pub active: AtomicU64, pub errors: AtomicU64, pub rejected_requests: AtomicU64, pub connections: AtomicU64, pub active_connections: AtomicU64, pub rejected_connections: AtomicU64, pub bytes_in: AtomicU64, pub bytes_out: AtomicU64, pub duration_us: AtomicU64, pub status_2xx: AtomicU64, pub status_3xx: AtomicU64, pub status_4xx: AtomicU64, pub status_5xx: AtomicU64, } #[derive(Serialize)] pub struct View { pub http1_requests: u64, pub http3_requests: u64, pub http3_connections: u64, pub http3_active_connections: u64, pub http3_rejected_connections: u64, pub compression_gzip: u64, pub compression_br: u64, pub compression_zstd: u64, pub compression_deflate: u64, pub http2_requests: u64, pub http2_connections: u64, pub http2_active_connections: u64, pub tls_connections: u64, pub tls_handshake_errors: u64, pub websocket_total: u64, pub websocket_active: u64, pub websocket_h1_total: u64, pub websocket_h1_active: u64, pub websocket_h2_total: u64, pub websocket_h2_active: u64, pub websocket_h3_total: u64, pub websocket_h3_active: u64, pub websocket_errors: u64, pub websocket_handshake_errors: u64, pub websocket_bytes_in: u64, pub websocket_bytes_out: u64, pub requests: u64, pub completed: u64, pub active: u64, pub errors: u64, pub rejected_requests: u64, pub connections: u64, pub active_connections: u64, pub rejected_connections: u64, pub bytes_in: u64, pub bytes_out: u64, pub mean_duration_ms: f64, } fn n(value: &AtomicU64) -> u64 { value.load(Ordering::Relaxed) } impl Metrics { pub fn record_compression(&self, encoding: crate::compression::AppliedEncoding) { let counter = match encoding { crate::compression::AppliedEncoding::Gzip => &self.compression_gzip, crate::compression::AppliedEncoding::Brotli => &self.compression_br, crate::compression::AppliedEncoding::Zstd => &self.compression_zstd, crate::compression::AppliedEncoding::Deflate => &self.compression_deflate, }; counter.fetch_add(1, Ordering::Relaxed); } pub fn view(&self) -> View { let completed = n(&self.completed); View { http1_requests: n(&self.http1_requests), http3_requests: n(&self.http3_requests), http3_connections: n(&self.http3_connections), http3_active_connections: n(&self.http3_active_connections), http3_rejected_connections: n(&self.http3_rejected_connections), compression_gzip: n(&self.compression_gzip), compression_br: n(&self.compression_br), compression_zstd: n(&self.compression_zstd), compression_deflate: n(&self.compression_deflate), http2_requests: n(&self.http2_requests), http2_connections: n(&self.http2_connections), http2_active_connections: n(&self.http2_active_connections), tls_connections: n(&self.tls_connections), tls_handshake_errors: n(&self.tls_handshake_errors), websocket_total: n(&self.websocket_total), websocket_active: n(&self.websocket_active), websocket_h1_total: n(&self.websocket_h1_total), websocket_h1_active: n(&self.websocket_h1_active), websocket_h2_total: n(&self.websocket_h2_total), websocket_h2_active: n(&self.websocket_h2_active), websocket_h3_total: n(&self.websocket_h3_total), websocket_h3_active: n(&self.websocket_h3_active), websocket_errors: n(&self.websocket_errors), websocket_handshake_errors: n(&self.websocket_handshake_errors), websocket_bytes_in: n(&self.websocket_bytes_in), websocket_bytes_out: n(&self.websocket_bytes_out), requests: n(&self.requests), completed, active: n(&self.active), errors: n(&self.errors), rejected_requests: n(&self.rejected_requests), connections: n(&self.connections), active_connections: n(&self.active_connections), rejected_connections: n(&self.rejected_connections), bytes_in: n(&self.bytes_in), bytes_out: n(&self.bytes_out), mean_duration_ms: if completed > 0 { n(&self.duration_us) as f64 / completed as f64 / 1000.0 } else { 0.0 }, } } pub fn prometheus(&self, logs_dropped: u64, revision: u64) -> String { let mut out = String::new(); let entries = [ ( "http1_requests_total", "counter", "Public requests received using HTTP/1.1.", n(&self.http1_requests), ), ( "http3_requests_total", "counter", "HTTP/3 requests", n(&self.http3_requests), ), ( "http3_connections_total", "counter", "HTTP/3 connections", n(&self.http3_connections), ), ( "http3_active_connections", "gauge", "HTTP/3 active connections", n(&self.http3_active_connections), ), ( "http3_rejected_connections_total", "counter", "HTTP/3 rejected connections", n(&self.http3_rejected_connections), ), ( "compression_gzip_total", "counter", "Gzip encoded responses.", n(&self.compression_gzip), ), ( "compression_br_total", "counter", "Brotli encoded responses.", n(&self.compression_br), ), ( "compression_zstd_total", "counter", "Zstd encoded responses.", n(&self.compression_zstd), ), ( "compression_deflate_total", "counter", "Deflate encoded responses.", n(&self.compression_deflate), ), ( "http2_requests_total", "counter", "Public requests received using HTTP/2.", n(&self.http2_requests), ), ( "http2_connections_total", "counter", "Public HTTP/2 connections over TLS or opt-in cleartext h2c prior knowledge.", n(&self.http2_connections), ), ( "http2_active_connections", "gauge", "Currently open public HTTP/2 connections.", n(&self.http2_active_connections), ), ( "tls_connections_total", "counter", "Successful public TLS handshakes.", n(&self.tls_connections), ), ( "tls_handshake_errors_total", "counter", "Failed or timed-out public TLS handshakes.", n(&self.tls_handshake_errors), ), ( "websocket_tunnels_total", "counter", "Established WebSocket tunnels across HTTP/1.1, HTTP/2 RFC 8441 and HTTP/3 RFC 9220.", n(&self.websocket_total), ), ( "websocket_tunnels_active", "gauge", "Open WebSocket tunnels, counted until byte transport ends.", n(&self.websocket_active), ), ( "websocket_h1_tunnels_total", "counter", "Established HTTP/1.1 Upgrade WebSocket tunnels.", n(&self.websocket_h1_total), ), ( "websocket_h1_tunnels_active", "gauge", "Active HTTP/1.1 Upgrade WebSocket tunnels.", n(&self.websocket_h1_active), ), ( "websocket_h2_rfc8441_tunnels_total", "counter", "Established HTTP/2 RFC 8441 WebSocket tunnels.", n(&self.websocket_h2_total), ), ( "websocket_h2_rfc8441_tunnels_active", "gauge", "Active HTTP/2 RFC 8441 WebSocket tunnels.", n(&self.websocket_h2_active), ), ( "websocket_h3_rfc9220_tunnels_total", "counter", "Established HTTP/3 RFC 9220 WebSocket tunnels.", n(&self.websocket_h3_total), ), ( "websocket_h3_rfc9220_tunnels_active", "gauge", "Active HTTP/3 RFC 9220 WebSocket tunnels.", n(&self.websocket_h3_active), ), ( "websocket_errors_total", "counter", "Upgrade transport, I/O or idle-timeout failures.", n(&self.websocket_errors), ), ( "websocket_handshake_errors_total", "counter", "WebSocket upgrade requests not answered with 101.", n(&self.websocket_handshake_errors), ), ( "websocket_client_bytes_total", "counter", "Client WebSocket frame bytes, including framing but excluding TLS/TCP.", n(&self.websocket_bytes_in), ), ( "websocket_backend_bytes_total", "counter", "Backend WebSocket frame bytes, including framing but excluding TLS/TCP.", n(&self.websocket_bytes_out), ), ( "requests_total", "counter", "Requests entering the public proxy, including capacity rejects.", n(&self.requests), ), ( "requests_completed_total", "counter", "Admitted exchanges released after their bodies finish or drop.", n(&self.completed), ), ( "requests_active", "gauge", "Admitted exchanges with a live request or response body.", n(&self.active), ), ( "errors_total", "counter", "Completed 5xx, timed-out, failed or aborted exchanges.", n(&self.errors), ), ( "requests_rejected_total", "counter", "Requests rejected by the in-flight limit.", n(&self.rejected_requests), ), ( "connections_total", "counter", "Accepted public TCP/QUIC connections.", n(&self.connections), ), ( "connections_active", "gauge", "Open public TCP/QUIC connections.", n(&self.active_connections), ), ( "connections_rejected_total", "counter", "Public TCP/QUIC connections rejected at the connection limit.", n(&self.rejected_connections), ), ( "request_body_bytes_total", "counter", "Request payload bytes pulled through the proxy, not wire bytes.", n(&self.bytes_in), ), ( "response_body_bytes_total", "counter", "Response payload bytes yielded by the proxy, not acknowledged wire bytes.", n(&self.bytes_out), ), ( "logs_dropped_total", "counter", "Access entries dropped due to contention, excluding ring eviction.", logs_dropped, ), ( "config_revision", "gauge", "Currently applied configuration revision.", revision, ), ]; for (name, kind, help, value) in entries { out.push_str(&format!("# HELP rust_proxy_{name} {help}\n# TYPE rust_proxy_{name} {kind}\nrust_proxy_{name} {value}\n")); } out.push_str("# HELP rust_proxy_exchange_duration_seconds Lifetime through body completion or cancellation.\n# TYPE rust_proxy_exchange_duration_seconds summary\n"); out.push_str(&format!("rust_proxy_exchange_duration_seconds_count {}\nrust_proxy_exchange_duration_seconds_sum {:.6}\n", n(&self.completed), n(&self.duration_us) as f64 / 1_000_000.0)); out.push_str("# HELP rust_proxy_responses_total Completed responses by status class.\n# TYPE rust_proxy_responses_total counter\n"); for (class, value) in [ ("1xx", &self.status_1xx), ("2xx", &self.status_2xx), ("3xx", &self.status_3xx), ("4xx", &self.status_4xx), ("5xx", &self.status_5xx), ] { out.push_str(&format!( "rust_proxy_responses_total{{class=\"{class}\"}} {}\n", n(value) )); } out.push_str("# HELP rust_proxy_websocket_duration_seconds Lifetime of closed WebSocket byte transports.\n# TYPE rust_proxy_websocket_duration_seconds summary\n"); out.push_str(&format!("rust_proxy_websocket_duration_seconds_count {}\nrust_proxy_websocket_duration_seconds_sum {:.6}\n", n(&self.websocket_closed), n(&self.websocket_duration_us) as f64 / 1_000_000.0)); out } } /// Fixed-cardinality per-vhost counters. Retained across config updates by ID, /// released when a vhost is removed, and finalized after complete body lifetime. pub const DURATION_BUCKETS_US: [u64; 10] = [ 5_000, 10_000, 25_000, 50_000, 100_000, 250_000, 500_000, 1_000_000, 2_500_000, 5_000_000, ]; #[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 status: [AtomicU64; 5], pub buckets: [AtomicU64; 11], } 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); for (index, threshold) in DURATION_BUCKETS_US.iter().enumerate() { if micros <= *threshold { self.buckets[index].fetch_add(1, Ordering::Relaxed); } } self.buckets[DURATION_BUCKETS_US.len()].fetch_add(1, Ordering::Relaxed); } } #[cfg(test)] mod vhost_tests { use super::*; #[test] fn streaming_completion_and_histogram() { let m = VhostMetrics::default(); m.active.store(1, Ordering::Relaxed); m.complete(503, false, std::time::Duration::from_millis(20), 10, 50); assert_eq!(n(&m.active), 0); assert_eq!(n(&m.errors), 1); assert_eq!(n(&m.status[4]), 1); assert_eq!(n(&m.buckets[1]), 0); assert_eq!(n(&m.buckets[2]), 1); assert_eq!(n(&m.buckets[10]), 1); } } #[cfg(test)] mod compression_counter_tests { use super::*; use crate::compression::AppliedEncoding; #[test] fn each_codec_has_an_independent_counter() { let metrics = Metrics::default(); for encoding in [ AppliedEncoding::Gzip, AppliedEncoding::Brotli, AppliedEncoding::Zstd, AppliedEncoding::Deflate, ] { metrics.record_compression(encoding); } let view = metrics.view(); assert_eq!(view.compression_gzip, 1); assert_eq!(view.compression_br, 1); assert_eq!(view.compression_zstd, 1); assert_eq!(view.compression_deflate, 1); } }