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..e70f98422 100644 --- a/crates/kms/src/cache.rs +++ b/crates/kms/src/cache.rs @@ -16,11 +16,84 @@ 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. Republished whenever the entry +/// set can have changed — every write path, plus the lookups that miss, since +/// TTL expiry drops entries without any write taking place. +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 +106,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 +138,26 @@ 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()); + + if cached.is_none() { + // TTL expiry is the one way the entry set shrinks without a write, + // and a miss is where it surfaces. moka expires lazily: the miss is + // reported at once, but the removal that decrements `entry_count` + // and reaches the eviction listener lands in the maintenance pass + // moka runs opportunistically on reads, on its own interval. So + // this converges the gauge within a housekeeping interval rather + // than on the first miss — enough to stop a cache that only ever + // expires, with no put, remove or clear to follow, from reporting a + // population that is gone. Forcing `run_pending_tasks` would + // tighten that at the cost of taking moka's maintenance lock on + // every miss, for a gauge `entry_count` only approximates anyway. + // Hits, the hot path, are left alone. + self.record_entry_count(); + } + + cached } /// Put key metadata into cache @@ -62,6 +169,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 +179,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 +192,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 +229,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 +337,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 +397,92 @@ 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)); + } + + /// Entries that expire are dropped outside every write path, so the gauge + /// has to be republished from the lookup that observes the expiry — nothing + /// else runs afterwards to correct it. + /// + /// moka's clock is internal and cannot be driven by tokio's paused time, + /// hence the one short real sleep. The explicit `run_pending_tasks` stands + /// in for the maintenance moka schedules by itself on reads, so the test + /// does not depend on moka's internal 300ms housekeeping interval. + #[test] + fn expiry_without_further_writes_converges_the_entry_gauge() { + let ttl = Duration::from_millis(100); + + let (stats, snapshot) = record_metrics(move || async move { + let mut cache = KmsCache::with_ttl(100, ttl); + + cache.put_key_metadata("expiring", &test_metadata("expiring")).await; + assert_eq!(cache.stats().entries, 1); + + tokio::time::sleep(Duration::from_millis(150)).await; + cache.key_metadata_cache.run_pending_tasks().await; + + // The lookup that observes the expiry is the last thing to touch + // the cache: no put, remove or clear follows it. + assert!(cache.get_key_metadata("expiring").await.is_none()); + + cache.stats() + }); + + assert_eq!(counter_value(&snapshot, METRIC_CACHE_EVICTIONS_TOTAL, &[("cause", "expired")]), 1); + assert_eq!(stats.entries, 0); + assert_eq!(gauge_value(&snapshot, METRIC_CACHE_ENTRIES), Some(0.0)); + } } diff --git a/crates/kms/src/lib.rs b/crates/kms/src/lib.rs index 6e9ff1c37..26b0c9a2a 100644 --- a/crates/kms/src/lib.rs +++ b/crates/kms/src/lib.rs @@ -89,6 +89,7 @@ pub use api_types::{ UpdateKeyDescriptionRequest, UpdateKeyDescriptionResponse, }; pub use audit::{KmsAuditOperation, KmsAuditOutcome, KmsAuditRecord, KmsAuditSink, redact_encryption_context}; +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 30355b75d..4a7eb2601 100644 --- a/crates/kms/src/manager.rs +++ b/crates/kms/src/manager.rs @@ -16,7 +16,7 @@ use crate::audit::{KmsAuditOperation, KmsAuditRecord, KmsAuditSink}; use crate::backends::KmsBackend; -use crate::cache::KmsCache; +use crate::cache::{KmsCache, KmsCacheStats}; use crate::config::KmsConfig; use crate::error::Result; use crate::types::{ @@ -205,8 +205,8 @@ impl KmsManager { result } - /// 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()) @@ -882,9 +882,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 20a10dd1c..2b6496177 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; @@ -207,9 +208,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 } diff --git a/rustfs/src/admin/handlers/kms_management.rs b/rustfs/src/admin/handlers/kms_management.rs index 88ee3a6f9..7fa45b372 100644 --- a/rustfs/src/admin/handlers/kms_management.rs +++ b/rustfs/src/admin/handlers/kms_management.rs @@ -82,6 +82,14 @@ pub struct KmsStatusResponse { pub struct CacheStatsResponse { pub hit_count: u64, pub miss_count: u64, + /// Entries currently cached. Additive field: omitted by older servers, so + /// it must stay defaulted for consumers. + #[serde(default)] + pub entry_count: u64, + /// Entries dropped since process start, whatever the cause. Additive + /// field, same compatibility rule as `entry_count`. + #[serde(default)] + pub eviction_count: u64, } #[derive(Debug, Serialize, Deserialize)] @@ -193,9 +201,11 @@ impl Operation for KmsStatusHandler { } }; - let cache_stats = service.cache_stats().await.map(|(hits, misses)| CacheStatsResponse { - hit_count: hits, - miss_count: misses, + let cache_stats = service.cache_stats().await.map(|stats| CacheStatsResponse { + hit_count: stats.hits, + miss_count: stats.misses, + entry_count: stats.entries, + eviction_count: stats.evictions, }); let config = kms_service_manager_from_context().get_redacted_config().await;