From 82af181dcf345db25100ff8d5d1c46e010d15434 Mon Sep 17 00:00:00 2001 From: houseme Date: Thu, 11 Jun 2026 22:26:02 +0800 Subject: [PATCH] refactor(logging): unify governance runtime events (#3367) --- crates/audit/src/observability.rs | 27 +- crates/audit/src/registry.rs | 45 ++- crates/audit/src/system.rs | 29 +- .../bucket/lifecycle/bucket_lifecycle_ops.rs | 100 ++++- crates/ecstore/src/bucket/lifecycle/core.rs | 14 +- .../ecstore/src/bucket/lifecycle/evaluator.rs | 13 +- .../replication/replication_resyncer.rs | 324 +++++++++++++--- crates/ecstore/src/disk/disk_store.rs | 101 ++++- crates/ecstore/src/disk/local.rs | 352 ++++++++++++++++-- crates/ecstore/src/store/heal.rs | 39 +- crates/notify/src/bucket_config_manager.rs | 40 +- crates/notify/src/integration.rs | 34 +- crates/notify/src/rule_engine.rs | 24 +- crates/notify/src/runtime_facade.rs | 9 +- crates/obs/src/logging.rs | 15 + rustfs/src/main.rs | 351 ++++++++++++++--- scripts/check_logging_guardrails.sh | 24 ++ 17 files changed, 1357 insertions(+), 184 deletions(-) diff --git a/crates/audit/src/observability.rs b/crates/audit/src/observability.rs index 03461f8e0..0fecc05ae 100644 --- a/crates/audit/src/observability.rs +++ b/crates/audit/src/observability.rs @@ -29,6 +29,9 @@ use tokio::sync::RwLock; use tracing::info; const RUSTFS_AUDIT_METRICS_NAMESPACE: &str = "rustfs.audit."; +const LOG_COMPONENT_AUDIT: &str = "audit"; +const LOG_SUBSYSTEM_OBSERVABILITY: &str = "observability"; +const EVENT_AUDIT_OBSERVABILITY_STATE: &str = "audit_observability_state"; const M_AUDIT_EVENTS_TOTAL: &str = const_str::concat!(RUSTFS_AUDIT_METRICS_NAMESPACE, "events.total"); const M_AUDIT_EVENTS_FAILED: &str = const_str::concat!(RUSTFS_AUDIT_METRICS_NAMESPACE, "events.failed"); @@ -152,14 +155,26 @@ impl AuditMetrics { pub fn record_config_reload(&self) { self.config_reload_count.fetch_add(1, Ordering::Relaxed); counter!(M_AUDIT_CONFIG_RELOADS).increment(1); - info!("Audit configuration reloaded"); + info!( + event = EVENT_AUDIT_OBSERVABILITY_STATE, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_OBSERVABILITY, + state = "config_reloaded", + "Audit observability state updated" + ); } /// Records a system start pub fn record_system_start(&self) { self.system_start_count.fetch_add(1, Ordering::Relaxed); counter!(M_AUDIT_SYSTEM_STARTS).increment(1); - info!("Audit system started"); + info!( + event = EVENT_AUDIT_OBSERVABILITY_STATE, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_OBSERVABILITY, + state = "system_started", + "Audit observability state updated" + ); } /// Gets the current events per second (EPS) @@ -229,7 +244,13 @@ impl AuditMetrics { // Reset EPS to zero after reset gauge!(M_AUDIT_EPS).set(0.0); - info!("Audit metrics reset"); + info!( + event = EVENT_AUDIT_OBSERVABILITY_STATE, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_OBSERVABILITY, + state = "metrics_reset", + "Audit observability state updated" + ); } /// Generates a comprehensive metrics report diff --git a/crates/audit/src/registry.rs b/crates/audit/src/registry.rs index d29442bff..7da7e0a1d 100644 --- a/crates/audit/src/registry.rs +++ b/crates/audit/src/registry.rs @@ -19,6 +19,11 @@ use rustfs_targets::arn::TargetID; use rustfs_targets::{SharedTarget, Target, TargetError, TargetPluginRegistry, TargetRuntimeManager}; use tracing::info; +const LOG_COMPONENT_AUDIT: &str = "audit"; +const LOG_SUBSYSTEM_REGISTRY: &str = "registry"; +const EVENT_AUDIT_TARGET_REGISTRY_KEY_CREATED: &str = "audit_target_registry_key_created"; +const EVENT_AUDIT_TARGET_REGISTRY_STATE: &str = "audit_target_registry_state"; + /// Registry for managing audit targets pub struct AuditRegistry { /// Storage for created targets @@ -155,7 +160,15 @@ impl AuditRegistry { if let Some(target) = self.targets.remove(&target_id) && let Err(err) = target.close().await { - tracing::error!(target_id = %target_id, error = %err, "Failed to close target during shutdown"); + tracing::error!( + event = EVENT_AUDIT_TARGET_REGISTRY_STATE, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_REGISTRY, + target_id = %target_id, + state = "close_failed", + error = %err, + "Failed to close target during shutdown" + ); if first_error.is_none() { first_error = Some(err); } @@ -178,7 +191,15 @@ impl AuditRegistry { /// * `String` - The unique key for the target. pub fn create_key(&self, target_type: &str, target_id: &str) -> String { let key = TargetID::new(target_id.to_string(), target_type.to_string()); - info!(target_type = %target_type, "Create key for {}", key); + info!( + event = EVENT_AUDIT_TARGET_REGISTRY_KEY_CREATED, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_REGISTRY, + target_type = %target_type, + target_id = %target_id, + registry_key = %key, + "Created audit target registry key" + ); key.to_string() } @@ -193,7 +214,15 @@ impl AuditRegistry { pub fn enable_target(&self, target_type: &str, target_id: &str) -> AuditResult<()> { let key = self.create_key(target_type, target_id); if self.get_target(&key).is_some() { - info!("Target {}-{} enabled", target_type, target_id); + info!( + event = EVENT_AUDIT_TARGET_REGISTRY_STATE, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_REGISTRY, + target_type = %target_type, + target_id = %target_id, + state = "enabled", + "Audit target registry state changed" + ); Ok(()) } else { Err(AuditError::Configuration( @@ -214,7 +243,15 @@ impl AuditRegistry { pub fn disable_target(&self, target_type: &str, target_id: &str) -> AuditResult<()> { let key = self.create_key(target_type, target_id); if self.get_target(&key).is_some() { - info!("Target {}-{} disabled", target_type, target_id); + info!( + event = EVENT_AUDIT_TARGET_REGISTRY_STATE, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_REGISTRY, + target_type = %target_type, + target_id = %target_id, + state = "disabled", + "Audit target registry state changed" + ); Ok(()) } else { Err(AuditError::Configuration( diff --git a/crates/audit/src/system.rs b/crates/audit/src/system.rs index d6bea4d41..6deca7586 100644 --- a/crates/audit/src/system.rs +++ b/crates/audit/src/system.rs @@ -183,7 +183,15 @@ impl AuditSystem { Ok(()) } Err(e) => { - error!(error = %e, "Failed to create audit targets"); + error!( + event = EVENT_AUDIT_SYSTEM_STATE, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_SYSTEM, + state = "stopped", + reason = "target_creation_failed", + error = %e, + "Failed to create audit targets" + ); let mut state = self.state.write().await; *state = AuditSystemState::Stopped; Err(e) @@ -265,7 +273,15 @@ impl AuditSystem { // Stop all stream tasks first if let Err(e) = self.clear_runtime_targets().await { - error!(error = %e, "Failed to close some audit targets"); + error!( + event = EVENT_AUDIT_SYSTEM_STATE, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_SYSTEM, + state = "stopping", + reason = "target_shutdown_failed", + error = %e, + "Failed to close some audit targets" + ); } // Clear configuration @@ -454,7 +470,14 @@ impl AuditSystem { Ok(()) } Err(e) => { - error!(error = %e, "Failed to reload audit configuration"); + error!( + event = EVENT_AUDIT_CONFIG_RELOADED, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_SYSTEM, + state = "reload_failed", + error = %e, + "Failed to reload audit configuration" + ); Err(e) } } diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 48ad2bd6c..b22b862f6 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -87,6 +87,9 @@ const EVENT_LIFECYCLE_WORKER_STATE: &str = "lifecycle_worker_state"; const EVENT_LIFECYCLE_TRANSITION_COMPENSATION: &str = "lifecycle_transition_compensation"; const EVENT_LIFECYCLE_STALE_MULTIPART_CLEANUP: &str = "lifecycle_stale_multipart_cleanup"; const EVENT_LIFECYCLE_SCAN_SKIPPED: &str = "lifecycle_scan_skipped"; +const EVENT_LIFECYCLE_TIER_AUDIT: &str = "lifecycle_tier_audit"; +const EVENT_LIFECYCLE_TIER_OPERATION_FAILED: &str = "lifecycle_tier_operation_failed"; +const EVENT_LIFECYCLE_DELETE_FAILED: &str = "lifecycle_delete_failed"; pub type TimeFn = Arc Pin + Send>> + Send + Sync + 'static>; pub type TraceFn = @@ -956,12 +959,16 @@ impl TransitionState { pub async fn init(api: Arc) { let (configured, absolute_max, n) = resolve_transition_worker_count(); info!( + event = EVENT_LIFECYCLE_WORKER_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, configured_transition_workers = configured, absolute_max_workers = absolute_max, effective_transition_workers = n, transition_queue_capacity = GLOBAL_TransitionState.transition_queue_capacity, transition_queue_send_timeout_ms = GLOBAL_TransitionState.transition_queue_send_timeout.as_millis() as u64, - "transition worker count resolved" + state = "configured", + "Lifecycle worker state resolved" ); //let mut transition_state = GLOBAL_TransitionState.write().await; @@ -1032,14 +1039,24 @@ impl TransitionState { ..Default::default() }; - if let Err(err) = transition_object(api.clone(), &task.obj_info, LcAuditEvent::new(task.event.clone(), task.src.clone())).await { - global_metrics().record_scanner_transition_failed(1); - if !is_err_version_not_found(&err) && !is_err_object_not_found(&err) && !is_network_or_host_down(&err.to_string(), false) && !err.to_string().contains("use of closed network connection") { - error!("Transition to {} failed for {}/{} version:{} with {}", - task.event.storage_class, task.obj_info.bucket, task.obj_info.name, task.obj_info.version_id.map(|v| v.to_string()).unwrap_or_default(), err.to_string()); - } - // Send s3:ObjectTransition:Failed event - send_event(EventArgs { + if let Err(err) = transition_object(api.clone(), &task.obj_info, LcAuditEvent::new(task.event.clone(), task.src.clone())).await { + global_metrics().record_scanner_transition_failed(1); + if !is_err_version_not_found(&err) && !is_err_object_not_found(&err) && !is_network_or_host_down(&err.to_string(), false) && !err.to_string().contains("use of closed network connection") { + error!( + event = EVENT_LIFECYCLE_TIER_OPERATION_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + bucket = %task.obj_info.bucket, + object = %task.obj_info.name, + version_id = %task.obj_info.version_id.map(|v| v.to_string()).unwrap_or_default(), + tier = %task.event.storage_class, + operation = "transition_object", + error = %err, + "Lifecycle tier operation failed" + ); + } + // Send s3:ObjectTransition:Failed event + send_event(EventArgs { event_name: EventName::ObjectTransitionFailed.to_string(), bucket_name: obj_info_for_event.bucket.clone(), object: obj_info_for_event, @@ -1142,13 +1159,17 @@ impl TransitionState { GLOBAL_TransitionState.record_scanner_transition_state(); info!( + event = EVENT_LIFECYCLE_WORKER_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, requested_transition_workers = requested, effective_transition_workers = n, absolute_max_workers = absolute_max, previous_transition_workers = previous_num_workers, current_transition_workers = current_workers, pruned_finished_transition_workers = pruned_finished_workers, - "transition workers updated" + state = "resized", + "Lifecycle worker state updated" ); } } @@ -1279,7 +1300,15 @@ async fn read_stale_multipart_candidate( ) { Ok(file_info) => (Some(file_info.metadata), file_info.mod_time), Err(err) => { - warn!(path = %metadata_path, error = ?err, "failed to parse multipart metadata during stale cleanup"); + warn!( + event = EVENT_LIFECYCLE_STALE_MULTIPART_CLEANUP, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + path = %metadata_path, + error = ?err, + reason = "multipart_metadata_parse_failed", + "Skipped multipart metadata parse during stale cleanup" + ); (None, None) } }; @@ -1865,14 +1894,35 @@ pub async fn expire_transitioned_object( ) .await; if let Err(e) = &ret { - error!("Failed to delete remote transitioned object {}: {:?}", oi.transitioned_object.name, e); + error!( + event = EVENT_LIFECYCLE_TIER_OPERATION_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + bucket = %oi.bucket, + object = %oi.name, + tier = %oi.transitioned_object.tier, + tier_object = %oi.transitioned_object.name, + tier_version_id = %oi.transitioned_object.version_id, + operation = "delete_remote_transitioned_object", + error = ?e, + "Lifecycle tier operation failed" + ); } mark_delete_opts_skip_decommissioned_on_remote_success(&mut opts, ret.is_ok()); let dobj = match api.delete_object(&oi.bucket, &oi.name, opts).await { Ok(obj) => obj, Err(e) => { - error!("Failed to delete transitioned object {}/{}: {:?}", oi.bucket, oi.name, e); + error!( + event = EVENT_LIFECYCLE_DELETE_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + bucket = %oi.bucket, + object = %oi.name, + operation = "delete_transitioned_object", + error = ?e, + "Lifecycle delete failed" + ); // Return the original object info if deletion fails oi.clone() } @@ -1959,10 +2009,13 @@ pub fn audit_tier_actions(_tier: &str, bytes: i64) -> TimeFn { let tier = tier.clone(); Box::pin(async move { info!( + event = EVENT_LIFECYCLE_TIER_AUDIT, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, tier = %tier, bytes = bytes, - "ILM tier transition audit: completed transition of {} bytes to tier '{}'", - bytes, tier + state = "transition_completed", + "Lifecycle tier transition audit completed" ); }) }) @@ -2009,13 +2062,17 @@ pub async fn get_transitioned_object_reader( .await .map_err(|e| { tracing::error!( + event = EVENT_LIFECYCLE_TIER_OPERATION_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, bucket = %bucket, object = %object, tier = %oi.transitioned_object.tier, tier_object = %oi.transitioned_object.name, tier_version_id = %oi.transitioned_object.version_id, error = %e, - "tier GET failed" + operation = "tier_get", + "Lifecycle tier operation failed" ); e })?; @@ -2332,7 +2389,16 @@ pub async fn apply_expiry_on_non_transitioned_objects( let mut dobj = match api.delete_object(&oi.bucket, &encode_dir_object(&oi.name), opts).await { Ok(dobj) => dobj, Err(e) => { - error!("delete_object error: {:?}", e); + error!( + event = EVENT_LIFECYCLE_DELETE_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + bucket = %oi.bucket, + object = %oi.name, + operation = "delete_object", + error = ?e, + "Lifecycle delete failed" + ); return false; } }; diff --git a/crates/ecstore/src/bucket/lifecycle/core.rs b/crates/ecstore/src/bucket/lifecycle/core.rs index cb4962b86..c484faf31 100644 --- a/crates/ecstore/src/bucket/lifecycle/core.rs +++ b/crates/ecstore/src/bucket/lifecycle/core.rs @@ -23,13 +23,16 @@ use std::collections::HashMap; use std::sync::Arc; use time::macros::offset; use time::{self, Duration, OffsetDateTime}; -use tracing::{debug, info}; +use tracing::debug; use uuid::Uuid; use crate::store_api::ObjectInfo; pub const TRANSITION_COMPLETE: &str = "complete"; pub const TRANSITION_PENDING: &str = "pending"; +const LOG_COMPONENT_ECSTORE: &str = "ecstore"; +const LOG_SUBSYSTEM_LIFECYCLE: &str = "lifecycle"; +const EVENT_LIFECYCLE_EXPIRY_COMPUTED: &str = "lifecycle_expiry_computed"; const ERR_LIFECYCLE_NO_RULE: &str = "Lifecycle configuration should have at least one rule"; const ERR_LIFECYCLE_DUPLICATE_ID: &str = "Rule ID must be unique. Found same ID for more than one rule"; const _ERR_XML_NOT_WELL_FORMED: &str = @@ -800,7 +803,14 @@ impl LifecycleCalculate for Transition { pub fn expected_expiry_time(mod_time: OffsetDateTime, days: i32) -> OffsetDateTime { if days == 0 { - info!("expected_expiry_time: days=0, returning UNIX_EPOCH for immediate expiry"); + debug!( + event = EVENT_LIFECYCLE_EXPIRY_COMPUTED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + days, + result = "unix_epoch", + "Computed immediate lifecycle expiry time" + ); return OffsetDateTime::UNIX_EPOCH; // Return epoch time to ensure immediate expiry } let t = mod_time diff --git a/crates/ecstore/src/bucket/lifecycle/evaluator.rs b/crates/ecstore/src/bucket/lifecycle/evaluator.rs index a2387789a..9b759ac94 100644 --- a/crates/ecstore/src/bucket/lifecycle/evaluator.rs +++ b/crates/ecstore/src/bucket/lifecycle/evaluator.rs @@ -23,6 +23,10 @@ use crate::bucket::object_lock::objectlock_sys::is_object_locked_by_metadata; use crate::bucket::replication::ReplicationConfig; use rustfs_common::metrics::IlmAction; +const LOG_COMPONENT_ECSTORE: &str = "ecstore"; +const LOG_SUBSYSTEM_LIFECYCLE: &str = "lifecycle"; +const EVENT_LIFECYCLE_VERSION_SCAN_SKIPPED: &str = "lifecycle_version_scan_skipped"; + /// Evaluator - evaluates lifecycle policy on objects for the given lifecycle /// configuration, lock retention configuration and replication configuration. pub struct Evaluator { @@ -112,7 +116,14 @@ impl Evaluator { // events after DeleteAllVersionsAction* events[i] = event; - info!("eval_inner: skipping remaining versions' lifecycle events after DeleteAllVersionsAction*"); + info!( + event = EVENT_LIFECYCLE_VERSION_SCAN_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + reason = "delete_all_versions_action", + action = ?events[i].action, + "Skipped remaining lifecycle version scan" + ); break 'top_loop; } diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index acdf26b53..591f09f93 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -96,6 +96,10 @@ const EVENT_RESYNC_OBJECT_PROCESSED: &str = "replication_resync_object_processed const EVENT_RESYNC_RUNTIME_SKIPPED: &str = "replication_resync_runtime_skipped"; const EVENT_REPLICATION_DELETE_SKIPPED: &str = "replication_delete_skipped"; const EVENT_REPLICATION_FORCE_DELETE_SKIPPED: &str = "replication_force_delete_skipped"; +const EVENT_RESYNC_WORKER_SIGNAL_FAILED: &str = "replication_resync_worker_signal_failed"; +const EVENT_RESYNC_TASK_FAILED: &str = "replication_resync_task_failed"; +const EVENT_RESYNC_TARGET_OPERATION_FAILED: &str = "replication_resync_target_operation_failed"; +const EVENT_RESYNC_RUNTIME_CHANNEL_FAILED: &str = "replication_resync_runtime_channel_failed"; pub(crate) const REPLICATION_DIR: &str = ".replication"; pub(crate) const RESYNC_FILE_NAME: &str = "resync.bin"; @@ -463,7 +467,14 @@ impl ReplicationResyncer { for _ in 0..RESYNC_WORKER_COUNT { if let Err(err) = worker_tx.send(()) { - error!("Failed to send worker message: {}", err); + error!( + event = EVENT_RESYNC_WORKER_SIGNAL_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + reason = "worker_bootstrap_signal_failed", + error = %err, + "Failed to signal replication resync worker" + ); } } @@ -639,7 +650,15 @@ impl ReplicationResyncer { if update { if let Err(err) = save_resync_status(bucket, status, api.clone()).await { - error!("Failed to save resync status: {}", err); + error!( + event = EVENT_RESYNC_STATUS_UPDATE_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + reason = "persist_failed", + error = %err, + "Failed to persist resync status" + ); } else { last_update_times.insert(bucket.clone(), status.last_update.unwrap()); } @@ -654,10 +673,28 @@ impl ReplicationResyncer { async fn resync_bucket_mark_status(&self, status: ResyncStatusType, opts: ResyncOpts, storage: Arc) { if let Err(err) = self.mark_status(status, opts.clone(), storage.clone()).await { - error!("Failed to mark resync status: {}", err); + error!( + event = EVENT_RESYNC_STATUS_UPDATE_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %opts.bucket, + arn = %opts.arn, + reason = "mark_status_failed", + error = %err, + "Failed to update resync status" + ); } if let Err(err) = self.worker_tx.send(()) { - error!("Failed to send worker message: {}", err); + error!( + event = EVENT_RESYNC_WORKER_SIGNAL_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %opts.bucket, + arn = %opts.arn, + reason = "worker_release_signal_failed", + error = %err, + "Failed to signal replication resync worker" + ); } // TODO: Metrics } @@ -683,7 +720,16 @@ impl ReplicationResyncer { let cfg = match get_replication_config(&opts.bucket).await { Ok(cfg) => cfg, Err(err) => { - error!("Failed to get replication config: {}", err); + error!( + event = EVENT_RESYNC_CONFIG_LOOKUP_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %opts.bucket, + arn = %opts.arn, + reason = "replication_config_lookup_failed", + error = %err, + "Failed to look up replication config during resync" + ); self.resync_bucket_mark_status(ResyncStatusType::ResyncFailed, opts.clone(), storage.clone()) .await; return; @@ -722,8 +768,13 @@ impl ReplicationResyncer { if target_arns.len() != 1 { error!( - "replication resync failed for {} - arn specified {} is missing in the replication config", - opts.bucket, opts.arn + event = EVENT_RESYNC_CONFIG_LOOKUP_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %opts.bucket, + arn = %opts.arn, + reason = "target_arn_missing_from_replication_config", + "Replication resync target ARN missing from replication config" ); self.resync_bucket_mark_status(ResyncStatusType::ResyncFailed, opts.clone(), storage.clone()) .await; @@ -735,8 +786,13 @@ impl ReplicationResyncer { .await else { error!( - "replication resync failed for {} - arn specified {} is missing in the bucket targets", - opts.bucket, opts.arn + event = EVENT_RESYNC_RUNTIME_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %opts.bucket, + arn = %opts.arn, + reason = "target_client_missing", + "Replication resync target client missing from bucket targets" ); self.resync_bucket_mark_status(ResyncStatusType::ResyncFailed, opts.clone(), storage.clone()) .await; @@ -748,7 +804,16 @@ impl ReplicationResyncer { .mark_status(ResyncStatusType::ResyncStarted, opts.clone(), storage.clone()) .await { - error!("Failed to mark resync status: {}", e); + error!( + event = EVENT_RESYNC_STATUS_UPDATE_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %opts.bucket, + arn = %opts.arn, + reason = "mark_started_failed", + error = %e, + "Failed to update resync status" + ); } let (tx, mut rx) = tokio::sync::mpsc::channel(100); @@ -758,7 +823,16 @@ impl ReplicationResyncer { .walk(cancellation_token.clone(), &opts.bucket, "", tx.clone(), WalkOptions::default()) .await { - error!("Failed to walk bucket {}: {}", opts.bucket, err); + error!( + event = EVENT_RESYNC_RUNTIME_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %opts.bucket, + arn = %opts.arn, + reason = "walk_failed", + error = %err, + "Replication resync bucket walk failed" + ); self.resync_bucket_mark_status(ResyncStatusType::ResyncFailed, opts.clone(), storage.clone()) .await; return; @@ -890,7 +964,15 @@ impl ReplicationResyncer { } if let Err(err) = results_tx.send(st) { - error!("Failed to send resync status: {}", err); + error!( + event = EVENT_RESYNC_RUNTIME_CHANNEL_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket_name, + reason = "status_channel_send_failed", + error = %err, + "Failed to send resync status" + ); } } }); @@ -900,7 +982,16 @@ impl ReplicationResyncer { while let Some(res) = rx.recv().await { if let Some(err) = res.err { - error!("Failed to get object info: {}", err); + error!( + event = EVENT_RESYNC_RUNTIME_CHANNEL_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %opts.bucket, + arn = %opts.arn, + reason = "object_info_failed", + error = %err, + "Failed to receive resync object info" + ); self.resync_bucket_mark_status(ResyncStatusType::ResyncFailed, opts.clone(), storage.clone()) .await; return; @@ -938,7 +1029,16 @@ impl ReplicationResyncer { let worker_idx = sip_hash(&roi.name, RESYNC_WORKER_COUNT, &DEFAULT_SIP_HASH_KEY); if let Err(err) = worker_txs[worker_idx].send(roi).await { - error!("Failed to send object info to worker: {}", err); + error!( + event = EVENT_RESYNC_RUNTIME_CHANNEL_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %opts.bucket, + arn = %opts.arn, + reason = "worker_queue_send_failed", + error = %err, + "Failed to send resync object to worker" + ); self.resync_bucket_mark_status(ResyncStatusType::ResyncFailed, opts.clone(), storage.clone()) .await; return; @@ -1368,7 +1468,15 @@ pub async fn check_replicate_delete( return ReplicateDecision::default(); } Err(err) => { - error!("Failed to get replication config for bucket {}: {}", bucket, err); + error!( + event = EVENT_RESYNC_CONFIG_LOOKUP_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + reason = "replication_config_lookup_failed", + error = %err, + "Failed to look up replication config for delete replication" + ); return ReplicateDecision::default(); } }; @@ -1853,7 +1961,16 @@ pub async fn replicate_delete(dobj: DeletedObjectReplicationInfo, rinfos.targets.push(tgt_info); } Err(e) => { - error!("replicate_delete task failed: {}", e); + error!( + event = EVENT_RESYNC_TASK_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + object = %dobj.delete_object.object_name, + operation = "replicate_delete", + error = %e, + "Replication resync task failed" + ); send_event(EventArgs { event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), @@ -1967,7 +2084,17 @@ pub async fn replicate_delete(dobj: DeletedObjectReplicationInfo, }); } Err(e) => { - error!("failed to delete object for bucket:{} arn:{} error:{}", bucket, dobj.target_arn, e); + error!( + event = EVENT_RESYNC_TARGET_OPERATION_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + arn = %dobj.target_arn, + object = %dobj.delete_object.object_name, + operation = "apply_replication_delete_state", + error = %e, + "Replication target operation failed" + ); send_event(EventArgs { event_name, bucket_name: bucket.clone(), @@ -2105,8 +2232,14 @@ async fn replicate_force_delete_to_targets(dobj: &DeletedObjectRe Ok(ns_lock) => ns_lock, Err(e) => { warn!( - "replicate force-delete: failed to get ns lock bucket:{} object:{} error:{}", - bucket, object_name, e + event = EVENT_REPLICATION_FORCE_DELETE_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + object = %object_name, + reason = "ns_lock_create_failed", + error = %e, + "Skipping replication force-delete" ); send_event(EventArgs { event_name: EventName::ObjectReplicationNotTracked.to_string(), @@ -2128,8 +2261,14 @@ async fn replicate_force_delete_to_targets(dobj: &DeletedObjectRe Ok(guard) => guard, Err(e) => { warn!( - "replicate force-delete: failed to get write lock bucket:{} object:{} error:{}", - bucket, object_name, e + event = EVENT_REPLICATION_FORCE_DELETE_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + object = %object_name, + reason = "write_lock_failed", + error = %e, + "Skipping replication force-delete" ); send_event(EventArgs { event_name: EventName::ObjectReplicationNotTracked.to_string(), @@ -2189,7 +2328,16 @@ async fn replicate_force_delete_to_targets(dobj: &DeletedObjectRe join_set.spawn(async move { if BucketTargetSys::get().is_offline(&tgt_client.to_url()).await { - error!("replicate force-delete: target offline bucket:{} arn:{}", bucket, tgt_client.arn); + error!( + event = EVENT_REPLICATION_FORCE_DELETE_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + arn = %tgt_client.arn, + reason = "target_offline", + endpoint = %tgt_client.to_url(), + "Skipping replication force-delete" + ); send_event(EventArgs { event_name: EventName::ObjectReplicationFailed.to_string(), bucket_name: bucket.clone(), @@ -2223,8 +2371,15 @@ async fn replicate_force_delete_to_targets(dobj: &DeletedObjectRe .await { error!( - "replicate force-delete failed bucket:{} object:{} arn:{} error:{}", - bucket, object_name, tgt_client.arn, e + event = EVENT_RESYNC_TARGET_OPERATION_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + object = %object_name, + arn = %tgt_client.arn, + operation = "force_delete_remove_object", + error = %e, + "Replication target operation failed" ); send_event(EventArgs { event_name: EventName::ObjectReplicationFailed.to_string(), @@ -2244,7 +2399,16 @@ async fn replicate_force_delete_to_targets(dobj: &DeletedObjectRe while let Some(result) = join_set.join_next().await { if let Err(e) = result { - error!("replicate force-delete task panicked: {}", e); + error!( + event = EVENT_RESYNC_TASK_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + object = %object_name, + operation = "force_delete", + error = %e, + "Replication resync task failed" + ); } } } @@ -2369,13 +2533,17 @@ async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_cli } Err(e) => { warn!( + event = EVENT_RESYNC_TARGET_OPERATION_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, bucket = tgt_client.bucket, object = dobj.delete_object.object_name, version_id = ?version_id, delete_marker = dobj.delete_object.delete_marker, is_version_purge, error = %e, - "replicate_delete_to_target failed" + operation = "replicate_delete_to_target", + "Replication target operation failed" ); rinfo.error = Some(e.to_string()); if !is_version_purge { @@ -2423,7 +2591,15 @@ pub async fn replicate_object(roi: ReplicateObjectInfo, storage: return; } Err(err) => { - error!("Failed to get replication config for bucket {}: {}", bucket, err); + error!( + event = EVENT_RESYNC_CONFIG_LOOKUP_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + reason = "replication_config_lookup_failed", + error = %err, + "Failed to look up replication config for object replication" + ); send_event(EventArgs { event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), @@ -2491,7 +2667,16 @@ pub async fn replicate_object(roi: ReplicateObjectInfo, storage: rinfos.targets.push(tgt_info); } Err(e) => { - error!("replicate_object task failed: {}", e); + error!( + event = EVENT_RESYNC_TASK_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + object = %object, + operation = "replicate_object", + error = %e, + "Replication resync task failed" + ); send_event(EventArgs { event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), @@ -2749,15 +2934,30 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { Err(e2) => { rinfo.error = Some(e2.to_string()); warn!( - "replication head_object fallback failed bucket:{} arn:{} error:{}", - bucket, tgt_client.arn, e2 + event = EVENT_RESYNC_TARGET_OPERATION_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + arn = %tgt_client.arn, + operation = "head_object_fallback", + error = %e2, + "Replication target operation failed" ); return rinfo; } } } else { rinfo.error = Some(e.to_string()); - warn!("replication head_object failed bucket:{} arn:{} error:{}", bucket, tgt_client.arn, e); + warn!( + event = EVENT_RESYNC_TARGET_OPERATION_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + arn = %tgt_client.arn, + operation = "head_object", + error = %e, + "Replication target operation failed" + ); return rinfo; } } @@ -2772,8 +2972,14 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { Ok((put_opts, is_mp)) => (put_opts, is_mp), Err(e) => { warn!( - "failed to get put replication opts for bucket:{} arn:{} error:{}", - bucket, tgt_client.arn, e + event = EVENT_RESYNC_TARGET_OPERATION_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + arn = %tgt_client.arn, + operation = "build_put_options", + error = %e, + "Replication target operation failed" ); send_event(EventArgs { event_name: EventName::ObjectReplicationNotTracked.to_string(), @@ -2823,8 +3029,16 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { rinfo.replication_status = ReplicationStatusType::Failed; rinfo.error = Some(err.to_string()); warn!( - "replication put_object failed src_bucket={} dest_bucket={} object={} err={:?}", - bucket, tgt_client.bucket, object, err + event = EVENT_RESYNC_TARGET_OPERATION_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + target_bucket = %tgt_client.bucket, + arn = %tgt_client.arn, + object = %object, + operation = "put_object", + error = ?err, + "Replication target operation failed" ); // TODO: check offline @@ -3022,10 +3236,15 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { && object_info.version_id.is_none() { warn!( - "unable to replicate {}/{} Newer version exists on target {}", - bucket, - object, - tgt_client.to_url() + event = EVENT_RESYNC_RUNTIME_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + object = %object, + arn = %tgt_client.arn, + endpoint = %tgt_client.to_url(), + reason = "newer_target_version_exists", + "Skipping replication because newer target version exists" ); send_event(EventArgs { event_name: EventName::ObjectReplicationNotTracked.to_string(), @@ -3139,8 +3358,14 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { Err(e) => { rinfo.error = Some(e.to_string()); warn!( - "failed to get put replication opts for bucket:{} arn:{} error:{}", - bucket, tgt_client.arn, e + event = EVENT_RESYNC_TARGET_OPERATION_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + arn = %tgt_client.arn, + operation = "build_put_options", + error = %e, + "Replication target operation failed" ); send_event(EventArgs { event_name: EventName::ObjectReplicationNotTracked.to_string(), @@ -3191,6 +3416,17 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { } { rinfo.replication_status = ReplicationStatusType::Failed; rinfo.error = Some(err.to_string()); + warn!( + event = EVENT_RESYNC_TARGET_OPERATION_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + arn = %tgt_client.arn, + object = %object, + operation = "put_object", + error = ?err, + "Replication target operation failed" + ); rinfo.duration = (OffsetDateTime::now_utc() - start_time).unsigned_abs(); // TODO: check offline @@ -3270,7 +3506,11 @@ fn wrap_with_bandwidth_monitor_with_header( } else { WARNED_MONITOR_UNINIT.call_once(|| { warn!( - "Global bucket monitor uninitialized; proceeding with unthrottled replication (bandwidth limits will be ignored)" + event = EVENT_RESYNC_RUNTIME_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + reason = "bucket_monitor_uninitialized", + "Skipping replication bandwidth monitor because global bucket monitor is uninitialized" ) }); stream diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index f7e6e2d4b..226b67f6b 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/crates/ecstore/src/disk/disk_store.rs @@ -45,6 +45,11 @@ use uuid::Uuid; /// Disk health status constants const DISK_HEALTH_OK: u32 = 0; const DISK_HEALTH_FAULTY: u32 = 1; +const LOG_COMPONENT_ECSTORE: &str = "ecstore"; +const LOG_SUBSYSTEM_DISK: &str = "disk"; +const EVENT_DISK_HEALTH_CHECK_FAILED: &str = "disk_health_check_failed"; +const EVENT_DISK_RECOVERY_PROBE_STATE: &str = "disk_recovery_probe_state"; +const EVENT_DISK_TIMEOUT_POLICY_FALLBACK: &str = "disk_timeout_policy_fallback"; #[derive(Clone, Copy, Debug, Eq, PartialEq)] enum TimeoutHealthAction { @@ -118,10 +123,14 @@ fn resolve_drive_timeout_profile_from_env() -> DriveTimeoutProfile { return profile; } warn!( + event = EVENT_DISK_TIMEOUT_POLICY_FALLBACK, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK, env = rustfs_config::ENV_DRIVE_TIMEOUT_PROFILE, value = %raw, default = rustfs_config::DEFAULT_DRIVE_TIMEOUT_PROFILE, - "Invalid drive timeout profile; falling back to default" + reason = "invalid_timeout_profile", + "Disk timeout policy fell back to default" ); DriveTimeoutProfile::parse(rustfs_config::DEFAULT_DRIVE_TIMEOUT_PROFILE).unwrap_or(DriveTimeoutProfile::Default) } @@ -211,10 +220,14 @@ fn resolve_drive_timeout_health_policy_from_env() -> TimeoutHealthPolicy { return policy; } warn!( + event = EVENT_DISK_TIMEOUT_POLICY_FALLBACK, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK, env = rustfs_config::ENV_DRIVE_TIMEOUT_HEALTH_ACTION, value = %raw, default = rustfs_config::DEFAULT_DRIVE_TIMEOUT_HEALTH_ACTION, - "Invalid drive timeout health action policy; falling back to default" + reason = "invalid_health_action_policy", + "Disk timeout policy fell back to default" ); TimeoutHealthPolicy::parse(rustfs_config::DEFAULT_DRIVE_TIMEOUT_HEALTH_ACTION).unwrap_or(TimeoutHealthPolicy::MarkFailure) } @@ -719,7 +732,14 @@ impl LocalDiskWrapper { && health.mark_failure(&disk.endpoint(), "active_health_check_failed") { // Health check failed, disk is considered faulty - warn!("health check: failed, disk is considered faulty"); + warn!( + event = EVENT_DISK_HEALTH_CHECK_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK, + endpoint = %disk.endpoint(), + reason = "faulty_disk", + "Disk health check marked disk faulty" + ); health.increment_waiting(); // Balance the increment from failed operation @@ -756,9 +776,14 @@ impl LocalDiskWrapper { // Verify data integrity if read_data.len() != test_data.len() { warn!( - "health check: test file data length mismatch: expected {} bytes, got {}", - test_data.len(), - read_data.len() + event = EVENT_DISK_HEALTH_CHECK_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK, + endpoint = %disk.endpoint(), + reason = "data_length_mismatch", + expected_bytes = test_data.len(), + actual_bytes = read_data.len(), + "Disk health check detected data length mismatch" ); if check_faulty_only { return Ok(()); @@ -787,7 +812,15 @@ impl LocalDiskWrapper { Ok(result) => match result { Ok(()) => Ok(()), Err(e) => { - warn!("health check: failed: {:?}", e); + warn!( + event = EVENT_DISK_HEALTH_CHECK_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK, + endpoint = %disk.endpoint(), + reason = "operation_failed", + error = ?e, + "Disk health check failed" + ); if e == DiskError::FaultyDisk { return Err(e); @@ -798,7 +831,15 @@ impl LocalDiskWrapper { }, Err(_) => { // Timeout occurred - warn!("health check: timeout after {:?}", timeout_duration); + warn!( + event = EVENT_DISK_HEALTH_CHECK_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK, + endpoint = %disk.endpoint(), + reason = "timeout", + timeout_secs = timeout_duration.as_secs(), + "Disk health check timed out" + ); Err(DiskError::FaultyDisk) } } @@ -835,17 +876,40 @@ impl LocalDiskWrapper { Ok(_) => { let state_before = health.runtime_state(); let is_online = health.mark_recovery_success(&disk.endpoint(), "recovery_probe_success"); - info!("Disk {} recovery probe succeeded; state={:?}", disk.to_string(), state_before); + info!( + event = EVENT_DISK_RECOVERY_PROBE_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK, + endpoint = %disk.endpoint(), + state = "probe_succeeded", + previous_state = ?state_before, + "Disk recovery probe state changed" + ); if !is_online { continue; } - info!("Disk {} is back online", disk.to_string()); + info!( + event = EVENT_DISK_RECOVERY_PROBE_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK, + endpoint = %disk.endpoint(), + state = "online", + "Disk recovery probe restored disk online" + ); health.decrement_waiting(); return; } Err(e) => { health.mark_failure(&disk.endpoint(), "recovery_probe_failed"); - warn!("Disk {} still faulty: {:?}", disk.to_string(), e); + warn!( + event = EVENT_DISK_RECOVERY_PROBE_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK, + endpoint = %disk.endpoint(), + state = "still_faulty", + error = ?e, + "Disk recovery probe detected disk still faulty" + ); } } } @@ -946,7 +1010,14 @@ impl LocalDiskWrapper { { // Check if disk is faulty if self.health.is_faulty() { - warn!("local disk {} health is faulty, returning error", self.to_string()); + warn!( + event = EVENT_DISK_HEALTH_CHECK_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK, + endpoint = %self.endpoint(), + reason = "disk_marked_faulty", + "Disk health check rejected operation because disk is marked faulty" + ); return Err(DiskError::FaultyDisk); } @@ -996,10 +1067,14 @@ impl LocalDiskWrapper { ) .increment(1); warn!( + event = EVENT_DISK_HEALTH_CHECK_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK, endpoint = %self.endpoint(), op, timeout_ms = timeout_duration.as_millis(), - "Local disk operation timed out" + reason = "operation_timeout", + "Disk operation timed out" ); Err(DiskError::Timeout) } diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 5806750a9..f12337fef 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -70,15 +70,31 @@ const ENV_BITROT_SIZE_MISMATCH_RETRY_COUNT: &str = "RUSTFS_BITROT_SIZE_MISMATCH_ const ENV_BITROT_SIZE_MISMATCH_RETRY_DELAY_MS: &str = "RUSTFS_BITROT_SIZE_MISMATCH_RETRY_DELAY_MS"; const DEFAULT_BITROT_SIZE_MISMATCH_RETRY_COUNT: u64 = 2; const DEFAULT_BITROT_SIZE_MISMATCH_RETRY_DELAY_MS: u64 = 100; +const LOG_COMPONENT_ECSTORE: &str = "ecstore"; +const LOG_SUBSYSTEM_DISK_LOCAL: &str = "disk_local"; +const EVENT_DISK_LOCAL_STARTUP_CLEANUP: &str = "disk_local_startup_cleanup"; +const EVENT_DISK_LOCAL_BACKGROUND_CLEANUP: &str = "disk_local_background_cleanup"; +const EVENT_DISK_LOCAL_SCAN_FAILED: &str = "disk_local_scan_failed"; +const EVENT_DISK_LOCAL_RENAME_REJECTED: &str = "disk_local_rename_rejected"; +const EVENT_DISK_LOCAL_READ_VERSION_FALLBACK: &str = "disk_local_read_version_fallback"; +const EVENT_DISK_LOCAL_DELETE_FAILED: &str = "disk_local_delete_failed"; +const EVENT_DISK_LOCAL_CHECK_PARTS: &str = "disk_local_check_parts"; +const EVENT_DISK_LOCAL_ACCESS_FAILED: &str = "disk_local_access_failed"; +const EVENT_DISK_LOCAL_VOLUME_SETUP_FAILED: &str = "disk_local_volume_setup_failed"; +const EVENT_DISK_LOCAL_FORMAT_DECODE_FAILED: &str = "disk_local_format_decode_failed"; fn log_startup_disk_io_error(stage: &str, path: &Path, err: &IoError) { warn!( + event = EVENT_DISK_LOCAL_STARTUP_CLEANUP, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, stage, path = %path.display(), error_kind = ?err.kind(), raw_os_error = ?err.raw_os_error(), error = ?err, - "local disk startup filesystem operation failed" + state = "io_failed", + "Disk local startup filesystem operation failed" ); } @@ -89,10 +105,14 @@ fn log_startup_disk_error(stage: &str, path: &Path, err: &DiskError) { } warn!( + event = EVENT_DISK_LOCAL_STARTUP_CLEANUP, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, stage, path = %path.display(), error = ?err, - "local disk startup operation failed" + state = "failed", + "Disk local startup operation failed" ); } @@ -474,7 +494,15 @@ impl LocalDisk { { startup_cleanup_ready.store(1, Ordering::Release); startup_cleanup_notify.notify_waiters(); - warn!(root = ?root, error = ?err, "failed to cleanup temporary data during disk startup"); + warn!( + event = EVENT_DISK_LOCAL_STARTUP_CLEANUP, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + root = ?root, + state = "failed", + error = ?err, + "Disk local startup cleanup failed" + ); } // Use optimized path resolution instead of absolutize_virtually @@ -519,7 +547,15 @@ impl LocalDisk { { Ok(ids) => ids, Err(err) => { - warn!(root = ?root, error = ?err, "failed to resolve physical device ids for disk root"); + warn!( + event = EVENT_DISK_LOCAL_STARTUP_CLEANUP, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + root = ?root, + state = "physical_device_id_lookup_failed", + error = ?err, + "Disk local startup metadata lookup failed" + ); Vec::new() } }; @@ -614,14 +650,37 @@ impl LocalDisk { tokio::select! { _ = interval.tick() => { if let Err(err) = Self::cleanup_deleted_objects(root.clone()).await { - error!("cleanup_deleted_objects error: {:?}", err); + error!( + event = EVENT_DISK_LOCAL_BACKGROUND_CLEANUP, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + task = "deleted_objects", + state = "failed", + error = ?err, + "Disk local background cleanup failed" + ); } if let Err(err) = Self::cleanup_stale_tmp_objects(root.clone()).await { - error!("cleanup_stale_tmp_objects error: {:?}", err); + error!( + event = EVENT_DISK_LOCAL_BACKGROUND_CLEANUP, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + task = "stale_tmp_objects", + state = "failed", + error = ?err, + "Disk local background cleanup failed" + ); } } _ = exit_rx.recv() => { - info!("cleanup_deleted_objects_loop exit"); + info!( + event = EVENT_DISK_LOCAL_BACKGROUND_CLEANUP, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + task = "deleted_objects_loop", + state = "stopped", + "Disk local background cleanup loop stopped" + ); break; } } @@ -683,9 +742,13 @@ impl LocalDisk { debug!(disk = %self.endpoint, "startup cleanup barrier released before walk_dir"); } else { warn!( + event = EVENT_DISK_LOCAL_STARTUP_CLEANUP, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, disk = %self.endpoint, timeout_ms = STARTUP_CLEANUP_WAIT_TIMEOUT.as_millis(), - "startup cleanup barrier timed out; continuing walk_dir" + state = "timed_out", + "Disk local startup cleanup barrier timed out" ); } } @@ -1059,7 +1122,15 @@ impl LocalDisk { ErrorKind::NotFound => (), ErrorKind::DirectoryNotEmpty => (), kind => { - warn!("delete_file remove_dir {:?} err {}", &delete_path, kind.to_string()); + warn!( + event = EVENT_DISK_LOCAL_DELETE_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + path = ?delete_path, + operation = "remove_dir", + error_kind = %kind, + "Disk local delete failed" + ); return Err(Error::other(FileAccessDeniedWithContext { path: delete_path.clone(), source: err, @@ -1073,7 +1144,15 @@ impl LocalDisk { match err.kind() { ErrorKind::NotFound => (), _ => { - warn!("delete_file remove_file {:?} err {:?}", &delete_path, &err); + warn!( + event = EVENT_DISK_LOCAL_DELETE_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + path = ?delete_path, + operation = "remove_file", + error = ?err, + "Disk local delete failed" + ); return Err(Error::other(FileAccessDeniedWithContext { path: delete_path.clone(), source: err, @@ -1182,7 +1261,14 @@ impl LocalDisk { && let Err(er) = access(volume_dir.as_ref()).await && er.kind() == ErrorKind::NotFound { - warn!("read_all_data_with_dmtime os err {:?}", &er); + warn!( + event = EVENT_DISK_LOCAL_READ_VERSION_FALLBACK, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + reason = "read_all_data_with_dmtime_volume_not_found", + error = ?er, + "Disk local read fallback failed" + ); return Err(DiskError::VolumeNotFound); } @@ -1408,13 +1494,17 @@ impl LocalDisk { Ok(()) => return Ok(()), Err(err) if attempt < retry_count && is_bitrot_size_mismatch_error(&err) => { info!( + event = EVENT_DISK_LOCAL_CHECK_PARTS, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, path = %part_path.display(), expected_size = part_size, actual_size = file_size, retry_attempt = attempt + 1, retry_count, retry_delay_ms = retry_delay.as_millis(), - "bitrot verify observed shard size mismatch; retrying" + state = "bitrot_retry", + "Disk local check_parts state changed" ); tokio::time::sleep(retry_delay).await; } @@ -1463,7 +1553,15 @@ impl LocalDisk { Ok(res) => res, Err(e) => { if e != DiskError::VolumeNotFound && e != Error::FileNotFound { - error!("scan list_dir {}, err {:?}", ¤t, &e); + error!( + event = EVENT_DISK_LOCAL_SCAN_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + path = %current, + operation = "list_dir", + error = ?e, + "Disk local scan failed" + ); return Err(e); } @@ -1610,11 +1708,20 @@ impl LocalDisk { }) .await?; + let scan_path = pop.clone(); if opts.recursive && let Err(er) = Box::pin(self.scan_dir(pop, prefix.clone(), opts, out, objs_returned, skip_object, dir_to_skip)).await { - error!("scan_dir err {:?}", er); + error!( + event = EVENT_DISK_LOCAL_SCAN_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + path = %scan_path, + operation = "scan_dir", + error = ?er, + "Disk local scan failed" + ); } } @@ -1702,11 +1809,20 @@ impl LocalDisk { }) .await?; + let scan_path = dir.clone(); if opts.recursive && let Err(er) = Box::pin(self.scan_dir(dir, prefix.clone(), opts, out, objs_returned, skip_object, dir_to_skip)).await { - warn!("scan_dir err {:?}", &er); + warn!( + event = EVENT_DISK_LOCAL_SCAN_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + path = %scan_path, + operation = "scan_dir", + error = ?er, + "Disk local recursive scan failed" + ); } } @@ -1933,7 +2049,13 @@ impl DiskAPI for LocalDisk { let b = fs::read(&self.format_path).await.map_err(to_unformatted_disk_error)?; let fm = FormatV3::try_from(b.as_slice()).map_err(|e| { - warn!("decode format.json err {:?}", e); + warn!( + event = EVENT_DISK_LOCAL_FORMAT_DECODE_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + error = ?e, + "Disk local format decode failed" + ); DiskError::CorruptedBackend })?; @@ -2045,11 +2167,29 @@ impl DiskAPI for LocalDisk { if resp.results[i] == CHECK_PART_UNKNOWN && let Some(err) = err { - error!("verify_file: failed to bitrot verify file: {:?}, error: {:?}", &part_path, &err); + error!( + event = EVENT_DISK_LOCAL_CHECK_PARTS, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + path = ?part_path, + part_number = part.number, + state = "bitrot_verify_failed", + error = ?err, + "Disk local check_parts state changed" + ); if err == DiskError::FileAccessDenied { continue; } - info!("part unknown, disk: {}, path: {:?}", self.to_string(), part_path); + info!( + event = EVENT_DISK_LOCAL_CHECK_PARTS, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + endpoint = %self.endpoint, + path = ?part_path, + part_number = part.number, + state = "unknown", + "Disk local check_parts state changed" + ); } } @@ -2142,7 +2282,15 @@ impl DiskAPI for LocalDisk { .as_str(), )?; - info!("check_parts: part_path: {:?}", &part_path); + debug!( + event = EVENT_DISK_LOCAL_CHECK_PARTS, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + path = ?part_path, + part_number = part.number, + state = "checking", + "Disk local check_parts state changed" + ); match lstat(&part_path).await { Ok(st) => { @@ -2158,7 +2306,16 @@ impl DiskAPI for LocalDisk { resp.results[i] = CHECK_PART_SUCCESS; } Err(err) => { - info!("check_parts: failed to stat file: {:?}, error: {:?}", &part_path, &err); + debug!( + event = EVENT_DISK_LOCAL_CHECK_PARTS, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + path = ?part_path, + part_number = part.number, + state = "part_stat_failed", + error = ?err, + "Disk local check_parts state changed" + ); let e: DiskError = to_file_error(err).into(); @@ -2172,7 +2329,16 @@ impl DiskAPI for LocalDisk { } resp.results[i] = CHECK_PART_FILE_NOT_FOUND; } else { - error!("check_parts: failed to stat file: {:?}, error: {:?}", &file_path, &e); + error!( + event = EVENT_DISK_LOCAL_CHECK_PARTS, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + path = ?file_path, + part_number = part.number, + state = "file_stat_failed", + error = ?e, + "Disk local check_parts state changed" + ); } continue; } @@ -2198,8 +2364,13 @@ impl DiskAPI for LocalDisk { if !src_is_dir && dst_is_dir || src_is_dir && !dst_is_dir { warn!( - "rename_part src and dst must be both dir or file src_is_dir:{}, dst_is_dir:{}", - src_is_dir, dst_is_dir + event = EVENT_DISK_LOCAL_RENAME_REJECTED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + reason = "src_dst_type_mismatch", + src_is_dir, + dst_is_dir, + "Disk local rename rejected" ); return Err(DiskError::FileAccessDenied); } @@ -2223,7 +2394,14 @@ impl DiskAPI for LocalDisk { if let Some(meta) = meta_op && !meta.is_dir() { - warn!("rename_part src is not dir {:?}", &src_file_path); + warn!( + event = EVENT_DISK_LOCAL_RENAME_REJECTED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + reason = "src_expected_dir_missing", + path = ?src_file_path, + "Disk local rename rejected" + ); return Err(DiskError::FileAccessDenied); } @@ -2231,7 +2409,14 @@ impl DiskAPI for LocalDisk { } else { let meta = lstat_std(&src_file_path).map_err(|e| -> DiskError { to_file_error(e).into() })?; if meta.is_dir() { - warn!("rename_part src is dir {:?}", &src_file_path); + warn!( + event = EVENT_DISK_LOCAL_RENAME_REJECTED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + reason = "src_unexpected_dir", + path = ?src_file_path, + "Disk local rename rejected" + ); return Err(DiskError::FileAccessDenied); } } @@ -2240,7 +2425,14 @@ impl DiskAPI for LocalDisk { let dst_meta = lstat_std(&dst_file_path).map_err(|e| -> DiskError { to_file_error(e).into() })?; if src_is_dir != dst_meta.is_dir() { - warn!("rename_part dst type changed after rename {:?}", &dst_file_path); + warn!( + event = EVENT_DISK_LOCAL_RENAME_REJECTED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + reason = "dst_type_changed_after_rename", + path = ?dst_file_path, + "Disk local rename rejected" + ); return Err(DiskError::FileAccessDenied); } @@ -2396,10 +2588,16 @@ impl DiskAPI for LocalDisk { let end_offset = offset.checked_add(length).ok_or(DiskError::FileCorrupt)?; if meta.len() < end_offset as u64 { error!( - "read_file_stream: file size is less than offset + length {} + {} = {}", + event = EVENT_DISK_LOCAL_READ_VERSION_FALLBACK, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + volume, + path, offset, length, - meta.len() + actual_size = meta.len(), + reason = "read_file_stream_out_of_bounds", + "Disk local read fallback failed" ); return Err(DiskError::FileCorrupt); } @@ -2438,10 +2636,16 @@ impl DiskAPI for LocalDisk { let end_offset = offset.checked_add(length).ok_or(DiskError::FileCorrupt)?; if meta.len() < end_offset as u64 { error!( - "read_file_zero_copy: file size is less than offset + length {} + {} = {}", + event = EVENT_DISK_LOCAL_READ_VERSION_FALLBACK, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + volume, + path, offset, length, - meta.len() + actual_size = meta.len(), + reason = "read_file_zero_copy_out_of_bounds", + "Disk local read fallback failed" ); return Err(DiskError::FileCorrupt); } @@ -2672,7 +2876,15 @@ impl DiskAPI for LocalDisk { if !skip_access_checks(src_volume) && let Err(e) = super::fs::access_std(&src_volume_dir) { - info!("access checks failed, src_volume_dir: {:?}, err: {}", src_volume_dir, e.to_string()); + info!( + event = EVENT_DISK_LOCAL_ACCESS_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + path = ?src_volume_dir, + operation = "rename_data_src_access", + error = %e, + "Disk local access check failed" + ); return Err(to_access_error(e, DiskError::VolumeAccessDenied).into()); } @@ -2680,7 +2892,15 @@ impl DiskAPI for LocalDisk { if !skip_access_checks(dst_volume) && let Err(e) = super::fs::access_std(&dst_volume_dir) { - info!("access checks failed, dst_volume_dir: {:?}, err: {}", dst_volume_dir, e.to_string()); + info!( + event = EVENT_DISK_LOCAL_ACCESS_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + path = ?dst_volume_dir, + operation = "rename_data_dst_access", + error = %e, + "Disk local access check failed" + ); return Err(to_access_error(e, DiskError::VolumeAccessDenied).into()); } @@ -2771,8 +2991,14 @@ impl DiskAPI for LocalDisk { { let _ = self.delete_file(&dst_volume_dir, dst_data_path, false, false).await; info!( - "rename all failed src_data_path: {:?}, dst_data_path: {:?}, err: {:?}", - src_data_path, dst_data_path, err + event = EVENT_DISK_LOCAL_RENAME_REJECTED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + reason = "rename_all_data_path_failed", + src_path = ?src_data_path, + dst_path = ?dst_data_path, + error = ?err, + "Disk local rename flow failed" ); return Err(err); } @@ -2781,7 +3007,16 @@ impl DiskAPI for LocalDisk { if let Some((_, dst_data_path)) = has_data_dir_path.as_ref() { let _ = self.delete_file(&dst_volume_dir, dst_data_path, false, false).await; } - info!("rename all failed err: {:?}", err); + info!( + event = EVENT_DISK_LOCAL_RENAME_REJECTED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + reason = "rename_all_metadata_failed", + src_path = ?src_file_path, + dst_path = ?dst_file_path, + error = ?err, + "Disk local rename flow failed" + ); return Err(err); } @@ -2797,7 +3032,14 @@ impl DiskAPI for LocalDisk { ) .await { - info!("write_all_private failed err: {:?}", err); + info!( + event = EVENT_DISK_LOCAL_RENAME_REJECTED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + reason = "write_old_metadata_failed", + error = ?err, + "Disk local rename flow failed" + ); return Err(err); } @@ -2908,7 +3150,15 @@ impl DiskAPI for LocalDisk { if let Err(e) = self.make_volume(vol).await && e != DiskError::VolumeExists { - error!("local disk make volumes failed: {e}"); + error!( + event = EVENT_DISK_LOCAL_VOLUME_SETUP_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + volume = vol, + operation = "make_volumes", + error = %e, + "Disk local volume setup failed" + ); return Err(e); } // TODO: health check @@ -2929,7 +3179,15 @@ impl DiskAPI for LocalDisk { os::make_dir_all(&volume_dir, self.root.as_path()).await?; return Ok(()); } - error!("local disk make volume failed: {e}"); + error!( + event = EVENT_DISK_LOCAL_VOLUME_SETUP_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + volume, + operation = "make_volume", + error = %e, + "Disk local volume setup failed" + ); return Err(to_volume_error(e).into()); } @@ -3136,7 +3394,15 @@ impl DiskAPI for LocalDisk { let part_path = self.get_object_path(volume, part_path.as_str())?; let data = self.read_all_data(volume, volume_dir, part_path.clone()).await.map_err(|e| { - warn!("read_version read_all_data {:?} failed: {e}", part_path); + warn!( + event = EVENT_DISK_LOCAL_READ_VERSION_FALLBACK, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + path = ?part_path, + reason = "inline_data_read_failed", + error = %e, + "Disk local read_version fallback failed" + ); e })?; fi.data = Some(Bytes::from(data)); @@ -3296,7 +3562,13 @@ impl DiskAPI for LocalDisk { res.mod_time = match meta.modified() { Ok(md) => Some(OffsetDateTime::from(md)), Err(_) => { - warn!("Not supported modified on this platform"); + warn!( + event = EVENT_DISK_LOCAL_FORMAT_DECODE_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_LOCAL, + reason = "modified_time_unsupported", + "Disk local modified time is unsupported on this platform" + ); None } }; diff --git a/crates/ecstore/src/store/heal.rs b/crates/ecstore/src/store/heal.rs index 5e524efab..ccef0de02 100644 --- a/crates/ecstore/src/store/heal.rs +++ b/crates/ecstore/src/store/heal.rs @@ -14,10 +14,14 @@ use super::*; +const LOG_COMPONENT_ECSTORE: &str = "ecstore"; +const LOG_SUBSYSTEM_HEAL: &str = "heal"; +const EVENT_HEAL_FORMAT_COMPLETED: &str = "heal_format_completed"; +const EVENT_HEAL_OBJECT_STARTED: &str = "heal_object_started"; + impl ECStore { #[instrument(skip(self))] pub(super) async fn handle_heal_format(&self, dry_run: bool) -> Result<(HealResultItem, Option)> { - info!("heal_format"); let mut r = HealResultItem { heal_item_type: HealItemType::Metadata.to_string(), detail: "disk-format".to_string(), @@ -43,10 +47,27 @@ impl ECStore { r.after.drives.append(&mut result.after.drives); } if count_no_heal == self.pools.len() { - info!("heal format success, NoHealRequired"); + info!( + event = EVENT_HEAL_FORMAT_COMPLETED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_HEAL, + dry_run, + result = "no_heal_required", + pool_count = self.pools.len(), + "Heal format completed" + ); return Ok((r, Some(StorageError::NoHealRequired))); } - info!("heal format success result: {:?}", r); + info!( + event = EVENT_HEAL_FORMAT_COMPLETED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_HEAL, + dry_run, + result = "healed_or_inspected", + disk_count = r.disk_count, + set_count = r.set_count, + "Heal format completed" + ); Ok((r, None)) } @@ -65,7 +86,17 @@ impl ECStore { version_id: &str, opts: &HealOpts, ) -> Result<(HealResultItem, Option)> { - info!("ECStore heal_object"); + info!( + event = EVENT_HEAL_OBJECT_STARTED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_HEAL, + bucket = %bucket, + object = %object, + version_id = %version_id, + remove = opts.remove, + scan_mode = ?opts.scan_mode, + "Heal object started" + ); let object = encode_dir_object(object); let mut futures = Vec::with_capacity(self.pools.len()); diff --git a/crates/notify/src/bucket_config_manager.rs b/crates/notify/src/bucket_config_manager.rs index bbfe1ebd9..f06db42f8 100644 --- a/crates/notify/src/bucket_config_manager.rs +++ b/crates/notify/src/bucket_config_manager.rs @@ -21,6 +21,11 @@ use rustfs_s3_types::EventName; use std::sync::Arc; use tracing::{debug, info, warn}; +const LOG_COMPONENT_NOTIFY: &str = "notify"; +const LOG_SUBSYSTEM_BUCKET_CONFIG: &str = "bucket_config"; +const EVENT_NOTIFY_BUCKET_CONFIG_VALIDATION: &str = "notify_bucket_config_validation"; +const EVENT_NOTIFY_BUCKET_CONFIG_LOADED: &str = "notify_bucket_config_loaded"; + #[derive(Clone)] pub struct NotifyBucketConfigManager { notifier: Arc, @@ -57,24 +62,53 @@ impl NotifyBucketConfigManager { if arn_list.is_empty() { return Err(NotificationError::Configuration(notify_configuration_hint())); } - info!("Available ARNs: {:?}", arn_list); + debug!( + event = EVENT_NOTIFY_BUCKET_CONFIG_VALIDATION, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_BUCKET_CONFIG, + bucket = %bucket, + region = %cfg.region, + available_arn_count = arn_list.len(), + "Loaded available notify target ARNs for bucket config validation" + ); if let Err(e) = cfg.validate(&cfg.region, &arn_list) { - debug!("Bucket notification config validation region:{} failed: {}", &cfg.region, e); + debug!( + event = EVENT_NOTIFY_BUCKET_CONFIG_VALIDATION, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_BUCKET_CONFIG, + bucket = %bucket, + region = %cfg.region, + error = %e, + result = "validation_failed", + "Bucket notification config validation failed" + ); if !matches!(e, ParseConfigError::ArnNotFound(_)) { return Err(NotificationError::BucketNotification(e.to_string())); } warn!( + event = EVENT_NOTIFY_BUCKET_CONFIG_VALIDATION, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_BUCKET_CONFIG, bucket = %bucket, region = %cfg.region, error = %e, + result = "missing_target_arn", "Bucket notification config references missing target ARN; keeping compatibility and loading remaining rules" ); } self.subscriber_view.apply_bucket_config(bucket, cfg); self.rule_engine.set_bucket_rules(bucket, cfg.get_rules_map().clone()).await; - info!("Loaded notification config for bucket: {}", bucket); + info!( + event = EVENT_NOTIFY_BUCKET_CONFIG_LOADED, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_BUCKET_CONFIG, + bucket = %bucket, + region = %cfg.region, + rule_count = cfg.get_rules_map().inner().len(), + "Loaded bucket notification config" + ); Ok(()) } diff --git a/crates/notify/src/integration.rs b/crates/notify/src/integration.rs index 35eb90cca..6db3aa9ef 100644 --- a/crates/notify/src/integration.rs +++ b/crates/notify/src/integration.rs @@ -36,6 +36,11 @@ const METRIC_NOTIFICATION_CURRENT_SEND_IN_PROGRESS: &str = "rustfs_notification_ const METRIC_NOTIFICATION_EVENTS_ERRORS_TOTAL: &str = "rustfs_notification_events_errors_total"; const METRIC_NOTIFICATION_EVENTS_SENT_TOTAL: &str = "rustfs_notification_events_sent_total"; const METRIC_NOTIFICATION_EVENTS_SKIPPED_TOTAL: &str = "rustfs_notification_events_skipped_total"; +const LOG_COMPONENT_NOTIFY: &str = "notify"; +const LOG_SUBSYSTEM_INTEGRATION: &str = "integration"; +const EVENT_NOTIFY_SYSTEM_DROP: &str = "notify_system_drop"; +const EVENT_NOTIFY_SYSTEM_SHUTDOWN_METRIC: &str = "notify_system_shutdown_metric"; +const EVENT_NOTIFY_SYSTEM_STATUS_SNAPSHOT: &str = "notify_system_status_snapshot"; #[derive(Clone)] pub struct LiveEventBatch { @@ -362,7 +367,13 @@ impl NotificationSystem { impl Drop for NotificationSystem { fn drop(&mut self) { // Asynchronous operation cannot be used here, but logs can be recorded. - info!("Notify the system instance to be destroyed"); + info!( + event = EVENT_NOTIFY_SYSTEM_DROP, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_INTEGRATION, + state = "dropping", + "Notification system instance is being dropped" + ); let snapshot = self.snapshot_metrics(); for (name, value, is_gauge) in [ @@ -376,15 +387,28 @@ impl Drop for NotificationSystem { } else { counter!(name).absolute(value); } - info!("shutdown metric {}={}", name, value); + info!( + event = EVENT_NOTIFY_SYSTEM_SHUTDOWN_METRIC, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_INTEGRATION, + metric_name = name, + metric_value = value, + metric_kind = if is_gauge { "gauge" } else { "counter" }, + "Notification shutdown metric snapshot" + ); } let status = self.get_status(); for (key, value) in status { - info!("key:{}, value:{}", key, value); + info!( + event = EVENT_NOTIFY_SYSTEM_STATUS_SNAPSHOT, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_INTEGRATION, + status_key = %key, + status_value = %value, + "Notification system status snapshot" + ); } - - info!("Notification system status at shutdown:"); } } diff --git a/crates/notify/src/rule_engine.rs b/crates/notify/src/rule_engine.rs index 0c5d1072d..792f091ca 100644 --- a/crates/notify/src/rule_engine.rs +++ b/crates/notify/src/rule_engine.rs @@ -20,6 +20,10 @@ use starshard::{AsyncShardedHashMap, DEFAULT_SHARDS, SnapshotMode}; use std::sync::Arc; use tracing::info; +const LOG_COMPONENT_NOTIFY: &str = "notify"; +const LOG_SUBSYSTEM_RULE_ENGINE: &str = "rule_engine"; +const EVENT_NOTIFY_RULES_UPDATED: &str = "notify_bucket_rules_updated"; + fn decoded_object_key_for_matching(object_key: &str) -> Option { if !object_key.contains('%') { return None; @@ -52,12 +56,21 @@ impl NotifyRuleEngine { } pub async fn set_bucket_rules(&self, bucket: &str, rules_map: RulesMap) { + let event_count = rules_map.iter_events().count(); if rules_map.is_empty() { self.bucket_rules_map.remove(&bucket.to_string()).await; } else { self.bucket_rules_map.insert(bucket.to_string(), rules_map).await; } - info!("Updated notification rules for bucket: {}", bucket); + info!( + event = EVENT_NOTIFY_RULES_UPDATED, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_RULE_ENGINE, + bucket = %bucket, + state = "updated", + event_count, + "Updated bucket notification rules" + ); } pub async fn get_bucket_rules(&self, bucket: &str) -> Option { @@ -66,7 +79,14 @@ impl NotifyRuleEngine { pub async fn clear_bucket_rules(&self, bucket: &str) { if self.bucket_rules_map.remove(&bucket.to_string()).await.is_some() { - info!("Removed all notification rules for bucket: {}", bucket); + info!( + event = EVENT_NOTIFY_RULES_UPDATED, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_RULE_ENGINE, + bucket = %bucket, + state = "removed", + "Removed bucket notification rules" + ); } } diff --git a/crates/notify/src/runtime_facade.rs b/crates/notify/src/runtime_facade.rs index 5cad6209c..098279d22 100644 --- a/crates/notify/src/runtime_facade.rs +++ b/crates/notify/src/runtime_facade.rs @@ -24,6 +24,7 @@ use tracing::{debug, info}; const LOG_COMPONENT_NOTIFY: &str = "notify"; const LOG_SUBSYSTEM_RUNTIME: &str = "runtime"; const EVENT_NOTIFY_RUNTIME_LIFECYCLE: &str = "notify_runtime_lifecycle"; +const EVENT_NOTIFY_RUNTIME_SHUTDOWN_FAILED: &str = "notify_runtime_shutdown_failed"; #[derive(Clone)] pub struct NotifyRuntimeFacade { @@ -143,7 +144,13 @@ impl NotifyRuntimeFacade { .shutdown(target_list.runtime_mut(), &mut replay_workers) .await { - tracing::error!(error = %err, "Failed to shutdown notify runtime cleanly"); + tracing::error!( + event = EVENT_NOTIFY_RUNTIME_SHUTDOWN_FAILED, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_RUNTIME, + error = %err, + "Failed to shutdown notify runtime cleanly" + ); } } tokio::time::sleep(Duration::from_millis(500)).await; diff --git a/crates/obs/src/logging.rs b/crates/obs/src/logging.rs index 2900a512c..4c683474e 100644 --- a/crates/obs/src/logging.rs +++ b/crates/obs/src/logging.rs @@ -217,5 +217,20 @@ mod tests { "Event processing initiated for {} targets for bucket: {}", ], ); + assert_no_unmasked_access_key_logging( + "crates/notify/src/bucket_config_manager.rs", + &["Available ARNs: {:?}", "Loaded notification config for bucket: {}"], + ); + assert_no_unmasked_access_key_logging( + "crates/notify/src/rule_engine.rs", + &[ + "Updated notification rules for bucket: {}", + "Removed all notification rules for bucket: {}", + ], + ); + assert_no_unmasked_access_key_logging( + "crates/audit/src/observability.rs", + &["Audit configuration reloaded", "Audit system started", "Audit metrics reset"], + ); } } diff --git a/rustfs/src/main.rs b/rustfs/src/main.rs index ab4cecf31..ad78e1c93 100644 --- a/rustfs/src/main.rs +++ b/rustfs/src/main.rs @@ -77,18 +77,36 @@ const LOG_COMPONENT_MAIN: &str = "main"; const LOG_SUBSYSTEM_STARTUP: &str = "startup"; const LOG_SUBSYSTEM_LICENSE: &str = "license"; const LOG_SUBSYSTEM_AUTH: &str = "auth"; +const LOG_SUBSYSTEM_STORAGE: &str = "storage"; const EVENT_EXTERNAL_ENV_COMPAT_CONFLICT: &str = "external_env_compat_conflict"; const EVENT_EXTERNAL_ENV_COMPAT_APPLIED: &str = "external_env_compat_applied"; const EVENT_DIAL9_RUNTIME_STATUS: &str = "dial9_runtime_status"; const EVENT_RUNTIME_LICENSE_STATUS: &str = "runtime_license_status"; +const EVENT_OBSERVABILITY_GUARD_SET_FAILED: &str = "observability_guard_set_failed"; const EVENT_TLS_OUTBOUND_INITIALIZED: &str = "tls_outbound_initialized"; +const EVENT_TLS_OUTBOUND_INITIALIZATION_FAILED: &str = "tls_outbound_initialization_failed"; const EVENT_DEFAULT_CREDENTIALS_DETECTED: &str = "default_credentials_detected"; const EVENT_SERVER_CONFIG_SANITIZED: &str = "server_config_sanitized"; const EVENT_SERVER_STARTING: &str = "server_starting"; +const EVENT_SERVER_RUNTIME_FAILED: &str = "server_runtime_failed"; +const EVENT_ACTION_CREDENTIALS_INITIALIZATION_FAILED: &str = "action_credentials_initialization_failed"; +const EVENT_ENDPOINT_PARSING_STARTED: &str = "endpoint_parsing_started"; +const EVENT_STORAGE_POOL_FORMATTING: &str = "storage_pool_formatting"; +const EVENT_STORAGE_POOL_HOST_RISK: &str = "storage_pool_host_risk"; +const EVENT_PROTOCOL_SYSTEM_STATE: &str = "protocol_system_state"; +const EVENT_AUDIT_SYSTEM_STATE: &str = "audit_system_state"; +const EVENT_DEADLOCK_DETECTOR_STATE: &str = "deadlock_detector_state"; const EVENT_KEYSTONE_AUTH_INITIALIZED: &str = "keystone_auth_initialized"; +const EVENT_KEYSTONE_AUTH_INITIALIZATION_FAILED: &str = "keystone_auth_initialization_failed"; const EVENT_OIDC_INITIALIZATION_FAILED: &str = "oidc_initialization_failed"; +const EVENT_NOTIFICATION_SYSTEM_INITIALIZATION_FAILED: &str = "notification_system_initialization_failed"; const EVENT_BACKGROUND_SERVICES_CONFIGURED: &str = "background_services_configured"; const EVENT_SERVER_READY: &str = "server_ready"; +const EVENT_SHUTDOWN_SIGNAL_RECEIVED: &str = "shutdown_signal_received"; +const EVENT_BACKGROUND_SERVICE_SHUTDOWN: &str = "background_service_shutdown"; +const EVENT_EVENT_NOTIFIER_SHUTDOWN: &str = "event_notifier_shutdown"; +const EVENT_PROFILING_SHUTDOWN: &str = "profiling_shutdown"; +const EVENT_SERVER_SHUTDOWN_STATE: &str = "server_shutdown_state"; #[cfg(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64"))] #[global_allocator] static GLOBAL: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc; @@ -173,7 +191,14 @@ async fn async_main() -> Result<()> { debug!(target: "rustfs::main", "Global observability guard set successfully."); } Err(e) => { - error!("Failed to set global observability guard: {}", e); + error!( + target: "rustfs::main", + event = EVENT_OBSERVABILITY_GUARD_SET_FAILED, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + error = %e, + "Failed to store global observability guard" + ); return Err(e); } } @@ -248,7 +273,15 @@ async fn async_main() -> Result<()> { ); } Err(e) => { - error!("Failed to initialize TLS from {}: {}", tls_path, e); + error!( + target: "rustfs::main", + event = EVENT_TLS_OUTBOUND_INITIALIZATION_FAILED, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + tls_path, + error = %e, + "Failed to initialize TLS outbound material" + ); return Err(Error::other(e.to_string())); } } @@ -261,7 +294,14 @@ async fn async_main() -> Result<()> { match run(*config).await { Ok(_) => Ok(()), Err(e) => { - error!("Server encountered an error and is shutting down: {}", e); + error!( + target: "rustfs::main", + event = EVENT_SERVER_RUNTIME_FAILED, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + error = %e, + "Server runtime failed" + ); Err(e) } } @@ -330,7 +370,14 @@ async fn run(config: rustfs::config::Config) -> Result<()> { } Err(e) => { let msg = format!("init global action credentials failed: {e:?}"); - error!(target: "rustfs::main::run","{msg}"); + error!( + target: "rustfs::main::run", + event = EVENT_ACTION_CREDENTIALS_INITIALIZATION_FAILED, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_AUTH, + error = %e, + "Failed to initialize global action credentials" + ); return Err(Error::other(msg)); } }; @@ -342,9 +389,12 @@ async fn run(config: rustfs::config::Config) -> Result<()> { // For RPC info!( target: "rustfs::main::run", + event = EVENT_ENDPOINT_PARSING_STARTED, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STORAGE, server_address = %server_address, - volumes = ?config.volumes, - "starting endpoint parsing" + volume_count = config.volumes.len(), + "Starting endpoint parsing" ); let (endpoint_pools, setup_type) = EndpointServerPools::from_volumes(server_address.clone().as_str(), config.volumes.clone()) .await @@ -386,19 +436,26 @@ async fn run(config: rustfs::config::Config) -> Result<()> { for (i, eps) in endpoint_pools.as_ref().iter().enumerate() { info!( target: "rustfs::main::run", + event = EVENT_STORAGE_POOL_FORMATTING, component = LOG_COMPONENT_MAIN, - subsystem = LOG_SUBSYSTEM_STARTUP, + subsystem = LOG_SUBSYSTEM_STORAGE, pool_id = i + 1, set_count = eps.set_count, drives_per_set = eps.drives_per_set, - "Formatting {}st pool, {} set(s), {} drives per set.", - i + 1, - eps.set_count, - eps.drives_per_set + "Formatting storage pool" ); if eps.drives_per_set > 1 { - warn!(target: "rustfs::main::run","WARNING: Host local has more than 0 drives of set. A host failure will result in data becoming unavailable."); + warn!( + target: "rustfs::main::run", + event = EVENT_STORAGE_POOL_HOST_RISK, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STORAGE, + pool_id = i + 1, + drives_per_set = eps.drives_per_set, + risk = "host_failure_data_unavailable", + "Detected multi-drive local set host failure risk" + ); } } @@ -485,15 +542,37 @@ async fn run(config: rustfs::config::Config) -> Result<()> { #[cfg(feature = "ftps")] let ftp_shutdown_tx = match init_ftp_system().await { Ok(Some(tx)) => { - info!("FTP system initialized successfully"); + info!( + event = EVENT_PROTOCOL_SYSTEM_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + protocol = "ftp", + state = "initialized", + "Protocol system state changed" + ); Some(tx) } Ok(None) => { - info!("FTP system disabled"); + info!( + event = EVENT_PROTOCOL_SYSTEM_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + protocol = "ftp", + state = "disabled", + "Protocol system state changed" + ); None } Err(e) => { - error!("Failed to initialize FTP system: {}", e); + error!( + event = EVENT_PROTOCOL_SYSTEM_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + protocol = "ftp", + state = "initialization_failed", + error = %e, + "Protocol system state changed" + ); return Err(Error::other(e)); } }; @@ -505,15 +584,37 @@ async fn run(config: rustfs::config::Config) -> Result<()> { #[cfg(feature = "ftps")] let ftps_shutdown_tx = match init_ftps_system().await { Ok(Some(tx)) => { - info!("FTPS system initialized successfully"); + info!( + event = EVENT_PROTOCOL_SYSTEM_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + protocol = "ftps", + state = "initialized", + "Protocol system state changed" + ); Some(tx) } Ok(None) => { - info!("FTPS system disabled"); + info!( + event = EVENT_PROTOCOL_SYSTEM_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + protocol = "ftps", + state = "disabled", + "Protocol system state changed" + ); None } Err(e) => { - error!("Failed to initialize FTPS system: {}", e); + error!( + event = EVENT_PROTOCOL_SYSTEM_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + protocol = "ftps", + state = "initialization_failed", + error = %e, + "Protocol system state changed" + ); return Err(Error::other(e)); } }; @@ -525,15 +626,37 @@ async fn run(config: rustfs::config::Config) -> Result<()> { #[cfg(feature = "webdav")] let webdav_shutdown_tx = match init_webdav_system().await { Ok(Some(tx)) => { - info!("WebDAV system initialized successfully"); + info!( + event = EVENT_PROTOCOL_SYSTEM_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + protocol = "webdav", + state = "initialized", + "Protocol system state changed" + ); Some(tx) } Ok(None) => { - info!("WebDAV system disabled"); + info!( + event = EVENT_PROTOCOL_SYSTEM_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + protocol = "webdav", + state = "disabled", + "Protocol system state changed" + ); None } Err(e) => { - error!("Failed to initialize WebDAV system: {}", e); + error!( + event = EVENT_PROTOCOL_SYSTEM_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + protocol = "webdav", + state = "initialization_failed", + error = %e, + "Protocol system state changed" + ); return Err(Error::other(e)); } }; @@ -545,15 +668,37 @@ async fn run(config: rustfs::config::Config) -> Result<()> { #[cfg(feature = "sftp")] let sftp_shutdown_tx = match init_sftp_system().await { Ok(Some(tx)) => { - info!("SFTP system initialized successfully"); + info!( + event = EVENT_PROTOCOL_SYSTEM_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + protocol = "sftp", + state = "initialized", + "Protocol system state changed" + ); Some(tx) } Ok(None) => { - info!("SFTP system disabled"); + info!( + event = EVENT_PROTOCOL_SYSTEM_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + protocol = "sftp", + state = "disabled", + "Protocol system state changed" + ); None } Err(e) => { - error!("Failed to initialize SFTP system: {}", e); + error!( + event = EVENT_PROTOCOL_SYSTEM_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + protocol = "sftp", + state = "initialization_failed", + error = %e, + "Protocol system state changed" + ); return Err(Error::other(e)); } }; @@ -569,17 +714,46 @@ async fn run(config: rustfs::config::Config) -> Result<()> { // Start the audit system match start_audit_system().await { - Ok(_) => info!(target: "rustfs::main::run","Audit system started successfully."), - Err(e) => error!(target: "rustfs::main::run","Failed to start audit system: {}", e), + Ok(_) => info!( + target: "rustfs::main::run", + event = EVENT_AUDIT_SYSTEM_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + state = "started", + "Audit system state changed" + ), + Err(e) => error!( + target: "rustfs::main::run", + event = EVENT_AUDIT_SYSTEM_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + state = "start_failed", + error = %e, + "Audit system state changed" + ), } // Initialize deadlock detector if enabled let detector = rustfs::storage::deadlock_detector::get_deadlock_detector(); if detector.is_enabled() { detector.start(); - info!(target: "rustfs::main::run","Deadlock detector started successfully."); + info!( + target: "rustfs::main::run", + event = EVENT_DEADLOCK_DETECTOR_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + state = "started", + "Deadlock detector state changed" + ); } else { - info!(target: "rustfs::main::run","Deadlock detector disabled."); + info!( + target: "rustfs::main::run", + event = EVENT_DEADLOCK_DETECTOR_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + state = "disabled", + "Deadlock detector state changed" + ); } let buckets_list = store @@ -625,7 +799,13 @@ async fn run(config: rustfs::config::Config) -> Result<()> { "Initialized Keystone authentication" ), Err(e) => { - error!("Failed to initialize Keystone authentication: {}", e); + error!( + event = EVENT_KEYSTONE_AUTH_INITIALIZATION_FAILED, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_AUTH, + error = %e, + "Failed to initialize Keystone authentication" + ); // Continue without Keystone - fall back to standard auth } } @@ -646,7 +826,13 @@ async fn run(config: rustfs::config::Config) -> Result<()> { // Initialize the global notification system new_global_notification_sys(endpoint_pools.clone()).await.map_err(|err| { - error!("new_global_notification_sys failed {:?}", &err); + error!( + event = EVENT_NOTIFICATION_SYSTEM_INITIALIZATION_FAILED, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + error = ?err, + "Failed to initialize notification system" + ); Error::other(err) })?; @@ -788,6 +974,9 @@ async fn handle_shutdown( info!( target: "rustfs::main::handle_shutdown", + event = EVENT_SHUTDOWN_SIGNAL_RECEIVED, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, signal = shutdown_signal.log_label(), "Shutdown signal received in main thread" ); @@ -802,7 +991,12 @@ async fn handle_shutdown( if enable_scanner { info!( target: "rustfs::main::handle_shutdown", - "Stopping background services (data scanner)..." + event = EVENT_BACKGROUND_SERVICE_SHUTDOWN, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + service = "data_scanner", + state = "stopping", + "Background service shutdown started" ); shutdown_background_services(); } @@ -810,7 +1004,12 @@ async fn handle_shutdown( if enable_heal || enable_scanner { info!( target: "rustfs::main::handle_shutdown", - "Stopping AHM services..." + event = EVENT_BACKGROUND_SERVICE_SHUTDOWN, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + service = "ahm", + state = "stopping", + "Background service shutdown started" ); shutdown_ahm_services(); } @@ -818,7 +1017,13 @@ async fn handle_shutdown( if !enable_scanner && !enable_heal { info!( target: "rustfs::main::handle_shutdown", - "Background services were disabled, skipping AHM shutdown" + event = EVENT_BACKGROUND_SERVICE_SHUTDOWN, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + service = "ahm", + state = "skipped", + reason = "disabled", + "Background service shutdown skipped" ); } @@ -827,7 +1032,12 @@ async fn handle_shutdown( if let Some(ftp_shutdown_tx) = ftp_shutdown_tx { info!( target: "rustfs::main::handle_shutdown", - "Shutting down FTP server..." + event = EVENT_PROTOCOL_SYSTEM_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + protocol = "ftp", + state = "stopping", + "Protocol system state changed" ); protocol_shutdowns.push(ftp_shutdown_tx.shutdown()); } @@ -835,7 +1045,12 @@ async fn handle_shutdown( if let Some(ftps_shutdown_tx) = ftps_shutdown_tx { info!( target: "rustfs::main::handle_shutdown", - "Shutting down FTPS server..." + event = EVENT_PROTOCOL_SYSTEM_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + protocol = "ftps", + state = "stopping", + "Protocol system state changed" ); protocol_shutdowns.push(ftps_shutdown_tx.shutdown()); } @@ -844,7 +1059,12 @@ async fn handle_shutdown( if let Some(webdav_shutdown_tx) = webdav_shutdown_tx { info!( target: "rustfs::main::handle_shutdown", - "Shutting down WebDAV server..." + event = EVENT_PROTOCOL_SYSTEM_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + protocol = "webdav", + state = "stopping", + "Protocol system state changed" ); protocol_shutdowns.push(webdav_shutdown_tx.shutdown()); } @@ -853,7 +1073,12 @@ async fn handle_shutdown( if let Some(sftp_shutdown_tx) = sftp_shutdown_tx { info!( target: "rustfs::main::handle_shutdown", - "Shutting down SFTP server..." + event = EVENT_PROTOCOL_SYSTEM_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + protocol = "sftp", + state = "stopping", + "Protocol system state changed" ); protocol_shutdowns.push(sftp_shutdown_tx.shutdown()); } @@ -861,30 +1086,61 @@ async fn handle_shutdown( // Stop the notification system info!( target: "rustfs::main::handle_shutdown", - "Shutting down event notifier system..." + event = EVENT_EVENT_NOTIFIER_SHUTDOWN, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + state = "stopping", + "Event notifier shutdown started" ); shutdown_event_notifier().await; // Stop the audit system info!( target: "rustfs::main::handle_shutdown", - "Stopping audit system..." + event = EVENT_AUDIT_SYSTEM_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + state = "stopping", + "Audit system state changed" ); match stop_audit_system().await { - Ok(_) => info!("Audit system stopped successfully."), - Err(e) => error!("Failed to stop audit system: {}", e), + Ok(_) => info!( + target: "rustfs::main::handle_shutdown", + event = EVENT_AUDIT_SYSTEM_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + state = "stopped", + "Audit system state changed" + ), + Err(e) => error!( + target: "rustfs::main::handle_shutdown", + event = EVENT_AUDIT_SYSTEM_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + state = "stop_failed", + error = %e, + "Audit system state changed" + ), } // Stop profiling tasks info!( target: "rustfs::main::handle_shutdown", - "Stopping profiling tasks..." + event = EVENT_PROFILING_SHUTDOWN, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + state = "stopping", + "Profiling shutdown started" ); rustfs::profiling::shutdown_profiling(); info!( target: "rustfs::main::handle_shutdown", - "Server is stopping..." + event = EVENT_SERVER_SHUTDOWN_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + state = "stopping", + "Server shutdown state changed" ); if let Some(s3_shutdown_handle) = s3_shutdown_handle { s3_shutdown_handle.shutdown().await; @@ -896,7 +1152,14 @@ async fn handle_shutdown( // the last updated status is stopped state_manager.update(ServiceState::Stopped); - info!(target: "rustfs::main::handle_shutdown", "Server stopped successfully."); + info!( + target: "rustfs::main::handle_shutdown", + event = EVENT_SERVER_SHUTDOWN_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + state = "stopped", + "Server shutdown state changed" + ); } #[cfg(test)] diff --git a/scripts/check_logging_guardrails.sh b/scripts/check_logging_guardrails.sh index 96795c965..2752120ca 100755 --- a/scripts/check_logging_guardrails.sh +++ b/scripts/check_logging_guardrails.sh @@ -33,6 +33,30 @@ forbidden_patterns=( 'info!("Sending event to targets: {:?}"' 'info!("Event processing initiated for {} targets for bucket: {}"' 'warn!("{}", notify_configuration_hint())' + 'info!("Available ARNs: {:?}"' + 'info!("Loaded notification config for bucket: {}"' + 'info!("Updated notification rules for bucket: {}"' + 'info!("Removed all notification rules for bucket: {}"' + 'info!("Audit configuration reloaded"' + 'info!("Audit system started"' + 'info!("Audit metrics reset"' + 'error!("Failed to set global observability guard: {}"' + 'error!("Failed to initialize TLS from {}: {}"' + 'error!("Server encountered an error and is shutting down: {}"' + 'error!("Failed to initialize Keystone authentication: {}"' + 'error!("new_global_notification_sys failed {:?}"' + 'info!("FTP system initialized successfully"' + 'info!("FTP system disabled"' + 'error!("Failed to initialize FTP system: {}"' + 'info!("FTPS system initialized successfully"' + 'info!("FTPS system disabled"' + 'error!("Failed to initialize FTPS system: {}"' + 'info!("WebDAV system initialized successfully"' + 'info!("WebDAV system disabled"' + 'error!("Failed to initialize WebDAV system: {}"' + 'info!("SFTP system initialized successfully"' + 'info!("SFTP system disabled"' + 'error!("Failed to initialize SFTP system: {}"' 'warn!("prewarm_local_disk_id_map: failed to load disk id for {}: {}"' 'info!("retrying get formats after {:?}"' )