feat(cubestore): Cache - auto trigger for full compaction (#7120)

Author: ovr

rust/cubestore/cubestore/src/cachestore/cache_eviction_manager.rs

@@ -150,6 +150,7 @@ pub struct CacheEvictionManager {
     // if ttl of a key is less then this value, key will be evicted
     // this help to delete upcoming keys for deleting
     eviction_min_ttl_threshold: u32,
+    compaction_trigger_size: u64,
     // background listener to track events
     _ttl_tl_loop_join_handle: Arc<AbortingJoinHandle<()>>,
 }
@@ -307,6 +308,7 @@ impl CacheEvictionManager {
             eviction_batch_size: config.cachestore_cache_eviction_batch_size(),
             eviction_below_threshold: config.cachestore_cache_eviction_below_threshold(),
             eviction_min_ttl_threshold: config.cachestore_cache_eviction_min_ttl_threshold(),
+            compaction_trigger_size: config.cachestore_cache_compaction_trigger_size(),
             //
             _ttl_tl_loop_join_handle: Arc::new(AbortingJoinHandle::new(join_handle)),
         }
@@ -506,6 +508,8 @@ impl CacheEvictionManager {
             } else {
                 log::trace!("Nothing to evict");
 
+                self.check_compaction_trigger(&store);
+
                 return Ok(EvictionResult::Finished(EvictionFinishedResult {
                     total_keys_removed: 0,
                     total_size_removed: 0,
@@ -518,6 +522,8 @@ impl CacheEvictionManager {
 
         let result = eviction_fut.await?;
 
+        self.check_compaction_trigger(&store);
+
         log::debug!(
             "Eviction finished, total_keys: {}, total_size: {}",
             self.get_stats_total_keys(),
@@ -1094,6 +1100,28 @@ impl CacheEvictionManager {
             Ok((weight, CACHE_ITEM_SIZE_WITHOUT_VALUE))
         }
     }
+
+    fn check_compaction_trigger(&self, store: &Arc<RocksStore>) {
+        let default_cf_metadata = store.db.get_column_family_metadata();
+
+        log::trace!(
+            "Compaction auto trigger, CF default size: {}",
+            humansize::format_size(default_cf_metadata.size, humansize::DECIMAL)
+        );
+
+        if default_cf_metadata.size > self.compaction_trigger_size {
+            let start: Option<&[u8]> = None;
+            let end: Option<&[u8]> = None;
+
+            log::debug!(
+                "Triggering compaction, CF default size: {} > {}",
+                humansize::format_size(default_cf_metadata.size, humansize::DECIMAL),
+                humansize::format_size(self.compaction_trigger_size, humansize::DECIMAL)
+            );
+
+            store.db.compact_range(start, end)
+        }
+    }
 }
 
 #[derive(Debug)]

rust/cubestore/cubestore/src/config/mod.rs

@@ -437,6 +437,8 @@ pub trait ConfigObj: DIService {
 
     fn cachestore_cache_max_size(&self) -> u64;
 
+    fn cachestore_cache_compaction_trigger_size(&self) -> u64;
+
     fn cachestore_cache_max_keys(&self) -> u32;
 
     fn cachestore_cache_eviction_policy(&self) -> CacheEvictionPolicy;
@@ -573,6 +575,7 @@ pub struct ConfigObjImpl {
     pub cachestore_cache_eviction_loop_interval: u64,
     pub cachestore_cache_ttl_persist_loop_interval: u64,
     pub cachestore_cache_max_size: u64,
+    pub cachestore_cache_compaction_trigger_size: u64,
     pub cachestore_cache_threshold_to_force_eviction: u8,
     pub cachestore_queue_results_expire: u64,
     pub cachestore_metrics_interval: u64,
@@ -780,6 +783,10 @@ impl ConfigObj for ConfigObjImpl {
         self.cachestore_cache_max_size
     }
 
+    fn cachestore_cache_compaction_trigger_size(&self) -> u64 {
+        self.cachestore_cache_compaction_trigger_size
+    }
+
     fn cachestore_cache_threshold_to_force_eviction(&self) -> u8 {
         self.cachestore_cache_threshold_to_force_eviction
     }
@@ -1098,6 +1105,20 @@ where
 }
 
 impl Config {
+    fn calculate_cache_compaction_trigger_size(cache_max_size: usize) -> usize {
+        match cache_max_size >> 20 {
+            // TODO: Enable this limits after moving to separate CF for cache
+            // d if d < 32 => 32 * 9,
+            // d if d < 64 => 64 * 8,
+            // d if d < 128 => 128 * 7,
+            // d if d < 256 => 256 * 6,
+            d if d < 512 => 512 * 5,
+            d if d < 1024 => cache_max_size * 4,
+            d if d < 4096 => cache_max_size * 3,
+            _ => cache_max_size * 2,
+        }
+    }
+
     pub fn default() -> Config {
         let query_timeout = env_parse("CUBESTORE_QUERY_TIMEOUT", 120);
         let query_cache_time_to_idle_secs = env_parse(
@@ -1106,6 +1127,23 @@ impl Config {
             60 * 60,
         );
 
+        let cachestore_cache_max_size = env_parse_size(
+            "CUBESTORE_CACHE_MAX_SIZE",
+            4096 << 20,
+            // 16384 mb
+            Some(16_384 << 20),
+            // 32 mb
+            Some(32 << 20),
+        );
+
+        let cachestore_cache_compaction_trigger_size = env_parse_size(
+            "CUBESTORE_CACHE_COMPACTION_TRIGGER_SIZE",
+            Self::calculate_cache_compaction_trigger_size(cachestore_cache_max_size),
+            None,
+            // 256 mb
+            Some(256 << 20),
+        ) as u64;
+
         Config {
             injector: Injector::new(),
             config_obj: Arc::new(ConfigObjImpl {
@@ -1256,14 +1294,8 @@ impl Config {
                     // 0 to disable
                     Some(0),
                 ),
-                cachestore_cache_max_size: env_parse_size(
-                    "CUBESTORE_CACHE_MAX_SIZE",
-                    4096 << 20,
-                    // 16384 mb
-                    Some(16384 << 20),
-                    // 32 mb
-                    Some(32 << 20),
-                ) as u64,
+                cachestore_cache_max_size: cachestore_cache_max_size as u64,
+                cachestore_cache_compaction_trigger_size,
                 cachestore_cache_threshold_to_force_eviction: 25,
                 cachestore_queue_results_expire: env_parse_duration(
                     "CUBESTORE_QUEUE_RESULTS_EXPIRE",
@@ -1474,7 +1506,8 @@ impl Config {
                 cachestore_gc_loop_interval: 30,
                 cachestore_cache_eviction_loop_interval: 60,
                 cachestore_cache_ttl_persist_loop_interval: 15,
-                cachestore_cache_max_size: 4096 << 32,
+                cachestore_cache_max_size: 4096 << 20,
+                cachestore_cache_compaction_trigger_size: 4096 * 2 << 20,
                 cachestore_cache_threshold_to_force_eviction: 25,
                 cachestore_queue_results_expire: 90,
                 cachestore_metrics_interval: 15,