diff --git a/rust/cubestore/cubestore/src/cachestore/cache_rocksstore.rs b/rust/cubestore/cubestore/src/cachestore/cache_rocksstore.rs index ec64edcdc744f..7b2ce7f850cc7 100644 --- a/rust/cubestore/cubestore/src/cachestore/cache_rocksstore.rs +++ b/rust/cubestore/cubestore/src/cachestore/cache_rocksstore.rs @@ -41,9 +41,17 @@ use std::time::Duration; use tokio::sync::broadcast::Sender; use tokio::task::JoinHandle; -pub(crate) struct RocksCacheStoreDetails {} +pub(crate) struct RocksCacheStoreDetails { + log_enabled: bool, +} impl RocksCacheStoreDetails { + pub fn new(config: &Arc) -> Self { + Self { + log_enabled: config.cachestore_log_enabled(), + } + } + pub fn get_compaction_state() -> CompactionPreloadedState { let mut indexes = HashMap::new(); @@ -157,6 +165,10 @@ impl RocksStoreDetails for RocksCacheStoreDetails { fn get_name(&self) -> &'static str { &"cachestore" } + + fn log_enabled(&self) -> bool { + self.log_enabled + } } pub struct RocksCacheStore { @@ -184,12 +196,13 @@ impl RocksCacheStore { metastore_fs: Arc, config: Arc, ) -> Result, CubeError> { + let details = Arc::new(RocksCacheStoreDetails::new(&config)); Self::new_from_store(RocksStore::with_listener( path, vec![], metastore_fs, config, - Arc::new(RocksCacheStoreDetails {}), + details, )?) } @@ -210,14 +223,10 @@ impl RocksCacheStore { metastore_fs: Arc, config: Arc, ) -> Result, CubeError> { - let store = RocksStore::load_from_dump( - path, - dump_path, - metastore_fs, - config, - Arc::new(RocksCacheStoreDetails {}), - ) - .await?; + let details = Arc::new(RocksCacheStoreDetails::new(&config)); + + let store = + RocksStore::load_from_dump(path, dump_path, metastore_fs, config, details).await?; Self::new_from_store(store) } @@ -227,8 +236,9 @@ impl RocksCacheStore { metastore_fs: Arc, config: Arc, ) -> Result, CubeError> { + let details = Arc::new(RocksCacheStoreDetails::new(&config)); let store = metastore_fs - .load_from_remote(&path, config, Arc::new(RocksCacheStoreDetails {})) + .load_from_remote(&path, config, details) .await?; Self::new_from_store(store) @@ -246,7 +256,7 @@ impl RocksCacheStore { let mut loops = vec![]; if self.store.config.upload_to_remote() { - let upload_interval = self.store.config.meta_store_log_upload_interval(); + let upload_interval = self.store.config.cachestore_log_upload_interval(); let cachestore = self.clone(); loops.push(cube_ext::spawn(async move { cachestore @@ -430,12 +440,13 @@ impl RocksCacheStore { let _ = std::fs::remove_dir_all(remote_store_path.clone()); - let details = Arc::new(RocksCacheStoreDetails {}); + let config_obj = config.config_obj(); + let details = Arc::new(RocksCacheStoreDetails::new(&config_obj)); let remote_fs = LocalDirRemoteFs::new(Some(remote_store_path.clone()), store_path.clone()); let store = RocksStore::new( store_path.clone().join(details.get_name()).as_path(), - BaseRocksStoreFs::new_for_cachestore(remote_fs.clone(), config.config_obj()), - config.config_obj(), + BaseRocksStoreFs::new_for_cachestore(remote_fs.clone(), config_obj.clone()), + config_obj, details, ) .unwrap(); diff --git a/rust/cubestore/cubestore/src/config/mod.rs b/rust/cubestore/cubestore/src/config/mod.rs index 2251b543f7b23..08f518b1fe2e6 100644 --- a/rust/cubestore/cubestore/src/config/mod.rs +++ b/rust/cubestore/cubestore/src/config/mod.rs @@ -430,6 +430,10 @@ pub trait ConfigObj: DIService { fn metastore_remote_address(&self) -> &Option; + fn cachestore_log_upload_interval(&self) -> u64; + + fn cachestore_log_enabled(&self) -> bool; + fn cachestore_rocksdb_config(&self) -> &RocksStoreConfig; fn cachestore_gc_loop_interval(&self) -> u64; @@ -581,6 +585,8 @@ pub struct ConfigObjImpl { pub metastore_bind_address: Option, pub metastore_remote_address: Option, pub metastore_rocks_store_config: RocksStoreConfig, + pub cachestore_log_upload_interval: u64, + pub cachestore_log_enabled: bool, pub cachestore_rocks_store_config: RocksStoreConfig, pub cachestore_gc_loop_interval: u64, pub cachestore_cache_eviction_loop_interval: u64, @@ -784,6 +790,14 @@ impl ConfigObj for ConfigObjImpl { &self.metastore_remote_address } + fn cachestore_log_upload_interval(&self) -> u64 { + self.cachestore_log_upload_interval + } + + fn cachestore_log_enabled(&self) -> bool { + self.cachestore_log_enabled + } + fn cachestore_rocksdb_config(&self) -> &RocksStoreConfig { &self.cachestore_rocks_store_config } @@ -1022,9 +1036,9 @@ fn env_bool(name: &str, default: bool) -> bool { env::var(name) .ok() .map(|x| match x.as_str() { - "0" => false, - "1" => true, - _ => panic!("expected '0' or '1' for '{}', found '{}'", name, &x), + "0" | "false" => false, + "1" | "true" => true, + _ => panic!("expected '0'/'1'/true/false for '{}', found '{}'", name, &x), }) .unwrap_or(default) } @@ -1329,6 +1343,13 @@ impl Config { }), metastore_remote_address: env::var("CUBESTORE_META_ADDR").ok(), metastore_rocks_store_config: RocksStoreConfig::metastore_default(), + cachestore_log_upload_interval: env_parse_duration( + "CACHESTORE_LOG_UPLOAD_INTERVAL", + 30, + Some(60), + Some(15), + ), + cachestore_log_enabled: env_bool("CACHESTORE_LOG_ENABLED", true), cachestore_rocks_store_config: RocksStoreConfig::cachestore_default(), cachestore_gc_loop_interval: env_parse_duration( "CUBESTORE_CACHESTORE_GC_LOOP", @@ -1559,6 +1580,8 @@ impl Config { metastore_bind_address: None, metastore_remote_address: None, metastore_rocks_store_config: RocksStoreConfig::metastore_default(), + cachestore_log_upload_interval: 30, + cachestore_log_enabled: true, cachestore_rocks_store_config: RocksStoreConfig::cachestore_default(), cachestore_gc_loop_interval: 30, cachestore_cache_eviction_loop_interval: 60, diff --git a/rust/cubestore/cubestore/src/metastore/mod.rs b/rust/cubestore/cubestore/src/metastore/mod.rs index 8584e043ef70b..77a8c12dd98d0 100644 --- a/rust/cubestore/cubestore/src/metastore/mod.rs +++ b/rust/cubestore/cubestore/src/metastore/mod.rs @@ -1335,6 +1335,10 @@ impl RocksStoreDetails for RocksMetaStoreDetails { fn get_name(&self) -> &'static str { &"metastore" } + + fn log_enabled(&self) -> bool { + true + } } pub struct RocksMetaStore { diff --git a/rust/cubestore/cubestore/src/metastore/rocks_store.rs b/rust/cubestore/cubestore/src/metastore/rocks_store.rs index 77c90d4a7b847..53898996ca3c7 100644 --- a/rust/cubestore/cubestore/src/metastore/rocks_store.rs +++ b/rust/cubestore/cubestore/src/metastore/rocks_store.rs @@ -785,6 +785,8 @@ pub trait RocksStoreDetails: Send + Sync { fn migrate(&self, table_ref: DbTableRef) -> Result<(), CubeError>; fn get_name(&self) -> &'static str; + + fn log_enabled(&self) -> bool; } #[derive(Clone)] @@ -1041,52 +1043,53 @@ impl RocksStore { pub async fn run_upload(&self) -> Result<(), CubeError> { let time = SystemTime::now(); - trace!("Persisting {} snapshot", self.details.get_name()); + trace!("Persisting {}", self.details.get_name()); let last_check_seq = self.last_check_seq().await; let last_db_seq = self.db.latest_sequence_number(); if last_check_seq == last_db_seq { - trace!( - "Persisting {} snapshot: nothing to update", - self.details.get_name() - ); + trace!("Persisting {}: nothing to update", self.details.get_name()); return Ok(()); } - let last_upload_seq = self.last_upload_seq().await; - let (serializer, min, max) = { - let updates = self.db.get_updates_since(last_upload_seq)?; - let mut serializer = WriteBatchContainer::new(); - - let mut seq_numbers = Vec::new(); - let size_limit = self.config.meta_store_log_upload_size_limit() as usize; - for update in updates.into_iter() { - let (n, write_batch) = update?; - seq_numbers.push(n); - write_batch.iterate(&mut serializer); - if serializer.size() > size_limit { - break; + if self.details.log_enabled() { + let last_upload_seq = self.last_upload_seq().await; + let (serializer, min, max) = { + let updates = self.db.get_updates_since(last_upload_seq)?; + let mut serializer = WriteBatchContainer::new(); + + let mut seq_numbers = Vec::new(); + let size_limit = self.config.meta_store_log_upload_size_limit() as usize; + for update in updates.into_iter() { + let (n, write_batch) = update?; + seq_numbers.push(n); + write_batch.iterate(&mut serializer); + if serializer.size() > size_limit { + break; + } } - } - ( - serializer, - seq_numbers.iter().min().map(|v| *v), - seq_numbers.iter().max().map(|v| *v), - ) - }; - if max.is_some() { - let snapshot_uploaded = self.snapshot_uploaded.read().await; - if *snapshot_uploaded { - let checkpoint_time = self.last_checkpoint_time.read().await; - let dir_name = format!("{}-logs", self.get_store_path(&checkpoint_time)); - self.metastore_fs - .upload_log(&dir_name, min.unwrap(), serializer) - .await?; + ( + serializer, + seq_numbers.iter().min().map(|v| *v), + seq_numbers.iter().max().map(|v| *v), + ) + }; + if max.is_some() { + let snapshot_uploaded = self.snapshot_uploaded.read().await; + if *snapshot_uploaded { + let checkpoint_time = self.last_checkpoint_time.read().await; + let dir_name = format!("{}-logs", self.get_store_path(&checkpoint_time)); + self.metastore_fs + .upload_log(&dir_name, min.unwrap(), serializer) + .await?; + } + let mut seq = self.last_upload_seq.write().await; + *seq = max.unwrap(); + self.write_completed_notify.notify_waiters(); } - let mut seq = self.last_upload_seq.write().await; - *seq = max.unwrap(); - self.write_completed_notify.notify_waiters(); + } else { + trace!("Persisting {}: logs are disabled", self.details.get_name()); } let last_checkpoint_time: SystemTime = self.last_checkpoint_time.read().await.clone();