diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index 198eafa6e..f1ba225d4 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -1916,6 +1916,23 @@ impl HealManager { } } + pub async fn active_progress_snapshot(&self) -> Option { + let active_heals = self.active_heals.lock().await; + if active_heals.is_empty() { + return None; + } + + let mut snapshot = HealProgress::default(); + for task in active_heals.values() { + let progress = task.get_progress().await; + snapshot.objects_scanned = snapshot.objects_scanned.saturating_add(progress.objects_scanned); + snapshot.objects_healed = snapshot.objects_healed.saturating_add(progress.objects_healed); + snapshot.objects_failed = snapshot.objects_failed.saturating_add(progress.objects_failed); + snapshot.bytes_processed = snapshot.bytes_processed.saturating_add(progress.bytes_processed); + } + Some(snapshot) + } + /// Start scheduler async fn start_scheduler(&self) -> Result<()> { let config = self.config.clone(); @@ -3796,6 +3813,43 @@ mod tests { assert_eq!(snapshot.active_by_source.scanner, 0); } + #[tokio::test] + async fn test_active_progress_snapshot_sums_active_task_progress() { + let storage: Arc = Arc::new(MockStorage); + let manager = HealManager::new(storage, None); + + let first = Arc::new(HealTask::from_request( + HealRequest::bucket("bucket-a".to_string()), + manager.storage.clone(), + )); + { + let mut progress = first.progress.write().await; + progress.update_progress(7, 3, 1, 4096); + } + + let second = Arc::new(HealTask::from_request( + HealRequest::bucket("bucket-b".to_string()), + manager.storage.clone(), + )); + { + let mut progress = second.progress.write().await; + progress.update_progress(11, 5, 2, 2048); + } + + manager.active_heals.lock().await.insert(first.id.clone(), first); + manager.active_heals.lock().await.insert(second.id.clone(), second); + + let progress = manager + .active_progress_snapshot() + .await + .expect("active progress should exist"); + + assert_eq!(progress.objects_scanned, 18); + assert_eq!(progress.objects_healed, 8); + assert_eq!(progress.objects_failed, 3); + assert_eq!(progress.bytes_processed, 6144); + } + #[tokio::test] async fn test_get_task_status_for_path_rejects_wrong_token_when_path_is_active() { let storage: Arc = Arc::new(MockStorage); diff --git a/crates/heal/src/lib.rs b/crates/heal/src/lib.rs index 65b3056b2..200a55760 100644 --- a/crates/heal/src/lib.rs +++ b/crates/heal/src/lib.rs @@ -18,7 +18,7 @@ pub mod heal; pub use error::{Error, Result}; pub use heal::{ HealManager, HealOperationsSnapshot, HealOptions, HealPriority, HealPriorityCounts, HealRequest, HealSourceCounts, HealType, - channel::HealChannelProcessor, + channel::HealChannelProcessor, progress::HealProgress, }; use rustfs_concurrency::WorkloadAdmissionSnapshotProvider; use std::sync::atomic::{AtomicU64, Ordering}; @@ -163,6 +163,14 @@ pub async fn current_heal_operations_snapshot() -> HealOperationsSnapshot { } } +pub async fn current_heal_progress_snapshot() -> Option { + if let Some(manager) = get_heal_manager() { + manager.active_progress_snapshot().await + } else { + None + } +} + fn usize_to_u64_saturated(value: usize) -> u64 { u64::try_from(value).unwrap_or(u64::MAX) } diff --git a/rustfs/src/admin/handlers/heal.rs b/rustfs/src/admin/handlers/heal.rs index 211fe51aa..9ea118e4f 100644 --- a/rustfs/src/admin/handlers/heal.rs +++ b/rustfs/src/admin/handlers/heal.rs @@ -206,6 +206,28 @@ struct BackgroundHealStatus<'a> { heal_queue_length: u64, heal_active_tasks: u64, heal_operations: rustfs_heal::HealOperationsSnapshot, + #[serde(skip_serializing_if = "Option::is_none")] + progress: Option, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct BackgroundHealProgress { + objects_scanned: u64, + objects_healed: u64, + objects_failed: u64, + bytes_processed: u64, +} + +impl From for BackgroundHealProgress { + fn from(progress: rustfs_heal::HealProgress) -> Self { + Self { + objects_scanned: progress.objects_scanned, + objects_healed: progress.objects_healed, + objects_failed: progress.objects_failed, + bytes_processed: progress.bytes_processed, + } + } } #[derive(Debug, Deserialize)] @@ -345,12 +367,14 @@ fn heal_channel_response_progress(response: &rustfs_common::heal_channel::HealCh fn encode_background_heal_status( info: &BackgroundHealInfo, heal_operations: rustfs_heal::HealOperationsSnapshot, + progress: Option, ) -> S3Result> { let status = BackgroundHealStatus { info, heal_queue_length: heal_operations.queue_length, heal_active_tasks: heal_operations.active_tasks, heal_operations, + progress, }; serde_json::to_vec(&status).map_err(|e| { warn!( @@ -727,7 +751,10 @@ impl Operation for BackgroundHealStatusHandler { let info = read_background_heal_info(store).await; let heal_operations = rustfs_heal::current_heal_operations_snapshot().await; - let body = encode_background_heal_status(&info, heal_operations)?; + let progress = rustfs_heal::current_heal_progress_snapshot() + .await + .map(BackgroundHealProgress::from); + let body = encode_background_heal_status(&info, heal_operations, progress)?; info!( event = EVENT_ADMIN_RESPONSE_EMITTED, component = LOG_COMPONENT_ADMIN_API, @@ -745,10 +772,10 @@ impl Operation for BackgroundHealStatusHandler { 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_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, 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; @@ -1190,7 +1217,7 @@ mod tests { ..Default::default() }; - let encoded = encode_background_heal_status(&info, operations).expect("background heal info should serialize"); + let encoded = encode_background_heal_status(&info, 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); @@ -1204,6 +1231,32 @@ mod tests { assert!(json["healOperations"]["queuedBySource"]["admin"].is_u64()); assert!(json["healOperations"]["queuedByPriority"]["low"].is_u64()); assert!(json["healOperations"]["queuedByPriority"]["high"].is_u64()); + assert!(json["progress"].is_null()); + } + + #[test] + fn test_encode_background_heal_status_includes_progress() { + let info = BackgroundHealInfo { + bitrot_start_time: None, + bitrot_start_cycle: 42, + current_scan_mode: HealScanMode::Deep, + }; + + let progress = BackgroundHealProgress { + objects_scanned: 7, + objects_healed: 3, + objects_failed: 1, + bytes_processed: 4096, + }; + + let encoded = encode_background_heal_status(&info, 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); + assert_eq!(json["progress"]["objectsHealed"], 3); + assert_eq!(json["progress"]["objectsFailed"], 1); + assert_eq!(json["progress"]["bytesProcessed"], 4096); } #[test]