Skip to content

feat(cubestore): CacheStore - allow to disable logs #8027

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ovr merged 3 commits into from
Mar 28, 2024
Merged

feat(cubestore): CacheStore - allow to disable logs #8027

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e162b3a
feat(cubestore): CacheStore - allow to disable logs
ovr Mar 27, 2024
5e4fd71
chore: correct name
ovr Mar 28, 2024
4266c5e
chore: support true/false
ovr Mar 28, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 26 additions & 15 deletions rust/cubestore/cubestore/src/cachestore/cache_rocksstore.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<dyn ConfigObj>) -> Self {
Self {
log_enabled: config.cachestore_log_enabled(),
}
}

pub fn get_compaction_state() -> CompactionPreloadedState {
let mut indexes = HashMap::new();

Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -184,12 +196,13 @@ impl RocksCacheStore {
metastore_fs: Arc<dyn MetaStoreFs>,
config: Arc<dyn ConfigObj>,
) -> Result<Arc<Self>, CubeError> {
let details = Arc::new(RocksCacheStoreDetails::new(&config));
Self::new_from_store(RocksStore::with_listener(
path,
vec![],
metastore_fs,
config,
Arc::new(RocksCacheStoreDetails {}),
details,
)?)
}

Expand All @@ -210,14 +223,10 @@ impl RocksCacheStore {
metastore_fs: Arc<dyn MetaStoreFs>,
config: Arc<dyn ConfigObj>,
) -> Result<Arc<Self>, 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)
}
Expand All @@ -227,8 +236,9 @@ impl RocksCacheStore {
metastore_fs: Arc<dyn MetaStoreFs>,
config: Arc<dyn ConfigObj>,
) -> Result<Arc<Self>, 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)
Expand All @@ -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
Expand Down Expand Up @@ -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();
Expand Down
29 changes: 26 additions & 3 deletions rust/cubestore/cubestore/src/config/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -430,6 +430,10 @@ pub trait ConfigObj: DIService {

fn metastore_remote_address(&self) -> &Option<String>;

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;
Expand Down Expand Up @@ -581,6 +585,8 @@ pub struct ConfigObjImpl {
pub metastore_bind_address: Option<String>,
pub metastore_remote_address: Option<String>,
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,
Expand Down Expand Up @@ -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
}
Expand Down Expand Up @@ -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)
}
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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,
Expand Down
4 changes: 4 additions & 0 deletions rust/cubestore/cubestore/src/metastore/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1335,6 +1335,10 @@ impl RocksStoreDetails for RocksMetaStoreDetails {
fn get_name(&self) -> &'static str {
&"metastore"
}

fn log_enabled(&self) -> bool {
true
}
}

pub struct RocksMetaStore {
Expand Down
75 changes: 39 additions & 36 deletions rust/cubestore/cubestore/src/metastore/rocks_store.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)]
Expand Down Expand Up @@ -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();
Expand Down