feat(cache): add metrics for new cache

src/query/storages/cache/Cargo.toml

@@ -17,6 +17,7 @@ common-cache = { path = "../../../common/cache" }
 common-exception = { path = "../../../common/exception" }
 
 async-trait = { version = "0.1.57", package = "async-trait-fn" }
+metrics = "0.20.1"
 opendal = "0.23"
 parking_lot = "0.12.1"
 serde = { workspace = true }

src/query/storages/cache/src/lib.rs

@@ -20,6 +20,7 @@ mod settings;
 pub use cache::ObjectCacheProvider;
 pub use object::CachedObject;
 pub use object::CachedObjectAccessor;
+pub use providers::metrics_reset;
 pub use providers::ByPassCache;
 pub use providers::FileCache;
 pub use providers::MemoryBytesCache;

src/query/storages/cache/src/object.rs

@@ -43,11 +43,11 @@ impl CachedObject for Vec<u8> {
 /// 2. create the CachedObjectAccessor with the cache provider
 /// 3. operation with object
 pub struct CachedObjectAccessor<T> {
-    cache: Arc<dyn ObjectCacheProvider<T>>,
+    cache: Arc<dyn ObjectCacheProvider<T> + Send + Sync>,
 }
 
 impl<T> CachedObjectAccessor<T> {
-    pub fn create(cache: Arc<dyn ObjectCacheProvider<T>>) -> CachedObjectAccessor<T> {
+    pub fn create(cache: Arc<dyn ObjectCacheProvider<T> + Send + Sync>) -> CachedObjectAccessor<T> {
         Self { cache }
     }
 

src/query/storages/cache/src/providers/by_pass_cache.rs

@@ -13,10 +13,17 @@
 // limitations under the License.
 
 use std::sync::Arc;
+use std::time::Instant;
 
 use common_exception::Result;
 use opendal::Object;
 
+use crate::providers::metrics_inc_bypass_read_milliseconds;
+use crate::providers::metrics_inc_bypass_reads;
+use crate::providers::metrics_inc_bypass_remove_milliseconds;
+use crate::providers::metrics_inc_bypass_removes;
+use crate::providers::metrics_inc_bypass_write_milliseconds;
+use crate::providers::metrics_inc_bypass_writes;
 use crate::CacheSettings;
 use crate::CachedObject;
 use crate::ObjectCacheProvider;
@@ -35,17 +42,44 @@ impl<T> ObjectCacheProvider<T> for ByPassCache
 where T: CachedObject + Send + Sync + 'static
 {
     async fn read_object(&self, object: &Object, start: u64, end: u64) -> Result<Arc<T>> {
+        let now = Instant::now();
+
         let data = object.range_read(start..end).await?;
+
+        // Perf.
+        {
+            metrics_inc_bypass_reads(1);
+            metrics_inc_bypass_read_milliseconds(now.elapsed().as_millis() as u64);
+        }
+
         T::from_bytes(data)
     }
 
     async fn write_object(&self, object: &Object, v: Arc<T>) -> Result<()> {
+        let now = Instant::now();
+
         object.write(v.to_bytes()?).await?;
+
+        // Perf.
+        {
+            metrics_inc_bypass_writes(1);
+            metrics_inc_bypass_write_milliseconds(now.elapsed().as_millis() as u64);
+        }
+
         Ok(())
     }
 
     async fn remove_object(&self, object: &Object) -> Result<()> {
+        let now = Instant::now();
+
         object.delete().await?;
+
+        // Perf.
+        {
+            metrics_inc_bypass_removes(1);
+            metrics_inc_bypass_remove_milliseconds(now.elapsed().as_millis() as u64);
+        }
+
         Ok(())
     }
 }

src/query/storages/cache/src/providers/memory_bytes_cache.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use std::sync::Arc;
+use std::time::Instant;
 
 use common_cache::BytesMeter;
 use common_cache::Cache;
@@ -22,6 +23,13 @@ use common_exception::Result;
 use opendal::Object;
 use parking_lot::RwLock;
 
+use crate::providers::metrics_inc_memory_bytes_hits;
+use crate::providers::metrics_inc_memory_bytes_load_milliseconds;
+use crate::providers::metrics_inc_memory_bytes_misses;
+use crate::providers::metrics_inc_memory_bytes_remove_milliseconds;
+use crate::providers::metrics_inc_memory_bytes_removes;
+use crate::providers::metrics_inc_memory_bytes_write_milliseconds;
+use crate::providers::metrics_inc_memory_bytes_writes;
 use crate::CacheSettings;
 use crate::ObjectCacheProvider;
 
@@ -57,14 +65,29 @@ impl ObjectCacheProvider<Vec<u8>> for MemoryBytesCache {
         let try_get_val = self.get_cache(&key);
         Ok(match try_get_val {
             None => {
+                let now = Instant::now();
+
                 let data = object.range_read(start..end).await?;
                 let v = Arc::new(data);
 
                 self.lru.write().put(key, v.clone());
 
+                // Perf.
+                {
+                    metrics_inc_memory_bytes_misses(1);
+                    metrics_inc_memory_bytes_load_milliseconds(now.elapsed().as_millis() as u64);
+                }
+
+                v
+            }
+            Some(v) => {
+                // Perf.
+                {
+                    metrics_inc_memory_bytes_hits(1);
+                }
+
                 v
             }
-            Some(v) => v,
         })
     }
 
@@ -74,13 +97,34 @@ impl ObjectCacheProvider<Vec<u8>> for MemoryBytesCache {
             self.lru.write().put(key, v.clone());
         }
 
+        let now = Instant::now();
+
         object.write(v.as_slice()).await?;
 
+        // Perf.
+        {
+            metrics_inc_memory_bytes_writes(1);
+            metrics_inc_memory_bytes_write_milliseconds(now.elapsed().as_millis() as u64);
+        }
+
         Ok(())
     }
 
     async fn remove_object(&self, object: &Object) -> Result<()> {
+        let key = object.path();
+
+        let now = Instant::now();
+
+        // Try to remove from the cache.
+        self.lru.write().pop(key);
         object.delete().await?;
+
+        // Perf.
+        {
+            metrics_inc_memory_bytes_removes(1);
+            metrics_inc_memory_bytes_remove_milliseconds(now.elapsed().as_millis() as u64);
+        }
+
         Ok(())
     }
 }

src/query/storages/cache/src/providers/memory_items_cache.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use std::sync::Arc;
+use std::time::Instant;
 
 use common_cache::Cache;
 use common_cache::Count;
@@ -22,6 +23,13 @@ use common_exception::Result;
 use opendal::Object;
 use parking_lot::RwLock;
 
+use crate::providers::metrics_inc_memory_item_hits;
+use crate::providers::metrics_inc_memory_item_load_milliseconds;
+use crate::providers::metrics_inc_memory_item_misses;
+use crate::providers::metrics_inc_memory_item_remove_milliseconds;
+use crate::providers::metrics_inc_memory_item_removes;
+use crate::providers::metrics_inc_memory_item_write_milliseconds;
+use crate::providers::metrics_inc_memory_item_writes;
 use crate::CacheSettings;
 use crate::CachedObject;
 use crate::ObjectCacheProvider;
@@ -57,14 +65,29 @@ where T: CachedObject + Send + Sync
 
         Ok(match try_get_val {
             None => {
+                let now = Instant::now();
+
                 let data = object.range_read(start..end).await?;
                 let v = T::from_bytes(data)?;
                 // Write to cache.
                 self.lru.write().put(key, v.clone());
 
+                // Perf.
+                {
+                    metrics_inc_memory_item_misses(1);
+                    metrics_inc_memory_item_load_milliseconds(now.elapsed().as_millis() as u64);
+                }
+
+                v
+            }
+            Some(v) => {
+                // Perf.
+                {
+                    metrics_inc_memory_item_hits(1);
+                }
+
                 v
             }
-            Some(v) => v,
         })
     }
 
@@ -74,19 +97,37 @@ where T: CachedObject + Send + Sync
             self.lru.write().put(key, v.clone());
         }
 
+        let now = Instant::now();
+
         let data = v.to_bytes()?;
 
         object.write(data).await?;
+
+        // Perf.
+        {
+            metrics_inc_memory_item_writes(1);
+            metrics_inc_memory_item_write_milliseconds(now.elapsed().as_millis() as u64);
+        }
+
         Ok(())
     }
 
     async fn remove_object(&self, object: &Object) -> Result<()> {
         let key = object.path();
 
+        let now = Instant::now();
+
         // Try to remove from the cache.
         self.lru.write().pop(key);
 
         object.delete().await?;
+
+        // Perf.
+        {
+            metrics_inc_memory_item_removes(1);
+            metrics_inc_memory_item_remove_milliseconds(now.elapsed().as_millis() as u64);
+        }
+
         Ok(())
     }
 }