diff --git a/crates/common/src/metrics.rs b/crates/common/src/metrics.rs index 142a678d9..824e4f44c 100644 --- a/crates/common/src/metrics.rs +++ b/crates/common/src/metrics.rs @@ -309,6 +309,62 @@ impl ScannerWorkSource { } } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ScannerReplicationRepairKind { + BucketObject, + BucketDeleteMarker, + BucketVersionPurge, + BucketExistingObject, + SitePassiveRequeue, + SiteActiveResync, +} + +impl ScannerReplicationRepairKind { + const ALL: [Self; 6] = [ + Self::BucketObject, + Self::BucketDeleteMarker, + Self::BucketVersionPurge, + Self::BucketExistingObject, + Self::SitePassiveRequeue, + Self::SiteActiveResync, + ]; + + pub fn source(self) -> ScannerWorkSource { + match self { + Self::BucketObject | Self::BucketDeleteMarker | Self::BucketVersionPurge | Self::BucketExistingObject => { + ScannerWorkSource::BucketReplication + } + Self::SitePassiveRequeue | Self::SiteActiveResync => ScannerWorkSource::SiteReplication, + } + } + + pub fn as_str(self) -> &'static str { + match self { + Self::BucketObject => "object", + Self::BucketDeleteMarker => "delete_marker", + Self::BucketVersionPurge => "version_purge", + Self::BucketExistingObject => "existing_object", + Self::SitePassiveRequeue => "passive_requeue", + Self::SiteActiveResync => "active_resync", + } + } + + fn all() -> &'static [Self] { + &Self::ALL + } + + fn index(self) -> usize { + match self { + Self::BucketObject => 0, + Self::BucketDeleteMarker => 1, + Self::BucketVersionPurge => 2, + Self::BucketExistingObject => 3, + Self::SitePassiveRequeue => 4, + Self::SiteActiveResync => 5, + } + } +} + #[derive(Debug, Default)] struct ScannerSourceWorkCounters { checked: AtomicU64, @@ -454,6 +510,19 @@ impl ScannerSourceWorkValues { missed: self.missed, } } + + fn replication_repair_snapshot(self, kind: ScannerReplicationRepairKind) -> ScannerReplicationRepairSnapshot { + ScannerReplicationRepairSnapshot { + source: kind.source().as_str().to_string(), + kind: kind.as_str().to_string(), + checked: self.checked, + queued: self.queued, + executed: self.executed, + failed: self.failed, + skipped: self.skipped, + missed: self.missed, + } + } } // --------------------------------------------------------------------------- @@ -687,6 +756,9 @@ pub struct Metrics { scanner_source_work: Vec, current_scan_cycle_source_work_start: Vec, last_scan_cycle_source_work: Vec, + scanner_replication_repair_work: Vec, + current_scan_cycle_replication_repair_work_start: Vec, + last_scan_cycle_replication_repair_work: Vec, partial_scan_cycles: AtomicU64, } @@ -781,6 +853,19 @@ pub struct ScannerSourceWorkSnapshot { pub missed: u64, } +#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)] +pub struct ScannerReplicationRepairSnapshot { + pub source: String, + pub kind: String, + pub checked: u64, + pub queued: u64, + pub executed: u64, + pub failed: u64, + pub skipped: u64, + #[serde(default)] + pub missed: u64, +} + #[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)] pub struct ScannerMaintenanceSourceSnapshot { pub source: String, @@ -1050,6 +1135,12 @@ pub struct ScannerMetricsReport { #[serde(default)] pub last_cycle_source_work: Vec, #[serde(default)] + pub replication_repair: Vec, + #[serde(default)] + pub current_cycle_replication_repair: Vec, + #[serde(default)] + pub last_cycle_replication_repair: Vec, + #[serde(default)] pub partial_cycles: u64, } @@ -1549,6 +1640,18 @@ impl Metrics { .iter() .map(|_| ScannerSourceWorkCounters::default()) .collect(), + scanner_replication_repair_work: ScannerReplicationRepairKind::all() + .iter() + .map(|_| ScannerSourceWorkCounters::default()) + .collect(), + current_scan_cycle_replication_repair_work_start: ScannerReplicationRepairKind::all() + .iter() + .map(|_| ScannerSourceWorkCounters::default()) + .collect(), + last_scan_cycle_replication_repair_work: ScannerReplicationRepairKind::all() + .iter() + .map(|_| ScannerSourceWorkCounters::default()) + .collect(), partial_scan_cycles: AtomicU64::new(0), } } @@ -1812,6 +1915,12 @@ impl Metrics { } } + pub fn record_scanner_replication_repair_work(&self, kind: ScannerReplicationRepairKind, work: ScannerSourceWorkUpdate) { + if let Some(counters) = self.scanner_replication_repair_work.get(kind.index()) { + counters.add(work); + } + } + fn record_last_cycle_scanner_source_work(&self, source: ScannerWorkSource, work: ScannerSourceWorkUpdate) { if let Some(counters) = self.last_scan_cycle_source_work.get(source.index()) { counters.add(work); @@ -2060,6 +2169,7 @@ impl Metrics { pub fn start_scan_cycle_work(&self) -> ScanCycleWorkSnapshot { let snapshot = self.scan_cycle_work_snapshot(); let source_snapshot = self.scanner_source_work_values(); + let replication_repair_snapshot = self.scanner_replication_repair_work_values(); self.current_scan_cycle_objects_start .store(snapshot.objects_scanned, Ordering::Relaxed); self.current_scan_cycle_directories_start @@ -2089,6 +2199,10 @@ impl Metrics { self.current_scan_cycle_usage_saves_start .store(snapshot.usage_saves, Ordering::Relaxed); self.store_scanner_source_work_values(&self.current_scan_cycle_source_work_start, &source_snapshot); + self.store_scanner_replication_repair_work_values( + &self.current_scan_cycle_replication_repair_work_start, + &replication_repair_snapshot, + ); self.current_scan_cycle_work_active.store(true, Ordering::Relaxed); snapshot } @@ -2096,8 +2210,11 @@ impl Metrics { pub fn finish_scan_cycle_work(&self, start: ScanCycleWorkSnapshot) { let work = self.scan_cycle_work_since(start); let source_work = self.scanner_source_work_since(&self.current_scan_cycle_source_work_start_values()); + let replication_repair_work = + self.scanner_replication_repair_work_since(&self.current_scan_cycle_replication_repair_work_start_values()); self.record_scan_cycle_work(work); self.record_scan_cycle_source_work(&source_work); + self.record_scan_cycle_replication_repair_work(&replication_repair_work); self.current_scan_cycle_work_active.store(false, Ordering::Relaxed); } @@ -2213,6 +2330,64 @@ impl Metrics { .collect() } + fn scanner_replication_repair_work_values(&self) -> Vec { + ScannerReplicationRepairKind::all() + .iter() + .filter_map(|kind| { + self.scanner_replication_repair_work + .get(kind.index()) + .map(ScannerSourceWorkCounters::values) + }) + .collect() + } + + fn current_scan_cycle_replication_repair_work_start_values(&self) -> Vec { + ScannerReplicationRepairKind::all() + .iter() + .filter_map(|kind| { + self.current_scan_cycle_replication_repair_work_start + .get(kind.index()) + .map(ScannerSourceWorkCounters::values) + }) + .collect() + } + + fn scanner_replication_repair_work_since(&self, start: &[ScannerSourceWorkValues]) -> Vec { + self.scanner_replication_repair_work_values() + .into_iter() + .enumerate() + .map(|(index, current)| current.saturating_sub(start.get(index).copied().unwrap_or_default())) + .collect() + } + + fn scanner_replication_repair_work_snapshots( + &self, + values: &[ScannerSourceWorkValues], + ) -> Vec { + ScannerReplicationRepairKind::all() + .iter() + .filter_map(|kind| { + values + .get(kind.index()) + .map(|values| values.replication_repair_snapshot(*kind)) + }) + .collect() + } + + fn scanner_replication_repair_work_counter_snapshots( + &self, + counters: &[ScannerSourceWorkCounters], + ) -> Vec { + ScannerReplicationRepairKind::all() + .iter() + .filter_map(|kind| { + counters + .get(kind.index()) + .map(|counters| counters.values().replication_repair_snapshot(*kind)) + }) + .collect() + } + fn store_scanner_source_work_values(&self, counters: &[ScannerSourceWorkCounters], values: &[ScannerSourceWorkValues]) { for source in ScannerWorkSource::all() { if let (Some(counter), Some(values)) = (counters.get(source.index()), values.get(source.index())) { @@ -2221,6 +2396,18 @@ impl Metrics { } } + fn store_scanner_replication_repair_work_values( + &self, + counters: &[ScannerSourceWorkCounters], + values: &[ScannerSourceWorkValues], + ) { + for kind in ScannerReplicationRepairKind::all() { + if let (Some(counter), Some(values)) = (counters.get(kind.index()), values.get(kind.index())) { + counter.store(*values); + } + } + } + pub fn record_scan_cycle_work(&self, work: ScanCycleWorkSnapshot) { // Telemetry-only gauges: readers may observe a transient mixed snapshot // while these independent atomic fields are updated. @@ -2254,6 +2441,10 @@ impl Metrics { self.store_scanner_source_work_values(&self.last_scan_cycle_source_work, work); } + fn record_scan_cycle_replication_repair_work(&self, work: &[ScannerSourceWorkValues]) { + self.store_scanner_replication_repair_work_values(&self.last_scan_cycle_replication_repair_work, work); + } + /// Snapshot of every path currently being scanned. pub async fn get_current_paths(&self) -> Vec { self.current_path_snapshots() @@ -2326,6 +2517,8 @@ impl Metrics { if self.current_scan_cycle_work_active.load(Ordering::Relaxed) { let current_work = self.scan_cycle_work_since(self.current_scan_cycle_work_start()); let current_source_work = self.scanner_source_work_since(&self.current_scan_cycle_source_work_start_values()); + let current_replication_repair_work = + self.scanner_replication_repair_work_since(&self.current_scan_cycle_replication_repair_work_start_values()); m.current_cycle_objects_scanned = current_work.objects_scanned; m.current_cycle_directories_scanned = current_work.directories_scanned; m.current_cycle_bucket_drive_scans = current_work.bucket_drive_scans; @@ -2341,6 +2534,7 @@ impl Metrics { m.current_cycle_replication_checks = current_work.replication_checks; m.current_cycle_usage_saves = current_work.usage_saves; m.current_cycle_source_work = self.scanner_source_work_snapshots(¤t_source_work); + m.current_cycle_replication_repair = self.scanner_replication_repair_work_snapshots(¤t_replication_repair_work); } let last_cycle_result = self.last_scan_cycle_result.load(Ordering::Relaxed); m.last_cycle_result = scan_cycle_result_label(last_cycle_result).to_string(); @@ -2371,6 +2565,8 @@ impl Metrics { m.last_cycle_replication_checks = self.last_scan_cycle_replication_checks.load(Ordering::Relaxed); m.last_cycle_usage_saves = self.last_scan_cycle_usage_saves.load(Ordering::Relaxed); m.last_cycle_source_work = self.scanner_source_work_counter_snapshots(&self.last_scan_cycle_source_work); + m.last_cycle_replication_repair = + self.scanner_replication_repair_work_counter_snapshots(&self.last_scan_cycle_replication_repair_work); m.failed_cycles = self.failed_scan_cycles.load(Ordering::Relaxed); m.partial_cycles_unknown = self.partial_scan_cycles_unknown.load(Ordering::Relaxed); m.partial_cycles_runtime = self.partial_scan_cycles_runtime.load(Ordering::Relaxed); @@ -2430,6 +2626,7 @@ impl Metrics { m.scan_checkpoint_ignored = self.scanner_checkpoint_ignored.load(Ordering::Relaxed); m.scan_checkpoint_stale = self.scanner_checkpoint_stale.load(Ordering::Relaxed); m.source_work = self.scanner_source_work_counter_snapshots(&self.scanner_source_work); + m.replication_repair = self.scanner_replication_repair_work_counter_snapshots(&self.scanner_replication_repair_work); m.partial_cycles = self.partial_scan_cycles.load(Ordering::Relaxed); // Lifetime operation counts @@ -3069,6 +3266,60 @@ mod tests { assert_eq!(heal.executed, 0); } + #[tokio::test] + async fn report_includes_scanner_replication_repair_work() { + let metrics = Metrics::new(); + + metrics.record_scanner_replication_repair_work( + ScannerReplicationRepairKind::BucketObject, + ScannerSourceWorkUpdate::queued(2), + ); + metrics.record_scanner_replication_repair_work( + ScannerReplicationRepairKind::BucketDeleteMarker, + ScannerSourceWorkUpdate::missed(1), + ); + metrics.record_scanner_replication_repair_work( + ScannerReplicationRepairKind::BucketVersionPurge, + ScannerSourceWorkUpdate { + skipped: 3, + ..Default::default() + }, + ); + metrics.record_scanner_replication_repair_work( + ScannerReplicationRepairKind::BucketExistingObject, + ScannerSourceWorkUpdate::queued(4), + ); + + let report = metrics.report().await; + let object = report + .replication_repair + .iter() + .find(|repair| repair.source == "bucket_replication" && repair.kind == "object") + .expect("bucket object repair work should be visible"); + assert_eq!(object.queued, 2); + + let delete_marker = report + .replication_repair + .iter() + .find(|repair| repair.source == "bucket_replication" && repair.kind == "delete_marker") + .expect("bucket delete-marker repair work should be visible"); + assert_eq!(delete_marker.missed, 1); + + let version_purge = report + .replication_repair + .iter() + .find(|repair| repair.source == "bucket_replication" && repair.kind == "version_purge") + .expect("bucket version purge repair work should be visible"); + assert_eq!(version_purge.skipped, 3); + + let existing_object = report + .replication_repair + .iter() + .find(|repair| repair.source == "bucket_replication" && repair.kind == "existing_object") + .expect("bucket existing object repair work should be visible"); + assert_eq!(existing_object.queued, 4); + } + #[tokio::test] async fn report_includes_scan_cycle_source_work() { let metrics = Metrics::new(); diff --git a/crates/ecstore/src/metrics_realtime.rs b/crates/ecstore/src/metrics_realtime.rs index a7d263bb4..a3f3376d4 100644 --- a/crates/ecstore/src/metrics_realtime.rs +++ b/crates/ecstore/src/metrics_realtime.rs @@ -24,6 +24,7 @@ use rustfs_madmin::metrics::{ ScannerMaintenanceControlSnapshot as MadminScannerMaintenanceControlSnapshot, ScannerMaintenanceSourceSnapshot as MadminScannerMaintenanceSourceSnapshot, ScannerMetrics as MadminScannerMetrics, ScannerPacingPressureSnapshot as MadminScannerPacingPressureSnapshot, + ScannerReplicationRepairSnapshot as MadminScannerReplicationRepairSnapshot, ScannerSourceCycleSnapshot as MadminScannerSourceCycleSnapshot, ScannerSourceWorkSnapshot as MadminScannerSourceWorkSnapshot, TimedAction as MadminTimedAction, }; @@ -276,6 +277,48 @@ fn to_madmin_scanner_metrics(metrics: rustfs_common::metrics::ScannerMetricsRepo missed: work.missed, }) .collect(), + replication_repair: metrics + .replication_repair + .into_iter() + .map(|repair| MadminScannerReplicationRepairSnapshot { + source: repair.source, + kind: repair.kind, + checked: repair.checked, + queued: repair.queued, + executed: repair.executed, + failed: repair.failed, + skipped: repair.skipped, + missed: repair.missed, + }) + .collect(), + current_cycle_replication_repair: metrics + .current_cycle_replication_repair + .into_iter() + .map(|repair| MadminScannerReplicationRepairSnapshot { + source: repair.source, + kind: repair.kind, + checked: repair.checked, + queued: repair.queued, + executed: repair.executed, + failed: repair.failed, + skipped: repair.skipped, + missed: repair.missed, + }) + .collect(), + last_cycle_replication_repair: metrics + .last_cycle_replication_repair + .into_iter() + .map(|repair| MadminScannerReplicationRepairSnapshot { + source: repair.source, + kind: repair.kind, + checked: repair.checked, + queued: repair.queued, + executed: repair.executed, + failed: repair.failed, + skipped: repair.skipped, + missed: repair.missed, + }) + .collect(), partial_cycles: metrics.partial_cycles, } } @@ -761,7 +804,37 @@ mod test { skipped: 60, missed: 61, }], - partial_cycles: 62, + replication_repair: vec![rustfs_common::metrics::ScannerReplicationRepairSnapshot { + source: "bucket_replication".to_string(), + kind: "object".to_string(), + checked: 62, + queued: 63, + executed: 64, + failed: 65, + skipped: 66, + missed: 67, + }], + current_cycle_replication_repair: vec![rustfs_common::metrics::ScannerReplicationRepairSnapshot { + source: "bucket_replication".to_string(), + kind: "delete_marker".to_string(), + checked: 68, + queued: 69, + executed: 70, + failed: 71, + skipped: 72, + missed: 73, + }], + last_cycle_replication_repair: vec![rustfs_common::metrics::ScannerReplicationRepairSnapshot { + source: "site_replication".to_string(), + kind: "active_resync".to_string(), + checked: 74, + queued: 75, + executed: 76, + failed: 77, + skipped: 78, + missed: 79, + }], + partial_cycles: 80, ..Default::default() }); @@ -831,6 +904,14 @@ mod test { assert_eq!(scanner.current_cycle_source_work[0].queued, 51); assert_eq!(scanner.last_cycle_source_work[0].source, "heal"); assert_eq!(scanner.last_cycle_source_work[0].skipped, 60); - assert_eq!(scanner.partial_cycles, 62); + assert_eq!(scanner.replication_repair[0].source, "bucket_replication"); + assert_eq!(scanner.replication_repair[0].kind, "object"); + assert_eq!(scanner.replication_repair[0].missed, 67); + assert_eq!(scanner.current_cycle_replication_repair[0].kind, "delete_marker"); + assert_eq!(scanner.current_cycle_replication_repair[0].queued, 69); + assert_eq!(scanner.last_cycle_replication_repair[0].source, "site_replication"); + assert_eq!(scanner.last_cycle_replication_repair[0].kind, "active_resync"); + assert_eq!(scanner.last_cycle_replication_repair[0].skipped, 78); + assert_eq!(scanner.partial_cycles, 80); } } diff --git a/crates/madmin/src/metrics.rs b/crates/madmin/src/metrics.rs index 3fc9259a5..0fcb371a8 100644 --- a/crates/madmin/src/metrics.rs +++ b/crates/madmin/src/metrics.rs @@ -158,6 +158,26 @@ pub struct ScannerSourceWorkSnapshot { pub missed: u64, } +#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)] +pub struct ScannerReplicationRepairSnapshot { + #[serde(rename = "source", default)] + pub source: String, + #[serde(rename = "kind", default)] + pub kind: String, + #[serde(rename = "checked", default)] + pub checked: u64, + #[serde(rename = "queued", default)] + pub queued: u64, + #[serde(rename = "executed", default)] + pub executed: u64, + #[serde(rename = "failed", default)] + pub failed: u64, + #[serde(rename = "skipped", default)] + pub skipped: u64, + #[serde(rename = "missed", default)] + pub missed: u64, +} + #[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)] pub struct ScannerMaintenanceSourceSnapshot { #[serde(rename = "source", default)] @@ -271,6 +291,17 @@ impl ScannerSourceWorkSnapshot { } } +impl ScannerReplicationRepairSnapshot { + fn merge(&mut self, other: &Self) { + self.checked = self.checked.saturating_add(other.checked); + self.queued = self.queued.saturating_add(other.queued); + self.executed = self.executed.saturating_add(other.executed); + self.failed = self.failed.saturating_add(other.failed); + self.skipped = self.skipped.saturating_add(other.skipped); + self.missed = self.missed.saturating_add(other.missed); + } +} + fn merge_source_work_snapshots(target: &mut Vec, source: &[ScannerSourceWorkSnapshot]) { for work in source { if let Some(existing) = target.iter_mut().find(|existing| existing.source == work.source) { @@ -282,6 +313,23 @@ fn merge_source_work_snapshots(target: &mut Vec, sour target.sort_by(|left, right| left.source.cmp(&right.source)); } +fn merge_replication_repair_snapshots( + target: &mut Vec, + source: &[ScannerReplicationRepairSnapshot], +) { + for repair in source { + if let Some(existing) = target + .iter_mut() + .find(|existing| existing.source == repair.source && existing.kind == repair.kind) + { + existing.merge(repair); + } else { + target.push(repair.clone()); + } + } + target.sort_by(|left, right| left.source.cmp(&right.source).then_with(|| left.kind.cmp(&right.kind))); +} + const SCANNER_PRIMARY_PRESSURE_NONE: &str = "none"; fn default_scanner_primary_pressure() -> String { @@ -599,6 +647,12 @@ pub struct ScannerMetrics { pub current_cycle_source_work: Vec, #[serde(rename = "last_cycle_source_work", default)] pub last_cycle_source_work: Vec, + #[serde(rename = "replication_repair", default)] + pub replication_repair: Vec, + #[serde(rename = "current_cycle_replication_repair", default)] + pub current_cycle_replication_repair: Vec, + #[serde(rename = "last_cycle_replication_repair", default)] + pub last_cycle_replication_repair: Vec, #[serde(rename = "partial_cycles", default)] pub partial_cycles: u64, } @@ -730,6 +784,9 @@ impl ScannerMetrics { merge_source_work_snapshots(&mut self.source_work, &other.source_work); merge_source_work_snapshots(&mut self.current_cycle_source_work, &other.current_cycle_source_work); merge_source_work_snapshots(&mut self.last_cycle_source_work, &other.last_cycle_source_work); + merge_replication_repair_snapshots(&mut self.replication_repair, &other.replication_repair); + merge_replication_repair_snapshots(&mut self.current_cycle_replication_repair, &other.current_cycle_replication_repair); + merge_replication_repair_snapshots(&mut self.last_cycle_replication_repair, &other.last_cycle_replication_repair); if self.ongoing_buckets < other.ongoing_buckets { self.ongoing_buckets = other.ongoing_buckets; @@ -1557,6 +1614,68 @@ mod tests { assert_eq!(usage.partial_cycles, 2); } + #[test] + fn scanner_metrics_merge_aggregates_replication_repair_status() { + let collected_at = Utc::now(); + let mut scanner = ScannerMetrics { + collected_at, + replication_repair: vec![ScannerReplicationRepairSnapshot { + source: "bucket_replication".to_string(), + kind: "object".to_string(), + queued: 2, + ..Default::default() + }], + ..Default::default() + }; + + scanner.merge(&ScannerMetrics { + collected_at: collected_at + chrono::Duration::seconds(1), + replication_repair: vec![ + ScannerReplicationRepairSnapshot { + source: "bucket_replication".to_string(), + kind: "object".to_string(), + missed: 3, + ..Default::default() + }, + ScannerReplicationRepairSnapshot { + source: "bucket_replication".to_string(), + kind: "delete_marker".to_string(), + skipped: 4, + ..Default::default() + }, + ScannerReplicationRepairSnapshot { + source: "site_replication".to_string(), + kind: "active_resync".to_string(), + queued: 5, + ..Default::default() + }, + ], + ..Default::default() + }); + + let object = scanner + .replication_repair + .iter() + .find(|repair| repair.source == "bucket_replication" && repair.kind == "object") + .expect("bucket object repair snapshot should be merged"); + assert_eq!(object.queued, 2); + assert_eq!(object.missed, 3); + + let delete_marker = scanner + .replication_repair + .iter() + .find(|repair| repair.source == "bucket_replication" && repair.kind == "delete_marker") + .expect("bucket delete-marker repair snapshot should be merged"); + assert_eq!(delete_marker.skipped, 4); + + let site_resync = scanner + .replication_repair + .iter() + .find(|repair| repair.source == "site_replication" && repair.kind == "active_resync") + .expect("site active resync snapshot should remain separate"); + assert_eq!(site_resync.queued, 5); + } + #[test] fn scanner_metrics_merge_preserves_distributed_status_fields() { let collected_at = Utc::now(); diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index 745ce52bf..3140a21ae 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -36,8 +36,8 @@ use rustfs_common::heal_channel::{ send_heal_request_with_admission, }; use rustfs_common::metrics::{ - IlmAction, Metric, Metrics, ScannerSourceWorkUpdate, ScannerWorkSource, UpdateCurrentPathFn, current_path_updater, - global_metrics, + IlmAction, Metric, Metrics, ScannerReplicationRepairKind, ScannerSourceWorkUpdate, ScannerWorkSource, UpdateCurrentPathFn, + current_path_updater, global_metrics, }; use rustfs_ecstore::bucket::lifecycle::bucket_lifecycle_audit::LcEventSrc; use rustfs_ecstore::bucket::lifecycle::bucket_lifecycle_ops::{GLOBAL_ExpiryState, apply_expiry_rule}; @@ -59,7 +59,9 @@ use rustfs_ecstore::global::is_erasure; use rustfs_ecstore::pools::{path2_bucket_object, path2_bucket_object_with_base_path}; use rustfs_ecstore::store_api::{ObjectInfo, ObjectToDelete}; use rustfs_ecstore::store_utils::is_reserved_or_invalid_bucket; -use rustfs_filemeta::{MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams, ReplicationStatusType}; +use rustfs_filemeta::{ + MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams, ReplicateObjectInfo, ReplicationStatusType, ReplicationType, +}; use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf}; use s3s::dto::{BucketLifecycleConfiguration, ObjectLockConfiguration}; use time::OffsetDateTime; @@ -205,16 +207,39 @@ fn record_scanner_ilm_action_if_queued(metrics: &Metrics, action: IlmAction, cou queued } -fn record_scanner_replication_admission(metrics: &Metrics, admission: ReplicationQueueAdmission) { - let work = match admission { +fn scanner_replication_work_update(admission: ReplicationQueueAdmission) -> ScannerSourceWorkUpdate { + match admission { ReplicationQueueAdmission::Queued => ScannerSourceWorkUpdate::queued(1), ReplicationQueueAdmission::Missed => ScannerSourceWorkUpdate::missed(1), ReplicationQueueAdmission::Skipped => ScannerSourceWorkUpdate { skipped: 1, ..Default::default() }, - }; + } +} + +fn scanner_replication_repair_kind(roi: &ReplicateObjectInfo) -> Option { + if roi.bucket.is_empty() && roi.name.is_empty() { + return None; + } + + if roi.op_type == ReplicationType::ExistingObject || roi.existing_obj_resync.must_resync() { + Some(ScannerReplicationRepairKind::BucketExistingObject) + } else if !roi.version_purge_status.is_empty() { + Some(ScannerReplicationRepairKind::BucketVersionPurge) + } else if roi.delete_marker { + Some(ScannerReplicationRepairKind::BucketDeleteMarker) + } else { + Some(ScannerReplicationRepairKind::BucketObject) + } +} + +fn record_scanner_replication_admission(metrics: &Metrics, roi: &ReplicateObjectInfo, admission: ReplicationQueueAdmission) { + let work = scanner_replication_work_update(admission); metrics.record_scanner_source_work(ScannerWorkSource::BucketReplication, work); + if let Some(kind) = scanner_replication_repair_kind(roi) { + metrics.record_scanner_replication_repair_work(kind, work); + } } fn scanner_heal_source(scan_mode: HealScanMode) -> ScannerWorkSource { @@ -920,8 +945,8 @@ impl ScannerItem { let done_replication = Metrics::time(Metric::CheckReplication); let replication_result = queue_replication_heal_internal(&oi.bucket, oi.clone(), (*replication).clone(), 0).await; done_replication(); - record_scanner_replication_admission(global_metrics(), replication_result.admission); let roi = replication_result.object_info; + record_scanner_replication_admission(global_metrics(), &roi, replication_result.admission); if !Self::should_account_replication_stats(oi) { return; } @@ -2409,7 +2434,7 @@ mod tests { use super::*; use rustfs_ecstore::disk::{DiskOption, endpoint::Endpoint, new_disk}; - use rustfs_filemeta::VersionPurgeStatusType; + use rustfs_filemeta::{ReplicateObjectInfo, ReplicationType, ResyncDecision, ResyncTargetDecision, VersionPurgeStatusType}; use serial_test::serial; #[cfg(unix)] use std::os::unix::fs::{PermissionsExt, symlink}; @@ -2605,10 +2630,15 @@ mod tests { #[tokio::test] async fn test_scanner_replication_admission_accounting_maps_source_work() { let metrics = Metrics::new(); + let object = ReplicateObjectInfo { + bucket: "bucket-a".to_string(), + name: "object-a".to_string(), + ..Default::default() + }; - record_scanner_replication_admission(&metrics, ReplicationQueueAdmission::Skipped); - record_scanner_replication_admission(&metrics, ReplicationQueueAdmission::Queued); - record_scanner_replication_admission(&metrics, ReplicationQueueAdmission::Missed); + record_scanner_replication_admission(&metrics, &object, ReplicationQueueAdmission::Skipped); + record_scanner_replication_admission(&metrics, &object, ReplicationQueueAdmission::Queued); + record_scanner_replication_admission(&metrics, &object, ReplicationQueueAdmission::Missed); let report = metrics.report().await; let replication = report @@ -2620,6 +2650,83 @@ mod tests { assert_eq!(replication.skipped, 1); assert_eq!(replication.queued, 1); assert_eq!(replication.missed, 1); + + let object_repair = report + .replication_repair + .iter() + .find(|repair| repair.source == ScannerWorkSource::BucketReplication.as_str() && repair.kind == "object") + .expect("bucket object repair work should be visible"); + assert_eq!(object_repair.skipped, 1); + assert_eq!(object_repair.queued, 1); + assert_eq!(object_repair.missed, 1); + } + + #[test] + fn test_scanner_replication_repair_kind_maps_bucket_variants() { + let object = ReplicateObjectInfo { + bucket: "bucket-a".to_string(), + name: "object-a".to_string(), + replication_status: ReplicationStatusType::Pending, + ..Default::default() + }; + assert_eq!(scanner_replication_repair_kind(&object), Some(ScannerReplicationRepairKind::BucketObject)); + + let delete_marker = ReplicateObjectInfo { + bucket: "bucket-a".to_string(), + name: "delete-marker-a".to_string(), + delete_marker: true, + replication_status: ReplicationStatusType::Failed, + ..Default::default() + }; + assert_eq!( + scanner_replication_repair_kind(&delete_marker), + Some(ScannerReplicationRepairKind::BucketDeleteMarker) + ); + + let version_purge = ReplicateObjectInfo { + bucket: "bucket-a".to_string(), + name: "version-purge-a".to_string(), + version_purge_status: VersionPurgeStatusType::Pending, + ..Default::default() + }; + assert_eq!( + scanner_replication_repair_kind(&version_purge), + Some(ScannerReplicationRepairKind::BucketVersionPurge) + ); + + let existing_object = ReplicateObjectInfo { + bucket: "bucket-a".to_string(), + name: "existing-object-a".to_string(), + op_type: ReplicationType::ExistingObject, + ..Default::default() + }; + assert_eq!( + scanner_replication_repair_kind(&existing_object), + Some(ScannerReplicationRepairKind::BucketExistingObject) + ); + + let mut existing_obj_resync = ResyncDecision::new(); + existing_obj_resync.targets.insert( + "arn:minio:replication:::target".to_string(), + ResyncTargetDecision { + replicate: true, + ..Default::default() + }, + ); + let existing_delete_marker = ReplicateObjectInfo { + bucket: "bucket-a".to_string(), + name: "existing-delete-marker-a".to_string(), + delete_marker: true, + replication_status: ReplicationStatusType::Completed, + existing_obj_resync, + ..Default::default() + }; + assert_eq!( + scanner_replication_repair_kind(&existing_delete_marker), + Some(ScannerReplicationRepairKind::BucketExistingObject) + ); + + assert_eq!(scanner_replication_repair_kind(&ReplicateObjectInfo::default()), None); } #[tokio::test] diff --git a/docs/operations/scanner-benchmark-runbook.md b/docs/operations/scanner-benchmark-runbook.md index d57025b52..c339ed02b 100644 --- a/docs/operations/scanner-benchmark-runbook.md +++ b/docs/operations/scanner-benchmark-runbook.md @@ -292,6 +292,9 @@ Compare these fields between baseline and tuned runs: | `metrics.source_work` | Shows cumulative work found, queued, skipped, missed, executed, and failed by source. | | `metrics.current_cycle_source_work` | Shows which source is consuming the current scan cycle. | | `metrics.last_cycle_source_work` | Shows which source consumed the previous scan cycle. | +| `metrics.replication_repair` | Splits scanner-discovered replication repair by source and kind, including bucket object, delete-marker, version-purge, existing-object repair, and site replication boundary states. | +| `metrics.current_cycle_replication_repair` | Shows which replication repair kind is being discovered or admitted in the current cycle. | +| `metrics.last_cycle_replication_repair` | Shows which replication repair kind consumed the previous cycle. | | `metrics.lifecycle_expiry.current_queued` | Shows scanner-driven expiry/delete work waiting in the expiry worker queue. | | `metrics.lifecycle_expiry.current_active` | Shows scanner-driven expiry/delete work currently running in expiry workers. | | `metrics.lifecycle_expiry.queue_missed` | Shows expiry/delete queue admission failures outside the scanner walk itself. | diff --git a/docs/operations/scanner-runtime-controls.md b/docs/operations/scanner-runtime-controls.md index 72b0d08a9..9b822ee17 100644 --- a/docs/operations/scanner-runtime-controls.md +++ b/docs/operations/scanner-runtime-controls.md @@ -99,6 +99,7 @@ metrics.lifecycle_transition.current_queued metrics.lifecycle_transition.scanner_missed metrics.maintenance_control.primary_control metrics.source_work +metrics.replication_repair metrics.scan_checkpoint ``` @@ -147,6 +148,28 @@ 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 Replication Repair + +`metrics.replication_repair`, `metrics.current_cycle_replication_repair`, and +`metrics.last_cycle_replication_repair` split scanner-discovered replication +repair work by source and repair kind. + +Each entry has the same `checked`, `queued`, `executed`, `failed`, `skipped`, +and `missed` counters used by `source_work`, plus: + +| Field | Meaning | +|---|---| +| `source` | `bucket_replication` for bucket replication repair, or `site_replication` for site replication boundary signals. | +| `kind` | Bucket repair kinds are `object`, `delete_marker`, `version_purge`, and `existing_object`. Site replication boundary kinds are `passive_requeue` and `active_resync`. | + +For bucket replication, `queued` means scanner-discovered repair was admitted +to the replication queue, `missed` means the queue or worker path could not +accept it, and `skipped` means the object did not require a new repair task. + +The site replication kinds keep passive scanner discovery separate from active +resync. Scanner status may report site replication boundary counters, but the +scanner should not be treated as the active site replication resync controller. + ## Reading Maintenance Control `metrics.maintenance_control` derives a source-level control snapshot from