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

chore(obs): standardize runtime logging batch one (#3438)

Browse Source
This commit is contained in:
cxymds
2026-06-14 18:55:14 +08:00
committed by GitHub
parent bed4c7288b
commit 8f6b1d47b5
5 changed files with 199 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/obs/src/logging.rs
+82
View File
@@ -83,6 +83,19 @@ mod tests {
}
}
fn assert_source_contains(rel_path: &str, required_patterns: &[&str]) {
let path = workspace_root().join(rel_path);
let source = fs::read_to_string(&path).unwrap_or_else(|err| panic!("failed to read {}: {}", path.display(), err));
for pattern in required_patterns {
assert!(
source.contains(pattern),
"missing required logging governance pattern `{}` in {}",
pattern,
path.display()
);
}
}
#[test]
fn logging_redaction_rules_are_valid() {
assert!(validate_logging_redaction_rules().is_ok());
@@ -156,6 +169,75 @@ mod tests {
);
}
#[test]
fn startup_fatal_stderr_uses_single_formatter_for_pre_observability_failures() {
assert_source_contains(
"rustfs/src/main.rs",
&[
"fn format_fatal_stderr_message(context: &str, error: impl std::fmt::Display) -> String",
"fn emit_fatal_stderr(context: &str, error: impl std::fmt::Display)",
"emit_fatal_stderr(\"Server runtime failed\", e)",
"emit_fatal_stderr(\"Command parse failed\", e)",
"emit_fatal_stderr(\"Observability initialization failed\", &e)",
],
);
}
#[test]
fn observability_runtime_logging_uses_tracing_for_fallback_and_guard_shutdown() {
assert_no_unmasked_access_key_logging(
"crates/obs/src/telemetry/local.rs",
&[
"[WARN] Failed to initialize file observability logging",
"Falling back to stdout logging.",
],
);
assert_source_contains(
"crates/obs/src/telemetry/local.rs",
&[
"warn!(",
"state = \"fallback_to_stdout\"",
"failed_sink = \"file\"",
"sink = \"stdout\"",
],
);
assert_no_unmasked_access_key_logging(
"crates/obs/src/telemetry/guard.rs",
&[
"eprintln!(\"Tracer shutdown error: {err:?}\")",
"eprintln!(\"Meter shutdown error: {err:?}\")",
"eprintln!(\"Logger shutdown error: {err:?}\")",
"eprintln!(\"Log cleanup task stopped\")",
"eprintln!(\"Tracing guard dropped, flushing logs.\")",
"eprintln!(\"Stdout guard dropped, flushing logs.\")",
],
);
assert_source_contains(
"crates/obs/src/telemetry/guard.rs",
&[
"EVENT_OBS_GUARD_SHUTDOWN",
"resource = \"tracer_provider\"",
"resource = \"meter_provider\"",
"resource = \"logger_provider\"",
"resource = \"log_cleaner\"",
"resource = \"tracing_guard\"",
"resource = \"stdout_guard\"",
],
);
}
#[test]
fn low_level_logging_sink_stderr_exceptions_remain_explicit() {
assert_source_contains(
"crates/obs/src/telemetry/rolling.rs",
&[
"Failed to flush log file before rotation",
"RollingAppender: Failed to rotate log file after",
"RollingAppender: failed to rotate log file",
],
);
}
#[test]
fn audit_notify_runtime_logging_does_not_use_previous_sentence_first_noise_patterns() {
assert_no_unmasked_access_key_logging(
crates/obs/src/telemetry/guard.rs
+76 -10
View File
@@ -27,6 +27,7 @@
//! 7. Stdout worker guard — flushes buffered log lines written to stdout.
use opentelemetry_sdk::{logs::SdkLoggerProvider, metrics::SdkMeterProvider, trace::SdkTracerProvider};
use tracing::{debug, error};
#[cfg(all(
feature = "pyroscope",
@@ -43,6 +44,10 @@ pub(crate) type MemoryProfilingAgent = pyroscope::PyroscopeAgent<pyroscope::pyro
#[cfg(not(all(feature = "pyroscope", target_os = "linux", target_env = "gnu", target_arch = "x86_64")))]
pub(crate) type MemoryProfilingAgent = ();
const LOG_COMPONENT_OBS: &str = "obs";
const LOG_SUBSYSTEM_GUARD: &str = "guard";
const EVENT_OBS_GUARD_SHUTDOWN: &str = "obs_guard_shutdown";
/// RAII guard that owns all active OpenTelemetry providers and the
/// `tracing_appender` worker guard.
///
@@ -85,25 +90,49 @@ impl std::fmt::Debug for OtelGuard {
impl Drop for OtelGuard {
/// Shut down all telemetry providers in order.
///
/// Errors during shutdown are printed to `stderr` so they are visible even
/// after the tracing subscriber has been torn down.
/// Errors are emitted before tracing resources are dropped so shutdown
/// diagnostics remain structured and low-noise.
fn drop(&mut self) {
if let Some(provider) = self.tracer_provider.take()
&& let Err(err) = provider.shutdown()
{
eprintln!("Tracer shutdown error: {err:?}");
error!(
event = EVENT_OBS_GUARD_SHUTDOWN,
component = LOG_COMPONENT_OBS,
subsystem = LOG_SUBSYSTEM_GUARD,
resource = "tracer_provider",
result = "shutdown_failed",
error = ?err,
"observability guard shutdown failed"
);
}
if let Some(provider) = self.meter_provider.take()
&& let Err(err) = provider.shutdown()
{
eprintln!("Meter shutdown error: {err:?}");
error!(
event = EVENT_OBS_GUARD_SHUTDOWN,
component = LOG_COMPONENT_OBS,
subsystem = LOG_SUBSYSTEM_GUARD,
resource = "meter_provider",
result = "shutdown_failed",
error = ?err,
"observability guard shutdown failed"
);
}
if let Some(provider) = self.logger_provider.take()
&& let Err(err) = provider.shutdown()
{
eprintln!("Logger shutdown error: {err:?}");
error!(
event = EVENT_OBS_GUARD_SHUTDOWN,
component = LOG_COMPONENT_OBS,
subsystem = LOG_SUBSYSTEM_GUARD,
resource = "logger_provider",
result = "shutdown_failed",
error = ?err,
"observability guard shutdown failed"
);
}
#[cfg(all(
@@ -112,7 +141,15 @@ impl Drop for OtelGuard {
))]
if let Some(agent) = self.profiling_agent.take() {
match agent.stop() {
Err(err) => eprintln!("Profiling agent stop error: {err:?}"),
Err(err) => error!(
event = EVENT_OBS_GUARD_SHUTDOWN,
component = LOG_COMPONENT_OBS,
subsystem = LOG_SUBSYSTEM_GUARD,
resource = "profiling_agent",
result = "shutdown_failed",
error = ?err,
"observability guard shutdown failed"
),
Ok(stopped) => {
stopped.shutdown();
}
@@ -122,7 +159,15 @@ impl Drop for OtelGuard {
#[cfg(all(feature = "pyroscope", target_os = "linux", target_env = "gnu", target_arch = "x86_64"))]
if let Some(agent) = self.memory_profiling_agent.take() {
match agent.stop() {
Err(err) => eprintln!("Memory profiling agent stop error: {err:?}"),
Err(err) => error!(
event = EVENT_OBS_GUARD_SHUTDOWN,
component = LOG_COMPONENT_OBS,
subsystem = LOG_SUBSYSTEM_GUARD,
resource = "memory_profiling_agent",
result = "shutdown_failed",
error = ?err,
"observability guard shutdown failed"
),
Ok(stopped) => {
stopped.shutdown();
}
@@ -130,18 +175,39 @@ impl Drop for OtelGuard {
}
if let Some(handle) = self.cleanup_handle.take() {
debug!(
event = EVENT_OBS_GUARD_SHUTDOWN,
component = LOG_COMPONENT_OBS,
subsystem = LOG_SUBSYSTEM_GUARD,
resource = "log_cleaner",
result = "abort_requested",
"observability guard resource shutdown requested"
);
handle.abort();
eprintln!("Log cleanup task stopped");
}
if let Some(guard) = self.tracing_guard.take() {
debug!(
event = EVENT_OBS_GUARD_SHUTDOWN,
component = LOG_COMPONENT_OBS,
subsystem = LOG_SUBSYSTEM_GUARD,
resource = "tracing_guard",
result = "flush_requested",
"observability guard resource flush requested"
);
drop(guard);
eprintln!("Tracing guard dropped, flushing logs.");
}
if let Some(guard) = self.stdout_guard.take() {
debug!(
event = EVENT_OBS_GUARD_SHUTDOWN,
component = LOG_COMPONENT_OBS,
subsystem = LOG_SUBSYSTEM_GUARD,
resource = "stdout_guard",
result = "flush_requested",
"observability guard resource flush requested"
);
drop(guard);
eprintln!("Stdout guard dropped, flushing logs.");
}
}
}
crates/obs/src/telemetry/local.rs
+14 -5
View File
@@ -51,7 +51,7 @@ use rustfs_config::{APP_NAME, DEFAULT_LOG_KEEP_FILES, DEFAULT_LOG_ROTATION_TIME,
use std::sync::Arc;
use std::{fs, io::IsTerminal, time::Duration};
use tracing::Subscriber;
use tracing::info;
use tracing::{info, warn};
use tracing_error::ErrorLayer;
use tracing_subscriber::{
fmt::{format::FmtSpan, time::LocalTime},
@@ -126,8 +126,9 @@ pub(super) fn init_local_logging(
match init_file_logging_internal(config, log_directory, logger_level, is_production) {
Ok(guard) => Ok(guard),
Err(error) if should_fallback_to_stdout(&error) => {
let guard = init_stdout_only(config, logger_level, is_production);
emit_file_logging_fallback_warning(log_directory, &error);
Ok(init_stdout_only(config, logger_level, is_production))
Ok(guard)
}
Err(error) => Err(error),
}
@@ -361,9 +362,17 @@ pub(super) fn should_fallback_to_stdout(error: &TelemetryError) -> bool {
}
pub(super) fn emit_file_logging_fallback_warning(log_directory: &str, error: &TelemetryError) {
eprintln!(
"[WARN] Failed to initialize file observability logging at '{}': {}. Falling back to stdout logging.",
log_directory, error
warn!(
event = EVENT_LOCAL_LOGGING_STATE,
component = LOG_COMPONENT_OBS,
subsystem = LOG_SUBSYSTEM_LOCAL_LOGGING,
state = "fallback_to_stdout",
backend = "local",
failed_sink = "file",
sink = "stdout",
log_directory,
error = %error,
"local logging state changed"
);
}
rustfs/src/app/context/compat.rs
+6
View File
@@ -122,6 +122,7 @@ mod tests {
use crate::config::{RustFSBufferConfig, WorkloadProfile};
use rustfs_ecstore::disk::endpoint::Endpoint;
use rustfs_ecstore::endpoints::{Endpoints, PoolEndpoints};
use rustfs_ecstore::new_object_layer_fn;
use rustfs_ecstore::store::init_local_disks;
use rustfs_iam::{store::object::ObjectStore, sys::IamSys};
use std::path::PathBuf;
@@ -211,6 +212,11 @@ mod tests {
}
async fn test_store() -> (TempDir, Arc<ECStore>, EndpointServerPools) {
if let Some(store) = new_object_layer_fn() {
let endpoints = EndpointServerPools(store.pools.iter().map(|pool| pool.endpoints.clone()).collect());
return (tempfile::tempdir().expect("compat test temp dir"), store, endpoints);
}
let temp_dir = tempfile::tempdir().expect("test temp dir");
let disk_paths = (0..4)
.map(|index| temp_dir.path().join(format!("disk{index}")))
rustfs/src/main.rs
+21 -5
View File
@@ -119,13 +119,21 @@ static GLOBAL: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc;
#[global_allocator]
static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
fn format_fatal_stderr_message(context: &str, error: impl std::fmt::Display) -> String {
format!("[FATAL] {context}: {error}")
}
fn emit_fatal_stderr(context: &str, error: impl std::fmt::Display) {
eprintln!("{}", format_fatal_stderr_message(context, error));
}
fn main() {
// Build Tokio runtime with optional dial9 telemetry support
let runtime = rustfs::server::build_tokio_runtime().expect("Failed to build Tokio runtime");
let result = runtime.block_on(async_main());
if let Err(ref e) = result {
// Use eprintln as tracing may not be initialized at this point
eprintln!("[FATAL] Server encountered an error and is shutting down: {e}");
// Tracing may not be initialized when startup fails this early.
emit_fatal_stderr("Server runtime failed", e);
std::process::exit(1);
}
}
@@ -158,7 +166,7 @@ async fn async_main() -> Result<()> {
let command_result = match rustfs::config::Opt::parse_command(args) {
Ok(result) => result,
Err(e) => {
eprintln!("Command parse failed, error: {}", e);
emit_fatal_stderr("Command parse failed", e);
std::process::exit(1);
}
};
@@ -183,8 +191,8 @@ async fn async_main() -> Result<()> {
let guard = match init_obs(Some(config.clone().obs_endpoint)).await {
Ok(g) => g,
Err(e) => {
// Use eprintln as tracing is not yet initialized
eprintln!("[FATAL] Failed to initialize observability: {e}");
// Structured logging is unavailable until observability initializes.
emit_fatal_stderr("Observability initialization failed", &e);
return Err(Error::other(e));
}
};
@@ -1284,6 +1292,14 @@ mod tests {
);
}
#[test]
fn fatal_stderr_message_uses_consistent_prefix_and_context() {
assert_eq!(
format_fatal_stderr_message("Observability initialization failed", "collector unavailable"),
"[FATAL] Observability initialization failed: collector unavailable"
);
}
#[test]
fn is_using_default_credentials_returns_true_for_default_keys() {
let mut config = rustfs::config::Config::new("127.0.0.1:9000", Vec::new());