From b387689f26927257862089579887175cab246331 Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Mon, 15 Jun 2026 22:28:53 +0800 Subject: [PATCH] feat(heal): expose scanner-aware operations status (#3483) Co-authored-by: Henry Guo --- crates/common/src/heal_channel.rs | 36 ++++ crates/heal/src/heal/channel.rs | 15 +- crates/heal/src/heal/manager.rs | 183 ++++++++++++++++++- crates/heal/src/heal/mod.rs | 2 +- crates/heal/src/heal/task.rs | 11 +- crates/heal/src/lib.rs | 25 ++- crates/scanner/src/scanner_folder.rs | 5 +- docs/operations/scanner-benchmark-runbook.md | 7 + docs/operations/scanner-runtime-controls.md | 27 +++ rustfs/src/admin/handlers/heal.rs | 34 +++- 10 files changed, 327 insertions(+), 18 deletions(-) diff --git a/crates/common/src/heal_channel.rs b/crates/common/src/heal_channel.rs index f19d6f9ff..c60fe8088 100644 --- a/crates/common/src/heal_channel.rs +++ b/crates/common/src/heal_channel.rs @@ -265,6 +265,27 @@ impl HealAdmissionResult { } } +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum HealRequestSource { + #[default] + Internal, + Admin, + Scanner, + AutoHeal, +} + +impl HealRequestSource { + pub const fn as_str(self) -> &'static str { + match self { + Self::Internal => "internal", + Self::Admin => "admin", + Self::Scanner => "scanner", + Self::AutoHeal => "auto_heal", + } + } +} + /// Heal channel command type #[derive(Debug)] pub enum HealChannelCommand { @@ -322,6 +343,8 @@ pub struct HealChannelRequest { pub dry_run: Option, /// Timeout in seconds (optional) pub timeout_seconds: Option, + /// Origin of the request for operational status and queue accounting + pub source: HealRequestSource, } /// Heal response from ahm to admin @@ -484,6 +507,7 @@ pub fn create_heal_request( recursive: None, dry_run: None, timeout_seconds: None, + source: HealRequestSource::Internal, disk: None, } } @@ -641,6 +665,7 @@ pub async fn send_heal_disk(set_disk_id: String, priority: Option, } +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct HealPriorityCounts { + pub low: u64, + pub normal: u64, + pub high: u64, + pub urgent: u64, +} + +impl HealPriorityCounts { + fn increment(&mut self, priority: HealPriority) { + match priority { + HealPriority::Low => self.low += 1, + HealPriority::Normal => self.normal += 1, + HealPriority::High => self.high += 1, + HealPriority::Urgent => self.urgent += 1, + } + } +} + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct HealSourceCounts { + pub scanner: u64, + pub admin: u64, + pub auto_heal: u64, + pub internal: u64, +} + +impl HealSourceCounts { + fn increment(&mut self, source: HealRequestSource) { + match source { + HealRequestSource::Scanner => self.scanner += 1, + HealRequestSource::Admin => self.admin += 1, + HealRequestSource::AutoHeal => self.auto_heal += 1, + HealRequestSource::Internal => self.internal += 1, + } + } +} + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct HealOperationsSnapshot { + pub queue_length: u64, + pub active_tasks: u64, + pub queued_by_priority: HealPriorityCounts, + pub active_by_priority: HealPriorityCounts, + pub queued_by_source: HealSourceCounts, + pub active_by_source: HealSourceCounts, +} + +fn usize_to_u64_saturated(value: usize) -> u64 { + u64::try_from(value).unwrap_or(u64::MAX) +} + impl PriorityHealQueue { fn new() -> Self { Self { @@ -212,6 +267,16 @@ impl PriorityHealQueue { stats } + fn operation_counts(&self) -> (HealPriorityCounts, HealSourceCounts) { + let mut priority = HealPriorityCounts::default(); + let mut source = HealSourceCounts::default(); + for item in &self.heap { + priority.increment(item.request.priority); + source.increment(item.request.source); + } + (priority, source) + } + #[cfg(test)] fn pop(&mut self) -> Option { self.heap.pop().map(|item| { @@ -1095,6 +1160,36 @@ impl HealManager { queue.len() } + pub async fn operations_snapshot(&self) -> HealOperationsSnapshot { + let (queue_length, queued_by_priority, queued_by_source) = { + let queue = self.heal_queue.lock().await; + let (priority, source) = queue.operation_counts(); + publish_heal_queue_length(&queue); + (usize_to_u64_saturated(queue.len()), priority, source) + }; + + let (active_tasks, active_by_priority, active_by_source) = { + let active_heals = self.active_heals.lock().await; + let mut priority = HealPriorityCounts::default(); + let mut source = HealSourceCounts::default(); + for task in active_heals.values() { + priority.increment(task.priority); + source.increment(task.source); + } + publish_active_heal_count(&active_heals); + (usize_to_u64_saturated(active_heals.len()), priority, source) + }; + + HealOperationsSnapshot { + queue_length, + active_tasks, + queued_by_priority, + active_by_priority, + queued_by_source, + active_by_source, + } + } + /// Start scheduler async fn start_scheduler(&self) -> Result<()> { let config = self.config.clone(); @@ -1304,7 +1399,7 @@ impl HealManager { } // enqueue erasure set heal request for this disk - let req = HealRequest::new( + let mut req = HealRequest::new( HealType::ErasureSet { buckets: buckets.clone(), set_disk_id: set_disk_id.clone(), @@ -1312,6 +1407,7 @@ impl HealManager { HealOptions::default(), HealPriority::Normal, ); + req.source = HealRequestSource::AutoHeal; let mut queue = heal_queue.lock().await; if matches!(queue.push(req), QueuePushOutcome::Accepted) { publish_heal_queue_length(&queue); @@ -1623,8 +1719,8 @@ fn can_schedule_request(request: &HealRequest, running_per_set: &HashMap = Arc::new(MockStorage); + let manager = HealManager::new(storage, None); + + let mut scanner_request = HealRequest::new( + HealType::Object { + bucket: "bucket-a".to_string(), + object: "object-a".to_string(), + version_id: None, + }, + HealOptions::default(), + HealPriority::Low, + ); + scanner_request.source = HealRequestSource::Scanner; + + let mut admin_request = HealRequest::bucket("bucket-b".to_string()); + admin_request.priority = HealPriority::High; + admin_request.source = HealRequestSource::Admin; + + let mut auto_request = HealRequest::new( + HealType::ErasureSet { + buckets: vec!["bucket-c".to_string()], + set_disk_id: "0-0".to_string(), + }, + HealOptions::default(), + HealPriority::Normal, + ); + auto_request.source = HealRequestSource::AutoHeal; + + manager + .submit_heal_request(scanner_request) + .await + .expect("scanner request should be accepted"); + manager + .submit_heal_request(admin_request) + .await + .expect("admin request should be accepted"); + manager + .submit_heal_request(auto_request) + .await + .expect("auto request should be accepted"); + + let snapshot = manager.operations_snapshot().await; + + assert_eq!(snapshot.queue_length, 3); + assert_eq!(snapshot.active_tasks, 0); + assert_eq!(snapshot.queued_by_priority.low, 1); + assert_eq!(snapshot.queued_by_priority.normal, 1); + assert_eq!(snapshot.queued_by_priority.high, 1); + assert_eq!(snapshot.queued_by_priority.urgent, 0); + assert_eq!(snapshot.queued_by_source.scanner, 1); + assert_eq!(snapshot.queued_by_source.admin, 1); + assert_eq!(snapshot.queued_by_source.auto_heal, 1); + assert_eq!(snapshot.queued_by_source.internal, 0); + } + + #[tokio::test] + async fn test_operations_snapshot_counts_active_by_source_and_priority() { + let storage: Arc = Arc::new(MockStorage); + let manager = HealManager::new(storage, None); + + let mut request = HealRequest::bucket("bucket-a".to_string()); + request.priority = HealPriority::High; + request.source = HealRequestSource::Admin; + let task = Arc::new(HealTask::from_request(request, manager.storage.clone())); + let task_id = task.id.clone(); + + manager.active_heals.lock().await.insert(task_id, task); + + let snapshot = manager.operations_snapshot().await; + + assert_eq!(snapshot.queue_length, 0); + assert_eq!(snapshot.active_tasks, 1); + assert_eq!(snapshot.active_by_priority.high, 1); + assert_eq!(snapshot.active_by_source.admin, 1); + assert_eq!(snapshot.active_by_source.scanner, 0); + } + #[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/heal/mod.rs b/crates/heal/src/heal/mod.rs index df03a3bda..59d7d25bf 100644 --- a/crates/heal/src/heal/mod.rs +++ b/crates/heal/src/heal/mod.rs @@ -23,6 +23,6 @@ pub mod task; pub mod utils; pub use erasure_healer::ErasureSetHealer; -pub use manager::HealManager; +pub use manager::{HealManager, HealOperationsSnapshot, HealPriorityCounts, HealSourceCounts}; pub use resume::{CheckpointManager, ResumeCheckpoint, ResumeManager, ResumeState, ResumeUtils}; pub use task::{HealOptions, HealPriority, HealRequest, HealTask, HealType}; diff --git a/crates/heal/src/heal/task.rs b/crates/heal/src/heal/task.rs index a8148b5df..7e7f8ca71 100644 --- a/crates/heal/src/heal/task.rs +++ b/crates/heal/src/heal/task.rs @@ -15,7 +15,7 @@ use crate::heal::{ErasureSetHealer, progress::HealProgress, storage::HealStorageAPI}; use crate::{Error, Result}; use metrics::{counter, histogram}; -use rustfs_common::heal_channel::{HealOpts, HealScanMode}; +use rustfs_common::heal_channel::{HealOpts, HealRequestSource, HealScanMode}; use rustfs_ecstore::{ data_usage::DATA_USAGE_CACHE_NAME, disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET}, @@ -169,6 +169,8 @@ pub struct HealRequest { pub options: HealOptions, /// Priority pub priority: HealPriority, + /// Origin of the request for operational status. + pub source: HealRequestSource, /// Whether this request should bypass queue admission dedup/full policies. pub force_start: bool, /// Created time @@ -185,6 +187,7 @@ impl HealRequest { heal_type, options, priority, + source: HealRequestSource::Internal, force_start: false, created_at: now, enqueued_at: now, @@ -232,6 +235,10 @@ pub struct HealTask { pub heal_type: HealType, /// Heal options pub options: HealOptions, + /// Priority inherited from the request + pub priority: HealPriority, + /// Origin inherited from the request + pub source: HealRequestSource, /// Task status pub status: Arc>, /// Progress tracking @@ -260,6 +267,8 @@ impl HealTask { id: request.id, heal_type: request.heal_type, options: request.options, + priority: request.priority, + source: request.source, status: Arc::new(RwLock::new(HealTaskStatus::Pending)), progress: Arc::new(RwLock::new(HealProgress::new())), result_items: Arc::new(RwLock::new(Vec::new())), diff --git a/crates/heal/src/lib.rs b/crates/heal/src/lib.rs index cd45b6236..9297f3fca 100644 --- a/crates/heal/src/lib.rs +++ b/crates/heal/src/lib.rs @@ -16,7 +16,10 @@ mod error; pub mod heal; pub use error::{Error, Result}; -pub use heal::{HealManager, HealOptions, HealPriority, HealRequest, HealType, channel::HealChannelProcessor}; +pub use heal::{ + HealManager, HealOperationsSnapshot, HealOptions, HealPriority, HealPriorityCounts, HealRequest, HealSourceCounts, HealType, + channel::HealChannelProcessor, +}; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, OnceLock}; use tokio_util::sync::CancellationToken; @@ -138,10 +141,26 @@ pub fn current_heal_queue_length() -> u64 { GLOBAL_HEAL_QUEUE_LENGTH.load(Ordering::Relaxed) } +pub async fn current_heal_operations_snapshot() -> HealOperationsSnapshot { + if let Some(manager) = get_heal_manager() { + manager.operations_snapshot().await + } else { + HealOperationsSnapshot { + queue_length: current_heal_queue_length(), + active_tasks: current_heal_active_tasks(), + ..Default::default() + } + } +} + +fn usize_to_u64_saturated(value: usize) -> u64 { + u64::try_from(value).unwrap_or(u64::MAX) +} + pub(crate) fn set_heal_active_tasks(count: usize) { - GLOBAL_HEAL_ACTIVE_TASKS.store(count as u64, Ordering::Relaxed); + GLOBAL_HEAL_ACTIVE_TASKS.store(usize_to_u64_saturated(count), Ordering::Relaxed); } pub(crate) fn set_heal_queue_length(count: usize) { - GLOBAL_HEAL_QUEUE_LENGTH.store(count as u64, Ordering::Relaxed); + GLOBAL_HEAL_QUEUE_LENGTH.store(usize_to_u64_saturated(count), Ordering::Relaxed); } diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index 3140a21ae..39b631812 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -32,7 +32,7 @@ use crate::scanner_io::{SCANNER_SKIP_FILE_ERROR, ScannerIODisk as _}; use crate::sleeper::DynamicSleeper; use metrics::{counter, describe_counter}; use rustfs_common::heal_channel::{ - HEAL_DELETE_DANGLING, HealAdmissionResult, HealChannelPriority, HealChannelRequest, HealScanMode, + HEAL_DELETE_DANGLING, HealAdmissionResult, HealChannelPriority, HealChannelRequest, HealRequestSource, HealScanMode, send_heal_request_with_admission, }; use rustfs_common::metrics::{ @@ -448,6 +448,7 @@ fn build_bucket_heal_request(bucket: String, priority: HealChannelPriority) -> H HealChannelRequest { bucket, priority, + source: HealRequestSource::Scanner, ..Default::default() } } @@ -466,6 +467,7 @@ fn build_object_heal_request( priority, scan_mode: Some(scan_mode), remove_corrupted: Some(HEAL_DELETE_DANGLING), + source: HealRequestSource::Scanner, ..Default::default() } } @@ -3078,6 +3080,7 @@ mod tests { assert!(request.object_version_id.is_none()); assert_eq!(request.scan_mode, Some(HealScanMode::Deep)); assert_eq!(request.priority, HealChannelPriority::Low); + assert_eq!(request.source, HealRequestSource::Scanner); assert_eq!(request.remove_corrupted, Some(HEAL_DELETE_DANGLING)); } diff --git a/docs/operations/scanner-benchmark-runbook.md b/docs/operations/scanner-benchmark-runbook.md index c339ed02b..61c570749 100644 --- a/docs/operations/scanner-benchmark-runbook.md +++ b/docs/operations/scanner-benchmark-runbook.md @@ -309,6 +309,13 @@ Compare these fields between baseline and tuned runs: Do not use a single CPU spike as the conclusion. Compare average and p95 CPU over the same observation window. +For heal or bitrot pressure investigations, also capture +`/v3/background-heal/status` and compare `healOperations.queueLength`, +`healOperations.activeTasks`, `healOperations.queuedBySource`, +`healOperations.activeBySource`, `healOperations.queuedByPriority`, and +`healOperations.activeByPriority`. These fields distinguish scanner-submitted +low-priority work from manual admin heal and auto-heal work. + ## Interpreting Results A useful tuning result has all of these properties: diff --git a/docs/operations/scanner-runtime-controls.md b/docs/operations/scanner-runtime-controls.md index 9b822ee17..1d63ddf27 100644 --- a/docs/operations/scanner-runtime-controls.md +++ b/docs/operations/scanner-runtime-controls.md @@ -148,6 +148,33 @@ Use these counters to decide whether scan progress is limited by scanner pacing or by a downstream subsystem such as lifecycle transition, replication repair, or heal admission. +## Reading Heal Operations + +The background heal status route is: + +```text +POST /v3/background-heal/status +``` + +It reports scanner-driven bitrot state together with heal queue execution +state. `healQueueLength` and `healActiveTasks` keep the legacy totals. +`healOperations` adds the same totals split by request source and priority: + +| Field | Meaning | +|---|---| +| `queueLength` | Total queued heal requests. | +| `activeTasks` | Total running heal tasks. | +| `queuedBySource` | Queued requests split into `scanner`, `admin`, `autoHeal`, and `internal`. | +| `activeBySource` | Running tasks split into `scanner`, `admin`, `autoHeal`, and `internal`. | +| `queuedByPriority` | Queued requests split into `low`, `normal`, `high`, and `urgent`. | +| `activeByPriority` | Running tasks split into `low`, `normal`, `high`, and `urgent`. | + +Use this route when `metrics.source_work` shows `heal` or `bitrot` queued or +missed work. Scanner-originated object checks should appear under +`scanner/low` for opportunistic work, while manual admin heal should appear +under `admin/high`. If scanner work grows but admin work remains blocked, treat +that as heal queue pressure rather than scanner pacing pressure. + ## Reading Replication Repair `metrics.replication_repair`, `metrics.current_cycle_replication_repair`, and diff --git a/rustfs/src/admin/handlers/heal.rs b/rustfs/src/admin/handlers/heal.rs index bb2a371ad..c47954845 100644 --- a/rustfs/src/admin/handlers/heal.rs +++ b/rustfs/src/admin/handlers/heal.rs @@ -22,7 +22,7 @@ use bytes::Bytes; use http::{HeaderMap, HeaderValue, Uri}; use hyper::{Method, StatusCode}; use matchit::Params; -use rustfs_common::heal_channel::{HealChannelPriority, HealChannelRequest, HealOpts}; +use rustfs_common::heal_channel::{HealChannelPriority, HealChannelRequest, HealOpts, HealRequestSource}; use rustfs_config::MAX_HEAL_REQUEST_SIZE; use rustfs_ecstore::bucket::utils::is_valid_object_prefix; use rustfs_ecstore::store_api::HealOperations; @@ -186,6 +186,7 @@ struct BackgroundHealStatus<'a> { info: &'a BackgroundHealInfo, heal_queue_length: u64, heal_active_tasks: u64, + heal_operations: rustfs_heal::HealOperationsSnapshot, } #[derive(Debug, Deserialize)] @@ -265,6 +266,7 @@ fn build_heal_channel_request(hip: &HealInitParams) -> HealChannelRequest { heal_request.update_parity = Some(hip.hs.update_parity); heal_request.recursive = Some(hip.hs.recursive); heal_request.dry_run = Some(hip.hs.dry_run); + heal_request.source = HealRequestSource::Admin; heal_request } @@ -301,11 +303,15 @@ fn heal_channel_response_items( heal_channel_response_status(response).1 } -fn encode_background_heal_status(info: &BackgroundHealInfo) -> S3Result> { +fn encode_background_heal_status( + info: &BackgroundHealInfo, + heal_operations: rustfs_heal::HealOperationsSnapshot, +) -> S3Result> { let status = BackgroundHealStatus { info, - heal_queue_length: rustfs_heal::current_heal_queue_length(), - heal_active_tasks: rustfs_heal::current_heal_active_tasks(), + heal_queue_length: heal_operations.queue_length, + heal_active_tasks: heal_operations.active_tasks, + heal_operations, }; serde_json::to_vec(&status).map_err(|e| { warn!( @@ -669,7 +675,8 @@ impl Operation for BackgroundHealStatusHandler { }; let info = read_background_heal_info(store).await; - let body = encode_background_heal_status(&info)?; + let heal_operations = rustfs_heal::current_heal_operations_snapshot().await; + let body = encode_background_heal_status(&info, heal_operations)?; info!( event = EVENT_ADMIN_RESPONSE_EMITTED, component = LOG_COMPONENT_ADMIN_API, @@ -695,7 +702,7 @@ mod tests { use http::StatusCode; use http::Uri; use matchit::Router; - use rustfs_common::heal_channel::{HealChannelPriority, HealOpts, HealScanMode}; + use rustfs_common::heal_channel::{HealChannelPriority, HealOpts, HealRequestSource, HealScanMode}; use rustfs_ecstore::error::StorageError; use rustfs_scanner::scanner::BackgroundHealInfo; use s3s::{ @@ -979,7 +986,13 @@ mod tests { current_scan_mode: HealScanMode::Deep, }; - let encoded = encode_background_heal_status(&info).expect("background heal info should serialize"); + let operations = rustfs_heal::HealOperationsSnapshot { + queue_length: 3, + active_tasks: 2, + ..Default::default() + }; + + let encoded = encode_background_heal_status(&info, operations).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); @@ -987,6 +1000,12 @@ mod tests { assert!(json["bitrotStartTime"].is_null()); assert!(json["healQueueLength"].is_u64()); assert!(json["healActiveTasks"].is_u64()); + assert_eq!(json["healOperations"]["queueLength"], json["healQueueLength"]); + assert_eq!(json["healOperations"]["activeTasks"], json["healActiveTasks"]); + assert!(json["healOperations"]["queuedBySource"]["scanner"].is_u64()); + assert!(json["healOperations"]["queuedBySource"]["admin"].is_u64()); + assert!(json["healOperations"]["queuedByPriority"]["low"].is_u64()); + assert!(json["healOperations"]["queuedByPriority"]["high"].is_u64()); } #[test] @@ -1041,6 +1060,7 @@ mod tests { assert_eq!(request.bucket, "bucket-a"); assert_eq!(request.object_prefix.as_deref(), Some("prefix-a")); assert_eq!(request.priority, HealChannelPriority::High); + assert_eq!(request.source, HealRequestSource::Admin); assert!(request.force_start); assert_eq!(request.pool_index, Some(1)); assert_eq!(request.set_index, Some(2));