diff --git a/.config/make/lint-fmt.mak b/.config/make/lint-fmt.mak index 9e857783a..5e96dc10a 100644 --- a/.config/make/lint-fmt.mak +++ b/.config/make/lint-fmt.mak @@ -26,6 +26,11 @@ architecture-migration-check: ## Check architecture migration guardrails @echo "🏗️ Checking architecture migration guardrails..." ./scripts/check_architecture_migration_rules.sh +.PHONY: logging-guardrails-check +logging-guardrails-check: ## Check logging guardrails for redaction and noise regressions + @echo "🪵 Checking logging guardrails..." + ./scripts/check_logging_guardrails.sh + .PHONY: compilation-check compilation-check: core-deps ## Run compilation check @echo "🔨 Running compilation check..." diff --git a/.config/make/pre-commit.mak b/.config/make/pre-commit.mak index 38a0012aa..1857f2030 100644 --- a/.config/make/pre-commit.mak +++ b/.config/make/pre-commit.mak @@ -7,5 +7,5 @@ setup-hooks: ## Set up git hooks @echo "✅ Git hooks setup complete!" .PHONY: pre-commit -pre-commit: fmt unsafe-code-check architecture-migration-check clippy-check compilation-check test ## Run pre-commit checks +pre-commit: fmt unsafe-code-check architecture-migration-check logging-guardrails-check clippy-check compilation-check test ## Run pre-commit checks @echo "✅ All pre-commit checks passed!" diff --git a/crates/audit/src/global.rs b/crates/audit/src/global.rs index 56fdaa25c..eedd8805a 100644 --- a/crates/audit/src/global.rs +++ b/crates/audit/src/global.rs @@ -15,7 +15,13 @@ use crate::{AuditEntry, AuditResult, AuditSystem, system::AuditTargetMetricSnapshot}; use rustfs_config::server_config::Config; use std::sync::{Arc, OnceLock}; -use tracing::{debug, error, trace, warn}; +use tracing::{debug, error, trace}; + +const LOG_COMPONENT_AUDIT: &str = "audit"; +const LOG_SUBSYSTEM_GLOBAL: &str = "global"; +const EVENT_AUDIT_GLOBAL_SKIPPED: &str = "audit_global_skipped"; +const EVENT_AUDIT_ENTRY_DROPPED: &str = "audit_entry_dropped"; +const EVENT_AUDIT_DISPATCH_FAILED: &str = "audit_dispatch_failed"; /// Global audit system instance static AUDIT_SYSTEM: OnceLock> = OnceLock::new(); @@ -37,7 +43,13 @@ macro_rules! with_audit_system { if let Some(system) = audit_system() { (async move { $async_closure(system).await }).await } else { - warn!("Audit system not initialized, operation skipped."); + debug!( + event = EVENT_AUDIT_GLOBAL_SKIPPED, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_GLOBAL, + reason = "system_not_initialized", + "Skipped audit system operation" + ); Ok(()) } }; @@ -70,16 +82,23 @@ pub async fn dispatch_audit_log(entry: Arc) -> AuditResult<()> { if system.is_running().await { system.dispatch(entry).await } else { - // The system is initialized but not running (for example, it is suspended). Silently discard log entries based on original logic. - // For debugging purposes, it can be useful to add a trace log here. - trace!("Audit system is not running, dropping audit entry."); + trace!( + event = EVENT_AUDIT_ENTRY_DROPPED, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_GLOBAL, + reason = "system_not_running", + "Dropped audit entry" + ); Ok(()) } } else { - // The system is not initialized at all. This is a more important state. - // It might be better to return an error or log a warning. - debug!("Audit system not initialized, dropping audit entry."); - // If this should be a hard failure, you can return Err(AuditError::NotInitialized("...")) + debug!( + event = EVENT_AUDIT_ENTRY_DROPPED, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_GLOBAL, + reason = "system_not_initialized", + "Dropped audit entry" + ); Ok(()) } } @@ -114,7 +133,13 @@ impl AuditLogger { /// Log an audit entry pub async fn log(entry: AuditEntry) { if let Err(e) = dispatch_audit_log(Arc::new(entry)).await { - error!(error = %e, "Failed to dispatch audit log entry"); + error!( + event = EVENT_AUDIT_DISPATCH_FAILED, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_GLOBAL, + error = %e, + "Failed to dispatch audit entry" + ); } } diff --git a/crates/audit/src/pipeline.rs b/crates/audit/src/pipeline.rs index 0457cfeb2..6b7f195b0 100644 --- a/crates/audit/src/pipeline.rs +++ b/crates/audit/src/pipeline.rs @@ -20,7 +20,20 @@ use rustfs_targets::{ use std::sync::Arc; use std::time::Duration; use tokio::sync::{Mutex, RwLock}; -use tracing::{error, info, warn}; +use tracing::{debug, error, info, warn}; + +const LOG_COMPONENT_AUDIT: &str = "audit"; +const LOG_SUBSYSTEM_PIPELINE: &str = "pipeline"; +const EVENT_AUDIT_DISPATCH_SKIPPED: &str = "audit_dispatch_skipped"; +const EVENT_AUDIT_DISPATCH_FAILED: &str = "audit_dispatch_failed"; +const EVENT_AUDIT_BATCH_DISPATCH_SKIPPED: &str = "audit_batch_dispatch_skipped"; +const EVENT_AUDIT_BATCH_DISPATCH_FAILED: &str = "audit_batch_dispatch_failed"; +const EVENT_AUDIT_BATCH_DISPATCH_COMPLETED: &str = "audit_batch_dispatch_completed"; +const EVENT_AUDIT_TARGET_STATE_CHANGED: &str = "audit_target_state_changed"; +const EVENT_AUDIT_REPLAY_DELIVERED: &str = "audit_replay_delivered"; +const EVENT_AUDIT_REPLAY_RETRY_SCHEDULED: &str = "audit_replay_retry_scheduled"; +const EVENT_AUDIT_REPLAY_DROPPED: &str = "audit_replay_dropped"; +const EVENT_AUDIT_REPLAY_STREAM_STATUS: &str = "audit_replay_stream_status"; #[derive(Clone)] pub struct AuditPipeline { @@ -40,7 +53,13 @@ impl AuditPipeline { let targets = registry.list_target_values(); if targets.is_empty() { - warn!("No audit targets configured for dispatch"); + debug!( + event = EVENT_AUDIT_DISPATCH_SKIPPED, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_PIPELINE, + reason = "no_targets_configured", + "Skipped audit dispatch" + ); return Ok(()); } @@ -77,7 +96,14 @@ impl AuditPipeline { observability::record_target_success(); } Err(e) => { - error!(target_id = %target_key, error = %e, "Failed to dispatch audit log to target"); + error!( + event = EVENT_AUDIT_DISPATCH_FAILED, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_PIPELINE, + target_id = %target_key, + error = %e, + "Failed to dispatch audit event" + ); errors.push(e); observability::record_target_failure(); } @@ -91,9 +117,13 @@ impl AuditPipeline { } else { observability::record_audit_failure(dispatch_time); warn!( + event = EVENT_AUDIT_DISPATCH_FAILED, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_PIPELINE, error_count = errors.len(), success_count = success_count, - "Some audit targets failed to receive log entry" + duration_ms = dispatch_time.as_millis() as u64, + "Some audit targets failed to receive audit event" ); } @@ -108,7 +138,14 @@ impl AuditPipeline { let targets = registry.list_target_values(); if targets.is_empty() { - warn!("No audit targets configured for batch dispatch"); + debug!( + event = EVENT_AUDIT_BATCH_DISPATCH_SKIPPED, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_PIPELINE, + entry_count = entries.len(), + reason = "no_targets_configured", + "Skipped audit batch dispatch" + ); return Ok(()); } @@ -142,21 +179,31 @@ impl AuditPipeline { let results = futures::future::join_all(tasks).await; let mut total_success = 0; let mut total_errors = 0; - for (_target_id, success_count, errors) in results { + for (target_id, success_count, errors) in results { total_success += success_count; total_errors += errors.len(); for e in errors { - error!("Batch dispatch error: {:?}", e); + error!( + event = EVENT_AUDIT_BATCH_DISPATCH_FAILED, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_PIPELINE, + target_id = %target_id, + error = ?e, + "Audit batch dispatch failed" + ); } } let dispatch_time = start_time.elapsed(); - info!( - "Batch dispatched {} entries, success: {}, errors: {}, time: {:?}", - entries.len(), - total_success, - total_errors, - dispatch_time + debug!( + event = EVENT_AUDIT_BATCH_DISPATCH_COMPLETED, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_PIPELINE, + entry_count = entries.len(), + success_count = total_success, + error_count = total_errors, + duration_ms = dispatch_time.as_millis() as u64, + "Completed audit batch dispatch" ); Ok(()) @@ -213,7 +260,14 @@ impl AuditRuntimeView { pub async fn enable_target(&self, target_id: &str) -> AuditResult<()> { let registry = self.registry.lock().await; if registry.get_target(target_id).is_some() { - info!(target_id = %target_id, "Target enabled"); + info!( + event = EVENT_AUDIT_TARGET_STATE_CHANGED, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_PIPELINE, + target_id = %target_id, + state = "enabled", + "Changed audit target state" + ); Ok(()) } else { Err(crate::AuditError::Configuration(format!("Target not found: {target_id}"), None)) @@ -223,7 +277,14 @@ impl AuditRuntimeView { pub async fn disable_target(&self, target_id: &str) -> AuditResult<()> { let registry = self.registry.lock().await; if registry.get_target(target_id).is_some() { - info!(target_id = %target_id, "Target disabled"); + info!( + event = EVENT_AUDIT_TARGET_STATE_CHANGED, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_PIPELINE, + target_id = %target_id, + state = "disabled", + "Changed audit target state" + ); Ok(()) } else { Err(crate::AuditError::Configuration(format!("Target not found: {target_id}"), None)) @@ -233,7 +294,14 @@ impl AuditRuntimeView { pub async fn remove_target(&self, target_id: &str) -> AuditResult<()> { let mut registry = self.registry.lock().await; if registry.remove_target(target_id).await.is_some() { - info!(target_id = %target_id, "Target removed"); + info!( + event = EVENT_AUDIT_TARGET_STATE_CHANGED, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_PIPELINE, + target_id = %target_id, + state = "removed", + "Changed audit target state" + ); Ok(()) } else { Err(crate::AuditError::Configuration(format!("Target not found: {target_id}"), None)) @@ -249,7 +317,14 @@ impl AuditRuntimeView { let mut registry = self.registry.lock().await; let _ = registry.remove_target(&target_id).await; registry.add_shared_target(target_id.clone(), shared_target); - info!(target_id = %target_id, "Target upserted"); + info!( + event = EVENT_AUDIT_TARGET_STATE_CHANGED, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_PIPELINE, + target_id = %target_id, + state = "upserted", + "Changed audit target state" + ); Ok(()) } } @@ -268,43 +343,111 @@ impl AuditRuntimeFacade { Box::pin(async move { match event { ReplayEvent::Delivered { key, target } => { - info!("Successfully sent audit entry, target: {}, key: {}", target.id(), key.to_string()); + debug!( + event = EVENT_AUDIT_REPLAY_DELIVERED, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_PIPELINE, + target_id = %target.id(), + replay_key = %key, + "Delivered queued audit event" + ); observability::record_target_success(); } ReplayEvent::RetryableError { error, target, .. } => match error { rustfs_targets::TargetError::NotConnected => { - warn!("Target {} not connected, retrying...", target.id()); + debug!( + event = EVENT_AUDIT_REPLAY_RETRY_SCHEDULED, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_PIPELINE, + target_id = %target.id(), + reason = "not_connected", + "Retrying queued audit event delivery" + ); } rustfs_targets::TargetError::Timeout(_) => { - warn!("Timeout sending to target {}, retrying...", target.id()); + debug!( + event = EVENT_AUDIT_REPLAY_RETRY_SCHEDULED, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_PIPELINE, + target_id = %target.id(), + reason = "timeout", + "Retrying queued audit event delivery" + ); } _ => {} }, ReplayEvent::Dropped { reason, target, .. } => { - warn!("Dropped queued payload for target {}: {}", target.id(), reason); + warn!( + event = EVENT_AUDIT_REPLAY_DROPPED, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_PIPELINE, + target_id = %target.id(), + reason = %reason, + "Dropped queued audit payload" + ); observability::record_target_failure(); } ReplayEvent::PermanentFailure { error, target, .. } => { - error!("Permanent error for target {}: {}", target.id(), error); + error!( + event = EVENT_AUDIT_REPLAY_DROPPED, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_PIPELINE, + target_id = %target.id(), + error = %error, + reason = "permanent_failure", + "Queued audit payload failed permanently" + ); target.record_final_failure(); observability::record_target_failure(); } ReplayEvent::RetryExhausted { key, target } => { - warn!("Max retries exceeded for key {}, target: {}, skipping", key.to_string(), target.id()); + warn!( + event = EVENT_AUDIT_REPLAY_DROPPED, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_PIPELINE, + target_id = %target.id(), + replay_key = %key, + reason = "retry_exhausted", + "Dropped queued audit payload after retry exhaustion" + ); target.record_final_failure(); observability::record_target_failure(); } ReplayEvent::UnreadableEntry { key, error, target } => { - warn!("Skipping unreadable audit store entry {} for target {}: {}", key, target.id(), error); + warn!( + event = EVENT_AUDIT_REPLAY_DROPPED, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_PIPELINE, + target_id = %target.id(), + replay_key = %key, + error = %error, + reason = "unreadable_entry", + "Skipped unreadable audit store entry" + ); } } }) }), Arc::new(|target_id, has_replay| { if has_replay { - info!(target_id = %target_id, "Audit stream processing started"); + info!( + event = EVENT_AUDIT_REPLAY_STREAM_STATUS, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_PIPELINE, + target_id = %target_id, + replay_enabled = true, + "Audit replay stream started" + ); } else { - info!(target_id = %target_id, "No store configured, skip audit stream processing"); + debug!( + event = EVENT_AUDIT_REPLAY_STREAM_STATUS, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_PIPELINE, + target_id = %target_id, + replay_enabled = false, + reason = "no_store_configured", + "Audit replay stream skipped" + ); } }), None, diff --git a/crates/audit/src/system.rs b/crates/audit/src/system.rs index 6c14c9c5b..d6bea4d41 100644 --- a/crates/audit/src/system.rs +++ b/crates/audit/src/system.rs @@ -20,7 +20,12 @@ use rustfs_config::server_config::Config; use rustfs_targets::{ReplayWorkerManager, Target}; use std::sync::Arc; use tokio::sync::{Mutex, RwLock}; -use tracing::{error, info, warn}; +use tracing::{debug, error, info, warn}; + +const LOG_COMPONENT_AUDIT: &str = "audit"; +const LOG_SUBSYSTEM_SYSTEM: &str = "system"; +const EVENT_AUDIT_SYSTEM_STATE: &str = "audit_system_state"; +const EVENT_AUDIT_CONFIG_RELOADED: &str = "audit_config_reloaded"; #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct AuditTargetMetricSnapshot { @@ -104,12 +109,19 @@ impl AuditSystem { final_state: AuditSystemState, ) -> AuditResult<()> { if targets.is_empty() { - info!("No enabled audit targets found, keeping audit system stopped"); + debug_audit_state("stopped", Some("no_enabled_targets"), None, 0); self.clear_runtime_targets().await?; return Ok(()); } - info!(target_count = targets.len(), "Created audit targets successfully"); + info!( + event = EVENT_AUDIT_SYSTEM_STATE, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_SYSTEM, + state = "targets_created", + target_count = targets.len(), + "Created audit targets" + ); let activation = self.runtime_facade().activate_targets_with_replay(targets).await; self.runtime_facade().replace_targets(activation).await?; @@ -134,7 +146,7 @@ impl AuditSystem { return Err(AuditError::AlreadyInitialized); } AuditSystemState::Starting => { - warn!("Audit system is already starting"); + warn_audit_state("starting", Some("already_starting")); return Ok(()); } _ => {} @@ -142,7 +154,13 @@ impl AuditSystem { drop(state); - info!("Starting audit system"); + info!( + event = EVENT_AUDIT_SYSTEM_STATE, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_SYSTEM, + state = "starting", + "Starting audit system" + ); // Record system start observability::record_system_start(); @@ -161,7 +179,7 @@ impl AuditSystem { } self.commit_runtime_targets(targets, AuditSystemState::Running).await?; - info!("Audit system started successfully"); + info_audit_state("running", None, None); Ok(()) } Err(e) => { @@ -183,11 +201,11 @@ impl AuditSystem { match *state { AuditSystemState::Running => { *state = AuditSystemState::Paused; - info!("Audit system paused"); + info_audit_state("paused", None, None); Ok(()) } AuditSystemState::Paused => { - warn!("Audit system is already paused"); + warn_audit_state("paused", Some("already_paused")); Ok(()) } _ => Err(AuditError::Configuration("Cannot pause audit system in current state".to_string(), None)), @@ -204,11 +222,11 @@ impl AuditSystem { match *state { AuditSystemState::Paused => { *state = AuditSystemState::Running; - info!("Audit system resumed"); + info_audit_state("running", Some("resumed"), None); Ok(()) } AuditSystemState::Running => { - warn!("Audit system is already running"); + warn_audit_state("running", Some("already_running")); Ok(()) } _ => Err(AuditError::Configuration("Cannot resume audit system in current state".to_string(), None)), @@ -224,11 +242,11 @@ impl AuditSystem { match *state { AuditSystemState::Stopped => { - warn!("Audit system is already stopped"); + warn_audit_state("stopped", Some("already_stopped")); return Ok(()); } AuditSystemState::Stopping => { - warn!("Audit system is already stopping"); + warn_audit_state("stopping", Some("already_stopping")); return Ok(()); } _ => {} @@ -237,7 +255,13 @@ impl AuditSystem { *state = AuditSystemState::Stopping; drop(state); - info!("Stopping audit system"); + info!( + event = EVENT_AUDIT_SYSTEM_STATE, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_SYSTEM, + state = "stopping", + "Stopping audit system" + ); // Stop all stream tasks first if let Err(e) = self.clear_runtime_targets().await { @@ -248,7 +272,7 @@ impl AuditSystem { let mut config_guard = self.config.write().await; *config_guard = None; - info!("Audit system stopped"); + info_audit_state("stopped", None, None); Ok(()) } @@ -396,7 +420,13 @@ impl AuditSystem { /// # Returns /// * `AuditResult<()>` - Result indicating success or failure pub async fn reload_config(&self, new_config: Config) -> AuditResult<()> { - info!("Reloading audit system configuration"); + info!( + event = EVENT_AUDIT_CONFIG_RELOADED, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_SYSTEM, + state = "reloading", + "Reloading audit configuration" + ); observability::record_config_reload(); @@ -414,7 +444,13 @@ impl AuditSystem { match self.create_targets_from_config(&new_config).await { Ok(targets) => { self.commit_runtime_targets(targets, final_state).await?; - info!("Audit configuration reloaded successfully"); + info!( + event = EVENT_AUDIT_CONFIG_RELOADED, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_SYSTEM, + state = "reloaded", + "Reloaded audit configuration" + ); Ok(()) } Err(e) => { @@ -446,6 +482,42 @@ impl AuditSystem { } } +fn info_audit_state(state: &str, reason: Option<&str>, target_count: Option) { + info!( + event = EVENT_AUDIT_SYSTEM_STATE, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_SYSTEM, + state, + reason = reason.unwrap_or_default(), + target_count = target_count.unwrap_or_default(), + "Changed audit system state" + ); +} + +fn debug_audit_state(state: &str, reason: Option<&str>, error: Option<&str>, target_count: usize) { + debug!( + event = EVENT_AUDIT_SYSTEM_STATE, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_SYSTEM, + state, + reason = reason.unwrap_or_default(), + error = error.unwrap_or_default(), + target_count, + "Observed audit system state" + ); +} + +fn warn_audit_state(state: &str, reason: Option<&str>) { + warn!( + event = EVENT_AUDIT_SYSTEM_STATE, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_SYSTEM, + state, + reason = reason.unwrap_or_default(), + "Audit system state transition skipped" + ); +} + #[cfg(test)] mod tests { use super::{AuditSystem, AuditSystemState}; diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index f8f1becde..48ad2bd6c 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -81,6 +81,13 @@ use tracing::{debug, error, info, warn}; use uuid::Uuid; use xxhash_rust::xxh64; +const LOG_COMPONENT_ECSTORE: &str = "ecstore"; +const LOG_SUBSYSTEM_LIFECYCLE: &str = "lifecycle"; +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"; + pub type TimeFn = Arc Pin + Send>> + Send + Sync + 'static>; pub type TraceFn = Arc) -> Pin + Send>> + Send + Sync + 'static>; @@ -189,7 +196,15 @@ impl LifecycleSys { Ok((lc, _)) => Some(lc), Err(Error::ConfigNotFound) => None, Err(err) => { - warn!(bucket, error = ?err, "failed to load lifecycle config"); + debug!( + event = EVENT_LIFECYCLE_SCAN_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + bucket, + error = ?err, + reason = "lifecycle_config_unavailable", + "Skipped lifecycle config lookup" + ); None } } @@ -204,13 +219,16 @@ impl LifecycleSys { let name = name.clone(); let version_id = version_id.clone(); Box::pin(async move { - info!( + debug!( bucket = %bucket, object = %name, version_id = %version_id, action = %_action, - "ILM lifecycle trace: {} on {}/{} (version: {})", - _action, bucket, name, version_id + event = EVENT_LIFECYCLE_WORKER_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + state = "trace", + "Lifecycle trace event" ); }) }) @@ -482,7 +500,14 @@ impl ExpiryState { loop { select! { _ = cancel_token.cancelled() => { - info!("lifecycle expiry worker received shutdown signal, exiting"); + debug!( + event = EVENT_LIFECYCLE_WORKER_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + state = "stopped", + reason = "shutdown_signal", + "Lifecycle expiry worker stopped" + ); break; } v = rx.recv() => { @@ -513,12 +538,16 @@ impl ExpiryState { else if v.as_any().is::() { let v = v.as_any().downcast_ref::().expect("Jentry downcast failed"); if let Err(err) = delete_object_from_remote_tier(&v.obj_name, &v.version_id, &v.tier_name).await { - warn!( + debug!( + event = EVENT_LIFECYCLE_WORKER_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, object = %v.obj_name, version_id = %v.version_id, tier = %v.tier_name, error = ?err, - "failed to delete transitioned object from remote tier" + reason = "remote_tier_delete_failed", + "Lifecycle worker skipped remote tier delete" ); } } @@ -532,14 +561,18 @@ impl ExpiryState { ) .await { - warn!( + debug!( bucket = %oi.bucket, object = %oi.name, remote_object = %oi.transitioned_object.name, remote_version_id = %oi.transitioned_object.version_id, tier = %oi.transitioned_object.tier, error = ?err, - "failed to sweep transitioned free version from remote tier" + event = EVENT_LIFECYCLE_WORKER_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + reason = "remote_tier_delete_failed", + "Lifecycle worker skipped remote tier delete" ); continue; } @@ -562,14 +595,18 @@ impl ExpiryState { } Err(err) if is_err_version_not_found(&err) || is_err_object_not_found(&err) => continue, Err(err) => { - warn!( + debug!( + event = EVENT_LIFECYCLE_WORKER_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, bucket = %oi.bucket, object = %oi.name, remote_object = %oi.transitioned_object.name, remote_version_id = %oi.transitioned_object.version_id, tier = %oi.transitioned_object.tier, error = ?err, - "failed to delete transitioned free version after remote tier sweep" + reason = "local_free_version_delete_failed", + "Lifecycle worker failed local free-version cleanup" ); break; } @@ -577,19 +614,29 @@ impl ExpiryState { } if !deleted_locally { - warn!( + debug!( + event = EVENT_LIFECYCLE_WORKER_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, bucket = %oi.bucket, object = %oi.name, remote_object = %oi.transitioned_object.name, remote_version_id = %oi.transitioned_object.version_id, tier = %oi.transitioned_object.tier, - "transitioned free version was not found during local cleanup" + reason = "local_free_version_missing", + "Lifecycle worker could not find transitioned free version locally" ); } } else { //info!("Invalid work type - {:?}", v); - warn!("lifecycle worker received unsupported operation type"); + debug!( + event = EVENT_LIFECYCLE_WORKER_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + state = "unsupported_task", + "Lifecycle worker received unsupported operation type" + ); } } } @@ -689,14 +736,37 @@ impl TransitionState { scheduled.lock().unwrap().remove(&bucket); Self::add_counter(&state.compensation_running_tasks, -1); state.record_scanner_transition_state(); - warn!(bucket = %bucket, "transition compensation skipped because object layer is unavailable"); + debug!( + event = EVENT_LIFECYCLE_TRANSITION_COMPENSATION, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + bucket = %bucket, + state = "skipped", + reason = "object_layer_unavailable", + "Skipped transition compensation" + ); return; }; if let Err(err) = enqueue_transition_for_existing_objects(api, &bucket).await { - warn!(bucket = %bucket, error = ?err, "transition compensation backfill failed"); + warn!( + event = EVENT_LIFECYCLE_TRANSITION_COMPENSATION, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + bucket = %bucket, + state = "failed", + error = ?err, + "Transition compensation backfill failed" + ); } else { - info!(bucket = %bucket, "transition compensation backfill completed"); + info!( + event = EVENT_LIFECYCLE_TRANSITION_COMPENSATION, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + bucket = %bucket, + state = "completed", + "Completed transition compensation backfill" + ); } scheduled.lock().unwrap().remove(&bucket); @@ -744,43 +814,59 @@ impl TransitionState { match failure { ImmediateEnqueueFailure::ForcedTimeout => { Self::inc_counter(&self.queue_send_timeout_tasks); - warn!( + debug!( bucket = %oi.bucket, object = %oi.name, source = ?src, compensation_scheduled = scheduled, + event = EVENT_LIFECYCLE_TRANSITION_COMPENSATION, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + state = "queue_timeout_forced", "transition enqueue forced into timeout path for test fault injection" ); } ImmediateEnqueueFailure::QueueClosed { timeout_ms } => match timeout_ms { Some(timeout_ms) => { - warn!( + debug!( bucket = %oi.bucket, object = %oi.name, source = ?src, timeout_ms, compensation_scheduled = scheduled, + event = EVENT_LIFECYCLE_TRANSITION_COMPENSATION, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + state = "queue_closed", "transition enqueue failed because the queue is closed" ); } None => { - warn!( + debug!( bucket = %oi.bucket, object = %oi.name, source = ?src, compensation_scheduled = scheduled, + event = EVENT_LIFECYCLE_TRANSITION_COMPENSATION, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + state = "queue_closed", "transition enqueue failed because the queue is closed" ); } }, ImmediateEnqueueFailure::QueueSendTimedOut { timeout_ms } => { Self::inc_counter(&self.queue_send_timeout_tasks); - warn!( + debug!( bucket = %oi.bucket, object = %oi.name, source = ?src, timeout_ms, compensation_scheduled = scheduled, + event = EVENT_LIFECYCLE_TRANSITION_COMPENSATION, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + state = "queue_send_timed_out", "transition enqueue timed out under backpressure" ); } @@ -850,10 +936,14 @@ impl TransitionState { ); } Err(async_channel::TrySendError::Closed(_)) => { - warn!( + debug!( bucket = %oi.bucket, object = %oi.name, source = ?src, + event = EVENT_LIFECYCLE_TRANSITION_COMPENSATION, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + state = "queue_closed", "transition enqueue failed because the queue is closed" ); } @@ -1229,7 +1319,14 @@ async fn cleanup_empty_multipart_sha_dirs_on_local_disks(set: &Arc) { Ok(entries) => entries, Err(err) => { if err != DiskError::FileNotFound && err != DiskError::VolumeNotFound { - warn!(error = ?err, "failed to list multipart root during empty sha cleanup"); + debug!( + event = EVENT_LIFECYCLE_STALE_MULTIPART_CLEANUP, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + error = ?err, + reason = "multipart_root_list_failed", + "Skipped empty multipart sha cleanup" + ); } continue; } @@ -1244,7 +1341,15 @@ async fn cleanup_empty_multipart_sha_dirs_on_local_disks(set: &Arc) { Ok(entries) => entries, Err(err) => { if err != DiskError::FileNotFound && err != DiskError::VolumeNotFound { - warn!(sha_dir = %sha_dir, error = ?err, "failed to list multipart sha dir during empty sha cleanup"); + debug!( + event = EVENT_LIFECYCLE_STALE_MULTIPART_CLEANUP, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + sha_dir = %sha_dir, + error = ?err, + reason = "multipart_sha_dir_list_failed", + "Skipped empty multipart sha cleanup" + ); } continue; } @@ -1260,7 +1365,15 @@ async fn cleanup_empty_multipart_sha_dirs_on_local_disks(set: &Arc) { && err != DiskError::FileNotFound && err != DiskError::VolumeNotFound { - warn!(sha_dir = %sha_dir, error = ?err, "failed to remove empty multipart sha dir"); + debug!( + event = EVENT_LIFECYCLE_STALE_MULTIPART_CLEANUP, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + sha_dir = %sha_dir, + error = ?err, + reason = "multipart_sha_dir_remove_failed", + "Failed to remove empty multipart sha dir" + ); } } } @@ -1282,7 +1395,14 @@ async fn cleanup_stale_multipart_uploads_in_set(set: &Arc, now: Offset Ok(entries) => entries, Err(err) => { if err != DiskError::FileNotFound && err != DiskError::VolumeNotFound { - warn!(error = ?err, "failed to list multipart root during stale cleanup"); + debug!( + event = EVENT_LIFECYCLE_STALE_MULTIPART_CLEANUP, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + error = ?err, + reason = "multipart_root_list_failed", + "Skipped stale multipart cleanup" + ); } continue; } @@ -1297,7 +1417,15 @@ async fn cleanup_stale_multipart_uploads_in_set(set: &Arc, now: Offset Ok(entries) => entries, Err(err) => { if err != DiskError::FileNotFound && err != DiskError::VolumeNotFound { - warn!(sha_dir = %sha_dir, error = ?err, "failed to list multipart sha dir during stale cleanup"); + debug!( + event = EVENT_LIFECYCLE_STALE_MULTIPART_CLEANUP, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + sha_dir = %sha_dir, + error = ?err, + reason = "multipart_sha_dir_list_failed", + "Skipped stale multipart cleanup" + ); } continue; } @@ -1317,7 +1445,15 @@ async fn cleanup_stale_multipart_uploads_in_set(set: &Arc, now: Offset Ok(candidate) => candidate, Err(err) => { if err != DiskError::FileNotFound { - warn!(path = %candidate_path, error = ?err, "failed to read multipart metadata during stale cleanup"); + debug!( + event = EVENT_LIFECYCLE_STALE_MULTIPART_CLEANUP, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + path = %candidate_path, + error = ?err, + reason = "multipart_metadata_read_failed", + "Multipart metadata unavailable during stale cleanup" + ); } let initiated = initiated_from_upload_dir(&upload_dir, None); StaleMultipartUploadCandidate { @@ -1351,18 +1487,39 @@ async fn cleanup_stale_multipart_uploads_in_set(set: &Arc, now: Offset deleted += 1; let upload_id = encode_stale_upload_id(&upload_dir); if let Some(metadata) = candidate.metadata.as_ref() { - info!( + debug!( bucket = metadata.get(RUSTFS_MULTIPART_BUCKET_KEY).cloned().unwrap_or_default(), object = metadata.get(RUSTFS_MULTIPART_OBJECT_KEY).cloned().unwrap_or_default(), upload_id = %upload_id, due = ?due, - "removed stale multipart upload" + event = EVENT_LIFECYCLE_STALE_MULTIPART_CLEANUP, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + state = "removed", + "Removed stale multipart upload" ); } else { - info!(path = %candidate.path, upload_id = %upload_id, due = ?due, "removed stale multipart upload"); + debug!( + path = %candidate.path, + upload_id = %upload_id, + due = ?due, + event = EVENT_LIFECYCLE_STALE_MULTIPART_CLEANUP, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + state = "removed", + "Removed stale multipart upload" + ); } } - Err(err) => warn!(path = %candidate.path, error = ?err, "failed to remove stale multipart upload"), + Err(err) => debug!( + event = EVENT_LIFECYCLE_STALE_MULTIPART_CLEANUP, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + path = %candidate.path, + error = ?err, + reason = "multipart_remove_failed", + "Failed to remove stale multipart upload" + ), } } @@ -1402,7 +1559,13 @@ pub fn init_background_stale_multipart_upload_cleanup(api: Arc) { let deleted = cleanup_stale_multipart_uploads_once_at(api, OffsetDateTime::now_utc(), default_expiry).await; if deleted > 0 { - info!(deleted, "completed stale multipart cleanup pass"); + debug!( + event = EVENT_LIFECYCLE_STALE_MULTIPART_CLEANUP, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + deleted, + "Completed stale multipart cleanup pass" + ); } } }); @@ -2061,9 +2224,14 @@ pub async fn eval_action_from_lifecycle( oi: &ObjectInfo, ) -> lifecycle::Event { let event = lc.eval(&oi.to_lifecycle_opts()).await; - //if serverDebugLog { - info!("lifecycle: Secondary scan: {}", event.action); - //} + debug!( + event = EVENT_LIFECYCLE_SCAN_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + action = ?event.action, + state = "evaluated", + "Evaluated lifecycle action during secondary scan" + ); let lock_enabled = if let Some(lr) = lr { lr.mode.is_some() } else { false }; @@ -2079,15 +2247,25 @@ pub async fn eval_action_from_lifecycle( if lock_enabled && check_object_lock_for_deletion(&oi.bucket, oi, false).await.is_some() { //if serverDebugLog { if oi.version_id.is_some() { - info!( - "lifecycle: {} v({}) is locked, not deleting", - oi.name, - oi.version_id.map(|v| v.to_string()).unwrap_or_default() + debug!( + event = EVENT_LIFECYCLE_SCAN_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + object = %oi.name, + version_id = %oi.version_id.map(|v| v.to_string()).unwrap_or_default(), + reason = "object_locked", + "Skipped lifecycle delete because object version is locked" ); } else { - info!("lifecycle: {} is locked, not deleting", oi.name); + debug!( + event = EVENT_LIFECYCLE_SCAN_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + object = %oi.name, + reason = "object_locked", + "Skipped lifecycle delete because object is locked" + ); } - //} return lifecycle::Event::default(); } if let Some(rcfg) = rcfg diff --git a/crates/ecstore/src/bucket/metadata.rs b/crates/ecstore/src/bucket/metadata.rs index 6d44af0b8..eb236628f 100644 --- a/crates/ecstore/src/bucket/metadata.rs +++ b/crates/ecstore/src/bucket/metadata.rs @@ -799,59 +799,139 @@ impl BucketMetadata { fn parse_all_configs(&mut self) -> Result<()> { if let Err(e) = self.parse_policy_config() { - tracing::warn!(bucket = %self.name, config = "policy", error = %e, "parse_all_configs: failed to parse"); + tracing::warn!( + event = "bucket_metadata_parse_failed", + component = "ecstore", + subsystem = "bucket_metadata", + bucket = %self.name, + config = "policy", + error = %e, + "Failed to parse bucket metadata config" + ); } if !self.notification_config_xml.is_empty() && let Err(e) = deserialize::(&self.notification_config_xml) .map(|c| self.notification_config = Some(c)) { - tracing::warn!(bucket = %self.name, config = "notification", error = %e, "parse_all_configs: failed to parse"); + tracing::warn!( + event = "bucket_metadata_parse_failed", + component = "ecstore", + subsystem = "bucket_metadata", + bucket = %self.name, + config = "notification", + error = %e, + "Failed to parse bucket metadata config" + ); } if !self.lifecycle_config_xml.is_empty() && let Err(e) = deserialize::(&self.lifecycle_config_xml).map(|c| self.lifecycle_config = Some(c)) { - tracing::warn!(bucket = %self.name, config = "lifecycle", error = %e, "parse_all_configs: failed to parse"); + tracing::warn!( + event = "bucket_metadata_parse_failed", + component = "ecstore", + subsystem = "bucket_metadata", + bucket = %self.name, + config = "lifecycle", + error = %e, + "Failed to parse bucket metadata config" + ); } if !self.object_lock_config_xml.is_empty() && let Err(e) = deserialize::(&self.object_lock_config_xml).map(|c| self.object_lock_config = Some(c)) { - tracing::warn!(bucket = %self.name, config = "object_lock", error = %e, "parse_all_configs: failed to parse"); + tracing::warn!( + event = "bucket_metadata_parse_failed", + component = "ecstore", + subsystem = "bucket_metadata", + bucket = %self.name, + config = "object_lock", + error = %e, + "Failed to parse bucket metadata config" + ); } if !self.versioning_config_xml.is_empty() && let Err(e) = deserialize::(&self.versioning_config_xml).map(|c| self.versioning_config = Some(c)) { - tracing::warn!(bucket = %self.name, config = "versioning", error = %e, "parse_all_configs: failed to parse"); + tracing::warn!( + event = "bucket_metadata_parse_failed", + component = "ecstore", + subsystem = "bucket_metadata", + bucket = %self.name, + config = "versioning", + error = %e, + "Failed to parse bucket metadata config" + ); } if !self.encryption_config_xml.is_empty() && let Err(e) = deserialize::(&self.encryption_config_xml).map(|c| self.sse_config = Some(c)) { - tracing::warn!(bucket = %self.name, config = "encryption", error = %e, "parse_all_configs: failed to parse"); + tracing::warn!( + event = "bucket_metadata_parse_failed", + component = "ecstore", + subsystem = "bucket_metadata", + bucket = %self.name, + config = "encryption", + error = %e, + "Failed to parse bucket metadata config" + ); } if !self.tagging_config_xml.is_empty() && let Err(e) = deserialize::(&self.tagging_config_xml).map(|c| self.tagging_config = Some(c)) { - tracing::warn!(bucket = %self.name, config = "tagging", error = %e, "parse_all_configs: failed to parse"); + tracing::warn!( + event = "bucket_metadata_parse_failed", + component = "ecstore", + subsystem = "bucket_metadata", + bucket = %self.name, + config = "tagging", + error = %e, + "Failed to parse bucket metadata config" + ); } if !self.quota_config_json.is_empty() && let Err(e) = serde_json::from_slice(&self.quota_config_json).map(|c| self.quota_config = Some(c)) { - tracing::warn!(bucket = %self.name, config = "quota", error = %e, "parse_all_configs: failed to parse"); + tracing::warn!( + event = "bucket_metadata_parse_failed", + component = "ecstore", + subsystem = "bucket_metadata", + bucket = %self.name, + config = "quota", + error = %e, + "Failed to parse bucket metadata config" + ); } if !self.replication_config_xml.is_empty() && let Err(e) = deserialize::(&self.replication_config_xml).map(|c| self.replication_config = Some(c)) { - tracing::warn!(bucket = %self.name, config = "replication", error = %e, "parse_all_configs: failed to parse"); + tracing::warn!( + event = "bucket_metadata_parse_failed", + component = "ecstore", + subsystem = "bucket_metadata", + bucket = %self.name, + config = "replication", + error = %e, + "Failed to parse bucket metadata config" + ); } if !self.bucket_targets_config_json.is_empty() { if let Err(e) = serde_json::from_slice::(&self.bucket_targets_config_json) .map(|t| self.bucket_target_config = Some(t)) { - tracing::warn!(bucket = %self.name, config = "bucket_targets", error = %e, "parse_all_configs: failed to parse"); + tracing::warn!( + event = "bucket_metadata_parse_failed", + component = "ecstore", + subsystem = "bucket_metadata", + bucket = %self.name, + config = "bucket_targets", + error = %e, + "Failed to parse bucket metadata config" + ); self.bucket_target_config = Some(BucketTargets::default()); } } else { @@ -860,45 +940,96 @@ impl BucketMetadata { if !self.cors_config_xml.is_empty() && let Err(e) = deserialize::(&self.cors_config_xml).map(|c| self.cors_config = Some(c)) { - tracing::warn!(bucket = %self.name, config = "cors", error = %e, "parse_all_configs: failed to parse"); + tracing::warn!( + event = "bucket_metadata_parse_failed", + component = "ecstore", + subsystem = "bucket_metadata", + bucket = %self.name, + config = "cors", + error = %e, + "Failed to parse bucket metadata config" + ); } if !self.logging_config_xml.is_empty() && let Err(e) = deserialize::(&self.logging_config_xml).map(|c| self.logging_config = Some(c)) { - tracing::warn!(bucket = %self.name, config = "logging", error = %e, "parse_all_configs: failed to parse"); + tracing::warn!( + event = "bucket_metadata_parse_failed", + component = "ecstore", + subsystem = "bucket_metadata", + bucket = %self.name, + config = "logging", + error = %e, + "Failed to parse bucket metadata config" + ); } if !self.website_config_xml.is_empty() && let Err(e) = deserialize::(&self.website_config_xml).map(|c| self.website_config = Some(c)) { - tracing::warn!(bucket = %self.name, config = "website", error = %e, "parse_all_configs: failed to parse"); + tracing::warn!( + event = "bucket_metadata_parse_failed", + component = "ecstore", + subsystem = "bucket_metadata", + bucket = %self.name, + config = "website", + error = %e, + "Failed to parse bucket metadata config" + ); } if !self.accelerate_config_xml.is_empty() && let Err(e) = deserialize::(&self.accelerate_config_xml).map(|c| self.accelerate_config = Some(c)) { - tracing::warn!(bucket = %self.name, config = "accelerate", error = %e, "parse_all_configs: failed to parse"); + tracing::warn!( + event = "bucket_metadata_parse_failed", + component = "ecstore", + subsystem = "bucket_metadata", + bucket = %self.name, + config = "accelerate", + error = %e, + "Failed to parse bucket metadata config" + ); } if !self.request_payment_config_xml.is_empty() && let Err(e) = deserialize::(&self.request_payment_config_xml) .map(|c| self.request_payment_config = Some(c)) { tracing::warn!( + event = "bucket_metadata_parse_failed", + component = "ecstore", + subsystem = "bucket_metadata", bucket = %self.name, config = "request_payment", error = %e, - "parse_all_configs: failed to parse" + "Failed to parse bucket metadata config" ); } if !self.public_access_block_config_xml.is_empty() && let Err(e) = deserialize::(&self.public_access_block_config_xml) .map(|c| self.public_access_block_config = Some(c)) { - tracing::warn!(bucket = %self.name, config = "public_access_block", error = %e, "parse_all_configs: failed to parse"); + tracing::warn!( + event = "bucket_metadata_parse_failed", + component = "ecstore", + subsystem = "bucket_metadata", + bucket = %self.name, + config = "public_access_block", + error = %e, + "Failed to parse bucket metadata config" + ); } if !self.bucket_acl_config_json.is_empty() && let Err(e) = String::from_utf8(self.bucket_acl_config_json.clone()).map(|acl| self.bucket_acl_config = Some(acl)) { - tracing::warn!(bucket = %self.name, config = "bucket_acl", error = %e, "parse_all_configs: failed to parse"); + tracing::warn!( + event = "bucket_metadata_parse_failed", + component = "ecstore", + subsystem = "bucket_metadata", + bucket = %self.name, + config = "bucket_acl", + error = %e, + "Failed to parse bucket metadata config" + ); } Ok(()) diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index 95fd81424..6c861daad 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -56,7 +56,15 @@ use tokio::sync::mpsc::Sender; use tokio::task::JoinHandle; use tokio::time::Duration; use tokio_util::sync::CancellationToken; -use tracing::{info, instrument, warn}; +use tracing::{debug, info, instrument}; + +const LOG_COMPONENT_ECSTORE: &str = "ecstore"; +const LOG_SUBSYSTEM_REPLICATION: &str = "replication"; +const EVENT_REPLICATION_WORKER_RESIZE_SKIPPED: &str = "replication_worker_resize_skipped"; +const EVENT_REPLICATION_WORKER_RESIZED: &str = "replication_worker_resized"; +const EVENT_REPLICATION_BACKPRESSURE: &str = "replication_backpressure"; +const EVENT_REPLICATION_RESYNC_LOAD_SKIPPED: &str = "replication_resync_load_skipped"; +const EVENT_REPLICATION_CONFIG_LOOKUP_SKIPPED: &str = "replication_config_lookup_skipped"; // Worker limits pub const WORKER_MAX_LIMIT: usize = 500; @@ -358,18 +366,31 @@ impl ReplicationPool { let mut workers = self.workers.write().await; if (check_old > 0 && workers.len() != check_old) || n == workers.len() || n < 1 { - warn!( - "resize_workers: skipping resize - check_old_mismatch={}, same_size={}, invalid_n={}", - check_old > 0 && workers.len() != check_old, - n == workers.len(), - n < 1 + debug!( + event = EVENT_REPLICATION_WORKER_RESIZE_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + check_old_mismatch = check_old > 0 && workers.len() != check_old, + same_size = n == workers.len(), + invalid_target_size = n < 1, + current_workers = workers.len(), + target_workers = n, + "Skipped replication worker resize" ); return; } // Add workers if needed if workers.len() < n { - info!("resize_workers: adding workers from {} to {}", workers.len(), n); + info!( + event = EVENT_REPLICATION_WORKER_RESIZED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + action = "increase", + from_workers = workers.len(), + to_workers = n, + "Resized replication workers" + ); } while workers.len() < n { @@ -416,7 +437,15 @@ impl ReplicationPool { // Remove workers if needed if workers.len() > n { - warn!("resize_workers: removing workers from {} to {}", workers.len(), n); + info!( + event = EVENT_REPLICATION_WORKER_RESIZED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + action = "decrease", + from_workers = workers.len(), + to_workers = n, + "Resized replication workers" + ); } while workers.len() > n { @@ -607,11 +636,26 @@ impl ReplicationPool { match priority { ReplicationPriority::Fast => { - // Log warning about unable to keep up - info!("Warning: Unable to keep up with incoming traffic"); + debug!( + event = EVENT_REPLICATION_BACKPRESSURE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + queue_type = "object", + priority = "fast", + recommendation = "none", + "Replication queue is backpressured" + ); } ReplicationPriority::Slow => { - info!("Warning: Unable to keep up with incoming traffic - recommend increasing replication priority to auto"); + debug!( + event = EVENT_REPLICATION_BACKPRESSURE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + queue_type = "object", + priority = "slow", + recommendation = "set_priority_auto", + "Replication queue is backpressured" + ); } ReplicationPriority::Auto => { let max_w = std::cmp::min(max_workers, WORKER_MAX_LIMIT); @@ -667,10 +711,26 @@ impl ReplicationPool { match priority { ReplicationPriority::Fast => { - info!("Warning: Unable to keep up with incoming deletes"); + debug!( + event = EVENT_REPLICATION_BACKPRESSURE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + queue_type = "delete", + priority = "fast", + recommendation = "none", + "Replication delete queue is backpressured" + ); } ReplicationPriority::Slow => { - info!("Warning: Unable to keep up with incoming deletes - recommend increasing replication priority to auto"); + debug!( + event = EVENT_REPLICATION_BACKPRESSURE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + queue_type = "delete", + priority = "slow", + recommendation = "set_priority_auto", + "Replication delete queue is backpressured" + ); } ReplicationPriority::Auto => { let max_w = std::cmp::min(max_workers, WORKER_MAX_LIMIT); @@ -930,7 +990,15 @@ impl ReplicationPool { Ok(meta) => meta, Err(err) => { if !matches!(err, EcstoreError::VolumeNotFound) { - warn!("Error loading resync metadata for bucket {bucket}: {err:?}"); + debug!( + event = EVENT_REPLICATION_RESYNC_LOAD_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + bucket, + error = ?err, + reason = "metadata_load_failed", + "Skipped replication resync metadata load" + ); } continue; } @@ -1179,7 +1247,15 @@ pub async fn queue_replication_heal(bucket: &str, oi: ObjectInfo, retry_count: u let rcfg = match metadata_sys::get_replication_config(bucket).await { Ok((config, _)) => config, Err(err) => { - warn!("Failed to get replication config for bucket {}: {}", bucket, err); + debug!( + event = EVENT_REPLICATION_CONFIG_LOOKUP_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + bucket, + error = %err, + reason = "config_lookup_failed", + "Skipped replication heal queue due to missing replication config" + ); return; } @@ -1188,7 +1264,15 @@ pub async fn queue_replication_heal(bucket: &str, oi: ObjectInfo, retry_count: u let tgts = match BucketTargetSys::get().list_bucket_targets(bucket).await { Ok(targets) => Some(targets), Err(err) => { - warn!("Failed to list bucket targets for bucket {}: {}", bucket, err); + debug!( + event = EVENT_REPLICATION_CONFIG_LOOKUP_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + bucket, + error = %err, + reason = "target_list_failed", + "Skipped bucket target list during replication heal queue setup" + ); None } }; diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 406f789e2..acdf26b53 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -85,9 +85,18 @@ use tokio::task::JoinSet; use tokio::time::Duration as TokioDuration; use tokio_util::io::ReaderStream; use tokio_util::sync::CancellationToken; -use tracing::{debug, error, info, instrument, warn}; +use tracing::{debug, error, instrument, warn}; use uuid::Uuid; +const LOG_COMPONENT_ECSTORE: &str = "ecstore"; +const LOG_SUBSYSTEM_REPLICATION_RESYNC: &str = "replication_resync"; +const EVENT_RESYNC_STATUS_UPDATE_SKIPPED: &str = "replication_resync_status_update_skipped"; +const EVENT_RESYNC_CONFIG_LOOKUP_SKIPPED: &str = "replication_resync_config_lookup_skipped"; +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"; + pub(crate) const REPLICATION_DIR: &str = ".replication"; pub(crate) const RESYNC_FILE_NAME: &str = "resync.bin"; pub(crate) const RESYNC_META_FORMAT: u16 = 1; @@ -508,12 +517,16 @@ impl ReplicationResyncer { }; if !resync_state_accepts_update(state, &opts) { - warn!( + debug!( + event = EVENT_RESYNC_STATUS_UPDATE_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, bucket = %opts.bucket, arn = %opts.arn, incoming_resync_id = %opts.resync_id, current_resync_id = %state.resync_id, - "ignoring stale resync status update" + reason = "stale_status_update", + "Skipped stale resync status update" ); return Ok(()); } @@ -565,12 +578,16 @@ impl ReplicationResyncer { }; if !resync_state_accepts_update(state, &opts) { - warn!( + debug!( + event = EVENT_RESYNC_STATUS_UPDATE_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, bucket = %opts.bucket, arn = %opts.arn, incoming_resync_id = %opts.resync_id, current_resync_id = %state.resync_id, - "ignoring stale resync stats update" + reason = "stale_stats_update", + "Skipped stale resync stats update" ); return; } @@ -676,7 +693,15 @@ impl ReplicationResyncer { let targets = match BucketTargetSys::get().list_bucket_targets(&opts.bucket).await { Ok(targets) => targets, Err(err) => { - warn!("Failed to list bucket targets: {}", err); + debug!( + event = EVENT_RESYNC_CONFIG_LOOKUP_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %opts.bucket, + error = %err, + reason = "target_list_failed", + "Failed to list bucket targets during resync" + ); self.resync_bucket_mark_status(ResyncStatusType::ResyncFailed, opts.clone(), storage.clone()) .await; return; @@ -847,14 +872,17 @@ impl ReplicationResyncer { (roi.size, None) }; - info!( - "resynced reset_id:{} object: {}/{}-{} size:{} err:{:?}", - reset_id, - bucket_name, - roi.name, - roi.version_id.unwrap_or_default(), + debug!( + event = EVENT_RESYNC_OBJECT_PROCESSED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + reset_id = %reset_id, + bucket = %bucket_name, + object = %roi.name, + version_id = %roi.version_id.unwrap_or_default(), size, - err, + error = ?err, + "Processed resync object" ); if cancel_token.is_cancelled() { @@ -1548,7 +1576,14 @@ pub async fn replicate_delete(dobj: DeletedObjectReplicationInfo, let _rcfg = match get_replication_config(&bucket).await { Ok(Some(config)) => config, Ok(None) => { - warn!("No replication config found for bucket: {}", bucket); + debug!( + event = EVENT_REPLICATION_DELETE_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + reason = "replication_config_missing", + "Skipping replication delete because replication config is missing" + ); send_event(EventArgs { event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), @@ -1567,7 +1602,15 @@ pub async fn replicate_delete(dobj: DeletedObjectReplicationInfo, return; } Err(err) => { - warn!("replication config for bucket: {} error: {}", bucket, err); + debug!( + event = EVENT_REPLICATION_DELETE_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + error = %err, + reason = "replication_config_lookup_failed", + "Skipping replication delete because replication config lookup failed" + ); send_event(EventArgs { event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), @@ -1605,30 +1648,42 @@ pub async fn replicate_delete(dobj: DeletedObjectReplicationInfo, match source_marker_state { Ok(info) if info.delete_marker && info.version_id == Some(delete_marker_version_id) => {} Ok(_) => { - warn!( + debug!( + event = EVENT_REPLICATION_DELETE_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, bucket, object = dobj.delete_object.object_name, version_id = %delete_marker_version_id, - "skipping stale delete-marker replication because source version is no longer a delete marker" + reason = "source_not_delete_marker", + "Skipping stale delete-marker replication" ); return; } Err(err) if is_err_object_not_found(&err) || is_err_version_not_found(&err) => { - warn!( + debug!( + event = EVENT_REPLICATION_DELETE_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, bucket, object = dobj.delete_object.object_name, version_id = %delete_marker_version_id, - "skipping stale delete-marker replication because source version no longer exists" + reason = "source_version_missing", + "Skipping stale delete-marker replication" ); return; } Err(err) => { - warn!( + debug!( + event = EVENT_REPLICATION_DELETE_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, bucket, object = dobj.delete_object.object_name, version_id = %delete_marker_version_id, error = %err, - "failed to verify source delete-marker state before replication" + reason = "source_state_verification_failed", + "Failed to verify source delete-marker state before replication" ); } } @@ -1645,9 +1700,15 @@ pub async fn replicate_delete(dobj: DeletedObjectReplicationInfo, ) { Ok(dsc) => dsc, Err(err) => { - warn!( - "failed to parse replicate decision for bucket:{} arn:{} error:{}", - bucket, dobj.target_arn, err + debug!( + event = EVENT_REPLICATION_DELETE_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + arn = %dobj.target_arn, + error = %err, + reason = "replicate_decision_parse_failed", + "Failed to parse replicate decision" ); send_event(EventArgs { event_name: EventName::ObjectReplicationNotTracked.to_string(), @@ -1672,9 +1733,15 @@ pub async fn replicate_delete(dobj: DeletedObjectReplicationInfo, { Ok(ns_lock) => ns_lock, Err(e) => { - warn!( - "failed to get ns lock for bucket:{} object:{} error:{}", - bucket, dobj.delete_object.object_name, e + debug!( + event = EVENT_REPLICATION_DELETE_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + object = %dobj.delete_object.object_name, + error = %e, + reason = "ns_lock_unavailable", + "Skipping replication delete" ); send_event(EventArgs { event_name: EventName::ObjectReplicationNotTracked.to_string(), @@ -1697,9 +1764,15 @@ pub async fn replicate_delete(dobj: DeletedObjectReplicationInfo, let _lock_guard = match ns_lock.get_write_lock(get_lock_acquire_timeout()).await { Ok(lock_guard) => lock_guard, Err(e) => { - warn!( - "failed to get write lock for bucket:{} object:{} error:{}", - bucket, dobj.delete_object.object_name, e + debug!( + event = EVENT_REPLICATION_DELETE_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + object = %dobj.delete_object.object_name, + error = %e, + reason = "write_lock_unavailable", + "Skipping replication delete" ); send_event(EventArgs { event_name: EventName::ObjectReplicationNotTracked.to_string(), @@ -1741,7 +1814,15 @@ pub async fn replicate_delete(dobj: DeletedObjectReplicationInfo, // Get the remote target client let Some(tgt_client) = BucketTargetSys::get().get_remote_target_client(&bucket, &tgt_entry.arn).await else { - warn!("failed to get target for bucket:{:?}, arn:{:?}", &bucket, &tgt_entry.arn); + debug!( + event = EVENT_REPLICATION_DELETE_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + arn = %tgt_entry.arn, + reason = "target_client_missing", + "Skipping replication delete because target client is unavailable" + ); send_event(EventArgs { event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), @@ -1969,7 +2050,14 @@ async fn replicate_force_delete_to_targets(dobj: &DeletedObjectRe let rcfg = match get_replication_config(bucket).await { Ok(Some(config)) => config, Ok(None) => { - warn!("replicate force-delete: no replication config for bucket:{}", bucket); + debug!( + event = EVENT_REPLICATION_FORCE_DELETE_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + reason = "replication_config_missing", + "Skipping replication force-delete because replication config is missing" + ); send_event(EventArgs { event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), @@ -1985,7 +2073,15 @@ async fn replicate_force_delete_to_targets(dobj: &DeletedObjectRe return; } Err(err) => { - warn!("replicate force-delete: replication config error bucket:{} error:{}", bucket, err); + debug!( + event = EVENT_REPLICATION_FORCE_DELETE_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + error = %err, + reason = "replication_config_lookup_failed", + "Skipping replication force-delete because replication config lookup failed" + ); send_event(EventArgs { event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), @@ -2064,7 +2160,15 @@ async fn replicate_force_delete_to_targets(dobj: &DeletedObjectRe for arn in tgt_arns { let Some(tgt_client) = BucketTargetSys::get().get_remote_target_client(bucket, &arn).await else { - warn!("replicate force-delete: failed to get target client bucket:{} arn:{}", bucket, arn); + debug!( + event = EVENT_REPLICATION_FORCE_DELETE_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + arn = %arn, + reason = "target_client_missing", + "Skipping replication force-delete because target client is unavailable" + ); send_event(EventArgs { event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), @@ -2300,7 +2404,14 @@ pub async fn replicate_object(roi: ReplicateObjectInfo, storage: let cfg = match get_replication_config(&bucket).await { Ok(Some(config)) => config, Ok(None) => { - warn!("No replication config found for bucket: {}", bucket); + debug!( + event = EVENT_RESYNC_CONFIG_LOOKUP_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + reason = "replication_config_missing", + "Skipping replication object because replication config is missing" + ); send_event(EventArgs { event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), @@ -2338,7 +2449,15 @@ pub async fn replicate_object(roi: ReplicateObjectInfo, storage: for arn in tgt_arns { let Some(tgt_client) = BucketTargetSys::get().get_remote_target_client(&bucket, &arn).await else { - warn!("failed to get target for bucket:{:?}, arn:{:?}", &bucket, &arn); + debug!( + event = EVENT_RESYNC_RUNTIME_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + arn = %arn, + reason = "target_client_missing", + "Skipping replication object target" + ); send_event(EventArgs { event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), @@ -2479,7 +2598,16 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { } if BucketTargetSys::get().is_offline(&tgt_client.to_url()).await { - warn!("target is offline: {}", tgt_client.to_url()); + debug!( + event = EVENT_RESYNC_RUNTIME_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 object target" + ); send_event(EventArgs { event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), @@ -2509,7 +2637,16 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { Ok(gr) => gr, Err(e) => { if !is_err_object_not_found(&e) || is_err_version_not_found(&e) { - warn!("failed to get object reader for bucket:{} arn:{} error:{}", bucket, tgt_client.arn, e); + debug!( + event = EVENT_RESYNC_RUNTIME_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + arn = %tgt_client.arn, + error = %e, + reason = "object_reader_unavailable", + "Skipping replication object target" + ); send_event(EventArgs { event_name: EventName::ObjectReplicationNotTracked.to_string(), @@ -2532,7 +2669,16 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { let size = match object_info.get_actual_size() { Ok(size) => size, Err(e) => { - warn!("failed to get actual size for bucket:{} arn:{} error:{}", bucket, tgt_client.arn, e); + debug!( + event = EVENT_RESYNC_RUNTIME_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + arn = %tgt_client.arn, + error = %e, + reason = "actual_size_unavailable", + "Skipping replication object target" + ); send_event(EventArgs { event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), @@ -2546,7 +2692,15 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { }; if tgt_client.bucket.is_empty() { - warn!("target bucket is empty: {}", tgt_client.bucket); + debug!( + event = EVENT_RESYNC_RUNTIME_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + arn = %tgt_client.arn, + reason = "target_bucket_empty", + "Skipping replication object target" + ); send_event(EventArgs { event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), @@ -2702,7 +2856,16 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { }; if BucketTargetSys::get().is_offline(&tgt_client.to_url()).await { - warn!("target is offline: {}", tgt_client.to_url()); + debug!( + event = EVENT_RESYNC_RUNTIME_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + arn = %tgt_client.arn, + target = %tgt_client.to_url(), + reason = "target_offline", + "Skipped replication because target is offline" + ); send_event(EventArgs { event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), @@ -2732,7 +2895,16 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { Ok(gr) => gr, Err(e) => { if !is_err_object_not_found(&e) || is_err_version_not_found(&e) { - warn!("failed to get object reader for bucket:{} arn:{} error:{}", bucket, tgt_client.arn, e); + debug!( + event = EVENT_RESYNC_RUNTIME_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + arn = %tgt_client.arn, + error = %e, + reason = "object_reader_unavailable", + "Skipped replication because object reader is unavailable" + ); send_event(EventArgs { event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), @@ -2763,7 +2935,16 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { let size = match object_info.get_actual_size() { Ok(size) => size, Err(e) => { - warn!("failed to get actual size for bucket:{} arn:{} error:{}", bucket, tgt_client.arn, e); + debug!( + event = EVENT_RESYNC_RUNTIME_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + arn = %tgt_client.arn, + error = %e, + reason = "actual_size_unavailable", + "Skipped replication because actual object size is unavailable" + ); send_event(EventArgs { event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), @@ -2779,7 +2960,15 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { // TODO: SSE if tgt_client.bucket.is_empty() { - warn!("target bucket is empty: {}", tgt_client.bucket); + debug!( + event = EVENT_RESYNC_RUNTIME_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + arn = %tgt_client.arn, + reason = "target_bucket_empty", + "Skipped replication because target bucket is empty" + ); send_event(EventArgs { event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), @@ -2801,7 +2990,16 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { }; if let Err(err) = sopts.set(AMZ_TAGGING_DIRECTIVE, "ACCESS") { - warn!("failed to set replication tagging directive header: {err}"); + debug!( + event = EVENT_RESYNC_RUNTIME_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + arn = %tgt_client.arn, + error = %err, + reason = "tagging_directive_header_invalid", + "Skipped replication tagging directive header detail" + ); } match head_object_with_proxy_stats( @@ -2877,9 +3075,15 @@ 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 + debug!( + event = EVENT_RESYNC_RUNTIME_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + arn = %tgt_client.arn, + error = %e2, + reason = "head_object_fallback_failed", + "Failed replication head-object fallback" ); send_event(EventArgs { event_name: EventName::ObjectReplicationNotTracked.to_string(), @@ -2897,7 +3101,16 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { replication_action = ReplicationAction::All; } else { rinfo.error = Some(e.to_string()); - warn!("failed to head object for bucket:{} arn:{} error:{}", bucket, tgt_client.arn, e); + debug!( + event = EVENT_RESYNC_RUNTIME_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + arn = %tgt_client.arn, + error = %e, + reason = "head_object_failed", + "Skipped replication because head-object failed" + ); send_event(EventArgs { event_name: EventName::ObjectReplicationNotTracked.to_string(), diff --git a/crates/ecstore/src/client/object_handlers_common.rs b/crates/ecstore/src/client/object_handlers_common.rs index cf57b9aa5..c4aa96272 100644 --- a/crates/ecstore/src/client/object_handlers_common.rs +++ b/crates/ecstore/src/client/object_handlers_common.rs @@ -13,7 +13,12 @@ // limitations under the License. use std::sync::Arc; -use tracing::warn; +use tracing::debug; + +const LOG_COMPONENT_ECSTORE: &str = "ecstore"; +const LOG_SUBSYSTEM_LIFECYCLE: &str = "lifecycle"; +const EVENT_LIFECYCLE_CLEANUP_SKIPPED: &str = "lifecycle_cleanup_skipped"; +const EVENT_LIFECYCLE_CLEANUP_FAILED: &str = "lifecycle_cleanup_failed"; use crate::bucket::lifecycle::lifecycle; use crate::bucket::replication::{DeletedObjectReplicationInfo, check_replicate_delete, schedule_replication_delete}; @@ -40,7 +45,15 @@ pub async fn delete_object_versions(api: &Arc, bucket: &str, to_del: &[ let version_suspended = match BucketVersioningSys::get(bucket).await { Ok(vc) => vc.suspended(), Err(err) => { - warn!(bucket, error = ?err, "failed to get versioning config during lifecycle noncurrent version cleanup"); + debug!( + event = EVENT_LIFECYCLE_CLEANUP_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + bucket, + error = ?err, + reason = "versioning_config_unavailable", + "Skipped lifecycle noncurrent version cleanup" + ); return; } }; @@ -70,12 +83,16 @@ pub async fn delete_object_versions(api: &Arc, bucket: &str, to_del: &[ .then(|| lifecycle_version_delete_replication_state(dsc.to_string(), dsc.pending_status())) } Err(err) => { - warn!( + debug!( + event = EVENT_LIFECYCLE_CLEANUP_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, bucket, object = %object.object_name, version_id = ?version_id, error = ?err, - "failed to get object info during lifecycle noncurrent version cleanup; skipping delete replication scheduling" + reason = "object_info_unavailable", + "Skipped lifecycle delete replication scheduling" ); None } @@ -119,7 +136,16 @@ pub async fn delete_object_versions(api: &Arc, bucket: &str, to_del: &[ .and_then(|o| o.version_id) .map(|v| v.to_string()) .unwrap_or_default(); - warn!(bucket, object = obj_name, version_id = %vid, error = ?e, "failed to delete noncurrent version during lifecycle cleanup"); + debug!( + event = EVENT_LIFECYCLE_CLEANUP_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + bucket, + object = obj_name, + version_id = %vid, + error = ?e, + "Failed lifecycle noncurrent version cleanup" + ); } } if remaining.is_empty() { diff --git a/crates/ecstore/src/rebalance.rs b/crates/ecstore/src/rebalance.rs index 14c94f1b6..065b026b3 100644 --- a/crates/ecstore/src/rebalance.rs +++ b/crates/ecstore/src/rebalance.rs @@ -39,7 +39,14 @@ use std::sync::Arc; use time::OffsetDateTime; use tokio::time::{Duration, Instant}; use tokio_util::sync::CancellationToken; -use tracing::{error, info, warn}; +use tracing::{debug, error, info, warn}; + +const LOG_COMPONENT_ECSTORE: &str = "ecstore"; +const LOG_SUBSYSTEM_REBALANCE: &str = "rebalance"; +const EVENT_REBALANCE_STATE: &str = "rebalance_state"; +const EVENT_REBALANCE_BUCKET: &str = "rebalance_bucket"; +const EVENT_REBALANCE_ENTRY: &str = "rebalance_entry"; +const EVENT_REBALANCE_LISTING: &str = "rebalance_listing"; use uuid::Uuid; const REBAL_META_FMT: u16 = 1; // Replace with actual format value @@ -573,7 +580,13 @@ impl RebalanceMeta { pub async fn load_with_opts(&mut self, store: Arc, opts: ObjectOptions) -> Result<()> { let (data, _) = read_config_with_metadata(store, REBAL_META_NAME, &opts).await?; if data.is_empty() { - info!("rebalanceMeta load_with_opts: no data"); + debug!( + event = EVENT_REBALANCE_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + state = "metadata_empty", + "Rebalance metadata is empty" + ); return Ok(()); } if data.len() <= 4 { @@ -595,7 +608,13 @@ impl RebalanceMeta { self.last_refreshed_at = Some(OffsetDateTime::now_utc()); - info!("rebalanceMeta load_with_opts: loaded meta done"); + debug!( + event = EVENT_REBALANCE_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + state = "metadata_loaded", + "Loaded rebalance metadata" + ); Ok(()) } @@ -605,7 +624,14 @@ impl RebalanceMeta { pub async fn save_with_opts(&self, store: Arc, opts: ObjectOptions) -> Result<()> { if self.pool_stats.is_empty() { - info!("rebalanceMeta save_with_opts: no pool stats"); + debug!( + event = EVENT_REBALANCE_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + state = "metadata_save_skipped", + reason = "no_pool_stats", + "Skipped rebalance metadata save" + ); return Ok(()); } @@ -658,10 +684,15 @@ impl ECStore { #[tracing::instrument(skip_all)] pub async fn load_rebalance_meta(&self) -> Result<()> { let mut meta = RebalanceMeta::new(); - info!("rebalanceMeta: store load rebalance meta"); + debug!( + event = EVENT_REBALANCE_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + state = "metadata_loading", + "Loading rebalance metadata" + ); let pool = clone_first_arc(&self.pools, "rebalanceMeta: no pools available")?; if resolve_rebalance_meta_load_result(meta.load(pool).await)? { - info!("rebalanceMeta: rebalance meta loaded0"); { let mut rebalance_meta = self.rebalance_meta.write().await; @@ -670,12 +701,23 @@ impl ECStore { drop(rebalance_meta); } - info!("rebalanceMeta: rebalance meta loaded1"); - resolve_load_rebalance_stats_update_result(self.update_rebalance_stats().await)?; - info!("rebalanceMeta: rebalance meta loaded2"); + debug!( + event = EVENT_REBALANCE_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + state = "metadata_loaded", + "Loaded rebalance metadata" + ); } else { - info!("rebalanceMeta: not found, rebalance not started"); + debug!( + event = EVENT_REBALANCE_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + state = "metadata_missing", + reason = "rebalance_not_started", + "Rebalance metadata not found" + ); } Ok(()) @@ -690,13 +732,25 @@ impl ECStore { clone_rebalance_pool_stats(rebalance_meta.as_ref())? }; - info!("update_rebalance_stats: pool_stats: {:?}", &pool_stats); + debug!( + event = EVENT_REBALANCE_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_count = pool_stats.len(), + "Refreshing rebalance stats snapshot" + ); for i in 0..self.pools.len() { if pool_stats.get(i).is_none() { - info!("update_rebalance_stats: pool {} not found", i); let mut rebalance_meta = self.rebalance_meta.write().await; - info!("update_rebalance_stats: pool {} not found, add", i); + debug!( + event = EVENT_REBALANCE_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index = i, + state = "pool_stat_missing", + "Adding missing rebalance pool stats entry" + ); if let Some(meta) = rebalance_meta.as_mut() { meta.pool_stats.push(RebalanceStats::default()); } @@ -706,7 +760,13 @@ impl ECStore { } if ok { - info!("update_rebalance_stats: save rebalance meta"); + debug!( + event = EVENT_REBALANCE_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + state = "metadata_saving", + "Saving rebalance metadata after stats refresh" + ); let rebalance_meta = self.rebalance_meta.read().await; if let Some(meta) = rebalance_meta.as_ref() { @@ -732,7 +792,14 @@ impl ECStore { #[tracing::instrument(skip(self))] pub async fn init_rebalance_meta(&self, bucktes: Vec) -> Result { - info!("init_rebalance_meta: start rebalance"); + info!( + event = EVENT_REBALANCE_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + state = "initializing", + bucket_count = bucktes.len(), + "Initializing rebalance metadata" + ); let si = StorageAdminApi::storage_info(self).await; let mut disk_stats = vec![DiskStat::default(); self.pools.len()]; @@ -924,7 +991,13 @@ impl ECStore { #[tracing::instrument(skip_all)] pub async fn start_rebalance(self: &Arc) -> Result<()> { - info!("start_rebalance: start rebalance"); + info!( + event = EVENT_REBALANCE_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + state = "starting", + "Starting rebalance" + ); let decommission_running = self.is_decommission_running().await; // let rebalance_meta = self.rebalance_meta.read().await; @@ -940,7 +1013,14 @@ impl ECStore { return Err(Error::ConfigNotFound); }; if should_skip_start_rebalance(meta.cancel.is_some(), is_rebalance_in_progress(meta)) { - info!("start_rebalance: already in progress, skip duplicate start"); + debug!( + event = EVENT_REBALANCE_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + state = "start_skipped", + reason = "already_in_progress", + "Skipped duplicate rebalance start" + ); return Ok(()); } if complete_rebalance_pools_at_goal(meta, OffsetDateTime::now_utc()) { @@ -963,24 +1043,45 @@ impl ECStore { let participants = if let Some(ref meta) = *self.rebalance_meta.read().await { resolve_rebalance_participants(meta.pool_stats.as_slice(), self.pools.len()) } else { - info!("start_rebalance:2 rebalance_meta is None exit"); + debug!( + event = EVENT_REBALANCE_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + state = "start_skipped", + reason = "metadata_missing", + "Skipped rebalance start because metadata is unavailable" + ); Vec::new() }; for (idx, participating) in participants.iter().enumerate() { if !*participating { - info!("start_rebalance: pool {} is not participating, skipping", idx); + debug!( + event = EVENT_REBALANCE_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index = idx, + state = "pool_skipped", + reason = "not_participating", + "Skipped rebalance pool" + ); continue; } - if !get_global_endpoints().as_ref().get(idx).is_some_and(|v| { - info!("start_rebalance: pool {} endpoints: {:?}", idx, v.endpoints); - v.endpoints.as_ref().first().is_some_and(|e| { - info!("start_rebalance: pool {} endpoint: {:?}, is_local: {}", idx, e, e.is_local); - e.is_local - }) - }) { - info!("start_rebalance: pool {} is not local, skipping", idx); + if !get_global_endpoints() + .as_ref() + .get(idx) + .is_some_and(|v| v.endpoints.as_ref().first().is_some_and(|e| e.is_local)) + { + debug!( + event = EVENT_REBALANCE_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index = idx, + state = "pool_skipped", + reason = "not_local", + "Skipped non-local rebalance pool" + ); continue; } @@ -991,12 +1092,25 @@ impl ECStore { if let Err(err) = store.rebalance_buckets(rx_clone, pool_idx).await { error!("Rebalance failed for pool {}: {}", pool_idx, err); } else { - info!("Rebalance completed for pool {}", pool_idx); + info!( + event = EVENT_REBALANCE_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index = pool_idx, + state = "completed", + "Completed rebalance pool" + ); } }); } - info!("start_rebalance: rebalance started done"); + info!( + event = EVENT_REBALANCE_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + state = "started", + "Started rebalance" + ); Ok(()) } @@ -1029,9 +1143,13 @@ impl ECStore { pool_stat.info.status, &terminal_event, ) { - info!( - "rebalance_buckets: preserving stopped status for pool {}", - pool_index + debug!( + event = EVENT_REBALANCE_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + state = "stopped_preserved", + "Preserved stopped rebalance status" ); } else { apply_rebalance_terminal_event( @@ -1058,11 +1176,27 @@ impl ECStore { return Err(wrapped); } } else { - info!(msg); + debug!( + event = EVENT_REBALANCE_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + state = "metadata_saved", + message = %msg, + "Saved rebalance metadata" + ); } if quit { - info!("{}: exiting save_task", msg); + debug!( + event = EVENT_REBALANCE_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + state = "save_task_exiting", + message = %msg, + "Exiting rebalance save task" + ); return Ok(()); } @@ -1070,13 +1204,28 @@ impl ECStore { } }); - info!("Pool {} rebalancing is started", pool_index); + info!( + event = EVENT_REBALANCE_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + state = "pool_started", + "Started rebalance worker" + ); let mut final_result: Result<()> = Ok(()); let mut deferred_buckets = HashSet::new(); loop { if rx.is_cancelled() { - info!("Pool {} rebalancing is stopped", pool_index); + info!( + event = EVENT_REBALANCE_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + state = "pool_stopped", + reason = "cancelled", + "Stopped rebalance worker" + ); let err = Error::OperationCanceled; final_result = Err(resolve_rebalance_terminal_error( err.clone(), @@ -1098,7 +1247,15 @@ impl ECStore { }; if let Some(bucket) = next_bucket { - info!("Rebalancing bucket: start {}", bucket); + debug!( + event = EVENT_REBALANCE_BUCKET, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + bucket = %bucket, + state = "started", + "Starting rebalance bucket" + ); let outcome = match resolve_rebalance_bucket_result( self.rebalance_bucket(rx.clone(), bucket.clone(), pool_index).await, @@ -1129,7 +1286,16 @@ impl ECStore { break; } - warn!("Rebalance bucket deferred due to transient object failures: {bucket}: {last_error}"); + warn!( + event = EVENT_REBALANCE_BUCKET, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + bucket = %bucket, + state = "deferred", + error = %last_error, + "Deferred rebalance bucket after transient object failures" + ); if let Err(err) = self.defer_rebalance_bucket(pool_index, bucket.clone(), last_error).await { error!("defer_rebalance_bucket failed for pool {}: {:?}", pool_index, err); final_result = Err(resolve_rebalance_terminal_error( @@ -1141,7 +1307,15 @@ impl ECStore { continue; } - info!("Rebalance bucket: done {} ", bucket); + debug!( + event = EVENT_REBALANCE_BUCKET, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + bucket = %bucket, + state = "completed", + "Completed rebalance bucket" + ); if let Err(err) = self.bucket_rebalance_done(pool_index, bucket).await { error!("bucket_rebalance_done failed for pool {}: {:?}", pool_index, err); final_result = Err(resolve_rebalance_terminal_error( @@ -1151,12 +1325,27 @@ impl ECStore { break; } } else { - info!("Rebalance bucket: no bucket to rebalance"); + debug!( + event = EVENT_REBALANCE_BUCKET, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + state = "idle", + reason = "no_bucket_to_rebalance", + "No rebalance bucket available" + ); break; } } - info!("Pool {} rebalancing is done", pool_index); + info!( + event = EVENT_REBALANCE_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + state = "pool_done", + "Finished rebalance worker" + ); if final_result.is_ok() && let Err(err) = send_rebalance_done_signal(&done_tx, Ok(()), pool_index).await @@ -1181,7 +1370,14 @@ impl ECStore { { // Check if the pool's rebalance status is already completed if pool_stat.info.status == RebalStatus::Completed { - info!("check_if_rebalance_done: pool {} is already completed", pool_index); + debug!( + event = EVENT_REBALANCE_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + state = "already_completed", + "Rebalance pool is already completed" + ); return true; } @@ -1202,7 +1398,15 @@ impl ECStore { { pool_stat.info.status = RebalStatus::Completed; pool_stat.info.end_time = Some(OffsetDateTime::now_utc()); - info!("check_if_rebalance_done: pool {} is completed, pfi: {}", pool_index, pfi); + info!( + event = EVENT_REBALANCE_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + state = "completed", + percent_free = pfi, + "Marked rebalance pool completed" + ); return true; } } @@ -1271,21 +1475,53 @@ fn resolve_next_rebalance_bucket(meta: Option<&RebalanceMeta>, pool_index: usize }; if pool_stat.info.status == RebalStatus::Completed || !pool_stat.participating { - info!("next_rebal_bucket: pool_index: {} completed or not participating", pool_index); + debug!( + event = EVENT_REBALANCE_BUCKET, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + state = "unavailable", + reason = "completed_or_not_participating", + "No rebalance bucket available" + ); return Ok(None); } if pool_stat.buckets.is_empty() { - info!("next_rebal_bucket: pool_index: {} buckets is empty", pool_index); + debug!( + event = EVENT_REBALANCE_BUCKET, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + state = "unavailable", + reason = "bucket_queue_empty", + "No rebalance bucket available" + ); return Ok(None); } if let Some(bucket) = next_rebal_bucket_from_stat(pool_stat) { - info!("next_rebal_bucket: pool_index: {} bucket: {}", pool_index, bucket); + debug!( + event = EVENT_REBALANCE_BUCKET, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + bucket = %bucket, + state = "selected", + "Selected rebalance bucket" + ); return Ok(Some(bucket)); } - info!("next_rebal_bucket: pool_index: {} None", pool_index); + debug!( + event = EVENT_REBALANCE_BUCKET, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + state = "unavailable", + reason = "selection_returned_none", + "No rebalance bucket available" + ); Ok(None) } @@ -1299,10 +1535,17 @@ fn mark_rebalance_bucket_done(meta: Option<&mut RebalanceMeta>, pool_index: usiz return Err(invalid_rebalance_pool_index_error(pool_index, meta.pool_stats.len())); }; - info!("bucket_rebalance_done: buckets {:?}", &pool_stat.buckets); - if take_bucket_from_rebalance_queue(pool_stat, bucket) { - info!("bucket_rebalance_done: bucket {} rebalanced", bucket); + debug!( + event = EVENT_REBALANCE_BUCKET, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + bucket = %bucket, + state = "queue_removed", + remaining_bucket_count = pool_stat.buckets.len(), + "Removed bucket from rebalance queue" + ); if has_deferred_rebalance_error(pool_stat) { pool_stat.info.last_error = None; } @@ -2036,7 +2279,16 @@ impl ECStore { bucket_configs: Arc, // wk: Arc, ) -> Result { - info!("rebalance_entry: start rebalance_entry"); + debug!( + event = EVENT_REBALANCE_ENTRY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + bucket = %bucket, + object = %entry.name, + pool_index, + state = "started", + "Starting rebalance entry" + ); // defer!(|| async { // warn!("rebalance_entry: defer give worker start"); @@ -2045,12 +2297,32 @@ impl ECStore { // }); if entry.is_dir() { - info!("rebalance_entry: entry is dir, skipping"); + debug!( + event = EVENT_REBALANCE_ENTRY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + bucket = %bucket, + object = %entry.name, + pool_index, + state = "skipped", + reason = "directory_entry", + "Skipped rebalance entry" + ); return Ok(RebalanceEntryOutcome::Completed); } if self.check_if_rebalance_done(pool_index).await { - info!("rebalance_entry: rebalance done, skipping pool {}", pool_index); + debug!( + event = EVENT_REBALANCE_ENTRY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + bucket = %bucket, + object = %entry.name, + state = "skipped", + reason = "pool_completed", + "Skipped rebalance entry" + ); return Ok(RebalanceEntryOutcome::Completed); } @@ -2077,16 +2349,33 @@ impl ECStore { .await { expired += 1; - info!("rebalance_entry {} Entry {} expired by lifecycle, skipping", &bucket, version.name); + debug!( + event = EVENT_REBALANCE_ENTRY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + bucket = %bucket, + object = %version.name, + state = "skipped", + reason = "expired_by_lifecycle", + "Skipped rebalance version" + ); continue; } let remaining_versions = fivs.versions.len() - expired; if should_skip_rebalance_delete_marker(version, remaining_versions, bucket_configs.replication_config.is_some()) { rebalanced += 1; - info!( - "rebalance_entry Entry {} is deleted and last version without replication, skipping", - version.name + debug!( + event = EVENT_REBALANCE_ENTRY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + bucket = %bucket, + object = %version.name, + state = "skipped", + reason = "last_delete_marker_without_replication", + "Skipped rebalance version" ); continue; } @@ -2112,7 +2401,17 @@ impl ECStore { if should_count_rebalance_version_complete(&result) { rebalanced += 1; } - info!("rebalance_entry {} Entry {} is already deleted, skipping", &bucket, version.name); + debug!( + event = EVENT_REBALANCE_ENTRY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + bucket = %bucket, + object = %version.name, + state = "skipped", + reason = "already_deleted", + "Skipped rebalance version" + ); continue; } @@ -2131,8 +2430,15 @@ impl ECStore { if should_defer_rebalance_entry_failure(&err) { let deferred_error = format!("{REBALANCE_DEFERRED_ENTRY_ERROR_PREFIX} {err}"); warn!( - "rebalance_entry {} deferring transient migration failure for {}/{:?}: {}", - &bucket, &version.name, &version.version_id, err + event = EVENT_REBALANCE_ENTRY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + bucket = %bucket, + object = %version.name, + state = "deferred", + error = %err, + "Deferred rebalance entry after transient migration failure" ); if let Err(stats_err) = self.update_rebalance_last_error(pool_index, deferred_error.clone()).await { error!( @@ -2192,7 +2498,16 @@ impl ECStore { entry.name.as_str(), ) .map_err(|err| with_rebalance_entry_context("cleanup_source", bucket.as_str(), entry.name.as_str(), err))?; - info!("rebalance_entry {} Entry {} deleted successfully", &bucket, &entry.name); + debug!( + event = EVENT_REBALANCE_ENTRY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + bucket = %bucket, + object = %entry.name, + state = "source_deleted", + "Deleted rebalance source entry" + ); } Ok(RebalanceEntryOutcome::Completed) @@ -2288,7 +2603,14 @@ impl ECStore { let task = tokio::spawn(async move { let _permit = permit; - info!("rebalance_entry: rebalance entry task start"); + debug!( + event = EVENT_REBALANCE_ENTRY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + set_index = set_idx, + state = "task_started", + "Started rebalance entry task" + ); let result = this.rebalance_entry(bucket, pool_index, entry, set, bucket_configs).await; if let Err(err) = &result { error!("rebalance_entry: rebalance entry failed: {err}"); @@ -2298,7 +2620,14 @@ impl ECStore { callback_rx.cancel(); } } - info!("rebalance_entry: rebalance entry task done"); + debug!( + event = EVENT_REBALANCE_ENTRY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + set_index = set_idx, + state = "task_completed", + "Completed rebalance entry task" + ); result }); @@ -2327,7 +2656,14 @@ impl ECStore { if let Err(err) = &result { error!("Rebalance worker {} error: {}", set_idx, err); } else { - info!("Rebalance worker {} done", set_idx); + debug!( + event = EVENT_REBALANCE_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + set_index = set_idx, + state = "worker_completed", + "Completed rebalance worker" + ); } result }); @@ -2355,7 +2691,15 @@ impl ECStore { return Ok(RebalanceBucketOutcome::Deferred { last_error }); } - info!("rebalance_bucket: rebalance_bucket done"); + debug!( + event = EVENT_REBALANCE_BUCKET, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + pool_index, + bucket = %bucket, + state = "completed", + "Finished rebalance bucket" + ); Ok(RebalanceBucketOutcome::Completed) } diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 9a43c869b..9df2f6e6f 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -18,6 +18,14 @@ use crate::global::{ GLOBAL_EventNotifier, GLOBAL_LOCAL_DISK_ID_MAP, GLOBAL_LOCAL_DISK_MAP, GLOBAL_LOCAL_DISK_SET_DRIVES, GLOBAL_TierConfigMgr, get_global_bucket_monitor, is_dist_erasure, is_first_cluster_node_local, }; +use tracing::{debug, error, info, warn}; + +const LOG_COMPONENT_ECSTORE: &str = "ecstore"; +const LOG_SUBSYSTEM_STORE_INIT: &str = "store_init"; +const EVENT_DECOMMISSION_RESUME_RETRY: &str = "decommission_resume_retry"; +const EVENT_DECOMMISSION_RESUME_FAILED: &str = "decommission_resume_failed"; +const EVENT_STORE_FORMAT_RETRY: &str = "store_format_retry"; +const EVENT_ECSTORE_INIT_STATUS: &str = "ecstore_init_status"; fn pool_first_endpoint_is_local(pool: &crate::endpoints::PoolEndpoints) -> bool { pool.endpoints.as_ref().first().is_some_and(|endpoint| endpoint.is_local) @@ -71,19 +79,27 @@ async fn resume_local_decommission_after_init(store: Arc, rx: Cancellat .await { error!( - "store init failed to resume decommission workers for pools {:?}: {}", - pool_indices, spawn_err + event = EVENT_DECOMMISSION_RESUME_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_STORE_INIT, + pool_indices = ?pool_indices, + error = %spawn_err, + reason = "spawn_workers_failed", + "Failed to resume decommission workers" ); } return; } Err(err) if should_retry_local_decommission_resume(&err, attempt) => { warn!( - "store init decommission resume missing config for pools {:?}, retry {}/{}: {}", - pool_indices, - attempt + 1, - LOCAL_DECOMMISSION_RESUME_MAX_CONFIG_RETRIES + 1, - err + event = EVENT_DECOMMISSION_RESUME_RETRY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_STORE_INIT, + pool_indices = ?pool_indices, + retry_count = attempt + 1, + retry_limit = LOCAL_DECOMMISSION_RESUME_MAX_CONFIG_RETRIES + 1, + error = %err, + "Retrying decommission resume after missing config" ); tokio::select! { _ = rx.cancelled() => return, @@ -91,7 +107,15 @@ async fn resume_local_decommission_after_init(store: Arc, rx: Cancellat } } Err(err) => { - error!("store init failed to resume decommission for pools {:?}: {}", pool_indices, err); + error!( + event = EVENT_DECOMMISSION_RESUME_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_STORE_INIT, + pool_indices = ?pool_indices, + error = %err, + reason = "resume_failed", + "Failed to resume decommission" + ); return; } } @@ -113,7 +137,13 @@ impl ECStore { let mut local_disks = Vec::new(); - info!("ECStore new address: {}", address.to_string()); + debug!( + event = EVENT_ECSTORE_INIT_STATUS, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_STORE_INIT, + address = %address, + "Initializing ECStore address" + ); let mut host = address.ip().to_string(); if host.is_empty() { host = GLOBAL_RUSTFS_HOST.read().await.to_string() @@ -122,7 +152,14 @@ impl ECStore { if port.is_empty() { port = GLOBAL_RUSTFS_PORT.read().await.to_string() } - info!("ECStore new host: {}, port: {}", host, port); + debug!( + event = EVENT_ECSTORE_INIT_STATUS, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_STORE_INIT, + host = %host, + port = %port, + "Initializing ECStore host" + ); init_local_peer(&endpoint_pools, &host, &port).await; // debug!("endpoint_pools: {:?}", endpoint_pools); @@ -179,10 +216,23 @@ impl ECStore { if interval < 16 { interval *= 2; } - info!("retrying get formats after {:?}", interval); + debug!( + event = EVENT_STORE_FORMAT_RETRY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_STORE_INIT, + retry_count = times, + retry_delay_secs = interval, + "Retrying storage format load" + ); select! { _ = tokio::signal::ctrl_c() => { - info!("got ctrl+c, exits"); + info!( + event = EVENT_STORE_FORMAT_RETRY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_STORE_INIT, + reason = "ctrl_c", + "Interrupted storage format retry loop" + ); exit(0); } _ = sleep(Duration::from_secs(interval)) => { diff --git a/crates/ecstore/src/store/peer.rs b/crates/ecstore/src/store/peer.rs index 13d64b838..fddde1a56 100644 --- a/crates/ecstore/src/store/peer.rs +++ b/crates/ecstore/src/store/peer.rs @@ -14,6 +14,12 @@ use super::*; use crate::global::GLOBAL_LOCAL_DISK_ID_MAP; +use tracing::{debug, error}; + +const LOG_COMPONENT_ECSTORE: &str = "ecstore"; +const LOG_SUBSYSTEM_DISK_STARTUP: &str = "disk_startup"; +const EVENT_LOCAL_DISK_ID_PREWARM_SKIPPED: &str = "local_disk_id_prewarm_skipped"; +const EVENT_LOCK_CLIENT_INITIALIZATION_FAILED: &str = "lock_client_initialization_failed"; async fn remember_local_disk_id(disk: &DiskStore) -> Option { let disk_id = disk.get_disk_id().await.ok().flatten()?; @@ -94,7 +100,14 @@ pub async fn all_local_disk() -> Vec { pub async fn prewarm_local_disk_id_map() { for disk in all_local_disk().await { if let Err(err) = disk.get_disk_id().await { - warn!("prewarm_local_disk_id_map: failed to load disk id for {}: {}", disk.endpoint(), err); + debug!( + event = EVENT_LOCAL_DISK_ID_PREWARM_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_STARTUP, + disk_endpoint = %disk.endpoint(), + error = %err, + "Skipped local disk id prewarm" + ); continue; } @@ -159,7 +172,14 @@ pub fn init_lock_clients(endpoint_pools: EndpointServerPools) { if !first_local_client_set { if let Err(e) = crate::global::set_global_lock_client(local_client.clone()) { // If already set, ignore the error (another thread may have set it) - warn!("set_global_lock_client error: {:?}", e); + debug!( + event = EVENT_LOCK_CLIENT_INITIALIZATION_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_STARTUP, + error = ?e, + reason = "global_lock_client_already_set", + "Skipped global lock client publication" + ); } else { first_local_client_set = true; } @@ -173,7 +193,13 @@ pub fn init_lock_clients(endpoint_pools: EndpointServerPools) { // Store the lock clients map globally if crate::global::set_global_lock_clients(clients).is_err() { - error!("init_lock_clients: error setting lock clients"); + error!( + event = EVENT_LOCK_CLIENT_INITIALIZATION_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_DISK_STARTUP, + reason = "set_global_lock_clients_failed", + "Failed to initialize lock clients" + ); } } diff --git a/crates/ecstore/src/tier/tier.rs b/crates/ecstore/src/tier/tier.rs index e5c2b2674..17e309af5 100644 --- a/crates/ecstore/src/tier/tier.rs +++ b/crates/ecstore/src/tier/tier.rs @@ -1230,10 +1230,20 @@ where return; } } - Err(err) => warn!("legacy tier config is incompatible, skip local migration: {}", err), + Err(err) => debug!( + bucket = RUSTFS_META_BUCKET, + path = %legacy_path, + error = %err, + "Skipping incompatible legacy tier config migration" + ), }, Ok(None) => {} - Err(err) => warn!("read legacy local tier config failed: {}", err), + Err(err) => debug!( + bucket = RUSTFS_META_BUCKET, + path = %legacy_path, + error = %err, + "Skipping legacy tier config migration after read failure" + ), } match read_tier_config_from_bucket(api.clone(), MIGRATING_META_BUCKET, &target_path, &opts).await { @@ -1243,10 +1253,20 @@ where info!("Migrated compatible tier config from migrating metadata bucket"); } } - Err(err) => warn!("migrating tier config is incompatible, skip migration: {}", err), + Err(err) => debug!( + bucket = MIGRATING_META_BUCKET, + path = %target_path, + error = %err, + "Skipping incompatible migrating tier config" + ), }, Ok(None) => {} - Err(err) => warn!("read migrating tier config failed: {}", err), + Err(err) => debug!( + bucket = MIGRATING_META_BUCKET, + path = %target_path, + error = %err, + "Skipping migrating tier config after read failure" + ), } } diff --git a/crates/notify/src/config_manager.rs b/crates/notify/src/config_manager.rs index db7e54259..9f0e2999d 100644 --- a/crates/notify/src/config_manager.rs +++ b/crates/notify/src/config_manager.rs @@ -23,7 +23,12 @@ use rustfs_config::server_config::{Config, KVS}; use rustfs_targets::{Target, arn::TargetID}; use std::sync::Arc; use tokio::sync::RwLock; -use tracing::{debug, info, warn}; +use tracing::{debug, info}; + +const LOG_COMPONENT_NOTIFY: &str = "notify"; +const LOG_SUBSYSTEM_CONFIG: &str = "config"; +const EVENT_NOTIFY_RUNTIME_LIFECYCLE: &str = "notify_runtime_lifecycle"; +const EVENT_NOTIFY_CONFIG_UPDATE: &str = "notify_config_update"; pub(crate) fn notify_configuration_hint() -> String { let webhook_enable_primary = format!("{}_PRIMARY", rustfs_config::notify::ENV_NOTIFY_WEBHOOK_ENABLE); @@ -81,7 +86,13 @@ impl NotifyConfigManager { } pub async fn init(&self) -> Result<(), NotificationError> { - info!("Initialize notification system..."); + info!( + event = EVENT_NOTIFY_RUNTIME_LIFECYCLE, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_CONFIG, + state = "initializing", + "Initializing notification system" + ); let config = { let guard = self.config.read().await; @@ -94,19 +105,48 @@ impl NotifyConfigManager { let targets: Vec + Send + Sync>> = self.registry.create_targets_from_config(&config).await?; - info!("{} notification targets were created", targets.len()); + info!( + event = EVENT_NOTIFY_RUNTIME_LIFECYCLE, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_CONFIG, + state = "targets_created", + target_count = targets.len(), + "Created notification targets" + ); if targets.is_empty() { - warn!("{}", notify_configuration_hint()); + debug!( + event = EVENT_NOTIFY_RUNTIME_LIFECYCLE, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_CONFIG, + state = "idle", + reason = "no_targets_configured", + hint = %notify_configuration_hint(), + "Notification runtime has no configured targets" + ); } let activation = self.runtime_facade.activate_targets_with_replay(targets).await; self.runtime_facade.replace_targets(activation).await?; - info!("Notification system initialized"); + info!( + event = EVENT_NOTIFY_RUNTIME_LIFECYCLE, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_CONFIG, + state = "initialized", + "Initialized notification system" + ); Ok(()) } pub async fn remove_target(&self, target_id: &TargetID, target_type: &str) -> Result<(), NotificationError> { - info!("Attempting to remove target: {}", target_id); + debug!( + event = EVENT_NOTIFY_CONFIG_UPDATE, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_CONFIG, + action = "remove_target", + target_id = %target_id, + target_type, + "Attempting to remove notification target" + ); let ttype = target_type.to_lowercase(); let tname = target_id.id.to_lowercase(); @@ -115,7 +155,15 @@ impl NotifyConfigManager { let mut changed = false; if let Some(targets_of_type) = config.0.get_mut(&ttype) { if targets_of_type.remove(&tname).is_some() { - info!("Removed target {} from configuration", target_id); + info!( + event = EVENT_NOTIFY_CONFIG_UPDATE, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_CONFIG, + action = "remove_target", + target_id = %target_id, + result = "removed", + "Removed notification target from configuration" + ); changed = true; } if targets_of_type.is_empty() { @@ -123,7 +171,15 @@ impl NotifyConfigManager { } } if !changed { - warn!("Target {} not found in configuration", target_id); + debug!( + event = EVENT_NOTIFY_CONFIG_UPDATE, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_CONFIG, + action = "remove_target", + target_id = %target_id, + result = "not_found", + "Notification target not found in configuration" + ); } changed }) @@ -131,7 +187,15 @@ impl NotifyConfigManager { } pub async fn set_target_config(&self, target_type: &str, target_name: &str, kvs: KVS) -> Result<(), NotificationError> { - info!("Setting config for target {} of type {}", target_name, target_type); + debug!( + event = EVENT_NOTIFY_CONFIG_UPDATE, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_CONFIG, + action = "set_target_config", + target_type, + target_name, + "Setting notification target configuration" + ); let ttype = target_type.to_lowercase(); let tname = target_name.to_lowercase(); self.update_config_and_reload(|config| { @@ -142,7 +206,15 @@ impl NotifyConfigManager { } pub async fn remove_target_config(&self, target_type: &str, target_name: &str) -> Result<(), NotificationError> { - info!("Removing config for target {} of type {}", target_name, target_type); + debug!( + event = EVENT_NOTIFY_CONFIG_UPDATE, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_CONFIG, + action = "remove_target_config", + target_type, + target_name, + "Removing notification target configuration" + ); let ttype = target_type.to_lowercase(); let tname = target_name.to_lowercase(); @@ -166,7 +238,16 @@ impl NotifyConfigManager { } } if !changed { - info!("Target {} of type {} not found, no changes made.", target_name, target_type); + debug!( + event = EVENT_NOTIFY_CONFIG_UPDATE, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_CONFIG, + action = "remove_target_config", + target_type = %target_type, + target_name = %target_name, + result = "not_found", + "Notification target configuration not found" + ); } debug!( subsystem_count = config.0.len(), @@ -178,7 +259,13 @@ impl NotifyConfigManager { } pub async fn reload_config(&self, new_config: Config) -> Result<(), NotificationError> { - info!("Reload notification configuration starts"); + info!( + event = EVENT_NOTIFY_RUNTIME_LIFECYCLE, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_CONFIG, + state = "reloading", + "Reloading notification configuration" + ); self.update_config(new_config.clone()).await; @@ -188,14 +275,35 @@ impl NotifyConfigManager { .await .map_err(NotificationError::Target)?; - info!("{} notification targets were created from the new configuration", targets.len()); + info!( + event = EVENT_NOTIFY_RUNTIME_LIFECYCLE, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_CONFIG, + state = "targets_created", + target_count = targets.len(), + "Created notification targets from reloaded configuration" + ); if targets.is_empty() { - warn!("{}", notify_configuration_hint()); + debug!( + event = EVENT_NOTIFY_RUNTIME_LIFECYCLE, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_CONFIG, + state = "idle", + reason = "no_targets_configured", + hint = %notify_configuration_hint(), + "Notification runtime has no configured targets after reload" + ); } let activation = self.runtime_facade.activate_targets_with_replay(targets).await; self.runtime_facade.replace_targets(activation).await?; - info!("Configuration reloaded end"); + info!( + event = EVENT_NOTIFY_RUNTIME_LIFECYCLE, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_CONFIG, + state = "reloaded", + "Reloaded notification configuration" + ); Ok(()) } @@ -219,7 +327,14 @@ impl NotifyConfigManager { .map_err(|e| NotificationError::ReadConfig(e.to_string()))?; if !modifier(&mut new_config) { - info!("Configuration not changed, skipping save and reload."); + debug!( + event = EVENT_NOTIFY_CONFIG_UPDATE, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_CONFIG, + action = "reload_if_changed", + result = "unchanged", + "Notification configuration unchanged; skipping reload" + ); return Ok(()); } @@ -227,7 +342,14 @@ impl NotifyConfigManager { .await .map_err(|e| NotificationError::SaveConfig(e.to_string()))?; - info!("Configuration updated. Reloading system..."); + info!( + event = EVENT_NOTIFY_CONFIG_UPDATE, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_CONFIG, + action = "reload_if_changed", + result = "updated", + "Notification configuration updated; reloading runtime" + ); self.reload_config(new_config).await } } diff --git a/crates/notify/src/notifier.rs b/crates/notify/src/notifier.rs index edba03854..c6c62f34e 100644 --- a/crates/notify/src/notifier.rs +++ b/crates/notify/src/notifier.rs @@ -21,6 +21,14 @@ use std::sync::Arc; use tokio::sync::{RwLock, Semaphore}; use tracing::{debug, error, info, instrument, warn}; +const LOG_COMPONENT_NOTIFY: &str = "notify"; +const LOG_SUBSYSTEM_DISPATCH: &str = "dispatch"; +const EVENT_NOTIFY_DISPATCH_SKIPPED: &str = "notify_dispatch_skipped"; +const EVENT_NOTIFY_DISPATCH_FAILED: &str = "notify_dispatch_failed"; +const EVENT_NOTIFY_DISPATCH_STARTED: &str = "notify_dispatch_started"; +const EVENT_NOTIFY_DISPATCH_COMPLETED: &str = "notify_dispatch_completed"; +const EVENT_NOTIFY_RUNTIME_LIFECYCLE: &str = "notify_runtime_lifecycle"; + pub type SharedNotifyTargetList = Arc>; /// Manages event notification to targets based on rules @@ -83,7 +91,13 @@ impl EventNotifier { // The logic for sending cancel signals via stream_cancel_senders would be removed. // TargetList::clear_targets_only already handles calling target.close(). target_list_guard.clear_targets_only().await; // Modified clear to not re-cancel - info!("Removed all targets and their streams"); + info!( + event = EVENT_NOTIFY_RUNTIME_LIFECYCLE, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_DISPATCH, + state = "targets_cleared", + "Removed all notify targets" + ); } /// Sends an event to the appropriate targets based on the bucket rules @@ -98,7 +112,15 @@ impl EventNotifier { let target_ids = self.rule_engine.match_targets(bucket_name, event_name, object_key).await; if target_ids.is_empty() { - debug!("No matching targets for event in bucket: {}", bucket_name); + debug!( + event = EVENT_NOTIFY_DISPATCH_SKIPPED, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_DISPATCH, + bucket = %bucket_name, + object = %object_key, + reason = "no_matching_targets", + "Skipped notify dispatch" + ); self.metrics.increment_skipped(); return; } @@ -107,7 +129,15 @@ impl EventNotifier { // Use scope to limit the borrow scope of target_list let target_list_guard = self.target_list.read().await; - info!("Sending event to targets: {:?}", target_ids); + debug!( + event = EVENT_NOTIFY_DISPATCH_STARTED, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_DISPATCH, + bucket = %bucket_name, + object = %object_key, + target_count = target_ids_len, + "Dispatching notify event" + ); for target_id in target_ids { // `get` now returns Option> if let Some(target_arc) = target_list_guard.get(&target_id) { @@ -115,7 +145,14 @@ impl EventNotifier { // target_arc is already Arc, clone it for the async task let target_for_task = target_arc.clone(); if !target_for_task.is_enabled() { - debug!("Skipping disabled target: {}", target_for_task.name()); + debug!( + event = EVENT_NOTIFY_DISPATCH_SKIPPED, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_DISPATCH, + target_id = %target_for_task.id(), + reason = "target_disabled", + "Skipped notify dispatch target" + ); continue; } let limiter = self.send_limiter.clone(); @@ -123,7 +160,14 @@ impl EventNotifier { let event_clone = event.clone(); let is_deferred = target_for_task.store().is_some(); let target_name_for_task = target_for_task.name(); // Get the name before generating the task - debug!("Preparing to send event to target: {}", target_name_for_task); + debug!( + event = EVENT_NOTIFY_DISPATCH_STARTED, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_DISPATCH, + target_id = %target_for_task.id(), + deferred = is_deferred, + "Prepared notify target dispatch" + ); // Use cloned data in closures to avoid borrowing conflicts // Create an EntityTarget from the event let entity_target: Arc> = Arc::new(EntityTarget { @@ -137,26 +181,56 @@ impl EventNotifier { let _permit = match limiter.acquire_owned().await { Ok(p) => p, Err(e) => { - error!("Failed to acquire send permit for target {}: {}", target_name_for_task, e); + error!( + event = EVENT_NOTIFY_DISPATCH_FAILED, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_DISPATCH, + target_id = %target_name_for_task, + error = %e, + reason = "permit_acquire_failed", + "Failed to acquire notify send permit" + ); metrics.increment_failed(); return; } }; if let Err(e) = target_for_task.save(entity_target.clone()).await { metrics.increment_failed(); - error!("Failed to send event to target {}: {}", target_name_for_task, e); + error!( + event = EVENT_NOTIFY_DISPATCH_FAILED, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_DISPATCH, + target_id = %target_name_for_task, + error = %e, + reason = "send_failed", + "Failed to dispatch notify event" + ); } else { if is_deferred { metrics.decrement_processing(); } else { metrics.increment_processed(); } - debug!("Successfully saved event to target {}", target_name_for_task); + debug!( + event = EVENT_NOTIFY_DISPATCH_COMPLETED, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_DISPATCH, + target_id = %target_name_for_task, + deferred = is_deferred, + "Completed notify target dispatch" + ); } }); handles.push(handle); } else { - warn!("Target ID {:?} found in rules but not in target list.", target_id); + warn!( + event = EVENT_NOTIFY_DISPATCH_FAILED, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_DISPATCH, + target_id = %target_id, + reason = "target_missing_from_runtime", + "Matched notify target is missing from runtime" + ); self.metrics.increment_skipped(); } } @@ -166,10 +240,24 @@ impl EventNotifier { // Wait for all tasks to be completed for handle in handles { if let Err(e) = handle.await { - error!("Task for sending/saving event failed: {}", e); + error!( + event = EVENT_NOTIFY_DISPATCH_FAILED, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_DISPATCH, + error = %e, + reason = "join_failed", + "Notify dispatch task failed" + ); } } - info!("Event processing initiated for {} targets for bucket: {}", target_ids_len, bucket_name); + debug!( + event = EVENT_NOTIFY_DISPATCH_COMPLETED, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_DISPATCH, + bucket = %bucket_name, + target_count = target_ids_len, + "Finished notify dispatch fan-out" + ); } /// Initializes the targets for buckets from shared target handles. @@ -179,14 +267,24 @@ impl EventNotifier { target_list_guard.clear(); for target in targets_to_init { - debug!("init bucket target: {}", target.name()); + debug!( + event = EVENT_NOTIFY_RUNTIME_LIFECYCLE, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_DISPATCH, + state = "target_init", + target_id = %target.id(), + "Initializing notify runtime target" + ); target_list_guard.add(target)?; } info!( - "Initialized {} shared targets, list size: {}", - target_list_guard.len(), - target_list_guard.len() + event = EVENT_NOTIFY_RUNTIME_LIFECYCLE, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_DISPATCH, + state = "targets_initialized", + target_count = target_list_guard.len(), + "Initialized notify runtime targets" ); Ok(()) } @@ -223,7 +321,14 @@ impl TargetList { let id = target.id(); if self.runtime.get_by_target_id(&id).is_some() { // Potentially update or log a warning/error if replacing an existing target. - warn!("Target with ID {} already exists in TargetList. It will be overwritten.", id); + warn!( + event = EVENT_NOTIFY_RUNTIME_LIFECYCLE, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_DISPATCH, + state = "target_overwrite", + target_id = %id, + "Overwriting existing notify runtime target" + ); } self.runtime.add_arc(target); Ok(()) diff --git a/crates/notify/src/runtime_facade.rs b/crates/notify/src/runtime_facade.rs index 1c6bd655d..5cad6209c 100644 --- a/crates/notify/src/runtime_facade.rs +++ b/crates/notify/src/runtime_facade.rs @@ -19,7 +19,11 @@ use rustfs_targets::{ use std::sync::Arc; use std::time::Duration; use tokio::sync::{RwLock, Semaphore}; -use tracing::info; +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"; #[derive(Clone)] pub struct NotifyRuntimeFacade { @@ -55,9 +59,24 @@ impl NotifyRuntimeFacade { }), Arc::new(|target_id, has_replay| { if has_replay { - info!("Event stream processing for target {} is started successfully", target_id); + info!( + event = EVENT_NOTIFY_RUNTIME_LIFECYCLE, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_RUNTIME, + target_id = %target_id, + state = "replay_started", + "Started notify replay worker" + ); } else { - info!("Target {} has no replay worker to start", target_id); + debug!( + event = EVENT_NOTIFY_RUNTIME_LIFECYCLE, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_RUNTIME, + target_id = %target_id, + state = "replay_skipped", + reason = "no_store_configured", + "Skipped notify replay worker startup" + ); } }), Some(concurrency_limiter), @@ -97,10 +116,23 @@ impl NotifyRuntimeFacade { } pub async fn shutdown(&self) { - info!("Turn off the notification system"); + info!( + event = EVENT_NOTIFY_RUNTIME_LIFECYCLE, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_RUNTIME, + state = "stopping", + "Stopping notification runtime" + ); let active_targets = self.replay_workers.read().await.len(); - info!("Stops {} active event stream processing tasks", active_targets); + info!( + event = EVENT_NOTIFY_RUNTIME_LIFECYCLE, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_RUNTIME, + state = "replay_stopping", + active_targets, + "Stopping notify replay workers" + ); { // Lock order: replay_workers -> target_list (matches notify AGENTS.md). @@ -116,7 +148,13 @@ impl NotifyRuntimeFacade { } tokio::time::sleep(Duration::from_millis(500)).await; - info!("Notify the system to be shut down completed"); + info!( + event = EVENT_NOTIFY_RUNTIME_LIFECYCLE, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_RUNTIME, + state = "stopped", + "Stopped notification runtime" + ); } } diff --git a/crates/obs/src/logging.rs b/crates/obs/src/logging.rs index 9b61b77f6..2900a512c 100644 --- a/crates/obs/src/logging.rs +++ b/crates/obs/src/logging.rs @@ -179,4 +179,43 @@ mod tests { &["Protocol storage client ListBuckets request: access_key={}"], ); } + + #[test] + fn startup_runtime_logging_does_not_dump_full_config_debug_output() { + let path = workspace_root().join("rustfs/src/main.rs"); + let source = fs::read_to_string(&path).unwrap_or_else(|err| panic!("failed to read {}: {}", path.display(), err)); + assert!( + !source.contains("debug!(\"config: {:?}\", &config)"), + "found forbidden full config debug output in {}", + path.display() + ); + } + + #[test] + fn audit_notify_runtime_logging_does_not_use_previous_sentence_first_noise_patterns() { + assert_no_unmasked_access_key_logging( + "crates/audit/src/pipeline.rs", + &[ + "No audit targets configured for dispatch", + "No audit targets configured for batch dispatch", + "Successfully sent audit entry, target: {}, key: {}", + "Target {} not connected, retrying...", + "Timeout sending to target {}, retrying...", + ], + ); + assert_no_unmasked_access_key_logging( + "crates/notify/src/runtime_facade.rs", + &[ + "Event stream processing for target {} is started successfully", + "Target {} has no replay worker to start", + ], + ); + assert_no_unmasked_access_key_logging( + "crates/notify/src/notifier.rs", + &[ + "Sending event to targets: {:?}", + "Event processing initiated for {} targets for bucket: {}", + ], + ); + } } diff --git a/scripts/check_logging_guardrails.sh b/scripts/check_logging_guardrails.sh new file mode 100755 index 000000000..96795c965 --- /dev/null +++ b/scripts/check_logging_guardrails.sh @@ -0,0 +1,48 @@ +#!/usr/bin/env bash +set -euo pipefail + +repo_root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +cd "$repo_root" + +checked_files=( + "rustfs/src/main.rs" + "rustfs/src/startup_iam.rs" + "rustfs/src/auth.rs" + "rustfs/src/protocols/client.rs" + "crates/audit/src/pipeline.rs" + "crates/audit/src/system.rs" + "crates/audit/src/global.rs" + "crates/notify/src/config_manager.rs" + "crates/notify/src/runtime_facade.rs" + "crates/notify/src/notifier.rs" + "crates/ecstore/src/store/peer.rs" + "crates/ecstore/src/store/init.rs" + "crates/ecstore/src/tier/tier.rs" +) + +forbidden_patterns=( + 'access_key={}' + 'secret_key={}' + 'Authorization={}' + 'token={}' + 'debug!("config: {:?}"' + 'warn!("No audit targets configured for dispatch"' + 'warn!("No audit targets configured for batch dispatch"' + 'info!("Event stream processing for target {} is started successfully"' + 'info!("Target {} has no replay worker to start"' + 'info!("Sending event to targets: {:?}"' + 'info!("Event processing initiated for {} targets for bucket: {}"' + 'warn!("{}", notify_configuration_hint())' + 'warn!("prewarm_local_disk_id_map: failed to load disk id for {}: {}"' + 'info!("retrying get formats after {:?}"' +) + +for pattern in "${forbidden_patterns[@]}"; do + if rg -n -F -- "$pattern" "${checked_files[@]}" >/dev/null; then + echo "❌ logging guardrail violation: found forbidden pattern '$pattern'" >&2 + rg -n -F -- "$pattern" "${checked_files[@]}" >&2 + exit 1 + fi +done + +echo "✅ Logging guardrails check passed"