fix: move some storage metrics to store instead of each engine (#1057)

Author: MrCroxx

foyer-common/src/metrics.rs

@@ -36,12 +36,14 @@ pub struct Metrics {
     pub storage_enqueue: BoxedCounter,
     pub storage_hit: BoxedCounter,
     pub storage_miss: BoxedCounter,
+    pub storage_throttled: BoxedCounter,
     pub storage_delete: BoxedCounter,
     pub storage_error: BoxedCounter,
 
     pub storage_enqueue_duration: BoxedHistogram,
     pub storage_hit_duration: BoxedHistogram,
     pub storage_miss_duration: BoxedHistogram,
+    pub storage_throttled_duration: BoxedHistogram,
     pub storage_delete_duration: BoxedHistogram,
 
     pub storage_queue_rotate: BoxedCounter,
@@ -217,11 +219,13 @@ impl Metrics {
         let storage_hit = foyer_storage_op_total.counter(&[name.clone(), "hit".into()]);
         let storage_miss = foyer_storage_op_total.counter(&[name.clone(), "miss".into()]);
         let storage_delete = foyer_storage_op_total.counter(&[name.clone(), "delete".into()]);
+        let storage_throttled = foyer_storage_op_total.counter(&[name.clone(), "throttled".into()]);
         let storage_error = foyer_storage_op_total.counter(&[name.clone(), "error".into()]);
 
         let storage_enqueue_duration = foyer_storage_op_duration.histogram(&[name.clone(), "enqueue".into()]);
         let storage_hit_duration = foyer_storage_op_duration.histogram(&[name.clone(), "hit".into()]);
         let storage_miss_duration = foyer_storage_op_duration.histogram(&[name.clone(), "miss".into()]);
+        let storage_throttled_duration = foyer_storage_op_duration.histogram(&[name.clone(), "throttled".into()]);
         let storage_delete_duration = foyer_storage_op_duration.histogram(&[name.clone(), "delete".into()]);
 
         let storage_queue_rotate = foyer_storage_inner_op_total.counter(&[name.clone(), "queue_rotate".into()]);
@@ -303,11 +307,13 @@ impl Metrics {
             storage_enqueue,
             storage_hit,
             storage_miss,
+            storage_throttled,
             storage_delete,
             storage_error,
             storage_enqueue_duration,
             storage_hit_duration,
             storage_miss_duration,
+            storage_throttled_duration,
             storage_delete_duration,
             storage_queue_rotate,
             storage_queue_rotate_duration,

foyer-storage/src/large/generic.rs

@@ -355,8 +355,6 @@ where
     fn load(&self, hash: u64) -> impl Future<Output = Result<Load<K, V>>> + Send + 'static {
         tracing::trace!(hash, "[lodc]: load");
 
-        let now = Instant::now();
-
         let indexer = self.inner.indexer.clone();
         let metrics = self.inner.metrics.clone();
         let region_manager = self.inner.region_manager.clone();
@@ -365,8 +363,6 @@ where
             let addr = match indexer.get(hash) {
                 Some(addr) => addr,
                 None => {
-                    metrics.storage_miss.increase(1);
-                    metrics.storage_miss_duration.record(now.elapsed().as_secs_f64());
                     return Ok(Load::Miss);
                 }
             };
@@ -381,13 +377,10 @@ where
                 Err(e @ Error::InvalidIoRange { .. }) => {
                     tracing::warn!(?e, "[lodc load]: invalid io range, remove this entry and skip");
                     indexer.remove(hash);
-                    metrics.storage_miss.increase(1);
-                    metrics.storage_miss_duration.record(now.elapsed().as_secs_f64());
                     return Ok(Load::Miss);
                 }
                 Err(e) => {
                     tracing::error!(hash, ?addr, ?e, "[lodc load]: load error");
-                    metrics.storage_error.increase(1);
                     return Err(e);
                 }
             }
@@ -406,13 +399,10 @@ where
                         "[lodc load]: deserialize read buffer raise error, remove this entry and skip"
                     );
                     indexer.remove(hash);
-                    metrics.storage_miss.increase(1);
-                    metrics.storage_miss_duration.record(now.elapsed().as_secs_f64());
                     return Ok(Load::Miss);
                 }
                 Err(e) => {
                     tracing::error!(hash, ?addr, ?e, "[lodc load]: load error");
-                    metrics.storage_error.increase(1);
                     return Err(e);
                 }
             };
@@ -439,14 +429,10 @@ where
                             "[lodc load]: deserialize read buffer raise error, remove this entry and skip"
                         );
                         indexer.remove(hash);
-                        metrics.storage_miss.increase(1);
-                        metrics.storage_miss_duration.record(now.elapsed().as_secs_f64());
-                        metrics.storage_error.increase(1);
                         return Ok(Load::Miss);
                     }
                     Err(e) => {
                         tracing::error!(hash, ?addr, ?header, ?e, "[lodc load]: load error");
-                        metrics.storage_error.increase(1);
                         return Err(e);
                     }
                 };
@@ -456,9 +442,6 @@ where
                 res
             };
 
-            metrics.storage_hit.increase(1);
-            metrics.storage_hit_duration.record(now.elapsed().as_secs_f64());
-
             let age = match region.statistics().probation.load(Ordering::Relaxed) {
                 true => Age::Old,
                 false => Age::Young,
@@ -476,8 +459,6 @@ where
     }
 
     fn delete(&self, hash: u64) {
-        let now = Instant::now();
-
         if !self.inner.active.load(Ordering::Relaxed) {
             tracing::warn!("cannot delete entry after closed");
             return;
@@ -500,12 +481,6 @@ where
                 stats,
             });
         });
-
-        self.inner.metrics.storage_delete.increase(1);
-        self.inner
-            .metrics
-            .storage_miss_duration
-            .record(now.elapsed().as_secs_f64());
     }
 
     fn may_contains(&self, hash: u64) -> bool {

foyer-storage/src/small/generic.rs

@@ -92,7 +92,7 @@ where
 
     active: AtomicBool,
 
-    metrics: Arc<Metrics>,
+    _metrics: Arc<Metrics>,
     _runtime: Runtime,
 }
 
@@ -155,7 +155,7 @@ where
             device: config.device,
             set_manager,
             active: AtomicBool::new(true),
-            metrics,
+            _metrics: metrics,
             _runtime: config.runtime,
         };
         let inner = Arc::new(inner);
@@ -189,15 +189,13 @@ where
 
     fn load(&self, hash: u64) -> impl Future<Output = Result<Load<K, V>>> + Send + 'static {
         let set_manager = self.inner.set_manager.clone();
-        let metrics = self.inner.metrics.clone();
 
         async move {
             set_manager
                 .load(hash)
                 .await
                 .inspect_err(|e| {
                     tracing::error!(hash, ?e, "[sodc load]: fail to load");
-                    metrics.storage_error.increase(1);
                 })
                 .map(|o| match o {
                     Some((key, value)) => Load::Entry {

foyer-storage/src/store.rs

@@ -215,10 +215,17 @@ where
     where
         Q: Hash + Equivalent<K> + ?Sized + Send + Sync + 'static,
     {
+        let now = Instant::now();
+
         let hash = self.inner.hasher.hash_one(key);
 
         #[cfg(feature = "test_utils")]
         if self.inner.load_throttle_switch.is_throttled() {
+            self.inner.metrics.storage_throttled.increase(1);
+            self.inner
+                .metrics
+                .storage_throttled_duration
+                .record(now.elapsed().as_secs_f64());
             return Ok(Load::Throttled);
         }
 
@@ -228,8 +235,18 @@ where
                 Pick::Reject => unreachable!(),
                 Pick::Throttled(_) => {
                     if self.inner.engine.may_contains(hash) {
+                        self.inner.metrics.storage_throttled.increase(1);
+                        self.inner
+                            .metrics
+                            .storage_throttled_duration
+                            .record(now.elapsed().as_secs_f64());
                         return Ok(Load::Throttled);
                     } else {
+                        self.inner.metrics.storage_miss.increase(1);
+                        self.inner
+                            .metrics
+                            .storage_miss_duration
+                            .record(now.elapsed().as_secs_f64());
                         return Ok(Load::Miss);
                     }
                 }
@@ -242,14 +259,38 @@ where
                 key: k,
                 value: v,
                 populated: p,
-            }) if key.equivalent(&k) => Ok(Load::Entry {
-                key: k,
-                value: v,
-                populated: p,
-            }),
-            Ok(Load::Entry { .. }) | Ok(Load::Miss) => Ok(Load::Miss),
-            Ok(Load::Throttled) => Ok(Load::Throttled),
-            Err(e) => Err(e),
+            }) if key.equivalent(&k) => {
+                self.inner.metrics.storage_hit.increase(1);
+                self.inner
+                    .metrics
+                    .storage_hit_duration
+                    .record(now.elapsed().as_secs_f64());
+                Ok(Load::Entry {
+                    key: k,
+                    value: v,
+                    populated: p,
+                })
+            }
+            Ok(Load::Entry { .. }) | Ok(Load::Miss) => {
+                self.inner.metrics.storage_miss.increase(1);
+                self.inner
+                    .metrics
+                    .storage_miss_duration
+                    .record(now.elapsed().as_secs_f64());
+                Ok(Load::Miss)
+            }
+            Ok(Load::Throttled) => {
+                self.inner.metrics.storage_throttled.increase(1);
+                self.inner
+                    .metrics
+                    .storage_throttled_duration
+                    .record(now.elapsed().as_secs_f64());
+                Ok(Load::Throttled)
+            }
+            Err(e) => {
+                self.inner.metrics.storage_error.increase(1);
+                Err(e)
+            }
         }
     }
 
@@ -258,8 +299,16 @@ where
     where
         Q: Hash + Equivalent<K> + ?Sized,
     {
+        let now = Instant::now();
+
         let hash = self.inner.hasher.hash_one(key);
-        self.inner.engine.delete(hash)
+        self.inner.engine.delete(hash);
+
+        self.inner.metrics.storage_delete.increase(1);
+        self.inner
+            .metrics
+            .storage_delete_duration
+            .record(now.elapsed().as_secs_f64());
     }
 
     /// Check if the disk cache contains a cached entry with the given key.