chore(cubestore): Set metastore deletion batch size using CUBESTORE_SNAPSHOTS_DELETION_BATCH_SIZE

Author: srh

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

@@ -518,6 +518,8 @@ pub trait ConfigObj: DIService {
 
     fn dump_dir(&self) -> &Option<PathBuf>;
 
+    fn snapshots_deletion_batch_size(&self) -> u64;
+
     fn minimum_metastore_snapshots_count(&self) -> u64;
 
     fn metastore_snapshots_lifetime(&self) -> u64;
@@ -630,6 +632,7 @@ pub struct ConfigObjImpl {
     pub drop_ws_processing_messages_after_secs: u64,
     pub drop_ws_complete_messages_after_secs: u64,
     pub skip_kafka_parsing_errors: bool,
+    pub snapshots_deletion_batch_size: u64,
     pub minimum_metastore_snapshots_count: u64,
     pub metastore_snapshots_lifetime: u64,
     pub minimum_cachestore_snapshots_count: u64,
@@ -953,6 +956,10 @@ impl ConfigObj for ConfigObjImpl {
         &self.dump_dir
     }
 
+    fn snapshots_deletion_batch_size(&self) -> u64 {
+        self.snapshots_deletion_batch_size
+    }
+
     fn minimum_metastore_snapshots_count(&self) -> u64 {
         self.minimum_metastore_snapshots_count
     }
@@ -1486,6 +1493,11 @@ impl Config {
                     10 * 60,
                 ),
                 skip_kafka_parsing_errors: env_parse("CUBESTORE_SKIP_KAFKA_PARSING_ERRORS", false),
+                // Presently, not useful to make more than upload_concurrency times constant
+                snapshots_deletion_batch_size: env_parse(
+                    "CUBESTORE_SNAPSHOTS_DELETION_BATCH_SIZE",
+                    80,
+                ),
                 minimum_metastore_snapshots_count: env_parse(
                     "CUBESTORE_MINIMUM_METASTORE_SNAPSHOTS_COUNT",
                     5,
@@ -1652,6 +1664,7 @@ impl Config {
                 drop_ws_processing_messages_after_secs: 60,
                 drop_ws_complete_messages_after_secs: 10,
                 skip_kafka_parsing_errors: false,
+                snapshots_deletion_batch_size: 80,
                 minimum_metastore_snapshots_count: 3,
                 metastore_snapshots_lifetime: 24 * 3600,
                 minimum_cachestore_snapshots_count: 3,

rust/cubestore/cubestore/src/metastore/rocks_fs.rs

@@ -55,7 +55,8 @@ pub struct BaseRocksStoreFs {
     name: &'static str,
     minimum_snapshots_count: u64,
     snapshots_lifetime: u64,
-    remote_files_cleanup_batch_size: u64,
+    // A copy of the upload-concurrency config -- we multiply this for our deletes.
+    snapshots_deletion_batch_size: u64,
 }
 
 impl BaseRocksStoreFs {
@@ -65,13 +66,13 @@ impl BaseRocksStoreFs {
     ) -> Arc<Self> {
         let minimum_snapshots_count = config.minimum_metastore_snapshots_count();
         let snapshots_lifetime = config.metastore_snapshots_lifetime();
-        let remote_files_cleanup_batch_size = config.remote_files_cleanup_batch_size();
+        let snapshots_deletion_batch_size = config.snapshots_deletion_batch_size();
         Arc::new(Self {
             remote_fs,
             name: "metastore",
             minimum_snapshots_count,
             snapshots_lifetime,
-            remote_files_cleanup_batch_size,
+            snapshots_deletion_batch_size,
         })
     }
     pub fn new_for_cachestore(
@@ -80,13 +81,13 @@ impl BaseRocksStoreFs {
     ) -> Arc<Self> {
         let minimum_snapshots_count = config.minimum_cachestore_snapshots_count();
         let snapshots_lifetime = config.cachestore_snapshots_lifetime();
-        let remote_files_cleanup_batch_size = config.remote_files_cleanup_batch_size();
+        let snapshots_deletion_batch_size = config.snapshots_deletion_batch_size();
         Arc::new(Self {
             remote_fs,
             name: "cachestore",
             minimum_snapshots_count,
             snapshots_lifetime,
-            remote_files_cleanup_batch_size,
+            snapshots_deletion_batch_size,
         })
     }
 
@@ -145,8 +146,8 @@ impl BaseRocksStoreFs {
         name: &str,
     ) -> Result<HashMap<u128, Vec<String>>, CubeError> {
         let existing_metastore_files = remote_fs.list(format!("{}-", name)).await?;
-        // Log a debug statement so that we can rule out the filename list itself being too large for memory.
-        log::debug!(
+        // Log an info statement so that we can rule out the filename list itself being too large for memory.
+        log::info!(
             "Listed existing {} files, count = {}",
             name,
             existing_metastore_files.len()
@@ -215,7 +216,7 @@ impl BaseRocksStoreFs {
         }
 
         for batch in to_delete.chunks(
-            self.remote_files_cleanup_batch_size
+            self.snapshots_deletion_batch_size
                 .try_into()
                 .unwrap_or(usize::MAX),
         ) {