diff --git a/crates/kms/examples/kms_local_demo.rs b/crates/kms/examples/kms_local_demo.rs index e5cd7b5ab..fa75ca7ad 100644 --- a/crates/kms/examples/kms_local_demo.rs +++ b/crates/kms/examples/kms_local_demo.rs @@ -227,10 +227,12 @@ async fn main() -> Result<(), Box> { // Step 12: Check cache statistics println!("12. Checking cache statistics..."); - if let Some((hits, misses)) = encryption_service.cache_stats().await { + if let Some(stats) = encryption_service.cache_stats().await { println!(" ✓ Cache statistics:"); - println!(" - Cache hits: {}", hits); - println!(" - Cache misses: {}\n", misses); + println!(" - Cached entries: {}", stats.entries); + println!(" - Cache hits: {}", stats.hits); + println!(" - Cache misses: {}", stats.misses); + println!(" - Entries evicted: {}\n", stats.evictions); } else { println!(" - Cache is disabled\n"); } diff --git a/crates/kms/examples/kms_vault_kv_demo.rs b/crates/kms/examples/kms_vault_kv_demo.rs index 9b518bfad..162bd4df1 100644 --- a/crates/kms/examples/kms_vault_kv_demo.rs +++ b/crates/kms/examples/kms_vault_kv_demo.rs @@ -263,10 +263,12 @@ async fn main() -> Result<(), Box> { // Step 12: Check cache statistics println!("12. Checking cache statistics..."); - if let Some((hits, misses)) = encryption_service.cache_stats().await { + if let Some(stats) = encryption_service.cache_stats().await { println!(" ✓ Cache statistics:"); - println!(" - Cache hits: {}", hits); - println!(" - Cache misses: {}\n", misses); + println!(" - Cached entries: {}", stats.entries); + println!(" - Cache hits: {}", stats.hits); + println!(" - Cache misses: {}", stats.misses); + println!(" - Entries evicted: {}\n", stats.evictions); } else { println!(" - Cache is disabled\n"); } diff --git a/crates/kms/src/cache.rs b/crates/kms/src/cache.rs index cbf9b35b2..5f8a7db75 100644 --- a/crates/kms/src/cache.rs +++ b/crates/kms/src/cache.rs @@ -16,11 +16,82 @@ use crate::types::KeyMetadata; use moka::future::Cache; +use moka::notification::RemovalCause; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; use std::time::Duration; +/// Default lifetime of a cached key metadata entry. +const DEFAULT_METADATA_TTL: Duration = Duration::from_secs(300); + +// --------------------------------------------------------------------------- +// Metrics +// +// Lookup outcomes and removals are counted here because moka exposes neither +// hit nor miss counts; the numbers below are the cache's own, not derived. +// Label values are exclusively static strings (lookup result, removal cause) — +// key identifiers and key metadata must never reach a metric label. +// --------------------------------------------------------------------------- + +/// Counter: key metadata lookups, by `result` (`hit` or `miss`). +const METRIC_CACHE_LOOKUPS_TOTAL: &str = "rustfs_kms_metadata_cache_lookups_total"; +/// Counter: entries dropped from the cache, by `cause` (`expired`, `size`, +/// `explicit`, `replaced`); only `expired` and `size` are true evictions, the +/// rest are invalidations driven by key lifecycle operations. +const METRIC_CACHE_EVICTIONS_TOTAL: &str = "rustfs_kms_metadata_cache_evictions_total"; +/// Gauge: entries currently held by the cache. +const METRIC_CACHE_ENTRIES: &str = "rustfs_kms_metadata_cache_entries"; + +/// Register metric descriptions once per process. +fn describe_metrics() { + static DESCRIBE: std::sync::Once = std::sync::Once::new(); + DESCRIBE.call_once(|| { + metrics::describe_counter!(METRIC_CACHE_LOOKUPS_TOTAL, "Total KMS key metadata cache lookups, by hit or miss"); + metrics::describe_counter!( + METRIC_CACHE_EVICTIONS_TOTAL, + "Total entries dropped from the KMS key metadata cache, by removal cause" + ); + metrics::describe_gauge!(METRIC_CACHE_ENTRIES, "Entries currently held by the KMS key metadata cache"); + }); +} + +fn removal_cause_label(cause: RemovalCause) -> &'static str { + match cause { + RemovalCause::Expired => "expired", + RemovalCause::Explicit => "explicit", + RemovalCause::Replaced => "replaced", + RemovalCause::Size => "size", + } +} + +/// Cumulative cache counters. Shared with moka's eviction listener, which runs +/// outside any `&self` borrow, hence the `Arc` and the atomics. +#[derive(Debug, Default)] +struct CacheCounters { + hits: AtomicU64, + misses: AtomicU64, + evictions: AtomicU64, +} + +/// Snapshot of the KMS key metadata cache counters. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct KmsCacheStats { + /// Entries currently held. Eventually consistent: moka applies pending + /// maintenance lazily. + pub entries: u64, + /// Lookups served from the cache since process start. + pub hits: u64, + /// Lookups that fell through to the backend since process start. + pub misses: u64, + /// Entries dropped since process start, whatever the cause (expiry, + /// capacity, invalidation, replacement). + pub evictions: u64, +} + /// KMS cache for storing frequently accessed keys and metadata pub struct KmsCache { key_metadata_cache: Cache, + counters: Arc, } impl KmsCache { @@ -33,11 +104,26 @@ impl KmsCache { /// A new instance of `KmsCache` /// pub fn new(capacity: u64) -> Self { + Self::with_ttl(capacity, DEFAULT_METADATA_TTL) + } + + /// Create a new KMS cache with an explicit metadata time-to-live + fn with_ttl(capacity: u64, metadata_ttl: Duration) -> Self { + describe_metrics(); + + let counters = Arc::new(CacheCounters::default()); + let eviction_counters = Arc::clone(&counters); + Self { key_metadata_cache: Cache::builder() .max_capacity(capacity) - .time_to_live(Duration::from_secs(300)) // 5 minutes default TTL + .time_to_live(metadata_ttl) + .eviction_listener(move |_key: Arc, _metadata: KeyMetadata, cause: RemovalCause| { + eviction_counters.evictions.fetch_add(1, Ordering::Relaxed); + metrics::counter!(METRIC_CACHE_EVICTIONS_TOTAL, "cause" => removal_cause_label(cause)).increment(1); + }) .build(), + counters, } } @@ -50,7 +136,9 @@ impl KmsCache { /// An `Option` containing the `KeyMetadata` if found, or `None` if not found /// pub async fn get_key_metadata(&self, key_id: &str) -> Option { - self.key_metadata_cache.get(key_id).await + let cached = self.key_metadata_cache.get(key_id).await; + self.record_lookup(cached.is_some()); + cached } /// Put key metadata into cache @@ -62,6 +150,7 @@ impl KmsCache { pub async fn put_key_metadata(&mut self, key_id: &str, metadata: &KeyMetadata) { self.key_metadata_cache.insert(key_id.to_string(), metadata.clone()).await; self.key_metadata_cache.run_pending_tasks().await; + self.record_entry_count(); } /// Remove key metadata from cache @@ -71,6 +160,11 @@ impl KmsCache { /// pub async fn remove_key_metadata(&mut self, key_id: &str) { self.key_metadata_cache.remove(key_id).await; + + // Flush maintenance so the removal notification and the entry gauge + // describe the cache as it is once this call returns. + self.key_metadata_cache.run_pending_tasks().await; + self.record_entry_count(); } /// Clear all cached entries @@ -79,18 +173,35 @@ impl KmsCache { // Wait for invalidation to complete self.key_metadata_cache.run_pending_tasks().await; + self.record_entry_count(); } - /// Get cache statistics (hit count, miss count) + /// Get cache statistics /// /// # Returns - /// A tuple containing total entries and total misses + /// A [`KmsCacheStats`] snapshot of entry count, hits, misses and evictions /// - pub fn stats(&self) -> (u64, u64) { - ( - self.key_metadata_cache.entry_count(), - 0u64, // moka doesn't provide miss count directly - ) + pub fn stats(&self) -> KmsCacheStats { + KmsCacheStats { + entries: self.key_metadata_cache.entry_count(), + hits: self.counters.hits.load(Ordering::Relaxed), + misses: self.counters.misses.load(Ordering::Relaxed), + evictions: self.counters.evictions.load(Ordering::Relaxed), + } + } + + fn record_lookup(&self, hit: bool) { + let (counter, result) = if hit { + (&self.counters.hits, "hit") + } else { + (&self.counters.misses, "miss") + }; + counter.fetch_add(1, Ordering::Relaxed); + metrics::counter!(METRIC_CACHE_LOOKUPS_TOTAL, "result" => result).increment(1); + } + + fn record_entry_count(&self) { + metrics::gauge!(METRIC_CACHE_ENTRIES).set(self.key_metadata_cache.entry_count() as f64); } } @@ -99,37 +210,90 @@ mod tests { use super::*; use crate::types::{KeyState, KeyUsage}; use jiff::Zoned; + use metrics_util::MetricKind; + use metrics_util::debugging::{DebugValue, DebuggingRecorder}; use std::time::Duration; - #[derive(Debug, Clone)] - struct CacheInfo { - key_metadata_count: u64, - } - - impl CacheInfo { - fn total_entries(&self) -> u64 { - self.key_metadata_count - } - } - impl KmsCache { - fn with_ttl_for_tests(capacity: u64, metadata_ttl: Duration) -> Self { - Self { - key_metadata_cache: Cache::builder().max_capacity(capacity).time_to_live(metadata_ttl).build(), - } - } - - fn info_for_tests(&self) -> CacheInfo { - CacheInfo { - key_metadata_count: self.key_metadata_cache.entry_count(), - } - } - fn contains_key_metadata_for_tests(&self, key_id: &str) -> bool { self.key_metadata_cache.contains_key(key_id) } } + fn test_metadata(key_id: &str) -> KeyMetadata { + KeyMetadata { + key_id: key_id.to_string(), + key_state: KeyState::Enabled, + key_usage: KeyUsage::EncryptDecrypt, + description: None, + creation_date: Zoned::now(), + deletion_date: None, + origin: "KMS".to_string(), + key_manager: "CUSTOMER".to_string(), + tags: std::collections::HashMap::new(), + } + } + + type MetricEntry = ( + metrics_util::CompositeKey, + Option, + Option, + DebugValue, + ); + + /// Drive `test` on a current-thread runtime under a thread-local debugging + /// recorder and return its output plus one snapshot of everything emitted. + /// + /// A single snapshot per test on purpose: `Snapshotter::snapshot` drains + /// the recorded state, so taking it per assertion would leave every + /// assertion after the first with nothing to look at. + fn record_metrics(test: F) -> (T, Vec) + where + F: FnOnce() -> Fut, + Fut: std::future::Future, + { + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let output = metrics::with_local_recorder(&recorder, || { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("current-thread runtime must build"); + runtime.block_on(test()) + }); + (output, snapshotter.snapshot().into_vec()) + } + + /// Sum of counters with `name` whose labels include every `(key, value)` pair. + fn counter_value(snapshot: &[MetricEntry], name: &str, labels: &[(&str, &str)]) -> u64 { + snapshot + .iter() + .filter_map(|(composite, _unit, _description, value)| { + let key = composite.key(); + let matches = composite.kind() == MetricKind::Counter + && key.name() == name + && labels + .iter() + .all(|(label, expected)| key.labels().any(|l| l.key() == *label && l.value() == *expected)); + match (matches, value) { + (true, DebugValue::Counter(count)) => Some(*count), + _ => None, + } + }) + .sum() + } + + /// Last value recorded for the unlabelled gauge `name`. + fn gauge_value(snapshot: &[MetricEntry], name: &str) -> Option { + snapshot.iter().find_map(|(composite, _unit, _description, value)| { + let matches = composite.kind() == MetricKind::Gauge && composite.key().name() == name; + match (matches, value) { + (true, DebugValue::Gauge(gauge)) => Some(**gauge), + _ => None, + } + }) + } + #[tokio::test] async fn test_cache_operations() { let mut cache = KmsCache::new(100); @@ -154,19 +318,16 @@ mod tests { assert_eq!(retrieved.expect("metadata should be cached").key_id, "test-key-1"); // Test cache info - let info = cache.info_for_tests(); - assert_eq!(info.key_metadata_count, 1); - assert_eq!(info.total_entries(), 1); + assert_eq!(cache.stats().entries, 1); // Test cache clearing cache.clear().await; - let info_after_clear = cache.info_for_tests(); - assert_eq!(info_after_clear.total_entries(), 0); + assert_eq!(cache.stats().entries, 0); } #[tokio::test] async fn test_cache_with_custom_ttl() { - let mut cache = KmsCache::with_ttl_for_tests( + let mut cache = KmsCache::with_ttl( 100, Duration::from_millis(100), // Short TTL for testing ); @@ -217,4 +378,59 @@ mod tests { assert!(cache.contains_key_metadata_for_tests("contains-test")); } + + #[test] + fn lookups_report_real_hit_and_miss_counts() { + let (stats, snapshot) = record_metrics(|| async { + let mut cache = KmsCache::new(100); + + assert!(cache.get_key_metadata("absent").await.is_none()); + cache.put_key_metadata("present", &test_metadata("present")).await; + assert!(cache.get_key_metadata("present").await.is_some()); + assert!(cache.get_key_metadata("absent").await.is_none()); + + cache.stats() + }); + + assert_eq!(stats.hits, 1); + assert_eq!(stats.misses, 2); + assert_eq!(counter_value(&snapshot, METRIC_CACHE_LOOKUPS_TOTAL, &[("result", "hit")]), 1); + assert_eq!(counter_value(&snapshot, METRIC_CACHE_LOOKUPS_TOTAL, &[("result", "miss")]), 2); + } + + #[test] + fn removals_report_their_cause_and_the_resulting_entry_count() { + let (stats, snapshot) = record_metrics(|| async { + let mut cache = KmsCache::new(100); + + cache.put_key_metadata("key", &test_metadata("key")).await; + cache.put_key_metadata("key", &test_metadata("key")).await; + cache.remove_key_metadata("key").await; + + cache.stats() + }); + + assert_eq!(counter_value(&snapshot, METRIC_CACHE_EVICTIONS_TOTAL, &[("cause", "replaced")]), 1); + assert_eq!(counter_value(&snapshot, METRIC_CACHE_EVICTIONS_TOTAL, &[("cause", "explicit")]), 1); + assert_eq!(stats.evictions, 2); + assert_eq!(stats.entries, 0); + assert_eq!(gauge_value(&snapshot, METRIC_CACHE_ENTRIES), Some(0.0)); + } + + #[test] + fn capacity_pressure_reports_size_evictions() { + let (stats, snapshot) = record_metrics(|| async { + let mut cache = KmsCache::with_ttl(1, DEFAULT_METADATA_TTL); + + cache.put_key_metadata("first", &test_metadata("first")).await; + cache.put_key_metadata("second", &test_metadata("second")).await; + + cache.stats() + }); + + assert_eq!(counter_value(&snapshot, METRIC_CACHE_EVICTIONS_TOTAL, &[("cause", "size")]), 1); + assert_eq!(stats.evictions, 1); + assert_eq!(stats.entries, 1); + assert_eq!(gauge_value(&snapshot, METRIC_CACHE_ENTRIES), Some(1.0)); + } } diff --git a/crates/kms/src/lib.rs b/crates/kms/src/lib.rs index 892ed5a35..820aa49f7 100644 --- a/crates/kms/src/lib.rs +++ b/crates/kms/src/lib.rs @@ -86,6 +86,7 @@ pub use api_types::{ StartKmsResponse, StopKmsResponse, TagKeyRequest, TagKeyResponse, UntagKeyRequest, UntagKeyResponse, UpdateKeyDescriptionRequest, UpdateKeyDescriptionResponse, }; +pub use cache::KmsCacheStats; pub use config::*; pub use deletion_worker::DeletionReferenceChecker; pub use encryption::is_data_key_envelope; diff --git a/crates/kms/src/manager.rs b/crates/kms/src/manager.rs index 9d77e07b4..2fde2a2e6 100644 --- a/crates/kms/src/manager.rs +++ b/crates/kms/src/manager.rs @@ -15,7 +15,7 @@ //! KMS manager for handling key operations and backend coordination use crate::backends::KmsBackend; -use crate::cache::KmsCache; +use crate::cache::{KmsCache, KmsCacheStats}; use crate::config::KmsConfig; use crate::error::Result; use crate::types::{ @@ -113,8 +113,8 @@ impl KmsManager { self.backend.list_keys(request).await } - /// Get cache statistics - pub async fn cache_stats(&self) -> Option<(u64, u64)> { + /// Get cache statistics, or `None` when caching is disabled + pub async fn cache_stats(&self) -> Option { if self.enable_cache { let cache = self.cache.read().await; Some(cache.stats()) @@ -254,9 +254,12 @@ mod tests { let describe_response = manager.describe_key(describe_request).await.expect("Failed to describe key"); assert_eq!(describe_response.key_metadata.key_id, create_response.key_id); - // Test cache stats - let stats = manager.cache_stats().await; - assert!(stats.is_some()); + // Creating the key populated the cache, so the describe above was + // served from it rather than from the backend. + let stats = manager.cache_stats().await.expect("cache is enabled"); + assert_eq!(stats.entries, 1); + assert_eq!(stats.hits, 1); + assert_eq!(stats.misses, 0); // Test health check let health = manager.health_check().await.expect("Health check failed"); diff --git a/crates/kms/src/service.rs b/crates/kms/src/service.rs index 01902541b..6f18e24b0 100644 --- a/crates/kms/src/service.rs +++ b/crates/kms/src/service.rs @@ -14,6 +14,7 @@ //! Object encryption service for S3-compatible encryption +use crate::cache::KmsCacheStats; use crate::encryption::ciphers::{create_cipher, generate_iv}; use crate::error::{KmsError, Result}; use crate::manager::KmsManager; @@ -160,9 +161,9 @@ impl ObjectEncryptionService { /// Get cache statistics /// /// # Returns - /// Option with (hits, misses) if caching is enabled + /// A [`KmsCacheStats`] snapshot if caching is enabled, `None` otherwise /// - pub async fn cache_stats(&self) -> Option<(u64, u64)> { + pub async fn cache_stats(&self) -> Option { self.kms_manager.cache_stats().await }