From 87bec2e6a6b53e2bf0b48baec1d4f48194a17b64 Mon Sep 17 00:00:00 2001 From: weisd Date: Sat, 26 Apr 2025 17:55:30 +0800 Subject: [PATCH] fix rebalance/decom --- ecstore/src/bitrot.rs | 5 +- ecstore/src/cache_value/metacache_set.rs | 7 +- ecstore/src/disk/local.rs | 2 +- ecstore/src/heal/data_usage_cache.rs | 3 +- ecstore/src/notification_sys.rs | 4 +- ecstore/src/peer_rest_client.rs | 2 + ecstore/src/pools.rs | 91 +++++++++++++++++++++--- ecstore/src/rebalance.rs | 70 ++++++++++++++---- ecstore/src/set_disk.rs | 2 +- rustfs/src/admin/handlers/rebalance.rs | 41 ++++++----- rustfs/src/grpc.rs | 12 +++- 11 files changed, 192 insertions(+), 47 deletions(-) diff --git a/ecstore/src/bitrot.rs b/ecstore/src/bitrot.rs index 68ccff71e..b64c84b3b 100644 --- a/ecstore/src/bitrot.rs +++ b/ecstore/src/bitrot.rs @@ -661,7 +661,10 @@ impl ReadAt for BitrotFileReader { } if offset != self.curr_offset { - error!("BitrotFileReader read_at offset != self.curr_offset, {} != {}", offset, self.curr_offset); + error!( + "BitrotFileReader read_at {}/{} offset != self.curr_offset, {} != {}", + &self.volume, &self.file_path, offset, self.curr_offset + ); return Err(Error::new(DiskError::Unexpected)); } diff --git a/ecstore/src/cache_value/metacache_set.rs b/ecstore/src/cache_value/metacache_set.rs index 48174d64f..e716a3051 100644 --- a/ecstore/src/cache_value/metacache_set.rs +++ b/ecstore/src/cache_value/metacache_set.rs @@ -294,7 +294,12 @@ pub async fn list_path_raw(mut rx: B_Receiver, opts: ListPathRawOptions) - jobs.push(revjob); - let _ = join_all(jobs).await; + let results = join_all(jobs).await; + for result in results { + if let Err(err) = result { + error!("list_path_raw err {:?}", err); + } + } Ok(()) } diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index 74cdaa690..45abdbf5f 100644 --- a/ecstore/src/disk/local.rs +++ b/ecstore/src/disk/local.rs @@ -455,7 +455,7 @@ impl LocalDisk { { if let Err(aerr) = access(volume_dir.as_ref()).await { if os_is_not_exist(&aerr) { - warn!("read_metadata_with_dmtime os err {:?}", &aerr); + // warn!("read_metadata_with_dmtime os err {:?}", &aerr); return Err(Error::new(DiskError::VolumeNotFound)); } } diff --git a/ecstore/src/heal/data_usage_cache.rs b/ecstore/src/heal/data_usage_cache.rs index 2dad9f55b..50a570f5e 100644 --- a/ecstore/src/heal/data_usage_cache.rs +++ b/ecstore/src/heal/data_usage_cache.rs @@ -18,7 +18,6 @@ use std::path::Path; use std::time::{Duration, SystemTime}; use tokio::sync::mpsc::Sender; use tokio::time::sleep; -use tracing::warn; use super::data_scanner::{SizeSummary, DATA_SCANNER_FORCE_COMPACT_AT_FOLDERS}; use super::data_usage::{BucketTargetUsageInfo, BucketUsageInfo, DataUsageInfo}; @@ -402,7 +401,7 @@ impl DataUsageCache { break; } Err(err) => { - warn!("Failed to load data usage cache from backend: {}", &err); + // warn!("Failed to load data usage cache from backend: {}", &err); match err.downcast_ref::() { Some(DiskError::FileNotFound) | Some(DiskError::VolumeNotFound) => { match store diff --git a/ecstore/src/notification_sys.rs b/ecstore/src/notification_sys.rs index d364cc281..ed4649069 100644 --- a/ecstore/src/notification_sys.rs +++ b/ecstore/src/notification_sys.rs @@ -9,7 +9,7 @@ use lazy_static::lazy_static; use madmin::{ItemState, ServerProperties}; use std::sync::OnceLock; use std::time::SystemTime; -use tracing::error; +use tracing::{error, warn}; lazy_static! { pub static ref GLOBAL_NotificationSys: OnceLock = OnceLock::new(); @@ -151,6 +151,8 @@ impl NotificationSys { for result in results { if let Err(err) = result { error!("notification load_rebalance_meta err {:?}", err); + } else { + warn!("notification load_rebalance_meta success"); } } } diff --git a/ecstore/src/peer_rest_client.rs b/ecstore/src/peer_rest_client.rs index 71f2e3f27..9a121c9b3 100644 --- a/ecstore/src/peer_rest_client.rs +++ b/ecstore/src/peer_rest_client.rs @@ -663,6 +663,8 @@ impl PeerRestClient { let request = Request::new(LoadRebalanceMetaRequest { start_rebalance }); let response = client.load_rebalance_meta(request).await?.into_inner(); + + warn!("load_rebalance_meta response {:?}", response); if !response.success { if let Some(msg) = response.error_info { return Err(Error::msg(msg)); diff --git a/ecstore/src/pools.rs b/ecstore/src/pools.rs index 3e868f19b..f6f996941 100644 --- a/ecstore/src/pools.rs +++ b/ecstore/src/pools.rs @@ -2,6 +2,7 @@ use crate::bucket::versioning_sys::BucketVersioningSys; use crate::cache_value::metacache_set::{list_path_raw, ListPathRawOptions}; use crate::config::com::{read_config, save_config, CONFIG_PREFIX}; use crate::config::error::ConfigError; +use crate::disk::error::is_err_volume_not_found; use crate::disk::{MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams, BUCKET_META_PREFIX, RUSTFS_META_BUCKET}; use crate::heal::data_usage::DATA_USAGE_CACHE_NAME; use crate::heal::heal_commands::HealOpts; @@ -609,6 +610,12 @@ impl ECStore { let mut lock = self.pool_meta.write().await; if lock.decommission_cancel(idx) { lock.save(self.pools.clone()).await?; + + drop(lock); + + if let Some(notification_sys) = get_global_notification_sys() { + notification_sys.reload_pool_meta().await; + } } Ok(()) @@ -628,6 +635,7 @@ impl ECStore { #[tracing::instrument(skip(self, rx))] pub async fn decommission(&self, rx: B_Receiver, indices: Vec) -> Result<()> { + warn!("decommission: {:?}", indices); if indices.is_empty() { return Err(Error::msg("errInvalidArgument")); } @@ -662,8 +670,10 @@ impl ECStore { wk: Arc, rcfg: Option, ) { + warn!("decommission_entry: {} {}", &bucket, &entry.name); wk.give().await; if entry.is_dir() { + warn!("decommission_entry: skip dir {}", &entry.name); return; } @@ -782,6 +792,9 @@ impl ECStore { } }; + let bucket_name = bucket.clone(); + let object_name = rd.object_info.name.clone(); + if let Err(err) = self.clone().decommission_object(idx, bucket, rd).await { if is_err_object_not_found(&err) || is_err_version_not_found(&err) || is_err_data_movement_overwrite(&err) { ignore = true; @@ -794,6 +807,11 @@ impl ECStore { continue; } + warn!( + "decommission_pool: decommission_object done {}/{} {}", + &bucket_name, &object_name, &version.name + ); + failure = false; break; } @@ -841,10 +859,16 @@ impl ECStore { .update_after(idx, self.pools.clone(), Duration::seconds(30)) .await .unwrap_or_default(); + + drop(pool_meta); if ok { - // TODO: ReloadPoolMeta + if let Some(notification_sys) = get_global_notification_sys() { + notification_sys.reload_pool_meta().await; + } } } + + warn!("decommission_pool: decommission_entry done {} {}", &bucket, &entry.name); } #[tracing::instrument(skip(self, rx))] @@ -872,6 +896,8 @@ impl ECStore { for (set_idx, set) in pool.disk_set.iter().enumerate() { wk.clone().take().await; + warn!("decommission_pool: decommission_pool {} {}", set_idx, &bi.name); + let decommission_entry: ListCallback = Arc::new({ let this = Arc::clone(self); let bucket = bi.name.clone(); @@ -895,20 +921,39 @@ impl ECStore { tokio::spawn(async move { loop { if rx.try_recv().is_ok() { + warn!("decommission_pool: cancel {}", set_id); break; } - if let Err(err) = set + warn!("decommission_pool: list_objects_to_decommission {} {}", set_id, &bi.name); + + match set .list_objects_to_decommission(rx.resubscribe(), bi.clone(), decommission_entry.clone()) .await { - error!("decommission_pool: list_objects_to_decommission {} err {:?}", set_id, &err); + Ok(_) => { + warn!("decommission_pool: list_objects_to_decommission {} done", set_id); + break; + } + Err(err) => { + error!("decommission_pool: list_objects_to_decommission {} err {:?}", set_id, &err); + if is_err_volume_not_found(&err) { + warn!("decommission_pool: list_objects_to_decommission {} volume not found", set_id); + break; + } + + tokio::time::sleep(tokio::time::Duration::from_secs(5)).await; + } } } }); } + warn!("decommission_pool: decommission_pool wait {} {}", idx, &bi.name); + wk.wait().await; + warn!("decommission_pool: decommission_pool done {} {}", idx, &bi.name); + Ok(()) } @@ -918,11 +963,15 @@ impl ECStore { error!("decom err {:?}", &err); if let Err(er) = self.decommission_failed(idx).await { error!("decom failed err {:?}", &er); + } else { + warn!("decommission: decommission_failed {}", idx); } return; } + warn!("decommission: decommission_in_background complete {}", idx); + let (failed, cmd_line) = { let pool_meta = self.pool_meta.read().await; let failed = { @@ -944,6 +993,8 @@ impl ECStore { } else if let Err(er) = self.complete_decommission(idx).await { error!("decom complete err {:?}", &er); } + + warn!("Decommissioning complete for pool {}", cmd_line); } #[tracing::instrument(skip(self))] @@ -955,6 +1006,9 @@ impl ECStore { let mut pool_meta = self.pool_meta.write().await; if pool_meta.decommission_failed(idx) { pool_meta.save(self.pools.clone()).await?; + + drop(pool_meta); + if let Some(notification_sys) = get_global_notification_sys() { notification_sys.reload_pool_meta().await; } @@ -972,6 +1026,7 @@ impl ECStore { let mut pool_meta = self.pool_meta.write().await; if pool_meta.decommission_complete(idx) { pool_meta.save(self.pools.clone()).await?; + drop(pool_meta); if let Some(notification_sys) = get_global_notification_sys() { notification_sys.reload_pool_meta().await; } @@ -996,7 +1051,7 @@ impl ECStore { }; if is_decommissioned { - info!("decommission: already done, moving on {}", bucket.to_string()); + warn!("decommission: already done, moving on {}", bucket.to_string()); { let mut pool_meta = self.pool_meta.write().await; @@ -1009,7 +1064,7 @@ impl ECStore { continue; } - info!("decommission: currently on bucket {}", &bucket.name); + warn!("decommission: currently on bucket {}", &bucket.name); if let Err(err) = self .decommission_pool(rx.resubscribe(), idx, pool.clone(), bucket.clone()) @@ -1017,6 +1072,8 @@ impl ECStore { { error!("decommission: decommission_pool err {:?}", &err); return Err(err); + } else { + warn!("decommission: decommission_pool done {}", &bucket.name); } { @@ -1026,6 +1083,8 @@ impl ECStore { error!("decom pool_meta.save err {:?}", err); } } + + warn!("decommission: decommission_pool bucket_done {}", &bucket.name); } } @@ -1107,6 +1166,7 @@ impl ECStore { #[tracing::instrument(skip(self, rd))] async fn decommission_object(self: Arc, pool_idx: usize, bucket: String, rd: GetObjectReader) -> Result<()> { + warn!("decommission_object: {} {}", &bucket, &rd.object_info.name); let object_info = rd.object_info.clone(); // TODO: check : use size or actual_size ? @@ -1269,10 +1329,23 @@ impl SetDisks { min_disks: listing_quorum, partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option]| { let resolver = resolver.clone(); - if let Ok(Some(entry)) = entries.resolve(resolver) { - cb_func(entry) - } else { - Box::pin(async {}) + let cb_func = cb_func.clone(); + + match entries.resolve(resolver) { + Ok(Some(entry)) => { + warn!("decommission_pool: list_objects_to_decommission get {}", &entry.name); + Box::pin(async move { + cb_func(entry).await; + }) + } + Ok(None) => { + warn!("decommission_pool: list_objects_to_decommission get none"); + Box::pin(async {}) + } + Err(err) => { + error!("decommission_pool: list_objects_to_decommission get err {:?}", &err); + Box::pin(async {}) + } } })), ..Default::default() diff --git a/ecstore/src/rebalance.rs b/ecstore/src/rebalance.rs index 56f10f0d0..62c5d696f 100644 --- a/ecstore/src/rebalance.rs +++ b/ecstore/src/rebalance.rs @@ -20,7 +20,7 @@ use http::HeaderMap; use serde::{Deserialize, Serialize}; use tokio::sync::broadcast::{self, Receiver as B_Receiver}; use tokio::time::{Duration, Instant}; -use tracing::{error, info}; +use tracing::{error, info, warn}; use uuid::Uuid; use workers::workers::Workers; @@ -162,6 +162,7 @@ impl RebalanceMeta { pub async fn load_with_opts(&mut self, store: Arc, opts: ObjectOptions) -> Result<()> { let (data, _) = read_config_with_metadata(store, REBAL_META_NAME, &opts).await?; if data.is_empty() { + warn!("rebalanceMeta: no data"); return Ok(()); } if data.len() <= 4 { @@ -183,6 +184,7 @@ impl RebalanceMeta { self.last_refreshed_at = Some(SystemTime::now()); + warn!("rebalanceMeta: loaded meta done"); Ok(()) } @@ -214,20 +216,33 @@ impl ECStore { #[tracing::instrument(skip_all)] pub async fn load_rebalance_meta(&self) -> Result<()> { let mut meta = RebalanceMeta::new(); + warn!("rebalanceMeta: load rebalance meta"); match meta.load(self.pools[0].clone()).await { Ok(_) => { - let mut rebalance_meta = self.rebalance_meta.write().await; + warn!("rebalanceMeta: rebalance meta loaded0"); + { + let mut rebalance_meta = self.rebalance_meta.write().await; - *rebalance_meta = Some(meta); + *rebalance_meta = Some(meta); + + drop(rebalance_meta); + } + + warn!("rebalanceMeta: rebalance meta loaded1"); if let Err(err) = self.update_rebalance_stats().await { error!("Failed to update rebalance stats: {}", err); + } else { + warn!("rebalanceMeta: rebalance meta loaded2"); } } Err(err) => { if !is_err_config_not_found(&err) { + error!("rebalanceMeta: load rebalance meta err {:?}", &err); return Err(err); } + + error!("rebalanceMeta: not found, rebalance not started"); } } @@ -237,21 +252,24 @@ impl ECStore { #[tracing::instrument(skip_all)] pub async fn update_rebalance_stats(&self) -> Result<()> { let mut ok = false; - let mut rebalance_meta = self.rebalance_meta.write().await; for i in 0..self.pools.len() { if self.find_index(i).await.is_none() { + let mut rebalance_meta = self.rebalance_meta.write().await; if let Some(meta) = rebalance_meta.as_mut() { meta.pool_stats.push(RebalanceStats::default()); } ok = true; + drop(rebalance_meta); } } if ok { + let mut rebalance_meta = self.rebalance_meta.write().await; if let Some(meta) = rebalance_meta.as_mut() { meta.save(self.pools[0].clone()).await?; } + drop(rebalance_meta); } Ok(()) @@ -267,6 +285,7 @@ impl ECStore { #[tracing::instrument(skip(self))] pub async fn init_rebalance_meta(&self, bucktes: Vec) -> Result { + warn!("init_rebalance_meta: start rebalance"); let si = self.storage_info().await; let mut disk_stats = vec![DiskStat::default(); self.pools.len()]; @@ -321,10 +340,15 @@ impl ECStore { meta.save(self.pools[0].clone()).await?; + warn!("init_rebalance_meta: rebalance meta saved"); + let id = meta.id.clone(); - let mut rebalance_meta = self.rebalance_meta.write().await; - *rebalance_meta = Some(meta); + { + let mut rebalance_meta = self.rebalance_meta.write().await; + *rebalance_meta = Some(meta); + drop(rebalance_meta); + } Ok(id) } @@ -424,6 +448,7 @@ impl ECStore { #[tracing::instrument(skip_all)] pub async fn start_rebalance(self: &Arc) { + warn!("start_rebalance: start rebalance"); // let rebalance_meta = self.rebalance_meta.read().await; let (tx, rx) = broadcast::channel::(1); @@ -433,13 +458,17 @@ impl ECStore { if let Some(meta) = rebalance_meta.as_mut() { meta.cancel = Some(tx) } else { + error!("start_rebalance: rebalance_meta is None exit"); return; } + + drop(rebalance_meta); } let participants = { if let Some(ref meta) = *self.rebalance_meta.read().await { if meta.stopped_at.is_some() { + warn!("start_rebalance: rebalance already stopped exit"); return; } @@ -457,6 +486,7 @@ impl ECStore { for (idx, participating) in participants.iter().enumerate() { if !*participating { + warn!("start_rebalance: pool {} is not participating, skipping", idx); continue; } @@ -465,6 +495,7 @@ impl ECStore { .get(idx) .map_or(true, |v| v.endpoints.as_ref().first().map_or(true, |e| e.is_local)) { + warn!("start_rebalance: pool {} is not local, skipping", idx); continue; } @@ -479,6 +510,8 @@ impl ECStore { } }); } + + warn!("start_rebalance: rebalance started done"); } #[tracing::instrument(skip(self, rx))] @@ -540,6 +573,7 @@ impl ECStore { } if quit { + warn!("{}: exiting save_task", msg); return; } @@ -547,13 +581,14 @@ impl ECStore { } }); - tracing::info!("Pool {} rebalancing is started", pool_index + 1); + tracing::warn!("Pool {} rebalancing is started", pool_index + 1); while let Some(bucket) = self.next_rebal_bucket(pool_index).await? { tracing::info!("Rebalancing bucket: {}", bucket); if let Err(err) = self.rebalance_bucket(rx.resubscribe(), bucket.clone(), pool_index).await { if err.to_string().contains("not initialized") { + warn!("rebalance_bucket: rebalance not initialized, continue"); continue; } tracing::error!("Error rebalancing bucket {}: {:?}", bucket, err); @@ -564,7 +599,7 @@ impl ECStore { self.bucket_rebalance_done(pool_index, bucket).await?; } - tracing::info!("Pool {} rebalancing is done", pool_index + 1); + tracing::warn!("Pool {} rebalancing is done", pool_index + 1); done_tx.send(Ok(())).await.ok(); save_task.await.ok(); @@ -1037,10 +1072,21 @@ impl SetDisks { partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option]| { // let cb = cb.clone(); let resolver = resolver.clone(); - if let Ok(Some(entry)) = entries.resolve(resolver) { - cb(entry) - } else { - Box::pin(async {}) + let cb = cb.clone(); + + match entries.resolve(resolver) { + Ok(Some(entry)) => { + warn!("rebalance: list_objects_to_decommission get {}", &entry.name); + Box::pin(async move { cb(entry).await }) + } + Ok(None) => { + warn!("rebalance: list_objects_to_decommission get none"); + Box::pin(async {}) + } + Err(err) => { + error!("rebalance: list_objects_to_decommission get err {:?}", &err); + Box::pin(async {}) + } } })), ..Default::default() diff --git a/ecstore/src/set_disk.rs b/ecstore/src/set_disk.rs index 19c244190..bd55ab39f 100644 --- a/ecstore/src/set_disk.rs +++ b/ecstore/src/set_disk.rs @@ -3667,7 +3667,7 @@ impl ObjectIO for SetDisks { error!("get_object_with_fileinfo err {:?}", e); }; - // error!("get_object_with_fileinfo end"); + // error!("get_object_with_fileinfo end {}/{}", bucket, object); }); Ok(reader) diff --git a/rustfs/src/admin/handlers/rebalance.rs b/rustfs/src/admin/handlers/rebalance.rs index fefd7df42..c33767781 100644 --- a/rustfs/src/admin/handlers/rebalance.rs +++ b/rustfs/src/admin/handlers/rebalance.rs @@ -34,9 +34,9 @@ pub struct RebalPoolProgress { #[serde(rename = "object")] pub object: String, #[serde(rename = "elapsed")] - pub elapsed: Duration, + pub elapsed: u64, #[serde(rename = "eta")] - pub eta: Duration, + pub eta: u64, } #[derive(Debug, Clone, Serialize, Deserialize, Default)] @@ -78,13 +78,13 @@ impl Operation for RebalanceStart { if store.is_decommission_running().await { return Err(s3_error!( - InternalError, + InvalidRequest, "Rebalance cannot be started, decommission is already in progress" )); } if store.is_rebalance_started().await { - return Err(s3_error!(InternalError, "Rebalance already in progress")); + return Err(s3_error!(OperationAborted, "Rebalance already in progress")); } let bucket_infos = store @@ -101,8 +101,11 @@ impl Operation for RebalanceStart { } }; + warn!("Rebalance started with id: {}", id); if let Some(notification_sys) = get_global_notification_sys() { + warn!("Loading rebalance meta"); notification_sys.load_rebalance_meta(true).await; + warn!("Rebalance meta loaded"); } let resp = RebalanceResp { id }; @@ -149,7 +152,7 @@ impl Operation for RebalanceStatus { disk_stats[disk.pool_index as usize].total_space += disk.total_space; } - let stop_time = meta.stopped_at; + let mut stop_time = meta.stopped_at; let mut admin_status = RebalanceAdminStatus { id: meta.id.clone(), stopped_at: meta.stopped_at, @@ -171,7 +174,7 @@ impl Operation for RebalanceStatus { // Calculate total bytes to be rebalanced let total_bytes_to_rebal = ps.init_capacity as f64 * meta.percent_free_goal - ps.init_free_space as f64; - let elapsed = if let Some(start_time) = ps.info.start_time { + let mut elapsed = if let Some(start_time) = ps.info.start_time { SystemTime::now() .duration_since(start_time) .map_err(|e| s3_error!(InternalError, "Failed to calculate elapsed time: {}", e))? @@ -179,21 +182,25 @@ impl Operation for RebalanceStatus { return Err(s3_error!(InternalError, "Start time is not available")); }; - let eta = if ps.bytes > 0 { + let mut eta = if ps.bytes > 0 { Duration::from_secs_f64(total_bytes_to_rebal * elapsed.as_secs_f64() / ps.bytes as f64) } else { Duration::ZERO }; - let stop_time = ps.info.end_time.unwrap_or(stop_time.unwrap_or(SystemTime::now())); + if ps.info.end_time.is_some() { + stop_time = ps.info.end_time; + } - let elapsed = if ps.info.end_time.is_some() || meta.stopped_at.is_some() { - stop_time - .duration_since(ps.info.start_time.unwrap_or(stop_time)) - .unwrap_or_default() - } else { - elapsed - }; + if let Some(stopped_at) = stop_time { + if let Ok(du) = stopped_at.duration_since(ps.info.start_time.unwrap_or(stopped_at)) { + elapsed = du; + } else { + return Err(s3_error!(InternalError, "Failed to calculate elapsed time")); + } + + eta = Duration::ZERO; + } admin_status.pools[i].progress = Some(RebalPoolProgress { num_objects: ps.num_objects, @@ -201,8 +208,8 @@ impl Operation for RebalanceStatus { bytes: ps.bytes, bucket: ps.bucket.clone(), object: ps.object.clone(), - elapsed, - eta, + elapsed: elapsed.as_secs(), + eta: eta.as_secs(), }); } diff --git a/rustfs/src/grpc.rs b/rustfs/src/grpc.rs index 6e702617b..ce8092dfb 100644 --- a/rustfs/src/grpc.rs +++ b/rustfs/src/grpc.rs @@ -40,7 +40,7 @@ use tokio::spawn; use tokio::sync::mpsc; use tokio_stream::wrappers::ReceiverStream; use tonic::{Request, Response, Status, Streaming}; -use tracing::{debug, error, info}; +use tracing::{debug, error, info, warn}; type ResponseStream = Pin> + Send>>; @@ -2388,19 +2388,27 @@ impl Node for NodeService { let LoadRebalanceMetaRequest { start_rebalance } = request.into_inner(); + warn!("handle LoadRebalanceMetaRequest"); + store.load_rebalance_meta().await.map_err(|err| { error!("load_rebalance_meta err {:?}", err); Status::internal(err.to_string()) })?; + warn!("load_rebalance_meta success"); + if start_rebalance { + warn!("start rebalance"); let store = store.clone(); tokio::spawn(async move { store.start_rebalance().await; }); } - unimplemented!() + Ok(tonic::Response::new(LoadRebalanceMetaResponse { + success: true, + error_info: None, + })) } async fn load_transition_tier_config(