cache: record per-level stats for client-side raw reads

src/cache/multilevel.rs

@@ -470,6 +470,24 @@ impl MultiLevelStorage {
         Ok(())
     }
 
+    fn record_read_hit(&self, hit_level: usize, duration: Duration) {
+        inc_stat!(self.atomic_stats.get(hit_level), hits, 1);
+        inc_stat!(
+            self.atomic_stats.get(hit_level),
+            hit_duration_nanos,
+            duration.as_nanos() as u64
+        );
+        for miss_idx in 0..hit_level {
+            inc_stat!(self.atomic_stats.get(miss_idx), misses, 1);
+        }
+    }
+
+    fn record_all_read_misses(&self) {
+        for idx in 0..self.levels.len() {
+            inc_stat!(self.atomic_stats.get(idx), misses, 1);
+        }
+    }
+
     /// Write to levels starting from `start_idx` asynchronously
     async fn write_remaining_levels_async(&self, key: &str, data: &Bytes, start_idx: usize) {
         for (idx, level) in self.levels.iter().enumerate().skip(start_idx) {
@@ -530,17 +548,7 @@ impl Storage for MultiLevelStorage {
                     let duration = start.elapsed();
                     debug!("Cache hit at level {} in {:?}", idx, duration);
 
-                    // Update stats
-                    inc_stat!(self.atomic_stats.get(idx), hits, 1);
-                    inc_stat!(
-                        self.atomic_stats.get(idx),
-                        hit_duration_nanos,
-                        duration.as_nanos() as u64
-                    );
-                    // Mark misses for all levels checked before this hit
-                    for miss_idx in 0..idx {
-                        inc_stat!(self.atomic_stats.get(miss_idx), misses, 1);
-                    }
+                    self.record_read_hit(idx, duration);
 
                     // If hit at level > 0, backfill to faster levels (L0 to L(idx-1))
                     if idx > 0 {
@@ -620,20 +628,30 @@ impl Storage for MultiLevelStorage {
         }
         debug!("Cache miss at all levels");
 
-        // Mark final miss for all checked levels
-        for idx in 0..self.levels.len() {
-            inc_stat!(self.atomic_stats.get(idx), misses, 1);
-        }
+        self.record_all_read_misses();
 
         Ok(Cache::Miss)
     }
 
     async fn get_raw(&self, key: &str) -> Result<Option<Bytes>> {
-        for level in &self.levels {
-            if let Some(bytes) = level.get_raw(key).await? {
-                return Ok(Some(bytes));
+        for (idx, level) in self.levels.iter().enumerate() {
+            let start = Instant::now();
+            match level.get_raw(key).await {
+                Ok(Some(bytes)) => {
+                    self.record_read_hit(idx, start.elapsed());
+                    return Ok(Some(bytes));
+                }
+                Ok(None) => {}
+                Err(error) => {
+                    warn!(
+                        "Error checking cache level {}: {}, trying next level",
+                        idx, error
+                    );
+                }
             }
         }
+
+        self.record_all_read_misses();
         Ok(None)
     }
 

src/cache/multilevel_test.rs

@@ -1080,6 +1080,10 @@ impl Storage for CheckFailingStorage {
         Err(anyhow!("intentional check failure"))
     }
 
+    async fn get_raw(&self, _key: &str) -> Result<Option<Bytes>> {
+        Err(anyhow!("intentional raw read failure"))
+    }
+
     fn location(&self) -> String {
         "CheckFailingStorage".to_owned()
     }
@@ -1715,6 +1719,73 @@ fn test_multilevel_get_raw_skips_levels_without_raw_support() {
     });
 }
 
+#[test]
+fn test_multilevel_get_raw_records_per_level_stats() {
+    let runtime = RuntimeBuilder::new_current_thread()
+        .enable_all()
+        .build()
+        .unwrap();
+    let l0 = Arc::new(InMemoryStorage::new());
+    let l1 = Arc::new(InMemoryStorage::new());
+    let storage =
+        MultiLevelStorage::new(vec![l0 as Arc<dyn Storage>, l1.clone() as Arc<dyn Storage>]);
+
+    runtime.block_on(async {
+        l1.put("hit", CacheWrite::default()).await.unwrap();
+        assert!(storage.get_raw("hit").await.unwrap().is_some());
+    });
+
+    let stats = storage.stats();
+    assert_eq!(stats.0[0].hits, 0);
+    assert_eq!(stats.0[0].misses, 1);
+    assert_eq!(stats.0[1].hits, 1);
+    assert_eq!(stats.0[1].misses, 0);
+}
+
+#[test]
+fn test_multilevel_get_raw_continues_after_level_error() {
+    let runtime = RuntimeBuilder::new_current_thread()
+        .enable_all()
+        .build()
+        .unwrap();
+    let l0 = Arc::new(CheckFailingStorage);
+    let l1 = Arc::new(InMemoryStorage::new());
+    let storage =
+        MultiLevelStorage::new(vec![l0 as Arc<dyn Storage>, l1.clone() as Arc<dyn Storage>]);
+
+    runtime.block_on(async {
+        l1.put("hit", CacheWrite::default()).await.unwrap();
+        assert!(storage.get_raw("hit").await.unwrap().is_some());
+    });
+
+    let stats = storage.stats();
+    assert_eq!(stats.0[0].hits, 0);
+    assert_eq!(stats.0[0].misses, 1);
+    assert_eq!(stats.0[1].hits, 1);
+    assert_eq!(stats.0[1].misses, 0);
+}
+
+#[test]
+fn test_multilevel_get_raw_records_all_level_misses() {
+    let runtime = RuntimeBuilder::new_current_thread()
+        .enable_all()
+        .build()
+        .unwrap();
+    let l0 = Arc::new(InMemoryStorage::new());
+    let l1 = Arc::new(InMemoryStorage::new());
+    let storage = MultiLevelStorage::new(vec![l0 as Arc<dyn Storage>, l1 as Arc<dyn Storage>]);
+
+    runtime.block_on(async {
+        assert!(storage.get_raw("missing").await.unwrap().is_none());
+    });
+
+    let stats = storage.stats();
+    assert_eq!(stats.0[0].hits, 0);
+    assert_eq!(stats.0[0].misses, 1);
+    assert_eq!(stats.0[1].hits, 0);
+    assert_eq!(stats.0[1].misses, 1);
+}
+
 #[test]
 fn test_multilevel_put_raw_writes_to_all_levels() {
     // Verifies that MultiLevelStorage::put_raw propagates the raw bytes