gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-17 10:17:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: move ecstore rebalance flow modules (#3658)

Browse Source
This commit is contained in:
安正超
2026-06-20 17:56:11 +08:00
committed by GitHub
parent 1595b2bc3f
commit d933d5e938
9 changed files with 5212 additions and 4983 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/rebalance.rs
+10 -4965
View File
File diff suppressed because it is too large Load Diff
crates/ecstore/src/rebalance/control.rs
+410
View File
@@ -0,0 +1,410 @@
use super::meta::{
clone_first_arc, clone_rebalance_pool_stats, defer_bucket_in_rebalance_queue, ensure_valid_rebalance_pool_index,
invalid_rebalance_pool_index_error, is_rebalance_conflicting_with_decommission, mark_rebalance_bucket_done,
merge_rebalance_meta, percent_free_ratio, rebalance_metadata_not_initialized_error, record_rebalance_cleanup_warning_in_meta,
resolve_next_rebalance_bucket, should_accept_rebalance_stats_update, should_pool_participate, stop_rebalance_meta_snapshot,
};
use super::worker::{
rebalance_meta_lock_error, resolve_load_rebalance_stats_update_result, resolve_rebalance_meta_load_result,
resolve_rebalance_meta_save_result,
};
use super::{
DiskStat, EVENT_REBALANCE_BUCKET, EVENT_REBALANCE_STATE, LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_REBALANCE, REBAL_META_NAME,
RebalStatus, RebalanceInfo, RebalanceMeta, RebalanceStats,
};
use crate::error::{Error, Result};
use crate::object_api::ObjectOptions;
use crate::set_disk::get_lock_acquire_timeout;
use crate::storage_api_contracts::EcstoreObjectIO;
use crate::store::ECStore;
use rustfs_filemeta::FileInfo;
use rustfs_storage_api::{NamespaceLocking as StorageNamespaceLocking, StorageAdminApi};
use std::sync::Arc;
use time::OffsetDateTime;
use tracing::{debug, info};
use uuid::Uuid;
impl ECStore {
pub(super) async fn save_rebalance_meta_with_merge<S>(
&self,
pool: Arc<S>,
local_snapshot: &RebalanceMeta,
stage: &str,
) -> Result<()>
where
S: EcstoreObjectIO + StorageNamespaceLocking<Error = Error, NamespaceLock = rustfs_lock::NamespaceLockWrapper>,
{
let ns_lock = pool.new_ns_lock(crate::disk::RUSTFS_META_BUCKET, REBAL_META_NAME).await?;
let _guard = ns_lock
.get_write_lock(get_lock_acquire_timeout())
.await
.map_err(rebalance_meta_lock_error)?;
let opts = ObjectOptions {
no_lock: true,
..Default::default()
};
let mut merged = RebalanceMeta::new();
match merged.load_with_opts(pool.clone(), opts.clone()).await {
Ok(()) => {
merge_rebalance_meta(&mut merged, local_snapshot);
}
Err(Error::ConfigNotFound) => {
merged = local_snapshot.clone();
}
Err(err) => return Err(Error::other(format!("rebalance meta load before save failed during {stage}: {err}"))),
}
merged.save_with_opts(pool, opts).await
}
#[tracing::instrument(skip_all)]
pub async fn load_rebalance_meta(&self) -> Result<()> {
let mut meta = RebalanceMeta::new();
debug!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
state = "metadata_loading",
"Loading rebalance metadata"
);
let pool = clone_first_arc(&self.pools, "rebalanceMeta: no pools available")?;
if resolve_rebalance_meta_load_result(meta.load(pool).await)? {
{
let mut rebalance_meta = self.rebalance_meta.write().await;
*rebalance_meta = Some(meta);
drop(rebalance_meta);
}
resolve_load_rebalance_stats_update_result(self.update_rebalance_stats().await)?;
debug!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
state = "metadata_loaded",
"Loaded rebalance metadata"
);
} else {
debug!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
state = "metadata_missing",
reason = "rebalance_not_started",
"Rebalance metadata not found"
);
}
Ok(())
}
#[tracing::instrument(skip_all)]
pub async fn update_rebalance_stats(&self) -> Result<()> {
let mut ok = false;
let pool_stats = {
let rebalance_meta = self.rebalance_meta.read().await;
clone_rebalance_pool_stats(rebalance_meta.as_ref())?
};
debug!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_count = pool_stats.len(),
"Refreshing rebalance stats snapshot"
);
for i in 0..self.pools.len() {
if pool_stats.get(i).is_none() {
let mut rebalance_meta = self.rebalance_meta.write().await;
debug!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index = i,
state = "pool_stat_missing",
"Adding missing rebalance pool stats entry"
);
if let Some(meta) = rebalance_meta.as_mut() {
meta.pool_stats.push(RebalanceStats::default());
}
ok = true;
drop(rebalance_meta);
}
}
if ok {
debug!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
state = "metadata_saving",
"Saving rebalance metadata after stats refresh"
);
let rebalance_meta = self.rebalance_meta.read().await;
if let Some(meta) = rebalance_meta.as_ref() {
let pool = clone_first_arc(&self.pools, "update_rebalance_stats: no pools available")?;
resolve_rebalance_meta_save_result(
self.save_rebalance_meta_with_merge(pool, meta, "update_rebalance_stats")
.await,
"update_rebalance_stats",
)?;
}
}
Ok(())
}
#[tracing::instrument(skip(self))]
pub async fn init_rebalance_meta(&self, bucktes: Vec<String>) -> Result<String> {
info!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
state = "initializing",
bucket_count = bucktes.len(),
"Initializing rebalance metadata"
);
let si = StorageAdminApi::storage_info(self).await;
let mut disk_stats = vec![DiskStat::default(); self.pools.len()];
let mut total_cap = 0;
let mut total_free = 0;
for disk in si.disks.iter() {
if disk.pool_index < 0 || disk_stats.len() <= disk.pool_index as usize {
continue;
}
total_cap += disk.total_space;
total_free += disk.available_space;
disk_stats[disk.pool_index as usize].total_space += disk.total_space;
disk_stats[disk.pool_index as usize].available_space += disk.available_space;
}
let percent_free_goal = percent_free_ratio(total_free, total_cap);
let mut pool_stats = Vec::with_capacity(self.pools.len());
let now = OffsetDateTime::now_utc();
for disk_stat in disk_stats.iter() {
let mut pool_stat = RebalanceStats {
init_free_space: disk_stat.available_space,
init_capacity: disk_stat.total_space,
buckets: bucktes.clone(),
rebalanced_buckets: Vec::with_capacity(bucktes.len()),
..Default::default()
};
if should_pool_participate(disk_stat.available_space, disk_stat.total_space, percent_free_goal) {
pool_stat.participating = true;
pool_stat.info = RebalanceInfo {
start_time: Some(now),
status: RebalStatus::Started,
..Default::default()
};
}
pool_stats.push(pool_stat);
}
let meta = RebalanceMeta {
id: Uuid::new_v4().to_string(),
percent_free_goal,
pool_stats,
..Default::default()
};
let pool = clone_first_arc(&self.pools, "init_rebalance_meta: no pools available")?;
resolve_rebalance_meta_save_result(
self.save_rebalance_meta_with_merge(pool, &meta, "init_rebalance_meta").await,
"init_rebalance_meta",
)?;
info!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
state = "metadata_initialized",
bucket_count = bucktes.len(),
"Rebalance metadata initialized"
);
let id = meta.id.clone();
{
let mut rebalance_meta = self.rebalance_meta.write().await;
*rebalance_meta = Some(meta);
drop(rebalance_meta);
}
Ok(id)
}
#[tracing::instrument(skip(self, fi))]
pub async fn update_pool_stats(&self, pool_index: usize, bucket: String, fi: &FileInfo) -> Result<()> {
self.update_pool_stats_batch(pool_index, bucket, &[fi]).await
}
#[tracing::instrument(skip(self, versions))]
pub async fn update_pool_stats_batch(&self, pool_index: usize, bucket: String, versions: &[&FileInfo]) -> Result<()> {
if versions.is_empty() {
return Ok(());
}
let mut rebalance_meta = self.rebalance_meta.write().await;
if let Some(meta) = rebalance_meta.as_mut() {
if !should_accept_rebalance_stats_update(meta, pool_index) {
return Ok(());
}
if let Some(pool_stat) = meta.pool_stats.get_mut(pool_index) {
pool_stat.update_batch(bucket, versions);
}
}
Ok(())
}
#[tracing::instrument(skip(self))]
pub async fn next_rebal_bucket(&self, pool_index: usize) -> Result<Option<String>> {
let rebalance_meta = self.rebalance_meta.read().await;
debug!(
event = EVENT_REBALANCE_BUCKET,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index,
has_meta = rebalance_meta.is_some(),
state = "next_bucket_lookup",
"Rebalance next bucket lookup"
);
resolve_next_rebalance_bucket(rebalance_meta.as_ref(), pool_index)
}
#[tracing::instrument(skip(self))]
pub async fn bucket_rebalance_done(&self, pool_index: usize, bucket: String) -> Result<()> {
let mut rebalance_meta = self.rebalance_meta.write().await;
mark_rebalance_bucket_done(rebalance_meta.as_mut(), pool_index, &bucket)
}
pub(super) async fn record_rebalance_cleanup_warning(
&self,
pool_index: usize,
bucket: &str,
object: &str,
message: String,
) -> Result<()> {
let mut rebalance_meta = self.rebalance_meta.write().await;
record_rebalance_cleanup_warning_in_meta(
rebalance_meta.as_mut(),
pool_index,
bucket,
object,
message,
OffsetDateTime::now_utc(),
)
}
pub(super) async fn defer_rebalance_bucket(&self, pool_index: usize, bucket: String, last_error: String) -> Result<()> {
let mut rebalance_meta = self.rebalance_meta.write().await;
let Some(meta) = rebalance_meta.as_mut() else {
return Err(rebalance_metadata_not_initialized_error("defer rebalance bucket"));
};
let pool_count = meta.pool_stats.len();
ensure_valid_rebalance_pool_index(pool_count, pool_index)?;
let Some(pool_stat) = meta.pool_stats.get_mut(pool_index) else {
return Err(invalid_rebalance_pool_index_error(pool_index, pool_count));
};
defer_bucket_in_rebalance_queue(pool_stat, &bucket)?;
pool_stat.info.last_error = Some(last_error);
meta.last_refreshed_at = Some(OffsetDateTime::now_utc());
Ok(())
}
pub async fn is_rebalance_started(&self) -> bool {
let rebalance_meta = self.rebalance_meta.read().await;
if let Some(meta) = rebalance_meta.as_ref() {
meta.pool_stats.iter().enumerate().for_each(|(i, v)| {
debug!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index = i,
participating = v.participating,
status = ?v.info.status,
state = "status_inspected",
"Rebalance status inspected"
);
});
let started = is_rebalance_conflicting_with_decommission(meta);
if started {
debug!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
state = "running",
"Rebalance is running"
);
return true;
}
}
debug!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
state = "not_running",
"Rebalance is not running"
);
false
}
pub async fn is_rebalance_conflicting_with_decommission(&self) -> bool {
let rebalance_meta = self.rebalance_meta.read().await;
rebalance_meta
.as_ref()
.is_some_and(is_rebalance_conflicting_with_decommission)
}
pub async fn is_pool_rebalancing(&self, pool_index: usize) -> bool {
let rebalance_meta = self.rebalance_meta.read().await;
if let Some(ref meta) = *rebalance_meta {
if meta.stopped_at.is_some() {
return false;
}
if let Some(pool_stat) = meta.pool_stats.get(pool_index) {
return pool_stat.participating && pool_stat.info.status == RebalStatus::Started;
}
}
false
}
#[tracing::instrument(skip(self))]
pub async fn stop_rebalance(self: &Arc<Self>) -> Result<()> {
let meta_to_save = {
let mut rebalance_meta = self.rebalance_meta.write().await;
stop_rebalance_meta_snapshot(rebalance_meta.as_mut(), OffsetDateTime::now_utc())
};
if let Some(meta_to_save) = meta_to_save {
let pool = clone_first_arc(self.pools.as_slice(), "stop_rebalance: no pools available")?;
resolve_rebalance_meta_save_result(
self.save_rebalance_meta_with_merge(pool, &meta_to_save, "stop_rebalance")
.await,
"stop_rebalance",
)?;
}
Ok(())
}
}
crates/ecstore/src/rebalance/entry.rs
+513
View File
@@ -0,0 +1,513 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use super::meta::{
clone_arc_by_index, ensure_valid_rebalance_pool_index, invalid_rebalance_pool_index_error,
rebalance_metadata_not_initialized_error, should_ignore_rebalance_data_usage_cache,
};
use super::migration::migrate_entry_version;
use super::worker::{
RebalanceEntryTask, load_rebalance_bucket_configs, rebalance_max_attempts, resolve_rebalance_bucket_error,
resolve_rebalance_entry_cleanup_delete_result, resolve_rebalance_file_info_versions_result,
resolve_rebalance_migrate_result_error, resolve_rebalance_stats_update_result, resolve_rebalance_worker_result,
run_rebalance_listing_with_retry, should_cleanup_rebalance_source_entry, should_count_rebalance_version_complete,
should_defer_rebalance_entry_failure, should_skip_rebalance_delete_marker, wait_rebalance_entry_tasks,
with_rebalance_entry_context,
};
use super::{
EVENT_REBALANCE_BUCKET, EVENT_REBALANCE_ENTRY, EVENT_REBALANCE_STATE, LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_REBALANCE,
REBALANCE_DEFERRED_ENTRY_ERROR_PREFIX, RebalanceBucketConfigs, RebalanceBucketOutcome, RebalanceEntryOutcome,
};
use crate::data_movement;
use crate::error::{Error, Result};
use crate::object_api::{GetObjectReader, ObjectOptions};
use crate::pools::ListCallback;
use crate::set_disk::SetDisks;
use crate::store::ECStore;
use rustfs_filemeta::MetaCacheEntry;
use rustfs_storage_api::ObjectOperations as _;
use rustfs_utils::path::encode_dir_object;
use std::sync::Arc;
use time::OffsetDateTime;
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, warn};
impl ECStore {
#[allow(unused_assignments)]
#[tracing::instrument(skip(self, set))]
async fn rebalance_entry(
self: Arc<Self>,
bucket: String,
pool_index: usize,
entry: MetaCacheEntry,
set: Arc<SetDisks>,
bucket_configs: Arc<RebalanceBucketConfigs>,
// wk: Arc<Workers>,
) -> Result<RebalanceEntryOutcome> {
debug!(
event = EVENT_REBALANCE_ENTRY,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
bucket = %bucket,
object = %entry.name,
pool_index,
state = "started",
"Starting rebalance entry"
);
// defer!(|| async {
// warn!("rebalance_entry: defer give worker start");
// wk.give().await;
// warn!("rebalance_entry: defer give worker done");
// });
if entry.is_dir() {
debug!(
event = EVENT_REBALANCE_ENTRY,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
bucket = %bucket,
object = %entry.name,
pool_index,
state = "skipped",
reason = "directory_entry",
"Skipped rebalance entry"
);
return Ok(RebalanceEntryOutcome::Completed);
}
if self.check_if_rebalance_done(pool_index).await {
debug!(
event = EVENT_REBALANCE_ENTRY,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index,
bucket = %bucket,
object = %entry.name,
state = "skipped",
reason = "pool_completed",
"Skipped rebalance entry"
);
return Ok(RebalanceEntryOutcome::Completed);
}
let mut fivs =
resolve_rebalance_file_info_versions_result(entry.file_info_versions(&bucket), bucket.as_str(), entry.name.as_str())?;
fivs.versions
.sort_by_key(|v| (v.mod_time.is_none(), std::cmp::Reverse(v.mod_time)));
let mut rebalanced: usize = 0;
let mut expired: usize = 0;
let mut stats_updates = Vec::with_capacity(fivs.versions.len());
for version in fivs.versions.iter() {
if crate::pools::should_skip_lifecycle_for_data_movement(
self.clone(),
&bucket,
version,
bucket_configs.lifecycle_config.as_ref(),
bucket_configs.lock_retention.clone(),
bucket_configs.replication_config.clone(),
true,
&crate::bucket::lifecycle::bucket_lifecycle_audit::LcEventSrc::Rebal,
)
.await
{
expired += 1;
debug!(
event = EVENT_REBALANCE_ENTRY,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index,
bucket = %bucket,
object = %version.name,
state = "skipped",
reason = "expired_by_lifecycle",
"Skipped rebalance version"
);
continue;
}
let remaining_versions = fivs.versions.len() - expired;
if should_skip_rebalance_delete_marker(version, remaining_versions, bucket_configs.replication_config.is_some()) {
rebalanced += 1;
debug!(
event = EVENT_REBALANCE_ENTRY,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index,
bucket = %bucket,
object = %version.name,
state = "skipped",
reason = "last_delete_marker_without_replication",
"Skipped rebalance version"
);
continue;
}
let version_id = version.version_id.map(|v| v.to_string());
let mut transfer = |src_pool_idx: usize, bucket: String, rd: GetObjectReader| {
let store = self.clone();
async move { store.rebalance_object(src_pool_idx, bucket, rd).await }
};
let result = migrate_entry_version(
set.as_ref(),
bucket.clone(),
pool_index,
version,
version_id.clone(),
rebalance_max_attempts(),
should_ignore_rebalance_data_usage_cache(bucket.as_str()),
&mut transfer,
)
.await;
if result.ignored {
if should_count_rebalance_version_complete(&result) {
rebalanced += 1;
}
debug!(
event = EVENT_REBALANCE_ENTRY,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index,
bucket = %bucket,
object = %version.name,
state = "skipped",
reason = "already_deleted",
"Skipped rebalance version"
);
continue;
}
if result.failed {
let err = resolve_rebalance_migrate_result_error(
result.error,
pool_index,
bucket.as_str(),
version.name.as_str(),
version_id.as_deref(),
);
error!(
"rebalance_entry {} Error rebalancing entry {}/{:?}: {:?}",
&bucket, &version.name, &version.version_id, err
);
if should_defer_rebalance_entry_failure(&err) {
let deferred_error = format!("{REBALANCE_DEFERRED_ENTRY_ERROR_PREFIX} {err}");
warn!(
event = EVENT_REBALANCE_ENTRY,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index,
bucket = %bucket,
object = %version.name,
state = "deferred",
error = %err,
"Deferred rebalance entry after transient migration failure"
);
if let Err(stats_err) = self.update_rebalance_last_error(pool_index, deferred_error.clone()).await {
error!(
"rebalance_entry {} failed to record deferred transient failure for {}: {}",
&bucket, &entry.name, stats_err
);
}
return Ok(RebalanceEntryOutcome::Deferred {
last_error: deferred_error,
});
}
let entry_err =
with_rebalance_entry_context(result.stage.unwrap_or("migrate"), bucket.as_str(), version.name.as_str(), err);
if !stats_updates.is_empty()
&& let Err(stats_err) = self
.update_pool_stats_batch(pool_index, bucket.clone(), stats_updates.as_slice())
.await
{
error!(
"rebalance_entry {} failed to update stats before returning migration error for {}: {}",
&bucket, &entry.name, stats_err
);
}
return Err(entry_err);
}
stats_updates.push(version);
if should_count_rebalance_version_complete(&result) {
rebalanced += 1;
}
}
resolve_rebalance_stats_update_result(
self.update_pool_stats_batch(pool_index, bucket.clone(), stats_updates.as_slice())
.await,
pool_index,
bucket.as_str(),
entry.name.as_str(),
)?;
if should_cleanup_rebalance_source_entry(rebalanced, fivs.versions.len()) {
let cleanup_warning = resolve_rebalance_entry_cleanup_delete_result(
set.delete_object(
bucket.as_str(),
&encode_dir_object(&entry.name),
ObjectOptions {
delete_prefix: true,
delete_prefix_object: true,
..Default::default()
},
)
.await,
bucket.as_str(),
entry.name.as_str(),
)?;
if let Some(message) = cleanup_warning {
warn!(
event = EVENT_REBALANCE_ENTRY,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index,
bucket = %bucket,
object = %entry.name,
stage = "cleanup_source",
cleanup_status = "failed_ignored",
error = %message,
"Ignored rebalance source cleanup failure"
);
if let Err(err) = self
.record_rebalance_cleanup_warning(pool_index, bucket.as_str(), entry.name.as_str(), message)
.await
{
error!(
event = EVENT_REBALANCE_ENTRY,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index,
bucket = %bucket,
object = %entry.name,
stage = "cleanup_source",
error = ?err,
"Failed to record rebalance source cleanup warning"
);
}
} else {
debug!(
event = EVENT_REBALANCE_ENTRY,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index,
bucket = %bucket,
object = %entry.name,
state = "source_deleted",
"Deleted rebalance source entry"
);
}
}
Ok(RebalanceEntryOutcome::Completed)
}
#[tracing::instrument(skip(self, rd))]
async fn rebalance_object(self: Arc<Self>, pool_idx: usize, bucket: String, rd: GetObjectReader) -> Result<()> {
data_movement::migrate_object(self, pool_idx, bucket, rd, "rebalance_object").await
}
async fn update_rebalance_last_error(&self, pool_idx: usize, message: String) -> Result<()> {
let mut rebalance_meta = self.rebalance_meta.write().await;
let Some(meta) = rebalance_meta.as_mut() else {
return Err(rebalance_metadata_not_initialized_error("record rebalance last error"));
};
let pool_count = meta.pool_stats.len();
ensure_valid_rebalance_pool_index(pool_count, pool_idx)?;
let Some(pool_stat) = meta.pool_stats.get_mut(pool_idx) else {
return Err(invalid_rebalance_pool_index_error(pool_idx, pool_count));
};
pool_stat.info.last_error = Some(message);
meta.last_refreshed_at = Some(OffsetDateTime::now_utc());
Ok(())
}
#[tracing::instrument(skip(self, rx))]
pub(super) async fn rebalance_bucket(
self: &Arc<Self>,
rx: CancellationToken,
bucket: String,
pool_index: usize,
) -> Result<RebalanceBucketOutcome> {
ensure_valid_rebalance_pool_index(self.pools.len(), pool_index)?;
// Placeholder for actual bucket rebalance logic
debug!(
event = EVENT_REBALANCE_BUCKET,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index,
bucket = %bucket,
state = "entry_scan_started",
"Rebalance bucket entry scan started"
);
// TODO: other config
// if bucket != RUSTFS_META_BUCKET{
// }
let pool = clone_arc_by_index(self.pools.as_slice(), pool_index, "invalid rebalance pool index")?;
let bucket_configs = Arc::new(load_rebalance_bucket_configs(&bucket).await?);
let mut jobs = Vec::new();
let entry_error = Arc::new(tokio::sync::Mutex::new(None::<Error>));
let entry_workers = Arc::new(tokio::sync::Semaphore::new(pool.disk_set.len().max(1)));
for (set_idx, set) in pool.disk_set.iter().enumerate() {
let entry_tasks = Arc::new(tokio::sync::Mutex::new(Vec::<RebalanceEntryTask>::new()));
let rebalance_entry: ListCallback = Arc::new({
let this = Arc::clone(self);
let bucket = bucket.clone();
let entry_error = entry_error.clone();
let callback_rx = rx.clone();
let set = set.clone();
let bucket_configs = bucket_configs.clone();
let entry_tasks = entry_tasks.clone();
let entry_workers = entry_workers.clone();
move |entry: MetaCacheEntry| {
let this = this.clone();
let bucket = bucket.clone();
let entry_error = entry_error.clone();
let callback_rx = callback_rx.clone();
let set = set.clone();
let bucket_configs = bucket_configs.clone();
let entry_tasks = entry_tasks.clone();
let entry_workers = entry_workers.clone();
Box::pin(async move {
if callback_rx.is_cancelled() {
return;
}
if entry_error.lock().await.is_some() {
return;
}
let permit = tokio::select! {
_ = callback_rx.cancelled() => return,
permit = entry_workers.clone().acquire_owned() => match permit {
Ok(permit) => permit,
Err(err) => {
error!("rebalance_entry: worker semaphore closed: {err}");
return;
}
},
};
if entry_error.lock().await.is_some() {
return;
}
let task = tokio::spawn(async move {
let _permit = permit;
debug!(
event = EVENT_REBALANCE_ENTRY,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
set_index = set_idx,
state = "task_started",
"Started rebalance entry task"
);
let result = this.rebalance_entry(bucket, pool_index, entry, set, bucket_configs).await;
if let Err(err) = &result {
error!("rebalance_entry: rebalance entry failed: {err}");
let mut first_err = entry_error.lock().await;
if first_err.is_none() {
*first_err = Some(err.clone());
callback_rx.cancel();
}
}
debug!(
event = EVENT_REBALANCE_ENTRY,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
set_index = set_idx,
state = "task_completed",
"Completed rebalance entry task"
);
result
});
entry_tasks.lock().await.push(task);
})
}
});
let set = set.clone();
let rx = rx.clone();
let bucket = bucket.clone();
let entry_tasks = entry_tasks.clone();
let job = tokio::spawn(async move {
let list_result =
run_rebalance_listing_with_retry(set, rx, bucket.clone(), rebalance_entry, set_idx, rebalance_max_attempts())
.await;
let entry_result = wait_rebalance_entry_tasks(set_idx, entry_tasks).await;
let result = list_result.and(entry_result);
if let Err(err) = &result {
error!("Rebalance worker {} error: {}", set_idx, err);
} else {
debug!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
set_index = set_idx,
state = "worker_completed",
"Completed rebalance worker"
);
}
result
});
jobs.push((set_idx, job));
}
let mut worker_error: Option<Error> = None;
let mut deferred_error: Option<String> = None;
for (set_idx, job) in jobs {
match resolve_rebalance_worker_result(set_idx, job.await) {
Ok(Some(last_error)) if deferred_error.is_none() => {
deferred_error = Some(last_error);
}
Ok(_) => {}
Err(err) if worker_error.is_none() => {
worker_error = Some(err);
}
Err(_) => {}
}
}
let entry_error = entry_error.lock().await.clone();
resolve_rebalance_bucket_error(entry_error, worker_error)?;
if let Some(last_error) = deferred_error {
return Ok(RebalanceBucketOutcome::Deferred { last_error });
}
debug!(
event = EVENT_REBALANCE_BUCKET,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index,
bucket = %bucket,
state = "completed",
"Finished rebalance bucket"
);
Ok(RebalanceBucketOutcome::Completed)
}
}
crates/ecstore/src/rebalance/meta.rs
+189 -4
View File
@@ -1,13 +1,198 @@
use super::{
EVENT_REBALANCE_BUCKET, Error, LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_REBALANCE, REBALANCE_DEFERRED_ENTRY_ERROR_PREFIX,
RebalSaveOpt, RebalStatus, RebalanceCleanupWarnings, RebalanceMeta, RebalanceStats, Result,
EVENT_REBALANCE_BUCKET, EVENT_REBALANCE_STATE, Error, GetObjectReader, LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_REBALANCE,
ObjectInfo, ObjectOptions, PutObjReader, REBAL_META_FMT, REBAL_META_NAME, REBAL_META_VER,
REBALANCE_DEFERRED_ENTRY_ERROR_PREFIX, RebalSaveOpt, RebalStatus, RebalanceCleanupWarnings, RebalanceMeta, RebalanceStats,
Result,
};
use crate::config::com::{read_config_with_metadata, save_config_with_opts};
use crate::error::is_err_operation_canceled;
use std::collections::HashSet;
use std::sync::Arc;
use http::HeaderMap;
use rustfs_filemeta::FileInfo;
use rustfs_storage_api::{HTTPRangeSpec, ObjectIO};
use std::{collections::HashSet, fmt, io::Cursor, sync::Arc};
use time::OffsetDateTime;
use tracing::{debug, info};
impl RebalanceStats {
pub fn update(&mut self, bucket: String, fi: &FileInfo) {
if fi.is_latest {
self.num_objects += 1;
}
self.num_versions += 1;
let on_disk_size = if fi.deleted || fi.erasure.data_blocks == 0 || fi.size <= 0 {
0
} else {
let data_blocks = fi.erasure.data_blocks as i64;
let total_blocks = fi.erasure.data_blocks.saturating_add(fi.erasure.parity_blocks) as i64;
fi.size
.saturating_mul(total_blocks)
.checked_div(data_blocks)
.unwrap_or(0)
.max(0) as u64
};
self.bytes = self.bytes.saturating_add(on_disk_size);
self.bucket = bucket;
self.object = fi.name.clone();
}
pub fn update_batch(&mut self, bucket: String, versions: &[&FileInfo]) {
for version in versions {
self.update(bucket.clone(), version);
}
}
}
impl fmt::Display for RebalStatus {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let status = match self {
RebalStatus::None => "None",
RebalStatus::Started => "Started",
RebalStatus::Completed => "Completed",
RebalStatus::Stopped => "Stopped",
RebalStatus::Failed => "Failed",
};
write!(f, "{status}")
}
}
impl From<u8> for RebalStatus {
fn from(value: u8) -> Self {
match value {
1 => RebalStatus::Started,
2 => RebalStatus::Completed,
3 => RebalStatus::Stopped,
4 => RebalStatus::Failed,
_ => RebalStatus::None,
}
}
}
impl RebalanceMeta {
pub fn new() -> Self {
Self::default()
}
pub async fn load<S>(&mut self, store: Arc<S>) -> Result<()>
where
S: ObjectIO<
Error = Error,
RangeSpec = HTTPRangeSpec,
HeaderMap = HeaderMap,
ObjectOptions = ObjectOptions,
ObjectInfo = ObjectInfo,
GetObjectReader = GetObjectReader,
PutObjectReader = PutObjReader,
>,
{
self.load_with_opts(store, ObjectOptions::default()).await
}
pub async fn load_with_opts<S>(&mut self, store: Arc<S>, opts: ObjectOptions) -> Result<()>
where
S: ObjectIO<
Error = Error,
RangeSpec = HTTPRangeSpec,
HeaderMap = HeaderMap,
ObjectOptions = ObjectOptions,
ObjectInfo = ObjectInfo,
GetObjectReader = GetObjectReader,
PutObjectReader = PutObjReader,
>,
{
let (data, _) = read_config_with_metadata(store, REBAL_META_NAME, &opts).await?;
if data.is_empty() {
debug!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
state = "metadata_empty",
"Rebalance metadata is empty"
);
return Ok(());
}
if data.len() <= 4 {
return Err(rebalance_meta_load_no_data_error());
}
// Read header
match u16::from_le_bytes([data[0], data[1]]) {
REBAL_META_FMT => {}
fmt => return Err(rebalance_meta_load_unknown_format_error(fmt)),
}
match u16::from_le_bytes([data[2], data[3]]) {
REBAL_META_VER => {}
ver => return Err(rebalance_meta_load_unknown_version_error(ver)),
}
let meta: Self = rmp_serde::from_read(Cursor::new(&data[4..]))?;
*self = meta;
self.last_refreshed_at = Some(OffsetDateTime::now_utc());
debug!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
state = "metadata_loaded",
"Loaded rebalance metadata"
);
Ok(())
}
pub async fn save<S>(&self, store: Arc<S>) -> Result<()>
where
S: ObjectIO<
Error = Error,
RangeSpec = HTTPRangeSpec,
HeaderMap = HeaderMap,
ObjectOptions = ObjectOptions,
ObjectInfo = ObjectInfo,
GetObjectReader = GetObjectReader,
PutObjectReader = PutObjReader,
>,
{
self.save_with_opts(store, ObjectOptions::default()).await
}
pub async fn save_with_opts<S>(&self, store: Arc<S>, opts: ObjectOptions) -> Result<()>
where
S: ObjectIO<
Error = Error,
RangeSpec = HTTPRangeSpec,
HeaderMap = HeaderMap,
ObjectOptions = ObjectOptions,
ObjectInfo = ObjectInfo,
GetObjectReader = GetObjectReader,
PutObjectReader = PutObjReader,
>,
{
if self.pool_stats.is_empty() {
debug!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
state = "metadata_save_skipped",
reason = "no_pool_stats",
"Skipped rebalance metadata save"
);
return Ok(());
}
let mut data = Vec::new();
// Initialize the header
data.extend(&REBAL_META_FMT.to_le_bytes());
data.extend(&REBAL_META_VER.to_le_bytes());
let msg = rmp_serde::to_vec(self)?;
data.extend(msg);
save_config_with_opts(store, REBAL_META_NAME, data, &opts).await?;
Ok(())
}
}
pub(super) fn is_rebalance_pool_started(pool_stat: &RebalanceStats) -> bool {
pool_stat.participating && pool_stat.info.status == RebalStatus::Started
}
crates/ecstore/src/rebalance/rebalance_unit_tests.rs
+3230
View File
File diff suppressed because it is too large Load Diff
crates/ecstore/src/rebalance/runtime.rs
+574
View File
@@ -0,0 +1,574 @@
use super::meta::{
apply_rebalance_save_option, apply_rebalance_terminal_event, classify_rebalance_terminal_event, clone_first_arc,
complete_rebalance_pools_at_goal, complete_rebalance_pools_with_empty_queue, ensure_valid_rebalance_pool_index,
has_deferred_rebalance_error, is_rebalance_in_progress, rebalance_goal_reached, resolve_rebalance_participants,
should_preserve_rebalance_stopped_state, should_skip_start_rebalance, validate_start_rebalance_state,
};
use super::worker::{
resolve_rebalance_bucket_result, resolve_rebalance_meta_save_result, resolve_rebalance_save_task_result,
resolve_rebalance_terminal_error, send_rebalance_done_signal,
};
use super::{
EVENT_REBALANCE_BUCKET, EVENT_REBALANCE_STATE, LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_REBALANCE, RebalSaveOpt, RebalStatus,
RebalanceBucketOutcome,
};
use crate::error::{Error, Result};
use crate::global::get_global_endpoints;
use crate::store::ECStore;
use std::collections::HashSet;
use std::sync::Arc;
use time::OffsetDateTime;
use tokio::time::{Duration, Instant};
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info, warn};
impl ECStore {
#[tracing::instrument(skip_all)]
pub async fn start_rebalance(self: &Arc<Self>) -> Result<()> {
info!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
state = "starting",
"Starting rebalance"
);
let decommission_running = self.is_decommission_running().await;
// let rebalance_meta = self.rebalance_meta.read().await;
let cancel_tx = CancellationToken::new();
let rx = cancel_tx.clone();
let mut meta_to_save = None;
{
let mut rebalance_meta = self.rebalance_meta.write().await;
validate_start_rebalance_state(decommission_running, rebalance_meta.is_some())?;
let Some(meta) = rebalance_meta.as_mut() else {
return Err(Error::ConfigNotFound);
};
if should_skip_start_rebalance(meta.cancel.is_some(), is_rebalance_in_progress(meta)) {
debug!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
state = "start_skipped",
reason = "already_in_progress",
"Skipped duplicate rebalance start"
);
return Ok(());
}
let now = OffsetDateTime::now_utc();
if complete_rebalance_pools_at_goal(meta, now) {
meta_to_save = Some(meta.clone());
}
if complete_rebalance_pools_with_empty_queue(meta, now) {
meta_to_save = Some(meta.clone());
}
meta.cancel = Some(cancel_tx);
drop(rebalance_meta);
}
if let Some(meta) = meta_to_save {
let pool = clone_first_arc(self.pools.as_slice(), "start_rebalance: no pools available")?;
resolve_rebalance_meta_save_result(
self.save_rebalance_meta_with_merge(pool, &meta, "start_rebalance complete pools at goal")
.await,
"start_rebalance complete pools at goal",
)?;
}
let participants = if let Some(ref meta) = *self.rebalance_meta.read().await {
resolve_rebalance_participants(meta.pool_stats.as_slice(), self.pools.len())
} else {
debug!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
state = "start_skipped",
reason = "metadata_missing",
"Skipped rebalance start because metadata is unavailable"
);
Vec::new()
};
if !participants.iter().any(|participating| *participating) {
debug!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
state = "start_skipped",
reason = "no_participants",
"Skipped rebalance start because no pools are participating"
);
return Ok(());
}
let mut workers_started = 0usize;
for (idx, participating) in participants.iter().enumerate() {
if !*participating {
debug!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index = idx,
state = "pool_skipped",
reason = "not_participating",
"Skipped rebalance pool"
);
continue;
}
if !get_global_endpoints()
.as_ref()
.get(idx)
.is_some_and(|v| v.endpoints.as_ref().first().is_some_and(|e| e.is_local))
{
debug!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index = idx,
state = "pool_skipped",
reason = "not_local",
"Skipped non-local rebalance pool"
);
continue;
}
let pool_idx = idx;
let store = self.clone();
let rx_clone = rx.clone();
workers_started += 1;
tokio::spawn(async move {
if let Err(err) = store.rebalance_buckets(rx_clone, pool_idx).await {
error!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index = pool_idx,
state = "pool_failed",
error = %err,
"Rebalance pool failed"
);
} else {
debug!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index = pool_idx,
state = "completed",
"Rebalance pool completed"
);
}
});
}
if workers_started == 0 {
debug!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
state = "start_skipped",
reason = "no_local_participants",
"Skipped rebalance start because no local pools are participating"
);
return Ok(());
}
info!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
state = "started",
worker_count = workers_started,
"Rebalance started"
);
Ok(())
}
#[tracing::instrument(skip(self, rx))]
async fn rebalance_buckets(self: &Arc<Self>, rx: CancellationToken, pool_index: usize) -> Result<()> {
ensure_valid_rebalance_pool_index(self.pools.len(), pool_index)?;
let (done_tx, mut done_rx) = tokio::sync::mpsc::channel::<Result<()>>(1);
// Save rebalance metadata periodically
let store = self.clone();
let save_task = tokio::spawn(async move {
let mut timer = tokio::time::interval_at(Instant::now() + Duration::from_secs(30), Duration::from_secs(10));
let mut msg: String;
let mut quit = false;
loop {
tokio::select! {
result = done_rx.recv() => {
quit = true;
let now = OffsetDateTime::now_utc();
let terminal_event = classify_rebalance_terminal_event(result, now);
msg = terminal_event.message().to_string();
let mut rebalance_meta = store.rebalance_meta.write().await;
if let Some(meta) = rebalance_meta.as_mut() {
let meta_stopped = meta.stopped_at.is_some();
if let Some(pool_stat) = meta.pool_stats.get_mut(pool_index) {
if should_preserve_rebalance_stopped_state(
meta_stopped,
pool_stat.info.status,
&terminal_event,
) {
debug!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index,
state = "stopped_preserved",
"Preserved stopped rebalance status"
);
} else {
apply_rebalance_terminal_event(
&mut pool_stat.info.status,
&mut pool_stat.info.end_time,
&mut pool_stat.info.last_error,
terminal_event,
now,
);
}
}
}
}
_ = timer.tick() => {
let now = OffsetDateTime::now_utc();
msg = format!("Saving rebalance metadata at {now:?}");
}
}
if let Err(err) = store.save_rebalance_stats(pool_index, RebalSaveOpt::Stats).await {
let wrapped = Error::other(format!("rebalance save_task stats save failed for pool {pool_index}: {err}"));
error!("{} err: {:?}", msg, wrapped);
if quit {
return Err(wrapped);
}
} else {
debug!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index,
state = "metadata_saved",
message = %msg,
"Saved rebalance metadata"
);
}
if quit {
debug!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index,
state = "save_task_exiting",
message = %msg,
"Exiting rebalance save task"
);
return Ok(());
}
timer.reset();
}
});
info!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index,
state = "pool_started",
"Rebalance worker started"
);
let mut final_result: Result<()> = Ok(());
let mut deferred_buckets = HashSet::new();
loop {
if rx.is_cancelled() {
info!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index,
state = "pool_stopped",
reason = "cancelled",
"Stopped rebalance worker"
);
let err = Error::OperationCanceled;
final_result = Err(resolve_rebalance_terminal_error(
err.clone(),
send_rebalance_done_signal(&done_tx, Err(err.clone()), pool_index).await,
));
break;
}
let next_bucket = match self.next_rebal_bucket(pool_index).await {
Ok(bucket) => bucket,
Err(err) => {
error!(
event = EVENT_REBALANCE_BUCKET,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index,
state = "next_bucket_failed",
error = ?err,
"Rebalance next bucket lookup failed"
);
final_result = Err(resolve_rebalance_terminal_error(
err.clone(),
send_rebalance_done_signal(&done_tx, Err(err.clone()), pool_index).await,
));
break;
}
};
if let Some(bucket) = next_bucket {
debug!(
event = EVENT_REBALANCE_BUCKET,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index,
bucket = %bucket,
state = "started",
"Starting rebalance bucket"
);
let outcome = match resolve_rebalance_bucket_result(
self.rebalance_bucket(rx.clone(), bucket.clone(), pool_index).await,
pool_index,
&bucket,
) {
Ok(outcome) => outcome,
Err(err) => {
error!(
event = EVENT_REBALANCE_BUCKET,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index,
bucket = %bucket,
state = "bucket_failed",
error = ?err,
"Rebalance bucket failed"
);
final_result = Err(resolve_rebalance_terminal_error(
err.clone(),
send_rebalance_done_signal(&done_tx, Err(err.clone()), pool_index).await,
));
break;
}
};
if let RebalanceBucketOutcome::Deferred { last_error } = outcome {
if !deferred_buckets.insert(bucket.clone()) {
let err = Error::other(format!(
"rebalance bucket {bucket} deferred repeatedly due to transient object failures: {last_error}"
));
error!(
event = EVENT_REBALANCE_BUCKET,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index,
bucket = %bucket,
state = "bucket_deferred_repeatedly",
error = ?err,
"Rebalance bucket failed after repeated deferral"
);
final_result = Err(resolve_rebalance_terminal_error(
err.clone(),
send_rebalance_done_signal(&done_tx, Err(err.clone()), pool_index).await,
));
break;
}
warn!(
event = EVENT_REBALANCE_BUCKET,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index,
bucket = %bucket,
state = "deferred",
error = %last_error,
"Deferred rebalance bucket after transient object failures"
);
if let Err(err) = self.defer_rebalance_bucket(pool_index, bucket.clone(), last_error).await {
error!(
event = EVENT_REBALANCE_BUCKET,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index,
bucket = %bucket,
state = "defer_failed",
error = ?err,
"Rebalance bucket defer failed"
);
final_result = Err(resolve_rebalance_terminal_error(
err.clone(),
send_rebalance_done_signal(&done_tx, Err(err.clone()), pool_index).await,
));
break;
}
continue;
}
debug!(
event = EVENT_REBALANCE_BUCKET,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index,
bucket = %bucket,
state = "completed",
"Completed rebalance bucket"
);
if let Err(err) = self.bucket_rebalance_done(pool_index, bucket).await {
error!(
event = EVENT_REBALANCE_BUCKET,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index,
state = "bucket_done_mark_failed",
error = ?err,
"Rebalance bucket completion mark failed"
);
final_result = Err(resolve_rebalance_terminal_error(
err.clone(),
send_rebalance_done_signal(&done_tx, Err(err.clone()), pool_index).await,
));
break;
}
} else {
debug!(
event = EVENT_REBALANCE_BUCKET,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index,
state = "idle",
reason = "no_bucket_to_rebalance",
"No rebalance bucket available"
);
break;
}
}
info!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index,
state = "pool_done",
"Rebalance worker finished"
);
if final_result.is_ok()
&& let Err(err) = send_rebalance_done_signal(&done_tx, Ok(()), pool_index).await
{
final_result = Err(err);
}
drop(done_tx);
if let Err(err) = resolve_rebalance_save_task_result(pool_index, save_task.await)
&& final_result.is_ok()
{
final_result = Err(err);
}
debug!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index,
state = "pool_result_returned",
"Rebalance worker result returned"
);
final_result
}
pub(super) async fn check_if_rebalance_done(&self, pool_index: usize) -> bool {
let mut rebalance_meta = self.rebalance_meta.write().await;
if let Some(meta) = rebalance_meta.as_mut()
&& let Some(pool_stat) = meta.pool_stats.get_mut(pool_index)
{
// Check if the pool's rebalance status is already completed
if pool_stat.info.status == RebalStatus::Completed {
debug!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index,
state = "already_completed",
"Rebalance pool is already completed"
);
return true;
}
// Mark pool rebalance as done only after it reaches the PercentFreeGoal.
let pfi = if pool_stat.init_capacity == 0 {
0.0
} else {
(pool_stat.init_free_space + pool_stat.bytes) as f64 / pool_stat.init_capacity as f64
};
if !has_deferred_rebalance_error(pool_stat)
&& rebalance_goal_reached(
pool_stat.init_free_space,
pool_stat.init_capacity,
pool_stat.bytes,
meta.percent_free_goal,
)
{
pool_stat.info.status = RebalStatus::Completed;
pool_stat.info.end_time = Some(OffsetDateTime::now_utc());
info!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index,
state = "completed",
percent_free = pfi,
"Marked rebalance pool completed"
);
return true;
}
}
false
}
}
impl ECStore {
#[tracing::instrument(skip(self))]
pub async fn save_rebalance_stats(&self, pool_idx: usize, opt: RebalSaveOpt) -> Result<()> {
let meta_to_save = {
let mut rebalance_meta = self.rebalance_meta.write().await;
let Some(meta) = rebalance_meta.as_mut() else {
return Ok(());
};
let now = OffsetDateTime::now_utc();
apply_rebalance_save_option(meta, pool_idx, opt, now);
meta.clone()
};
let pool = clone_first_arc(&self.pools, "save_rebalance_stats: no pools available")?;
debug!(
event = EVENT_REBALANCE_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
pool_index = pool_idx,
save_opt = ?opt,
state = "metadata_save_requested",
"Rebalance metadata save requested"
);
let stage = format!("save_rebalance_stats for pool {pool_idx} opt {opt:?}");
resolve_rebalance_meta_save_result(
self.save_rebalance_meta_with_merge(pool, &meta_to_save, stage.as_str()).await,
stage.as_str(),
)?;
Ok(())
}
}
crates/ecstore/src/rebalance/types.rs
+119
View File
@@ -0,0 +1,119 @@
use serde::{Deserialize, Serialize};
use std::sync::Arc;
use time::OffsetDateTime;
use tokio_util::sync::CancellationToken;
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct RebalanceStats {
#[serde(rename = "ifs")]
pub init_free_space: u64, // Pool free space at the start of rebalance
#[serde(rename = "ic")]
pub init_capacity: u64, // Pool capacity at the start of rebalance
#[serde(rename = "bus")]
pub buckets: Vec<String>, // Buckets being rebalanced or to be rebalanced
#[serde(rename = "rbs")]
pub rebalanced_buckets: Vec<String>, // Buckets rebalanced
#[serde(rename = "bu")]
pub bucket: String, // Last rebalanced bucket
#[serde(rename = "ob")]
pub object: String, // Last rebalanced object
#[serde(rename = "no")]
pub num_objects: u64, // Number of objects rebalanced
#[serde(rename = "nv")]
pub num_versions: u64, // Number of versions rebalanced
#[serde(rename = "bs")]
pub bytes: u64, // Number of bytes rebalanced
#[serde(rename = "par")]
pub participating: bool, // Whether the pool is participating in rebalance
#[serde(rename = "inf")]
pub info: RebalanceInfo, // Rebalance operation info
#[serde(rename = "cw", default)]
pub cleanup_warnings: RebalanceCleanupWarnings,
}
pub type RStats = Vec<Arc<RebalanceStats>>;
#[derive(Debug, Default)]
pub(super) struct RebalanceBucketConfigs {
pub(super) lifecycle_config: Option<s3s::dto::BucketLifecycleConfiguration>,
pub(super) lock_retention: Option<s3s::dto::DefaultRetention>,
pub(super) replication_config: Option<(s3s::dto::ReplicationConfiguration, OffsetDateTime)>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(super) enum RebalanceBucketOutcome {
Completed,
Deferred { last_error: String },
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(super) enum RebalanceEntryOutcome {
Completed,
Deferred { last_error: String },
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
pub enum RebalStatus {
#[default]
None,
Started,
Completed,
Stopped,
Failed,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
pub enum RebalSaveOpt {
#[default]
Stats,
StoppedAt,
}
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct RebalanceInfo {
#[serde(rename = "startTs")]
pub start_time: Option<OffsetDateTime>, // Time at which rebalance-start was issued
#[serde(rename = "stopTs")]
pub end_time: Option<OffsetDateTime>, // Time at which rebalance operation completed or rebalance-stop was called
#[serde(rename = "err")]
pub last_error: Option<String>, // Last rebalance error message
#[serde(rename = "status")]
pub status: RebalStatus, // Current state of rebalance operation
}
#[derive(Debug, Default, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct RebalanceCleanupWarnings {
#[serde(rename = "count", default)]
pub count: u64,
#[serde(rename = "lastMsg", default)]
pub last_message: Option<String>,
#[serde(rename = "lastBucket", default)]
pub last_bucket: Option<String>,
#[serde(rename = "lastObject", default)]
pub last_object: Option<String>,
#[serde(rename = "lastAt", default)]
pub last_at: Option<OffsetDateTime>,
}
#[allow(dead_code)]
#[derive(Debug, Clone, Default)]
pub struct DiskStat {
pub total_space: u64,
pub available_space: u64,
}
#[derive(Debug, Default, Serialize, Deserialize, Clone)]
pub struct RebalanceMeta {
#[serde(skip)]
pub cancel: Option<CancellationToken>, // To be invoked on rebalance-stop
#[serde(skip)]
pub last_refreshed_at: Option<OffsetDateTime>,
#[serde(rename = "stopTs")]
pub stopped_at: Option<OffsetDateTime>, // Time when rebalance-stop was issued
#[serde(rename = "id")]
pub id: String, // ID of the ongoing rebalance operation
#[serde(rename = "pf")]
pub percent_free_goal: f64, // Computed from total free space and capacity at the start of rebalance
#[serde(rename = "rss")]
pub pool_stats: Vec<RebalanceStats>, // Per-pool rebalance stats keyed by pool index
}
crates/utils/src/os/windows.rs
+1 -1
View File
@@ -170,7 +170,7 @@ pub fn get_physical_device_ids(disk: &str) -> std::io::Result<Vec<String>> {
/// This is the Windows equivalent of Linux's `st_dev` major:minor pair and is
/// useful for diagnostic purposes when validating physical disk independence.
// SAFETY: `path_wide` is a valid null-terminated UTF-16 string and all output
// pointers target initialized stack variables with documented MAX_PATH sizing.
// SAFETY: pointers target initialized stack variables with documented MAX_PATH sizing.
#[allow(unsafe_code)]
pub fn get_volume_serial_number(path: &str) -> std::io::Result<u32> {
let path_wide = to_wide_path(Path::new(path));