chore(cubestore): Remove old local files refactoring (#6688)

Author: waralexrom

rust/cubestore/cubestore/src/app_metrics.rs

@@ -56,3 +56,6 @@ pub static METASTORE_READ_OUT_QUEUE_OPERATION: Histogram =
 
 /// RemoteFs metrics
 pub static REMOTE_FS_OPERATION_CORE: Counter = metrics::counter("cs.remote_fs.operations.core");
+pub static REMOTE_FS_FILES_TO_REMOVE: Gauge = metrics::gauge("cs.remote_fs.files_to_remove.count");
+pub static REMOTE_FS_FILES_SIZE_TO_REMOVE: Gauge =
+    metrics::gauge("cs.remote_fs.files_to_remove.size");

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

@@ -20,6 +20,7 @@ use crate::metastore::{
 use crate::mysql::{MySqlServer, SqlAuthDefaultImpl, SqlAuthService};
 use crate::queryplanner::query_executor::{QueryExecutor, QueryExecutorImpl};
 use crate::queryplanner::{QueryPlanner, QueryPlannerImpl};
+use crate::remotefs::cleanup::RemoteFsCleanup;
 use crate::remotefs::gcs::GCSRemoteFs;
 use crate::remotefs::minio::MINIORemoteFs;
 use crate::remotefs::queue::QueueRemoteFs;
@@ -109,6 +110,14 @@ impl CubeServices {
             QueueRemoteFs::wait_processing_loops(remote_fs.clone()).await
         }));
 
+        if self.injector.has_service_typed::<RemoteFsCleanup>().await {
+            let cleanup = self.injector.get_service_typed::<RemoteFsCleanup>().await;
+            futures.push(cube_ext::spawn(async move {
+                cleanup.wait_local_cleanup_loop().await;
+                Ok(())
+            }));
+        }
+
         if !self.cluster.is_select_worker() {
             let rocks_meta_store = self.rocks_meta_store.clone().unwrap();
             futures.push(cube_ext::spawn(async move {
@@ -220,6 +229,11 @@ impl CubeServices {
             scheduler.stop_processing_loops()?;
         }
 
+        if self.injector.has_service_typed::<RemoteFsCleanup>().await {
+            let cleanup = self.injector.get_service_typed::<RemoteFsCleanup>().await;
+            cleanup.stop();
+        }
+
         if self
             .injector
             .has_service_typed::<CacheStoreSchedulerImpl>()
@@ -470,7 +484,13 @@ pub trait ConfigObj: DIService {
 
     fn local_files_cleanup_interval_secs(&self) -> u64;
 
+    fn remote_files_cleanup_interval_secs(&self) -> u64;
+
     fn local_files_cleanup_size_threshold(&self) -> u64;
+
+    fn local_files_cleanup_delay_secs(&self) -> u64;
+
+    fn remote_files_cleanup_delay_secs(&self) -> u64;
 }
 
 #[derive(Debug, Clone)]
@@ -544,7 +564,10 @@ pub struct ConfigObjImpl {
     pub transport_max_message_size: usize,
     pub transport_max_frame_size: usize,
     pub local_files_cleanup_interval_secs: u64,
+    pub remote_files_cleanup_interval_secs: u64,
     pub local_files_cleanup_size_threshold: u64,
+    pub local_files_cleanup_delay_secs: u64,
+    pub remote_files_cleanup_delay_secs: u64,
 }
 
 crate::di_service!(ConfigObjImpl, [ConfigObj]);
@@ -812,9 +835,21 @@ impl ConfigObj for ConfigObjImpl {
         self.local_files_cleanup_interval_secs
     }
 
+    fn remote_files_cleanup_interval_secs(&self) -> u64 {
+        self.remote_files_cleanup_interval_secs
+    }
+
     fn local_files_cleanup_size_threshold(&self) -> u64 {
         self.local_files_cleanup_size_threshold
     }
+
+    fn local_files_cleanup_delay_secs(&self) -> u64 {
+        self.local_files_cleanup_delay_secs
+    }
+
+    fn remote_files_cleanup_delay_secs(&self) -> u64 {
+        self.remote_files_cleanup_delay_secs
+    }
 }
 
 lazy_static! {
@@ -1187,14 +1222,26 @@ impl Config {
                 ),
                 local_files_cleanup_interval_secs: env_parse(
                     "CUBESTORE_LOCAL_FILES_CLEANUP_INTERVAL_SECS",
-                    3 * 600,
+                    600,
+                ),
+                remote_files_cleanup_interval_secs: env_parse(
+                    "CUBESTORE_REMOTE_FILES_CLEANUP_INTERVAL_SECS",
+                    6 * 600,
                 ),
                 local_files_cleanup_size_threshold: env_parse_size(
                     "CUBESTORE_LOCAL_FILES_CLEANUP_SIZE_THRESHOLD",
                     1024 * 1024 * 1024,
                     None,
                     None,
                 ) as u64,
+                local_files_cleanup_delay_secs: env_parse(
+                    "CUBESTORE_LOCAL_FILES_CLEANUP_DELAY_SECS",
+                    600,
+                ),
+                remote_files_cleanup_delay_secs: env_parse(
+                    "CUBESTORE_REMOTE_FILES_CLEANUP_DELAY_SECS",
+                    3600,
+                ),
             }),
         }
     }
@@ -1280,7 +1327,10 @@ impl Config {
                 transport_max_message_size: 64 << 20,
                 transport_max_frame_size: 16 << 20,
                 local_files_cleanup_interval_secs: 600,
+                remote_files_cleanup_interval_secs: 600,
                 local_files_cleanup_size_threshold: 0,
+                local_files_cleanup_delay_secs: 600,
+                remote_files_cleanup_delay_secs: 3600,
             }),
         }
     }
@@ -1739,6 +1789,16 @@ impl Config {
                 ))
             })
             .await;
+
+        self.injector
+            .register_typed::<RemoteFsCleanup, _, _, _>(async move |i| {
+                Arc::new(RemoteFsCleanup::new(
+                    i.get_service_typed().await,
+                    i.get_service_typed().await,
+                    i.get_service_typed().await,
+                ))
+            })
+            .await;
     }
 
     pub async fn configure_common(&self) {

rust/cubestore/cubestore/src/metastore/mod.rs

@@ -969,6 +969,8 @@ pub trait MetaStore: DIService + Send + Sync {
         &self,
     ) -> Result<Vec<(IdRow<Partition>, Vec<IdRow<Chunk>>)>, CubeError>;
 
+    async fn get_all_filenames(&self) -> Result<Vec<String>, CubeError>;
+
     fn chunks_table(&self) -> ChunkMetaStoreTable;
     async fn create_chunk(
         &self,
@@ -3313,6 +3315,26 @@ impl MetaStore for RocksMetaStore {
         })
         .await
     }
+    async fn get_all_filenames(&self) -> Result<Vec<String>, CubeError> {
+        self.read_operation_out_of_queue(|db| {
+            let mut filenames = Vec::new();
+            for c in ChunkRocksTable::new(db.clone()).table_scan(db.snapshot)? {
+                let c = c?;
+                if !c.row.in_memory {
+                    filenames.push(c.row.get_full_name(c.id));
+                }
+            }
+
+            for p in PartitionRocksTable::new(db.clone()).table_scan(db.snapshot)? {
+                let p = p?;
+                if let Some(filename) = p.row.get_full_name(p.id) {
+                    filenames.push(filename);
+                }
+            }
+            Ok(filenames)
+        })
+        .await
+    }
 
     #[tracing::instrument(level = "trace", skip(self))]
     async fn create_chunk(

rust/cubestore/cubestore/src/queryplanner/test_utils.rs

@@ -735,6 +735,10 @@ impl MetaStore for MetaStoreMock {
     ) -> Result<Vec<IdRow<Index>>, CubeError> {
         panic!("MetaStore mock!")
     }
+
+    async fn get_all_filenames(&self) -> Result<Vec<String>, CubeError> {
+        panic!("MetaStore mock!")
+    }
 }
 
 crate::di_service!(MetaStoreMock, [MetaStore]);