From ec135f8c4c7fa133221b6ec677b878f6fbe6c4ce Mon Sep 17 00:00:00 2001 From: anthonymartin Date: Tue, 4 Aug 2026 22:25:59 -0400 Subject: [PATCH] fix(heal): bound per-object logging (#5719) Co-authored-by: Anthony Martin <949506+anthonymartin@users.noreply.github.com> --- crates/ecstore/src/core/sets.rs | 2 +- crates/ecstore/src/erasure/coding/heal.rs | 19 +- crates/ecstore/src/set_disk/mod.rs | 48 ++- crates/ecstore/src/set_disk/ops/heal.rs | 493 +++++++++++++++++++--- crates/ecstore/src/store/heal.rs | 5 +- crates/ecstore/src/store/mod.rs | 2 +- scripts/check_logging_guardrails.sh | 96 +++++ 7 files changed, 587 insertions(+), 78 deletions(-) diff --git a/crates/ecstore/src/core/sets.rs b/crates/ecstore/src/core/sets.rs index efe36f732..5c57677d3 100644 --- a/crates/ecstore/src/core/sets.rs +++ b/crates/ecstore/src/core/sets.rs @@ -1093,7 +1093,7 @@ impl crate::storage_api_contracts::heal::HealOperations for Sets { Ok(result) } - #[tracing::instrument(skip(self))] + #[tracing::instrument(level = "trace", skip(self, opts), fields(bucket = %bucket, object = %object, version_id = %version_id))] async fn heal_object( &self, bucket: &str, diff --git a/crates/ecstore/src/erasure/coding/heal.rs b/crates/ecstore/src/erasure/coding/heal.rs index d4d18401a..8055965e4 100644 --- a/crates/ecstore/src/erasure/coding/heal.rs +++ b/crates/ecstore/src/erasure/coding/heal.rs @@ -24,7 +24,11 @@ use std::io; use std::io::ErrorKind; use std::time::Duration; use tokio::io::AsyncRead; -use tracing::{info, warn}; +use tracing::{trace, warn}; + +const LOG_COMPONENT_ECSTORE: &str = "ecstore"; +const LOG_SUBSYSTEM_ERASURE: &str = "erasure"; +const EVENT_ERASURE_HEAL_STARTED: &str = "erasure_heal_started"; async fn read_heal_shards( readers: &mut [Option>], @@ -115,11 +119,14 @@ impl super::Erasure { where R: AsyncRead + Unpin + Send + Sync, { - info!( - "Erasure heal, writers len: {}, readers len: {}, total_length: {}", - writers.len(), - readers.len(), - total_length + trace!( + event = EVENT_ERASURE_HEAL_STARTED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_ERASURE, + writer_count = writers.len(), + reader_count = readers.len(), + total_length, + "Erasure heal started" ); if writers.len() != self.parity_shards + self.data_shards { return Err(Error::other("invalid argument")); diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 4261073cf..585952a58 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -3964,9 +3964,15 @@ async fn disks_with_all_parts( }; if corrupted { - info!( - "disks_with_all_partsv2: metadata is corrupted, object_name={}, index: {index}", - object_name + debug!( + event = EVENT_SET_DISK_HEAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, + bucket, + object = %object_name, + disk_index = index, + state = "metadata_corrupt", + "Set disk object metadata is corrupt" ); meta_errs[index] = Some(DiskError::FileCorrupt); parts_metadata[index] = FileInfo::default(); @@ -3977,9 +3983,15 @@ async fn disks_with_all_parts( if erasure_distribution_reliable { if !file_info_is_valid_for_metadata(meta) { - info!( - "disks_with_all_partsv2: metadata is not valid, object_name={}, index: {index}", - object_name + debug!( + event = EVENT_SET_DISK_HEAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, + bucket, + object = %object_name, + disk_index = index, + state = "metadata_invalid", + "Set disk object metadata is invalid" ); parts_metadata[index] = FileInfo::default(); meta_errs[index] = Some(DiskError::FileCorrupt); @@ -3991,9 +4003,15 @@ async fn disks_with_all_parts( // Erasure distribution is not the same as onlineDisks // attempt a fix if possible, assuming other entries // might have the right erasure distribution. - info!( - "disks_with_all_partsv2: erasure distribution is not the same as onlineDisks, object_name={}, index: {index}", - object_name + debug!( + event = EVENT_SET_DISK_HEAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, + bucket, + object = %object_name, + disk_index = index, + state = "erasure_distribution_mismatch", + "Set disk erasure distribution mismatched online disks" ); parts_metadata[index] = FileInfo::default(); meta_errs[index] = Some(DiskError::FileCorrupt); @@ -4066,6 +4084,7 @@ async fn disks_with_all_parts( event = EVENT_SET_DISK_HEAL, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_SET_DISK, + bucket, object = %object_name, disk_index = index, state = "verify_failed", @@ -4085,6 +4104,7 @@ async fn disks_with_all_parts( event = EVENT_SET_DISK_HEAL, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_SET_DISK, + bucket, object = %object_name, disk_index = index, state = "check_parts_failed", @@ -4153,9 +4173,13 @@ pub fn should_heal_object_on_disk( } if !meta.equals(latest_meta) { - warn!( - "should_heal_object_on_disk: metadata is outdated, object_name={}, meta: {:?}, latest_meta: {:?}", - meta.name, meta, latest_meta + debug!( + event = EVENT_SET_DISK_HEAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, + object = %meta.name, + state = "metadata_outdated", + "Set disk object metadata is outdated" ); return (true, true, Some(DiskError::OutdatedXLMeta)); } diff --git a/crates/ecstore/src/set_disk/ops/heal.rs b/crates/ecstore/src/set_disk/ops/heal.rs index ab70adc18..2e164a6f4 100644 --- a/crates/ecstore/src/set_disk/ops/heal.rs +++ b/crates/ecstore/src/set_disk/ops/heal.rs @@ -15,6 +15,7 @@ use super::super::*; use crate::io_support::bitrot::object_mmap_read_enabled; use crate::storage_api_contracts::namespace::NamespaceLocking as _; +use tracing::trace; const LOG_COMPONENT_ECSTORE: &str = "ecstore"; const LOG_SUBSYSTEM_HEAL: &str = "heal"; @@ -85,6 +86,67 @@ fn should_fail_heal_rename(_bucket: &str, _object: &str, _disk_index: usize) -> false } +#[cfg(test)] +static HEAL_WRITER_FAILURES: std::sync::Mutex> = std::sync::Mutex::new(Vec::new()); + +#[cfg(test)] +struct HealWriterFailureScope { + bucket: String, + object: String, +} + +#[cfg(test)] +impl HealWriterFailureScope { + fn install(bucket: &str, object: &str, disk_indexes: &[usize], error: DiskError) -> Self { + let mut failures = HEAL_WRITER_FAILURES + .lock() + .expect("heal writer failure registry should not poison"); + assert!( + !failures.iter().any(|(registered_bucket, registered_object, _, _)| { + registered_bucket == bucket && registered_object == object + }), + "heal writer failures must be installed once per object" + ); + failures.extend( + disk_indexes + .iter() + .map(|index| (bucket.to_string(), object.to_string(), *index, error.clone())), + ); + Self { + bucket: bucket.to_string(), + object: object.to_string(), + } + } +} + +#[cfg(test)] +impl Drop for HealWriterFailureScope { + fn drop(&mut self) { + HEAL_WRITER_FAILURES + .lock() + .expect("heal writer failure registry should not poison") + .retain(|(bucket, object, _, _)| bucket != &self.bucket || object != &self.object); + } +} + +#[cfg(test)] +fn injected_heal_writer_error(bucket: &str, object: &str, disk_index: usize) -> Option { + let mut failures = HEAL_WRITER_FAILURES + .lock() + .expect("heal writer failure registry should not poison"); + failures + .iter() + .position(|(registered_bucket, registered_object, registered_index, _)| { + registered_bucket == bucket && registered_object == object && *registered_index == disk_index + }) + .map(|position| failures.swap_remove(position).3) +} + +#[cfg(not(test))] +fn injected_heal_writer_error(_bucket: &str, _object: &str, _disk_index: usize) -> Option { + None +} + #[derive(Clone, Copy, Debug, PartialEq, Eq)] struct PartFailureSummary { part_number: usize, @@ -233,8 +295,41 @@ fn first_unhealthy_part_summary( .map(|(_, summary)| summary) } +fn heal_writer_error_summary(error: &DiskError) -> String { + match error { + DiskError::Io(io_error) => format!("io::{:?}", io_error.kind()), + _ => error.to_string(), + } +} + +fn warn_heal_writer_failures( + bucket: &str, + object: &str, + version_id: &str, + writer_failure_count: usize, + result: &'static str, + first_failure: &(usize, usize, String), +) { + let (first_part_number, first_disk_index, first_error) = first_failure; + warn!( + event = EVENT_SET_DISK_HEAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, + bucket, + object, + version_id, + writer_failure_count, + first_part_number, + first_disk_index, + error = %first_error, + result, + state = "writer_unavailable", + "Set disk object heal writer failures" + ); +} + impl SetDisks { - #[tracing::instrument(skip(self, opts), fields(bucket = %bucket, object = %object, version_id = %version_id))] + #[tracing::instrument(level = "trace", skip(self, opts), fields(bucket = %bucket, object = %object, version_id = %version_id))] pub(in crate::set_disk) async fn heal_object( &self, bucket: &str, @@ -254,7 +349,16 @@ impl SetDisks { opts: &HealOpts, allow_explicit_version_regen: bool, ) -> disk::error::Result<(HealResultItem, Option)> { - info!(?opts, "Starting heal_object"); + trace!( + event = EVENT_SET_DISK_HEAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, + scan_mode = %opts.scan_mode.as_str(), + dry_run = opts.dry_run, + remove = opts.remove, + state = "started", + "Set disk object heal started" + ); let disks = self.get_disks_internal().await; @@ -290,13 +394,14 @@ impl SetDisks { let (mut parts_metadata, errs) = Self::read_all_fileinfo(&disks, "", bucket, object, version_id, true, true, false).await?; - info!( + trace!( + event = EVENT_SET_DISK_HEAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, parts_count = parts_metadata.len(), - bucket = bucket, - object = object, - version_id = version_id, - ?errs, - "File info read complete" + error_count = errs.iter().flatten().count(), + state = "metadata_read", + "Set disk object metadata read" ); if DiskError::is_all_not_found(&errs) { debug!(bucket, object, version_id, "heal_object skipped missing object"); @@ -313,7 +418,14 @@ impl SetDisks { )); } - info!(parts_count = parts_metadata.len(), "heal_object Initiating quorum check"); + trace!( + event = EVENT_SET_DISK_HEAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, + parts_count = parts_metadata.len(), + state = "quorum_check", + "Set disk object quorum check started" + ); match Self::object_quorum_from_meta(&parts_metadata, &errs, self.default_parity_count) { Ok((read_quorum, _)) => { result.parity_blocks = result.disk_count - read_quorum as usize; @@ -325,14 +437,35 @@ impl SetDisks { (Self::list_online_disks(&disks, &parts_metadata, &errs, read_quorum as usize), disk_len) }; - info!(?parts_metadata, ?errs, ?read_quorum, ?disk_len, "heal_object List disks metadata"); - - info!(?online_disks, ?quorum_mod_time, ?quorum_etag, "heal_object List online disks"); + trace!( + event = EVENT_SET_DISK_HEAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, + metadata_count = parts_metadata.len(), + error_count = errs.iter().flatten().count(), + read_quorum, + disk_count = disk_len, + online_disk_count = online_disks.iter().flatten().count(), + state = "disk_metadata_resolved", + "Set disk object metadata resolved" + ); let filter_by_etag = quorum_etag.is_some(); match Self::pick_valid_fileinfo(&parts_metadata, quorum_mod_time, quorum_etag.clone(), read_quorum as usize) { Ok(latest_meta) => { - info!("heal_object latest_meta: {:?}", latest_meta); + trace!( + event = EVENT_SET_DISK_HEAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, + deleted = latest_meta.deleted, + remote = latest_meta.is_remote(), + inline = latest_meta.inline_data(), + part_count = latest_meta.parts.len(), + data_shards = latest_meta.erasure.data_blocks, + parity_shards = latest_meta.erasure.parity_blocks, + state = "canonical_metadata_selected", + "Set disk canonical object metadata selected" + ); let (data_errs_by_disk, data_errs_by_part) = disks_with_all_parts( &mut online_disks, @@ -346,10 +479,14 @@ impl SetDisks { ) .await?; - info!( - "disks_with_all_parts heal_object results: available_disks count={}, total_disks={}", - online_disks.iter().filter(|d| d.is_some()).count(), - online_disks.len() + trace!( + event = EVENT_SET_DISK_HEAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, + available_disk_count = online_disks.iter().flatten().count(), + disk_count = online_disks.len(), + state = "parts_checked", + "Set disk object parts checked" ); let erasure = if !latest_meta.deleted && !latest_meta.is_remote() { @@ -558,11 +695,23 @@ impl SetDisks { } if !latest_meta.deleted && latest_meta.erasure.distribution.len() != online_disks.len() { + let distribution_len = latest_meta.erasure.distribution.len(); + let disk_slot_count = online_disks.len(); let err_str = format!( - "unexpected file distribution ({:?}) from available disks ({:?}), looks like backend disks have been manually modified refusing to heal {}/{}({})", - latest_meta.erasure.distribution, online_disks, bucket, object, version_id + "unexpected file distribution length {distribution_len} for {disk_slot_count} disk slots; backend disks may have been manually modified; refusing to heal {bucket}/{object}({version_id})" + ); + warn!( + event = EVENT_SET_DISK_HEAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, + bucket, + object, + version_id, + distribution_len, + disk_slot_count, + state = "invalid_distribution", + "Set disk object heal refused due to invalid erasure distribution" ); - warn!(err_str); let err = DiskError::other(err_str); return Ok(( self.default_heal_result(latest_meta, &errs, bucket, object, version_id).await, @@ -572,11 +721,23 @@ impl SetDisks { let latest_disks = Self::shuffle_disks(&online_disks, &latest_meta.erasure.distribution); if !latest_meta.deleted && latest_meta.erasure.distribution.len() != out_dated_disks.len() { + let distribution_len = latest_meta.erasure.distribution.len(); + let disk_slot_count = out_dated_disks.len(); let err_str = format!( - "unexpected file distribution ({:?}) from outdated disks ({:?}), looks like backend disks have been manually modified refusing to heal {}/{}({})", - latest_meta.erasure.distribution, out_dated_disks, bucket, object, version_id + "unexpected file distribution length {distribution_len} for {disk_slot_count} disk slots; backend disks may have been manually modified; refusing to heal {bucket}/{object}({version_id})" + ); + warn!( + event = EVENT_SET_DISK_HEAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, + bucket, + object, + version_id, + distribution_len, + disk_slot_count, + state = "invalid_distribution", + "Set disk object heal refused due to invalid erasure distribution" ); - warn!(err_str); let err = DiskError::other(err_str); return Ok(( self.default_heal_result(latest_meta, &errs, bucket, object, version_id).await, @@ -585,15 +746,23 @@ impl SetDisks { } if !latest_meta.deleted && latest_meta.erasure.distribution.len() != parts_metadata.len() { + let distribution_len = latest_meta.erasure.distribution.len(); + let metadata_count = parts_metadata.len(); let err_str = format!( - "unexpected file distribution ({:?}) from metadata entries ({:?}), looks like backend disks have been manually modified refusing to heal {}/{}({})", - latest_meta.erasure.distribution, - parts_metadata.len(), + "unexpected file distribution length {distribution_len} for {metadata_count} metadata entries; backend disks may have been manually modified; refusing to heal {bucket}/{object}({version_id})" + ); + warn!( + event = EVENT_SET_DISK_HEAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, bucket, object, - version_id + version_id, + distribution_len, + metadata_count, + state = "invalid_distribution", + "Set disk object heal refused due to invalid erasure distribution" ); - warn!(err_str); let err = DiskError::other(err_str); return Ok(( self.default_heal_result(latest_meta, &errs, bucket, object, version_id).await, @@ -652,6 +821,9 @@ impl SetDisks { if !latest_meta.deleted && !latest_meta.is_remote() { let erasure_info = latest_meta.erasure.clone(); + let mut writer_failure_count = 0usize; + let mut first_writer_failure = None; + let mut writer_failure_warned = false; for (part_index, part) in latest_meta.parts.iter().enumerate() { let till_offset = erasure.shard_file_offset(0, part.size, part.size); @@ -666,9 +838,15 @@ impl SetDisks { let this_part_errs = Self::shuffle_check_parts(&data_errs_by_part[&part_index], &erasure_info.distribution); if this_part_errs[index] != CHECK_PART_SUCCESS { - info!( - "reading part {}: index={}, part_errs={:?}, skipping", - part.number, index, this_part_errs[index] + trace!( + event = EVENT_SET_DISK_HEAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, + part_number = part.number, + disk_index = index, + part_status = this_part_errs[index], + state = "source_shard_skipped", + "Set disk source shard skipped" ); readers.push(None); continue; @@ -726,28 +904,33 @@ impl SetDisks { // create writers for all disk positions, but only for outdated disks for (index, disk_op) in out_dated_disks.iter().enumerate() { if let Some(outdated_disk) = disk_op { - let writer = match create_bitrot_writer( - is_inline_buffer, - Some(outdated_disk), - RUSTFS_META_TMP_BUCKET, - &path_join_buf(&[ - &tmp_id.to_string(), - &dst_data_dir.to_string(), - &format!("part.{}", part.number), - ]), - erasure.shard_file_size(part.size as i64), - erasure.shard_size(), - HashAlgorithm::HighwayHash256S, - ) - .await + let writer_result = if let Some(error) = injected_heal_writer_error(bucket, object, index) { + Err(error) + } else { + create_bitrot_writer( + is_inline_buffer, + Some(outdated_disk), + RUSTFS_META_TMP_BUCKET, + &path_join_buf(&[ + &tmp_id.to_string(), + &dst_data_dir.to_string(), + &format!("part.{}", part.number), + ]), + erasure.shard_file_size(part.size as i64), + erasure.shard_size(), + HashAlgorithm::HighwayHash256S, + ) + .await + }; + let writer = match writer_result { Ok(writer) => writer, Err(err) => { - info!( - "create_bitrot_writer disk {}, err {:?}, skipping operation", - outdated_disk.to_string(), - err - ); + writer_failure_count += 1; + if first_writer_failure.is_none() { + first_writer_failure = + Some((part.number, index, heal_writer_error_summary(&err))); + } writers.push(None); continue; } @@ -761,6 +944,20 @@ impl SetDisks { // Heal each part. erasure.Heal() will write the healed // part to .rustfs/tmp/uuid/ which needs to be renamed // later to the final location. + if writer_failure_count > 0 + && writers.iter().all(Option::is_none) + && let Some(first_failure) = first_writer_failure.as_ref() + { + warn_heal_writer_failures( + bucket, + object, + version_id, + writer_failure_count, + "all_targets_unavailable", + first_failure, + ); + writer_failure_warned = true; + } if let Err(e) = erasure.heal(&mut writers, readers, part.size, &prefer).await { // Don't leak the partially-written healed shards in // .rustfs/tmp when heal fails midway (backlog#799 B20). @@ -805,6 +1002,16 @@ impl SetDisks { } if disks_to_heal_count == 0 { + if !writer_failure_warned && let Some(first_failure) = first_writer_failure.as_ref() { + warn_heal_writer_failures( + bucket, + object, + version_id, + writer_failure_count, + "all_targets_unavailable", + first_failure, + ); + } // Clean up healed shards written to .rustfs/tmp before bailing (B20). let _ = self.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_id).await; return Ok(( @@ -815,6 +1022,17 @@ impl SetDisks { )); } } + + if !writer_failure_warned && let Some(first_failure) = first_writer_failure.as_ref() { + warn_heal_writer_failures( + bucket, + object, + version_id, + writer_failure_count, + "partial_targets_unavailable", + first_failure, + ); + } } // Rename from tmp location to the actual location. // MinIO stops on the first RenameData error. RustFS intentionally @@ -1095,7 +1313,16 @@ impl SetDisks { async fn reclaim_orphan_data_dirs_best_effort(&self, bucket: &str, object: &str) { match self.reclaim_orphan_data_dirs(bucket, object).await { Ok(removed) if removed > 0 => { - info!(bucket, object, removed, "heal_object: reclaimed orphaned data directories"); + debug!( + event = EVENT_SET_DISK_HEAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, + bucket, + object, + removed, + state = "orphan_data_reclaimed", + "Set disk orphaned data reclaimed" + ); } Ok(_) => {} Err(e) => { @@ -1338,7 +1565,7 @@ impl SetDisks { Ok((result, None)) } - #[tracing::instrument(skip(self))] + #[tracing::instrument(level = "trace", skip(self), fields(bucket = %bucket, object = %object))] pub(in crate::set_disk) async fn heal_object_dir( &self, bucket: &str, @@ -1532,7 +1759,7 @@ impl crate::storage_api_contracts::heal::HealOperations for SetDisks { Ok(result) } - #[tracing::instrument(skip(self))] + #[tracing::instrument(level = "trace", skip(self, opts), fields(bucket = %bucket, object = %object, version_id = %version_id))] async fn heal_object( &self, bucket: &str, @@ -1636,8 +1863,8 @@ impl crate::storage_api_contracts::heal::HealOperations for SetDisks { #[cfg(test)] mod heal_result_report_tests { - use super::{DanglingCheckPartsFailure, DanglingDeleteFailure, DanglingDeleteSafety, SetDisks}; - use super::{HEAL_RENAME_INCOMPLETE, HealRenameFailureScope}; + use super::{DanglingCheckPartsFailure, DanglingDeleteFailure, DanglingDeleteSafety, SetDisks, heal_writer_error_summary}; + use super::{HEAL_RENAME_INCOMPLETE, HealRenameFailureScope, HealWriterFailureScope}; use crate::disk::endpoint::Endpoint; use crate::disk::error::DiskError; use crate::disk::format::FormatV3; @@ -1654,12 +1881,166 @@ mod heal_result_report_tests { }; use rustfs_common::heal_channel::{DriveState, HealOpts, HealScanMode}; use rustfs_filemeta::{BLOCK_SIZE_V2, FileInfo, ObjectPartInfo, TRANSITION_COMPLETE}; - use std::sync::Arc; + use std::sync::{Arc, Mutex}; use tempfile::TempDir; use time::OffsetDateTime; use tokio::sync::RwLock; + use tracing_subscriber::fmt::MakeWriter; use uuid::Uuid; + #[derive(Clone, Default)] + struct CapturedLogs { + buffer: Arc>>, + } + + struct CapturedLogWriter { + buffer: Arc>>, + } + + impl CapturedLogs { + fn contents(&self) -> String { + let buffer = self + .buffer + .lock() + .expect("captured logs mutex should not be poisoned") + .clone(); + String::from_utf8(buffer).expect("captured logs should be valid UTF-8") + } + } + + impl std::io::Write for CapturedLogWriter { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + self.buffer + .lock() + .expect("captured logs mutex should not be poisoned") + .extend_from_slice(buf); + Ok(buf.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + impl<'a> MakeWriter<'a> for CapturedLogs { + type Writer = CapturedLogWriter; + + fn make_writer(&'a self) -> Self::Writer { + CapturedLogWriter { + buffer: Arc::clone(&self.buffer), + } + } + } + + #[test] + fn heal_writer_error_summary_redacts_io_message() { + let error = DiskError::Io(std::io::Error::new(std::io::ErrorKind::PermissionDenied, "/sensitive/storage/path")); + + let summary = heal_writer_error_summary(&error); + + assert_eq!(summary, "io::PermissionDenied"); + assert!(!summary.contains("sensitive")); + } + + #[tokio::test(flavor = "current_thread")] + #[serial_test::serial] + async fn heal_writer_failures_emit_one_aggregate_warning_per_object() { + for (case, failed_target_count, expected_result, expect_error) in [ + ("partial", 1usize, "partial_targets_unavailable", false), + ("all", 2usize, "all_targets_unavailable", true), + ] { + let (temp_dirs, disks, set) = hermetic_set_disks_isolated(4).await; + let bucket = format!("heal-writer-{case}"); + let object = "object.bin"; + for disk in &disks { + disk.make_volume(&bucket).await.expect("bucket volume should be created"); + } + + let mut reader = PutObjReader::from_vec(vec![0x5a; 1024 * 1024]); + set.put_object(&bucket, object, &mut reader, &ObjectOptions::default()) + .await + .expect("source object should be written"); + let source = disks[2] + .read_version("", &bucket, object, "", &ReadOptions::default()) + .await + .expect("source metadata should be readable"); + let data_dir = source.data_dir.expect("non-inline source should have a data directory"); + let mut target_slots = [source.erasure.distribution[0] - 1, source.erasure.distribution[1] - 1]; + target_slots.sort_unstable(); + + for index in [0, 1] { + tokio::fs::remove_file( + temp_dirs[index] + .path() + .join(&bucket) + .join(object) + .join(data_dir.to_string()) + .join("part.1"), + ) + .await + .expect("target shard should be removed before heal"); + } + + let failed_slots = &target_slots[..failed_target_count]; + let logs = CapturedLogs::default(); + let subscriber = tracing_subscriber::fmt() + .with_max_level(tracing::Level::WARN) + .with_writer(logs.clone()) + .with_ansi(false) + .without_time() + .finish(); + let subscriber_guard = tracing::subscriber::set_default(subscriber); + let failure_scope = HealWriterFailureScope::install(&bucket, object, failed_slots, DiskError::DiskFull); + + let heal_outcome = set + .heal_object( + &bucket, + object, + "", + &HealOpts { + no_lock: true, + scan_mode: HealScanMode::Deep, + ..Default::default() + }, + ) + .await; + drop(failure_scope); + drop(subscriber_guard); + + assert_eq!( + heal_outcome.is_err(), + expect_error, + "{case}: aggregate heal result should match writer outcomes" + ); + let output = logs.contents(); + assert_eq!( + output.matches("Set disk object heal writer failures").count(), + 1, + "{case}: writer failures must emit one aggregate warning per object: {output}" + ); + assert!( + output.contains(&format!("writer_failure_count={failed_target_count}")), + "{case}: warning must report the aggregate failure count: {output}" + ); + assert!( + output.contains(&format!("first_disk_index={}", failed_slots[0])), + "{case}: warning must report the first failed target: {output}" + ); + assert!( + output.contains("first_part_number=1"), + "{case}: warning must report the first failed part: {output}" + ); + assert!( + output.contains(&format!("result=\"{expected_result}\"")), + "{case}: warning must distinguish partial from all-target failure: {output}" + ); + assert!( + output.contains("error=drive path full"), + "{case}: warning must preserve a redacted failure reason: {output}" + ); + } + } + async fn real_disk() -> (TempDir, Endpoint, DiskStore) { let dir = tempfile::tempdir().expect("tempdir should be created"); let endpoint = diff --git a/crates/ecstore/src/store/heal.rs b/crates/ecstore/src/store/heal.rs index fe99c1291..2c90e744e 100644 --- a/crates/ecstore/src/store/heal.rs +++ b/crates/ecstore/src/store/heal.rs @@ -14,6 +14,7 @@ use super::*; use crate::storage_api_contracts::heal::HealOperations as _; +use tracing::trace; const LOG_COMPONENT_ECSTORE: &str = "ecstore"; const LOG_SUBSYSTEM_HEAL: &str = "heal"; @@ -83,7 +84,7 @@ impl ECStore { Ok(res) } - #[instrument(skip(self))] + #[instrument(level = "trace", skip(self, opts), fields(bucket = %bucket, object = %object, version_id = %version_id))] pub(super) async fn handle_heal_object( &self, bucket: &str, @@ -91,7 +92,7 @@ impl ECStore { version_id: &str, opts: &HealOpts, ) -> Result<(HealResultItem, Option)> { - info!( + trace!( event = EVENT_HEAL_OBJECT_STARTED, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_HEAL, diff --git a/crates/ecstore/src/store/mod.rs b/crates/ecstore/src/store/mod.rs index 0c4be2476..120a6ae0c 100644 --- a/crates/ecstore/src/store/mod.rs +++ b/crates/ecstore/src/store/mod.rs @@ -787,7 +787,7 @@ impl crate::storage_api_contracts::heal::HealOperations for ECStore { async fn heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result { self.handle_heal_bucket(bucket, opts).await } - #[instrument(skip(self))] + #[instrument(level = "trace", skip(self, opts), fields(bucket = %bucket, object = %object, version_id = %version_id))] async fn heal_object( &self, bucket: &str, diff --git a/scripts/check_logging_guardrails.sh b/scripts/check_logging_guardrails.sh index 9511c0636..340a5e962 100755 --- a/scripts/check_logging_guardrails.sh +++ b/scripts/check_logging_guardrails.sh @@ -690,6 +690,102 @@ if [[ -n "$unmasked_revoke_fields" ]]; then exit 1 fi +heal_hotpath_files=( + "crates/ecstore/src/store/heal.rs" + "crates/ecstore/src/store/mod.rs" + "crates/ecstore/src/core/sets.rs" + "crates/ecstore/src/set_disk/ops/heal.rs" +) + +heal_function_pattern='async fn (handle_)?heal_object(_dir)?\(' +trace_heal_instrumentation_pattern='#\[(tracing::)?instrument\(\s*level\s*=\s*"trace"[^]]*\)\]\s*(pub(\([^)]*\))?\s+)?async fn (handle_)?heal_object(_dir)?\(' +heal_function_count="$(rg -n "$heal_function_pattern" "${heal_hotpath_files[@]}" | wc -l | tr -d ' ')" +trace_heal_instrumentation_count="$( + rg -U -o "$trace_heal_instrumentation_pattern" "${heal_hotpath_files[@]}" | + rg -c 'async fn (handle_)?heal_object' || true +)" +if [[ "$heal_function_count" != "$trace_heal_instrumentation_count" ]]; then + echo "❌ logging guardrail violation: per-object heal instrumentation must be TRACE-only" >&2 + echo "found $heal_function_count per-object heal functions but $trace_heal_instrumentation_count TRACE spans" >&2 + exit 1 +fi + +unexpected_heal_info="$( + rg -n '\binfo!' crates/ecstore/src/set_disk/ops/heal.rs crates/ecstore/src/erasure/coding/heal.rs | + rg -v 'set disk formats success, NoHealRequired' || true +)" +if [[ -n "$unexpected_heal_info" ]]; then + echo "❌ logging guardrail violation: per-object set-disk heal events must not be emitted at INFO" >&2 + echo "$unexpected_heal_info" >&2 + exit 1 +fi + +heal_info_event_pattern='info!\([^;]*(EVENT_HEAL_OBJECT_STARTED|state\s*=\s*"(metadata_corrupt|metadata_invalid|erasure_distribution_mismatch)"|disks_with_all_partsv2: metadata is corrupted|disks_with_all_partsv2: metadata is not valid|disks_with_all_partsv2: erasure distribution is not the same as onlineDisks)[^;]*\);' +if rg -n -U "$heal_info_event_pattern" crates/ecstore/src/store/heal.rs crates/ecstore/src/set_disk/mod.rs >/dev/null; then + echo "❌ logging guardrail violation: per-object heal diagnostics must not be emitted at INFO" >&2 + rg -n -U "$heal_info_event_pattern" crates/ecstore/src/store/heal.rs crates/ecstore/src/set_disk/mod.rs >&2 + exit 1 +fi + +raw_heal_metadata_pattern='(\?(parts_metadata|online_disks|out_dated_disks|latest_meta|meta)\b|(parts_metadata|online_disks|out_dated_disks|latest_meta|meta)\s*=\s*\?|(?:latest_meta|meta)\s*:\s*\{:#?\?\}|\{(parts_metadata|online_disks|out_dated_disks|latest_meta|meta):#?\?\}|\{:#?\?\}[^;]*(parts_metadata|online_disks|out_dated_disks|latest_meta|meta)\b|unexpected file distribution \(\{:#?\?\}\))' +if rg -n -U "$raw_heal_metadata_pattern" crates/ecstore/src/set_disk/ops/heal.rs crates/ecstore/src/set_disk/mod.rs >/dev/null; then + echo "❌ logging guardrail violation: heal logs must not dump raw object or disk metadata" >&2 + rg -n -U "$raw_heal_metadata_pattern" crates/ecstore/src/set_disk/ops/heal.rs crates/ecstore/src/set_disk/mod.rs >&2 + exit 1 +fi + +# Keep the matchers honest. These unsafe equivalents previously bypassed the +# guard when attributes/macros were multiline, long, or used structured Debug. +for fixture in \ + $'#[tracing::instrument(\n level = "info",\n skip(self),\n)]\nasync fn heal_object(' \ + $'#[instrument(skip(self), fields(level = "trace"))]\nasync fn heal_object('; do + fixture_function_count="$(printf '%s\n' "$fixture" | rg -c "$heal_function_pattern" || true)" + fixture_trace_count="$( + printf '%s\n' "$fixture" | + rg -U -o "$trace_heal_instrumentation_pattern" | + rg -c 'async fn (handle_)?heal_object' || true + )" + if [[ "$fixture_function_count" == "$fixture_trace_count" ]]; then + echo "❌ logging guardrail self-test failed: unsafe heal span was accepted" >&2 + echo "$fixture" >&2 + exit 1 + fi +done + +printf -v heal_info_padding '%*s' 600 '' +long_info_event_fixture="info!(${heal_info_padding} event = EVENT_HEAL_OBJECT_STARTED);" +if ! printf '%s\n' "$long_info_event_fixture" | rg -U "$heal_info_event_pattern" >/dev/null; then + echo "❌ logging guardrail self-test failed: long per-object INFO event was not detected" >&2 + exit 1 +fi + +for fixture in \ + 'info!("disks_with_all_partsv2: metadata is corrupted, object_name={}", object);' \ + 'info!("disks_with_all_partsv2: metadata is not valid, object_name={}", object);' \ + 'info!("disks_with_all_partsv2: erasure distribution is not the same as onlineDisks");'; do + if ! printf '%s\n' "$fixture" | rg -U "$heal_info_event_pattern" >/dev/null; then + echo "❌ logging guardrail self-test failed: retired per-object INFO event was not detected" >&2 + echo "$fixture" >&2 + exit 1 + fi +done + +for fixture in \ + 'debug!(latest_meta = ?latest_meta, "raw metadata")' \ + 'trace!("latest_meta: {:#?}", latest_meta)' \ + 'warn!("available disks: {:?}", online_disks)' \ + 'debug!("{:#?}", parts_metadata)' \ + 'debug!("raw metadata: {latest_meta:?}")' \ + 'trace!("raw metadata: {meta:#?}")' \ + 'warn!("raw disks: {online_disks:?}")' \ + 'warn!("unexpected file distribution ({:?})", online_disks)'; do + if ! printf '%s\n' "$fixture" | rg -U "$raw_heal_metadata_pattern" >/dev/null; then + echo "❌ logging guardrail self-test failed: raw heal metadata fixture was not detected" >&2 + echo "$fixture" >&2 + exit 1 + fi +done + # Secret material must never be interpolated into log or error strings. # Error messages are log content: they propagate via `?` and are printed by # startup/error logging far from the construction site, and a value that fails