diff --git a/sgl-model-gateway/src/observability/inflight_tracker.rs b/sgl-model-gateway/src/observability/inflight_tracker.rs index 84abfc5b4..3cc61bb12 100644 --- a/sgl-model-gateway/src/observability/inflight_tracker.rs +++ b/sgl-model-gateway/src/observability/inflight_tracker.rs @@ -8,40 +8,15 @@ use std::{ use dashmap::DashMap; -use super::metrics::Metrics; +use super::gauge_histogram::{BucketBounds, GaugeHistogramHandle, GaugeHistogramVec}; use crate::policies::utils::PeriodicTask; -struct NumericalBuckets { - upper_bounds: &'static [u64], - le_labels: Vec<&'static str>, - gt_labels: Vec<&'static str>, -} - -impl NumericalBuckets { - fn new(upper_bounds: &'static [u64]) -> Self { - let leak_str = |n: u64| Box::leak(n.to_string().into_boxed_str()) as &'static str; - - let mut le_labels: Vec<&'static str> = upper_bounds.iter().map(|&b| leak_str(b)).collect(); - le_labels.push("+Inf"); - - let mut gt_labels: Vec<&'static str> = vec!["0"]; - gt_labels.extend(upper_bounds.iter().map(|&b| leak_str(b))); - - Self { - upper_bounds, - le_labels, - gt_labels, - } - } - - fn len(&self) -> usize { - self.le_labels.len() - } -} - -const AGE_BUCKET_BOUNDS: &[u64] = &[30, 60, 180, 300, 600, 1200, 3600, 7200, 14400, 28800, 86400]; -static AGE_BUCKETS: LazyLock = - LazyLock::new(|| NumericalBuckets::new(AGE_BUCKET_BOUNDS)); +static INFLIGHT_AGE_BOUNDS: BucketBounds<11> = + BucketBounds::new([30, 60, 180, 300, 600, 1200, 3600, 7200, 14400, 28800, 86400]); +static INFLIGHT_AGE_HISTOGRAM: GaugeHistogramVec<11> = + GaugeHistogramVec::new("smg_http_inflight_request_age_count", &INFLIGHT_AGE_BOUNDS); +static INFLIGHT_AGE_HANDLE: LazyLock = + LazyLock::new(|| INFLIGHT_AGE_HISTOGRAM.register_no_labels()); pub struct InFlightRequestTracker { requests: DashMap, @@ -84,33 +59,21 @@ impl InFlightRequestTracker { } pub fn compute_bucket_counts(&self) -> Vec { - let buckets = &*AGE_BUCKETS; + let ages = self.collect_ages(); + INFLIGHT_AGE_BOUNDS.compute_counts(&ages) + } + + fn collect_ages(&self) -> Vec { let now = Instant::now(); - let inf_idx = buckets.len() - 1; - - let mut counts = vec![0usize; buckets.len()]; - for entry in self.requests.iter() { - let age_secs = now.duration_since(*entry.value()).as_secs(); - let bucket_idx = buckets - .upper_bounds - .iter() - .position(|&bound| age_secs <= bound) - .unwrap_or(inf_idx); - counts[bucket_idx] += 1; - } - - counts + self.requests + .iter() + .map(|entry| now.duration_since(*entry.value()).as_secs()) + .collect() } fn sample_and_record(&self) { - let buckets = &*AGE_BUCKETS; let counts = self.compute_bucket_counts(); - for ((>, &le), &count) in std::iter::zip( - std::iter::zip(&buckets.gt_labels, &buckets.le_labels), - &counts, - ) { - Metrics::set_inflight_request_age_count(gt, le, count); - } + INFLIGHT_AGE_HANDLE.set_counts(&counts); } } @@ -169,45 +132,29 @@ mod tests { } #[test] - fn test_empty_tracker_buckets() { + fn test_collect_ages_empty() { let tracker = InFlightRequestTracker::new(); - let counts = tracker.compute_bucket_counts(); - assert!(counts.iter().all(|&c| c == 0)); + let ages = tracker.collect_ages(); + assert!(ages.is_empty()); } #[test] - fn test_bucket_counts() { + fn test_collect_ages() { let tracker = InFlightRequestTracker::new(); let now = Instant::now(); tracker.insert_with_time(1, now); tracker.insert_with_time(2, now - Duration::from_secs(45)); tracker.insert_with_time(3, now - Duration::from_secs(100)); - tracker.insert_with_time(4, now - Duration::from_secs(250)); - tracker.insert_with_time(5, now - Duration::from_secs(500)); - tracker.insert_with_time(6, now - Duration::from_secs(700)); - let counts = tracker.compute_bucket_counts(); - assert_eq!(counts[0], 1, "bucket <=30s"); - assert_eq!(counts[1], 1, "bucket <=60s"); - assert_eq!(counts[2], 1, "bucket <=180s"); - assert_eq!(counts[3], 1, "bucket <=300s"); - assert_eq!(counts[4], 1, "bucket <=600s"); - assert_eq!(counts[5], 1, "bucket <=1200s"); - assert_eq!(*counts.last().unwrap(), 0, "bucket +Inf"); - } - - #[test] - fn test_bucket_boundary_values() { - let tracker = InFlightRequestTracker::new(); - let now = Instant::now(); - - tracker.insert_with_time(1, now - Duration::from_secs(30)); - tracker.insert_with_time(2, now - Duration::from_secs(31)); - - let counts = tracker.compute_bucket_counts(); - assert_eq!(counts[0], 1, "bucket <=30s includes exact boundary"); - assert_eq!(counts[1], 1, "bucket <=60s has one request (31s)"); + let ages = tracker.collect_ages(); + assert_eq!(ages.len(), 3); + // Ages should be approximately 0, 45, 100 (order may vary due to DashMap) + let mut sorted_ages = ages.clone(); + sorted_ages.sort(); + assert!(sorted_ages[0] <= 1); // ~0s + assert!((44..=46).contains(&sorted_ages[1])); // ~45s + assert!((99..=101).contains(&sorted_ages[2])); // ~100s } #[test] @@ -245,24 +192,4 @@ mod tests { assert_ne!(g2.request_id, g3.request_id); assert_eq!(tracker.len(), 3); } - - #[test] - fn test_age_buckets_labels() { - let buckets = NumericalBuckets::new(&[10, 30, 60]); - - assert_eq!(buckets.le_labels, vec!["10", "30", "60", "+Inf"]); - assert_eq!(buckets.gt_labels, vec!["0", "10", "30", "60"]); - assert_eq!(buckets.len(), 4); - } - - #[test] - fn test_age_buckets_global() { - let buckets = &*AGE_BUCKETS; - - assert_eq!(buckets.le_labels.first(), Some(&"30")); - assert_eq!(buckets.le_labels.last(), Some(&"+Inf")); - assert_eq!(buckets.gt_labels.first(), Some(&"0")); - assert_eq!(buckets.gt_labels.last(), Some(&"86400")); - assert_eq!(buckets.le_labels.len(), buckets.gt_labels.len()); - } } diff --git a/sgl-model-gateway/src/observability/metrics.rs b/sgl-model-gateway/src/observability/metrics.rs index aee39cb5c..f5814abcb 100644 --- a/sgl-model-gateway/src/observability/metrics.rs +++ b/sgl-model-gateway/src/observability/metrics.rs @@ -497,15 +497,6 @@ impl Metrics { .record(duration.as_secs_f64()); } - pub fn set_inflight_request_age_count(gt: &'static str, le: &'static str, count: usize) { - gauge!( - "smg_http_inflight_request_age_count", - "gt" => gt, - "le" => le - ) - .set(count as f64); - } - /// Set active HTTP connections count pub fn set_http_connections_active(count: usize) { gauge!("smg_http_connections_active").set(count as f64);