feat(cubestore): add 'no upload' mode (#2405)

We avoid any uploads of metastore or remote filesystem in this mode.
To enable, set `CUBESTORE_NO_UPLOAD` environment variable to any value.

Useful for debugging, not for production use.

rust/cubestore/src/config/mod.rs

@@ -209,6 +209,8 @@ pub trait ConfigObj: DIService {
     fn server_name(&self) -> &String;
 
     fn max_ingestion_data_frames(&self) -> usize;
+
+    fn upload_to_remote(&self) -> bool;
 }
 
 #[derive(Debug, Clone)]
@@ -233,6 +235,7 @@ pub struct ConfigObjImpl {
     pub connection_timeout: u64,
     pub server_name: String,
     pub max_ingestion_data_frames: usize,
+    pub upload_to_remote: bool,
 }
 
 crate::di_service!(ConfigObjImpl, [ConfigObj]);
@@ -318,6 +321,10 @@ impl ConfigObj for ConfigObjImpl {
     fn max_ingestion_data_frames(&self) -> usize {
         self.max_ingestion_data_frames
     }
+
+    fn upload_to_remote(&self) -> bool {
+        self.upload_to_remote
+    }
 }
 
 lazy_static! {
@@ -411,6 +418,7 @@ impl Config {
                 server_name: env::var("CUBESTORE_SERVER_NAME")
                     .ok()
                     .unwrap_or("localhost".to_string()),
+                upload_to_remote: !env::var("CUBESTORE_NO_UPLOAD").ok().is_some(),
             }),
         }
     }
@@ -447,6 +455,7 @@ impl Config {
                 wal_split_threshold: 262144,
                 connection_timeout: 60,
                 server_name: "localhost".to_string(),
+                upload_to_remote: true,
             }),
         }
     }

rust/cubestore/src/metastore/mod.rs

@@ -1675,6 +1675,10 @@ impl RocksMetaStore {
     }
 
     pub async fn wait_upload_loop(meta_store: Arc<Self>) {
+        if !meta_store.config.upload_to_remote() {
+            log::info!("Not running metastore upload loop");
+            return;
+        }
         meta_store
             .upload_loop
             .process(
@@ -2088,6 +2092,11 @@ impl RocksMetaStore {
 #[async_trait]
 impl MetaStore for RocksMetaStore {
     async fn wait_for_current_seq_to_sync(&self) -> Result<(), CubeError> {
+        if !self.config.upload_to_remote() {
+            return Err(CubeError::internal(
+                "waiting for metastore to upload in noupload mode".to_string(),
+            ));
+        }
         while self.has_pending_changes().await? {
             tokio::time::timeout(
                 Duration::from_secs(30),

rust/cubestore/src/remotefs/queue.rs

@@ -291,6 +291,12 @@ impl RemoteFs for QueueRemoteFs {
         local_upload_path: &str,
         remote_path: &str,
     ) -> Result<(), CubeError> {
+        if !self.config.upload_to_remote() {
+            return Err(CubeError::internal(format!(
+                "Refusing to upload {}",
+                remote_path
+            )));
+        }
         let mut receiver = self.result_sender.subscribe();
         self.upload_queue.push(RemoteFsOp::Upload {
             temp_upload_path: local_upload_path.to_string(),
@@ -334,6 +340,12 @@ impl RemoteFs for QueueRemoteFs {
     }
 
     async fn delete_file(&self, remote_path: &str) -> Result<(), CubeError> {
+        if !self.config.upload_to_remote() {
+            return Err(CubeError::internal(format!(
+                "Refusing to delete {}",
+                remote_path
+            )));
+        }
         let mut receiver = self.result_sender.subscribe();
         self.upload_queue
             .push(RemoteFsOp::Delete(remote_path.to_string()));