Refactor cache metrics for consistency and completeness (#1721)

This PR streamlines cache metrics collection across disk and express
cache implementations.

This change renames caching metrics for consistency with other OTLP
metrics. This change also captures latency, bytes_transferred from/to
cache and errors consistently across both disk and express cache
implementations.

### Does this change impact existing behavior?

Yes, updates cache metrics in logs

### Does this change need a changelog entry? Does it require a version
change?

Yes

---

By submitting this pull request, I confirm that my contribution is made
under the terms of the Apache 2.0 license and I agree to the terms of
the [Developer Certificate of Origin
(DCO)](https://developercertificate.org/).

---------

Signed-off-by: Sahitya Damera <sahityad@amazon.com>

Author: sahityadg

mountpoint-s3-fs/CHANGELOG.md

@@ -1,7 +1,8 @@
 ## Unreleased (v0.8.4)
 
 * Add metric to track cache hit rate in logs. ([#1716](https://github.com/awslabs/mountpoint-s3/pull/1716))
-* Remove redundant cache merics in logs. ([#1716](https://github.com/awslabs/mountpoint-s3/pull/1716))
+* Remove redundant cache merics in logs. ([#1716](https://github.com/awslabs/mountpoint-s3/pull/1716), [#1721](https://github.com/awslabs/mountpoint-s3/pull/1721))
+* Rename cache metrics for consistency. ([#1721](https://github.com/awslabs/mountpoint-s3/pull/1721))
 
 ## v0.8.3 (October 30, 2025)
 

mountpoint-s3-fs/src/data_cache.rs

@@ -42,19 +42,6 @@ pub enum DataCacheError {
     EvictionFailure,
 }
 
-impl DataCacheError {
-    fn reason(&self) -> &'static str {
-        match self {
-            DataCacheError::IoFailure(_) => "io_failure",
-            DataCacheError::InvalidBlockHeader(_) => "invalid_block_header",
-            DataCacheError::InvalidBlockChecksum => "invalid_block_checksum",
-            DataCacheError::InvalidBlockContent => "invalid_block_content",
-            DataCacheError::InvalidBlockOffset => "invalid_block_offset",
-            DataCacheError::EvictionFailure => "eviction_failure",
-        }
-    }
-}
-
 pub type DataCacheResult<Value> = Result<Value, DataCacheError>;
 
 /// Data cache for fixed-size checksummed buffers.

mountpoint-s3-fs/src/data_cache/disk_data_cache.rs

@@ -21,6 +21,10 @@ use tracing::{trace, warn};
 use crate::checksums::IntegrityError;
 use crate::data_cache::DataCacheError;
 use crate::memory::{BufferKind, PagedPool};
+use crate::metrics::defs::{
+    ATTR_CACHE, CACHE_DISK, CACHE_EVICT_LATENCY, CACHE_GET_ERRORS, CACHE_GET_IO_SIZE, CACHE_GET_LATENCY,
+    CACHE_PUT_ERRORS, CACHE_PUT_IO_SIZE, CACHE_PUT_LATENCY, CACHE_TOTAL_SIZE,
+};
 use crate::object::ObjectId;
 use crate::sync::Mutex;
 
@@ -387,7 +391,7 @@ impl DiskDataCache {
     }
 
     fn is_limit_exceeded(&self, size: usize) -> bool {
-        metrics::gauge!("disk_data_cache.disk_usage_mib").set((size / 1024 / 1024) as f64);
+        metrics::gauge!(CACHE_TOTAL_SIZE, ATTR_CACHE => CACHE_DISK).set(size as f64);
         match self.config.limit {
             CacheLimit::Unbounded => false,
             CacheLimit::TotalSize { max_size } => size > max_size,
@@ -461,29 +465,27 @@ impl DataCache for DiskDataCache {
         let start = Instant::now();
         let block_key = DiskBlockKey::new(cache_key, block_idx);
         let path = self.get_path_for_block_key(&block_key);
-        match self.read_block(&path, cache_key, block_idx, block_offset) {
+        let result = match self.read_block(&path, cache_key, block_idx, block_offset) {
             Ok(None) => {
                 // Cache miss.
-                metrics::counter!("disk_data_cache.block_hit").increment(0);
                 Ok(None)
             }
             Ok(Some(bytes)) => {
                 // Cache hit.
-                metrics::counter!("disk_data_cache.block_hit").increment(1);
-                metrics::counter!("disk_data_cache.total_bytes", "type" => "read").increment(bytes.len() as u64);
-                metrics::histogram!("disk_data_cache.read_duration_us").record(start.elapsed().as_micros() as f64);
+                metrics::counter!(CACHE_GET_IO_SIZE, ATTR_CACHE => CACHE_DISK).increment(bytes.len() as u64);
                 if let Some(usage) = &self.usage {
                     usage.lock().unwrap().refresh(&block_key);
                 }
                 Ok(Some(bytes))
             }
             Err(err) => {
-                // Invalid block. Count as cache miss.
-                metrics::counter!("disk_data_cache.block_hit").increment(0);
-                metrics::counter!("disk_data_cache.block_err").increment(1);
+                // Invalid block.
+                metrics::counter!(CACHE_GET_ERRORS, ATTR_CACHE => CACHE_DISK).increment(1);
                 Err(err)
             }
-        }
+        };
+        metrics::histogram!(CACHE_GET_LATENCY, ATTR_CACHE => CACHE_DISK).record(start.elapsed().as_micros() as f64);
+        result
     }
 
     async fn put_block(
@@ -497,37 +499,48 @@ impl DataCache for DiskDataCache {
         if block_offset != block_idx * self.config.block_size {
             return Err(DataCacheError::InvalidBlockOffset);
         }
-
+        let start = Instant::now();
         let bytes_len = bytes.len();
         let block_key = DiskBlockKey::new(&cache_key, block_idx);
         let path = self.get_path_for_block_key(&block_key);
         trace!(?cache_key, ?path, "new block will be created in disk cache");
 
-        let block = DiskBlock::new(cache_key, block_idx, block_offset, bytes).map_err(|err| match err {
-            DiskBlockCreationError::IntegrityError(_e) => DataCacheError::InvalidBlockContent,
-        })?;
-
-        {
-            let eviction_start = Instant::now();
-            let result = self.evict_if_needed();
-            metrics::histogram!("disk_data_cache.eviction_duration_us")
-                .record(eviction_start.elapsed().as_micros() as f64);
-            result
-        }?;
-
-        let write_start = Instant::now();
-        let (temp_file, size) = self.write_block(&path, block)?;
-        metrics::histogram!("disk_data_cache.write_duration_us").record(write_start.elapsed().as_micros() as f64);
-        metrics::counter!("disk_data_cache.total_bytes", "type" => "write").increment(bytes_len as u64);
-        if let Some(usage) = &self.usage {
-            let mut usage = usage.lock().unwrap();
-            _ = temp_file.persist(path).map_err(|e| e.error)?;
-            usage.add(block_key, size);
+        // Capture the put operation result separately from metrics recording
+        // to ensure we can record both success and error metrics consistently
+        let put_result = (|| -> DataCacheResult<()> {
+            let block = DiskBlock::new(cache_key, block_idx, block_offset, bytes).map_err(|err| match err {
+                DiskBlockCreationError::IntegrityError(_e) => DataCacheError::InvalidBlockContent,
+            })?;
+
+            {
+                let eviction_start = Instant::now();
+                let result = self.evict_if_needed();
+                metrics::histogram!(CACHE_EVICT_LATENCY, ATTR_CACHE => CACHE_DISK)
+                    .record(eviction_start.elapsed().as_micros() as f64);
+                result
+            }?;
+
+            let result = self.write_block(&path, block);
+            let (temp_file, size) = result?;
+
+            if let Some(usage) = &self.usage {
+                let mut usage = usage.lock().unwrap();
+                _ = temp_file.persist(path).map_err(|e| e.error)?;
+                usage.add(block_key, size);
+            } else {
+                _ = temp_file.persist(path).map_err(|e| e.error)?;
+            }
+
+            Ok(())
+        })();
+
+        if put_result.is_ok() {
+            metrics::counter!(CACHE_PUT_IO_SIZE, ATTR_CACHE => CACHE_DISK).increment(bytes_len as u64);
         } else {
-            _ = temp_file.persist(path).map_err(|e| e.error)?;
+            metrics::counter!(CACHE_PUT_ERRORS, ATTR_CACHE => CACHE_DISK).increment(1);
         }
-
-        Ok(())
+        metrics::histogram!(CACHE_PUT_LATENCY, ATTR_CACHE => CACHE_DISK).record(start.elapsed().as_micros() as f64);
+        put_result
     }
 
     fn block_size(&self) -> u64 {

mountpoint-s3-fs/src/data_cache/express_data_cache.rs

@@ -1,5 +1,9 @@
 use super::{BlockIndex, ChecksummedBytes, DataCache, DataCacheError, DataCacheResult};
 use crate::ServerSideEncryption;
+use crate::metrics::defs::{
+    ATTR_CACHE, CACHE_EXPRESS, CACHE_GET_ERRORS, CACHE_GET_IO_SIZE, CACHE_GET_LATENCY, CACHE_OVERSIZED_OBJECTS,
+    CACHE_PUT_ERRORS, CACHE_PUT_IO_SIZE, CACHE_PUT_LATENCY,
+};
 use crate::object::ObjectId;
 use std::collections::HashMap;
 use std::time::Instant;
@@ -167,7 +171,7 @@ where
         object_size: usize,
     ) -> DataCacheResult<Option<ChecksummedBytes>> {
         if object_size > self.config.max_object_size {
-            metrics::counter!("express_data_cache.over_max_object_size", "type" => "read").increment(1);
+            metrics::counter!(CACHE_OVERSIZED_OBJECTS, ATTR_CACHE => CACHE_EXPRESS).increment(1);
             return Ok(None);
         }
 
@@ -259,7 +263,7 @@ where
         object_size: usize,
     ) -> DataCacheResult<()> {
         if object_size > self.config.max_object_size {
-            metrics::counter!("express_data_cache.over_max_object_size", "type" => "write").increment(1);
+            metrics::counter!(CACHE_OVERSIZED_OBJECTS, ATTR_CACHE => CACHE_EXPRESS).increment(1);
             return Ok(());
         }
 
@@ -296,25 +300,18 @@ where
         object_size: usize,
     ) -> DataCacheResult<Option<ChecksummedBytes>> {
         let start = Instant::now();
-        let (result, result_type) = match self.read_block(cache_key, block_idx, block_offset, object_size).await {
+        let result = match self.read_block(cache_key, block_idx, block_offset, object_size).await {
             Ok(Some(data)) => {
-                metrics::counter!("express_data_cache.block_hit").increment(1);
-                metrics::counter!("express_data_cache.total_bytes", "type" => "read").increment(data.len() as u64);
-                (Ok(Some(data)), "ok")
-            }
-            Ok(None) => {
-                metrics::counter!("express_data_cache.block_hit").increment(0);
-                (Ok(None), "miss")
+                metrics::counter!(CACHE_GET_IO_SIZE, ATTR_CACHE => CACHE_EXPRESS).increment(data.len() as u64);
+                Ok(Some(data))
             }
+            Ok(None) => Ok(None),
             Err(err) => {
-                metrics::counter!("express_data_cache.block_hit").increment(0);
-                metrics::counter!("express_data_cache.block_err", "reason" => err.reason(), "type" => "read")
-                    .increment(1);
-                (Err(err), "error")
+                metrics::counter!(CACHE_GET_ERRORS, ATTR_CACHE => CACHE_EXPRESS).increment(1);
+                Err(err)
             }
         };
-        metrics::histogram!("express_data_cache.read_duration_us", "type" => result_type)
-            .record(start.elapsed().as_micros() as f64);
+        metrics::histogram!(CACHE_GET_LATENCY, ATTR_CACHE => CACHE_EXPRESS).record(start.elapsed().as_micros() as f64);
         result
     }
 
@@ -327,22 +324,21 @@ where
         object_size: usize,
     ) -> DataCacheResult<()> {
         let start = Instant::now();
-        let (result, result_type) = match self
+        let bytes_len = bytes.len();
+        let result = match self
             .write_block(cache_key, block_idx, block_offset, bytes, object_size)
             .await
         {
             Ok(()) => {
-                metrics::counter!("express_data_cache.total_bytes", "type" => "write").increment(object_size as u64);
-                (Ok(()), "ok")
+                metrics::counter!(CACHE_PUT_IO_SIZE, ATTR_CACHE => CACHE_EXPRESS).increment(bytes_len as u64);
+                Ok(())
             }
             Err(err) => {
-                metrics::counter!("express_data_cache.block_err", "reason" => err.reason(), "type" => "write")
-                    .increment(1);
-                (Err(err), "error")
+                metrics::counter!(CACHE_PUT_ERRORS, ATTR_CACHE => CACHE_EXPRESS).increment(1);
+                Err(err)
             }
         };
-        metrics::histogram!("express_data_cache.write_duration_us", "type" => result_type)
-            .record(start.elapsed().as_micros() as f64);
+        metrics::histogram!(CACHE_PUT_LATENCY, ATTR_CACHE => CACHE_EXPRESS).record(start.elapsed().as_micros() as f64);
         result
     }
 

mountpoint-s3-fs/src/metrics/defs.rs

@@ -29,8 +29,25 @@ pub const PROCESS_MEMORY_USAGE: &str = "process.memory_usage";
 
 pub const PREFETCH_RESET_STATE: &str = "prefetch.reset_state";
 
+// Cache metric constants
+pub const CACHE_GET_IO_SIZE: &str = "cache.get_io_size";
+pub const CACHE_PUT_IO_SIZE: &str = "cache.put_io_size";
+pub const CACHE_GET_LATENCY: &str = "cache.get_latency";
+pub const CACHE_PUT_LATENCY: &str = "cache.put_latency";
+pub const CACHE_GET_ERRORS: &str = "cache.get_errors";
+pub const CACHE_PUT_ERRORS: &str = "cache.put_errors";
+pub const CACHE_TOTAL_SIZE: &str = "cache.total_size";
+pub const CACHE_EVICT_LATENCY: &str = "cache.evict_latency";
+pub const CACHE_BLOCK_HIT: &str = "cache.block_hit";
+pub const CACHE_OVERSIZED_OBJECTS: &str = "cache.oversized_objects";
+
 // Attribute constants
 pub const ATTR_FUSE_REQUEST: &str = "fuse_request";
+pub const ATTR_CACHE: &str = "cache";
+
+// Cache type constants
+pub const CACHE_DISK: &str = "disk";
+pub const CACHE_EXPRESS: &str = "express";
 
 pub fn lookup_config(name: &str) -> MetricConfig {
     match name {

mountpoint-s3-fs/tests/fuse_tests/metrics_test.rs

@@ -7,6 +7,11 @@ use mountpoint_s3_fs::S3FilesystemConfig;
 use mountpoint_s3_fs::data_cache::{
     DiskDataCache, DiskDataCacheConfig, ExpressDataCache, ExpressDataCacheConfig, MultilevelDataCache,
 };
+use mountpoint_s3_fs::metrics::defs::{
+    ATTR_CACHE, CACHE_DISK, CACHE_EVICT_LATENCY, CACHE_EXPRESS, CACHE_GET_IO_SIZE, CACHE_GET_LATENCY,
+    CACHE_PUT_IO_SIZE, CACHE_PUT_LATENCY,
+};
+
 use mountpoint_s3_fs::metrics::defs::{
     ATTR_FUSE_REQUEST, FUSE_IO_SIZE, FUSE_REQUEST_ERRORS, FUSE_REQUEST_LATENCY, PREFETCH_RESET_STATE,
 };
@@ -239,23 +244,22 @@ rusty_fork_test! {
         test_session.client().put_object("file.txt", &content).unwrap();
         let path = test_session.mount_path().join("file.txt");
 
-        // First read - populate cache and wait for asynchronous cache updates
+        // First read - populate cache
         File::open(&path).unwrap().read_to_end(&mut Vec::new()).unwrap();
         sleep(Duration::from_millis(100));
 
         // No cache hits but cache should be updated
         assert!(recorder.get("fuse.cache_hit", &[]).is_none(), "Cache hit metric should not exist after cache miss");
-        assert_metric_exists(&recorder, "disk_data_cache.write_duration_us", &[]);
-        assert_metric_exists(&recorder, "disk_data_cache.total_bytes", &[("type", "write")]);
+        assert_metric_exists(&recorder, CACHE_PUT_LATENCY, &[(ATTR_CACHE, CACHE_DISK)]);
+        assert_metric_exists(&recorder, CACHE_PUT_IO_SIZE, &[(ATTR_CACHE, CACHE_DISK)]);
 
         // Second read - cache hit
         File::open(&path).unwrap().read_to_end(&mut Vec::new()).unwrap();
 
         // Verify cache hit metrics
         assert_metric_exists(&recorder, "fuse.cache_hit", &[]);
-        assert_metric_exists(&recorder, "disk_data_cache.block_hit", &[]);
-        assert_metric_exists(&recorder, "disk_data_cache.read_duration_us", &[]);
-        assert_metric_exists(&recorder, "disk_data_cache.total_bytes", &[("type", "read")]);
+        assert_metric_exists(&recorder, CACHE_GET_LATENCY, &[(ATTR_CACHE, CACHE_DISK)]);
+        assert_metric_exists(&recorder, CACHE_GET_IO_SIZE, &[(ATTR_CACHE, CACHE_DISK)]);
 
         // Read a large file to trigger eviction metrics
         let large_content = vec![b'z'; 2 * 1024 * 1024];
@@ -265,7 +269,7 @@ rusty_fork_test! {
         sleep(Duration::from_millis(100));
 
         // Verify eviction metrics
-        assert_metric_exists(&recorder, "disk_data_cache.eviction_duration_us", &[]);
+        assert_metric_exists(&recorder, CACHE_EVICT_LATENCY, &[(ATTR_CACHE, CACHE_DISK)]);
     }
 
     #[test]
@@ -304,22 +308,21 @@ rusty_fork_test! {
         sleep(Duration::from_millis(100));
 
         // Verify cache population metrics
-        // Currently, disk_data_cache.read_duration_us is only recorded on cache hits, not misses.
         // But they are recorded for express.
         assert!(recorder.get("fuse.cache_hit", &[]).is_none(), "Cache hit metric should not exist after cache miss");
-        assert_metric_exists(&recorder, "disk_data_cache.write_duration_us", &[]);
-        assert_metric_exists(&recorder, "disk_data_cache.total_bytes", &[("type", "write")]);
-        assert_metric_exists(&recorder, "express_data_cache.write_duration_us", &[("type", "ok")]);
-        assert_metric_exists(&recorder, "express_data_cache.total_bytes", &[("type", "write")]);
-        assert_metric_exists(&recorder, "express_data_cache.read_duration_us", &[("type", "miss")]);
+        assert_metric_exists(&recorder, CACHE_PUT_LATENCY, &[(ATTR_CACHE, CACHE_DISK)]);
+        assert_metric_exists(&recorder, CACHE_PUT_IO_SIZE, &[(ATTR_CACHE, CACHE_DISK)]);
+        assert_metric_exists(&recorder, CACHE_GET_LATENCY, &[(ATTR_CACHE, CACHE_DISK)]);
+        assert_metric_exists(&recorder, CACHE_PUT_LATENCY, &[(ATTR_CACHE, CACHE_EXPRESS)]);
+        assert_metric_exists(&recorder, CACHE_PUT_IO_SIZE, &[(ATTR_CACHE, CACHE_EXPRESS)]);
+        assert_metric_exists(&recorder, CACHE_GET_LATENCY, &[(ATTR_CACHE, CACHE_EXPRESS)]);
 
         // Second read should hit disk cache after async write completes
         File::open(&path).unwrap().read_to_end(&mut Vec::new()).unwrap();
         sleep(Duration::from_millis(100));
 
         assert_metric_exists(&recorder, "fuse.cache_hit", &[]);
-        assert_metric_exists(&recorder, "disk_data_cache.block_hit", &[]);
-        assert_metric_exists(&recorder, "disk_data_cache.read_duration_us", &[]);
-        assert_metric_exists(&recorder, "disk_data_cache.total_bytes", &[("type", "read")]);
+        assert_metric_exists(&recorder, CACHE_GET_LATENCY, &[(ATTR_CACHE, CACHE_DISK)]);
+        assert_metric_exists(&recorder, CACHE_GET_IO_SIZE, &[(ATTR_CACHE, CACHE_DISK)]);
     }
 }

mountpoint-s3/CHANGELOG.md

@@ -1,7 +1,8 @@
 ## Unreleased (v1.22.0)
 
 * Add metric to track cache hit rate in logs. ([#1716](https://github.com/awslabs/mountpoint-s3/pull/1716))
-* Remove redundant cache merics in logs. ([#1716](https://github.com/awslabs/mountpoint-s3/pull/1716))
+* Remove redundant cache merics in logs. ([#1716](https://github.com/awslabs/mountpoint-s3/pull/1716), [#1721](https://github.com/awslabs/mountpoint-s3/pull/1721))
+* Rename cache metrics for consistency. ([#1721](https://github.com/awslabs/mountpoint-s3/pull/1721))
 
 ## v1.21.0 (Oct 30, 2025)