gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-24 13:16:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(logging): reduce runtime noise (#3363)

Browse Source
This commit is contained in:
houseme
2026-06-11 19:49:01 +08:00
committed by GitHub
parent a0b6636b61
commit 0a987d870b
19 changed files with 1974 additions and 305 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/audit/src/global.rs
+35 -10
View File
@@ -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<Arc<AuditSystem>> = 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<AuditEntry>) -> 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"
);
}
}
crates/audit/src/pipeline.rs
+169 -26
View File
@@ -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,
crates/audit/src/system.rs
+88 -16
View File
@@ -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<usize>) {
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};