Merge remote-tracking branch 'origin/main' into wrj/metrics-js-udf-me…

…mory-usage

Signed-off-by: Runji Wang <[email protected]>

Cargo.toml

@@ -75,7 +75,7 @@ license = "Apache-2.0"
 repository = "https://github.com/risingwavelabs/risingwave"
 
 [workspace.dependencies]
-foyer = { version = "0.9.1", features = ["nightly"] }
+foyer = { version = "0.9.2", features = ["nightly"] }
 auto_enums = { version = "0.8", features = ["futures03", "tokio1"] }
 await-tree = "0.2.1"
 aws-config = { version = "1", default-features = false, features = [

grafana/common.py

@@ -514,11 +514,15 @@ def table_metric(name, filter=None):
 
 def quantile(f, percentiles):
     quantile_map = {
-        "60": ["0.6", "60"],
+        "10": ["0.1", "10"],
+        "25": ["0.25", "25"],
         "50": ["0.5", "50"],
+        "60": ["0.6", "60"],
+        "75": ["0.75", "75"],
         "90": ["0.9", "90"],
         "99": ["0.99", "99"],
         "999": ["0.999", "999"],
+        "100": ["1.0", "100"],
         "max": ["1.0", "max"],
     }
     return list(

grafana/risingwave-dev-dashboard.dashboard.py

@@ -1960,6 +1960,20 @@ def section_hummock_read(outer_panels):
                         ),
                     ],
                 ),
+                panels.timeseries_percentage(
+                    "Block Cache Efficiency",
+                    "Histogram of the estimated hit ratio of a block while in the block cache.",
+                    [
+                        *quantile(
+                            lambda quantile, legend: panels.target(
+                                f"clamp_max(histogram_quantile({quantile}, sum(rate({metric('block_efficiency_histogram_bucket')}[$__rate_interval])) by (le,{COMPONENT_LABEL},{NODE_LABEL})), 1)",
+                                f"block cache efficienfy - p{legend}"
+                                + " - {{%s}} @ {{%s}}" % (COMPONENT_LABEL, NODE_LABEL),
+                            ),
+                            [10, 25, 50, 75, 90, 100],
+                        ),
+                    ]
+                ),
                 panels.timeseries_ops(
                     "Iter keys flow",
                     "",

src/storage/src/hummock/sstable_store.rs

@@ -20,7 +20,9 @@ use std::sync::Arc;
 use await_tree::InstrumentAwait;
 use bytes::Bytes;
 use fail::fail_point;
-use foyer::{CacheContext, FetchState, HybridCache, HybridCacheBuilder, HybridCacheEntry};
+use foyer::{
+    CacheContext, EventListener, FetchState, HybridCache, HybridCacheBuilder, HybridCacheEntry,
+};
 use futures::{future, StreamExt};
 use itertools::Itertools;
 use risingwave_common::config::StorageMemoryConfig;
@@ -53,6 +55,31 @@ pub struct SstableBlockIndex {
     pub block_idx: u64,
 }
 
+pub struct BlockCacheEventListener {
+    metrics: Arc<HummockStateStoreMetrics>,
+}
+
+impl BlockCacheEventListener {
+    pub fn new(metrics: Arc<HummockStateStoreMetrics>) -> Self {
+        Self { metrics }
+    }
+}
+
+impl EventListener for BlockCacheEventListener {
+    type Key = SstableBlockIndex;
+    type Value = Box<Block>;
+
+    fn on_memory_release(&self, _key: Self::Key, value: Self::Value)
+    where
+        Self::Key: foyer::Key,
+        Self::Value: foyer::Value,
+    {
+        self.metrics
+            .block_efficiency_histogram
+            .observe(value.efficiency());
+    }
+}
+
 // TODO: Define policy based on use cases (read / compaction / ...).
 #[derive(Clone, Copy, Eq, PartialEq)]
 pub enum CachePolicy {

src/storage/src/monitor/hummock_state_store_metrics.rs

@@ -84,7 +84,7 @@ pub struct HummockStateStoreMetrics {
     pub old_value_size: IntGauge,
 
     // block statistics
-    pub block_efficiency_histogram: RelabeledHistogramVec,
+    pub block_efficiency_histogram: Histogram,
 
     pub event_handler_pending_event: IntGauge,
     pub event_handler_latency: HistogramVec,
@@ -417,13 +417,7 @@ impl HummockStateStoreMetrics {
             "Access ratio of in-memory block.",
             exponential_buckets(0.001, 2.0, 11).unwrap(),
         );
-        let block_efficiency_histogram =
-            register_histogram_vec_with_registry!(opts, &["table_id"], registry).unwrap();
-        let block_efficiency_histogram = RelabeledHistogramVec::with_metric_level(
-            MetricLevel::Info,
-            block_efficiency_histogram,
-            metric_level,
-        );
+        let block_efficiency_histogram = register_histogram_with_registry!(opts, registry).unwrap();
 
         let event_handler_pending_event = register_int_gauge_with_registry!(
             "state_store_event_handler_pending_event",

src/storage/src/store_impl.rs

@@ -30,8 +30,8 @@ use crate::error::StorageResult;
 use crate::filter_key_extractor::{RemoteTableAccessor, RpcFilterKeyExtractorManager};
 use crate::hummock::hummock_meta_client::MonitoredHummockMetaClient;
 use crate::hummock::{
-    Block, HummockError, HummockStorage, RecentFilter, Sstable, SstableBlockIndex, SstableStore,
-    SstableStoreConfig,
+    Block, BlockCacheEventListener, HummockError, HummockStorage, RecentFilter, Sstable,
+    SstableBlockIndex, SstableStore, SstableStoreConfig,
 };
 use crate::memory::sled::SledStateStore;
 use crate::memory::MemoryStateStore;
@@ -673,6 +673,9 @@ impl StateStoreImpl {
         let block_cache_v2 = {
             let mut builder = HybridCacheBuilder::new()
                 .with_name("foyer.data")
+                .with_event_listener(Arc::new(BlockCacheEventListener::new(
+                    state_store_metrics.clone(),
+                )))
                 .memory(opts.block_cache_capacity_mb * MB)
                 .with_shards(opts.block_cache_shard_num)
                 .with_eviction_config(opts.block_cache_eviction_config.clone())