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

fix(admin): report heal runtime state (#4786)

Browse Source
This commit is contained in:
Zhengchao An
2026-07-13 11:02:13 +08:00
committed by GitHub
parent 2e85709634
commit 535d672b1f
3 changed files with 78 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/heal/src/lib.rs
+4
View File
@@ -143,6 +143,10 @@ pub fn get_heal_channel_processor() -> Option<&'static Arc<tokio::sync::Mutex<He
GLOBAL_HEAL_CHANNEL_PROCESSOR.get()
}
pub fn heal_runtime_initialized() -> bool {
get_heal_manager().is_some() && get_heal_channel_processor().is_some()
}
pub fn current_heal_active_tasks() -> u64 {
GLOBAL_HEAL_ACTIVE_TASKS.load(Ordering::Relaxed)
}
rustfs/src/admin/handlers/heal.rs
+67 -8
View File
@@ -21,6 +21,7 @@ use crate::admin::storage_api::bucket::utils::is_valid_object_prefix;
use crate::admin::storage_api::contract::heal::HealOperations as _;
use crate::server::ADMIN_PREFIX;
use crate::server::RemoteAddr;
use crate::startup_background::{heal_enabled_from_env, scanner_enabled_from_env};
use bytes::Bytes;
use http::{HeaderMap, HeaderValue, Uri};
use hyper::{Method, StatusCode};
@@ -203,6 +204,7 @@ struct HealTaskStatus {
struct BackgroundHealStatus<'a> {
#[serde(flatten)]
info: &'a BackgroundHealInfo,
state: HealRuntimeState,
heal_queue_length: u64,
heal_active_tasks: u64,
heal_operations: rustfs_heal::HealOperationsSnapshot,
@@ -210,6 +212,33 @@ struct BackgroundHealStatus<'a> {
progress: Option<BackgroundHealProgress>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
enum HealRuntimeState {
Disabled,
Uninitialized,
Idle,
Active,
}
fn background_heal_runtime_state(
services_enabled: bool,
initialized: bool,
operations: &rustfs_heal::HealOperationsSnapshot,
) -> HealRuntimeState {
if !services_enabled {
return HealRuntimeState::Disabled;
}
if !initialized {
return HealRuntimeState::Uninitialized;
}
if operations.queue_length > 0 || operations.active_tasks > 0 {
HealRuntimeState::Active
} else {
HealRuntimeState::Idle
}
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct BackgroundHealProgress {
@@ -366,11 +395,13 @@ fn heal_channel_response_progress(response: &rustfs_common::heal_channel::HealCh
fn encode_background_heal_status(
info: &BackgroundHealInfo,
state: HealRuntimeState,
heal_operations: rustfs_heal::HealOperationsSnapshot,
progress: Option<BackgroundHealProgress>,
) -> S3Result<Vec<u8>> {
let status = BackgroundHealStatus {
info,
state,
heal_queue_length: heal_operations.queue_length,
heal_active_tasks: heal_operations.active_tasks,
heal_operations,
@@ -751,10 +782,15 @@ impl Operation for BackgroundHealStatusHandler {
let info = read_background_heal_info(store).await;
let heal_operations = rustfs_heal::current_heal_operations_snapshot().await;
let state = background_heal_runtime_state(
scanner_enabled_from_env() || heal_enabled_from_env(),
rustfs_heal::heal_runtime_initialized(),
&heal_operations,
);
let progress = rustfs_heal::current_heal_progress_snapshot()
.await
.map(BackgroundHealProgress::from);
let body = encode_background_heal_status(&info, heal_operations, progress)?;
let body = encode_background_heal_status(&info, state, heal_operations, progress)?;
info!(
event = EVENT_ADMIN_RESPONSE_EMITTED,
component = LOG_COMPONENT_ADMIN_API,
@@ -772,10 +808,11 @@ impl Operation for BackgroundHealStatusHandler {
mod tests {
use super::extract_heal_init_params;
use super::{
BackgroundHealProgress, 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_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,
BackgroundHealProgress, HealInitParams, HealResp, HealRuntimeState, background_heal_runtime_state,
build_heal_channel_request, encode_background_heal_status, encode_heal_start_success, 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 crate::admin::storage_api::error::StorageError;
use bytes::Bytes;
@@ -1225,7 +1262,8 @@ mod tests {
..Default::default()
};
let encoded = encode_background_heal_status(&info, operations, None).expect("background heal info should serialize");
let encoded = encode_background_heal_status(&info, HealRuntimeState::Active, operations, None)
.expect("background heal info should serialize");
let json: serde_json::Value = serde_json::from_slice(&encoded).expect("json should deserialize");
assert_eq!(json["bitrotStartCycle"], 42);
@@ -1239,6 +1277,7 @@ mod tests {
assert!(json["healOperations"]["queuedBySource"]["admin"].is_u64());
assert!(json["healOperations"]["queuedByPriority"]["low"].is_u64());
assert!(json["healOperations"]["queuedByPriority"]["high"].is_u64());
assert_eq!(json["state"], "active");
assert!(json["progress"].is_null());
}
@@ -1257,8 +1296,13 @@ mod tests {
bytes_processed: 4096,
};
let encoded = encode_background_heal_status(&info, rustfs_heal::HealOperationsSnapshot::default(), Some(progress))
.expect("background heal info should serialize");
let encoded = encode_background_heal_status(
&info,
HealRuntimeState::Idle,
rustfs_heal::HealOperationsSnapshot::default(),
Some(progress),
)
.expect("background heal info should serialize");
let json: serde_json::Value = serde_json::from_slice(&encoded).expect("json should deserialize");
assert_eq!(json["progress"]["objectsScanned"], 7);
@@ -1267,6 +1311,21 @@ mod tests {
assert_eq!(json["progress"]["bytesProcessed"], 4096);
}
#[test]
fn test_background_heal_runtime_state_distinguishes_disabled_and_uninitialized() {
let operations = rustfs_heal::HealOperationsSnapshot::default();
assert_eq!(background_heal_runtime_state(false, false, &operations), HealRuntimeState::Disabled);
assert_eq!(background_heal_runtime_state(true, false, &operations), HealRuntimeState::Uninitialized);
assert_eq!(background_heal_runtime_state(true, true, &operations), HealRuntimeState::Idle);
let active = rustfs_heal::HealOperationsSnapshot {
active_tasks: 1,
..operations
};
assert_eq!(background_heal_runtime_state(true, true, &active), HealRuntimeState::Active);
}
#[test]
fn test_encode_heal_start_success_uses_client_wire_shape() {
let encoded = encode_heal_start_success("token-1".to_string(), "127.0.0.1:9000".to_string())
rustfs/src/startup_background.rs
+7 -3
View File
@@ -24,8 +24,8 @@ use tracing::{debug, info};
pub(crate) const ENV_SCANNER_ENABLED: &str = "RUSTFS_SCANNER_ENABLED";
pub(crate) const ENV_SCANNER_ENABLED_DEPRECATED: &str = "RUSTFS_ENABLE_SCANNER";
const ENV_HEAL_ENABLED: &str = "RUSTFS_HEAL_ENABLED";
const ENV_HEAL_ENABLED_DEPRECATED: &str = "RUSTFS_ENABLE_HEAL";
pub(crate) const ENV_HEAL_ENABLED: &str = "RUSTFS_HEAL_ENABLED";
pub(crate) const ENV_HEAL_ENABLED_DEPRECATED: &str = "RUSTFS_ENABLE_HEAL";
const LOG_COMPONENT_MAIN: &str = "main";
const LOG_SUBSYSTEM_STARTUP: &str = "startup";
const EVENT_BACKGROUND_SERVICES_CONFIGURED: &str = "background_services_configured";
@@ -34,11 +34,15 @@ pub(crate) fn scanner_enabled_from_env() -> bool {
get_env_bool_with_aliases(ENV_SCANNER_ENABLED, &[ENV_SCANNER_ENABLED_DEPRECATED], true)
}
pub(crate) fn heal_enabled_from_env() -> bool {
get_env_bool_with_aliases(ENV_HEAL_ENABLED, &[ENV_HEAL_ENABLED_DEPRECATED], true)
}
pub(crate) async fn init_background_service_runtime(store: Arc<ECStore>) -> Result<bool> {
let _ = create_ahm_services_cancel_token();
let enable_scanner = scanner_enabled_from_env();
let enable_heal = get_env_bool_with_aliases(ENV_HEAL_ENABLED, &[ENV_HEAL_ENABLED_DEPRECATED], true);
let enable_heal = heal_enabled_from_env();
info!(
target: "rustfs::main::run",