gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-05 21:07:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(heal): recover shards after node rejoin (#3814)

Browse Source
This commit is contained in:
cxymds
2026-06-24 12:51:16 +08:00
committed by GitHub
parent aa089f18f2
commit 5eda332fee
6 changed files with 638 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/admin/handlers/heal.rs
+126 -27
View File
@@ -194,6 +194,8 @@ struct HealTaskStatus {
heal_settings: HealOpts,
#[serde(skip_serializing_if = "Vec::is_empty")]
items: Vec<rustfs_madmin::heal_commands::HealResultItem>,
#[serde(skip_serializing_if = "Option::is_none")]
progress: Option<serde_json::Value>,
}
#[derive(Debug, Serialize)]
@@ -211,6 +213,8 @@ struct HealTaskStatusPayload {
summary: String,
#[serde(default)]
items: Vec<rustfs_madmin::heal_commands::HealResultItem>,
#[serde(default)]
progress: Option<serde_json::Value>,
}
fn map_heal_response(result: Option<HealResp>) -> S3Result<(StatusCode, Vec<u8>)> {
@@ -253,6 +257,7 @@ fn encode_heal_task_status(
failure_detail: String,
heal_settings: HealOpts,
items: Vec<rustfs_madmin::heal_commands::HealResultItem>,
progress: Option<serde_json::Value>,
) -> S3Result<Vec<u8>> {
encode_json(&HealTaskStatus {
summary,
@@ -260,12 +265,14 @@ fn encode_heal_task_status(
start_time: current_rfc3339_time()?,
heal_settings,
items,
progress,
})
}
fn build_heal_channel_request(hip: &HealInitParams) -> HealChannelRequest {
let root_erasure_set_target =
hip.bucket.is_empty() && hip.obj_prefix.is_empty() && matches!((hip.hs.pool, hip.hs.set), (Some(_), Some(_)));
let root_cluster_target = hip.bucket.is_empty() && hip.obj_prefix.is_empty() && hip.hs.pool.is_none() && hip.hs.set.is_none();
let recursive = if (!hip.bucket.is_empty() && hip.obj_prefix.is_empty()) || root_erasure_set_target {
true
} else {
@@ -291,7 +298,7 @@ fn build_heal_channel_request(hip: &HealInitParams) -> HealChannelRequest {
heal_request.remove_corrupted = Some(hip.hs.remove);
heal_request.recreate_missing = Some(hip.hs.recreate);
heal_request.update_parity = Some(hip.hs.update_parity);
heal_request.recursive = Some(recursive);
heal_request.recursive = Some(recursive || root_cluster_target);
heal_request.dry_run = Some(hip.hs.dry_run);
heal_request.source = HealRequestSource::Admin;
heal_request
@@ -299,15 +306,15 @@ fn build_heal_channel_request(hip: &HealInitParams) -> HealChannelRequest {
fn heal_channel_response_status(
response: &rustfs_common::heal_channel::HealChannelResponse,
) -> (String, Vec<rustfs_madmin::heal_commands::HealResultItem>) {
) -> (String, Vec<rustfs_madmin::heal_commands::HealResultItem>, Option<serde_json::Value>) {
let Some(data) = response.data.as_deref() else {
return ("running".to_string(), Vec::new());
return ("running".to_string(), Vec::new(), None);
};
if let Ok(payload) = serde_json::from_slice::<HealTaskStatusPayload>(data)
&& !payload.summary.is_empty()
{
return (payload.summary, payload.items);
return (payload.summary, payload.items, payload.progress);
}
let summary = std::str::from_utf8(data)
@@ -315,7 +322,7 @@ fn heal_channel_response_status(
.filter(|summary| !summary.is_empty())
.unwrap_or("running")
.to_string();
(summary, Vec::new())
(summary, Vec::new(), None)
}
#[cfg(test)]
@@ -330,6 +337,11 @@ fn heal_channel_response_items(
heal_channel_response_status(response).1
}
#[cfg(test)]
fn heal_channel_response_progress(response: &rustfs_common::heal_channel::HealChannelResponse) -> Option<serde_json::Value> {
heal_channel_response_status(response).2
}
fn encode_background_heal_status(
info: &BackgroundHealInfo,
heal_operations: rustfs_heal::HealOperationsSnapshot,
@@ -362,20 +374,16 @@ fn validate_heal_request_mode(hip: &HealInitParams) -> S3Result<()> {
(Some(_), None) | (None, Some(_)) => {
Err(s3_error!(InvalidRequest, "root heal erasure-set target requires both pool and set"))
}
(None, None) => Err(s3_error!(InvalidRequest, "starting heal without a bucket target is not supported")),
(None, None) if hip.hs.recursive => Ok(()),
(None, None) => Err(s3_error!(InvalidRequest, "root heal requires recursive=true or a bucket target")),
};
}
Ok(())
}
fn should_handle_root_heal_directly(hip: &HealInitParams) -> bool {
hip.bucket.is_empty()
&& hip.obj_prefix.is_empty()
&& hip.client_token.is_empty()
&& !hip.force_stop
&& hip.hs.pool.is_none()
&& hip.hs.set.is_none()
fn should_handle_root_heal_directly(_hip: &HealInitParams) -> bool {
false
}
fn map_root_heal_status(heal_err: Option<super::super::Error>) -> S3Result<()> {
@@ -529,9 +537,14 @@ impl Operation for HealHandler {
spawn_traced(async move {
match rustfs_common::heal_channel::query_heal_status(heal_path_str, client_token).await {
Ok(response) if response.success => {
let (summary, items) = heal_channel_response_status(&response);
let resp_bytes =
encode_heal_task_status(summary, response.error.unwrap_or_default(), HealOpts::default(), items);
let (summary, items, progress) = heal_channel_response_status(&response);
let resp_bytes = encode_heal_task_status(
summary,
response.error.unwrap_or_default(),
HealOpts::default(),
items,
progress,
);
match resp_bytes {
Ok(resp_bytes) => {
let _ = tx_clone
@@ -582,8 +595,8 @@ impl Operation for HealHandler {
let resp_bytes = if client_token.is_empty() {
encode_heal_start_success(response.request_id, client_address)
} else {
let (summary, items) = heal_channel_response_status(&response);
encode_heal_task_status(summary, response.error.unwrap_or_default(), heal_settings, items)
let (summary, items, progress) = heal_channel_response_status(&response);
encode_heal_task_status(summary, response.error.unwrap_or_default(), heal_settings, items, progress)
};
match resp_bytes {
Ok(resp_bytes) => {
@@ -734,8 +747,9 @@ mod tests {
use super::extract_heal_init_params;
use super::{
HealInitParams, HealResp, build_heal_channel_request, encode_background_heal_status, encode_heal_start_success,
encode_heal_task_status, heal_channel_response_items, heal_channel_response_summary, json_response, map_heal_response,
map_root_heal_status, should_handle_root_heal_directly, validate_heal_request_mode, validate_heal_target,
encode_heal_task_status, heal_channel_response_items, heal_channel_response_progress, heal_channel_response_summary,
json_response, map_heal_response, map_root_heal_status, should_handle_root_heal_directly, validate_heal_request_mode,
validate_heal_target,
};
use bytes::Bytes;
use http::StatusCode;
@@ -918,6 +932,31 @@ mod tests {
assert_eq!(request.recreate_missing, Some(true));
}
#[test]
fn test_root_recursive_heal_channel_request_targets_cluster() {
let hip = HealInitParams {
hs: HealOpts {
recursive: true,
scan_mode: HealScanMode::Deep,
recreate: true,
..Default::default()
},
force_start: true,
..Default::default()
};
let request = build_heal_channel_request(&hip);
assert_eq!(request.bucket, "");
assert_eq!(request.disk, None);
assert_eq!(request.object_prefix, None);
assert_eq!(request.priority, HealChannelPriority::High);
assert_eq!(request.source, HealRequestSource::Admin);
assert_eq!(request.scan_mode, Some(HealScanMode::Deep));
assert_eq!(request.recursive, Some(true));
assert_eq!(request.recreate_missing, Some(true));
}
#[test]
fn test_bucket_heal_channel_request_defaults_to_recursive() {
let hip = HealInitParams {
@@ -973,10 +1012,22 @@ mod tests {
assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
assert!(
err.to_string()
.contains("starting heal without a bucket target is not supported")
.contains("root heal requires recursive=true or a bucket target")
);
}
#[test]
fn test_validate_heal_request_mode_allows_root_recursive_heal_start() {
validate_heal_request_mode(&HealInitParams {
hs: HealOpts {
recursive: true,
..Default::default()
},
..Default::default()
})
.expect("root recursive heal should start tracked cluster heal");
}
#[test]
fn test_validate_heal_request_mode_allows_root_erasure_set_target() {
validate_heal_request_mode(&HealInitParams {
@@ -1009,9 +1060,9 @@ mod tests {
}
#[test]
fn test_should_handle_root_heal_directly_for_root_start_modes() {
assert!(should_handle_root_heal_directly(&HealInitParams::default()));
assert!(should_handle_root_heal_directly(&HealInitParams {
fn test_should_handle_root_heal_directly_is_disabled_for_root_start_modes() {
assert!(!should_handle_root_heal_directly(&HealInitParams::default()));
assert!(!should_handle_root_heal_directly(&HealInitParams {
force_start: true,
..Default::default()
}));
@@ -1169,9 +1220,14 @@ mod tests {
#[test]
fn test_encode_heal_task_status_uses_client_wire_shape() {
let encoded =
encode_heal_task_status("Heal status query accepted".to_string(), String::new(), HealOpts::default(), Vec::new())
.expect("status response should serialize");
let encoded = encode_heal_task_status(
"Heal status query accepted".to_string(),
String::new(),
HealOpts::default(),
Vec::new(),
None,
)
.expect("status response should serialize");
let json: serde_json::Value = serde_json::from_slice(&encoded).expect("json should deserialize");
assert_eq!(json["summary"], "Heal status query accepted");
@@ -1182,6 +1238,26 @@ mod tests {
OffsetDateTime::parse(start_time, &Rfc3339).expect("startTime should be RFC3339");
}
#[test]
fn test_encode_heal_task_status_preserves_progress() {
let progress = json!({
"objectsScanned": 7,
"objectsHealed": 3,
"currentObject": "bucket-a/object-a"
});
let encoded = encode_heal_task_status(
"running".to_string(),
String::new(),
HealOpts::default(),
Vec::new(),
Some(progress.clone()),
)
.expect("status response should serialize");
let json: serde_json::Value = serde_json::from_slice(&encoded).expect("json should deserialize");
assert_eq!(json["progress"], progress);
}
#[test]
fn test_build_heal_channel_request_preserves_client_options() {
let hip = HealInitParams {
@@ -1264,6 +1340,29 @@ mod tests {
assert_eq!(items[0].object_size, 1024);
}
#[test]
fn test_heal_channel_response_status_preserves_progress() {
let progress = serde_json::json!({
"objectsScanned": 4,
"objectsHealed": 2,
"currentObject": "bucket-a/object-a"
});
let payload = serde_json::json!({
"summary": "running",
"items": [],
"progress": progress
});
let response = rustfs_common::heal_channel::create_heal_response(
"token".to_string(),
true,
Some(serde_json::to_vec(&payload).expect("payload should serialize")),
None,
);
assert_eq!(heal_channel_response_summary(&response), "running");
assert_eq!(heal_channel_response_progress(&response), Some(progress));
}
#[test]
fn test_json_response_sets_application_json_content_type() {
let response = json_response(StatusCode::OK, b"{}".to_vec());