Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 12 additions & 6 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
6 changes: 6 additions & 0 deletions crates/colander-cache/src/fifo.rs
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,12 @@ impl FifoCache {
}

impl CachePolicy for FifoCache {
fn peek(&self, key: &str) -> Option<Arc<CachedResponse>> {
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<Arc<CachedResponse>> {
if let Some(&index) = self.map.get(key) {
let node = self.arena.get(index).unwrap();
Expand Down
6 changes: 6 additions & 0 deletions crates/colander-cache/src/lru.rs
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,12 @@ impl LruCache {
}

impl CachePolicy for LruCache {
fn peek(&self, key: &str) -> Option<Arc<CachedResponse>> {
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<Arc<CachedResponse>> {
if let Some(&index) = self.map.get(key) {
let node = self.arena.get(index).unwrap();
Expand Down
5 changes: 5 additions & 0 deletions crates/colander-cache/src/sharded.rs
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,11 @@ impl<T: CachePolicy> ShardedCache<T> {
self.shards[idx].write().get(key)
}

/// Inspect without counting another request or updating eviction policy state.
pub fn peek(&self, key: &str) -> Option<Arc<CachedResponse>> {
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);
Expand Down
6 changes: 6 additions & 0 deletions crates/colander-cache/src/sieve.rs
Original file line number Diff line number Diff line change
Expand Up @@ -95,6 +95,12 @@ impl SieveCache {
}

impl CachePolicy for SieveCache {
fn peek(&self, key: &str) -> Option<Arc<CachedResponse>> {
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<Arc<CachedResponse>> {
if let Some(&index) = self.map.get(key) {
let node = self.arena.get(index).unwrap();
Expand Down
3 changes: 3 additions & 0 deletions crates/colander-cache/src/traits.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<Arc<CachedResponse>>;

/// Inspect a live entry without changing statistics or eviction state.
fn peek(&self, key: &str) -> Option<Arc<CachedResponse>>;

/// Insert a key-value pair. May trigger eviction if at capacity.
fn insert(&mut self, key: String, value: CachedResponse);

Expand Down
81 changes: 79 additions & 2 deletions crates/proxy-server/src/cache_layer.rs
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,14 @@ enum CacheInner {
}

impl CacheInner {
fn peek(&self, key: &str) -> Option<Arc<CachedResponse>> {
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<Arc<CachedResponse>> {
match self {
CacheInner::Sieve(c) => c.get(key),
Expand Down Expand Up @@ -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
}
Expand All @@ -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);
Expand Down Expand Up @@ -275,3 +294,61 @@ pub struct CacheControl {
pub cacheable: bool,
pub max_age: Option<Duration>,
}

#[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));
}
}
135 changes: 128 additions & 7 deletions crates/proxy-server/src/metrics.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<Arc<CacheLayer>> = None;
let mut previous_primary: Option<PolicyMetrics> = None;
let mut previous_comparison: Option<PolicyMetrics> = 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,
Expand All @@ -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,
Expand Down Expand Up @@ -188,3 +253,59 @@ pub async fn stats_handler(State(state): State<MetricsState>) -> 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}"
);
}
}
Loading