fran/rustfs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(heal): expose replacement recovery status (#5912)

Browse Source 
Add a v4 admin status endpoint for local durable automatic replacement recovery records without changing the v3 background heal status or peer v1 payloads.

Co-authored-by: heihutu <[email protected]>
This commit is contained in:
houseme
2026-08-10 01:48:18 +00:00
committed by GitHub
co-authored by heihutu
parent f17ea7f146
commit 1aeb84dd6b
6 changed files with 125 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/admin/handlers/heal.rs
+113 -5
View File
@@ -54,6 +54,8 @@ const EVENT_ADMIN_REQUEST_REJECTED: &str = "admin_request_rejected";
const EVENT_ADMIN_REQUEST_FAILED: &str = "admin_request_failed";
const EVENT_ADMIN_RESPONSE_EMITTED: &str = "admin_response_emitted";
const PEER_HEAL_STATUS_TIMEOUT: Duration = Duration::from_secs(5);
pub(crate) const REPLACEMENT_RECOVERY_STATUS_ROUTE_SUFFIX: &str = "/v4/heal/replacement-recovery";
const REPLACEMENT_RECOVERY_STATUS_CONTRACT_VERSION: u32 = 1;
#[derive(Debug, Default, Serialize, Deserialize)]
struct HealInitParams {
@@ -175,6 +177,12 @@ pub fn register_heal_route(r: &mut S3Router<AdminOperation>) -> std::io::Result<
AdminOperation(&BackgroundHealStatusHandler {}),
)?;
r.insert(
Method::GET,
format!("{}{}", ADMIN_PREFIX, REPLACEMENT_RECOVERY_STATUS_ROUTE_SUFFIX).as_str(),
AdminOperation(&ReplacementRecoveryStatusHandler {}),
)?;
Ok(())
}
@@ -1014,6 +1022,54 @@ fn json_response(status: StatusCode, body: Vec<u8>) -> S3Response<(StatusCode, B
S3Response::with_headers((status, Body::from(body)), headers)
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct ReplacementRecoveryStatusResponse {
pub contract_version: u32,
pub path: String,
pub scope: &'static str,
pub local: rustfs_heal::ReplacementRecoverySnapshot,
pub cluster: ReplacementRecoveryClusterStatus,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct ReplacementRecoveryClusterStatus {
pub definitive: bool,
pub reason: &'static str,
}
fn build_replacement_recovery_status_response(
local: rustfs_heal::ReplacementRecoverySnapshot,
) -> ReplacementRecoveryStatusResponse {
ReplacementRecoveryStatusResponse {
contract_version: REPLACEMENT_RECOVERY_STATUS_CONTRACT_VERSION,
path: format!("{}{}", ADMIN_PREFIX, REPLACEMENT_RECOVERY_STATUS_ROUTE_SUFFIX),
scope: "local_survivor_disks",
local,
cluster: ReplacementRecoveryClusterStatus {
definitive: false,
reason: "distributed replacement recovery status requires a peer capability RPC",
},
}
}
fn encode_replacement_recovery_status(response: &ReplacementRecoveryStatusResponse) -> S3Result<Vec<u8>> {
serde_json::to_vec(response).map_err(|e| {
warn!(
event = EVENT_ADMIN_REQUEST_FAILED,
component = LOG_COMPONENT_ADMIN_API,
subsystem = LOG_SUBSYSTEM_HEAL_ADMIN,
operation = "replacement_recovery_status",
result = "failed",
reason = "serialize_replacement_recovery_status_failed",
error = %e,
"admin request failed"
);
s3_error!(InternalError, "failed to serialize replacement recovery status: {e}")
})
}
async fn validate_heal_admin_request(req: &S3Request<Body>) -> S3Result<()> {
let Some(input_cred) = req.credentials.as_ref() else {
return Err(s3_error!(InvalidRequest, "authentication required"));
@@ -1284,16 +1340,39 @@ impl Operation for BackgroundHealStatusHandler {
}
}
pub struct ReplacementRecoveryStatusHandler {}
#[async_trait::async_trait]
impl Operation for ReplacementRecoveryStatusHandler {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
validate_heal_admin_request(&req).await?;
let local = rustfs_heal::current_replacement_recovery_snapshot().await;
let response = build_replacement_recovery_status_response(local);
let body = encode_replacement_recovery_status(&response)?;
info!(
event = EVENT_ADMIN_RESPONSE_EMITTED,
component = LOG_COMPONENT_ADMIN_API,
subsystem = LOG_SUBSYSTEM_HEAL_ADMIN,
operation = "replacement_recovery_status",
result = "success",
"admin response emitted"
);
Ok(json_response(StatusCode::OK, body))
}
}
#[cfg(test)]
mod tests {
use super::extract_heal_init_params;
use super::{
BackgroundHealProgress, HealInitParams, HealResp, HealRuntimeState, aggregate_cluster_heal_status,
background_heal_runtime_state, build_heal_channel_request, encode_background_heal_status, encode_heal_start_success,
encode_heal_task_status, execute_after_heal_control_capability, heal_channel_response_items,
heal_channel_response_progress, heal_channel_response_summary, json_response, map_heal_response, map_root_heal_status,
merge_peer_heal_statuses, peer_topology_complete, query_peer_heal_status, reject_heal_admission,
should_handle_root_heal_directly, validate_heal_request_mode, validate_heal_target,
background_heal_runtime_state, build_heal_channel_request, build_replacement_recovery_status_response,
encode_background_heal_status, encode_heal_start_success, encode_heal_task_status, execute_after_heal_control_capability,
heal_channel_response_items, heal_channel_response_progress, heal_channel_response_summary, json_response,
map_heal_response, map_root_heal_status, merge_peer_heal_statuses, peer_topology_complete, query_peer_heal_status,
reject_heal_admission, should_handle_root_heal_directly, validate_heal_request_mode, validate_heal_target,
};
use crate::admin::storage_api::error::StorageError;
use crate::storage::rpc::node_service::heal::{NodeHealProgress, NodeHealStatusSnapshot};
@@ -1341,6 +1420,35 @@ mod tests {
assert!(executed.load(Ordering::SeqCst));
}
#[test]
fn replacement_recovery_status_response_never_claims_cluster_completion() {
let response = build_replacement_recovery_status_response(rustfs_heal::ReplacementRecoverySnapshot {
records: vec![rustfs_heal::ReplacementRecoveryRecord {
task_id: "11111111-1111-4111-8111-111111111111".to_string(),
state: rustfs_heal::ReplacementRecoveryState::Completed,
generation: Some("11111111-1111-4111-8111-111111111111".to_string()),
set_disk_id: Some("pool_0_set_0".to_string()),
target_slots: vec!["http://node-a:9000/mnt/disk1".to_string()],
reason: None,
verified_at: Some(42),
}],
definitive: true,
reason: None,
});
let value = serde_json::to_value(response).expect("replacement status response should serialize");
assert_eq!(value["contractVersion"], 1);
assert_eq!(value["path"], "/rustfs/admin/v4/heal/replacement-recovery");
assert_eq!(value["scope"], "local_survivor_disks");
assert_eq!(value["local"]["definitive"], true);
assert_eq!(value["local"]["records"][0]["state"], "completed");
assert_eq!(value["cluster"]["definitive"], false);
assert_eq!(
value["cluster"]["reason"],
"distributed replacement recovery status requires a peer capability RPC"
);
}
#[test]
fn test_reject_heal_admission_preserves_retry_semantics() {
for admission in [
rustfs/src/admin/handlers/mod.rs
+1
View File
@@ -119,6 +119,7 @@ mod tests {
let _tls_status_handler = tls_debug::TlsStatusHandler {};
let _heal_handler = heal::HealHandler {};
let _bg_heal_handler = heal::BackgroundHealStatusHandler {};
let _replacement_recovery_status_handler = heal::ReplacementRecoveryStatusHandler {};
let _replication_metrics_handler = replication::GetReplicationMetricsHandler {};
let _set_remote_target_handler = replication::SetRemoteTargetHandler {};
let _list_remote_target_handler = replication::ListRemoteTargetHandler {};
rustfs/src/admin/route_policy.rs
+6
View File
@@ -335,6 +335,12 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
admin(HttpMethod::Post, "/rustfs/admin/v3/heal/{bucket}", HEAL, RouteRiskLevel::High),
admin(HttpMethod::Post, "/rustfs/admin/v3/heal/{bucket}/{prefix}", HEAL, RouteRiskLevel::High),
admin(HttpMethod::Post, "/rustfs/admin/v3/background-heal/status", HEAL, RouteRiskLevel::High),
admin(
HttpMethod::Get,
"/rustfs/admin/v4/heal/replacement-recovery",
HEAL,
RouteRiskLevel::Sensitive,
),
admin(HttpMethod::Get, "/rustfs/admin/v3/tier", LIST_TIER, RouteRiskLevel::Sensitive),
admin(HttpMethod::Get, "/rustfs/admin/v3/tier-stats", LIST_TIER, RouteRiskLevel::Sensitive),
admin(HttpMethod::Get, "/rustfs/admin/v3/tier/{tier}", LIST_TIER, RouteRiskLevel::Sensitive),
rustfs/src/admin/route_registration_test.rs
+2
View File
@@ -188,6 +188,7 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
admin_route_sample(Method::POST, "/v3/heal/{bucket}", "/v3/heal/test-bucket"),
admin_route_sample(Method::POST, "/v3/heal/{bucket}/{prefix}", "/v3/heal/test-bucket/prefix"),
admin_route(Method::POST, "/v3/background-heal/status"),
admin_route(Method::GET, "/v4/heal/replacement-recovery"),
admin_route(Method::GET, "/v3/tier"),
admin_route(Method::GET, "/v3/tier-stats"),
admin_route_sample(Method::GET, "/v3/tier/{tier}", "/v3/tier/HOT"),
@@ -1224,6 +1225,7 @@ fn test_register_routes_cover_representative_admin_paths() {
assert_route(&router, Method::POST, &admin_path("/v3/heal/test-bucket"));
assert_route(&router, Method::POST, &admin_path("/v3/heal/test-bucket/prefix"));
assert_route(&router, Method::POST, &admin_path("/v3/background-heal/status"));
assert_route(&router, Method::GET, &admin_path("/v4/heal/replacement-recovery"));
assert_route(&router, Method::GET, &admin_path("/v3/tier"));
assert_route(&router, Method::GET, &admin_path("/v3/tier/HOT"));