fix(kms): report real cache counters through the admin status API

KmsStatusResponse.cache_stats mapped the old (entry_count, 0) tuple onto
hit_count and miss_count, so operators polling KMS status read the entry
count as a hit count and a miss count that was always zero.

Map the fields to the counters they claim to be, and add entry_count and
eviction_count as additive, defaulted fields so the entry number that
hit_count used to carry is still available.

Refs rustfs/backlog#1584

crates/kms/examples/kms_local_demo.rs

@@ -227,10 +227,12 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     // 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");
     }

crates/kms/examples/kms_vault_kv_demo.rs

@@ -263,10 +263,12 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     // 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");
     }

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<String, KeyMetadata>,
+    counters: Arc<CacheCounters>,
 }
 
 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<String>, _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<KeyMetadata> {
-        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<metrics::Unit>,
+        Option<metrics::SharedString>,
+        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<F, Fut, T>(test: F) -> (T, Vec<MetricEntry>)
+    where
+        F: FnOnce() -> Fut,
+        Fut: std::future::Future<Output = T>,
+    {
+        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<f64> {
+        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));
+    }
 }

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;

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<KmsCacheStats> {
         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");

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<KmsCacheStats> {
         self.kms_manager.cache_stats().await
     }
 

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;