diff --git a/README.md b/README.md index 124c04d..9e7b571 100644 --- a/README.md +++ b/README.md @@ -261,12 +261,18 @@ curl http://localhost:9090/metrics | Metric | Type | Labels | Description | |--------|------|--------|-------------| -| `colander_cache_hits_total` | counter | `policy` | Total cache hits | -| `colander_cache_misses_total` | counter | `policy` | Total cache misses | -| `colander_cache_keys` | gauge | `policy` | Current number of cached entries | -| `colander_cache_evictions_total` | gauge | `policy` | Total evictions | -| `colander_request_duration_seconds` | histogram | — | End-to-end request latency | -| `colander_upstream_duration_seconds` | histogram | — | Upstream (origin) latency on cache misses | +| `colander_cache_hits_total` | counter | `policy`, `role` | Observed cache hits | +| `colander_cache_misses_total` | counter | `policy`, `role` | Observed cache misses | +| `colander_cache_keys` | gauge | `policy`, `role` | Current cached entries | +| `colander_cache_capacity` | gauge | `policy`, `role` | Configured cache slots | +| `colander_cache_evictions_total` | counter | `policy`, `role` | Observed evictions | +| `colander_cache_lookups_per_second` | gauge | — | Primary cache lookups per second | + +Metrics are sampled every 500 ms. `role` distinguishes the primary and comparison +caches, including when both use the same policy. Counters remain monotonic when a +cache is replaced; increments that occur entirely between the last sample and a +replacement cannot be observed. Dashboard throughput measures primary cache lookups, +not requests that bypass the cache. Latency histograms are not currently exported. ### Grafana diff --git a/crates/colander-cache/src/fifo.rs b/crates/colander-cache/src/fifo.rs index 05d988c..13d38cf 100644 --- a/crates/colander-cache/src/fifo.rs +++ b/crates/colander-cache/src/fifo.rs @@ -31,6 +31,12 @@ impl FifoCache { } impl CachePolicy for FifoCache { + fn peek(&self, key: &str) -> Option> { + let index = *self.map.get(key)?; + let node = self.arena.get(index)?; + (!node.value.is_expired()).then(|| Arc::clone(&node.value)) + } + fn get(&mut self, key: &str) -> Option> { if let Some(&index) = self.map.get(key) { let node = self.arena.get(index).unwrap(); diff --git a/crates/colander-cache/src/lru.rs b/crates/colander-cache/src/lru.rs index 2237df8..449c27f 100644 --- a/crates/colander-cache/src/lru.rs +++ b/crates/colander-cache/src/lru.rs @@ -34,6 +34,12 @@ impl LruCache { } impl CachePolicy for LruCache { + fn peek(&self, key: &str) -> Option> { + let index = *self.map.get(key)?; + let node = self.arena.get(index)?; + (!node.value.is_expired()).then(|| Arc::clone(&node.value)) + } + fn get(&mut self, key: &str) -> Option> { if let Some(&index) = self.map.get(key) { let node = self.arena.get(index).unwrap(); diff --git a/crates/colander-cache/src/sharded.rs b/crates/colander-cache/src/sharded.rs index dec560e..6719358 100644 --- a/crates/colander-cache/src/sharded.rs +++ b/crates/colander-cache/src/sharded.rs @@ -46,6 +46,11 @@ impl ShardedCache { self.shards[idx].write().get(key) } + /// Inspect without counting another request or updating eviction policy state. + pub fn peek(&self, key: &str) -> Option> { + self.shards[self.shard_index(key)].read().peek(key) + } + /// Insert a key-value pair. Takes a write lock on one shard. pub fn insert(&self, key: String, value: CachedResponse) { let idx = self.shard_index(&key); diff --git a/crates/colander-cache/src/sieve.rs b/crates/colander-cache/src/sieve.rs index fb14525..2f525f2 100644 --- a/crates/colander-cache/src/sieve.rs +++ b/crates/colander-cache/src/sieve.rs @@ -95,6 +95,12 @@ impl SieveCache { } impl CachePolicy for SieveCache { + fn peek(&self, key: &str) -> Option> { + let index = *self.map.get(key)?; + let node = self.arena.get(index)?; + (!node.value.is_expired()).then(|| Arc::clone(&node.value)) + } + fn get(&mut self, key: &str) -> Option> { if let Some(&index) = self.map.get(key) { let node = self.arena.get(index).unwrap(); diff --git a/crates/colander-cache/src/traits.rs b/crates/colander-cache/src/traits.rs index cf2693b..e335792 100644 --- a/crates/colander-cache/src/traits.rs +++ b/crates/colander-cache/src/traits.rs @@ -36,6 +36,9 @@ pub trait CachePolicy: Send { /// Look up a key. Returns the cached response if found and not expired. fn get(&mut self, key: &str) -> Option>; + /// Inspect a live entry without changing statistics or eviction state. + fn peek(&self, key: &str) -> Option>; + /// Insert a key-value pair. May trigger eviction if at capacity. fn insert(&mut self, key: String, value: CachedResponse); diff --git a/crates/proxy-server/src/cache_layer.rs b/crates/proxy-server/src/cache_layer.rs index 8b38d32..78be15e 100644 --- a/crates/proxy-server/src/cache_layer.rs +++ b/crates/proxy-server/src/cache_layer.rs @@ -26,6 +26,14 @@ enum CacheInner { } impl CacheInner { + fn peek(&self, key: &str) -> Option> { + match self { + CacheInner::Sieve(c) => c.peek(key), + CacheInner::Lru(c) => c.peek(key), + CacheInner::Fifo(c) => c.peek(key), + } + } + fn get(&self, key: &str) -> Option> { match self { CacheInner::Sieve(c) => c.get(key), @@ -132,7 +140,14 @@ impl CacheLayer { let comparison_hit = if self.is_demo_mode() { if let Some(comp) = &self.comparison { - comp.get(key).is_some() + let hit = comp.get(key).is_some(); + // A shadow miss still needs the object when primary serves a hit. + if !hit { + if let Some(value) = &primary_result { + comp.insert(key.to_owned(), (**value).clone()); + } + } + hit } else { false } @@ -150,7 +165,11 @@ impl CacheLayer { pub fn insert(&self, key: String, value: CachedResponse) { if self.is_demo_mode() { if let Some(comp) = &self.comparison { - comp.insert(key.clone(), value.clone()); + // A primary miss may already be a shadow hit. Preserve its entry + // and policy state; a second lookup must not skew hit counters. + if comp.peek(&key).is_none() { + comp.insert(key.clone(), value.clone()); + } } } self.primary.insert(key, value); @@ -275,3 +294,61 @@ pub struct CacheControl { pub cacheable: bool, pub max_age: Option, } + +#[cfg(test)] +mod comparison_tests { + use super::*; + + fn cache() -> CacheLayer { + CacheLayer::new("sieve", Some("lru"), 1024, Duration::from_secs(60), 1024) + } + + #[test] + fn primary_hit_populates_a_shadow_miss_without_extra_statistics() { + let cache = cache(); + cache.insert( + "key".into(), + cache.build_response(200, vec![], Bytes::from_static(b"value"), None), + ); + cache.comparison.as_ref().unwrap().remove("key"); + let lookup = cache.get("key"); + assert!(lookup.is_hit()); + assert!(!lookup.comparison_hit); + let stats = cache.comparison_stats().unwrap(); + assert_eq!((stats.hits, stats.misses, stats.current_size), (0, 1, 1)); + assert!(cache.get("key").comparison_hit); + } + + #[test] + fn primary_miss_does_not_replace_a_shadow_hit_or_count_an_extra_lookup() { + let cache = cache(); + cache.insert( + "key".into(), + cache.build_response(200, vec![], Bytes::from_static(b"old"), None), + ); + cache.primary.remove("key"); + let lookup = cache.get("key"); + assert!(!lookup.is_hit()); + assert!(lookup.comparison_hit); + cache.insert( + "key".into(), + cache.build_response(200, vec![], Bytes::from_static(b"fresh"), None), + ); + let comparison = cache.comparison.as_ref().unwrap(); + assert_eq!(comparison.peek("key").unwrap().body, "old"); + assert_eq!((comparison.stats().hits, comparison.stats().misses), (1, 0)); + } + + #[test] + fn bench_mode_does_not_populate_or_touch_shadow_cache() { + let cache = cache(); + cache.set_mode(CacheMode::Bench); + cache.insert( + "key".into(), + cache.build_response(200, vec![], Bytes::from_static(b"value"), None), + ); + cache.get("key"); + let stats = cache.comparison_stats().unwrap(); + assert_eq!((stats.hits, stats.misses, stats.current_size), (0, 0, 0)); + } +} diff --git a/crates/proxy-server/src/metrics.rs b/crates/proxy-server/src/metrics.rs index 6c12684..9b7c64e 100644 --- a/crates/proxy-server/src/metrics.rs +++ b/crates/proxy-server/src/metrics.rs @@ -85,26 +85,54 @@ pub async fn metrics_broadcaster( start_time: Instant, ) { let mut interval = tokio::time::interval(tokio::time::Duration::from_millis(500)); - let mut prev_total_requests: u64 = 0; + let mut previous_cache: Option> = None; + let mut previous_primary: Option = None; + let mut previous_comparison: Option = None; + let mut previous_sample = Instant::now(); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); loop { interval.tick().await; - let cache = state.cache.load(); + let cache = state.cache.load_full(); let primary = PolicyMetrics::from_cache(&cache, true).unwrap(); // primary always Some let comparison = PolicyMetrics::from_cache(&cache, false); - let current_total = primary.hits + primary.misses; - let delta = current_total.saturating_sub(prev_total_requests); - let throughput = delta as f64 * 2.0; // 500ms window → multiply by 2 for per-second - prev_total_requests = current_total; + let now = Instant::now(); + let window = now.duration_since(previous_sample); + previous_sample = now; + let same_cache = previous_cache + .as_ref() + .is_some_and(|previous| Arc::ptr_eq(previous, &cache)); + if !same_cache { + for (old, role) in [ + (previous_primary.as_ref(), "primary"), + (previous_comparison.as_ref(), "comparison"), + ] { + if let Some(old) = old { + ::metrics::gauge!("colander_cache_keys", "policy" => old.name.clone(), "role" => role).set(0.0); + ::metrics::gauge!("colander_cache_capacity", "policy" => old.name.clone(), "role" => role).set(0.0); + } + } + previous_primary = None; + previous_comparison = None; + } + let throughput = lookup_rate(&primary, previous_primary.as_ref(), window); + publish_policy(&primary, previous_primary.as_ref(), "primary"); + if let Some(comparison) = &comparison { + publish_policy(comparison, previous_comparison.as_ref(), "comparison"); + } + ::metrics::gauge!("colander_cache_lookups_per_second").set(throughput); + previous_cache = Some(cache.clone()); + previous_primary = Some(primary.clone()); + previous_comparison = comparison.clone(); let snapshot = MetricsSnapshot { timestamp_ms: std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap() // safe: clock is after 1970 .as_millis(), - window_ms: 500, + window_ms: window.as_millis().max(1) as u64, primary, comparison, throughput_rps: throughput, @@ -117,6 +145,43 @@ pub async fn metrics_broadcaster( } } +fn lookup_rate( + current: &PolicyMetrics, + previous: Option<&PolicyMetrics>, + window: std::time::Duration, +) -> f64 { + let old = previous.map_or(0, |p| p.hits + p.misses); + let count = (current.hits + current.misses).saturating_sub(old); + count as f64 / window.as_secs_f64().max(0.001) +} + +fn publish_policy(current: &PolicyMetrics, previous: Option<&PolicyMetrics>, role: &'static str) { + for (metric, value, old) in [ + ( + "colander_cache_hits_total", + current.hits, + previous.map_or(0, |p| p.hits), + ), + ( + "colander_cache_misses_total", + current.misses, + previous.map_or(0, |p| p.misses), + ), + ( + "colander_cache_evictions_total", + current.evictions, + previous.map_or(0, |p| p.evictions), + ), + ] { + ::metrics::counter!(metric, "policy" => current.name.clone(), "role" => role) + .increment(value.saturating_sub(old)); + } + ::metrics::gauge!("colander_cache_keys", "policy" => current.name.clone(), "role" => role) + .set(current.size as f64); + ::metrics::gauge!("colander_cache_capacity", "policy" => current.name.clone(), "role" => role) + .set(current.capacity as f64); +} + /// WebSocket upgrade handler for /ws/metrics. pub async fn ws_metrics_handler( ws: WebSocketUpgrade, @@ -188,3 +253,59 @@ pub async fn stats_handler(State(state): State) -> impl IntoRespon "mode": format!("{:?}", cache.mode()).to_lowercase(), })) } + +#[cfg(test)] +mod tests { + use super::*; + use std::time::Duration; + + fn sample(hits: u64) -> PolicyMetrics { + PolicyMetrics { + name: "SIEVE".into(), + hits, + misses: 0, + evictions: 0, + hit_rate: 1.0, + size: 2, + capacity: 64, + } + } + + #[test] + fn uses_actual_window_and_resets_baseline_after_cache_replacement() { + assert_eq!( + lookup_rate(&sample(10), Some(&sample(4)), Duration::from_secs(2)), + 3.0 + ); + assert_eq!( + lookup_rate(&sample(2), None, Duration::from_millis(500)), + 4.0 + ); + } + + #[test] + fn prometheus_exports_real_counters_and_gauges_without_double_counting() { + let recorder = metrics_exporter_prometheus::PrometheusBuilder::new().build_recorder(); + let handle = recorder.handle(); + ::metrics::with_local_recorder(&recorder, || { + publish_policy(&sample(4), None, "primary"); + publish_policy(&sample(6), Some(&sample(4)), "primary"); + publish_policy(&sample(6), Some(&sample(6)), "primary"); + // A replaced cache adds its new counts to the monotonic series. + publish_policy(&sample(2), None, "primary"); + }); + let output = handle.render(); + assert!( + output.contains("colander_cache_hits_total{policy=\"SIEVE\",role=\"primary\"} 8"), + "{output}" + ); + assert!( + output.contains("colander_cache_keys{policy=\"SIEVE\",role=\"primary\"} 2"), + "{output}" + ); + assert!( + output.contains("colander_cache_capacity{policy=\"SIEVE\",role=\"primary\"} 64"), + "{output}" + ); + } +}