From 360bceafced5bd63a39f013f7fc30aef162c50f2 Mon Sep 17 00:00:00 2001 From: houseme Date: Tue, 18 Aug 2026 08:29:29 +0800 Subject: [PATCH] feat(heal): add progress and trace observability (#6179) * feat(heal): track erasure set progress baseline Record erasure-set heal byte progress from per-object results and seed progress totals from complete usage-cache snapshots when available. Keep usage-cache failures observational so heal execution continues without a baseline. Co-Authored-By: heihutu * feat(heal): skip filtered erasure set versions Skip erasure-set versions written after the durable heal start time, and queue lifecycle-expired versions for expiry before skipping them. Track new-version and ILM-expired skips separately so progress can explain completed baseline work without treating these skips as retry-blocking failures. Co-Authored-By: heihutu * feat(heal): wire abandoned data-dir cleanup check Connect check_abandoned_parts through ECStore, pool, and set layers so heal can invoke the existing orphan data-dir reclaim path instead of returning NotImplemented. Add dry-run support to the reclaim scan and cover dry-run plus scoped set behavior with regression tests. Co-Authored-By: heihutu * feat(obs): add heal scanner trace bus Introduce an in-process broadcast trace bus with typed heal and scanner events, lazy event construction, and bounded lagged-subscriber behavior. Cover zero-subscriber publishing, subscription delivery, drop accounting, and lagged receivers with focused common-crate tests. Co-Authored-By: heihutu * feat(obs): stream heal trace events from admin API Wire the admin trace endpoint to the common trace bus for heal/scanner events, including kind, regex, and threshold filtering. Co-Authored-By: heihutu * feat(obs): emit heal trace events Publish heal task lifecycle and abandoned-parts cleanup events through the common trace bus so the admin trace stream has live heal diagnostics. Co-Authored-By: heihutu * feat(obs): emit scanner trace events Publish scanner folder, lifecycle action, and heal-candidate events through the common trace bus for live admin scanner diagnostics. Co-Authored-By: heihutu * fix(heal): route data usage loader through storage api Keep ECStore data-usage facade access behind the heal storage_api boundary so architecture migration guards can validate the heal progress path. Co-Authored-By: heihutu * perf(heal): avoid lifecycle snapshots on ordinary heal pages Only request lifecycle object snapshots when the heal pass has lifecycle expiry context. This keeps ordinary listing and disk-walk pages from cloning FileInfo/ObjectInfo payloads while preserving the skip path that queues expired versions. Co-Authored-By: heihutu * test(heal): update bug-fix mocks for lifecycle snapshots Carry the lifecycle snapshot opt-in argument through the remaining heal bug-fix test mocks so all-targets clippy covers the updated storage trait. Co-Authored-By: heihutu * test(rustfs): sync heal storage mock signature Update the rustfs storage RPC test mock for the lifecycle snapshot opt-in argument and cover it with rustfs all-targets clippy. Co-Authored-By: heihutu * test(e2e): allocate smoke ports across nextest processes Serialize E2E port selection with a small /tmp allocator so nextest workers do not reuse the same just-released ephemeral port before RustFS binds it. Co-Authored-By: heihutu --------- Co-authored-by: heihutu --- Cargo.lock | 1 + crates/common/Cargo.toml | 1 + crates/common/src/lib.rs | 1 + crates/common/src/trace_bus.rs | 333 ++++++++++++++ crates/e2e_test/src/common.rs | 83 +++- crates/ecstore/src/api/mod.rs | 1 + crates/ecstore/src/bucket/lifecycle/mod.rs | 2 +- crates/ecstore/src/core/pools.rs | 91 ++++ crates/ecstore/src/core/sets.rs | 29 +- .../src/set_disk/core/io_primitives.rs | 22 + crates/ecstore/src/set_disk/mod.rs | 109 ++++- crates/ecstore/src/set_disk/ops/heal.rs | 61 ++- crates/ecstore/src/set_disk/ops/heal_walk.rs | 50 ++- crates/ecstore/src/store/heal.rs | 42 +- crates/ecstore/src/store/heal_walk.rs | 3 +- crates/heal/src/heal/channel.rs | 1 + crates/heal/src/heal/erasure_healer.rs | 344 ++++++++++++-- crates/heal/src/heal/manager.rs | 30 ++ crates/heal/src/heal/progress.rs | 166 ++++++- crates/heal/src/heal/storage.rs | 176 +++++++- crates/heal/src/heal/storage_api.rs | 13 +- crates/heal/src/heal/task.rs | 238 +++++++++- crates/heal/src/lib.rs | 1 + .../heal_b5_versioned_regression_test.rs | 2 +- .../tests/heal_b920_subquorum_union_test.rs | 6 +- crates/heal/tests/heal_bug_fixes_test.rs | 2 + crates/madmin/src/service_commands.rs | 10 +- crates/scanner/src/scanner_folder.rs | 254 +++++++++++ rustfs/src/admin/handlers/profile_admin.rs | 422 ++++++++++++++++-- rustfs/src/storage/rpc/node_service.rs | 1 + 30 files changed, 2383 insertions(+), 112 deletions(-) create mode 100644 crates/common/src/trace_bus.rs diff --git a/Cargo.lock b/Cargo.lock index b521a731d..85c91d523 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9280,6 +9280,7 @@ dependencies = [ "s3s", "serde", "serde_json", + "smallvec", "tokio", "tonic", "tracing", diff --git a/crates/common/Cargo.toml b/crates/common/Cargo.toml index 5c9094d61..ddc02cf2e 100644 --- a/crates/common/Cargo.toml +++ b/crates/common/Cargo.toml @@ -42,6 +42,7 @@ chrono = { workspace = true, features = ["serde"] } jiff = { workspace = true, features = ["serde"] } metrics = { workspace = true } serde = { workspace = true, features = ["derive"] } +smallvec = { workspace = true } rmp-serde = { workspace = true } s3s = { workspace = true, features = ["minio"] } tracing = { workspace = true } diff --git a/crates/common/src/lib.rs b/crates/common/src/lib.rs index 09240e25b..1ae200f24 100644 --- a/crates/common/src/lib.rs +++ b/crates/common/src/lib.rs @@ -19,6 +19,7 @@ pub mod last_minute; pub mod metrics; mod readiness; pub mod table_catalog; +pub mod trace_bus; pub use globals::*; pub use readiness::{GlobalReadiness, SystemStage}; diff --git a/crates/common/src/trace_bus.rs b/crates/common/src/trace_bus.rs new file mode 100644 index 000000000..5e6a90d60 --- /dev/null +++ b/crates/common/src/trace_bus.rs @@ -0,0 +1,333 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use smallvec::SmallVec; +use std::{ + sync::{ + Arc, OnceLock, + atomic::{AtomicUsize, Ordering}, + }, + time::{Duration, SystemTime}, +}; +use tokio::sync::broadcast; + +const DEFAULT_TRACE_BUS_CAPACITY: usize = 1024; +const TRACE_ATTR_INLINE_CAPACITY: usize = 8; + +static GLOBAL_TRACE_BUS: OnceLock = OnceLock::new(); + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum TraceKind { + Heal, + Scanner, +} + +impl TraceKind { + pub const fn as_str(self) -> &'static str { + match self { + Self::Heal => "heal", + Self::Scanner => "scanner", + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum TraceFunc { + HealTask, + HealBucket, + HealObject, + HealCheckAbandonedParts, + HealErasureSetPage, + ScannerFolder, + ScannerIlmAction, + ScannerHealCandidate, + Dropped, +} + +impl TraceFunc { + pub const fn as_str(self) -> &'static str { + match self { + Self::HealTask => "heal.Task", + Self::HealBucket => "heal.Bucket", + Self::HealObject => "heal.Object", + Self::HealCheckAbandonedParts => "heal.CheckAbandonedParts", + Self::HealErasureSetPage => "heal.ErasureSetPage", + Self::ScannerFolder => "scanner.Folder", + Self::ScannerIlmAction => "scanner.IlmAction", + Self::ScannerHealCandidate => "scanner.HealCandidate", + Self::Dropped => "trace.Dropped", + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum TraceVal { + Bool(bool), + U64(u64), + I64(i64), + Str(Arc), +} + +impl From for TraceVal { + fn from(value: bool) -> Self { + Self::Bool(value) + } +} + +impl From for TraceVal { + fn from(value: u64) -> Self { + Self::U64(value) + } +} + +impl From for TraceVal { + fn from(value: i64) -> Self { + Self::I64(value) + } +} + +impl From<&str> for TraceVal { + fn from(value: &str) -> Self { + Self::Str(Arc::from(value)) + } +} + +impl From for TraceVal { + fn from(value: String) -> Self { + Self::Str(Arc::from(value)) + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TraceAttr { + pub key: &'static str, + pub value: TraceVal, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TraceEvent { + pub kind: TraceKind, + pub func: TraceFunc, + pub time: SystemTime, + pub bucket: Option>, + pub object: Option>, + pub duration: Duration, + pub bytes: u64, + pub attrs: SmallVec<[TraceAttr; TRACE_ATTR_INLINE_CAPACITY]>, +} + +impl TraceEvent { + pub fn new(kind: TraceKind, func: TraceFunc) -> Self { + Self { + kind, + func, + time: SystemTime::now(), + bucket: None, + object: None, + duration: Duration::ZERO, + bytes: 0, + attrs: SmallVec::new(), + } + } + + pub fn with_bucket(mut self, bucket: impl Into>) -> Self { + self.bucket = Some(bucket.into()); + self + } + + pub fn with_object(mut self, object: impl Into>) -> Self { + self.object = Some(object.into()); + self + } + + pub fn with_duration(mut self, duration: Duration) -> Self { + self.duration = duration; + self + } + + pub fn with_bytes(mut self, bytes: u64) -> Self { + self.bytes = bytes; + self + } + + pub fn with_attr(mut self, key: &'static str, value: impl Into) -> Self { + self.attrs.push(TraceAttr { + key, + value: value.into(), + }); + self + } +} + +#[derive(Debug)] +pub struct TraceBus { + sender: broadcast::Sender>, + subscriber_count: Arc, +} + +impl TraceBus { + pub fn new(capacity: usize) -> Self { + let capacity = capacity.max(1); + let (sender, _receiver) = broadcast::channel(capacity); + Self { + sender, + subscriber_count: Arc::new(AtomicUsize::new(0)), + } + } + + pub fn subscriber_count(&self) -> usize { + self.subscriber_count.load(Ordering::Acquire) + } + + pub fn subscribe(&self) -> TraceSubscription { + let receiver = self.sender.subscribe(); + self.subscriber_count.fetch_add(1, Ordering::AcqRel); + TraceSubscription { + receiver, + subscriber_count: Arc::clone(&self.subscriber_count), + } + } + + pub fn emit(&self, build: impl FnOnce() -> TraceEvent) -> bool { + if self.subscriber_count() == 0 { + return false; + } + + self.sender.send(Arc::new(build())).is_ok() + } +} + +impl Default for TraceBus { + fn default() -> Self { + Self::new(DEFAULT_TRACE_BUS_CAPACITY) + } +} + +#[derive(Debug)] +pub struct TraceSubscription { + receiver: broadcast::Receiver>, + subscriber_count: Arc, +} + +impl TraceSubscription { + pub async fn recv(&mut self) -> Result, broadcast::error::RecvError> { + self.receiver.recv().await + } + + pub fn try_recv(&mut self) -> Result, broadcast::error::TryRecvError> { + self.receiver.try_recv() + } +} + +impl Drop for TraceSubscription { + fn drop(&mut self) { + self.subscriber_count.fetch_sub(1, Ordering::AcqRel); + } +} + +pub fn global_trace_bus() -> &'static TraceBus { + GLOBAL_TRACE_BUS.get_or_init(TraceBus::default) +} + +pub fn subscribe_trace_events() -> TraceSubscription { + global_trace_bus().subscribe() +} + +pub fn trace_emit(build: impl FnOnce() -> TraceEvent) -> bool { + global_trace_bus().emit(build) +} + +pub fn trace_subscriber_count() -> usize { + global_trace_bus().subscriber_count() +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::AtomicUsize; + + #[test] + fn trace_emit_skips_builder_without_subscribers() { + let bus = TraceBus::new(4); + let built = AtomicUsize::new(0); + + let sent = bus.emit(|| { + built.fetch_add(1, Ordering::Relaxed); + TraceEvent::new(TraceKind::Heal, TraceFunc::HealTask) + }); + + assert!(!sent); + assert_eq!(built.load(Ordering::Relaxed), 0); + } + + #[tokio::test] + async fn trace_subscriber_receives_event() { + let bus = TraceBus::new(4); + let mut subscription = bus.subscribe(); + + assert!(bus.emit(|| { + TraceEvent::new(TraceKind::Heal, TraceFunc::HealObject) + .with_bucket("bucket") + .with_object("object") + .with_duration(Duration::from_millis(7)) + .with_bytes(11) + .with_attr("dry", true) + })); + + let event = subscription + .recv() + .await + .expect("subscriber should receive emitted trace event"); + + assert_eq!(event.kind, TraceKind::Heal); + assert_eq!(event.func, TraceFunc::HealObject); + assert_eq!(event.bucket.as_deref(), Some("bucket")); + assert_eq!(event.object.as_deref(), Some("object")); + assert_eq!(event.duration, Duration::from_millis(7)); + assert_eq!(event.bytes, 11); + assert_eq!( + event.attrs.as_slice(), + &[TraceAttr { + key: "dry", + value: TraceVal::Bool(true) + }] + ); + } + + #[test] + fn trace_subscription_drop_decrements_count() { + let bus = TraceBus::new(4); + let subscription = bus.subscribe(); + + assert_eq!(bus.subscriber_count(), 1); + drop(subscription); + assert_eq!(bus.subscriber_count(), 0); + } + + #[tokio::test] + async fn lagged_subscriber_drops_events_without_blocking_publishers() { + let bus = TraceBus::new(2); + let mut subscription = bus.subscribe(); + + for index in 0_u64..4 { + assert!(bus.emit(|| { TraceEvent::new(TraceKind::Scanner, TraceFunc::ScannerFolder).with_attr("index", index) })); + } + + let err = subscription + .recv() + .await + .expect_err("receiver should observe lag instead of blocking publishers"); + assert!(matches!(err, broadcast::error::RecvError::Lagged(_))); + } +} diff --git a/crates/e2e_test/src/common.rs b/crates/e2e_test/src/common.rs index f1fcaa20a..460c764ed 100644 --- a/crates/e2e_test/src/common.rs +++ b/crates/e2e_test/src/common.rs @@ -32,6 +32,7 @@ use rustfs_signer::sign_v4; use s3s::Body; use std::ffi::OsStr; use std::fs as stdfs; +use std::io::ErrorKind; use std::path::{Path, PathBuf}; use std::process::{Child, Command, Stdio}; use std::sync::Once; @@ -51,6 +52,11 @@ pub(crate) const FAST_DATA_USAGE_SCANNER_ENV: &[(&str, &str)] = &[("RUSTFS_SCANNER_CYCLE", "1"), ("RUSTFS_SCANNER_START_DELAY_SECS", "0")]; pub const TEST_BUCKET: &str = "e2e-test-bucket"; const RUSTFS_FULL_FEATURE: &str = "full"; +const TEST_PORT_MIN: u16 = 20_000; +const TEST_PORT_RANGE: u16 = 40_000; +const TEST_PORT_COUNTER_PATH: &str = "/tmp/rustfs_e2e_next_port"; +const TEST_PORT_LOCK_DIR: &str = "/tmp/rustfs_e2e_port_allocator.lock"; +const TEST_PORT_LOCK_STALE_AFTER: Duration = Duration::from_secs(30); fn capture_log_path(log_dir: &Path, temp_dir: &str) -> Option { let temp_name = Path::new(temp_dir).file_name()?.to_string_lossy(); @@ -67,6 +73,64 @@ fn configured_capture_log_path(temp_dir: &str) -> Option { capture_log_path(Path::new(&log_dir), temp_dir).map(|path| path.to_string_lossy().into_owned()) } +struct PortAllocatorGuard; + +impl PortAllocatorGuard { + async fn acquire() -> Result> { + loop { + match stdfs::create_dir(TEST_PORT_LOCK_DIR) { + Ok(()) => return Ok(Self), + Err(err) if err.kind() == ErrorKind::AlreadyExists => { + remove_stale_port_allocator_lock(); + sleep(Duration::from_millis(10)).await; + } + Err(err) => return Err(err.into()), + } + } + } +} + +impl Drop for PortAllocatorGuard { + fn drop(&mut self) { + let _ = stdfs::remove_dir(TEST_PORT_LOCK_DIR); + } +} + +fn advance_test_port(port: u16) -> u16 { + let offset = (port - TEST_PORT_MIN + 1) % TEST_PORT_RANGE; + TEST_PORT_MIN + offset +} + +fn seeded_test_port() -> u16 { + let offset = (Uuid::new_v4().as_u128() % u128::from(TEST_PORT_RANGE)) as u16; + TEST_PORT_MIN + offset +} + +fn read_next_test_port() -> u16 { + stdfs::read_to_string(TEST_PORT_COUNTER_PATH) + .ok() + .and_then(|value| value.trim().parse::().ok()) + .filter(|port| (TEST_PORT_MIN..TEST_PORT_MIN + TEST_PORT_RANGE).contains(port)) + .unwrap_or_else(seeded_test_port) +} + +fn remove_stale_port_allocator_lock() { + let Ok(metadata) = stdfs::metadata(TEST_PORT_LOCK_DIR) else { + return; + }; + let Ok(modified) = metadata.modified() else { + return; + }; + if modified.elapsed().is_ok_and(|elapsed| elapsed > TEST_PORT_LOCK_STALE_AFTER) { + let _ = stdfs::remove_dir(TEST_PORT_LOCK_DIR); + } +} + +fn write_next_test_port(port: u16) -> Result<(), Box> { + stdfs::write(TEST_PORT_COUNTER_PATH, port.to_string())?; + Ok(()) +} + pub(crate) fn capture_command_logs( command: &mut Command, log_path: Option<&str>, @@ -508,10 +572,21 @@ impl RustFSTestEnvironment { /// Find an available port for the test pub async fn find_available_port() -> Result> { use std::net::TcpListener; - let listener = TcpListener::bind("127.0.0.1:0")?; - let port = listener.local_addr()?.port(); - drop(listener); - Ok(port) + let _guard = PortAllocatorGuard::acquire().await?; + let mut next_port = read_next_test_port(); + + for _ in 0..TEST_PORT_RANGE { + let port = next_port; + next_port = advance_test_port(next_port); + write_next_test_port(next_port)?; + + if let Ok(listener) = TcpListener::bind(("127.0.0.1", port)) { + drop(listener); + return Ok(port); + } + } + + Err("no available E2E test port found".into()) } /// Kill any existing RustFS processes diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 5526b7d00..4b8bc3249 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -483,6 +483,7 @@ pub mod store_list { } pub mod storage { + pub use crate::core::pools::HealLifecycleExpiryContext; pub use crate::store::HealWalkVersion; pub use crate::store::{ ECStore, all_local_disk, all_local_disk_path, find_local_disk_by_ref, init_local_disks, diff --git a/crates/ecstore/src/bucket/lifecycle/mod.rs b/crates/ecstore/src/bucket/lifecycle/mod.rs index 823263072..6d8e64f1b 100644 --- a/crates/ecstore/src/bucket/lifecycle/mod.rs +++ b/crates/ecstore/src/bucket/lifecycle/mod.rs @@ -19,7 +19,7 @@ pub mod core; pub mod evaluator; pub mod manual_transition_job; mod metadata_boundary; -pub(crate) use metadata_boundary::get_expiry_configs; +pub(crate) use metadata_boundary::{LifecycleExpiryConfigs, get_expiry_configs}; mod object_lock_boundary; pub use self::core as lifecycle; mod replication_sink; diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index ec9da3708..02fe70f29 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -16,6 +16,7 @@ use crate::bucket::replication::replication_state_from_filemeta; use crate::bucket::versioning_sys::BucketVersioningSys; use crate::bucket::{ lifecycle::{ + LifecycleExpiryConfigs, bucket_lifecycle_audit::LcEventSrc, bucket_lifecycle_ops::{ LifecycleOps, apply_expiry_on_transitioned_object, apply_expiry_rule_in, eval_action_from_lifecycle, @@ -2335,6 +2336,10 @@ fn lifecycle_action_removes_data_movement_version(action: IlmAction) -> bool { ) } +fn lifecycle_action_skips_heal_version(action: IlmAction) -> bool { + action.delete() +} + fn resolve_data_movement_lifecycle_expiry_result(action: IlmAction, apply_actions: bool, applied: bool) -> Result { if !apply_actions || applied { return Ok(true); @@ -2385,7 +2390,80 @@ pub(crate) async fn should_skip_lifecycle_for_data_movement( } } +pub struct HealLifecycleExpiryContext { + configs: LifecycleExpiryConfigs, +} + impl ECStore { + pub async fn load_heal_lifecycle_expiry_context(&self, bucket: &str) -> Result> { + if bucket == RUSTFS_META_BUCKET { + return Ok(None); + } + + let configs = get_expiry_configs(self, bucket).await?; + if configs.lifecycle.is_none() { + return Ok(None); + } + + Ok(Some(HealLifecycleExpiryContext { configs })) + } + + pub async fn enqueue_heal_lifecycle_expiry( + self: &Arc, + context: &HealLifecycleExpiryContext, + bucket: &str, + object: &str, + version_id: Option<&str>, + object_info: Option<&crate::object_api::ObjectInfo>, + ) -> Result { + let Some(lifecycle_config) = context.configs.lifecycle.as_ref() else { + return Ok(false); + }; + + let object_info = if let Some(object_info) = object_info { + if object_info.bucket != bucket || object_info.name != object { + return Ok(false); + } + let snapshot_version_id = object_info + .version_id + .filter(|version_id| !version_id.is_nil()) + .map(|version_id| version_id.to_string()); + if snapshot_version_id.as_deref() != version_id { + return Ok(false); + } + object_info.clone() + } else { + match self + .get_object_info( + bucket, + object, + &ObjectOptions { + version_id: version_id.map(str::to_string), + versioned: version_id.is_some(), + expected_bucket_incarnation_id: Some(context.configs.bucket_incarnation_id), + ..Default::default() + }, + ) + .await + { + Ok(object_info) => object_info, + Err(err) if is_err_object_not_found(&err) || is_err_version_not_found(&err) => return Ok(false), + Err(err) => return Err(err), + } + }; + + let event = eval_action_from_lifecycle(lifecycle_config, context.configs.object_lock.as_deref(), &object_info).await; + if !lifecycle_action_skips_heal_version(event.action) { + return Ok(false); + } + + if lifecycle_delete_all_versions_blocked_by_replication(self.clone(), bucket, &object_info.name, event.action).await? { + return Ok(false); + } + + Ok(apply_expiry_rule_in(self.clone(), &event, &LcEventSrc::Scanner, &object_info).await) + } + async fn save_current_pool_meta(&self) -> Result<()> { let _save_guard = self.pool_meta_save_gate.lock().await; let snapshot = { @@ -4287,6 +4365,19 @@ mod tests { )); } + #[test] + fn lifecycle_action_skips_heal_version_for_every_delete_action() { + assert!(lifecycle_action_skips_heal_version(IlmAction::DeleteAction)); + assert!(lifecycle_action_skips_heal_version(IlmAction::DeleteVersionAction)); + assert!(lifecycle_action_skips_heal_version(IlmAction::DeleteRestoredAction)); + assert!(lifecycle_action_skips_heal_version(IlmAction::DeleteRestoredVersionAction)); + assert!(lifecycle_action_skips_heal_version(IlmAction::DeleteAllVersionsAction)); + assert!(lifecycle_action_skips_heal_version(IlmAction::DelMarkerDeleteAllVersionsAction)); + assert!(!lifecycle_action_skips_heal_version(IlmAction::TransitionAction)); + assert!(!lifecycle_action_skips_heal_version(IlmAction::TransitionVersionAction)); + assert!(!lifecycle_action_skips_heal_version(IlmAction::NoneAction)); + } + #[test] fn resolve_data_movement_lifecycle_expiry_result_allows_dry_run_skip() { let skip = resolve_data_movement_lifecycle_expiry_result(IlmAction::DeleteVersionAction, false, false) diff --git a/crates/ecstore/src/core/sets.rs b/crates/ecstore/src/core/sets.rs index acb8b53b9..1d1bcedeb 100644 --- a/crates/ecstore/src/core/sets.rs +++ b/crates/ecstore/src/core/sets.rs @@ -1140,11 +1140,11 @@ impl crate::storage_api_contracts::heal::HealOperations for Sets { Err(Error::DiskNotFound) } - #[tracing::instrument(skip(self))] - async fn check_abandoned_parts(&self, _bucket: &str, _object: &str, _opts: &HealOpts) -> Result<()> { - // Multipart orphan reconciliation is intentionally retained above the pool/set layers - // until there is a concrete caller and a stable lower-level contract to implement. - Err(StorageError::NotImplemented) + #[tracing::instrument(level = "debug", skip(self, opts), fields(bucket = %bucket, object = %object, dry_run = opts.dry_run))] + async fn check_abandoned_parts(&self, bucket: &str, object: &str, opts: &HealOpts) -> Result<()> { + self.get_disks_for_heal_object(object, opts)? + .check_abandoned_parts(bucket, object, opts) + .await } } @@ -1996,7 +1996,7 @@ mod tests { } #[tokio::test] - async fn sets_check_abandoned_parts_returns_typed_not_implemented_error() { + async fn sets_check_abandoned_parts_rejects_invalid_set_scope() { let format = FormatV3::new(1, 1); let sets = Sets { id: format.id, @@ -2021,10 +2021,21 @@ mod tests { }; let err = sets - .check_abandoned_parts("bucket", "object", &HealOpts::default()) + .check_abandoned_parts( + "bucket", + "object", + &HealOpts { + set: Some(1), + ..Default::default() + }, + ) .await - .expect_err("abandoned-parts ownership should stay above the pool/set storage layers"); - assert!(matches!(err, StorageError::NotImplemented)); + .expect_err("out-of-range abandoned-parts set scope must fail closed"); + assert!( + matches!(err, StorageError::InvalidArgument(_, ref field, ref reason) + if field == "set" && reason.contains("invalid heal set index 1")), + "unexpected invalid set error: {err:?}" + ); } // Builds a single-set `Sets` over `SET_DRIVE_COUNT` local temp-dir disks, diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index 41bd6dac7..7af25a64c 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -4860,6 +4860,14 @@ impl SetDisks { /// is best-effort maintenance: individual delete failures are logged and /// skipped rather than propagated. pub(crate) async fn reclaim_orphan_data_dirs(&self, bucket: &str, object: &str) -> disk::error::Result { + self.reclaim_orphan_data_dirs_inner(bucket, object, false).await + } + + pub(crate) async fn dry_run_reclaim_orphan_data_dirs(&self, bucket: &str, object: &str) -> disk::error::Result { + self.reclaim_orphan_data_dirs_inner(bucket, object, true).await + } + + async fn reclaim_orphan_data_dirs_inner(&self, bucket: &str, object: &str, dry_run: bool) -> disk::error::Result { let disks = self.get_disks_internal().await; // Phase 1 (read-only): build the referenced-data-dir union and record the @@ -4967,6 +4975,20 @@ impl SetDisks { continue; } let stray = format!("{object}/{dir}"); + if dry_run { + removed += 1; + debug!( + target: "rustfs_ecstore::set_disk", + event = "heal_abandoned_parts", + component = "ecstore", + subsystem = "heal", + state = "dry_run_matched", + result = "matched", + bucket, object, data_dir = %dir, + "Heal abandoned parts dry-run matched orphaned data directory" + ); + continue; + } match disk .delete( bucket, diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index a04c5db82..5a56ea36e 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -6998,6 +6998,100 @@ mod tests { assert!(object_dir.join(STORAGE_FORMAT_FILE).exists(), "metadata must be preserved"); } + async fn recv_abandoned_parts_trace( + trace: &mut rustfs_common::trace_bus::TraceSubscription, + bucket: &str, + object: &str, + state: &str, + ) -> rustfs_common::trace_bus::TraceEvent { + for _ in 0..32 { + let event = tokio::time::timeout(std::time::Duration::from_secs(1), trace.recv()) + .await + .expect("abandoned-parts trace event should arrive") + .expect("trace bus should stay open"); + if event.kind == rustfs_common::trace_bus::TraceKind::Heal + && event.func == rustfs_common::trace_bus::TraceFunc::HealCheckAbandonedParts + && event.bucket.as_deref() == Some(bucket) + && event.object.as_deref() == Some(object) + && trace_attr_string(&event, "state").as_deref() == Some(state) + { + return (*event).clone(); + } + } + + panic!("expected abandoned-parts trace state {state} for {bucket}/{object}"); + } + + fn trace_attr_string(event: &rustfs_common::trace_bus::TraceEvent, key: &str) -> Option { + event.attrs.iter().find_map(|attr| { + if attr.key != key { + return None; + } + Some(match &attr.value { + rustfs_common::trace_bus::TraceVal::Bool(value) => value.to_string(), + rustfs_common::trace_bus::TraceVal::U64(value) => value.to_string(), + rustfs_common::trace_bus::TraceVal::I64(value) => value.to_string(), + rustfs_common::trace_bus::TraceVal::Str(value) => value.to_string(), + }) + }) + } + + #[tokio::test] + async fn check_abandoned_parts_dry_run_counts_without_deleting() { + let mut trace = rustfs_common::trace_bus::subscribe_trace_events(); + let (dir, disk) = make_single_local_disk().await; + let live = Uuid::new_v4(); + let orphan = Uuid::new_v4(); + + let object_dir = dir.path().join("bucket").join("obj"); + write_object_meta_with_data_dirs(&object_dir, "bucket", "obj", &[live]).await; + fs::create_dir_all(object_dir.join(live.to_string())) + .await + .expect("live data dir should be created"); + fs::create_dir_all(object_dir.join(orphan.to_string())) + .await + .expect("orphan data dir should be created"); + + let set = make_set_disks_with(vec![Some(disk)]).await; + set.check_abandoned_parts( + "bucket", + "obj", + &HealOpts { + dry_run: true, + no_lock: true, + ..Default::default() + }, + ) + .await + .expect("dry-run abandoned-parts check should succeed"); + let dry_run_trace = recv_abandoned_parts_trace(&mut trace, "bucket", "obj", "dry_run_matched").await; + assert_eq!(trace_attr_string(&dry_run_trace, "dry_run").as_deref(), Some("true")); + assert_eq!(trace_attr_string(&dry_run_trace, "data_dirs").as_deref(), Some("1")); + + assert!(object_dir.join(live.to_string()).exists(), "referenced data dir must be preserved"); + assert!(object_dir.join(orphan.to_string()).exists(), "dry-run must not remove orphaned data dir"); + + set.check_abandoned_parts( + "bucket", + "obj", + &HealOpts { + no_lock: true, + ..Default::default() + }, + ) + .await + .expect("abandoned-parts check should reclaim stale data dir"); + let reclaim_trace = recv_abandoned_parts_trace(&mut trace, "bucket", "obj", "reclaimed").await; + assert_eq!(trace_attr_string(&reclaim_trace, "dry_run").as_deref(), Some("false")); + assert_eq!(trace_attr_string(&reclaim_trace, "data_dirs").as_deref(), Some("1")); + + assert!( + object_dir.join(live.to_string()).exists(), + "referenced data dir must remain after reclaim" + ); + assert!(!object_dir.join(orphan.to_string()).exists(), "orphaned data dir must be removed"); + } + #[tokio::test] async fn reclaim_orphan_data_dirs_recovers_deferred_cleanup_after_restart() { let (dir, disk) = make_single_local_disk().await; @@ -12233,11 +12327,18 @@ mod tests { .expect_err("unsupported copy_object_part should return a typed error"); assert!(matches!(copy_part_err, StorageError::NotImplemented)); - let abandoned_err = set_disks - .check_abandoned_parts("bucket", "object", &HealOpts::default()) + set_disks + .check_abandoned_parts( + "bucket", + "object", + &HealOpts { + dry_run: true, + no_lock: true, + ..Default::default() + }, + ) .await - .expect_err("abandoned-parts check should stay in the upper reconciliation layer"); - assert!(matches!(abandoned_err, StorageError::NotImplemented)); + .expect("abandoned-parts check should be callable on empty disk sets"); } #[tokio::test] diff --git a/crates/ecstore/src/set_disk/ops/heal.rs b/crates/ecstore/src/set_disk/ops/heal.rs index 274b33f43..de688fbdb 100644 --- a/crates/ecstore/src/set_disk/ops/heal.rs +++ b/crates/ecstore/src/set_disk/ops/heal.rs @@ -16,6 +16,7 @@ use super::super::*; use crate::disk::disk_store::DiskStoreRenameDataExt; use crate::io_support::bitrot::object_mmap_read_enabled; use crate::storage_api_contracts::namespace::NamespaceLocking as _; +use rustfs_common::trace_bus::{TraceEvent, TraceFunc, TraceKind, trace_emit}; use tracing::trace; const LOG_COMPONENT_ECSTORE: &str = "ecstore"; @@ -2057,11 +2058,61 @@ impl crate::storage_api_contracts::heal::HealOperations for SetDisks { Err(Error::DiskNotFound) } - #[tracing::instrument(skip(self))] - async fn check_abandoned_parts(&self, _bucket: &str, _object: &str, _opts: &HealOpts) -> Result<()> { - // Multipart orphan reconciliation is intentionally retained above the set layer - // until there is a concrete caller and a stable lower-level contract to implement. - Err(StorageError::NotImplemented) + #[tracing::instrument(level = "debug", skip(self, opts), fields(bucket = %bucket, object = %object, dry_run = opts.dry_run))] + async fn check_abandoned_parts(&self, bucket: &str, object: &str, opts: &HealOpts) -> Result<()> { + let started_at = std::time::Instant::now(); + let _write_lock_guard = if !opts.no_lock { + let ns_lock = self.new_ns_lock(bucket, object).await?; + Some( + ns_lock + .get_write_lock(get_lock_acquire_timeout()) + .await + .map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?, + ) + } else { + None + }; + + let removed = if opts.dry_run { + self.dry_run_reclaim_orphan_data_dirs(bucket, object).await? + } else { + self.reclaim_orphan_data_dirs(bucket, object).await? + }; + let state = if opts.dry_run && removed > 0 { + "dry_run_matched" + } else if removed > 0 { + "reclaimed" + } else { + "checked" + }; + let data_dirs = u64::try_from(removed).unwrap_or(u64::MAX); + + trace_emit(|| { + TraceEvent::new(TraceKind::Heal, TraceFunc::HealCheckAbandonedParts) + .with_bucket(bucket) + .with_object(object) + .with_duration(started_at.elapsed()) + .with_attr("state", state) + .with_attr("dry_run", opts.dry_run) + .with_attr("data_dirs", data_dirs) + }); + + if removed > 0 { + trace!( + event = "heal_abandoned_parts", + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_HEAL, + state = if opts.dry_run { "dry_run_matched" } else { "reclaimed" }, + result = "ok", + bucket, + object, + dry_run = opts.dry_run, + data_dirs = removed, + "Heal abandoned parts checked object data directories" + ); + } + + Ok(()) } } diff --git a/crates/ecstore/src/set_disk/ops/heal_walk.rs b/crates/ecstore/src/set_disk/ops/heal_walk.rs index a39a9abe7..ea1ea66cf 100644 --- a/crates/ecstore/src/set_disk/ops/heal_walk.rs +++ b/crates/ecstore/src/set_disk/ops/heal_walk.rs @@ -23,6 +23,7 @@ //! per-version `SetDisks::heal_object`. use super::super::*; +use crate::object_api::ObjectInfo; use std::collections::HashSet; use std::sync::Mutex; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; @@ -39,12 +40,16 @@ const BACKGROUND_WALKDIR_STALL_TIMEOUT: Duration = Duration::from_secs(60); /// it must not gate healing logic — the delete-marker vs data path is chosen /// inside `ops/heal.rs` from the resolved latest metadata. `version_id` is /// normalized (nil/absent UUID => `None`). -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone)] pub struct HealWalkVersion { /// object key pub name: String, /// normalized version id (`None` when the version is nil/absent) pub version_id: Option, + /// version modification time as Unix nanoseconds + pub mod_time_unix_nanos: Option, + /// object snapshot for lifecycle evaluation + pub lifecycle_object_info: Option, /// whether this version is a delete marker (observability only) pub is_delete_marker: bool, } @@ -63,6 +68,7 @@ struct HealWalkCollector { bucket: String, batch_objects: usize, version_budget: usize, + include_lifecycle_object_info: bool, objects: Mutex>, decode_error: Mutex>, version_total: AtomicUsize, @@ -116,10 +122,25 @@ impl HealWalkCollector { let mut versions = Vec::with_capacity(fiv.versions.len() + fiv.free_versions.len()); for fi in fiv.versions.iter().chain(fiv.free_versions.iter()) { + let version_uuid = fi.version_id.filter(|version_id| !version_id.is_nil()); + let lifecycle_object_info = if self.include_lifecycle_object_info { + let mut lifecycle_fi = fi.clone(); + lifecycle_fi.version_id = version_uuid; + Some(ObjectInfo::from_file_info( + &lifecycle_fi, + &self.bucket, + &entry.name, + version_uuid.is_some(), + )) + } else { + None + }; versions.push(HealWalkVersion { name: entry.name.clone(), // Normalize: nil/absent version id => None. - version_id: fi.version_id.filter(|u| !u.is_nil()).map(|u| u.to_string()), + version_id: version_uuid.map(|u| u.to_string()), + mod_time_unix_nanos: fi.mod_time.map(|mod_time| mod_time.unix_timestamp_nanos()), + lifecycle_object_info, is_delete_marker: fi.deleted, }); } @@ -173,11 +194,26 @@ impl HealWalkCollector { } }; for fi in fiv.versions.iter().chain(fiv.free_versions.iter()) { - let vid = fi.version_id.filter(|u| !u.is_nil()).map(|u| u.to_string()); + let version_uuid = fi.version_id.filter(|version_id| !version_id.is_nil()); + let vid = version_uuid.map(|u| u.to_string()); if seen.insert(vid.clone()) { + let lifecycle_object_info = if self.include_lifecycle_object_info { + let mut lifecycle_fi = fi.clone(); + lifecycle_fi.version_id = version_uuid; + Some(ObjectInfo::from_file_info( + &lifecycle_fi, + &self.bucket, + &entry.name, + version_uuid.is_some(), + )) + } else { + None + }; versions.push(HealWalkVersion { name: entry.name.clone(), version_id: vid, + mod_time_unix_nanos: fi.mod_time.map(|mod_time| mod_time.unix_timestamp_nanos()), + lifecycle_object_info, is_delete_marker: fi.deleted, }); } @@ -255,6 +291,7 @@ impl SetDisks { forward_to: Option<&str>, batch_objects: usize, version_budget: usize, + include_lifecycle_object_info: bool, ) -> disk::error::Result<(Vec, Option, bool)> { assert!(batch_objects >= 2, "heal_walk_versions_page requires batch_objects >= 2"); @@ -264,6 +301,7 @@ impl SetDisks { bucket: bucket.to_string(), batch_objects, version_budget: version_budget.max(1), + include_lifecycle_object_info, objects: Mutex::new(Vec::new()), decode_error: Mutex::new(None), version_total: AtomicUsize::new(0), @@ -347,6 +385,7 @@ mod tests { bucket: "bucket".to_string(), batch_objects: 2, version_budget: 2, + include_lifecycle_object_info: false, objects: Mutex::new(Vec::new()), decode_error: Mutex::new(None), version_total: AtomicUsize::new(0), @@ -388,6 +427,8 @@ mod tests { HealWalkVersion { name: name.to_string(), version_id: Some(id.to_string()), + mod_time_unix_nanos: None, + lifecycle_object_info: None, is_delete_marker: dm, } } @@ -491,6 +532,7 @@ mod tests { bucket: "bucket".to_string(), batch_objects: 1000, version_budget: 10_000, + include_lifecycle_object_info: false, objects: Mutex::new(Vec::new()), version_total: AtomicUsize::new(0), decode_error: Mutex::new(None), @@ -567,7 +609,7 @@ mod tests { .expect("corrupt test metadata should be written"); let error = set_disks - .heal_walk_versions_page(bucket, "", None, 2, 2) + .heal_walk_versions_page(bucket, "", None, 2, 2, false) .await .expect_err("semantic metadata corruption must fail the heal disk walk"); diff --git a/crates/ecstore/src/store/heal.rs b/crates/ecstore/src/store/heal.rs index 3efa4ed5f..d10abe740 100644 --- a/crates/ecstore/src/store/heal.rs +++ b/crates/ecstore/src/store/heal.rs @@ -18,6 +18,7 @@ use tracing::trace; const LOG_COMPONENT_ECSTORE: &str = "ecstore"; const LOG_SUBSYSTEM_HEAL: &str = "heal"; +const EVENT_HEAL_ABANDONED_PARTS: &str = "heal_abandoned_parts"; const EVENT_HEAL_FORMAT_COMPLETED: &str = "heal_format_completed"; const EVENT_HEAL_OBJECT_STARTED: &str = "heal_object_started"; @@ -256,13 +257,40 @@ impl ECStore { #[instrument(skip(self))] pub(super) async fn handle_check_abandoned_parts(&self, bucket: &str, object: &str, opts: &HealOpts) -> Result<()> { - let _ = (bucket, object, opts); - // Stale multipart reconciliation is already owned by the lifecycle-driven - // background cleanup path in `bucket_lifecycle_ops.rs`. There is currently - // no stable object-heal contract that should fan this request out through - // pool/set storage layers, so keep the placeholder explicit at the ECStore - // boundary instead of dispatching into lower layers. - Err(StorageError::NotImplemented) + let object = encode_dir_object(object); + let pools = self.get_pools_for_heal_object(opts)?; + + let mut futures = Vec::with_capacity(pools.len()); + for pool in pools.iter() { + futures.push(pool.check_abandoned_parts(bucket, &object, opts)); + } + + let mut first_error = None; + for result in join_all(futures).await { + if let Err(err) = result + && first_error.is_none() + { + first_error = Some(err); + } + } + + if let Some(err) = first_error { + return Err(err); + } + + trace!( + event = EVENT_HEAL_ABANDONED_PARTS, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_HEAL, + state = "completed", + result = "ok", + bucket, + object, + dry_run = opts.dry_run, + "Heal abandoned parts completed" + ); + + Ok(()) } } diff --git a/crates/ecstore/src/store/heal_walk.rs b/crates/ecstore/src/store/heal_walk.rs index 39d50007d..191e87e54 100644 --- a/crates/ecstore/src/store/heal_walk.rs +++ b/crates/ecstore/src/store/heal_walk.rs @@ -34,6 +34,7 @@ impl ECStore { forward_to: Option<&str>, batch_objects: usize, version_budget: usize, + include_lifecycle_object_info: bool, ) -> Result<(Vec, Option, bool)> { if pool_idx >= self.pools.len() || set_idx >= self.pools[pool_idx].disk_set.len() { return Err(Error::other(format!( @@ -43,7 +44,7 @@ impl ECStore { } self.pools[pool_idx].disk_set[set_idx] - .heal_walk_versions_page(bucket, prefix, forward_to, batch_objects, version_budget) + .heal_walk_versions_page(bucket, prefix, forward_to, batch_objects, version_budget, include_lifecycle_object_info) .await .map_err(Error::from) } diff --git a/crates/heal/src/heal/channel.rs b/crates/heal/src/heal/channel.rs index d2be2545d..23cf2f168 100644 --- a/crates/heal/src/heal/channel.rs +++ b/crates/heal/src/heal/channel.rs @@ -767,6 +767,7 @@ mod tests { _bucket: &str, _prefix: &str, _continuation_token: Option<&str>, + _include_lifecycle_object_info: bool, ) -> crate::Result<(Vec, Option, bool)> { Ok((vec![], None, false)) } diff --git a/crates/heal/src/heal/erasure_healer.rs b/crates/heal/src/heal/erasure_healer.rs index 8f1810bb6..04075837c 100644 --- a/crates/heal/src/heal/erasure_healer.rs +++ b/crates/heal/src/heal/erasure_healer.rs @@ -23,13 +23,14 @@ use crate::heal::{ }; use crate::{Error, Result}; use futures::{StreamExt, stream::FuturesUnordered}; -use metrics::gauge; +use metrics::{counter, gauge}; use rustfs_common::heal_channel::{HealOpts, HealRequestSource, HealScanMode}; use rustfs_madmin::heal_commands::HealResultItem; use std::sync::{ Arc, atomic::{AtomicUsize, Ordering}, }; +use std::time::{Duration, UNIX_EPOCH}; use tokio::sync::{RwLock, Semaphore}; use tracing::{debug, error, warn}; @@ -47,6 +48,21 @@ enum HealObjectOutcome { Failed, } +fn result_object_size_u64(result: &HealResultItem) -> u64 { + u64::try_from(result.object_size).unwrap_or(u64::MAX) +} + +const NEW_VERSION_SKIP_GRACE_SECS: u64 = 60; +const NANOS_PER_SECOND: i128 = 1_000_000_000; + +fn should_skip_new_version(mod_time_unix_nanos: Option, started_at_secs: u64) -> bool { + let Some(mod_time_unix_nanos) = mod_time_unix_nanos else { + return false; + }; + let cutoff_secs = started_at_secs.saturating_add(NEW_VERSION_SKIP_GRACE_SECS); + mod_time_unix_nanos > i128::from(cutoff_secs).saturating_mul(NANOS_PER_SECOND) +} + struct PageConcurrencyGuard { in_flight: Arc, set_label: String, @@ -492,6 +508,7 @@ impl ErasureSetHealer { &mut skipped_objects, resume_manager, checkpoint_manager, + state.start_time, ) .await; @@ -658,6 +675,7 @@ impl ErasureSetHealer { skipped_objects: &mut u64, resume_manager: &ResumeManager, checkpoint_manager: &CheckpointManager, + started_at_secs: u64, ) -> Result<()> { debug!( target: "rustfs::heal::erasure_healer", @@ -710,6 +728,7 @@ impl ErasureSetHealer { // The end-of-pass summary reports the full failed/skipped counts. let mut transient_skip_samples_logged = 0_u64; let mut failure_samples_logged = 0_u64; + let mut bytes_processed = self.progress.read().await.bytes_processed; // backlog#920: select the per-erasure-set DISK-WALK union enumerator when // the scan is Deep OR the request came from AutoHeal — these are the paths @@ -718,17 +737,25 @@ impl ErasureSetHealer { // which stays the default. let use_disk_walk = matches!(self.heal_opts.scan_mode, HealScanMode::Deep) || matches!(self.source, HealRequestSource::AutoHeal); + let lifecycle_expiry_context = self.storage.load_heal_lifecycle_expiry_context(bucket).await?; + let include_lifecycle_object_info = lifecycle_expiry_context.is_some(); loop { self.verify_replacement_identity_fence("page scan").await?; // Get one page of object versions let (objects, next_token, is_truncated) = if use_disk_walk { self.storage - .list_versions_for_heal_page_disk_walk(set_disk_id, bucket, "", continuation_token.as_deref()) + .list_versions_for_heal_page_disk_walk( + set_disk_id, + bucket, + "", + continuation_token.as_deref(), + include_lifecycle_object_info, + ) .await? } else { self.storage - .list_objects_for_heal_page(bucket, "", continuation_token.as_deref()) + .list_objects_for_heal_page(bucket, "", continuation_token.as_deref(), include_lifecycle_object_info) .await? }; let page_is_empty = objects.is_empty(); @@ -736,6 +763,7 @@ impl ErasureSetHealer { let page_resume_index = *current_object_index; let semaphore = Arc::new(Semaphore::new(page_concurrency_limit)); let mut page_tasks = FuturesUnordered::new(); + let mut completed_in_page = 0usize; // Capture the last version identity of this page for the anti-loop guard. let page_last = objects.last().map(|item| (item.name.clone(), item.version_id.clone())); @@ -751,6 +779,75 @@ impl ErasureSetHealer { continue; } + if should_skip_new_version(item.mod_time_unix_nanos, started_at_secs) { + checkpoint_manager.add_processed_object(key).await?; + *processed_objects = processed_objects.saturating_add(1); + completed_in_page = completed_in_page.saturating_add(1); + counter!("rustfs_heal_skipped_new_versions_total").increment(1); + { + let mut progress = self.progress.write().await; + progress.record_skipped_new_version(); + progress.set_current_object(Some(format!("skipped_new: {bucket}/{}", item.name))); + progress.update_progress(*processed_objects, *successful_objects, *failed_objects, bytes_processed); + } + debug!( + target: "rustfs::heal::erasure_healer", + event = EVENT_HEAL_ERASURE_OBJECT_STATE, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_ERASURE_HEALER, + set_disk_id, + bucket, + object = %item.name, + version_id = ?item.version_id, + state = "skipped_new_version", + "Erasure set object version skipped because it was written after heal started" + ); + if completed_in_page.is_multiple_of(100) { + checkpoint_manager.update_position(bucket_index, page_resume_index).await?; + } + continue; + } + + if let Some(context) = lifecycle_expiry_context.as_ref() + && self + .storage + .enqueue_heal_lifecycle_expiry( + context, + bucket, + &item.name, + item.version_id.as_deref(), + item.lifecycle_object_info.as_ref(), + ) + .await? + { + checkpoint_manager.add_processed_object(key).await?; + *processed_objects = processed_objects.saturating_add(1); + completed_in_page = completed_in_page.saturating_add(1); + counter!("rustfs_heal_skipped_ilm_expired_total").increment(1); + { + let mut progress = self.progress.write().await; + progress.record_skipped_ilm_expired(); + progress.set_current_object(Some(format!("skipped_ilm: {bucket}/{}", item.name))); + progress.update_progress(*processed_objects, *successful_objects, *failed_objects, bytes_processed); + } + debug!( + target: "rustfs::heal::erasure_healer", + event = EVENT_HEAL_ERASURE_OBJECT_STATE, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_ERASURE_HEALER, + set_disk_id, + bucket, + object = %item.name, + version_id = ?item.version_id, + state = "skipped_ilm_expired", + "Erasure set object version skipped because lifecycle expiry was queued" + ); + if completed_in_page.is_multiple_of(100) { + checkpoint_manager.update_position(bucket_index, page_resume_index).await?; + } + continue; + } + resume_manager .set_current_item(Some(bucket.to_string()), Some(item.name.clone())) .await?; @@ -777,7 +874,7 @@ impl ErasureSetHealer { let _permit = match permit { Ok(permit) => permit, - Err(err) => return (dedup_key, object_name, version_id, Err(err)), + Err(err) => return (dedup_key, object_name, version_id, (0, Err(err))), }; let _in_flight_guard = PageConcurrencyGuard::new(in_flight, set_label); @@ -788,7 +885,7 @@ impl ErasureSetHealer { // recorded as skipped-ok rather than failed. The delete-marker // vs data path is chosen internally in ops/heal.rs. let result = if cancel_token.is_cancelled() { - Err(Error::TaskCancelled) + (0, Err(Error::TaskCancelled)) } else { match storage .heal_object(&bucket_name, &object_name, version_id.as_deref(), &heal_opts) @@ -797,8 +894,9 @@ impl ErasureSetHealer { Ok((result, None)) if target_outcomes_complete(&result, &target_endpoints) => { + let object_size = result_object_size_u64(&result); if !replacement_commit_evidence_required { - Ok(true) + (object_size, Ok(true)) } else { match storage .replacement_targets_have_version( @@ -810,27 +908,42 @@ impl ErasureSetHealer { ) .await { - Ok(true) => Ok(true), - Ok(false) => Err(Error::transient_skip(format!( + Ok(true) => (object_size, Ok(true)), + Ok(false) => (object_size, Err(Error::transient_skip(format!( "Skipped heal for {bucket_name}/{object_name} because replacement target readback did not confirm the committed version" - ))), - Err(err) => Err(Error::transient_skip(format!( + )))), + Err(err) => (object_size, Err(Error::transient_skip(format!( "Skipped heal for {bucket_name}/{object_name} because replacement target readback failed: {err}" - ))), + )))), } } - } - Ok((_result, None)) if !target_endpoints.is_empty() => Err(Error::transient_skip(format!( - "Skipped heal for {bucket_name}/{object_name} because a replacement target was not committed" - ))), - Ok((_result, None)) => Ok(true), - Ok((_, Some(err))) if is_missing_object_dir_heal_result(&object_name, &err) => Ok(false), - Ok((_, Some(err))) | Err(err) => match Self::classify_heal_object_error(&err) { - HealObjectOutcome::Absent => Ok(false), - HealObjectOutcome::Transient => Err(Error::transient_skip(format!( - "Skipped heal for {bucket_name}/{object_name} due to transient error: {err}" + }, + Ok((result, None)) if !target_endpoints.is_empty() => ( + result_object_size_u64(&result), + Err(Error::transient_skip(format!( + "Skipped heal for {bucket_name}/{object_name} because a replacement target was not committed" ))), - HealObjectOutcome::Failed => Err(err), + ), + Ok((result, None)) => (result_object_size_u64(&result), Ok(true)), + Ok((result, Some(err))) if is_missing_object_dir_heal_result(&object_name, &err) => { + (result_object_size_u64(&result), Ok(false)) + } + Ok((result, Some(err))) => { + let object_size = result_object_size_u64(&result); + match Self::classify_heal_object_error(&err) { + HealObjectOutcome::Absent => (object_size, Ok(false)), + HealObjectOutcome::Transient => (object_size, Err(Error::transient_skip(format!( + "Skipped heal for {bucket_name}/{object_name} due to transient error: {err}" + )))), + HealObjectOutcome::Failed => (object_size, Err(err)), + } + } + Err(err) => match Self::classify_heal_object_error(&err) { + HealObjectOutcome::Absent => (0, Ok(false)), + HealObjectOutcome::Transient => (0, Err(Error::transient_skip(format!( + "Skipped heal for {bucket_name}/{object_name} due to transient error: {err}" + )))), + HealObjectOutcome::Failed => (0, Err(err)), }, } }; @@ -839,11 +952,12 @@ impl ErasureSetHealer { }); } - let mut completed_in_page = 0usize; while let Some((key, object, version_id, result)) = page_tasks.next().await { + let (object_size, result) = result; match result { Ok(true) => { *successful_objects += 1; + bytes_processed = bytes_processed.saturating_add(object_size); checkpoint_manager.add_processed_object(key).await?; debug!( target: "rustfs::heal::erasure_healer", @@ -861,6 +975,7 @@ impl ErasureSetHealer { Ok(false) => { checkpoint_manager.add_processed_object(key).await?; *successful_objects += 1; + bytes_processed = bytes_processed.saturating_add(object_size); debug!( target: "rustfs::heal::erasure_healer", event = EVENT_HEAL_ERASURE_OBJECT_STATE, @@ -877,6 +992,7 @@ impl ErasureSetHealer { Err(err @ Error::TaskCancelled) | Err(err @ Error::TaskTimeout) => return Err(err), Err(Error::TransientSkip { message }) => { *skipped_objects += 1; + bytes_processed = bytes_processed.saturating_add(object_size); checkpoint_manager.add_skipped_object(key).await?; demote_to_debug_when!(!take_failure_log_sample(&mut transient_skip_samples_logged), warn, target: "rustfs::heal::erasure_healer", { event = EVENT_HEAL_ERASURE_OBJECT_STATE, @@ -893,6 +1009,7 @@ impl ErasureSetHealer { } Err(err) => { *failed_objects += 1; + bytes_processed = bytes_processed.saturating_add(object_size); checkpoint_manager.add_failed_object(key).await?; demote_to_debug_when!(!take_failure_log_sample(&mut failure_samples_logged), warn, target: "rustfs::heal::erasure_healer", { event = EVENT_HEAL_ERASURE_OBJECT_STATE, @@ -911,6 +1028,11 @@ impl ErasureSetHealer { *processed_objects += 1; completed_in_page += 1; + { + let mut progress = self.progress.write().await; + progress.set_current_object(Some(format!("{bucket}/{object}"))); + progress.update_progress(*processed_objects, *successful_objects, *failed_objects, bytes_processed); + } if completed_in_page.is_multiple_of(100) { checkpoint_manager.update_position(bucket_index, page_resume_index).await?; @@ -964,7 +1086,9 @@ impl ErasureSetHealer { progress.objects_scanned = state.total_objects; progress.objects_healed = state.successful_objects; progress.objects_failed = state.failed_objects; - progress.bytes_processed = 0; // set to 0 for now, can be extended later + progress.bytes_processed = 0; // Resume state tracks object counts, not byte counters. + progress.start_time = UNIX_EPOCH.checked_add(Duration::from_secs(state.start_time)); + progress.last_update_time = UNIX_EPOCH.checked_add(Duration::from_secs(state.last_update)); progress.set_current_object(state.current_object.clone()); } } @@ -1135,13 +1259,15 @@ mod resume_loop_tests { //! that emits programmable multi-version pages. These exercise the real loop //! logic (cursor seeding, per-version dedup, anti-loop guard, absence //! handling) — not merely a mock's own output. - use super::{ErasureSetHealer, target_outcomes_complete}; + use super::{ + ErasureSetHealer, NANOS_PER_SECOND, NEW_VERSION_SKIP_GRACE_SECS, should_skip_new_version, target_outcomes_complete, + }; use crate::heal::progress::HealProgress; use crate::heal::resume::{ CheckpointManager, RESUME_CHECKPOINT_FILE, ReplacementTargetIdentity, ResumeDeleteFailure, ResumeManager, ResumeUtils, compose_key, }; - use crate::heal::storage::{DiskStatus, HealListItem, HealObjectInfo, HealStorageAPI}; + use crate::heal::storage::{DiskStatus, HealLifecycleExpiryContext, HealListItem, HealObjectInfo, HealStorageAPI}; use crate::heal::storage_api::status::BucketInfo; use crate::heal::{ BUCKET_META_PREFIX, DiskOption, DiskStore, EcstoreError, Endpoint, HealDiskExt as _, RUSTFS_META_BUCKET, new_disk, @@ -1149,7 +1275,7 @@ mod resume_loop_tests { use crate::{Error, Result}; use rustfs_common::heal_channel::{HealOpts, HealRequestSource}; use rustfs_madmin::heal_commands::{HealDriveInfo, HealResultItem, Infos}; - use std::collections::{HashMap, VecDeque}; + use std::collections::{HashMap, HashSet, VecDeque}; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; use tempfile::TempDir; @@ -1160,10 +1286,37 @@ mod resume_loop_tests { HealListItem { name: name.to_string(), version_id: version.map(str::to_string), + mod_time_unix_nanos: None, + lifecycle_object_info: None, is_delete_marker: delete_marker, } } + fn item_with_mod_time(name: &str, version: Option<&str>, mod_time_secs: u64) -> HealListItem { + HealListItem { + name: name.to_string(), + version_id: version.map(str::to_string), + mod_time_unix_nanos: Some(i128::from(mod_time_secs).saturating_mul(NANOS_PER_SECOND)), + lifecycle_object_info: None, + is_delete_marker: false, + } + } + + #[test] + fn new_version_filter_respects_grace_boundary() { + let started_at = 1_700_000_000; + + assert!(!should_skip_new_version(None, started_at)); + assert!(!should_skip_new_version( + Some(i128::from(started_at + NEW_VERSION_SKIP_GRACE_SECS).saturating_mul(NANOS_PER_SECOND)), + started_at, + )); + assert!(should_skip_new_version( + Some(i128::from(started_at + NEW_VERSION_SKIP_GRACE_SECS + 1).saturating_mul(NANOS_PER_SECOND)), + started_at, + )); + } + #[test] fn target_outcomes_require_each_requested_endpoint_once_and_ok() { let result = HealResultItem { @@ -1246,8 +1399,10 @@ mod resume_loop_tests { /// Target-specific physical readback evidence per `compose_key`; the /// fake models a healthy backend unless a test explicitly revokes it. replacement_commit_evidence: Mutex>, + lifecycle_expired: Mutex>, /// every heal_object call recorded as (name, version_id) heal_calls: Mutex)>>, + list_include_lifecycle_object_info: Mutex>, replacement_target_identity_sequences: Mutex>>, fail_listing: AtomicBool, } @@ -1274,9 +1429,15 @@ mod resume_loop_tests { .unwrap() .insert(compose_key(name, version), ReplacementCommitEvidence::Error(message.to_string())); } + fn set_lifecycle_expired(&self, name: &str, version: Option<&str>) { + self.lifecycle_expired.lock().unwrap().insert(compose_key(name, version)); + } fn calls(&self) -> Vec<(String, Option)> { self.heal_calls.lock().unwrap().clone() } + fn list_include_lifecycle_object_info_calls(&self) -> Vec { + self.list_include_lifecycle_object_info.lock().unwrap().clone() + } fn fail_listing(&self) { self.fail_listing.store(true, Ordering::SeqCst); } @@ -1330,6 +1491,23 @@ mod resume_loop_tests { async fn get_object_checksum(&self, _b: &str, _o: &str) -> Result> { Ok(None) } + async fn load_heal_lifecycle_expiry_context(&self, _bucket: &str) -> Result> { + Ok((!self.lifecycle_expired.lock().unwrap().is_empty()).then(HealLifecycleExpiryContext::test)) + } + async fn enqueue_heal_lifecycle_expiry( + &self, + _context: &HealLifecycleExpiryContext, + _bucket: &str, + object: &str, + version_id: Option<&str>, + _object_info: Option<&HealObjectInfo>, + ) -> Result { + Ok(self + .lifecycle_expired + .lock() + .unwrap() + .contains(&compose_key(object, version_id))) + } async fn heal_object( &self, _bucket: &str, @@ -1386,7 +1564,12 @@ mod resume_loop_tests { _bucket: &str, _prefix: &str, continuation_token: Option<&str>, + include_lifecycle_object_info: bool, ) -> Result<(Vec, Option, bool)> { + self.list_include_lifecycle_object_info + .lock() + .unwrap() + .push(include_lifecycle_object_info); if self.fail_listing.load(Ordering::SeqCst) { return Err(Error::other("injected listing failure")); } @@ -1476,6 +1659,7 @@ mod resume_loop_tests { /// Drive one bucket heal pass; returns (processed, successful, failed, skipped, result). async fn run(env: &Env) -> (u64, u64, u64, u64, Result<()>) { + let state = env.resume.get_state().await; let mut current_object_index = 0usize; let mut processed = 0u64; let mut successful = 0u64; @@ -1494,6 +1678,7 @@ mod resume_loop_tests { &mut skipped, &env.resume, &env.checkpoint, + state.start_time, ) .await; (processed, successful, failed, skipped, result) @@ -1559,6 +1744,7 @@ mod resume_loop_tests { let mut successful = 0; let mut failed = 0; let mut skipped = 0; + let started_at = env.resume.get_state().await.start_time; let error = healer .heal_bucket_with_resume( @@ -1572,6 +1758,7 @@ mod resume_loop_tests { &mut skipped, &env.resume, &env.checkpoint, + started_at, ) .await .expect_err("a remounted target must not begin a new page scan"); @@ -1641,6 +1828,109 @@ mod resume_loop_tests { assert_eq!(skipped, 0); } + #[tokio::test] + async fn erasure_set_progress_accumulates_healed_object_bytes() { + let env = make_env().await; + env.storage.set_page( + None, + Page { + items: vec![item("first", Some("v1"), false), item("second", Some("v2"), false)], + next: None, + truncated: false, + }, + ); + env.storage.set_result( + "first", + Some("v1"), + HealResultItem { + object_size: 1024, + ..Default::default() + }, + ); + env.storage.set_result( + "second", + Some("v2"), + HealResultItem { + object_size: 2048, + ..Default::default() + }, + ); + + let (processed, successful, failed, skipped, result) = run(&env).await; + + result.expect("page heal should succeed"); + assert_eq!(processed, 2); + assert_eq!(successful, 2); + assert_eq!(failed, 0); + assert_eq!(skipped, 0); + let progress = env.healer.progress.read().await; + assert_eq!(progress.objects_scanned, 2); + assert_eq!(progress.objects_healed, 2); + assert_eq!(progress.objects_failed, 0); + assert_eq!(progress.bytes_processed, 3072); + assert!(matches!(progress.current_object.as_deref(), Some("b/first" | "b/second"))); + } + + #[tokio::test] + async fn erasure_set_skips_versions_written_after_heal_started() { + let env = make_env().await; + let started_at = env.resume.get_state().await.start_time; + env.storage.set_page( + None, + Page { + items: vec![ + item_with_mod_time("old", Some("v1"), started_at + NEW_VERSION_SKIP_GRACE_SECS), + item_with_mod_time("new", Some("v2"), started_at + NEW_VERSION_SKIP_GRACE_SECS + 1), + ], + next: None, + truncated: false, + }, + ); + + let (processed, successful, failed, skipped, result) = run(&env).await; + + result.expect("page heal should succeed"); + assert_eq!(processed, 2); + assert_eq!(successful, 1); + assert_eq!(failed, 0); + assert_eq!(skipped, 0); + assert_eq!(env.storage.calls(), vec![("old".to_string(), Some("v1".to_string()))]); + let progress = env.healer.progress.read().await; + assert_eq!(progress.skipped_new_versions, 1); + assert_eq!(progress.objects_scanned, 2); + assert_eq!(progress.objects_healed, 1); + assert_eq!(progress.objects_failed, 0); + } + + #[tokio::test] + async fn erasure_set_skips_versions_queued_for_lifecycle_expiry() { + let env = make_env().await; + env.storage.set_page( + None, + Page { + items: vec![item("expired", Some("v1"), false), item("kept", Some("v2"), false)], + next: None, + truncated: false, + }, + ); + env.storage.set_lifecycle_expired("expired", Some("v1")); + + let (processed, successful, failed, skipped, result) = run(&env).await; + + result.expect("page heal should succeed"); + assert_eq!(processed, 2); + assert_eq!(successful, 1); + assert_eq!(failed, 0); + assert_eq!(skipped, 0); + assert_eq!(env.storage.calls(), vec![("kept".to_string(), Some("v2".to_string()))]); + assert_eq!(env.storage.list_include_lifecycle_object_info_calls(), vec![true]); + let progress = env.healer.progress.read().await; + assert_eq!(progress.skipped_ilm_expired, 1); + assert_eq!(progress.objects_scanned, 2); + assert_eq!(progress.objects_healed, 1); + assert_eq!(progress.objects_failed, 0); + } + #[tokio::test] async fn bucket_listing_failure_does_not_mark_set_completed() { let env = make_env().await; diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index 79ba6b169..66b8f637f 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -2385,8 +2385,27 @@ impl HealManager { 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.skipped_new_versions = snapshot.skipped_new_versions.saturating_add(progress.skipped_new_versions); + snapshot.skipped_ilm_expired = snapshot.skipped_ilm_expired.saturating_add(progress.skipped_ilm_expired); + snapshot.objects_total_count = snapshot.objects_total_count.saturating_add(progress.objects_total_count); + snapshot.objects_total_size = snapshot.objects_total_size.saturating_add(progress.objects_total_size); snapshot.bytes_processed = snapshot.bytes_processed.saturating_add(progress.bytes_processed); + snapshot.start_time = match (snapshot.start_time, progress.start_time) { + (Some(current), Some(next)) => Some(current.min(next)), + (None, next) => next, + (current, None) => current, + }; + snapshot.last_update_time = match (snapshot.last_update_time, progress.last_update_time) { + (Some(current), Some(next)) => Some(current.max(next)), + (None, next) => next, + (current, None) => current, + }; + if progress.current_object.is_some() { + snapshot.current_object = progress.current_object; + } } + snapshot.refresh_progress_percentage(); + snapshot.refresh_estimated_completion_time(); Some(snapshot) } @@ -3208,6 +3227,7 @@ impl HealManager { } else { completed_task.get_status().await }; + let completed_progress = completed_task.get_progress().await; let completed_status_entry = CompletedHealStatus { heal_type: completed_task.heal_type.clone(), status: completed_status.clone(), @@ -3223,6 +3243,7 @@ impl HealManager { match completed_status { HealTaskStatus::Completed => { stats.update_task_completion(true); + stats.add_healed_objects(completed_progress.objects_healed, completed_progress.bytes_processed); } HealTaskStatus::Retrying { .. } => {} _ => { @@ -3749,6 +3770,7 @@ mod tests { _bucket: &str, _prefix: &str, _continuation_token: Option<&str>, + _include_lifecycle_object_info: bool, ) -> Result<(Vec, Option, bool)> { Ok((Vec::new(), None, false)) } @@ -5396,6 +5418,8 @@ mod tests { )); { let mut progress = first.progress.write().await; + progress.start_time = Some(SystemTime::now() - Duration::from_secs(20)); + progress.set_total_baseline(12, 8192); progress.update_progress(7, 3, 1, 4096); } @@ -5405,6 +5429,8 @@ mod tests { )); { let mut progress = second.progress.write().await; + progress.start_time = Some(SystemTime::now() - Duration::from_secs(10)); + progress.set_total_baseline(8, 4096); progress.update_progress(11, 5, 2, 2048); } @@ -5419,7 +5445,11 @@ mod tests { assert_eq!(progress.objects_scanned, 18); assert_eq!(progress.objects_healed, 8); assert_eq!(progress.objects_failed, 3); + assert_eq!(progress.objects_total_count, 20); + assert_eq!(progress.objects_total_size, 12288); assert_eq!(progress.bytes_processed, 6144); + assert!((progress.progress_percentage - 50.0).abs() < 0.001); + assert!(progress.estimated_completion_time.is_some()); } #[tokio::test] diff --git a/crates/heal/src/heal/progress.rs b/crates/heal/src/heal/progress.rs index cb602b5de..981aa01fe 100644 --- a/crates/heal/src/heal/progress.rs +++ b/crates/heal/src/heal/progress.rs @@ -13,7 +13,7 @@ // limitations under the License. use serde::{Deserialize, Serialize}; -use std::time::SystemTime; +use std::time::{Duration, SystemTime}; #[derive(Debug, Default, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] @@ -24,6 +24,14 @@ pub struct HealProgress { pub objects_healed: u64, /// Objects failed pub objects_failed: u64, + /// Versions skipped because they were written after this heal started + pub skipped_new_versions: u64, + /// Versions skipped because lifecycle already selected them for expiry + pub skipped_ilm_expired: u64, + /// Baseline object count from the latest complete usage snapshot + pub objects_total_count: u64, + /// Baseline object bytes from the latest complete usage snapshot + pub objects_total_size: u64, /// Bytes processed pub bytes_processed: u64, /// Current object @@ -54,10 +62,56 @@ impl HealProgress { self.bytes_processed = bytes; self.last_update_time = Some(SystemTime::now()); - // calculate progress percentage - let total = scanned + healed + failed; + self.refresh_progress_percentage(); + self.refresh_estimated_completion_time(); + } + + pub fn set_total_baseline(&mut self, objects_total_count: u64, objects_total_size: u64) { + self.objects_total_count = objects_total_count; + self.objects_total_size = objects_total_size; + self.last_update_time = Some(SystemTime::now()); + self.refresh_progress_percentage(); + self.refresh_estimated_completion_time(); + } + + pub fn record_skipped_new_version(&mut self) { + self.skipped_new_versions = self.skipped_new_versions.saturating_add(1); + self.last_update_time = Some(SystemTime::now()); + self.refresh_progress_percentage(); + self.refresh_estimated_completion_time(); + } + + pub fn record_skipped_ilm_expired(&mut self) { + self.skipped_ilm_expired = self.skipped_ilm_expired.saturating_add(1); + self.last_update_time = Some(SystemTime::now()); + self.refresh_progress_percentage(); + self.refresh_estimated_completion_time(); + } + + fn completed_for_baseline(&self) -> u64 { + self.objects_healed + .saturating_add(self.objects_failed) + .saturating_add(self.skipped_new_versions) + .saturating_add(self.skipped_ilm_expired) + } + + pub(crate) fn refresh_progress_percentage(&mut self) { + if self.objects_total_size > 0 { + self.progress_percentage = ((self.bytes_processed as f64 / self.objects_total_size as f64) * 100.0).min(100.0); + return; + } + if self.objects_total_count > 0 { + let completed = self.completed_for_baseline(); + self.progress_percentage = ((completed as f64 / self.objects_total_count as f64) * 100.0).min(100.0); + return; + } + + let total = self + .objects_scanned + .saturating_add(self.objects_healed) + .saturating_add(self.objects_failed); if total > 0 { - self.progress_percentage = (healed as f64 / total as f64) * 100.0; + self.progress_percentage = (self.objects_healed as f64 / total as f64) * 100.0; } } @@ -66,9 +120,36 @@ impl HealProgress { self.last_update_time = Some(SystemTime::now()); } + pub fn refresh_estimated_completion_time(&mut self) { + let Some(start_time) = self.start_time else { + self.estimated_completion_time = None; + return; + }; + if self.is_completed() || !(0.0..100.0).contains(&self.progress_percentage) || self.bytes_processed == 0 { + self.estimated_completion_time = None; + return; + } + + let elapsed = match SystemTime::now().duration_since(start_time) { + Ok(elapsed) if !elapsed.is_zero() => elapsed, + _ => { + self.estimated_completion_time = None; + return; + } + }; + let estimated_total_secs = elapsed.as_secs_f64() * 100.0 / self.progress_percentage; + self.estimated_completion_time = start_time.checked_add(Duration::from_secs_f64(estimated_total_secs)); + } + pub fn is_completed(&self) -> bool { - self.progress_percentage >= 100.0 - || self.objects_scanned > 0 && self.objects_healed + self.objects_failed >= self.objects_scanned + if self.progress_percentage >= 100.0 { + return true; + } + if self.objects_total_count > 0 || self.objects_total_size > 0 { + return false; + } + + self.objects_scanned > 0 && self.objects_healed.saturating_add(self.objects_failed) >= self.objects_scanned } pub fn get_success_rate(&self) -> f64 { @@ -158,6 +239,10 @@ mod tests { assert_eq!(progress.objects_scanned, 0); assert_eq!(progress.objects_healed, 0); assert_eq!(progress.objects_failed, 0); + assert_eq!(progress.skipped_new_versions, 0); + assert_eq!(progress.skipped_ilm_expired, 0); + assert_eq!(progress.objects_total_count, 0); + assert_eq!(progress.objects_total_size, 0); assert_eq!(progress.bytes_processed, 0); assert_eq!(progress.progress_percentage, 0.0); assert!(progress.start_time.is_some()); @@ -181,6 +266,73 @@ mod tests { assert!(progress.last_update_time.is_some()); } + #[test] + fn test_heal_progress_estimates_completion_time_from_progress() { + let mut progress = HealProgress::new(); + progress.start_time = Some(SystemTime::now() - Duration::from_secs(10)); + + progress.update_progress(100, 25, 0, 4096); + + let eta = progress + .estimated_completion_time + .expect("partial byte progress should estimate completion"); + assert!(eta > SystemTime::now()); + } + + #[test] + fn test_heal_progress_uses_byte_baseline_for_percentage() { + let mut progress = HealProgress::new(); + progress.set_total_baseline(10, 8192); + + progress.update_progress(100, 25, 0, 4096); + + assert!((progress.progress_percentage - 50.0).abs() < 0.001); + } + + #[test] + fn test_heal_progress_uses_object_baseline_when_bytes_unknown() { + let mut progress = HealProgress::new(); + progress.set_total_baseline(10, 0); + + progress.update_progress(100, 3, 2, 0); + + assert!((progress.progress_percentage - 50.0).abs() < 0.001); + } + + #[test] + fn test_heal_progress_counts_skipped_versions_for_object_baseline() { + let mut progress = HealProgress::new(); + progress.set_total_baseline(10, 0); + + progress.update_progress(100, 3, 2, 0); + progress.record_skipped_new_version(); + + assert_eq!(progress.skipped_new_versions, 1); + assert!((progress.progress_percentage - 60.0).abs() < 0.001); + } + + #[test] + fn test_heal_progress_does_not_estimate_completion_without_bytes() { + let mut progress = HealProgress::new(); + progress.start_time = Some(SystemTime::now() - Duration::from_secs(10)); + + progress.update_progress(100, 25, 0, 0); + + assert!(progress.estimated_completion_time.is_none()); + } + + #[test] + fn test_heal_progress_with_baseline_is_not_completed_by_processed_count() { + let mut progress = HealProgress::new(); + progress.start_time = Some(SystemTime::now() - Duration::from_secs(10)); + progress.set_total_baseline(10, 8192); + + progress.update_progress(1, 1, 0, 1024); + + assert!(!progress.is_completed()); + assert!(progress.estimated_completion_time.is_some()); + } + #[test] fn test_heal_progress_update_progress_zero_total() { let mut progress = HealProgress::new(); @@ -251,6 +403,8 @@ mod tests { assert_eq!(json["objectsScanned"], 10); assert_eq!(json["objectsHealed"], 8); assert_eq!(json["objectsFailed"], 2); + assert_eq!(json["skippedNewVersions"], 0); + assert_eq!(json["skippedIlmExpired"], 0); assert_eq!(json["bytesProcessed"], 1024); assert_eq!(json["currentObject"], "test-bucket/test-object"); assert!(json["progressPercentage"].is_number()); diff --git a/crates/heal/src/heal/storage.rs b/crates/heal/src/heal/storage.rs index f5101ac1c..fbe51f050 100644 --- a/crates/heal/src/heal/storage.rs +++ b/crates/heal/src/heal/storage.rs @@ -22,6 +22,7 @@ use serde::{Deserialize, Serialize}; use std::sync::Arc; use tracing::{debug, error, warn}; +use super::storage_api::owner::{EcstoreHealLifecycleExpiryContext, ecstore_load_admin_data_usage_from_backend_cached}; use super::storage_api::storage::{ BucketInfo, BucketOperations, DiskSetSelector, HealOperations as _, ListOperations as _, ObjectIO as _, ObjectOperations as _, StorageAdminApi, @@ -29,6 +30,37 @@ use super::storage_api::storage::{ use super::{DiskStore, ECStore, Endpoint, HealDiskExt as _, StorageError, resume::ReplacementTargetIdentity}; pub use super::{HealObjectInfo, HealObjectOptions, HealPutObjReader}; +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct HealBucketUsageBaseline { + pub objects_count: u64, + pub bytes: u64, +} + +pub struct HealLifecycleExpiryContext { + inner: HealLifecycleExpiryContextInner, +} + +enum HealLifecycleExpiryContextInner { + Ecstore(EcstoreHealLifecycleExpiryContext), + #[allow(dead_code)] + Test, +} + +impl HealLifecycleExpiryContext { + fn ecstore(inner: EcstoreHealLifecycleExpiryContext) -> Self { + Self { + inner: HealLifecycleExpiryContextInner::Ecstore(inner), + } + } + + #[cfg(test)] + pub(crate) fn test() -> Self { + Self { + inner: HealLifecycleExpiryContextInner::Test, + } + } +} + const LOG_COMPONENT_HEAL: &str = "heal"; const LOG_SUBSYSTEM_STORAGE: &str = "storage"; const EVENT_HEAL_STORAGE_OBJECT_IO: &str = "heal_storage_object_io"; @@ -272,6 +304,10 @@ pub struct HealListItem { pub name: String, /// normalized version id (`None` when the version is nil/absent) pub version_id: Option, + /// version modification time as Unix nanoseconds + pub mod_time_unix_nanos: Option, + /// object snapshot for lifecycle evaluation + pub lifecycle_object_info: Option, /// whether this version is a delete marker (observability only) pub is_delete_marker: bool, } @@ -329,6 +365,28 @@ pub trait HealStorageAPI: Send + Sync { /// Get bucket info async fn get_bucket_info(&self, bucket: &str) -> Result>; + /// Aggregate usage-cache baselines for the requested buckets. + async fn erasure_set_usage_baseline(&self, _buckets: &[String]) -> Result> { + Ok(None) + } + + /// Load per-bucket lifecycle expiry context for heal skips. + async fn load_heal_lifecycle_expiry_context(&self, _bucket: &str) -> Result> { + Ok(None) + } + + /// Queue lifecycle expiry for a version that heal can skip. + async fn enqueue_heal_lifecycle_expiry( + &self, + _context: &HealLifecycleExpiryContext, + _bucket: &str, + _object: &str, + _version_id: Option<&str>, + _object_info: Option<&HealObjectInfo>, + ) -> Result { + Ok(false) + } + /// Fix bucket metadata async fn heal_bucket_metadata(&self, bucket: &str) -> Result<()>; @@ -409,6 +467,7 @@ pub trait HealStorageAPI: Send + Sync { bucket: &str, prefix: &str, continuation_token: Option<&str>, + include_lifecycle_object_info: bool, ) -> Result<(Vec, Option, bool)>; /// List versions for healing via a per-erasure-set DISK-WALK union enumerator @@ -427,8 +486,10 @@ pub trait HealStorageAPI: Send + Sync { bucket: &str, prefix: &str, continuation_token: Option<&str>, + include_lifecycle_object_info: bool, ) -> Result<(Vec, Option, bool)> { - self.list_objects_for_heal_page(bucket, prefix, continuation_token).await + self.list_objects_for_heal_page(bucket, prefix, continuation_token, include_lifecycle_object_info) + .await } /// Get disk for resume functionality. @@ -1021,6 +1082,85 @@ impl HealStorageAPI for ECStoreHealStorage { } } + async fn erasure_set_usage_baseline(&self, buckets: &[String]) -> Result> { + if buckets.is_empty() { + return Ok(None); + } + + let info = match ecstore_load_admin_data_usage_from_backend_cached(self.ecstore.clone()).await { + Ok(info) if info.is_complete_bucket_usage_snapshot() => info, + Ok(_) | Err(_) => return Ok(None), + }; + + let mut baseline = HealBucketUsageBaseline::default(); + for bucket in buckets { + if let Some(usage) = info.buckets_usage.get(bucket) { + baseline.objects_count = baseline.objects_count.saturating_add(usage.objects_count); + baseline.bytes = baseline.bytes.saturating_add(usage.size); + } + } + + Ok(Some(baseline)) + } + + async fn load_heal_lifecycle_expiry_context(&self, bucket: &str) -> Result> { + match self.ecstore.load_heal_lifecycle_expiry_context(bucket).await { + Ok(Some(context)) => Ok(Some(HealLifecycleExpiryContext::ecstore(context))), + Ok(None) => Ok(None), + Err(err) => { + debug!( + target: "rustfs::heal::storage", + event = EVENT_HEAL_STORAGE_ADMIN_OP, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_STORAGE, + operation = "load_heal_lifecycle_expiry_context", + bucket, + result = "failed", + error = %err, + "Heal storage lifecycle expiry context load failed" + ); + Ok(None) + } + } + } + + async fn enqueue_heal_lifecycle_expiry( + &self, + context: &HealLifecycleExpiryContext, + bucket: &str, + object: &str, + version_id: Option<&str>, + object_info: Option<&HealObjectInfo>, + ) -> Result { + let context = match &context.inner { + HealLifecycleExpiryContextInner::Ecstore(context) => context, + HealLifecycleExpiryContextInner::Test => return Ok(false), + }; + match self + .ecstore + .enqueue_heal_lifecycle_expiry(context, bucket, object, version_id, object_info) + .await + { + Ok(queued) => Ok(queued), + Err(err) => { + debug!( + target: "rustfs::heal::storage", + event = EVENT_HEAL_STORAGE_ADMIN_OP, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_STORAGE, + operation = "enqueue_heal_lifecycle_expiry", + bucket, + object, + version_id = ?version_id, + result = "failed", + error = %err, + "Heal storage lifecycle expiry check failed" + ); + Ok(false) + } + } + } + async fn heal_bucket_metadata(&self, bucket: &str) -> Result<()> { debug!( target: "rustfs::heal::storage", @@ -1436,7 +1576,7 @@ impl HealStorageAPI for ECStoreHealStorage { loop { let (page_objects, next_token, is_truncated) = self - .list_objects_for_heal_page(bucket, prefix, continuation_token.as_deref()) + .list_objects_for_heal_page(bucket, prefix, continuation_token.as_deref(), false) .await?; all_objects.extend(page_objects); @@ -1471,6 +1611,7 @@ impl HealStorageAPI for ECStoreHealStorage { bucket: &str, prefix: &str, continuation_token: Option<&str>, + include_lifecycle_object_info: bool, ) -> Result<(Vec, Option, bool)> { debug!( target: "rustfs::heal::storage", @@ -1522,10 +1663,19 @@ impl HealStorageAPI for ECStoreHealStorage { let page_objects: Vec = list_info .objects .into_iter() - .map(|obj| HealListItem { - name: obj.name, - version_id: obj.version_id.filter(|u| !u.is_nil()).map(|u| u.to_string()), - is_delete_marker: obj.delete_marker, + .map(|mut obj| { + obj.version_id = obj.version_id.filter(|u| !u.is_nil()); + let version_id = obj.version_id.map(|u| u.to_string()); + let mod_time_unix_nanos = obj.mod_time.map(|mod_time| mod_time.unix_timestamp_nanos()); + let is_delete_marker = obj.delete_marker; + let lifecycle_object_info = include_lifecycle_object_info.then(|| obj.clone()); + HealListItem { + name: obj.name, + version_id, + mod_time_unix_nanos, + lifecycle_object_info, + is_delete_marker, + } }) .collect(); let page_count = page_objects.len(); @@ -1562,6 +1712,7 @@ impl HealStorageAPI for ECStoreHealStorage { bucket: &str, prefix: &str, continuation_token: Option<&str>, + include_lifecycle_object_info: bool, ) -> Result<(Vec, Option, bool)> { // Per-page bounds for the disk-walk union enumerator. Objects are atomic // (never split across pages), so version_budget only bounds how many @@ -1590,7 +1741,16 @@ impl HealStorageAPI for ECStoreHealStorage { let (versions, next_forward, is_truncated) = self .ecstore - .heal_walk_versions_page(pool_idx, set_idx, bucket, prefix, forward_to.as_deref(), BATCH_OBJECTS, VERSION_BUDGET) + .heal_walk_versions_page( + pool_idx, + set_idx, + bucket, + prefix, + forward_to.as_deref(), + BATCH_OBJECTS, + VERSION_BUDGET, + include_lifecycle_object_info, + ) .await .map_err(|e| { error!( @@ -1614,6 +1774,8 @@ impl HealStorageAPI for ECStoreHealStorage { .map(|v| HealListItem { name: v.name, version_id: v.version_id, + mod_time_unix_nanos: v.mod_time_unix_nanos, + lifecycle_object_info: v.lifecycle_object_info, is_delete_marker: v.is_delete_marker, }) .collect(); diff --git a/crates/heal/src/heal/storage_api.rs b/crates/heal/src/heal/storage_api.rs index 417998e39..ff722b418 100644 --- a/crates/heal/src/heal/storage_api.rs +++ b/crates/heal/src/heal/storage_api.rs @@ -12,7 +12,10 @@ // See the License for the specific language governing permissions and // limitations under the License. -pub(crate) use rustfs_ecstore::api::data_usage::DATA_USAGE_CACHE_NAME as ECSTORE_DATA_USAGE_CACHE_NAME; +pub(crate) use rustfs_ecstore::api::data_usage::{ + DATA_USAGE_CACHE_NAME as ECSTORE_DATA_USAGE_CACHE_NAME, + load_admin_data_usage_from_backend_cached as ecstore_load_admin_data_usage_from_backend_cached, +}; pub(crate) use rustfs_ecstore::api::disk::endpoint::Endpoint as EcstoreEndpoint; pub(crate) use rustfs_ecstore::api::disk::error::{DiskError as EcstoreDiskError, Result as EcstoreDiskResult}; pub(crate) use rustfs_ecstore::api::disk::{ @@ -25,7 +28,9 @@ pub(crate) use rustfs_ecstore::api::disk::{ pub(crate) use rustfs_ecstore::api::disk::{DiskOption as EcstoreDiskOption, new_disk as ecstore_new_disk}; pub(crate) use rustfs_ecstore::api::error::{Error as EcstoreErrorType, StorageError as EcstoreStorageError}; pub(crate) use rustfs_ecstore::api::runtime::local_disk_map_read as ecstore_local_disk_map_read; -pub(crate) use rustfs_ecstore::api::storage::ECStore as EcstoreStore; +pub(crate) use rustfs_ecstore::api::storage::{ + ECStore as EcstoreStore, HealLifecycleExpiryContext as EcstoreHealLifecycleExpiryContext, +}; use rustfs_storage_api as storage_contracts; pub(crate) mod owner { @@ -34,8 +39,8 @@ pub(crate) mod owner { pub(crate) use super::{ ECSTORE_BUCKET_META_PREFIX, ECSTORE_DATA_USAGE_CACHE_NAME, ECSTORE_HEALING_MARKER_PATH, ECSTORE_RUSTFS_META_BUCKET, EcstoreConditionalFileUpdate, EcstoreDeleteOptions, EcstoreDiskAPI, EcstoreDiskBytes, EcstoreDiskError, - EcstoreDiskResult, EcstoreDiskStore, EcstoreEndpoint, EcstoreErrorType, EcstoreStorageError, EcstoreStore, - ecstore_local_disk_map_read, + EcstoreDiskResult, EcstoreDiskStore, EcstoreEndpoint, EcstoreErrorType, EcstoreHealLifecycleExpiryContext, + EcstoreStorageError, EcstoreStore, ecstore_load_admin_data_usage_from_backend_cached, ecstore_local_disk_map_read, }; #[cfg(test)] diff --git a/crates/heal/src/heal/task.rs b/crates/heal/src/heal/task.rs index 62123c418..f6472eb65 100644 --- a/crates/heal/src/heal/task.rs +++ b/crates/heal/src/heal/task.rs @@ -19,11 +19,12 @@ use crate::heal::{ resume::{ CheckpointManager, ReplacementPhase, ReplacementTargetIdentity, ResumeManager, replacement_target_identities_match, }, - storage::{HealStorageAPI, next_heal_listing_token}, + storage::{HealBucketUsageBaseline, HealStorageAPI, next_heal_listing_token}, }; use crate::{Error, Result}; use metrics::{counter, histogram}; use rustfs_common::heal_channel::{HealOpts, HealRequestSource, HealScanMode}; +use rustfs_common::trace_bus::{TraceEvent, TraceFunc, TraceKind, trace_emit}; use rustfs_madmin::heal_commands::HealResultItem; use rustfs_utils::path::SLASH_SEPARATOR; use serde::{Deserialize, Serialize}; @@ -178,6 +179,17 @@ pub enum HealPriority { Urgent = 3, } +impl HealPriority { + fn as_str(self) -> &'static str { + match self { + Self::Low => "low", + Self::Normal => "normal", + Self::High => "high", + Self::Urgent => "urgent", + } + } +} + /// Heal options #[derive(Debug, Clone, Serialize, Deserialize)] pub struct HealOptions { @@ -498,6 +510,61 @@ impl HealTask { } } + fn emit_trace_task_state(&self, state: &'static str, duration: Duration, error: Option<&Error>) { + trace_emit(|| { + let mut event = TraceEvent::new(TraceKind::Heal, TraceFunc::HealTask) + .with_duration(duration) + .with_attr("task_id", self.id.as_str()) + .with_attr("heal_type", self.heal_type.log_kind()) + .with_attr("state", state) + .with_attr("source", self.source.as_str()) + .with_attr("priority", self.priority.as_str()) + .with_attr("retry_attempts", u64::from(self.retry_attempts)) + .with_attr("dry_run", self.options.dry_run); + + event = match &self.heal_type { + HealType::Cluster => event, + HealType::Object { + bucket, + object, + version_id, + } => { + let event = event.with_bucket(bucket.as_str()).with_object(object.as_str()); + match version_id { + Some(version_id) => event.with_attr("version_id", version_id.as_str()), + None => event, + } + } + HealType::Bucket { bucket } => event.with_bucket(bucket.as_str()), + HealType::Prefix { bucket, prefix } => event.with_bucket(bucket.as_str()).with_object(prefix.as_str()), + HealType::ErasureSet { buckets, set_disk_id } => { + let bucket_count = u64::try_from(buckets.len()).unwrap_or(u64::MAX); + event + .with_attr("set_disk_id", set_disk_id.as_str()) + .with_attr("bucket_count", bucket_count) + } + HealType::Metadata { bucket, object } => event.with_bucket(bucket.as_str()).with_object(object.as_str()), + HealType::ECDecode { + bucket, + object, + version_id, + } => { + let event = event.with_bucket(bucket.as_str()).with_object(object.as_str()); + match version_id { + Some(version_id) => event.with_attr("version_id", version_id.as_str()), + None => event, + } + } + HealType::MRF { meta_path } => event.with_object(meta_path.as_str()), + }; + + match error { + Some(error) => event.with_attr("error", error.to_string()), + None => event, + } + }); + } + async fn remaining_timeout(&self) -> Result> { if let Some(total) = self.options.timeout { let start_instant = { *self.task_start_instant.read().await }; @@ -717,6 +784,7 @@ impl HealTask { queue_delay = ?queue_delay, "Heal task started" }); + self.emit_trace_task_state("started", Duration::ZERO, None); let result = match &self.heal_type { HealType::Cluster => self.heal_cluster().await, @@ -805,6 +873,14 @@ impl HealTask { } } + let terminal_state = match &result { + Ok(_) => "completed", + Err(Error::TaskCancelled) => "cancelled", + Err(Error::TaskTimeout) => "timed_out", + Err(_) => "failed", + }; + self.emit_trace_task_state(terminal_state, start_instant.elapsed(), result.as_ref().err()); + result } @@ -1535,7 +1611,7 @@ impl HealTask { let (objects, next_token, is_truncated) = self .await_with_control( self.storage - .list_objects_for_heal_page(bucket, prefix, continuation_token.as_deref()), + .list_objects_for_heal_page(bucket, prefix, continuation_token.as_deref(), false), ) .await?; @@ -1697,6 +1773,23 @@ impl HealTask { Ok(()) } + async fn apply_erasure_set_usage_baseline(&self, buckets: &[String]) -> Result<()> { + let baseline = match self + .await_with_control(self.storage.erasure_set_usage_baseline(buckets)) + .await + { + Ok(Some(baseline)) => baseline, + Ok(None) => return Ok(()), + Err(err @ Error::TaskCancelled) | Err(err @ Error::TaskTimeout) => return Err(err), + Err(_) => return Ok(()), + }; + + let HealBucketUsageBaseline { objects_count, bytes } = baseline; + let mut progress = self.progress.write().await; + progress.set_total_baseline(objects_count, bytes); + Ok(()) + } + async fn heal_metadata(&self, bucket: &str, object: &str) -> Result<()> { debug!( target: "rustfs::heal::task", @@ -2298,6 +2391,8 @@ impl HealTask { None }; + self.apply_erasure_set_usage_baseline(&buckets).await?; + let healing_marker = format!("{set_disk_id}:{}", self.id); if let Some((disk, resume_manager, _)) = replacement_resume.as_ref() { let state = resume_manager.get_state().await; @@ -2602,7 +2697,8 @@ impl HealTask { { let mut progress = self.progress.write().await; - progress.update_progress(4, 4, 0, 0); + let bytes_processed = progress.bytes_processed; + progress.update_progress(4, 4, 0, bytes_processed); } match result { @@ -2658,6 +2754,7 @@ mod tests { use super::super::{DiskOption, DiskStore, Endpoint, HealDiskExt as _, new_disk}; use super::*; use crate::heal::storage::{DiskStatus, HealListItem, HealObjectInfo}; + use rustfs_common::trace_bus::{TraceEvent, TraceFunc, TraceKind, TraceSubscription, TraceVal, subscribe_trace_events}; use rustfs_madmin::heal_commands::{HealDriveInfo, HealResultItem, Infos}; use std::collections::{HashMap, VecDeque}; use std::sync::Mutex; @@ -3203,6 +3300,8 @@ mod tests { block_heal_object: Mutex, resume_disk: Mutex>, replacement_resume_disk: Mutex>, + usage_baseline: Mutex>, + usage_baseline_error: Mutex, } #[test] @@ -3265,11 +3364,69 @@ mod tests { assert_eq!(samples_logged, MAX_BUCKET_FAILURE_LOG_SAMPLES); } + #[tokio::test] + async fn execute_emits_heal_trace_task_state() { + let mut trace = subscribe_trace_events(); + let storage = Arc::new(MockStorage::default()); + let task = HealTask::from_request( + HealRequest::object("bucket-a".to_string(), "object-a".to_string(), Some("version-a".to_string())), + storage, + ); + + task.execute().await.expect("mock object heal should complete"); + + let started = recv_trace_task_state(&mut trace, &task.id, "started").await; + assert_eq!(started.kind, TraceKind::Heal); + assert_eq!(started.func, TraceFunc::HealTask); + assert_eq!(started.bucket.as_deref(), Some("bucket-a")); + assert_eq!(started.object.as_deref(), Some("object-a")); + assert_eq!(trace_attr_string(&started, "heal_type").as_deref(), Some("object")); + assert_eq!(trace_attr_string(&started, "source").as_deref(), Some("internal")); + assert_eq!(trace_attr_string(&started, "version_id").as_deref(), Some("version-a")); + + let completed = recv_trace_task_state(&mut trace, &task.id, "completed").await; + assert_eq!(completed.kind, TraceKind::Heal); + assert_eq!(completed.func, TraceFunc::HealTask); + assert_eq!(trace_attr_string(&completed, "state").as_deref(), Some("completed")); + } + + async fn recv_trace_task_state(trace: &mut TraceSubscription, task_id: &str, state: &str) -> TraceEvent { + for _ in 0..32 { + let event = tokio::time::timeout(Duration::from_secs(1), trace.recv()) + .await + .expect("trace event should arrive") + .expect("trace bus should stay open"); + if trace_attr_string(&event, "task_id").as_deref() == Some(task_id) + && trace_attr_string(&event, "state").as_deref() == Some(state) + { + return (*event).clone(); + } + } + + panic!("expected trace state {state} for task {task_id}"); + } + + fn trace_attr_string(event: &TraceEvent, key: &str) -> Option { + event.attrs.iter().find_map(|attr| { + if attr.key != key { + return None; + } + Some(match &attr.value { + TraceVal::Bool(value) => value.to_string(), + TraceVal::U64(value) => value.to_string(), + TraceVal::I64(value) => value.to_string(), + TraceVal::Str(value) => value.to_string(), + }) + }) + } + /// Build a latest, non-delete-marker heal list item with no version id. fn heal_item(name: &str) -> HealListItem { HealListItem { name: name.to_string(), version_id: None, + mod_time_unix_nanos: None, + lifecycle_object_info: None, is_delete_marker: false, } } @@ -3357,6 +3514,13 @@ mod tests { })) } + async fn erasure_set_usage_baseline(&self, _buckets: &[String]) -> Result> { + if *self.usage_baseline_error.lock().unwrap() { + return Err(Error::Other("usage baseline unavailable".to_string())); + } + Ok(*self.usage_baseline.lock().unwrap()) + } + async fn heal_bucket_metadata(&self, _bucket: &str) -> Result<()> { Ok(()) } @@ -3540,6 +3704,7 @@ mod tests { bucket: &str, prefix: &str, continuation_token: Option<&str>, + _include_lifecycle_object_info: bool, ) -> Result<(Vec, Option, bool)> { self.listed_prefixes.lock().unwrap().push(prefix.to_string()); if *self.truncate_without_token.lock().unwrap() { @@ -4654,6 +4819,73 @@ mod tests { assert!(storage.object_heal_opts.lock().unwrap().is_empty()); } + #[tokio::test] + async fn erasure_set_heal_applies_usage_baseline_to_progress() { + let temp = TempDir::new().expect("temporary directory should be created"); + let disk = make_resume_disk(&temp).await; + let storage = Arc::new(MockStorage { + resume_disk: Mutex::new(Some(disk)), + usage_baseline: Mutex::new(Some(HealBucketUsageBaseline { + objects_count: 10, + bytes: 8, + })), + ..Default::default() + }); + let request = HealRequest::new( + HealType::ErasureSet { + buckets: vec!["bucket-a".to_string()], + set_disk_id: "pool_0_set_0".to_string(), + }, + HealOptions { + timeout: None, + ..Default::default() + }, + HealPriority::Normal, + ); + let task = HealTask::from_request(request, storage); + + task.heal_erasure_set(vec!["bucket-a".to_string()], "pool_0_set_0".to_string()) + .await + .expect("erasure set heal should complete"); + + let progress = task.get_progress().await; + assert_eq!(progress.objects_total_count, 10); + assert_eq!(progress.objects_total_size, 8); + assert_eq!(progress.bytes_processed, 2); + assert!((progress.progress_percentage - 25.0).abs() < 0.001); + } + + #[tokio::test] + async fn erasure_set_heal_ignores_usage_baseline_errors() { + let temp = TempDir::new().expect("temporary directory should be created"); + let disk = make_resume_disk(&temp).await; + let storage = Arc::new(MockStorage { + resume_disk: Mutex::new(Some(disk)), + usage_baseline_error: Mutex::new(true), + ..Default::default() + }); + let request = HealRequest::new( + HealType::ErasureSet { + buckets: vec!["bucket-a".to_string()], + set_disk_id: "pool_0_set_0".to_string(), + }, + HealOptions { + timeout: None, + ..Default::default() + }, + HealPriority::Normal, + ); + let task = HealTask::from_request(request, storage); + + task.heal_erasure_set(vec!["bucket-a".to_string()], "pool_0_set_0".to_string()) + .await + .expect("usage baseline failures should not fail erasure set heal"); + + let progress = task.get_progress().await; + assert_eq!(progress.objects_total_count, 0); + assert_eq!(progress.objects_total_size, 0); + } + #[tokio::test] async fn resumable_erasure_set_execution_is_cancelled_while_object_heal_is_pending() { let temp = TempDir::new().expect("temporary directory should be created"); diff --git a/crates/heal/src/lib.rs b/crates/heal/src/lib.rs index 7c156304d..3dd29b064 100644 --- a/crates/heal/src/lib.rs +++ b/crates/heal/src/lib.rs @@ -445,6 +445,7 @@ mod tests { _bucket: &str, _prefix: &str, _continuation_token: Option<&str>, + _include_lifecycle_object_info: bool, ) -> Result<(Vec, Option, bool), Error> { Ok((Vec::new(), None, false)) } diff --git a/crates/heal/tests/heal_b5_versioned_regression_test.rs b/crates/heal/tests/heal_b5_versioned_regression_test.rs index 61a542955..95e0f0a90 100644 --- a/crates/heal/tests/heal_b5_versioned_regression_test.rs +++ b/crates/heal/tests/heal_b5_versioned_regression_test.rs @@ -176,7 +176,7 @@ async fn enumerate_all_versions(heal_storage: &Arc, bucket: let mut token: Option = None; loop { let (page, next, truncated) = heal_storage - .list_objects_for_heal_page(bucket, "", token.as_deref()) + .list_objects_for_heal_page(bucket, "", token.as_deref(), false) .await .expect("list_objects_for_heal_page failed"); items.extend(page); diff --git a/crates/heal/tests/heal_b920_subquorum_union_test.rs b/crates/heal/tests/heal_b920_subquorum_union_test.rs index 6d188b3f3..8d4ac65f2 100644 --- a/crates/heal/tests/heal_b920_subquorum_union_test.rs +++ b/crates/heal/tests/heal_b920_subquorum_union_test.rs @@ -166,7 +166,7 @@ async fn enumerate_b5(heal_storage: &Arc, bucket: &str) -> V let mut token: Option = None; loop { let (page, next, truncated) = heal_storage - .list_objects_for_heal_page(bucket, "", token.as_deref()) + .list_objects_for_heal_page(bucket, "", token.as_deref(), false) .await .expect("b5 list page failed"); items.extend(page); @@ -187,7 +187,7 @@ async fn enumerate_disk_walk(heal_storage: &Arc, bucket: &st let mut token: Option = None; loop { let (page, next, truncated) = heal_storage - .list_versions_for_heal_page_disk_walk(SET_DISK_ID, bucket, "", token.as_deref()) + .list_versions_for_heal_page_disk_walk(SET_DISK_ID, bucket, "", token.as_deref(), false) .await .expect("disk-walk list page failed"); items.extend(page); @@ -418,7 +418,7 @@ mod serial_tests { let mut pages = 0usize; loop { let (versions, next_forward, truncated) = ecstore - .heal_walk_versions_page(0, 0, bucket, "", forward.as_deref(), 2, 100_000) + .heal_walk_versions_page(0, 0, bucket, "", forward.as_deref(), 2, 100_000, false) .await .expect("heal_walk_versions_page failed"); pages += 1; diff --git a/crates/heal/tests/heal_bug_fixes_test.rs b/crates/heal/tests/heal_bug_fixes_test.rs index 7b8737f19..ba59142fd 100644 --- a/crates/heal/tests/heal_bug_fixes_test.rs +++ b/crates/heal/tests/heal_bug_fixes_test.rs @@ -242,6 +242,7 @@ fn test_heal_task_status_atomic_update() { _bucket: &str, _prefix: &str, _continuation_token: Option<&str>, + _include_lifecycle_object_info: bool, ) -> rustfs_heal::Result<(Vec, Option, bool)> { Ok((vec![], None, false)) } @@ -385,6 +386,7 @@ async fn test_heal_task_transient_object_exists_skip_avoids_recreate() { _bucket: &str, _prefix: &str, _continuation_token: Option<&str>, + _include_lifecycle_object_info: bool, ) -> rustfs_heal::Result<(Vec, Option, bool)> { Ok((Vec::new(), None, false)) } diff --git a/crates/madmin/src/service_commands.rs b/crates/madmin/src/service_commands.rs index ddda978be..1b5d0b2a9 100644 --- a/crates/madmin/src/service_commands.rs +++ b/crates/madmin/src/service_commands.rs @@ -43,7 +43,7 @@ pub struct ServiceTraceOpts { #[allow(dead_code)] impl ServiceTraceOpts { - fn trace_types(&self) -> TraceType { + pub fn trace_types(&self) -> TraceType { let mut tt = TraceType::default(); tt.set_if(self.s3, &TraceType::S3); tt.set_if(self.internal, &TraceType::INTERNAL); @@ -72,6 +72,14 @@ impl ServiceTraceOpts { tt } + pub fn only_errors(&self) -> bool { + self.only_errors + } + + pub fn threshold(&self) -> Duration { + self.threshold + } + pub fn parse_params(&mut self, uri: &Uri) -> Result<(), String> { let query_pairs: HashMap<_, _> = uri .query() diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index 25f1e758d..820cbb12f 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -41,6 +41,7 @@ use rustfs_common::metrics::{ CloseDiskGuard, IlmAction, Metric, Metrics, ScannerReplicationRepairKind, ScannerSourceWorkUpdate, ScannerWorkSource, UpdateCurrentPathFn, current_path_updater, global_metrics, }; +use rustfs_common::trace_bus::{TraceEvent, TraceFunc, TraceKind, trace_emit, trace_subscriber_count}; use rustfs_filemeta::{MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams}; use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf}; use s3s::dto::{BucketLifecycleConfiguration, ObjectLockConfiguration}; @@ -430,6 +431,113 @@ fn non_negative_i64_to_u64(value: i64) -> u64 { value.max(0) as u64 } +fn trace_start_instant() -> Option { + (trace_subscriber_count() > 0).then(Instant::now) +} + +fn emit_scanner_folder_trace(root: &str, folder: &str, objects: u64, started_at: Option, state: &'static str) { + let Some(started_at) = started_at else { + return; + }; + + trace_emit(|| { + let (bucket, prefix) = path2_bucket_object_with_base_path(root, folder); + TraceEvent::new(TraceKind::Scanner, TraceFunc::ScannerFolder) + .with_bucket(bucket) + .with_object(prefix) + .with_duration(started_at.elapsed()) + .with_attr("state", state) + .with_attr("objects", objects) + }); +} + +fn emit_scanner_ilm_action_trace( + bucket: &str, + object: &str, + action: IlmAction, + count: u64, + queued: bool, + started_at: Option, +) { + let Some(started_at) = started_at else { + return; + }; + + let state = if queued { "queued" } else { "not_queued" }; + trace_emit(|| { + TraceEvent::new(TraceKind::Scanner, TraceFunc::ScannerIlmAction) + .with_bucket(bucket) + .with_object(object) + .with_duration(started_at.elapsed()) + .with_attr("state", state) + .with_attr("action", action.as_str()) + .with_attr("count", count) + .with_attr("queued", queued) + }); +} + +struct ScannerHealCandidateTraceContext { + bucket: String, + object: Option, + version_id: Option, + scan_mode: Option, + started_at: Instant, +} + +fn scanner_heal_candidate_trace_context(request: &HealChannelRequest) -> Option { + let started_at = trace_start_instant()?; + Some(ScannerHealCandidateTraceContext { + bucket: request.bucket.clone(), + object: request.object_prefix.clone(), + version_id: request.object_version_id.clone(), + scan_mode: request.scan_mode, + started_at, + }) +} + +struct ScannerHealCandidateTrace<'a> { + candidate_type: &'static str, + bucket: &'a str, + object: Option<&'a str>, + version_id: Option<&'a str>, + priority: HealChannelPriority, + scan_mode: Option, + result: Result, + started_at: Instant, +} + +fn emit_scanner_heal_candidate_trace(trace: ScannerHealCandidateTrace<'_>) { + trace_emit(|| { + let (state, admission, error) = match trace.result { + Ok(result) if result.is_admitted() => ("admitted", describe_heal_admission(result), None), + Ok(result) => ("not_admitted", describe_heal_admission(result), None), + Err(error) => ("submit_failed", "channel_error".to_string(), Some(error)), + }; + let mut event = TraceEvent::new(TraceKind::Scanner, TraceFunc::ScannerHealCandidate) + .with_bucket(trace.bucket) + .with_duration(trace.started_at.elapsed()) + .with_attr("state", state) + .with_attr("candidate_type", trace.candidate_type) + .with_attr("priority", heal_priority_label(trace.priority)) + .with_attr("admission", admission); + + if let Some(object) = trace.object { + event = event.with_object(object); + } + if let Some(version_id) = trace.version_id { + event = event.with_attr("version_id", version_id); + } + if let Some(scan_mode) = trace.scan_mode { + event = event.with_attr("scan_mode", scan_mode.as_str()); + } + if let Some(error) = error { + event = event.with_attr("error", error); + } + + event + }); +} + fn apply_scanner_size_summary(into: &mut DataUsageEntry, summary: &SizeSummary) { into.size = into.size.saturating_add(summary.total_size); into.versions = into.versions.saturating_add(summary.versions); @@ -677,9 +785,22 @@ async fn send_scanner_heal_request( request: HealChannelRequest, ) -> Result { let priority = request.priority; + let trace_context = scanner_heal_candidate_trace_context(&request); match send_heal_request_with_admission(request).await { Ok(result) => { record_heal_candidate_admission(candidate_type, priority, result); + if let Some(trace_context) = trace_context.as_ref() { + emit_scanner_heal_candidate_trace(ScannerHealCandidateTrace { + candidate_type, + bucket: &trace_context.bucket, + object: trace_context.object.as_deref(), + version_id: trace_context.version_id.as_deref(), + priority, + scan_mode: trace_context.scan_mode, + result: Ok(result), + started_at: trace_context.started_at, + }); + } Ok(result) } Err(err) => { @@ -690,6 +811,18 @@ async fn send_scanner_heal_request( "result" => "channel_error".to_string() ) .increment(1); + if let Some(trace_context) = trace_context.as_ref() { + emit_scanner_heal_candidate_trace(ScannerHealCandidateTrace { + candidate_type, + bucket: &trace_context.bucket, + object: trace_context.object.as_deref(), + version_id: trace_context.version_id.as_deref(), + priority, + scan_mode: trace_context.scan_mode, + result: Err(err.as_str()), + started_at: trace_context.started_at, + }); + } Err(ScannerError::Other(err)) } } @@ -905,7 +1038,9 @@ impl ScannerItem { "Scanner lifecycle action dispatched" ); let done_ilm = Metrics::time_ilm(event.action); + let trace_started_at = trace_start_instant(); let queued = apply_expiry_rule(event, &LcEventSrc::Scanner, oi).await; + emit_scanner_ilm_action_trace(&self.bucket, &oi.name, event.action, 1, queued, trace_started_at); if record_scanner_ilm_action_if_queued(global_metrics(), event.action, 1, queued) { done_ilm(1)(); remaining_versions = 0; @@ -957,7 +1092,9 @@ impl ScannerItem { "Scanner lifecycle action dispatched" ); let done_ilm = Metrics::time_ilm(event.action); + let trace_started_at = trace_start_instant(); let queued = apply_expiry_rule(event, &LcEventSrc::Scanner, oi).await; + emit_scanner_ilm_action_trace(&self.bucket, &oi.name, event.action, 1, queued, trace_started_at); if record_scanner_ilm_action_if_queued(global_metrics(), event.action, 1, queued) { done_ilm(1)(); if !versioning_config.prefix_enabled(&self.object_path()) && event.action == IlmAction::DeleteAction { @@ -995,7 +1132,9 @@ impl ScannerItem { "Scanner lifecycle action dispatched" ); let done_ilm = Metrics::time_ilm(event.action); + let trace_started_at = trace_start_instant(); let queued = apply_transition_rule(event, &LcEventSrc::Scanner, oi).await; + emit_scanner_ilm_action_trace(&self.bucket, &oi.name, event.action, 1, queued, trace_started_at); if record_scanner_ilm_action_if_queued(global_metrics(), event.action, 1, queued) { done_ilm(1)(); } @@ -1019,7 +1158,21 @@ impl ScannerItem { let action = event.action; let count = u64::try_from(to_delete_objs.len()).unwrap_or(u64::MAX); let done_ilm = Metrics::time_ilm(action); + let trace_started_at = trace_start_instant(); let queued = enqueue_runtime_newer_noncurrent(&self.bucket, to_delete_objs, event, &LcEventSrc::Scanner).await; + if let Some(trace_started_at) = trace_started_at { + let state = if queued { "queued" } else { "not_queued" }; + trace_emit(|| { + TraceEvent::new(TraceKind::Scanner, TraceFunc::ScannerIlmAction) + .with_bucket(self.bucket.as_str()) + .with_object(self.object_path()) + .with_duration(trace_started_at.elapsed()) + .with_attr("state", state) + .with_attr("action", action.as_str()) + .with_attr("count", count) + .with_attr("queued", queued) + }); + } if record_scanner_ilm_action_if_queued(global_metrics(), action, count, queued) { done_ilm(count)(); remaining_versions = remaining_versions.saturating_sub(noncurrent_accounting.len()); @@ -1830,6 +1983,7 @@ impl FolderScanner { into: &mut DataUsageEntry, ) -> Result<(), ScannerError> { let done_folder = Metrics::time(Metric::ScanFolder); + let trace_started_at = trace_start_instant(); if ctx.is_cancelled() { return Err(ScannerError::Other("Operation cancelled".to_string())); @@ -2895,6 +3049,8 @@ impl FolderScanner { } done_folder(); + let scanned_objects = u64::try_from(into.objects).unwrap_or(u64::MAX); + emit_scanner_folder_trace(&self.root, &folder.name, scanned_objects, trace_started_at, "completed"); Ok(()) } @@ -4400,6 +4556,104 @@ mod tests { ); } + #[tokio::test] + async fn scanner_trace_helpers_emit_expected_events() { + let mut trace = rustfs_common::trace_bus::subscribe_trace_events(); + + emit_scanner_folder_trace( + "/tmp/rustfs-scanner-trace", + "/tmp/rustfs-scanner-trace/bucket-a/folder-a", + 7, + Some(Instant::now()), + "completed", + ); + let folder = recv_scanner_trace_event( + &mut trace, + TraceFunc::ScannerFolder, + Some("bucket-a"), + Some("folder-a"), + Some("completed"), + ) + .await; + assert_eq!(trace_attr_string(&folder, "objects").as_deref(), Some("7")); + + emit_scanner_ilm_action_trace("bucket-a", "object-a", IlmAction::DeleteAction, 2, true, Some(Instant::now())); + let ilm = recv_scanner_trace_event( + &mut trace, + TraceFunc::ScannerIlmAction, + Some("bucket-a"), + Some("object-a"), + Some("queued"), + ) + .await; + assert_eq!(trace_attr_string(&ilm, "action").as_deref(), Some("delete")); + assert_eq!(trace_attr_string(&ilm, "count").as_deref(), Some("2")); + assert_eq!(trace_attr_string(&ilm, "queued").as_deref(), Some("true")); + + emit_scanner_heal_candidate_trace(ScannerHealCandidateTrace { + candidate_type: "object", + bucket: "bucket-a", + object: Some("object-a"), + version_id: Some("version-a"), + priority: HealChannelPriority::High, + scan_mode: Some(HealScanMode::Deep), + result: Ok(HealAdmissionResult::Merged), + started_at: Instant::now(), + }); + let heal_candidate = recv_scanner_trace_event( + &mut trace, + TraceFunc::ScannerHealCandidate, + Some("bucket-a"), + Some("object-a"), + Some("admitted"), + ) + .await; + assert_eq!(trace_attr_string(&heal_candidate, "candidate_type").as_deref(), Some("object")); + assert_eq!(trace_attr_string(&heal_candidate, "priority").as_deref(), Some("high")); + assert_eq!(trace_attr_string(&heal_candidate, "scan_mode").as_deref(), Some("deep")); + assert_eq!(trace_attr_string(&heal_candidate, "version_id").as_deref(), Some("version-a")); + assert_eq!(trace_attr_string(&heal_candidate, "admission").as_deref(), Some("merged")); + } + + async fn recv_scanner_trace_event( + trace: &mut rustfs_common::trace_bus::TraceSubscription, + func: TraceFunc, + bucket: Option<&str>, + object: Option<&str>, + state: Option<&str>, + ) -> TraceEvent { + for _ in 0..32 { + let event = tokio::time::timeout(Duration::from_secs(1), trace.recv()) + .await + .expect("scanner trace event should arrive") + .expect("trace bus should stay open"); + if event.kind == TraceKind::Scanner + && event.func == func + && event.bucket.as_deref() == bucket + && event.object.as_deref() == object + && state.is_none_or(|state| trace_attr_string(&event, "state").as_deref() == Some(state)) + { + return (*event).clone(); + } + } + + panic!("expected scanner trace event {func:?} for bucket {bucket:?} object {object:?}"); + } + + fn trace_attr_string(event: &TraceEvent, key: &str) -> Option { + event.attrs.iter().find_map(|attr| { + if attr.key != key { + return None; + } + Some(match &attr.value { + rustfs_common::trace_bus::TraceVal::Bool(value) => value.to_string(), + rustfs_common::trace_bus::TraceVal::U64(value) => value.to_string(), + rustfs_common::trace_bus::TraceVal::I64(value) => value.to_string(), + rustfs_common::trace_bus::TraceVal::Str(value) => value.to_string(), + }) + }) + } + #[test] fn test_build_high_priority_heal_admission_error_contains_context() { let err = build_high_priority_heal_admission_error( diff --git a/rustfs/src/admin/handlers/profile_admin.rs b/rustfs/src/admin/handlers/profile_admin.rs index d4a85360a..d14af9f3b 100644 --- a/rustfs/src/admin/handlers/profile_admin.rs +++ b/rustfs/src/admin/handlers/profile_admin.rs @@ -23,18 +23,24 @@ use futures::{Stream, StreamExt}; use http::{HeaderMap, HeaderValue}; use hyper::{Method, StatusCode}; use matchit::Params; +use regex::Regex; +use rustfs_common::trace_bus::{TraceEvent, TraceKind, TraceVal, subscribe_trace_events}; use rustfs_madmin::service_commands::ServiceTraceOpts; +use rustfs_madmin::trace::TraceType; use rustfs_policy::policy::action::{Action, AdminAction}; use s3s::header::CONTENT_TYPE; use s3s::stream::{ByteStream, DynByteStream}; use s3s::{Body, S3Request, S3Response, S3Result, StdError, s3_error}; use serde::Serialize; +use std::collections::HashMap; use std::pin::Pin; use std::task::{Context, Poll}; -use std::time::Duration; +use std::time::{Duration, SystemTime}; +use time::{OffsetDateTime, format_description::well_known::Rfc3339}; use tokio::sync::mpsc; use tokio_stream::wrappers::ReceiverStream; use tracing::error; +use url::form_urlencoded; #[derive(Serialize)] struct ProfileStatus { @@ -206,16 +212,164 @@ impl Stream for TraceStream { impl ByteStream for TraceStream {} +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct TraceKindFilter { + heal: bool, + scanner: bool, +} + +impl TraceKindFilter { + const ALL_SUPPORTED: Self = Self { + heal: true, + scanner: true, + }; + + fn from_request(uri: &hyper::Uri, trace_types: TraceType) -> S3Result { + let mut has_kind = false; + let mut filter = Self { + heal: false, + scanner: false, + }; + + for (key, value) in trace_query_pairs(uri) { + if key != "kind" { + continue; + } + has_kind = true; + for item in value.split(',') { + match item.trim().to_ascii_lowercase().as_str() { + "heal" | "healing" => filter.heal = true, + "scanner" => filter.scanner = true, + "all" => return Ok(Self::ALL_SUPPORTED), + _ => return Err(s3_error!(InvalidRequest, "invalid trace kind")), + } + } + } + + if has_kind { + return Ok(filter); + } + + if trace_types.mask() == 0 || trace_query_flag(uri, "all") { + return Ok(Self::ALL_SUPPORTED); + } + + Ok(Self { + heal: trace_types.overlaps(&TraceType::HEALING), + scanner: trace_types.overlaps(&TraceType::SCANNER), + }) + } + + const fn matches(self, kind: TraceKind) -> bool { + match kind { + TraceKind::Heal => self.heal, + TraceKind::Scanner => self.scanner, + } + } +} + +#[derive(Debug)] +struct TraceStreamFilter { + kinds: TraceKindFilter, + regex: Option, + threshold: Duration, +} + +impl TraceStreamFilter { + fn from_request(uri: &hyper::Uri, opts: &ServiceTraceOpts) -> S3Result { + if opts.only_errors() { + return Err(s3_error!( + InvalidRequest, + "trace error-only filter is not supported for heal/scanner trace" + )); + } + + Ok(Self { + kinds: TraceKindFilter::from_request(uri, opts.trace_types())?, + regex: trace_regex_filter(uri)?, + threshold: opts.threshold(), + }) + } + + fn matches_kind(&self, kind: TraceKind) -> bool { + self.kinds.matches(kind) + } + + fn matches_record(&self, record: &TraceWireRecord) -> bool { + record.duration >= self.threshold && self.regex.as_ref().is_none_or(|regex| record.matches_regex(regex)) + } +} + +#[derive(Serialize)] +struct TraceWireRecord { + #[serde(rename = "type")] + trace_type: u64, + #[serde(rename = "nodename")] + node_name: String, + #[serde(rename = "funcname")] + func_name: String, + #[serde(rename = "time")] + time: String, + #[serde(rename = "path")] + path: String, + #[serde(rename = "dur")] + duration: Duration, + #[serde(rename = "bytes", skip_serializing_if = "Option::is_none")] + bytes: Option, + #[serde(rename = "msg", skip_serializing_if = "Option::is_none")] + message: Option, + #[serde(rename = "custom", skip_serializing_if = "Option::is_none")] + custom: Option>, +} + +impl TraceWireRecord { + fn from_event(node_name: &str, event: &TraceEvent) -> Self { + Self { + trace_type: trace_type_mask(event.kind), + node_name: node_name.to_owned(), + func_name: event.func.as_str().to_owned(), + time: trace_time_string(event.time), + path: trace_path(event), + duration: event.duration, + bytes: trace_bytes(event.bytes), + message: None, + custom: trace_custom_attrs(event), + } + } + + fn dropped(node_name: &str, dropped: u64) -> Self { + let mut custom = HashMap::new(); + custom.insert("dropped_events".to_string(), dropped.to_string()); + + Self { + trace_type: 0, + node_name: node_name.to_owned(), + func_name: "trace.Dropped".to_string(), + time: trace_time_string(SystemTime::now()), + path: String::new(), + duration: Duration::ZERO, + bytes: None, + message: Some("trace subscriber lagged".to_string()), + custom: Some(custom), + } + } + + fn matches_regex(&self, regex: &Regex) -> bool { + regex.is_match(&self.func_name) + || regex.is_match(&self.path) + || self.message.as_ref().is_some_and(|message| regex.is_match(message)) + || self + .custom + .as_ref() + .is_some_and(|custom| custom.iter().any(|(key, value)| regex.is_match(key) || regex.is_match(value))) + } +} + /// `GET /v3/trace` — stream real-time server trace events. /// -/// RustFS emits diagnostics through the `tracing` pipeline but does not expose -/// an in-process subscriber that can fan trace events out to an admin client -/// (there is no request-trace broadcast channel). Rather than return an opaque -/// `501` — which would make `mc admin trace` fail to connect — this honors the -/// streaming NDJSON contract: it validates the requested trace filters, opens -/// the stream, emits a single capability record explaining that live tracing is -/// not wired, then holds the connection open with keep-alives. No fabricated -/// trace records are ever sent. +/// RustFS currently publishes heal and scanner diagnostics through the common +/// trace bus. The admin endpoint exposes those events as MinIO-shaped NDJSON +/// records while keeping unsupported trace classes filtered out. pub struct TraceHandler {} #[async_trait::async_trait] @@ -228,24 +382,13 @@ impl Operation for TraceHandler { let mut opts = ServiceTraceOpts::default(); opts.parse_params(&req.uri) .map_err(|_| s3_error!(InvalidRequest, "invalid trace parameters"))?; + let filter = TraceStreamFilter::from_request(&req.uri, &opts)?; let node_name = sysinfo::System::host_name().unwrap_or_else(|| "rustfs".to_string()); - let (tx, rx) = mpsc::channel::>(8); + let mut subscription = subscribe_trace_events(); + let (tx, rx) = mpsc::channel::>(64); spawn_traced(async move { - let notice = serde_json::json!({ - "nodename": node_name, - "funcname": "admin.Trace", - "msg": "RustFS does not expose an in-process trace-event subscriber; live tracing is not yet available", - "err": "trace_streaming_unsupported", - }); - if let Ok(mut encoded) = serde_json::to_vec(¬ice) { - encoded.push(b'\n'); - if tx.send(Ok(Bytes::from(encoded))).await.is_err() { - return; - } - } - let mut ticker = tokio::time::interval(Duration::from_secs(15)); ticker.tick().await; loop { @@ -256,6 +399,26 @@ impl Operation for TraceHandler { break; } } + received = subscription.recv() => { + match received { + Ok(event) => { + if !filter.matches_kind(event.kind) { + continue; + } + let record = TraceWireRecord::from_event(&node_name, &event); + if filter.matches_record(&record) && send_trace_record(&tx, &record).await.is_err() { + break; + } + } + Err(tokio::sync::broadcast::error::RecvError::Lagged(dropped)) => { + let record = TraceWireRecord::dropped(&node_name, dropped); + if send_trace_record(&tx, &record).await.is_err() { + break; + } + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => break, + } + } } } }); @@ -269,17 +432,115 @@ impl Operation for TraceHandler { } } +async fn send_trace_record(tx: &mpsc::Sender>, record: &TraceWireRecord) -> Result<(), ()> { + let Some(encoded) = encode_ndjson(record) else { + return Ok(()); + }; + tx.send(Ok(encoded)).await.map_err(|_| ()) +} + +fn encode_ndjson(value: &impl Serialize) -> Option { + let mut encoded = serde_json::to_vec(value).ok()?; + encoded.push(b'\n'); + Some(Bytes::from(encoded)) +} + +fn trace_query_pairs(uri: &hyper::Uri) -> impl Iterator + '_ { + uri.query() + .into_iter() + .flat_map(|query| form_urlencoded::parse(query.as_bytes())) + .map(|(key, value)| (key.into_owned(), value.into_owned())) +} + +fn trace_query_flag(uri: &hyper::Uri, flag: &str) -> bool { + trace_query_pairs(uri).any(|(key, value)| key == flag && value == "true") +} + +fn trace_regex_filter(uri: &hyper::Uri) -> S3Result> { + trace_query_pairs(uri) + .find_map(|(key, value)| { + if key == "filter" && !value.is_empty() { + Some(value) + } else { + None + } + }) + .map(|pattern| Regex::new(&pattern).map_err(|_| s3_error!(InvalidRequest, "invalid trace filter"))) + .transpose() +} + +fn trace_type_mask(kind: TraceKind) -> u64 { + match kind { + TraceKind::Heal => TraceType::HEALING.mask(), + TraceKind::Scanner => TraceType::SCANNER.mask(), + } +} + +fn trace_time_string(time: SystemTime) -> String { + match OffsetDateTime::from(time).format(&Rfc3339) { + Ok(value) => value, + Err(_) => "1970-01-01T00:00:00Z".to_string(), + } +} + +fn trace_path(event: &TraceEvent) -> String { + match (event.bucket.as_deref(), event.object.as_deref()) { + (Some(bucket), Some(object)) if !object.is_empty() => format!("{bucket}/{object}"), + (Some(bucket), _) => bucket.to_owned(), + (None, Some(object)) => object.to_owned(), + (None, None) => String::new(), + } +} + +fn trace_bytes(bytes: u64) -> Option { + if bytes == 0 { + return None; + } + + match i64::try_from(bytes) { + Ok(value) => Some(value), + Err(_) => Some(i64::MAX), + } +} + +fn trace_custom_attrs(event: &TraceEvent) -> Option> { + if event.attrs.is_empty() { + return None; + } + + Some( + event + .attrs + .iter() + .map(|attr| (attr.key.to_string(), trace_value_string(&attr.value))) + .collect(), + ) +} + +fn trace_value_string(value: &TraceVal) -> String { + match value { + TraceVal::Bool(value) => value.to_string(), + TraceVal::U64(value) => value.to_string(), + TraceVal::I64(value) => value.to_string(), + TraceVal::Str(value) => value.to_string(), + } +} + #[cfg(test)] mod tests { use super::{ ProfileControlHandler, ProfileHandler, ProfileStatusHandler, ProfilingDownloadHandler, ProfilingStartHandler, - TraceHandler, + TraceHandler, TraceKindFilter, TraceStreamFilter, TraceWireRecord, }; use crate::admin::router::Operation; use http::{Extensions, HeaderMap, Uri}; use hyper::Method; use matchit::Params; - use s3s::{Body, S3ErrorCode, S3Request}; + use rustfs_common::trace_bus::{TraceEvent, TraceFunc, TraceKind}; + use rustfs_madmin::service_commands::ServiceTraceOpts; + use rustfs_madmin::trace::TraceType; + use s3s::{Body, S3ErrorCode, S3Request, S3Result}; + use std::time::{Duration, UNIX_EPOCH}; fn build_profile_request(uri: &'static str) -> S3Request { S3Request { @@ -295,6 +556,13 @@ mod tests { } } + fn build_trace_stream_filter(uri: &'static str) -> S3Result { + let uri = Uri::from_static(uri); + let mut opts = ServiceTraceOpts::default(); + opts.parse_params(&uri).expect("test trace params should parse"); + TraceStreamFilter::from_request(&uri, &opts) + } + #[tokio::test] async fn profile_handler_rejects_missing_credentials() { let result = ProfileHandler {} @@ -358,4 +626,108 @@ mod tests { .expect_err("trace must reject anonymous requests"); assert_eq!(err.code(), &S3ErrorCode::AccessDenied); } + + #[test] + fn trace_kind_filter_supports_kind_query() { + let uri = Uri::from_static("/rustfs/admin/v3/trace?kind=heal"); + let filter = TraceKindFilter::from_request(&uri, TraceType::default()).expect("kind filter should parse"); + + assert!(filter.matches(TraceKind::Heal)); + assert!(!filter.matches(TraceKind::Scanner)); + } + + #[test] + fn trace_kind_filter_defaults_to_supported_events_without_type_flags() { + let uri = Uri::from_static("/rustfs/admin/v3/trace"); + let filter = TraceKindFilter::from_request(&uri, TraceType::default()).expect("empty filter should parse"); + + assert!(filter.matches(TraceKind::Heal)); + assert!(filter.matches(TraceKind::Scanner)); + } + + #[test] + fn trace_kind_filter_rejects_unknown_kind() { + let uri = Uri::from_static("/rustfs/admin/v3/trace?kind=s3"); + let err = TraceKindFilter::from_request(&uri, TraceType::default()).expect_err("unknown kind should fail"); + + assert_eq!(err.code(), &S3ErrorCode::InvalidRequest); + } + + #[test] + fn trace_stream_filter_matches_regex_against_path_and_attrs() { + let filter = build_trace_stream_filter("/rustfs/admin/v3/trace?kind=heal&filter=data/.%2Bxl.meta") + .expect("regex filter should parse"); + let event = TraceEvent::new(TraceKind::Heal, TraceFunc::HealObject) + .with_bucket("data") + .with_object("dir/xl.meta") + .with_attr("dry_run", true); + let record = TraceWireRecord::from_event("node-a", &event); + + assert!(filter.matches_kind(event.kind)); + assert!(filter.matches_record(&record)); + } + + #[test] + fn trace_stream_filter_rejects_invalid_regex() { + let err = build_trace_stream_filter("/rustfs/admin/v3/trace?kind=heal&filter=[").expect_err("invalid regex should fail"); + + assert_eq!(err.code(), &S3ErrorCode::InvalidRequest); + } + + #[test] + fn trace_stream_filter_applies_threshold() { + let filter = + build_trace_stream_filter("/rustfs/admin/v3/trace?kind=heal&threshold=10ms").expect("threshold should parse"); + let short = TraceWireRecord::from_event( + "node-a", + &TraceEvent::new(TraceKind::Heal, TraceFunc::HealObject).with_duration(Duration::from_millis(9)), + ); + let long = TraceWireRecord::from_event( + "node-a", + &TraceEvent::new(TraceKind::Heal, TraceFunc::HealObject).with_duration(Duration::from_millis(10)), + ); + + assert!(!filter.matches_record(&short)); + assert!(filter.matches_record(&long)); + } + + #[test] + fn trace_stream_filter_rejects_error_only_filter() { + let err = build_trace_stream_filter("/rustfs/admin/v3/trace?kind=heal&err=true").expect_err("err filter should fail"); + + assert_eq!(err.code(), &S3ErrorCode::InvalidRequest); + } + + #[test] + fn trace_wire_record_contains_madmin_trace_fields() { + let event = TraceEvent::new(TraceKind::Heal, TraceFunc::HealObject) + .with_bucket("bucket") + .with_object("object") + .with_duration(Duration::from_millis(3)) + .with_bytes(17) + .with_attr("dry", true); + let mut record = TraceWireRecord::from_event("node-a", &event); + record.time = "1970-01-01T00:00:00Z".to_string(); + + let value = serde_json::to_value(&record).expect("trace record should serialize"); + + assert_eq!(value["type"], TraceType::HEALING.mask()); + assert_eq!(value["nodename"], "node-a"); + assert_eq!(value["funcname"], "heal.Object"); + assert_eq!(value["time"], "1970-01-01T00:00:00Z"); + assert_eq!(value["path"], "bucket/object"); + assert_eq!(value["bytes"], 17); + assert_eq!(value["custom"]["dry"], "true"); + } + + #[test] + fn trace_wire_record_formats_epoch_time() { + let event = TraceEvent { + time: UNIX_EPOCH, + ..TraceEvent::new(TraceKind::Scanner, TraceFunc::ScannerFolder) + }; + let record = TraceWireRecord::from_event("node-a", &event); + + assert_eq!(record.time, "1970-01-01T00:00:00Z"); + } } diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index d06e0892f..a39ca6c3a 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -2420,6 +2420,7 @@ mod tests { _bucket: &str, _prefix: &str, _continuation_token: Option<&str>, + _include_lifecycle_object_info: bool, ) -> rustfs_heal::Result<(Vec, Option, bool)> { Ok((Vec::new(), None, false)) }