gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-09 06:39:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(rebalance): fence peers before activation (#5842)

Browse Source
This commit is contained in:
cxymds
2026-08-08 19:55:50 +08:00
committed by GitHub
parent c2e23411e8
commit a7de957eb8
4 changed files with 374 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/services/rebalance/control.rs
+43 -10
View File
@@ -327,16 +327,8 @@ impl ECStore {
#[tracing::instrument(skip(self, bucktes))]
pub async fn init_and_start_rebalance(self: &Arc<Self>, bucktes: Vec<String>) -> Result<String> {
let _start_guard = self.start_gate.lock().await;
let decommission_running = self.is_decommission_running().await;
{
let rebalance_meta = self.rebalance_meta.read().await;
validate_init_rebalance_state(decommission_running, rebalance_meta.as_ref())?;
}
let id = self.init_rebalance_meta(bucktes).await?;
if let Err(start_err) = self.start_rebalance().await {
let id = self.init_rebalance_start(bucktes).await?;
if let Err(start_err) = self.start_rebalance_for_id(&id).await {
if let Err(rollback_err) = self
.rollback_rebalance_start_without_worker_for_id(Some(&id), start_err.to_string())
.await
@@ -354,6 +346,47 @@ impl ECStore {
Ok(id)
}
#[tracing::instrument(skip(self, bucktes))]
pub async fn init_rebalance_start(self: &Arc<Self>, bucktes: Vec<String>) -> Result<String> {
let _start_guard = self.start_gate.lock().await;
let decommission_running = self.is_decommission_running().await;
{
let rebalance_meta = self.rebalance_meta.read().await;
validate_init_rebalance_state(decommission_running, rebalance_meta.as_ref())?;
}
self.init_rebalance_meta(bucktes).await
}
#[tracing::instrument(skip(self))]
pub async fn start_rebalance_for_id(self: &Arc<Self>, expected_id: &str) -> Result<()> {
let _start_guard = self.start_gate.lock().await;
{
let rebalance_meta = self.rebalance_meta.read().await;
let Some(meta) = rebalance_meta.as_ref() else {
return Err(Error::ConfigNotFound);
};
if meta.id != expected_id {
return Err(Error::other(format!(
"rebalance metadata changed before start: expected {expected_id}, found {}",
meta.id
)));
}
if meta.stopped_at.is_some() {
return Err(Error::other(format!("rebalance {expected_id} was stopped before start")));
}
}
self.start_rebalance().await
}
pub async fn rollback_rebalance_start_for_id(self: &Arc<Self>, expected_id: Option<&str>, start_error: String) -> Result<()> {
self.rollback_rebalance_start_without_worker_for_id(expected_id, start_error)
.await
}
#[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
crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs
+67 -15
View File
@@ -2584,21 +2584,7 @@ async fn test_init_and_start_rebalance_rejects_second_start_after_gate() {
}],
..Default::default()
};
let endpoint_pools: crate::layout::endpoints::EndpointServerPools = Vec::new().into();
let store = Arc::new(crate::store::ECStore {
id: uuid::Uuid::new_v4(),
disk_map: std::collections::HashMap::new(),
pools: Vec::new(),
peer_sys: crate::cluster::rpc::S3PeerSys::new(&endpoint_pools),
pool_meta: tokio::sync::RwLock::new(crate::core::pools::PoolMeta::default()),
rebalance_meta: tokio::sync::RwLock::new(Some(active_meta)),
decommission_cancelers: tokio::sync::RwLock::new(Vec::new()),
start_gate: tokio::sync::Mutex::new(()),
pool_meta_save_gate: tokio::sync::Mutex::new(()),
ctx: crate::runtime::instance::bootstrap_ctx(),
bucket_fence_registry: std::sync::Arc::default(),
});
let store = test_store_with_rebalance_meta(active_meta);
let err = store
.init_and_start_rebalance(vec!["bucket".to_string()])
@@ -2608,6 +2594,72 @@ async fn test_init_and_start_rebalance_rejects_second_start_after_gate() {
assert!(matches!(err, Error::RebalanceAlreadyRunning));
}
#[tokio::test]
async fn test_start_rebalance_for_id_rejects_changed_metadata() {
let meta = RebalanceMeta {
id: "rebalance-a".to_string(),
pool_stats: vec![RebalanceStats {
participating: true,
info: RebalanceInfo {
status: RebalStatus::Started,
..Default::default()
},
..Default::default()
}],
..Default::default()
};
let store = test_store_with_rebalance_meta(meta);
let err = store
.start_rebalance_for_id("rebalance-b")
.await
.expect_err("staged start must not start changed metadata");
assert!(err.to_string().contains("rebalance metadata changed before start"));
}
#[tokio::test]
async fn test_start_rebalance_for_id_rejects_stopped_metadata() {
let meta = RebalanceMeta {
id: "rebalance-a".to_string(),
stopped_at: Some(OffsetDateTime::now_utc()),
pool_stats: vec![RebalanceStats {
participating: true,
info: RebalanceInfo {
status: RebalStatus::Started,
..Default::default()
},
..Default::default()
}],
..Default::default()
};
let store = test_store_with_rebalance_meta(meta);
let err = store
.start_rebalance_for_id("rebalance-a")
.await
.expect_err("staged start must not restart stopped metadata");
assert!(err.to_string().contains("was stopped before start"));
}
fn test_store_with_rebalance_meta(meta: RebalanceMeta) -> Arc<crate::store::ECStore> {
let endpoint_pools: crate::layout::endpoints::EndpointServerPools = Vec::new().into();
Arc::new(crate::store::ECStore {
id: uuid::Uuid::new_v4(),
disk_map: std::collections::HashMap::new(),
pools: Vec::new(),
peer_sys: crate::cluster::rpc::S3PeerSys::new(&endpoint_pools),
pool_meta: tokio::sync::RwLock::new(crate::core::pools::PoolMeta::default()),
rebalance_meta: tokio::sync::RwLock::new(Some(meta)),
decommission_cancelers: tokio::sync::RwLock::new(Vec::new()),
start_gate: tokio::sync::Mutex::new(()),
pool_meta_save_gate: tokio::sync::Mutex::new(()),
ctx: crate::runtime::instance::bootstrap_ctx(),
bucket_fence_registry: std::sync::Arc::default(),
})
}
#[test]
fn test_percent_free_ratio_zero_capacity_is_zero() {
assert_eq!(percent_free_ratio(100, 0), 0.0);
rustfs/src/admin/handlers/rebalance.rs
+263 -69
View File
@@ -39,7 +39,7 @@ use rustfs_utils::{
http::{AMZ_REQUEST_ID, REQUEST_ID_HEADER},
};
use s3s::{
Body, S3Request, S3Response, S3Result,
Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result,
header::{CONTENT_LENGTH, CONTENT_TYPE},
s3_error,
};
@@ -102,6 +102,10 @@ fn rebalance_start_rollback_error(start_err: &str, rollback_result: &Result<(),
}
}
fn rebalance_internal_error(message: impl Into<String>) -> S3Error {
S3Error::with_message(S3ErrorCode::InternalError, message.into())
}
fn rebalance_rollback_stop_failure_message(rebalance_id: &str, failures: &[String]) -> String {
format!("cluster stop_rebalance rollback for {rebalance_id} partial: {}", failures.join("; "))
}
@@ -128,6 +132,28 @@ fn rebalance_rollback_failure_message(
failures.join("; ")
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum RebalanceStartStep {
PropagateFence,
StartLocal,
PropagateWorkers,
}
const DISTRIBUTED_REBALANCE_START_STEPS: [RebalanceStartStep; 3] = [
RebalanceStartStep::PropagateFence,
RebalanceStartStep::StartLocal,
RebalanceStartStep::PropagateWorkers,
];
const LOCAL_REBALANCE_START_STEPS: [RebalanceStartStep; 1] = [RebalanceStartStep::StartLocal];
fn rebalance_start_steps(has_notification_sys: bool) -> &'static [RebalanceStartStep] {
if has_notification_sys {
&DISTRIBUTED_REBALANCE_START_STEPS
} else {
&LOCAL_REBALANCE_START_STEPS
}
}
async fn rollback_cluster_rebalance_start(
store: &Arc<ECStore>,
notification_sys: Option<&NotificationSys>,
@@ -177,6 +203,50 @@ async fn rollback_cluster_rebalance_start(
Ok(())
}
async fn rollback_rebalance_start_for_admin(
store: &Arc<ECStore>,
notification_sys: Option<&NotificationSys>,
rebalance_id: &str,
start_err: &str,
request_id: &str,
actor: &str,
remote_addr: &str,
) -> S3Result<()> {
let rollback_result = rollback_cluster_rebalance_start(store, notification_sys, rebalance_id).await;
let rollback_label = rollback_result_label(&rollback_result);
match &rollback_result {
Ok(_) => info!(
event = EVENT_ADMIN_REBALANCE_STATE,
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_REBALANCE,
action = "start",
result = rollback_label,
request_id = %request_id,
actor = %actor,
remote_addr = %remote_addr,
rebalance_id = %rebalance_id,
propagation_error = %start_err,
"admin rebalance state"
),
Err(rollback_err) => error!(
event = EVENT_ADMIN_REBALANCE_STATE,
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_REBALANCE,
action = "start",
result = rollback_label,
request_id = %request_id,
actor = %actor,
remote_addr = %remote_addr,
rebalance_id = %rebalance_id,
propagation_error = %start_err,
rollback_error = %rollback_err,
"admin rebalance state"
),
}
Err(rebalance_internal_error(rebalance_start_rollback_error(start_err, &rollback_result)))
}
pub fn register_rebalance_route(r: &mut S3Router<AdminOperation>) -> std::io::Result<()> {
r.insert(
Method::POST,
@@ -473,7 +543,7 @@ impl Operation for RebalanceStart {
let buckets: Vec<String> = bucket_infos.into_iter().map(|bucket| bucket.name).collect();
let id = match store.init_and_start_rebalance(buckets).await {
let id = match store.init_rebalance_start(buckets).await {
Ok(id) => id,
Err(StorageError::DecommissionAlreadyRunning) => {
log_rebalance_request_rejected("start", "decommission_in_progress", &request_id, &actor, &remote_addr);
@@ -484,7 +554,7 @@ impl Operation for RebalanceStart {
return Err(s3_error!(OperationAborted, "rebalance is already in progress"));
}
Err(e) => {
return Err(s3_error!(InternalError, "failed to start rebalance: {}", e));
return Err(s3_error!(InternalError, "failed to initialize rebalance: {}", e));
}
};
@@ -493,80 +563,186 @@ impl Operation for RebalanceStart {
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_REBALANCE,
action = "start",
state = "started",
state = "metadata_initialized",
request_id = %request_id,
actor = %actor,
remote_addr = %remote_addr,
rebalance_id = %id,
"admin rebalance state"
);
if let Some(notification_sys) = current_notification_system() {
info!(
event = EVENT_ADMIN_REBALANCE_STATE,
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_REBALANCE,
action = "start",
state = "propagation_started",
request_id = %request_id,
actor = %actor,
remote_addr = %remote_addr,
rebalance_id = %id,
"admin rebalance state"
);
if let Err(err) = notification_sys.load_rebalance_meta(true).await {
error!(
event = EVENT_ADMIN_REBALANCE_STATE,
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_REBALANCE,
action = "start",
result = "propagation_failed",
request_id = %request_id,
actor = %actor,
remote_addr = %remote_addr,
rebalance_id = %id,
error = %err,
"admin rebalance state"
);
let notification_sys = current_notification_system();
for step in rebalance_start_steps(notification_sys.is_some()) {
match step {
RebalanceStartStep::PropagateFence => {
if let Some(notification_sys) = notification_sys.as_ref() {
info!(
event = EVENT_ADMIN_REBALANCE_STATE,
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_REBALANCE,
action = "start",
state = "fence_propagation_started",
request_id = %request_id,
actor = %actor,
remote_addr = %remote_addr,
rebalance_id = %id,
"admin rebalance state"
);
if let Err(err) = notification_sys.load_rebalance_meta(false).await {
error!(
event = EVENT_ADMIN_REBALANCE_STATE,
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_REBALANCE,
action = "start",
result = "fence_propagation_failed",
request_id = %request_id,
actor = %actor,
remote_addr = %remote_addr,
rebalance_id = %id,
error = %err,
"admin rebalance state"
);
let start_err = err.to_string();
let rollback_result = rollback_cluster_rebalance_start(&store, Some(&notification_sys), &id).await;
let rollback_label = rollback_result_label(&rollback_result);
match &rollback_result {
Ok(_) => info!(
event = EVENT_ADMIN_REBALANCE_STATE,
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_REBALANCE,
action = "start",
result = rollback_label,
request_id = %request_id,
actor = %actor,
remote_addr = %remote_addr,
rebalance_id = %id,
propagation_error = %start_err,
"admin rebalance state"
),
Err(rollback_err) => error!(
event = EVENT_ADMIN_REBALANCE_STATE,
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_REBALANCE,
action = "start",
result = rollback_label,
request_id = %request_id,
actor = %actor,
remote_addr = %remote_addr,
rebalance_id = %id,
propagation_error = %start_err,
rollback_error = %rollback_err,
"admin rebalance state"
),
let start_err = err.to_string();
rollback_rebalance_start_for_admin(
&store,
Some(notification_sys),
&id,
&start_err,
&request_id,
&actor,
&remote_addr,
)
.await?;
}
}
}
RebalanceStartStep::StartLocal => {
if let Err(err) = store.start_rebalance_for_id(&id).await {
error!(
event = EVENT_ADMIN_REBALANCE_STATE,
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_REBALANCE,
action = "start",
result = "local_start_failed",
request_id = %request_id,
actor = %actor,
remote_addr = %remote_addr,
rebalance_id = %id,
error = %err,
"admin rebalance state"
);
return Err(s3_error!(
InternalError,
"{}",
rebalance_start_rollback_error(&start_err, &rollback_result)
));
let start_err = err.to_string();
if let Err(rollback_err) = store.rollback_rebalance_start_for_id(Some(&id), start_err.clone()).await {
error!(
event = EVENT_ADMIN_REBALANCE_STATE,
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_REBALANCE,
action = "start",
result = "local_start_rollback_failed",
request_id = %request_id,
actor = %actor,
remote_addr = %remote_addr,
rebalance_id = %id,
start_error = %start_err,
rollback_error = %rollback_err,
"admin rebalance state"
);
return Err(rebalance_internal_error(format!(
"failed to start rebalance after metadata initialized for {id}; rollback failed: {rollback_err}"
)));
}
info!(
event = EVENT_ADMIN_REBALANCE_STATE,
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_REBALANCE,
action = "start",
result = "local_start_rollback_success",
request_id = %request_id,
actor = %actor,
remote_addr = %remote_addr,
rebalance_id = %id,
start_error = %start_err,
"admin rebalance state"
);
if let Some(notification_sys) = notification_sys.as_ref() {
let terminal_reload_attempt_at = OffsetDateTime::now_utc();
let terminal_reload_failures = match notification_sys.load_rebalance_meta_failures(false).await {
Ok(failures) => failures,
Err(err) => vec![format!("terminal rebalance reload rollback for {id} failed: {err}")],
};
if !terminal_reload_failures.is_empty() {
let record = RebalanceStopPropagationRecord {
stop_attempt_at: None,
stop_failures: Vec::new(),
terminal_reload_attempt_at: Some(terminal_reload_attempt_at),
terminal_reload_failures: terminal_reload_failures.clone(),
};
store.record_rebalance_stop_propagation(record).await.map_err(|err| {
rebalance_internal_error(format!(
"failed to persist rebalance local-start rollback propagation metadata: {err}"
))
})?;
return Err(rebalance_internal_error(format!(
"failed to start rebalance after metadata initialized for {}; local metadata was finalized as failed, but terminal peer reload was incomplete: {}",
id,
rebalance_rollback_terminal_reload_failure_message(&id, &terminal_reload_failures)
)));
}
}
return Err(rebalance_internal_error(format!(
"failed to start rebalance after metadata initialized for {id}; local metadata was finalized as failed: {start_err}"
)));
}
}
RebalanceStartStep::PropagateWorkers => {
if let Some(notification_sys) = notification_sys.as_ref() {
info!(
event = EVENT_ADMIN_REBALANCE_STATE,
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_REBALANCE,
action = "start",
state = "worker_propagation_started",
request_id = %request_id,
actor = %actor,
remote_addr = %remote_addr,
rebalance_id = %id,
"admin rebalance state"
);
if let Err(err) = notification_sys.load_rebalance_meta(true).await {
error!(
event = EVENT_ADMIN_REBALANCE_STATE,
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_REBALANCE,
action = "start",
result = "worker_propagation_failed",
request_id = %request_id,
actor = %actor,
remote_addr = %remote_addr,
rebalance_id = %id,
error = %err,
"admin rebalance state"
);
let start_err = err.to_string();
rollback_rebalance_start_for_admin(
&store,
Some(notification_sys),
&id,
&start_err,
&request_id,
&actor,
&remote_addr,
)
.await?;
}
}
}
}
}
if notification_sys.is_some() {
info!(
event = EVENT_ADMIN_REBALANCE_STATE,
component = LOG_COMPONENT_ADMIN,
@@ -902,10 +1078,11 @@ mod rebalance_handler_tests {
use super::build_rebalance_pool_progress;
use super::calculate_rebalance_progress;
use super::{
RebalPoolProgress, RebalanceAdminStatus, RebalancePoolStatus, RebalanceStopPropagationStatus,
RebalPoolProgress, RebalanceAdminStatus, RebalancePoolStatus, RebalanceStartStep, RebalanceStopPropagationStatus,
build_rebalance_admin_status, build_rebalance_pool_statuses, build_rebalance_stop_propagation_status,
rebalance_pool_used, rebalance_query_present, rebalance_remaining_buckets, rebalance_rollback_failure_message,
rebalance_rollback_stop_failure_message, rebalance_start_rollback_error, rebalance_used_pct, rollback_result_label,
rebalance_rollback_stop_failure_message, rebalance_start_rollback_error, rebalance_start_steps, rebalance_used_pct,
rollback_result_label,
};
use crate::admin::storage_api::rebalance::{
DiskStat, RebalStatus, RebalanceCleanupWarningEntry, RebalanceCleanupWarnings, RebalanceInfo, RebalanceMeta,
@@ -994,6 +1171,23 @@ mod rebalance_handler_tests {
assert_eq!(rollback_result_label(&rollback_result), "rollback_failed");
}
#[test]
fn test_distributed_rebalance_start_fences_peers_before_workers() {
assert_eq!(
rebalance_start_steps(true),
&[
RebalanceStartStep::PropagateFence,
RebalanceStartStep::StartLocal,
RebalanceStartStep::PropagateWorkers
]
);
}
#[test]
fn test_local_rebalance_start_has_no_peer_propagation_steps() {
assert_eq!(rebalance_start_steps(false), &[RebalanceStartStep::StartLocal]);
}
#[test]
fn test_calculate_rebalance_progress_stopped_by_end_time() {
let start = OffsetDateTime::from_unix_timestamp(1_000).unwrap();
scripts/check_s3s_footprint.sh
+1 -1
View File
@@ -24,7 +24,7 @@ cd "$(dirname "$0")/.."
# Baselines verified on 2026-08-06. Lower-only; see header.
S3S_IMPORT_FILES_BASELINE=236
S3_ERROR_LINES_BASELINE=1679
S3_ERROR_LINES_BASELINE=1678
S3S_PATH_PATTERN='(^|[^"[:alnum:]_])s3s::'
TMP_DIR="$(mktemp -d)"