From 7dc15b8ca6200dc0e616a9e6002a69de1f683336 Mon Sep 17 00:00:00 2001 From: houseme Date: Tue, 8 Sep 2026 07:21:26 +0800 Subject: [PATCH] feat(heal): expose admission observability (#7418) * feat(heal): expose admission observability Track heal admission outcomes and bounded lock-phase latency through the existing operations snapshot so distributed E2E gates can assert duplicate, forceStart, and displacement behavior without relying on logs. Co-Authored-By: heihutu Co-Authored-By: zhi22915 * fix(error): merge equivalent api message branches Combine the MaxVersionsExceeded and internal IO message branches so Clippy no longer flags identical if blocks while preserving the existing response messages. Co-Authored-By: heihutu Co-Authored-By: zhi22915 --------- Co-authored-by: zhi22915 Co-authored-by: Zhengchao An --- crates/heal/src/heal/manager.rs | 153 +++++++++++++++++++- crates/heal/src/heal/manager/tests.rs | 82 +++++++++++ crates/heal/src/heal/mod.rs | 2 +- crates/heal/src/lib.rs | 3 +- rustfs/src/admin/handlers/heal.rs | 30 ++++ rustfs/src/storage/rpc/node_service/heal.rs | 1 + 6 files changed, 264 insertions(+), 7 deletions(-) diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index 74ec6d546..b54049a14 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -20,7 +20,7 @@ use crate::heal::{ task::{HealOptions, HealPriority, HealRequest, HealTask, HealTaskStatus, HealType, demote_to_debug_when}, }; use crate::{Error, Result}; -use metrics::{counter, gauge}; +use metrics::{counter, gauge, histogram}; use rustfs_concurrency::WorkloadAdmissionSnapshotProvider; use rustfs_concurrency::workload::{ForegroundPressure, foreground_pressure}; #[cfg(test)] @@ -34,7 +34,7 @@ use std::sync::LazyLock; use std::{ collections::{BinaryHeap, HashMap, HashSet}, sync::{Arc, Mutex as StdMutex, MutexGuard as StdMutexGuard}, - time::{Duration, SystemTime}, + time::{Duration, Instant, SystemTime}, }; use tokio::{ sync::{Mutex, Notify, RwLock}, @@ -181,6 +181,13 @@ fn lock_displaced_terminals( } } +fn lock_admission_telemetry(registry: &StdMutex) -> StdMutexGuard<'_, HealAdmissionTelemetry> { + match registry.lock() { + Ok(guard) => guard, + Err(poisoned) => poisoned.into_inner(), + } +} + fn record_displaced_terminal( registry: &StdMutex>>, request: &HealRequest, @@ -384,6 +391,61 @@ impl HealSourceCounts { } } +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Deserialize, serde::Serialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct HealAdmissionTelemetry { + pub accepted: u64, + pub merged: u64, + pub full: u64, + pub dropped: u64, + pub duplicate: u64, + pub overlap_rejected: u64, + pub displaced: u64, + pub force_start: u64, + pub max_start_duration_micros: u64, + pub max_lock_phase_micros: u64, +} + +impl HealAdmissionTelemetry { + fn record(&mut self, observation: HealAdmissionObservation) { + match observation.result { + HealAdmissionResult::Accepted => self.accepted = self.accepted.saturating_add(1), + HealAdmissionResult::Merged => self.merged = self.merged.saturating_add(1), + HealAdmissionResult::Full => self.full = self.full.saturating_add(1), + HealAdmissionResult::Dropped(_) => self.dropped = self.dropped.saturating_add(1), + } + if observation.context == "duplicate" { + self.duplicate = self.duplicate.saturating_add(1); + } + if observation.context == "overlap_rejected" { + self.overlap_rejected = self.overlap_rejected.saturating_add(1); + } + if observation.displaced { + self.displaced = self.displaced.saturating_add(1); + } + if observation.force_start { + self.force_start = self.force_start.saturating_add(1); + } + self.max_start_duration_micros = self + .max_start_duration_micros + .max(duration_micros_saturated(observation.start_duration)); + self.max_lock_phase_micros = self + .max_lock_phase_micros + .max(duration_micros_saturated(observation.lock_phase)); + } +} + +#[derive(Debug, Clone, Copy)] +struct HealAdmissionObservation { + source: HealRequestSource, + result: HealAdmissionResult, + context: &'static str, + force_start: bool, + displaced: bool, + start_duration: Duration, + lock_phase: Duration, +} + #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Deserialize, serde::Serialize)] #[serde(rename_all = "camelCase", deny_unknown_fields)] pub struct HealOperationsSnapshot { @@ -396,12 +458,18 @@ pub struct HealOperationsSnapshot { pub queued_by_source: HealSourceCounts, pub active_by_source: HealSourceCounts, pub retrying_by_source: HealSourceCounts, + #[serde(default)] + pub admission: HealAdmissionTelemetry, } fn usize_to_u64_saturated(value: usize) -> u64 { u64::try_from(value).unwrap_or(u64::MAX) } +fn duration_micros_saturated(duration: Duration) -> u64 { + u64::try_from(duration.as_micros()).unwrap_or(u64::MAX) +} + fn heal_type_matches_path(heal_type: &HealType, heal_path: &str) -> bool { let heal_path = heal_path.trim_matches('/'); if heal_path.is_empty() || heal_path == LEGACY_ROOT_HEAL_PATH { @@ -764,6 +832,9 @@ pub struct HealManager { notify: Arc, /// Optional runtime workload snapshot provider used to protect foreground data-plane work. workload_provider: Option, + /// Bounded, low-cardinality admission telemetry exposed through the + /// existing operations snapshot for cluster E2E assertions. + admission_telemetry: Arc>, } /// Where a task-id lookup resolved. The variants carry the resolved state @@ -919,6 +990,33 @@ impl HealManager { .increment(1); } + fn record_admission_observation(&self, observation: HealAdmissionObservation) { + let result = observation.result.result_label().to_string(); + let reason = observation.result.reason_label().to_string(); + let source = observation.source.as_str().to_string(); + let context = observation.context.to_string(); + let force_start = observation.force_start.to_string(); + histogram!( + "rustfs_heal_admission_start_duration_seconds", + "source" => source.clone(), + "result" => result.clone(), + "reason" => reason.clone(), + "context" => context.clone(), + "force_start" => force_start.clone() + ) + .record(observation.start_duration.as_secs_f64()); + histogram!( + "rustfs_heal_admission_lock_phase_seconds", + "source" => source, + "result" => result, + "reason" => reason, + "context" => context, + "force_start" => force_start + ) + .record(observation.lock_phase.as_secs_f64()); + lock_admission_telemetry(&self.admission_telemetry).record(observation); + } + fn remove_mrf_repair_notice_targets_for_task(&self, task_id: &str) { let targets = lock_mrf_repair_notice_targets(&self.mrf_repair_notice_targets).remove(task_id); if let Some(targets) = targets { @@ -1265,6 +1363,7 @@ impl HealManager { statistics: Arc::new(RwLock::new(HealStatistics::new())), notify: Arc::new(Notify::new()), workload_provider, + admission_telemetry: Arc::new(StdMutex::new(HealAdmissionTelemetry::default())), } } @@ -1455,6 +1554,9 @@ impl HealManager { preserve_alias: bool, mrf_notice_target: Option, ) -> Result { + let admission_start = Instant::now(); + let source = request.source; + let force_start = request.force_start; // HS-06 forceStart semantics (admin only): MinIO stops the old task // first and then starts the new one. Cancel any active admin task // overlapping this request's path before entering admission, so the @@ -1505,6 +1607,7 @@ impl HealManager { // Match the scheduler's active -> queue order and keep retry ownership // in the same atomic view. Otherwise queue -> active and // active -> retrying transitions can slip between duplicate checks. + let lock_phase_start = Instant::now(); let active_heals = self.active_heals.lock().await; #[cfg(test)] pause_duplicate_admission_after_active_lock(&request.id).await; @@ -1539,7 +1642,17 @@ impl HealManager { drop(retrying_heals); drop(queue); drop(active_heals); + let lock_phase = lock_phase_start.elapsed(); Self::record_admission_metric(request.source, admission, "duplicate"); + self.record_admission_observation(HealAdmissionObservation { + source, + result: admission, + context: "duplicate", + force_start, + displaced: false, + start_duration: admission_start.elapsed(), + lock_phase, + }); match admission { HealAdmissionResult::Merged => { @@ -1618,7 +1731,17 @@ impl HealManager { drop(retrying_heals); drop(queue); drop(active_heals); + let lock_phase = lock_phase_start.elapsed(); Self::record_admission_metric(request.source, HealAdmissionResult::Dropped(reason), "overlap_rejected"); + self.record_admission_observation(HealAdmissionObservation { + source, + result: HealAdmissionResult::Dropped(reason), + context: "overlap_rejected", + force_start, + displaced: false, + start_duration: admission_start.elapsed(), + lock_phase, + }); warn!( target: "rustfs::heal::manager", event = EVENT_HEAL_QUEUE_ADMISSION, @@ -1663,6 +1786,8 @@ impl HealManager { drop(retrying_heals); drop(queue); drop(active_heals); + let lock_phase = lock_phase_start.elapsed(); + let displaced = displaced_terminal.is_some(); if let (Some(displaced_task_id), Some(displaced_terminal)) = (displaced_task_id, displaced_terminal) { // The queue has already removed the displaced request, so the @@ -1676,6 +1801,16 @@ impl HealManager { self.notify.notify_one(); } + self.record_admission_observation(HealAdmissionObservation { + source, + result: admission, + context: "submit", + force_start, + displaced, + start_duration: admission_start.elapsed(), + lock_phase, + }); + Ok(HealAdmissionReceipt { result: admission, task_id, @@ -2111,17 +2246,25 @@ impl HealManager { } publish_active_heal_count(&active_heals); publish_heal_queue_length(&queue); + let queue_length = usize_to_u64_saturated(queue.len()); + let active_tasks = usize_to_u64_saturated(active_heals.len()); + let retrying_tasks = usize_to_u64_saturated(retrying_heals.len()); + drop(retrying_heals); + drop(queue); + drop(active_heals); + let admission = *lock_admission_telemetry(&self.admission_telemetry); HealOperationsSnapshot { - queue_length: usize_to_u64_saturated(queue.len()), - active_tasks: usize_to_u64_saturated(active_heals.len()), - retrying_tasks: usize_to_u64_saturated(retrying_heals.len()), + queue_length, + active_tasks, + retrying_tasks, queued_by_priority, active_by_priority, retrying_by_priority, queued_by_source, active_by_source, retrying_by_source, + admission, } } diff --git a/crates/heal/src/heal/manager/tests.rs b/crates/heal/src/heal/manager/tests.rs index 12495bc2c..c1431472a 100644 --- a/crates/heal/src/heal/manager/tests.rs +++ b/crates/heal/src/heal/manager/tests.rs @@ -2792,6 +2792,88 @@ async fn admin_force_start_cancels_overlapping_active_task_first() { ); } +#[tokio::test] +async fn admission_snapshot_tracks_start_duplicate_force_start_and_displacement() { + let storage: Arc = Arc::new(MockStorage); + let manager = Arc::new(HealManager::new( + storage, + Some(HealConfig { + queue_size: 1, + ..Default::default() + }), + )); + + let mut paused = admin_prefix_request("bucket-a", "logs/"); + paused.priority = HealPriority::Low; + let hook = Arc::new(DuplicateAdmissionTestHook { + request_id: paused.id.clone(), + active_lock_reached: Notify::new(), + active_lock_release: Notify::new(), + }); + *DUPLICATE_ADMISSION_TEST_HOOK.lock().await = Some(hook.clone()); + + let submit_manager = Arc::clone(&manager); + let mut paused_submission = tokio::spawn(async move { submit_manager.submit_heal_request(paused).await }); + tokio::time::timeout(Duration::from_secs(1), hook.active_lock_reached.notified()) + .await + .expect("admission should reach the test-only lock phase hook"); + assert!( + tokio::time::timeout(Duration::from_millis(10), &mut paused_submission) + .await + .is_err(), + "admission must wait while the lock-phase hook is held" + ); + hook.active_lock_release.notify_one(); + assert_eq!( + paused_submission + .await + .expect("paused admission task should join") + .expect("paused admission should succeed"), + HealAdmissionResult::Accepted + ); + *DUPLICATE_ADMISSION_TEST_HOOK.lock().await = None; + + let duplicate = admin_prefix_request("bucket-a", "logs/"); + let duplicate_receipt = manager + .submit_heal_request_with_receipt(duplicate) + .await + .expect("duplicate admission should return a canonical receipt"); + assert_eq!(duplicate_receipt.result, HealAdmissionResult::Merged); + + let mut high = admin_prefix_request("bucket-b", "logs/"); + high.priority = HealPriority::High; + assert_eq!( + manager + .submit_heal_request(high) + .await + .expect("higher priority admin request should displace queued low-priority work"), + HealAdmissionResult::Accepted + ); + + let mut forced = admin_prefix_request("bucket-c", "logs/"); + forced.force_start = true; + assert_eq!( + manager + .submit_heal_request(forced) + .await + .expect("forceStart should keep explicit admission semantics"), + HealAdmissionResult::Accepted + ); + + let admission = manager.operations_snapshot().await.admission; + assert_eq!(admission.accepted, 3); + assert_eq!(admission.merged, 1); + assert_eq!(admission.full, 0); + assert_eq!(admission.dropped, 0); + assert_eq!(admission.duplicate, 1); + assert_eq!(admission.displaced, 1); + assert_eq!(admission.force_start, 1); + assert!( + admission.max_lock_phase_micros > 0, + "snapshot should expose a measurable queue/admission lock phase for p95-style external aggregation" + ); +} + #[tokio::test] async fn test_operations_snapshot_counts_active_by_source_and_priority() { let storage: Arc = Arc::new(MockStorage); diff --git a/crates/heal/src/heal/mod.rs b/crates/heal/src/heal/mod.rs index ce2058086..e7885fb91 100644 --- a/crates/heal/src/heal/mod.rs +++ b/crates/heal/src/heal/mod.rs @@ -34,7 +34,7 @@ use storage_api::owner::{ }; pub use erasure_healer::ErasureSetHealer; -pub use manager::{HealManager, HealOperationsSnapshot, HealPriorityCounts, HealSourceCounts}; +pub use manager::{HealAdmissionTelemetry, 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/lib.rs b/crates/heal/src/lib.rs index 762c6c0c6..733701e45 100644 --- a/crates/heal/src/lib.rs +++ b/crates/heal/src/lib.rs @@ -19,7 +19,8 @@ pub mod heal; pub use error::{Error, Result}; pub use heal::{ - HealManager, HealOperationsSnapshot, HealOptions, HealPriority, HealPriorityCounts, HealRequest, HealSourceCounts, HealType, + HealAdmissionTelemetry, HealManager, HealOperationsSnapshot, HealOptions, HealPriority, HealPriorityCounts, HealRequest, + HealSourceCounts, HealType, channel::HealChannelProcessor, progress::{HealProgress, aggregate_heal_progress}, resume::{ReplacementRecoveryRecord, ReplacementRecoveryState, ResumeUtils}, diff --git a/rustfs/src/admin/handlers/heal.rs b/rustfs/src/admin/handlers/heal.rs index b0d8173c9..2b78eff01 100644 --- a/rustfs/src/admin/handlers/heal.rs +++ b/rustfs/src/admin/handlers/heal.rs @@ -339,6 +339,19 @@ fn add_source_counts(total: &mut rustfs_heal::HealSourceCounts, next: rustfs_hea total.mrf = total.mrf.saturating_add(next.mrf); } +fn add_admission_telemetry(total: &mut rustfs_heal::HealAdmissionTelemetry, next: rustfs_heal::HealAdmissionTelemetry) { + total.accepted = total.accepted.saturating_add(next.accepted); + total.merged = total.merged.saturating_add(next.merged); + total.full = total.full.saturating_add(next.full); + total.dropped = total.dropped.saturating_add(next.dropped); + total.duplicate = total.duplicate.saturating_add(next.duplicate); + total.overlap_rejected = total.overlap_rejected.saturating_add(next.overlap_rejected); + total.displaced = total.displaced.saturating_add(next.displaced); + total.force_start = total.force_start.saturating_add(next.force_start); + total.max_start_duration_micros = total.max_start_duration_micros.max(next.max_start_duration_micros); + total.max_lock_phase_micros = total.max_lock_phase_micros.max(next.max_lock_phase_micros); +} + fn add_operations(total: &mut rustfs_heal::HealOperationsSnapshot, next: rustfs_heal::HealOperationsSnapshot) { total.queue_length = total.queue_length.saturating_add(next.queue_length); total.active_tasks = total.active_tasks.saturating_add(next.active_tasks); @@ -349,6 +362,7 @@ fn add_operations(total: &mut rustfs_heal::HealOperationsSnapshot, next: rustfs_ add_source_counts(&mut total.queued_by_source, next.queued_by_source); add_source_counts(&mut total.active_by_source, next.active_by_source); add_source_counts(&mut total.retrying_by_source, next.retrying_by_source); + add_admission_telemetry(&mut total.admission, next.admission); } fn aggregate_cluster_heal_status(snapshots: Vec) -> ClusterHealStatusSnapshot { @@ -2307,6 +2321,10 @@ 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["healOperations"]["admission"]["accepted"].is_u64()); + assert!(json["healOperations"]["admission"]["duplicate"].is_u64()); + assert!(json["healOperations"]["admission"]["forceStart"].is_u64()); + assert!(json["healOperations"]["admission"]["maxLockPhaseMicros"].is_u64()); assert_eq!(json["state"], "active"); assert_eq!(json["clusterStatusComplete"], true); assert!(json["progress"].is_null()); @@ -2486,6 +2504,18 @@ mod tests { queued_by_source: sources(value), active_by_source: sources(value), retrying_by_source: sources(value), + admission: rustfs_heal::HealAdmissionTelemetry { + accepted: value, + merged: value, + full: value, + dropped: value, + duplicate: value, + overlap_rejected: value, + displaced: value, + force_start: value, + max_start_duration_micros: value, + max_lock_phase_micros: value, + }, }; let progress = |value| NodeHealProgress { objects_scanned: value, diff --git a/rustfs/src/storage/rpc/node_service/heal.rs b/rustfs/src/storage/rpc/node_service/heal.rs index e5aaec78b..fe9637453 100644 --- a/rustfs/src/storage/rpc/node_service/heal.rs +++ b/rustfs/src/storage/rpc/node_service/heal.rs @@ -726,6 +726,7 @@ mod tests { assert_eq!(decoded.info().bitrot_start_cycle, 9); assert_eq!(decoded.operations.queue_length, 2); assert_eq!(decoded.operations.queued_by_source.mrf, 0); + assert_eq!(decoded.operations.admission, rustfs_heal::HealAdmissionTelemetry::default()); let progress = decoded.progress.expect("legacy progress should decode"); assert_eq!(progress.objects_scanned, 7); assert!(!progress.baseline_known);