From 0d68851f7a426e10570266eac0b1a36483cc67c3 Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 13 Jun 2026 08:39:47 +0800 Subject: [PATCH] refactor(logging): standardize rustfs runtime events (#3396) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(rpc): adjust log levels and reduce noise in storage RPC layer Issue #682: pool rebalance normal flow logs were using warn! instead of info!. Additionally, several high-frequency RPC entry points (ping, write_stream, read_at, walk_dir) were logging at info! level, causing unnecessary log noise in production. Changes: - load_rebalance_meta: warn! → info! for normal flow (3 occurrences) - ping body decode: info! → debug! (fires on every health check) - write_stream/read_at/walk_dir entry: info! → debug! - metrics.rs: lowercase error messages, add structured error field - http_service.rs: add structured error field to walk_dir failure - Add tracing::instrument fields for start_rebalance context * refactor(logging): standardize rustfs runtime events * build(deps): bump workspace dependency versions * build(deps): pin time to 0.3.47 * build(deps): update postgres client versions --- Cargo.lock | 52 +-- Cargo.toml | 12 +- crates/e2e_test/Cargo.toml | 2 +- rustfs/src/admin/handlers/kms_dynamic.rs | 281 +++++++++++-- rustfs/src/admin/handlers/policies.rs | 154 ++++++-- rustfs/src/admin/handlers/service_account.rs | 70 +++- rustfs/src/admin/handlers/trace.rs | 3 - rustfs/src/admin/router.rs | 73 +++- rustfs/src/app/bucket_usecase.rs | 57 ++- rustfs/src/app/object_usecase.rs | 26 +- rustfs/src/auth_keystone.rs | 86 +++- rustfs/src/capacity/service.rs | 125 +++++- rustfs/src/embedded.rs | 99 ++++- rustfs/src/init.rs | 391 ++++++++++++++++--- rustfs/src/main.rs | 66 +++- rustfs/src/profiling.rs | 136 ++++++- rustfs/src/server/http.rs | 219 +++++++++-- rustfs/src/server/tls_material.rs | 186 ++++++++- rustfs/src/storage/ecfs.rs | 96 ++++- rustfs/src/storage/ecfs_extend.rs | 31 +- rustfs/src/storage/rpc/http_service.rs | 2 +- rustfs/src/storage/rpc/metrics.rs | 4 +- rustfs/src/storage/rpc/node_service.rs | 70 +++- rustfs/src/storage/sse.rs | 33 +- scripts/check_logging_guardrails.sh | 52 +++ 25 files changed, 1980 insertions(+), 346 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index a9cbbb7ce..b83cf02ca 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1239,9 +1239,9 @@ dependencies = [ [[package]] name = "aws-smithy-types" -version = "1.4.9" +version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "53f93074121a1be41317b9aa607143ae17900631f7f59a99f2b905d519d6783b" +checksum = "32b42fcf341259d85ca10fac9a2f6448a8ec691c6955a18e45bc3b71a85fab85" dependencies = [ "base64-simd", "bytes", @@ -3576,7 +3576,7 @@ dependencies = [ "libc", "option-ext", "redox_users", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -3912,7 +3912,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -4348,9 +4348,9 @@ dependencies = [ [[package]] name = "google-cloud-auth" -version = "1.12.0" +version = "1.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4595b69c95c727ac896a4f2f16d0cdaa6f547de0bc5d64dbe9201487fc3226d5" +checksum = "a300d4011cb53573eafe2419630d303ced54aab6c194a6d9e4156de375800372" dependencies = [ "async-trait", "aws-lc-rs", @@ -4471,9 +4471,9 @@ dependencies = [ [[package]] name = "google-cloud-lro" -version = "1.7.0" +version = "1.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "eb2dba01e667af12137514f0673abb861f3785a32a685d70d63f7bbed335ccae" +checksum = "f40d4311479512e408b67777cf497dac6e41fbe8c96730416ed9d47855584b29" dependencies = [ "google-cloud-gax", "google-cloud-longrunning", @@ -4499,9 +4499,9 @@ dependencies = [ [[package]] name = "google-cloud-storage" -version = "1.14.0" +version = "1.15.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4993a4d81d6f350a735a6eaab9cc441c143e3e78fc351bc9bcc1a2448557e8fa" +checksum = "f796d8a7a5b1457ecd5ebb0451bf0502e15127cf42195591ffea90092fa82551" dependencies = [ "async-trait", "base64 0.22.1", @@ -5345,7 +5345,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -6571,7 +6571,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -6743,7 +6743,7 @@ version = "5.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "51e219e79014df21a225b1860a479e2dcd7cbd9130f4defd4bd0e191ea31d67d" dependencies = [ - "base64 0.22.1", + "base64 0.21.7", "chrono", "getrandom 0.2.17", "http 1.4.2", @@ -7664,9 +7664,9 @@ dependencies = [ [[package]] name = "postgres-protocol" -version = "0.6.11" +version = "0.6.12" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "56201207dac53e2f38e848e31b4b91616a6bb6e0c7205b77718994a7f49e70fc" +checksum = "08808e3c483c46e999108051c78334f473d5adb59d78bb80a1268c7e6aa6c514" dependencies = [ "base64 0.22.1", "byteorder", @@ -7682,9 +7682,9 @@ dependencies = [ [[package]] name = "postgres-types" -version = "0.2.13" +version = "0.2.14" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8dc729a129e682e8d24170cd30ae1aa01b336b096cbb56df6d534ffec133d186" +checksum = "851ca9db4932932d69f3ea811b1abe63087a0f740a47692619dd40d4899b68be" dependencies = [ "bytes", "fallible-iterator 0.2.0", @@ -10216,7 +10216,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -10290,7 +10290,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -10958,9 +10958,9 @@ dependencies = [ [[package]] name = "smallvec" -version = "1.15.1" +version = "1.15.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" +checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" dependencies = [ "serde", ] @@ -11441,10 +11441,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.4.2", + "getrandom 0.3.4", "once_cell", "rustix 1.1.4", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -11711,9 +11711,9 @@ dependencies = [ [[package]] name = "tokio-postgres" -version = "0.7.17" +version = "0.7.18" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4dd8df5ef180f6364759a6f00f7aadda4fbbac86cdee37480826a6ff9f3574ce" +checksum = "a528f7d280f6d5b9cd149635c8705b0dd049754bc67d81d31fa25169a93809d3" dependencies = [ "async-trait", "byteorder", @@ -12608,7 +12608,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 67d2cbdc7..eb1b4d980 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -194,11 +194,11 @@ zeroize = { version = "1.8.2", features = ["derive"] } chrono = { version = "0.4.45", features = ["serde"] } humantime = "2.3.0" jiff = { version = "0.2.28", features = ["serde"] } -time = { version = "0.3.47", features = ["std", "parsing", "formatting", "macros", "serde"] } +time = { version = "=0.3.47", features = ["std", "parsing", "formatting", "macros", "serde"] } # Database deadpool-postgres = { version = "0.14", features = ["rt_tokio_1"] } -tokio-postgres = { version = "0.7", default-features = false, features = ["runtime", "with-serde_json-1"] } +tokio-postgres = { version = "0.7.18", default-features = false, features = ["runtime", "with-serde_json-1"] } tokio-postgres-rustls = "0.14.0" # Utilities and Tools @@ -211,7 +211,7 @@ aws-config = { version = "1.8.18" } aws-credential-types = { version = "1.2.14" } aws-sdk-s3 = { version = "1.135.0", default-features = false, features = ["sigv4a", "default-https-client", "rt-tokio"] } aws-smithy-http-client = { version = "1.1.13", default-features = false, features = ["default-client", "rustls-aws-lc"] } -aws-smithy-types = { version = "1.4.9" } +aws-smithy-types = { version = "1.5.0" } base64 = "0.22.1" base64-simd = "0.8.0" brotli = "8.0.3" @@ -229,8 +229,8 @@ enumset = "1.1.13" faster-hex = "0.10.0" flate2 = "1.1.9" glob = "0.3.3" -google-cloud-storage = "1.14.0" -google-cloud-auth = "1.12.0" +google-cloud-storage = "1.15.0" +google-cloud-auth = "1.13.0" hashbrown = { version = "0.17.1", features = ["serde", "rayon"] } hex = "0.4.3" hex-simd = "0.8.0" @@ -272,7 +272,7 @@ s3s = { git = "https://github.com/rustfs/s3s", rev = "cf4c3346ed2554daa7fc6437f9 serial_test = "3.5.0" shadow-rs = { version = "2.0.0", default-features = false } siphasher = "1.0.3" -smallvec = { version = "1.15.1", features = ["serde"] } +smallvec = { version = "1.15.2", features = ["serde"] } smartstring = "1.0.1" snafu = "0.9.1" snap = "1.1.1" diff --git a/crates/e2e_test/Cargo.toml b/crates/e2e_test/Cargo.toml index 6855657bf..91316b920 100644 --- a/crates/e2e_test/Cargo.toml +++ b/crates/e2e_test/Cargo.toml @@ -67,7 +67,7 @@ chrono = { workspace = true } md5 = { workspace = true } sha2 = { workspace = true } astral-tokio-tar = { workspace = true } -s3s.workspace = true +s3s = { workspace = true } zstd.workspace = true time.workspace = true suppaftp = { workspace = true, features = ["tokio", "rustls-aws-lc-rs"] } diff --git a/rustfs/src/admin/handlers/kms_dynamic.rs b/rustfs/src/admin/handlers/kms_dynamic.rs index 7bb056553..990189253 100644 --- a/rustfs/src/admin/handlers/kms_dynamic.rs +++ b/rustfs/src/admin/handlers/kms_dynamic.rs @@ -34,10 +34,17 @@ use tracing::{error, info, instrument, warn}; /// Path to store KMS configuration in the cluster metadata const KMS_CONFIG_PATH: &str = "config/kms_config.json"; +const LOG_COMPONENT_ADMIN: &str = "admin"; +const LOG_SUBSYSTEM_KMS: &str = "kms"; fn kms_service_manager_from_context() -> std::sync::Arc { resolve_kms_runtime_service_manager().unwrap_or_else(|| { - warn!("KMS service manager not initialized, initializing now as fallback"); + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_service_manager_fallback", + "KMS service manager fallback initialized" + ); rustfs_kms::init_global_kms_service_manager() }) } @@ -105,7 +112,13 @@ async fn save_kms_config(config: &KmsConfig) -> Result<(), String> { .await .map_err(|e| format!("Failed to save KMS config to storage: {e}"))?; - info!("KMS configuration persisted to cluster storage at {}", KMS_CONFIG_PATH); + info!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_config_persisted", + storage_path = KMS_CONFIG_PATH, + "Persisted KMS configuration" + ); Ok(()) } @@ -113,27 +126,60 @@ async fn save_kms_config(config: &KmsConfig) -> Result<(), String> { #[instrument] pub async fn load_kms_config() -> Option { let Some(store) = new_object_layer_fn() else { - warn!("Storage layer not initialized, cannot load KMS config"); + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_config_load_skipped", + reason = "storage_uninitialized", + "Skipped KMS configuration load" + ); return None; }; match read_config(store, KMS_CONFIG_PATH).await { Ok(data) => match serde_json::from_slice::(&data) { Ok(config) => { - info!("Loaded KMS configuration from cluster storage"); + info!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_config_loaded", + storage_path = KMS_CONFIG_PATH, + "Loaded persisted KMS configuration" + ); Some(config) } Err(e) => { - error!("Failed to deserialize KMS config: {}", e); + error!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_config_deserialize_failed", + storage_path = KMS_CONFIG_PATH, + error = %e, + "Failed to deserialize KMS configuration" + ); None } }, Err(e) => { // Config not found is normal on first run if e.to_string().contains("ConfigNotFound") || e.to_string().contains("not found") { - info!("No persisted KMS configuration found (first run or not configured yet)"); + info!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_config_loaded", + state = "not_found", + storage_path = KMS_CONFIG_PATH, + "Persisted KMS configuration not found" + ); } else { - warn!("Failed to load KMS config from storage: {}", e); + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_config_load_failed", + storage_path = KMS_CONFIG_PATH, + error = %e, + "Failed to load KMS configuration" + ); } None } @@ -212,13 +258,27 @@ impl Operation for ConfigureKmsHandler { match serde_json::from_slice(&body) { Ok(req) => req, Err(e) => { - error!("Invalid JSON in configure request: {}", e); + error!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_request_decode_failed", + operation = "configure", + error = %e, + "Failed to decode KMS admin request" + ); return Ok(S3Response::new((StatusCode::BAD_REQUEST, Body::from(format!("Invalid JSON: {e}"))))); } } }; - info!("Configuring KMS from admin request"); + info!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_service_state", + operation = "configure", + state = "requested", + "KMS service state changed" + ); let service_manager = kms_service_manager_from_context(); let existing_config = service_manager.get_config().await; @@ -236,18 +296,42 @@ impl Operation for ConfigureKmsHandler { // Persist the configuration to cluster storage if let Err(e) = save_kms_config(&kms_config).await { let error_msg = format!("KMS configured in memory but failed to persist: {e}"); - error!("{}", error_msg); + error!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_service_state", + operation = "configure", + state = "persist_failed", + error = %e, + "KMS service state changed" + ); let status = service_manager.get_status().await; (false, error_msg, status) } else { let status = service_manager.get_status().await; - info!("KMS configured successfully and persisted with status: {:?}", status); + info!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_service_state", + operation = "configure", + state = "configured", + status = ?status, + "KMS service state changed" + ); (true, "KMS configured successfully".to_string(), status) } } Err(e) => { let error_msg = format!("Failed to configure KMS: {e}"); - error!("{}", error_msg); + error!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_service_state", + operation = "configure", + state = "configure_failed", + error = %e, + "KMS service state changed" + ); let status = service_manager.get_status().await; (false, error_msg, status) } @@ -309,20 +393,42 @@ impl Operation for StartKmsHandler { match serde_json::from_slice(&body) { Ok(req) => req, Err(e) => { - error!("Invalid JSON in start request: {}", e); + error!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_request_decode_failed", + operation = "start", + error = %e, + "Failed to decode KMS admin request" + ); return Ok(S3Response::new((StatusCode::BAD_REQUEST, Body::from(format!("Invalid JSON: {e}"))))); } } }; - info!("Starting KMS service with force: {:?}", start_request.force); + info!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_service_state", + operation = "start", + state = "requested", + force = start_request.force.unwrap_or(false), + "KMS service state changed" + ); let service_manager = kms_service_manager_from_context(); // Check if already running and force flag let current_status = service_manager.get_status().await; if matches!(current_status, KmsServiceStatus::Running) && !start_request.force.unwrap_or(false) { - warn!("KMS service is already running"); + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_service_state", + operation = "start", + state = "already_running", + "KMS service state changed" + ); let response = StartKmsResponse { success: false, message: "KMS service is already running. Use force=true to restart.".to_string(), @@ -349,19 +455,43 @@ impl Operation for StartKmsHandler { Ok(()) => match service_manager.start().await { Ok(()) => { let status = service_manager.get_status().await; - info!("KMS service restarted successfully"); + info!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_service_state", + operation = "restart", + state = "running", + status = ?status, + "KMS service state changed" + ); (true, "KMS service restarted successfully".to_string(), status) } Err(e) => { let error_msg = format!("Failed to restart KMS service: {e}"); - error!("{}", error_msg); + error!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_service_state", + operation = "restart", + state = "start_failed", + error = %e, + "KMS service state changed" + ); let status = service_manager.get_status().await; (false, error_msg, status) } }, Err(e) => { let error_msg = format!("Failed to stop KMS service for restart: {e}"); - error!("{}", error_msg); + error!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_service_state", + operation = "restart", + state = "stop_failed", + error = %e, + "KMS service state changed" + ); let status = service_manager.get_status().await; (false, error_msg, status) } @@ -371,12 +501,28 @@ impl Operation for StartKmsHandler { match service_manager.start().await { Ok(()) => { let status = service_manager.get_status().await; - info!("KMS service started successfully"); + info!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_service_state", + operation = "start", + state = "running", + status = ?status, + "KMS service state changed" + ); (true, "KMS service started successfully".to_string(), status) } Err(e) => { let error_msg = format!("Failed to start KMS service: {e}"); - error!("{}", error_msg); + error!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_service_state", + operation = "start", + state = "start_failed", + error = %e, + "KMS service state changed" + ); let status = service_manager.get_status().await; (false, error_msg, status) } @@ -427,19 +573,42 @@ impl Operation for StopKmsHandler { ) .await?; - info!("Stopping KMS service"); + info!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_service_state", + operation = "stop", + state = "requested", + "KMS service state changed" + ); let service_manager = kms_service_manager_from_context(); let (success, message, status) = match service_manager.stop().await { Ok(()) => { let status = service_manager.get_status().await; - info!("KMS service stopped successfully"); + info!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_service_state", + operation = "stop", + state = "stopped", + status = ?status, + "KMS service state changed" + ); (true, "KMS service stopped successfully".to_string(), status) } Err(e) => { let error_msg = format!("Failed to stop KMS service: {e}"); - error!("{}", error_msg); + error!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_service_state", + operation = "stop", + state = "stop_failed", + error = %e, + "KMS service state changed" + ); let status = service_manager.get_status().await; (false, error_msg, status) } @@ -489,7 +658,12 @@ impl Operation for GetKmsStatusHandler { ) .await?; - info!("Getting KMS service status"); + info!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_status_requested", + "KMS status requested" + ); let service_manager = kms_service_manager_from_context(); @@ -517,7 +691,16 @@ impl Operation for GetKmsStatusHandler { config_summary, }; - info!("KMS status: {:?}", response); + info!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_status_resolved", + status = ?response.status, + backend_type = ?response.backend_type, + healthy = response.healthy, + has_config_summary = response.config_summary.is_some(), + "KMS status resolved" + ); let json_response = match serde_json::to_string(&response) { Ok(json) => json, @@ -572,13 +755,27 @@ impl Operation for ReconfigureKmsHandler { match serde_json::from_slice(&body) { Ok(req) => req, Err(e) => { - error!("Invalid JSON in reconfigure request: {}", e); + error!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_request_decode_failed", + operation = "reconfigure", + error = %e, + "Failed to decode KMS admin request" + ); return Ok(S3Response::new((StatusCode::BAD_REQUEST, Body::from(format!("Invalid JSON: {e}"))))); } } }; - info!("Reconfiguring KMS"); + info!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_service_state", + operation = "reconfigure", + state = "requested", + "KMS service state changed" + ); let service_manager = kms_service_manager_from_context(); let existing_config = service_manager.get_config().await; @@ -596,18 +793,42 @@ impl Operation for ReconfigureKmsHandler { // Persist the configuration to cluster storage if let Err(e) = save_kms_config(&kms_config).await { let error_msg = format!("KMS reconfigured in memory but failed to persist: {e}"); - error!("{}", error_msg); + error!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_service_state", + operation = "reconfigure", + state = "persist_failed", + error = %e, + "KMS service state changed" + ); let status = service_manager.get_status().await; (false, error_msg, status) } else { let status = service_manager.get_status().await; - info!("KMS reconfigured successfully and persisted with status: {:?}", status); + info!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_service_state", + operation = "reconfigure", + state = "reconfigured", + status = ?status, + "KMS service state changed" + ); (true, "KMS reconfigured and restarted successfully".to_string(), status) } } Err(e) => { let error_msg = format!("Failed to reconfigure KMS: {e}"); - error!("{}", error_msg); + error!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_service_state", + operation = "reconfigure", + state = "reconfigure_failed", + error = %e, + "KMS service state changed" + ); let status = service_manager.get_status().await; (false, error_msg, status) } diff --git a/rustfs/src/admin/handlers/policies.rs b/rustfs/src/admin/handlers/policies.rs index 1b91da5fd..a8decfd29 100644 --- a/rustfs/src/admin/handlers/policies.rs +++ b/rustfs/src/admin/handlers/policies.rs @@ -50,6 +50,9 @@ use time::OffsetDateTime; use tracing::warn; use url::form_urlencoded; +const LOG_COMPONENT_ADMIN: &str = "admin"; +const LOG_SUBSYSTEM_POLICY: &str = "policy"; + pub fn register_iam_policy_route(r: &mut S3Router) -> std::io::Result<()> { r.insert( Method::GET, @@ -113,8 +116,6 @@ pub struct ListCannedPolicies {} #[async_trait::async_trait] impl Operation for ListCannedPolicies { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { - warn!("handle ListCannedPolicies"); - let Some(input_cred) = req.credentials else { return Err(s3_error!(InvalidRequest, "get cred failed")); }; @@ -145,7 +146,14 @@ impl Operation for ListCannedPolicies { let Ok(iam_store) = rustfs_iam::get() else { return Err(s3_error!(InternalError, "iam not init")) }; let policies = iam_store.list_polices(&query.bucket).await.map_err(|e| { - warn!("list policies failed, e: {:?}", e); + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_POLICY, + event = "policy_list_failed", + bucket = %query.bucket, + error = ?e, + "Failed to list canned policies" + ); S3Error::with_message(S3ErrorCode::InternalError, e.to_string()) })?; @@ -172,8 +180,6 @@ pub struct AddCannedPolicy {} #[async_trait::async_trait] impl Operation for AddCannedPolicy { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { - warn!("handle AddCannedPolicy"); - let Some(input_cred) = req.credentials else { return Err(s3_error!(InvalidRequest, "get cred failed")); }; @@ -213,13 +219,27 @@ impl Operation for AddCannedPolicy { let policy_bytes = match input.store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE).await { Ok(b) => b, Err(e) => { - warn!("get body failed, e: {:?}", e); + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_POLICY, + event = "policy_body_read_failed", + policy = %query.name, + error = ?e, + "Failed to read policy request body" + ); return Err(s3_error!(InvalidRequest, "policy configuration body too large or failed to read")); } }; let policy = Policy::parse_config(policy_bytes.as_ref()).map_err(|e| { - warn!("parse policy failed, e: {:?}", e); + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_POLICY, + event = "policy_parse_failed", + policy = %query.name, + error = ?e, + "Failed to parse canned policy" + ); S3Error::with_message(S3ErrorCode::InvalidRequest, e.to_string()) })?; @@ -229,7 +249,14 @@ impl Operation for AddCannedPolicy { let Ok(iam_store) = rustfs_iam::get() else { return Err(s3_error!(InternalError, "iam not init")) }; let updated_at = iam_store.set_policy(&query.name, policy.clone()).await.map_err(|e| { - warn!("set policy failed, e: {:?}", e); + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_POLICY, + event = "policy_set_failed", + policy = %query.name, + error = ?e, + "Failed to persist canned policy" + ); S3Error::with_message(S3ErrorCode::InternalError, e.to_string()) })?; @@ -259,8 +286,6 @@ pub struct InfoCannedPolicy {} #[async_trait::async_trait] impl Operation for InfoCannedPolicy { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { - warn!("handle InfoCannedPolicy"); - let Some(input_cred) = req.credentials else { return Err(s3_error!(InvalidRequest, "get cred failed")); }; @@ -300,7 +325,14 @@ impl Operation for InfoCannedPolicy { let Ok(iam_store) = rustfs_iam::get() else { return Err(s3_error!(InternalError, "iam not init")) }; let pd = iam_store.info_policy(&query.name).await.map_err(|e| { - warn!("info policy failed, e: {:?}", e); + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_POLICY, + event = "policy_info_failed", + policy = %query.name, + error = ?e, + "Failed to load canned policy" + ); S3Error::with_message(S3ErrorCode::InternalError, e.to_string()) })?; @@ -317,8 +349,6 @@ pub struct RemoveCannedPolicy {} #[async_trait::async_trait] impl Operation for RemoveCannedPolicy { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { - warn!("handle RemoveCannedPolicy"); - let Some(input_cred) = req.credentials else { return Err(s3_error!(InvalidRequest, "get cred failed")); }; @@ -353,7 +383,14 @@ impl Operation for RemoveCannedPolicy { let Ok(iam_store) = rustfs_iam::get() else { return Err(s3_error!(InternalError, "iam not init")) }; iam_store.delete_policy(&query.name, true).await.map_err(|e| { - warn!("delete policy failed, e: {:?}", e); + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_POLICY, + event = "policy_delete_failed", + policy = %query.name, + error = ?e, + "Failed to delete canned policy" + ); S3Error::with_message(S3ErrorCode::InternalError, e.to_string()) })?; @@ -391,8 +428,6 @@ pub struct SetPolicyForUserOrGroup {} #[async_trait::async_trait] impl Operation for SetPolicyForUserOrGroup { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { - warn!("handle SetPolicyForUserOrGroup"); - let Some(input_cred) = req.credentials else { return Err(s3_error!(InvalidRequest, "get cred failed")); }; @@ -435,7 +470,15 @@ impl Operation for SetPolicyForUserOrGroup { } Err(err) => { if !is_err_no_such_user(&err) { - warn!("is temp user failed, e: {:?}", err); + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_POLICY, + event = "policy_target_validation_failed", + target = %query.user_or_group, + check = "is_temp_user", + error = ?err, + "Failed to validate policy target" + ); return Err(S3Error::with_message(S3ErrorCode::InternalError, err.to_string())); } } @@ -456,7 +499,15 @@ impl Operation for SetPolicyForUserOrGroup { } } else { iam_store.get_group_description(&query.user_or_group).await.map_err(|e| { - warn!("get group description failed, e: {:?}", e); + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_POLICY, + event = "policy_target_validation_failed", + target = %query.user_or_group, + check = "group_description", + error = ?e, + "Failed to validate policy target" + ); iam_error_to_s3_error(e) })?; } @@ -465,7 +516,16 @@ impl Operation for SetPolicyForUserOrGroup { .policy_db_set(&query.user_or_group, rustfs_iam::store::UserType::Reg, query.is_group, &query.policy_name) .await .map_err(|e| { - warn!("policy db set failed, e: {:?}", e); + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_POLICY, + event = "policy_mapping_set_failed", + target = %query.user_or_group, + policy = %query.policy_name, + is_group = query.is_group, + error = ?e, + "Failed to update policy mapping" + ); S3Error::with_message(S3ErrorCode::InternalError, e.to_string()) })?; @@ -674,7 +734,14 @@ async fn collect_group_policy_mappings( for group in groups { let group_desc = iam_store.get_group_description(&group).await.map_err(|e| { - warn!("get group description failed, e: {:?}", e); + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_POLICY, + event = "policy_group_lookup_failed", + group = %group, + error = ?e, + "Failed to load group policy mapping" + ); iam_error_to_s3_error(e) })?; let policies = split_policy_names(&group_desc.policy); @@ -715,7 +782,13 @@ async fn handle_builtin_policy_entities(req: S3Request) -> S3Result, is_attach: bool Ok((false, _)) => {} Err(err) => { if !is_err_no_such_user(&err) { - warn!("is temp user failed, e: {:?}", err); + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_POLICY, + event = "policy_target_validation_failed", + target = %assoc_req.user, + check = "is_temp_user", + error = ?err, + "Failed to validate policy association target" + ); return Err(S3Error::with_message(S3ErrorCode::InternalError, err.to_string())); } } @@ -866,14 +947,28 @@ async fn handle_builtin_policy_association(req: S3Request, is_attach: bool } let user_info = iam_store.get_user_info(&assoc_req.user).await.map_err(|e| { - warn!("get user info failed, e: {:?}", e); + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_POLICY, + event = "policy_user_info_failed", + target = %assoc_req.user, + error = ?e, + "Failed to load policy association target" + ); iam_error_to_s3_error(e) })?; (assoc_req.user, false, direct_user_policy_names(&user_info)) } else { let group_desc = iam_store.get_group_description(&assoc_req.group).await.map_err(|e| { - warn!("get group description failed, e: {:?}", e); + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_POLICY, + event = "policy_group_lookup_failed", + group = %assoc_req.group, + error = ?e, + "Failed to load policy association target" + ); iam_error_to_s3_error(e) })?; @@ -890,7 +985,16 @@ async fn handle_builtin_policy_association(req: S3Request, is_attach: bool .policy_db_set(&target_name, rustfs_iam::store::UserType::Reg, is_group, &updated_policies.join(",")) .await .map_err(|e| { - warn!("policy db set failed, e: {:?}", e); + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_POLICY, + event = "policy_mapping_set_failed", + target = %target_name, + policy_count = updated_policies.len(), + is_group, + error = ?e, + "Failed to update policy association" + ); S3Error::with_message(S3ErrorCode::InternalError, e.to_string()) })?; diff --git a/rustfs/src/admin/handlers/service_account.rs b/rustfs/src/admin/handlers/service_account.rs index 32bbc5701..c440e0f3c 100644 --- a/rustfs/src/admin/handlers/service_account.rs +++ b/rustfs/src/admin/handlers/service_account.rs @@ -48,6 +48,9 @@ use time::OffsetDateTime; use tracing::{debug, warn}; use url::form_urlencoded; +const LOG_COMPONENT_ADMIN: &str = "admin"; +const LOG_SUBSYSTEM_SERVICE_ACCOUNT: &str = "service_account"; + fn sr_session_policy_from_value(value: Option<&serde_json::Value>) -> S3Result { let Some(value) = value else { return Ok(SRSessionPolicy::default()); @@ -104,7 +107,14 @@ fn is_service_account_owner_of(caller: &StoredCredentials, target_parent_user: & } fn map_service_account_lookup_error(err: rustfs_iam::error::Error, action: &str) -> S3Error { - debug!("{action}, e: {:?}", err); + debug!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_SERVICE_ACCOUNT, + event = "service_account_lookup_failed", + action, + error = ?err, + "Service account lookup failed" + ); if is_err_no_such_service_account(&err) { iam_error_to_s3_error(err) } else { @@ -113,7 +123,14 @@ fn map_service_account_lookup_error(err: rustfs_iam::error::Error, action: &str) } fn map_temp_account_lookup_error(err: rustfs_iam::error::Error, action: &str) -> S3Error { - debug!("{action}, e: {:?}", err); + debug!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_SERVICE_ACCOUNT, + event = "temporary_account_lookup_failed", + action, + error = ?err, + "Temporary account lookup failed" + ); if is_err_no_such_temp_account(&err) { iam_error_to_s3_error(err) } else { @@ -124,7 +141,13 @@ fn map_temp_account_lookup_error(err: rustfs_iam::error::Error, action: &str) -> fn parse_service_account_policy(policy: &serde_json::Value) -> S3Result { let policy_bytes = serde_json::to_vec(policy).map_err(|e| s3_error!(InvalidArgument, "marshal policy failed: {:?}", e))?; Policy::parse_config(&policy_bytes).map_err(|e| { - debug!("parse service account policy failed, e: {:?}", e); + debug!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_SERVICE_ACCOUNT, + event = "service_account_policy_parse_failed", + error = ?e, + "Failed to parse service account policy" + ); match e { rustfs_policy::error::Error::PolicyError(rustfs_policy::policy::Error::NonResource) => { s3_error!(InvalidArgument, "invalid service account policy: Resource is empty") @@ -214,7 +237,6 @@ pub struct AddServiceAccount {} #[async_trait::async_trait] impl Operation for AddServiceAccount { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { - warn!("handle AddServiceAccount "); let Some(req_cred) = req.credentials else { return Err(s3_error!(InvalidRequest, "get cred failed")); }; @@ -347,7 +369,14 @@ impl Operation for AddServiceAccount { .new_service_account(&target_user, target_groups, opts) .await .map_err(|e| { - debug!("create service account failed, e: {:?}", e); + debug!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_SERVICE_ACCOUNT, + event = "service_account_create_failed", + target_user = %target_user, + error = ?e, + "Failed to create service account" + ); s3_error!(InternalError, "create service account failed, e: {:?}", e) })?; @@ -449,8 +478,6 @@ pub struct UpdateServiceAccount {} #[async_trait::async_trait] impl Operation for UpdateServiceAccount { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { - warn!("handle UpdateServiceAccount"); - let query = { if let Some(query) = req.uri.query() { let input: AccessKeyQuery = @@ -583,8 +610,6 @@ pub struct InfoServiceAccount {} #[async_trait::async_trait] impl Operation for InfoServiceAccount { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { - warn!("handle InfoServiceAccount"); - let query = { if let Some(query) = req.uri.query() { let input: AccessKeyQuery = @@ -657,8 +682,6 @@ pub struct TemporaryAccountInfo {} #[async_trait::async_trait] impl Operation for TemporaryAccountInfo { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { - warn!("handle TemporaryAccountInfo"); - let query = { if let Some(query) = req.uri.query() { let input: AccessKeyQuery = @@ -727,8 +750,6 @@ pub struct InfoAccessKey {} #[async_trait::async_trait] impl Operation for InfoAccessKey { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { - warn!("handle InfoAccessKey"); - let query = { if let Some(query) = req.uri.query() { let input: AccessKeyQuery = @@ -832,8 +853,6 @@ pub struct ListServiceAccount {} #[async_trait::async_trait] impl Operation for ListServiceAccount { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { - warn!("handle ListServiceAccount"); - let query = { if let Some(query) = req.uri.query() { let input: ListServiceAccountQuery = from_bytes(query.as_bytes()) @@ -986,8 +1005,6 @@ pub struct ListAccessKeysBulk {} #[async_trait::async_trait] impl Operation for ListAccessKeysBulk { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { - warn!("handle ListAccessKeysBulk"); - let query = { parse_list_access_keys_query(req.uri.query()) }; if query.all && !query.users.is_empty() { @@ -1155,7 +1172,6 @@ pub struct DeleteServiceAccount {} #[async_trait::async_trait] impl Operation for DeleteServiceAccount { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { - warn!("handle DeleteServiceAccount"); let Some(input_cred) = req.credentials else { return Err(s3_error!(InvalidRequest, "get cred failed")); }; @@ -1164,7 +1180,14 @@ impl Operation for DeleteServiceAccount { check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key) .await .map_err(|e| { - debug!("check key failed: {e:?}"); + debug!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_SERVICE_ACCOUNT, + event = "service_account_auth_failed", + action = "delete", + error = ?e, + "Service account authentication failed" + ); s3_error!(InternalError, "check key failed") })?; @@ -1229,7 +1252,14 @@ impl Operation for DeleteServiceAccount { } iam_store.delete_service_account(&query.access_key, true).await.map_err(|e| { - debug!("delete service account failed, e: {:?}", e); + debug!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_SERVICE_ACCOUNT, + event = "service_account_delete_failed", + access_key = %query.access_key, + error = ?e, + "Failed to delete service account" + ); s3_error!(InternalError, "delete service account failed") })?; diff --git a/rustfs/src/admin/handlers/trace.rs b/rustfs/src/admin/handlers/trace.rs index 22b4a4b4a..08c297faf 100644 --- a/rustfs/src/admin/handlers/trace.rs +++ b/rustfs/src/admin/handlers/trace.rs @@ -20,7 +20,6 @@ use matchit::Params; use rustfs_ecstore::rpc::PeerRestClient; use rustfs_madmin::service_commands::ServiceTraceOpts; use s3s::{Body, S3Request, S3Response, S3Result, s3_error}; -use tracing::warn; #[allow(dead_code)] fn extract_trace_options(uri: &Uri) -> S3Result { @@ -38,8 +37,6 @@ pub struct Trace {} #[async_trait::async_trait] impl Operation for Trace { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { - warn!("handle Trace"); - let _trace_opts = extract_trace_options(&req.uri)?; // let (tx, rx) = mpsc::channel(10000); diff --git a/rustfs/src/admin/router.rs b/rustfs/src/admin/router.rs index 63cfa251f..a1fc32762 100644 --- a/rustfs/src/admin/router.rs +++ b/rustfs/src/admin/router.rs @@ -101,6 +101,10 @@ use tracing::{error, warn}; use url::form_urlencoded; use uuid::Uuid; +const LOG_COMPONENT_ADMIN: &str = "admin"; +const LOG_SUBSYSTEM_OBJECT_LAMBDA: &str = "object_lambda"; +const LOG_SUBSYSTEM_LIVE_EVENTS: &str = "live_events"; + #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum ReplicationExtRoute { MetricsV1, @@ -629,7 +633,13 @@ async fn load_current_server_config() -> S3Result { match read_config_without_migrate(store).await { Ok(config) => return Ok(config), Err(err) => { - warn!("failed to reload current server config for object lambda request: {err}"); + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_OBJECT_LAMBDA, + event = "object_lambda_config_reload_failed", + error = %err, + "Failed to reload current server config for object lambda request" + ); } } } @@ -653,8 +663,11 @@ fn build_object_lambda_http_client(config: &ObjectLambdaWebhookConfig) -> S3Resu if config.skip_tls_verify { warn!( - "Object Lambda webhook target '{}' is configured to skip TLS certificate verification. This permits MITM attacks and should not be used in production.", - config.endpoint + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_OBJECT_LAMBDA, + event = "object_lambda_tls_verification_disabled", + endpoint = %config.endpoint, + "Object Lambda target is configured to skip TLS certificate verification" ); builder = builder.danger_accept_invalid_certs(true); } else if !config.client_ca.is_empty() { @@ -1206,11 +1219,25 @@ async fn fan_in_remote_live_events( { Ok(Ok(batch)) => batch, Ok(Err(err)) => { - warn!("failed to fetch live events from peer {}: {err}", peer.client.host); + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_LIVE_EVENTS, + event = "peer_live_events_fetch_failed", + peer = %peer.client.host, + error = %err, + "Failed to fetch live events from peer" + ); break; } Err(_) => { - warn!("timed out fetching live events from peer {}", peer.client.host); + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_LIVE_EVENTS, + event = "peer_live_events_fetch_failed", + peer = %peer.client.host, + error = "timeout", + "Failed to fetch live events from peer" + ); break; } }; @@ -1231,13 +1258,28 @@ async fn fan_in_remote_live_events( } } Err(err) => { - warn!("failed to serialize remote listen notification event: {err}"); + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_LIVE_EVENTS, + event = "live_event_serialize_failed", + source = "remote_peer", + peer = %peer.client.host, + error = %err, + "Failed to serialize live event" + ); } } } } Err(err) => { - warn!("failed to decode live events from peer {}: {err}", peer.client.host); + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_LIVE_EVENTS, + event = "peer_live_events_decode_failed", + peer = %peer.client.host, + error = %err, + "Failed to decode live events from peer" + ); } } } @@ -1290,12 +1332,25 @@ fn build_listen_notification_response(uri: &Uri, bucket: Option<&str>) -> S3Resu } } Err(err) => { - warn!("failed to serialize listen notification event: {err}"); + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_LIVE_EVENTS, + event = "live_event_serialize_failed", + source = "local_stream", + error = %err, + "Failed to serialize live event" + ); } } } Err(broadcast::error::RecvError::Lagged(skipped)) => { - warn!("listen notification stream lagged and skipped {skipped} events"); + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_LIVE_EVENTS, + event = "live_event_stream_lagged", + skipped, + "Live event stream lagged" + ); } Err(broadcast::error::RecvError::Closed) => break, } diff --git a/rustfs/src/app/bucket_usecase.rs b/rustfs/src/app/bucket_usecase.rs index d168c647c..d377c56cb 100644 --- a/rustfs/src/app/bucket_usecase.rs +++ b/rustfs/src/app/bucket_usecase.rs @@ -81,6 +81,9 @@ use std::{ sync::Arc, }; use tracing::{debug, error, info, instrument, warn}; + +const LOG_COMPONENT_APP: &str = "app"; +const LOG_SUBSYSTEM_BUCKET: &str = "bucket"; use urlencoding::encode; fn serialize_config(value: &T) -> S3Result> { @@ -1148,7 +1151,14 @@ impl DefaultBucketUsecase { if err == StorageError::ConfigNotFound { return Err(s3_error!(ServerSideEncryptionConfigurationNotFoundError)); } - warn!("get_sse_config err {:?}", err); + warn!( + component = LOG_COMPONENT_APP, + subsystem = LOG_SUBSYSTEM_BUCKET, + event = "bucket_sse_config_load_failed", + bucket = %bucket, + error = ?err, + "Failed to load bucket SSE configuration" + ); return Err(ApiError::from(err).into()); } }; @@ -1180,7 +1190,14 @@ impl DefaultBucketUsecase { "The CORS configuration does not exist".to_string(), )); } - warn!("get_cors_config err {:?}", &err); + warn!( + component = LOG_COMPONENT_APP, + subsystem = LOG_SUBSYSTEM_BUCKET, + event = "bucket_cors_config_load_failed", + bucket = %bucket, + error = ?err, + "Failed to load bucket CORS configuration" + ); return Err(ApiError::from(err).into()); } }; @@ -1235,7 +1252,14 @@ impl DefaultBucketUsecase { .map_err(ApiError::from)?; let has_notification_config = metadata_sys::get_notification_config(&bucket).await.unwrap_or_else(|err| { - warn!("get_notification_config err {:?}", err); + warn!( + component = LOG_COMPONENT_APP, + subsystem = LOG_SUBSYSTEM_BUCKET, + event = "bucket_notification_config_load_failed", + bucket = %bucket, + error = ?err, + "Failed to load bucket notification configuration" + ); None }); @@ -1428,7 +1452,14 @@ impl DefaultBucketUsecase { if err == StorageError::ConfigNotFound { return Err(S3Error::with_message(S3ErrorCode::NoSuchTagSet, "The TagSet does not exist".to_string())); } - warn!("get_tagging_config err {:?}", &err); + warn!( + component = LOG_COMPONENT_APP, + subsystem = LOG_SUBSYSTEM_BUCKET, + event = "bucket_tagging_config_load_failed", + bucket = %bucket, + error = ?err, + "Failed to load bucket tagging configuration" + ); return Err(ApiError::from(err).into()); } }; @@ -1571,7 +1602,14 @@ impl DefaultBucketUsecase { Ok((cfg, _)) => cfg, Err(StorageError::ConfigNotFound) => ObjectLockConfiguration::default(), Err(err) => { - warn!("get_object_lock_config err {:?}", err); + warn!( + component = LOG_COMPONENT_APP, + subsystem = LOG_SUBSYSTEM_BUCKET, + event = "bucket_object_lock_config_load_failed", + bucket = %bucket, + error = ?err, + "Failed to load bucket object lock configuration" + ); return Err(ApiError::from(err).into()); } }; @@ -1740,7 +1778,14 @@ impl DefaultBucketUsecase { } Err(err) => { if err != StorageError::ConfigNotFound { - warn!("get_public_access_block_config err {:?}", &err); + warn!( + component = LOG_COMPONENT_APP, + subsystem = LOG_SUBSYSTEM_BUCKET, + event = "bucket_public_access_block_load_failed", + bucket = %bucket, + error = ?err, + "Failed to load bucket public access block configuration" + ); return Err(ApiError::from(err).into()); } } diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 942514954..c8bc05c12 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -1822,19 +1822,22 @@ impl DefaultObjectUsecase { let store = get_validated_store(&bucket).await?; - // TDD: Get bucket default encryption configuration let bucket_sse_config = metadata_sys::get_sse_config(&bucket).await.ok(); - debug!("TDD: bucket_sse_config={:?}", bucket_sse_config); + debug!( + target: "rustfs::app::object_usecase", + component = "app", + subsystem = "object", + event = "bucket_sse_config_lookup", + bucket = %bucket, + found = bucket_sse_config.is_some(), + "Bucket SSE configuration lookup completed" + ); - // TDD: Determine effective encryption configuration (request overrides bucket default) let original_sse = server_side_encryption.clone(); let mut effective_sse = server_side_encryption.or_else(|| { bucket_sse_config.as_ref().and_then(|(config, _timestamp)| { - debug!("TDD: Processing bucket SSE config: {:?}", config); config.rules.first().and_then(|rule| { - debug!("TDD: Processing SSE rule: {:?}", rule); rule.apply_server_side_encryption_by_default.as_ref().map(|sse| { - debug!("TDD: Found SSE default: {:?}", sse); match sse.sse_algorithm.as_str() { "AES256" => ServerSideEncryption::from_static(ServerSideEncryption::AES256), "aws:kms" => ServerSideEncryption::from_static(ServerSideEncryption::AWS_KMS), @@ -1844,7 +1847,16 @@ impl DefaultObjectUsecase { }) }) }); - debug!("TDD: effective_sse={:?} (original={:?})", effective_sse, original_sse); + debug!( + target: "rustfs::app::object_usecase", + component = "app", + subsystem = "object", + event = "effective_sse_resolved", + bucket = %bucket, + requested = ?original_sse, + effective = ?effective_sse, + "Resolved effective SSE configuration" + ); let mut effective_kms_key_id = ssekms_key_id.or_else(|| { bucket_sse_config.as_ref().and_then(|(config, _timestamp)| { diff --git a/rustfs/src/auth_keystone.rs b/rustfs/src/auth_keystone.rs index d75e2ca42..04a0d11fa 100644 --- a/rustfs/src/auth_keystone.rs +++ b/rustfs/src/auth_keystone.rs @@ -17,22 +17,42 @@ use http::HeaderMap; use rustfs_credentials::Credentials; use rustfs_keystone::{KeystoneAuthProvider, KeystoneClient, KeystoneConfig, KeystoneIdentityMapper}; +use rustfs_obs::MaskedAccessKey; use s3s::{S3Result, s3_error}; use std::sync::{Arc, OnceLock}; -use tracing::{debug, error, info}; +use tracing::{error, info}; static KEYSTONE_AUTH: OnceLock> = OnceLock::new(); static KEYSTONE_MAPPER: OnceLock> = OnceLock::new(); static KEYSTONE_CONFIG: OnceLock = OnceLock::new(); +const LOG_COMPONENT_AUTH: &str = "auth"; +const LOG_SUBSYSTEM_KEYSTONE: &str = "keystone"; + /// Initialize Keystone authentication pub async fn init_keystone_auth(config: KeystoneConfig) -> Result<(), Box> { if !config.enable { - info!("Keystone authentication disabled"); + info!( + component = LOG_COMPONENT_AUTH, + subsystem = LOG_SUBSYSTEM_KEYSTONE, + event = "keystone_state", + state = "disabled", + "Keystone authentication state changed" + ); return Ok(()); } - info!("Initializing Keystone authentication..."); + info!( + component = LOG_COMPONENT_AUTH, + subsystem = LOG_SUBSYSTEM_KEYSTONE, + event = "keystone_state", + state = "initializing", + auth_url = %config.auth_url, + version = %config.version, + enable_tenant_prefix = config.enable_tenant_prefix, + enable_cache = config.enable_cache, + "Keystone authentication state changed" + ); // Validate configuration config.validate()?; @@ -71,11 +91,17 @@ pub async fn init_keystone_auth(config: KeystoneConfig) -> Result<(), Box S3Result { - info!("Keystone token authentication successful: user={}", cred.parent_user); + info!( + component = LOG_COMPONENT_AUTH, + subsystem = LOG_SUBSYSTEM_KEYSTONE, + event = "keystone_token_auth", + token_type = "x_auth_token", + principal = %MaskedAccessKey(&cred.parent_user), + result = "success", + "Keystone token authentication completed" + ); Ok(Some(cred)) } Err(e) => { - error!("Keystone token authentication failed: {}", e); + error!( + component = LOG_COMPONENT_AUTH, + subsystem = LOG_SUBSYSTEM_KEYSTONE, + event = "keystone_token_auth", + token_type = "x_auth_token", + result = "failed", + error = %e, + "Keystone token authentication completed" + ); Err(s3_error!(InvalidToken, "Invalid Keystone token: {}", e)) } }; @@ -141,15 +181,29 @@ pub async fn authenticate_keystone(headers: &HeaderMap) -> S3Result { - info!("Keystone Swift token authentication successful: user={}", cred.parent_user); + info!( + component = LOG_COMPONENT_AUTH, + subsystem = LOG_SUBSYSTEM_KEYSTONE, + event = "keystone_token_auth", + token_type = "x_storage_token", + principal = %MaskedAccessKey(&cred.parent_user), + result = "success", + "Keystone token authentication completed" + ); Ok(Some(cred)) } Err(e) => { - error!("Keystone Swift token authentication failed: {}", e); + error!( + component = LOG_COMPONENT_AUTH, + subsystem = LOG_SUBSYSTEM_KEYSTONE, + event = "keystone_token_auth", + token_type = "x_storage_token", + result = "failed", + error = %e, + "Keystone token authentication completed" + ); Err(s3_error!(InvalidToken, "Invalid Keystone token: {}", e)) } }; diff --git a/rustfs/src/capacity/service.rs b/rustfs/src/capacity/service.rs index 7d64d3173..07796043a 100644 --- a/rustfs/src/capacity/service.rs +++ b/rustfs/src/capacity/service.rs @@ -22,6 +22,9 @@ use std::sync::Arc; use std::time::Instant; use tracing::{debug, info, warn}; +const LOG_COMPONENT_CAPACITY: &str = "capacity"; +const LOG_SUBSYSTEM_CAPACITY: &str = "capacity"; + pub fn capacity_disk_ref(endpoint: impl Into, drive_path: impl Into) -> CapacityDiskRef { CapacityDiskRef { endpoint: endpoint.into(), @@ -50,7 +53,15 @@ async fn refresh_admin_disks_with_subset_fallback( match scan::refresh_capacity_with_scope(refresh_disks.clone(), dirty_subset).await { Ok(update) => Ok(update), Err(err) if dirty_subset => { - warn!("Dirty-subset capacity refresh failed: {}. Retrying full-disk refresh for recovery", err); + warn!( + component = LOG_COMPONENT_CAPACITY, + subsystem = LOG_SUBSYSTEM_CAPACITY, + event = "capacity_refresh_retry", + scope = "dirty_subset", + fallback_scope = "full_disk", + error = %err, + "Capacity refresh failed and will retry with full-disk scope" + ); scan::refresh_capacity_with_scope(all_disks, false).await } Err(err) => Err(err), @@ -133,13 +144,27 @@ pub async fn resolve_admin_used_capacity(disks: &[rustfs_madmin::Disk], fallback Ok(update) => { let elapsed = start.elapsed(); debug!( - "Foreground capacity refresh completed in {:?} (files={}, estimated={})", - elapsed, update.file_count, update.is_estimated + component = LOG_COMPONENT_CAPACITY, + subsystem = LOG_SUBSYSTEM_CAPACITY, + event = "capacity_refresh_completed", + mode = "foreground", + duration_ms = elapsed.as_millis() as u64, + file_count = update.file_count, + is_estimated = update.is_estimated, + "Capacity refresh completed" ); update.total_used } Err(err) => { - warn!("Foreground capacity refresh failed: {}, using cached value", err); + warn!( + component = LOG_COMPONENT_CAPACITY, + subsystem = LOG_SUBSYSTEM_CAPACITY, + event = "capacity_refresh_failed", + mode = "foreground", + fallback = "cached_value", + error = %err, + "Capacity refresh failed" + ); record_capacity_cache_served("stale"); cached.total_used } @@ -148,17 +173,40 @@ pub async fn resolve_admin_used_capacity(disks: &[rustfs_madmin::Disk], fallback record_capacity_cache_served("stale"); debug!( - "Using stale cached capacity: {} bytes (age: {:?}, source: {:?}, files={}, estimated={}, needs_update={}, blocking={})", - cached.total_used, cache_age, cached.source, cached.file_count, cached.is_estimated, needs_update, should_block + component = LOG_COMPONENT_CAPACITY, + subsystem = LOG_SUBSYSTEM_CAPACITY, + event = "capacity_cache_served", + cache_state = "stale", + total_used = cached.total_used, + age_ms = cache_age.as_millis() as u64, + source = ?cached.source, + file_count = cached.file_count, + is_estimated = cached.is_estimated, + needs_update, + blocking = should_block, + "Served cached capacity" ); record_capacity_refresh_request("background", capacity_manager::DataSource::Scheduled.as_metric_label()); if spawn_refresh_if_needed_admin_disks(capacity_manager.clone(), capacity_manager::DataSource::Scheduled, disks, true) .await { - debug!("Background capacity update started"); + debug!( + component = LOG_COMPONENT_CAPACITY, + subsystem = LOG_SUBSYSTEM_CAPACITY, + event = "capacity_refresh_background", + state = "started", + "Background capacity refresh state changed" + ); } else { - debug!("Background update already in progress, skipping spawn"); + debug!( + component = LOG_COMPONENT_CAPACITY, + subsystem = LOG_SUBSYSTEM_CAPACITY, + event = "capacity_refresh_background", + state = "skipped", + reason = "already_running", + "Background capacity refresh state changed" + ); } return cached.total_used; @@ -171,15 +219,27 @@ pub async fn resolve_admin_used_capacity(disks: &[rustfs_madmin::Disk], fallback Ok(update) => { let elapsed = start.elapsed(); info!( - "Initial capacity calculation completed: {} bytes in {:?} (files={}, estimated={})", - update.total_used, elapsed, update.file_count, update.is_estimated + component = LOG_COMPONENT_CAPACITY, + subsystem = LOG_SUBSYSTEM_CAPACITY, + event = "capacity_refresh_completed", + mode = "initial", + total_used = update.total_used, + duration_ms = elapsed.as_millis() as u64, + file_count = update.file_count, + is_estimated = update.is_estimated, + "Capacity refresh completed" ); update.total_used } Err(err) => { warn!( - "Failed to calculate data directory used capacity: {}, falling back to disk used capacity", - err + component = LOG_COMPONENT_CAPACITY, + subsystem = LOG_SUBSYSTEM_CAPACITY, + event = "capacity_refresh_failed", + mode = "initial", + fallback = "disk_used_capacity", + error = %err, + "Capacity refresh failed" ); record_capacity_cache_served("fallback"); record_capacity_scan_mode("fallback"); @@ -195,25 +255,56 @@ pub async fn resolve_admin_used_capacity(disks: &[rustfs_madmin::Disk], fallback } pub async fn init_capacity_management_for_local_disks() { - info!("Initializing capacity management system..."); + info!( + component = LOG_COMPONENT_CAPACITY, + subsystem = LOG_SUBSYSTEM_CAPACITY, + event = "capacity_manager_state", + state = "initializing", + "Capacity manager state changed" + ); let disks = rustfs_ecstore::store::all_local_disk().await; if disks.is_empty() { - warn!("No local disks found, capacity management will not run"); + warn!( + component = LOG_COMPONENT_CAPACITY, + subsystem = LOG_SUBSYSTEM_CAPACITY, + event = "capacity_manager_state", + state = "skipped", + reason = "no_local_disks", + "Capacity manager state changed" + ); return; } - info!("Found {} local disk(s)", disks.len()); + info!( + component = LOG_COMPONENT_CAPACITY, + subsystem = LOG_SUBSYSTEM_CAPACITY, + event = "capacity_manager_disks_detected", + disk_count = disks.len(), + "Detected local disks for capacity management" + ); let disk_refs = disks .iter() .map(|ds| capacity_disk_ref(ds.endpoint().to_string(), ds.to_string())) .collect(); - info!("Starting background capacity update task..."); + info!( + component = LOG_COMPONENT_CAPACITY, + subsystem = LOG_SUBSYSTEM_CAPACITY, + event = "capacity_manager_state", + state = "starting_background_task", + "Capacity manager state changed" + ); capacity_manager::start_background_task(disk_refs).await; - info!("Capacity management system initialized successfully"); + info!( + component = LOG_COMPONENT_CAPACITY, + subsystem = LOG_SUBSYSTEM_CAPACITY, + event = "capacity_manager_state", + state = "initialized", + "Capacity manager state changed" + ); } pub async fn get_cached_capacity_with_metrics() -> Option<(u64, &'static str)> { diff --git a/rustfs/src/embedded.rs b/rustfs/src/embedded.rs index 5d99e3e69..dcce405ec 100644 --- a/rustfs/src/embedded.rs +++ b/rustfs/src/embedded.rs @@ -85,6 +85,9 @@ use std::sync::{ use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, warn}; +const LOG_COMPONENT_EMBEDDED: &str = "embedded"; +const LOG_SUBSYSTEM_EMBEDDED: &str = "embedded"; + /// Tracks whether a server has been started in this process. static SERVER_STARTED: AtomicBool = AtomicBool::new(false); @@ -299,7 +302,14 @@ impl RustFSServerBuilder { // Crypto provider. if let Err(err) = default_provider().install_default() { - debug!("Ignoring crypto provider installation error: {err:?}"); + debug!( + component = LOG_COMPONENT_EMBEDDED, + subsystem = LOG_SUBSYSTEM_EMBEDDED, + event = "crypto_provider_state", + state = "already_installed", + error = ?err, + "Embedded crypto provider state changed" + ); } // Trusted proxies. @@ -370,7 +380,14 @@ impl RustFSServerBuilder { let store = match ECStore::new(server_addr, endpoint_pools.clone(), ctx.clone()).await { Ok(store) => store, Err(e) => { - error!("ECStore::new {:?}", e); + error!( + component = LOG_COMPONENT_EMBEDDED, + subsystem = LOG_SUBSYSTEM_EMBEDDED, + event = "embedded_storage_init_failed", + stage = "ecstore_new", + error = ?e, + "Embedded storage initialization failed" + ); shutdown_embedded_server(); return Err(ServerError::Init(format!("ECStore: {e}"))); } @@ -387,7 +404,15 @@ impl RustFSServerBuilder { shutdown_embedded_server(); return Err(ServerError::Init(format!("init_global_config_sys failed after 15 retries: {e}"))); } - debug!("init_global_config_sys retry {retry}: {e}"); + debug!( + component = LOG_COMPONENT_EMBEDDED, + subsystem = LOG_SUBSYSTEM_EMBEDDED, + event = "embedded_storage_init_retry", + stage = "global_config_sys", + retry, + error = %e, + "Embedded storage initialization retry scheduled" + ); tokio::time::sleep(std::time::Duration::from_secs(1)).await; } readiness.mark_stage(SystemStage::StorageReady); @@ -397,7 +422,14 @@ impl RustFSServerBuilder { // KMS (optional, non-fatal for embedded). if let Err(e) = init_kms_system(&config).await { - warn!("KMS initialization skipped: {e}"); + warn!( + component = LOG_COMPONENT_EMBEDDED, + subsystem = LOG_SUBSYSTEM_EMBEDDED, + event = "embedded_optional_service_skipped", + service = "kms", + error = %e, + "Embedded optional service initialization skipped" + ); } // Buffer profiles. @@ -408,7 +440,14 @@ impl RustFSServerBuilder { // Audit (non-fatal). if let Err(e) = start_audit_system().await { - warn!("Audit system: {e}"); + warn!( + component = LOG_COMPONENT_EMBEDDED, + subsystem = LOG_SUBSYSTEM_EMBEDDED, + event = "embedded_optional_service_skipped", + service = "audit", + error = %e, + "Embedded optional service initialization skipped" + ); } // Bucket listing for metadata + notification init. @@ -445,7 +484,14 @@ impl RustFSServerBuilder { // Notification system. if let Err(e) = new_global_notification_sys(endpoint_pools.clone()).await { - warn!("notification system: {e}"); + warn!( + component = LOG_COMPONENT_EMBEDDED, + subsystem = LOG_SUBSYSTEM_EMBEDDED, + event = "embedded_optional_service_skipped", + service = "notification", + error = %e, + "Embedded optional service initialization skipped" + ); } if iam_bootstrap == IamBootstrapDisposition::ReadyInline { @@ -471,6 +517,10 @@ impl RustFSServerBuilder { info!( target: "rustfs::embedded", + component = LOG_COMPONENT_EMBEDDED, + subsystem = LOG_SUBSYSTEM_EMBEDDED, + event = "embedded_server_state", + state = "ready", "RustFS embedded server ready at http://{}", server.endpoint_address() ); @@ -541,7 +591,14 @@ impl RustFSServer { } async fn do_shutdown(&mut self) { - info!(target: "rustfs::embedded", "Shutting down embedded RustFS server..."); + info!( + target: "rustfs::embedded", + component = LOG_COMPONENT_EMBEDDED, + subsystem = LOG_SUBSYSTEM_EMBEDDED, + event = "embedded_server_state", + state = "stopping", + "Embedded server state changed" + ); // Cancel background services. self.cancel_token.cancel(); @@ -551,7 +608,14 @@ impl RustFSServer { // Stop the audit system. if let Err(e) = stop_audit_system().await { - warn!("Failed to stop audit system during shutdown: {e}"); + warn!( + component = LOG_COMPONENT_EMBEDDED, + subsystem = LOG_SUBSYSTEM_EMBEDDED, + event = "embedded_shutdown_cleanup_failed", + service = "audit", + error = %e, + "Embedded shutdown cleanup failed" + ); } // Signal HTTP server to stop. @@ -563,10 +627,25 @@ impl RustFSServer { if let Some(ref dir) = self.temp_dir && let Err(e) = tokio::fs::remove_dir_all(dir).await { - warn!("Failed to clean up temp dir {}: {e}", dir.display()); + warn!( + component = LOG_COMPONENT_EMBEDDED, + subsystem = LOG_SUBSYSTEM_EMBEDDED, + event = "embedded_shutdown_cleanup_failed", + service = "temp_dir", + path = %dir.display(), + error = %e, + "Embedded shutdown cleanup failed" + ); } - info!(target: "rustfs::embedded", "Embedded RustFS server stopped."); + info!( + target: "rustfs::embedded", + component = LOG_COMPONENT_EMBEDDED, + subsystem = LOG_SUBSYSTEM_EMBEDDED, + event = "embedded_server_state", + state = "stopped", + "Embedded server state changed" + ); } } diff --git a/rustfs/src/init.rs b/rustfs/src/init.rs index 520a1d682..06c797735 100644 --- a/rustfs/src/init.rs +++ b/rustfs/src/init.rs @@ -28,15 +28,30 @@ use std::env; use std::io::Error; use tracing::{debug, error, info, instrument, warn}; +const LOG_COMPONENT_INIT: &str = "init"; +const LOG_SUBSYSTEM_STARTUP: &str = "startup"; +const LOG_SUBSYSTEM_UPDATE: &str = "update_check"; +const LOG_SUBSYSTEM_NOTIFICATION: &str = "notification"; +const LOG_SUBSYSTEM_KMS: &str = "kms"; +const LOG_SUBSYSTEM_BUFFER: &str = "buffer_profile"; +const LOG_SUBSYSTEM_AUTOTUNER: &str = "autotuner"; +const LOG_SUBSYSTEM_PROTOCOL: &str = "protocol"; + #[instrument] pub fn print_server_info() { let current_year = jiff::Zoned::now().year(); - // Use custom macros to print server information - info!("RustFS Object Storage Server"); - info!("Copyright: 2024-{} RustFS, Inc", current_year); - info!("License: Apache-2.0 https://www.apache.org/licenses/LICENSE-2.0"); - info!("Version: {}", version::get_version()); - info!("Docs: https://rustfs.com/docs/"); + info!( + target: "rustfs::init", + event = "server_identity", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_STARTUP, + product = "RustFS Object Storage Server", + version = %version::get_version(), + copyright_year = current_year, + license = "Apache-2.0", + docs_url = "https://rustfs.com/docs/", + "Server identity loaded" + ); } /// Initialize the asynchronous update check system. @@ -63,28 +78,62 @@ pub fn init_update_check() { if result.update_available { if let Some(latest) = &result.latest_version { info!( - "🚀 Version check: New version available: {} -> {} (current: {})", - result.current_version, latest.version, result.current_version + target: "rustfs::init", + event = "update_check_result", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_UPDATE, + result = "update_available", + current_version = %result.current_version, + latest_version = %latest.version, + has_release_notes = latest.release_notes.is_some(), + download_url = latest.download_url.as_deref().unwrap_or_default(), + "Update check completed" ); - if let Some(notes) = &latest.release_notes { - info!("📝 Release notes: {}", notes); - } - if let Some(url) = &latest.download_url { - info!("🔗 Download URL: {}", url); - } } } else { - debug!("✅ Version check: Current version is up to date: {}", result.current_version); + debug!( + target: "rustfs::init", + event = "update_check_result", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_UPDATE, + result = "up_to_date", + current_version = %result.current_version, + "Update check completed" + ); } } Ok(Err(UpdateCheckError::HttpError(e))) => { - debug!("Version check: network error (this is normal): {}", e); + debug!( + target: "rustfs::init", + event = "update_check_result", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_UPDATE, + result = "http_error", + error = %e, + "Update check skipped" + ); } Ok(Err(e)) => { - debug!("Version check: failed (this is normal): {}", e); + debug!( + target: "rustfs::init", + event = "update_check_result", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_UPDATE, + result = "failed", + error = %e, + "Update check failed" + ); } Err(_) => { - debug!("Version check: timeout after 30 seconds (this is normal)"); + debug!( + target: "rustfs::init", + event = "update_check_result", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_UPDATE, + result = "timeout", + timeout_secs = 30, + "Update check timed out" + ); } } }); @@ -113,51 +162,110 @@ pub async fn add_bucket_notification_configuration(buckets: Vec) { .map(|r| r.as_str()) .unwrap_or_else(|| { warn!( - "Global region is not set; attempting notification configuration for all buckets using default region '{}'.", - RUSTFS_REGION + target: "rustfs::init", + event = "notification_region_fallback", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_NOTIFICATION, + fallback_region = RUSTFS_REGION, + "Notification configuration falling back to default region" ); RUSTFS_REGION }); for bucket in buckets.iter() { let has_notification_config = metadata_sys::get_notification_config(bucket).await.unwrap_or_else(|err| { - warn!("get_notification_config err {:?}", err); + warn!( + target: "rustfs::init", + event = "notification_config_load_failed", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_NOTIFICATION, + bucket = %bucket, + error = ?err, + "Failed to load bucket notification configuration" + ); None }); match has_notification_config { Some(cfg) => { info!( - target: "rustfs::main::add_bucket_notification_configuration", + target: "rustfs::init", + event = "notification_config_loaded", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_NOTIFICATION, bucket = %bucket, - "Bucket '{}' has existing notification configuration: {:?}", bucket, cfg); + queue_configuration_count = cfg.queue_configurations.as_ref().map_or(0, Vec::len), + topic_configuration_count = cfg.topic_configurations.as_ref().map_or(0, Vec::len), + lambda_configuration_count = cfg.lambda_function_configurations.as_ref().map_or(0, Vec::len), + "Loaded bucket notification configuration" + ); let mut event_rules = Vec::new(); if let Err(e) = process_queue_configurations(&mut event_rules, cfg.queue_configurations.clone(), arn_to_target_id) { - error!("Failed to parse queue notification config for bucket '{}': {:?}", bucket, e); + error!( + target: "rustfs::init", + event = "notification_config_parse_failed", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_NOTIFICATION, + bucket = %bucket, + config_type = "queue", + error = ?e, + "Failed to parse bucket notification configuration" + ); } if let Err(e) = process_topic_configurations(&mut event_rules, cfg.topic_configurations.clone(), arn_to_target_id) { - error!("Failed to parse topic notification config for bucket '{}': {:?}", bucket, e); + error!( + target: "rustfs::init", + event = "notification_config_parse_failed", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_NOTIFICATION, + bucket = %bucket, + config_type = "topic", + error = ?e, + "Failed to parse bucket notification configuration" + ); } if let Err(e) = process_lambda_configurations(&mut event_rules, cfg.lambda_function_configurations.clone(), arn_to_target_id) { - error!("Failed to parse lambda notification config for bucket '{}': {:?}", bucket, e); + error!( + target: "rustfs::init", + event = "notification_config_parse_failed", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_NOTIFICATION, + bucket = %bucket, + config_type = "lambda", + error = ?e, + "Failed to parse bucket notification configuration" + ); } if let Err(e) = notifier_global::add_event_specific_rules(bucket, region, &event_rules) .await .map_err(|e| s3_error!(InternalError, "Failed to add rules: {e}")) { - error!("Failed to add rules for bucket '{}': {:?}", bucket, e); + error!( + target: "rustfs::init", + event = "notification_rules_registration_failed", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_NOTIFICATION, + bucket = %bucket, + region, + error = ?e, + "Failed to register bucket notification rules" + ); } } None => { info!( - target: "rustfs::main::add_bucket_notification_configuration", + target: "rustfs::init", + event = "notification_config_missing", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_NOTIFICATION, bucket = %bucket, - "Bucket '{}' has no existing notification configuration.", bucket); + "Bucket notification configuration not found" + ); } } } @@ -266,7 +374,15 @@ async fn configure_and_start_kms( .await .map_err(|e| Error::other(format!("Failed to start KMS: {e}")))?; - info!("KMS service configured and started successfully from {}", config_source); + info!( + target: "rustfs::init", + event = "kms_service_state", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_KMS, + state = "started", + config_source, + "KMS service state changed" + ); Ok(()) } @@ -287,7 +403,15 @@ pub async fn init_kms_system(config: &config::Config) -> std::io::Result<()> { // If KMS is enabled in configuration, configure and start the service if config.kms_enable { - info!("KMS is enabled via command line, configuring and starting service..."); + info!( + target: "rustfs::init", + event = "kms_service_state", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_KMS, + state = "configuring", + config_source = "command_line", + "KMS service state changed" + ); // Create KMS configuration from command line options let kms_config = match config.kms_backend.as_str() { @@ -300,22 +424,49 @@ pub async fn init_kms_system(config: &config::Config) -> std::io::Result<()> { configure_and_start_kms(&service_manager, kms_config, "command line options").await?; } else { // Try to load persisted KMS configuration from cluster storage - info!("Attempting to load persisted KMS configuration from cluster storage..."); + info!( + target: "rustfs::init", + event = "kms_persisted_config_lookup", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_KMS, + state = "loading", + "Loading persisted KMS configuration" + ); if let Some(persisted_config) = admin::handlers::kms_dynamic::load_kms_config().await { - info!("Found persisted KMS configuration, attempting to configure and start service..."); + info!( + target: "rustfs::init", + event = "kms_persisted_config_lookup", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_KMS, + state = "found", + "Loaded persisted KMS configuration" + ); // Configure the KMS service with persisted config match configure_and_start_kms(&service_manager, persisted_config, "persisted configuration").await { - Ok(()) => { - info!("KMS service configured and started successfully from persisted configuration"); - } + Ok(()) => {} Err(e) => { - warn!("Failed to configure KMS with persisted configuration: {}", e); + warn!( + target: "rustfs::init", + event = "kms_service_state", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_KMS, + state = "persisted_config_failed", + error = %e, + "KMS service state changed" + ); } } } else { - info!("No persisted KMS configuration found. KMS is ready for dynamic configuration via API."); + info!( + target: "rustfs::init", + event = "kms_persisted_config_lookup", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_KMS, + state = "not_found", + "No persisted KMS configuration found" + ); } } @@ -341,11 +492,28 @@ pub fn init_buffer_profile_system(config: &config::Config) { // Whether buffer profiling is disabled or not, it is enabled by default, unless the user explicitly sets '--buffer-profile-disable' or 'RUSTFS_BUFFER_PROFILE_DISABLE=true' if config.buffer_profile_disable { // User explicitly disabled buffer profiling - use GeneralPurpose profile in disabled mode - info!("Buffer profiling disabled via --buffer-profile-disable, using GeneralPurpose profile"); + info!( + target: "rustfs::init", + event = "buffer_profile_state", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_BUFFER, + state = "disabled", + profile = "GeneralPurpose", + reason = "flag_override", + "Buffer profile state changed" + ); set_buffer_profile_enabled(false); } else { // Enabled by default: use configured workload profile - info!("Buffer profiling enabled with profile: {}", config.buffer_profile); + info!( + target: "rustfs::init", + event = "buffer_profile_state", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_BUFFER, + state = "configuring", + profile = %config.buffer_profile, + "Buffer profile state changed" + ); // Parse the workload profile from configuration string // Support a custom profile when buffer_profile is set to "custom"; @@ -358,8 +526,14 @@ pub fn init_buffer_profile_system(config: &config::Config) { let default_unknown = get_env_usize(ENV_RUSTFS_BUFFER_DEFAULT_SIZE, DEFAULT_BUFFER_UNKNOWN_SIZE); info!( - "Creating custom buffer profile: min={}, max={}, default={}", - min_size, max_size, default_unknown + target: "rustfs::init", + event = "buffer_profile_custom", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_BUFFER, + min_size, + max_size, + default_size = default_unknown, + "Creating custom buffer profile" ); WorkloadProfile::custom( min_size, @@ -376,20 +550,47 @@ pub fn init_buffer_profile_system(config: &config::Config) { }; // Log the selected profile for operational visibility - info!("Active buffer profile: {:?}", profile); + info!( + target: "rustfs::init", + event = "buffer_profile_selected", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_BUFFER, + profile = ?profile, + "Selected buffer profile" + ); // Create and validate buffer configuration let mut buffer_config = RustFSBufferConfig::new(profile); if let Err(e) = buffer_config.validate() { - warn!("Buffer configuration validation failed: {}. Falling back to GeneralPurpose profile.", e); + warn!( + target: "rustfs::init", + event = "buffer_profile_validation_failed", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_BUFFER, + error = %e, + fallback_profile = DEFAULT_BUFFER_PROFILE, + "Buffer profile validation failed" + ); // Fall back to a known-good profile to avoid installing an invalid configuration let fallback_profile = WorkloadProfile::from_name(DEFAULT_BUFFER_PROFILE); - info!("Using fallback buffer profile: {:?}", fallback_profile); + info!( + target: "rustfs::init", + event = "buffer_profile_fallback", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_BUFFER, + profile = ?fallback_profile, + "Using fallback buffer profile" + ); let fallback_config = RustFSBufferConfig::new(fallback_profile); if let Err(e2) = fallback_config.validate() { error!( - "Fallback buffer configuration validation failed: {}. Aborting buffer profiling initialization.", - e2 + target: "rustfs::init", + event = "buffer_profile_validation_failed", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_BUFFER, + error = %e2, + fallback_profile = DEFAULT_BUFFER_PROFILE, + "Fallback buffer profile validation failed" ); panic!("Failed to initialize a valid RustFS buffer configuration"); } @@ -397,7 +598,15 @@ pub fn init_buffer_profile_system(config: &config::Config) { } // Log the workload profile name - info!("Workload profile: {}", buffer_config.workload_name()); + let workload_name = buffer_config.workload_name(); + info!( + target: "rustfs::init", + event = "buffer_profile_workload", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_BUFFER, + workload = %workload_name, + "Buffer profile workload selected" + ); // Initialize the global buffer configuration init_global_buffer_config(buffer_config); @@ -405,7 +614,15 @@ pub fn init_buffer_profile_system(config: &config::Config) { // Enable buffer profiling globally set_buffer_profile_enabled(true); - info!("Buffer profiling system initialized successfully"); + info!( + target: "rustfs::init", + event = "buffer_profile_state", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_BUFFER, + state = "initialized", + workload = %workload_name, + "Buffer profile state changed" + ); } } @@ -441,9 +658,26 @@ where tokio::spawn(async move { if let Err(e) = server.await { - error!("{} server error: {}", protocol_name, e); + error!( + target: "rustfs::init", + event = "protocol_server_state", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_PROTOCOL, + protocol = protocol_name, + state = "runtime_failed", + error = %e, + "Protocol server state changed" + ); } - info!("{} server shutdown completed", protocol_name); + info!( + target: "rustfs::init", + event = "protocol_server_state", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_PROTOCOL, + protocol = protocol_name, + state = "stopped", + "Protocol server state changed" + ); }); shutdown_tx @@ -463,7 +697,14 @@ pub async fn init_auto_tuner(ctx: tokio_util::sync::CancellationToken) { let autotuner_enabled = rustfs_utils::get_env_bool("RUSTFS_AUTOTUNER_ENABLED", false); if autotuner_enabled { - info!(target: "rustfs::main::run", "Starting auto-tuner for performance optimization"); + info!( + target: "rustfs::init", + event = "autotuner_state", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_AUTOTUNER, + state = "starting", + "Auto-tuner state changed" + ); let config = TunerConfig::default(); let manager = get_concurrency_manager(); @@ -475,23 +716,59 @@ pub async fn init_auto_tuner(ctx: tokio_util::sync::CancellationToken) { loop { tokio::select! { _ = ctx.cancelled() => { - info!(target: "rustfs::autotuner", "Auto-tuner shutting down"); + info!( + target: "rustfs::init", + event = "autotuner_state", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_AUTOTUNER, + state = "stopping", + "Auto-tuner state changed" + ); break; } _ = tokio::time::sleep(tokio::time::Duration::from_secs(60)) => { if let Err(e) = tuner.tune().await { - error!(target: "rustfs::autotuner", "Auto-tuner iteration failed: {}", e); + error!( + target: "rustfs::init", + event = "autotuner_iteration", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_AUTOTUNER, + result = "failed", + error = %e, + "Auto-tuner iteration completed" + ); } else { - debug!(target: "rustfs::autotuner", "Auto-tuner iteration completed"); + debug!( + target: "rustfs::init", + event = "autotuner_iteration", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_AUTOTUNER, + result = "ok", + "Auto-tuner iteration completed" + ); } } } } }); - info!(target: "rustfs::main::run", "Auto-tuner started successfully"); + info!( + target: "rustfs::init", + event = "autotuner_state", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_AUTOTUNER, + state = "started", + "Auto-tuner state changed" + ); } else { - info!(target: "rustfs::main::run", "Auto-tuner disabled (set RUSTFS_AUTOTUNER_ENABLED=true to enable)"); + info!( + target: "rustfs::init", + event = "autotuner_state", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_AUTOTUNER, + state = "disabled", + "Auto-tuner state changed" + ); } } diff --git a/rustfs/src/main.rs b/rustfs/src/main.rs index ad78e1c93..8aa2eda1b 100644 --- a/rustfs/src/main.rs +++ b/rustfs/src/main.rs @@ -90,7 +90,10 @@ const EVENT_SERVER_CONFIG_SANITIZED: &str = "server_config_sanitized"; const EVENT_SERVER_STARTING: &str = "server_starting"; const EVENT_SERVER_RUNTIME_FAILED: &str = "server_runtime_failed"; const EVENT_ACTION_CREDENTIALS_INITIALIZATION_FAILED: &str = "action_credentials_initialization_failed"; +const EVENT_OBSERVABILITY_GUARD_SET: &str = "observability_guard_set"; +const EVENT_CRYPTO_PROVIDER_STATE: &str = "crypto_provider_state"; const EVENT_ENDPOINT_PARSING_STARTED: &str = "endpoint_parsing_started"; +const EVENT_STARTUP_STORAGE_STAGE: &str = "startup_storage_stage"; const EVENT_STORAGE_POOL_FORMATTING: &str = "storage_pool_formatting"; const EVENT_STORAGE_POOL_HOST_RISK: &str = "storage_pool_host_risk"; const EVENT_PROTOCOL_SYSTEM_STATE: &str = "protocol_system_state"; @@ -188,7 +191,14 @@ async fn async_main() -> Result<()> { // Store in global storage match set_global_guard(guard).map_err(Error::other) { Ok(_) => { - debug!(target: "rustfs::main", "Global observability guard set successfully."); + debug!( + target: "rustfs::main", + event = EVENT_OBSERVABILITY_GUARD_SET, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + result = "ok", + "Stored global observability guard" + ); } Err(e) => { error!( @@ -249,7 +259,15 @@ async fn async_main() -> Result<()> { // Make sure to use a modern encryption suite if default_provider().install_default().is_err() { // A crypto provider is already installed (e.g. by the host process); this is fine. - debug!("rustls crypto provider already installed, skipping aws-lc-rs default install"); + debug!( + target: "rustfs::main", + event = EVENT_CRYPTO_PROVIDER_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + provider = "aws_lc_rs", + state = "already_installed", + "Rustls crypto provider state checked" + ); } // Initialize TLS outbound material (root CAs, mTLS identity) if configured. // Server-side TLS acceptor is built separately inside start_http_server(). @@ -415,6 +433,11 @@ async fn run(config: rustfs::config::Config) -> Result<()> { // Initialize the local disk debug!( target: "rustfs::main::run", + event = EVENT_STARTUP_STORAGE_STAGE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STORAGE, + stage = "local_disk_initialization", + state = "starting", "starting local disk initialization" ); init_local_disks(endpoint_pools.clone()) @@ -505,6 +528,11 @@ async fn run(config: rustfs::config::Config) -> Result<()> { // 2. Start Storage Engine (ECStore) debug!( target: "rustfs::main::run", + event = EVENT_STARTUP_STORAGE_STAGE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STORAGE, + stage = "ecstore_initialization", + state = "starting", "starting ECStore initialization" ); let store = ECStore::new(server_addr, endpoint_pools.clone(), ctx.clone()) @@ -524,7 +552,17 @@ async fn run(config: rustfs::config::Config) -> Result<()> { // // Initialize global configuration system let mut retry_count = 0; while let Err(e) = ecconfig::init_global_config_sys(store.clone()).await { - error!("ecstore config::init_global_config_sys failed {:?}", e); + error!( + target: "rustfs::main::run", + event = EVENT_STARTUP_STORAGE_STAGE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STORAGE, + stage = "global_config_initialization", + state = "retrying", + retry_count = retry_count + 1, + error = ?e, + "Startup storage stage failed" + ); // TODO: check error type retry_count += 1; if retry_count > 15 { @@ -860,7 +898,17 @@ async fn run(config: rustfs::config::Config) -> Result<()> { } if !enable_heal && !enable_scanner { - debug!(target: "rustfs::main::run","Both scanner and heal are disabled, skipping AHM service initialization"); + debug!( + target: "rustfs::main::run", + event = EVENT_BACKGROUND_SERVICES_CONFIGURED, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + enable_scanner = false, + enable_heal = false, + ahm_state = "skipped", + reason = "disabled", + "Background services configured" + ); } // print server info @@ -916,7 +964,15 @@ async fn run(config: rustfs::config::Config) -> Result<()> { ) .await; - info!(target: "rustfs::main::run","server is stopped state: {:?}", state_manager.current_state()); + info!( + target: "rustfs::main::run", + event = EVENT_SERVER_SHUTDOWN_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + state = ?state_manager.current_state(), + result = "stopped", + "Server shutdown state changed" + ); Ok(()) } diff --git a/rustfs/src/profiling.rs b/rustfs/src/profiling.rs index c2d170cee..cafc13982 100644 --- a/rustfs/src/profiling.rs +++ b/rustfs/src/profiling.rs @@ -79,6 +79,9 @@ mod linux_impl { use tokio_util::sync::CancellationToken; use tracing::{debug, info, warn}; + const LOG_COMPONENT_PROFILING: &str = "profiling"; + const LOG_SUBSYSTEM_PROFILING: &str = "profiling"; + static CPU_CONT_GUARD: OnceLock>>>> = OnceLock::new(); static PROFILING_CANCEL_TOKEN: OnceLock = OnceLock::new(); @@ -95,7 +98,15 @@ mod linux_impl { let dir = get_env_str(ENV_OUTPUT_DIR, DEFAULT_OUTPUT_DIR); let p = PathBuf::from(dir); if let Err(e) = create_dir_all(&p) { - warn!("profiling: create output dir {} failed: {}, fallback to current dir", p.display(), e); + warn!( + component = LOG_COMPONENT_PROFILING, + subsystem = LOG_SUBSYSTEM_PROFILING, + event = "profiling_output_dir_fallback", + path = %p.display(), + error = %e, + fallback = ".", + "Profiling output directory fallback applied" + ); return PathBuf::from("."); } p @@ -130,7 +141,14 @@ mod linux_impl { let report = guard.report().build().map_err(|e| format!("build report failed: {e}"))?; let out = output_dir().join(format!("cpu_profile_{}.pb", ts())); write_pprof_report_pb(&report, &out)?; - info!("CPU profile exported: {}", out.display()); + info!( + component = LOG_COMPONENT_PROFILING, + subsystem = LOG_SUBSYSTEM_PROFILING, + event = "profiling_dump_exported", + profile_type = "cpu", + path = %out.display(), + "Profiling dump exported" + ); Ok(out) } @@ -139,7 +157,14 @@ mod linux_impl { if let Some(cell) = CPU_CONT_GUARD.get() { let guard_slot = cell.lock().await; if let Some(ref guard) = *guard_slot { - debug!("profiling: using continuous profiler guard for CPU dump"); + debug!( + component = LOG_COMPONENT_PROFILING, + subsystem = LOG_SUBSYSTEM_PROFILING, + event = "profiling_dump_source", + profile_type = "cpu", + source = "continuous_guard", + "Using continuous CPU profiling guard for dump" + ); return dump_cpu_with_guard(guard).await; } } @@ -168,7 +193,14 @@ mod linux_impl { let bytes = prof_ctl.dump_pprof().map_err(|e| format!("dump pprof failed: {e}"))?; f.write_all(&bytes).map_err(|e| format!("write file failed: {e}"))?; - info!("Memory profile exported: {}", out.display()); + info!( + component = LOG_COMPONENT_PROFILING, + subsystem = LOG_SUBSYSTEM_PROFILING, + event = "profiling_dump_exported", + profile_type = "memory", + path = %out.display(), + "Profiling dump exported" + ); Ok(out) } @@ -229,7 +261,14 @@ mod linux_impl { let cell = CPU_CONT_GUARD.get_or_init(|| Arc::new(Mutex::new(None))).clone(); let mut slot = cell.lock().await; if slot.is_some() { - warn!("profiling: continuous CPU guard already running"); + warn!( + component = LOG_COMPONENT_PROFILING, + subsystem = LOG_SUBSYSTEM_PROFILING, + event = "profiling_state", + profile_type = "cpu_continuous", + state = "already_running", + "Profiling state changed" + ); return; } match pprof::ProfilerGuardBuilder::default() @@ -239,20 +278,53 @@ mod linux_impl { { Ok(guard) => { *slot = Some(guard); - info!(freq = freq_hz, "start continuous CPU profiling"); + info!( + component = LOG_COMPONENT_PROFILING, + subsystem = LOG_SUBSYSTEM_PROFILING, + event = "profiling_state", + profile_type = "cpu_continuous", + state = "started", + freq_hz, + "Profiling state changed" + ); } - Err(e) => warn!("start continuous CPU profiling failed: {e}"), + Err(e) => warn!( + component = LOG_COMPONENT_PROFILING, + subsystem = LOG_SUBSYSTEM_PROFILING, + event = "profiling_state", + profile_type = "cpu_continuous", + state = "start_failed", + error = %e, + "Profiling state changed" + ), } } // Internal: start periodic CPU sampling loop async fn start_cpu_periodic(freq_hz: i32, interval: Duration, duration: Duration, token: CancellationToken) { - info!(freq = freq_hz, ?interval, ?duration, "start periodic CPU profiling"); + info!( + component = LOG_COMPONENT_PROFILING, + subsystem = LOG_SUBSYSTEM_PROFILING, + event = "profiling_state", + profile_type = "cpu_periodic", + state = "started", + freq_hz, + ?interval, + ?duration, + "Profiling state changed" + ); tokio::spawn(async move { loop { tokio::select! { _ = token.cancelled() => { - info!("periodic CPU profiling task cancelled"); + info!( + component = LOG_COMPONENT_PROFILING, + subsystem = LOG_SUBSYSTEM_PROFILING, + event = "profiling_state", + profile_type = "cpu_periodic", + state = "cancelled", + "Profiling state changed" + ); break; } _ = sleep(interval) => {} @@ -272,7 +344,14 @@ mod linux_impl { tokio::select! { _ = token.cancelled() => { - info!("periodic CPU profiling task cancelled during capture"); + info!( + component = LOG_COMPONENT_PROFILING, + subsystem = LOG_SUBSYSTEM_PROFILING, + event = "profiling_state", + profile_type = "cpu_periodic", + state = "cancelled_during_capture", + "Profiling state changed" + ); break; } _ = sleep(duration) => {} @@ -284,7 +363,14 @@ mod linux_impl { if let Err(e) = write_pprof_report_pb(&report, &out) { warn!("write periodic CPU pprof failed: {e}"); } else { - info!("periodic CPU profile exported: {}", out.display()); + info!( + component = LOG_COMPONENT_PROFILING, + subsystem = LOG_SUBSYSTEM_PROFILING, + event = "profiling_dump_exported", + profile_type = "cpu_periodic", + path = %out.display(), + "Profiling dump exported" + ); } } Err(e) => warn!("periodic CPU report build failed: {e}"), @@ -296,12 +382,27 @@ mod linux_impl { // Internal: start periodic memory dump when jemalloc profiling is active #[cfg(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64"))] async fn start_memory_periodic(interval: Duration, token: CancellationToken) { - info!(?interval, "start periodic memory pprof dump"); + info!( + component = LOG_COMPONENT_PROFILING, + subsystem = LOG_SUBSYSTEM_PROFILING, + event = "profiling_state", + profile_type = "memory_periodic", + state = "started", + ?interval, + "Profiling state changed" + ); tokio::spawn(async move { loop { tokio::select! { _ = token.cancelled() => { - info!("periodic memory profiling task cancelled"); + info!( + component = LOG_COMPONENT_PROFILING, + subsystem = LOG_SUBSYSTEM_PROFILING, + event = "profiling_state", + profile_type = "memory_periodic", + state = "cancelled", + "Profiling state changed" + ); break; } _ = sleep(interval) => {} @@ -328,7 +429,14 @@ mod linux_impl { if let Err(e) = f.write_all(&bytes) { tracing::error!("periodic mem dump write failed: {}", e); } else { - info!("periodic memory profile dumped to {}", out.display()); + info!( + component = LOG_COMPONENT_PROFILING, + subsystem = LOG_SUBSYSTEM_PROFILING, + event = "profiling_dump_exported", + profile_type = "memory_periodic", + path = %out.display(), + "Profiling dump exported" + ); } } Err(e) => tracing::error!("periodic mem dump failed: {}", e), diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index 772b2c990..f458e8595 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -84,9 +84,21 @@ const LOG_COMPONENT_SERVER: &str = "server"; const LOG_SUBSYSTEM_HTTP: &str = "http"; const LOG_SUBSYSTEM_TRANSPORT: &str = "transport"; const LOG_SUBSYSTEM_TLS: &str = "tls"; +const LOG_SUBSYSTEM_STARTUP: &str = "startup"; const EVENT_TLS_HANDSHAKE_FAILED: &str = "tls_handshake_failed"; const EVENT_HTTP_TRANSPORT_CLOSED: &str = "http_transport_closed"; const EVENT_HTTP_TRANSPORT_FAILED: &str = "http_transport_failed"; +const EVENT_SOCKET_FALLBACK: &str = "socket_fallback"; +const EVENT_HTTP_BIND_FAILED: &str = "http_bind_failed"; +const EVENT_HTTP_STARTUP_ENDPOINTS: &str = "http_startup_endpoints"; +const EVENT_HTTP_HOST_ROUTING: &str = "http_host_routing"; +const EVENT_HTTP_COMPRESSION_STATE: &str = "http_compression_state"; +const EVENT_HTTP_TRANSPORT_PARAMETERS: &str = "http_transport_parameters"; +const EVENT_HTTP_ACCEPT_LOOP_STATE: &str = "http_accept_loop_state"; +const EVENT_HTTP_CONNECTION_DRAIN: &str = "http_connection_drain"; +const EVENT_PEER_ADDR_UNAVAILABLE: &str = "peer_addr_unavailable"; +const EVENT_RPC_SIGNATURE_VERIFICATION_FAILED: &str = "rpc_signature_verification_failed"; +const EVENT_GRPC_TRACE_CONTEXT_PROPAGATION_FAILED: &str = "grpc_trace_context_propagation_failed"; static ACTIVE_HTTP_REQUESTS: AtomicU64 = AtomicU64::new(0); @@ -223,7 +235,15 @@ pub async fn start_http_server(config: &config::Config, readiness: Arc s, Err(e) => { - warn!("Failed to create socket for {:?}: {}, falling back to IPv4", server_addr, e); + warn!( + event = EVENT_SOCKET_FALLBACK, + component = LOG_COMPONENT_SERVER, + subsystem = LOG_SUBSYSTEM_STARTUP, + from_addr = %server_addr, + fallback = "ipv4", + error = %e, + "Socket creation fell back to IPv4" + ); let ipv4_addr = SocketAddr::new(std::net::Ipv4Addr::UNSPECIFIED.into(), server_addr.port()); server_addr = ipv4_addr; socket2::Socket::new(socket2::Domain::IPV4, socket2::Type::STREAM, Some(socket2::Protocol::TCP))? @@ -235,7 +255,15 @@ pub async fn start_http_server(config: &config::Config, readiness: Arc ip, None => { - warn!("Unable to obtain local IP address, using fallback IP: {}", local_addr.ip()); + warn!( + event = "local_ip_fallback", + component = LOG_COMPONENT_SERVER, + subsystem = LOG_SUBSYSTEM_STARTUP, + fallback_ip = %local_addr.ip(), + "Falling back to listener IP for startup endpoint logging" + ); local_addr.ip() } }; @@ -357,18 +405,36 @@ pub async fn start_http_server(config: &config::Config, readiness: Arc match res { Ok(conn) => conn, Err(err) => { - error!("error accepting connection: {err}"); + error!( + event = EVENT_HTTP_ACCEPT_LOOP_STATE, + component = LOG_COMPONENT_SERVER, + subsystem = LOG_SUBSYSTEM_TRANSPORT, + state = "accept_failed", + error = %err, + "HTTP accept loop state changed" + ); continue; } }, _ = shutdown_rx.recv() => { - info!("Shutdown signal received in worker thread"); + info!( + event = EVENT_HTTP_ACCEPT_LOOP_STATE, + component = LOG_COMPONENT_SERVER, + subsystem = LOG_SUBSYSTEM_TRANSPORT, + state = "shutdown_signal_received", + "HTTP accept loop state changed" + ); break; } } @@ -527,12 +633,25 @@ pub async fn start_http_server(config: &config::Config, readiness: Arc match res { Ok(conn) => conn, Err(err) => { - error!("error accepting connection: {err}"); + error!( + event = EVENT_HTTP_ACCEPT_LOOP_STATE, + component = LOG_COMPONENT_SERVER, + subsystem = LOG_SUBSYSTEM_TRANSPORT, + state = "accept_failed", + error = %err, + "HTTP accept loop state changed" + ); continue; } }, _ = shutdown_rx.recv() => { - info!("Shutdown signal received in worker thread"); + info!( + event = EVENT_HTTP_ACCEPT_LOOP_STATE, + component = LOG_COMPONENT_SERVER, + subsystem = LOG_SUBSYSTEM_TRANSPORT, + state = "shutdown_signal_received", + "HTTP accept loop state changed" + ); break; } } @@ -581,14 +700,35 @@ pub async fn start_http_server(config: &config::Config, readiness: Arc 0 { - info!(active_connections, "Draining active HTTP connections before shutdown"); + info!( + event = EVENT_HTTP_CONNECTION_DRAIN, + component = LOG_COMPONENT_SERVER, + subsystem = LOG_SUBSYSTEM_TRANSPORT, + state = "draining", + active_connections, + "HTTP connection drain started" + ); } tokio::select! { () = graceful.shutdown() => { - debug!("Gracefully shutdown!"); + debug!( + event = EVENT_HTTP_CONNECTION_DRAIN, + component = LOG_COMPONENT_SERVER, + subsystem = LOG_SUBSYSTEM_TRANSPORT, + state = "completed", + "HTTP connection drain completed" + ); }, () = tokio::time::sleep(Duration::from_secs(10)) => { - warn!(active_connections, "Timed out waiting for HTTP connections to drain during shutdown"); + warn!( + event = EVENT_HTTP_CONNECTION_DRAIN, + component = LOG_COMPONENT_SERVER, + subsystem = LOG_SUBSYSTEM_TRANSPORT, + state = "timeout", + active_connections, + timeout_secs = 10, + "HTTP connection drain timed out" + ); } } }); @@ -726,6 +866,9 @@ fn process_connection( Ok(addr) => Some(RemoteAddr(addr)), Err(e) => { warn!( + event = EVENT_PEER_ADDR_UNAVAILABLE, + component = LOG_COMPONENT_SERVER, + subsystem = LOG_SUBSYSTEM_HTTP, error = %e, "Failed to obtain peer address; policy evaluation may fall back to a default source IP" ); @@ -1084,7 +1227,13 @@ fn handle_connection_error(peer_addr: Option<&str>, err: &(dyn std::error::Error #[allow(clippy::result_large_err)] fn check_auth(req: Request<()>) -> std::result::Result, Status> { verify_rpc_signature(TONIC_RPC_PREFIX, &Method::GET, req.metadata().as_ref()).map_err(|e| { - error!("RPC signature verification failed: {}", e); + error!( + event = EVENT_RPC_SIGNATURE_VERIFICATION_FAILED, + component = LOG_COMPONENT_SERVER, + subsystem = LOG_SUBSYSTEM_HTTP, + error = %e, + "RPC signature verification failed" + ); Status::unauthenticated("No valid auth token") })?; @@ -1099,7 +1248,13 @@ fn check_auth(req: Request<()>) -> std::result::Result, Status> { "Extracted trace context from incoming gRPC metadata" ); if let Err(e) = tracing::Span::current().set_parent(parent_context) { - warn!("Failed to propagate tracing context from gRPC metadata: `{:?}`", e); + warn!( + event = EVENT_GRPC_TRACE_CONTEXT_PROPAGATION_FAILED, + component = LOG_COMPONENT_SERVER, + subsystem = LOG_SUBSYSTEM_HTTP, + error = ?e, + "Failed to propagate tracing context from gRPC metadata" + ); } } Ok(req) diff --git a/rustfs/src/server/tls_material.rs b/rustfs/src/server/tls_material.rs index 44146d7a7..2b3d6d430 100644 --- a/rustfs/src/server/tls_material.rs +++ b/rustfs/src/server/tls_material.rs @@ -41,6 +41,9 @@ use std::time::Duration; use tokio_rustls::TlsAcceptor; use tracing::{debug, info, warn}; +const LOG_COMPONENT_TLS: &str = "tls"; +const LOG_SUBSYSTEM_TLS: &str = "tls_material"; + /// System CA certificate search paths (platform-specific). const SYSTEM_CA_PATHS: &[&str] = &[ "/etc/ssl/certs/ca-certificates.crt", // Debian/Ubuntu/Alpine @@ -98,7 +101,13 @@ pub(crate) async fn build_acceptor_from_loaded( match server { Some(RuntimeServerTlsMaterial::SingleCert { certs, key }) => { let config = build_server_config(ServerCertSource::SingleCert { certs, key }, mtls_verifier)?; - info!("Created TLS acceptor with root single certificate"); + info!( + component = LOG_COMPONENT_TLS, + subsystem = LOG_SUBSYSTEM_TLS, + event = "tls_acceptor_created", + mode = "single_cert", + "TLS acceptor created" + ); let acceptor = Arc::new(TlsAcceptor::from(Arc::new(config))); Ok(Some(Arc::new(TlsAcceptorHolder::new(acceptor)))) } @@ -106,7 +115,13 @@ pub(crate) async fn build_acceptor_from_loaded( let resolver = create_multi_cert_resolver(cert_key_pairs) .map_err(|e| TlsMaterialError::Parse(format!("build multi-cert resolver: {e}")))?; let config = build_server_config(ServerCertSource::Resolver(Arc::new(resolver)), mtls_verifier)?; - info!("Created TLS acceptor with SNI resolver"); + info!( + component = LOG_COMPONENT_TLS, + subsystem = LOG_SUBSYSTEM_TLS, + event = "tls_acceptor_created", + mode = "sni_resolver", + "TLS acceptor created" + ); let acceptor = Arc::new(TlsAcceptor::from(Arc::new(config))); Ok(Some(Arc::new(TlsAcceptorHolder::new(acceptor)))) } @@ -126,7 +141,13 @@ async fn enrich_outbound(outbound: &mut rustfs_tls_runtime::OutboundTlsMaterial, if get_env_bool(ENV_TRUST_LEAF_CERT_AS_CA, DEFAULT_TRUST_LEAF_CERT_AS_CA) && load_cert_file_by_name(tls_dir, RUSTFS_TLS_CERT, &mut outbound.root_ca_pem).await { - info!("Loaded leaf certificate(s) as root CA as per RUSTFS_TRUST_LEAF_CERT_AS_CA"); + info!( + component = LOG_COMPONENT_TLS, + subsystem = LOG_SUBSYSTEM_TLS, + event = "tls_root_ca_enriched", + source = "leaf_certificate", + "TLS root CA set enriched" + ); } // 2. Optional: load system root CAs @@ -135,15 +156,36 @@ async fn enrich_outbound(outbound: &mut rustfs_tls_runtime::OutboundTlsMaterial, for path in SYSTEM_CA_PATHS { if load_cert_file(Path::new(path), &mut outbound.root_ca_pem, "system root certificates").await { system_loaded = true; - info!("Loaded system root certificates from {}", path); + info!( + component = LOG_COMPONENT_TLS, + subsystem = LOG_SUBSYSTEM_TLS, + event = "tls_root_ca_enriched", + source = "system_roots", + path, + "TLS root CA set enriched" + ); break; } } if !system_loaded { - debug!("Could not find system root certificates in common locations."); + debug!( + component = LOG_COMPONENT_TLS, + subsystem = LOG_SUBSYSTEM_TLS, + event = "tls_root_ca_enriched", + source = "system_roots", + state = "not_found", + "No system root certificates were loaded" + ); } } else { - info!("Loading system root certificates disabled via RUSTFS_TRUST_SYSTEM_CA"); + info!( + component = LOG_COMPONENT_TLS, + subsystem = LOG_SUBSYSTEM_TLS, + event = "tls_root_ca_enriched", + source = "system_roots", + state = "disabled", + "System root CA loading disabled" + ); } // 3. Optional: override mTLS identity from env-var paths @@ -177,8 +219,13 @@ async fn load_mtls_identity_from_overridden_paths( if !client_cert_path.exists() || !client_key_path.exists() { info!( - "mTLS client identity not configured (missing {:?} and/or {:?}); proceeding with server-only TLS", - client_cert_path, client_key_path + component = LOG_COMPONENT_TLS, + subsystem = LOG_SUBSYSTEM_TLS, + event = "mtls_identity_state", + state = "not_configured", + cert_path = ?client_cert_path, + key_path = ?client_key_path, + "mTLS client identity not configured" ); return Ok(None); } @@ -197,7 +244,15 @@ async fn load_mtls_identity_from_overridden_paths( let mut reader = std::io::Cursor::new(&key_pem); PrivateKeyDer::from_pem_reader(&mut reader).map_err(|e| TlsMaterialError::Parse(format!("invalid client key PEM: {e}")))?; - info!("Loaded mTLS client identity cert={:?} key={:?}", client_cert_path, client_key_path); + info!( + component = LOG_COMPONENT_TLS, + subsystem = LOG_SUBSYSTEM_TLS, + event = "mtls_identity_state", + state = "loaded", + cert_path = ?client_cert_path, + key_path = ?client_key_path, + "mTLS client identity loaded" + ); Ok(Some(MtlsIdentityPem { cert_pem, key_pem })) } @@ -216,18 +271,43 @@ fn map_runtime_tls_error(err: rustfs_tls_runtime::TlsRuntimeError) -> TlsMateria /// Returns true if the file was successfully loaded. async fn load_cert_file(path: &Path, pem_data: &mut Vec, desc: &str) -> bool { if tokio::fs::metadata(path).await.is_err() { - debug!("{} file not found at {:?}", desc, path); + debug!( + component = LOG_COMPONENT_TLS, + subsystem = LOG_SUBSYSTEM_TLS, + event = "tls_material_file_state", + description = desc, + path = ?path, + state = "missing", + "TLS material file state changed" + ); return false; } match tokio::fs::read(path).await { Ok(data) => { pem_data.extend_from_slice(&data); pem_data.push(b'\n'); - info!("Loaded {} from {:?}", desc, path); + info!( + component = LOG_COMPONENT_TLS, + subsystem = LOG_SUBSYSTEM_TLS, + event = "tls_material_file_state", + description = desc, + path = ?path, + state = "loaded", + "TLS material file state changed" + ); true } Err(e) => { - debug!("Failed to read {} from {:?}: {}", desc, path, e); + debug!( + component = LOG_COMPONENT_TLS, + subsystem = LOG_SUBSYSTEM_TLS, + event = "tls_material_file_state", + description = desc, + path = ?path, + state = "read_failed", + error = %e, + "TLS material file state changed" + ); false } } @@ -238,7 +318,14 @@ async fn load_cert_file(path: &Path, pem_data: &mut Vec, desc: &str) -> bool /// Returns `true` if at least one matching file was loaded. async fn load_cert_file_by_name(dir: &Path, cert_name: &str, pem_data: &mut Vec) -> bool { let Ok(mut rd) = tokio::fs::read_dir(dir).await else { - debug!("Certificate directory not found: {}", dir.display()); + debug!( + component = LOG_COMPONENT_TLS, + subsystem = LOG_SUBSYSTEM_TLS, + event = "tls_material_directory_state", + path = %dir.display(), + state = "missing", + "TLS material directory state changed" + ); return false; }; @@ -421,14 +508,28 @@ impl TlsAcceptorHolder { pub(crate) fn spawn_reload_loop(tls_path: String, holder: Arc) { let enabled = get_env_bool(ENV_TLS_RELOAD_ENABLE, DEFAULT_TLS_RELOAD_ENABLE); if !enabled { - debug!("TLS certificate hot reload is disabled (set {}=1 to enable)", ENV_TLS_RELOAD_ENABLE); + debug!( + component = LOG_COMPONENT_TLS, + subsystem = LOG_SUBSYSTEM_TLS, + event = "tls_reload_state", + state = "disabled", + env_var = ENV_TLS_RELOAD_ENABLE, + "TLS reload state changed" + ); return; } let interval_secs = rustfs_utils::get_env_u64(ENV_TLS_RELOAD_INTERVAL, DEFAULT_TLS_RELOAD_INTERVAL).max(5); let tls_source = TlsSource::from_directory(&tls_path); - info!("TLS certificate hot reload enabled, checking every {}s", interval_secs); + info!( + component = LOG_COMPONENT_TLS, + subsystem = LOG_SUBSYSTEM_TLS, + event = "tls_reload_state", + state = "enabled", + interval_secs, + "TLS reload state changed" + ); tokio::spawn(async move { let mut interval = tokio::time::interval(Duration::from_secs(interval_secs)); @@ -440,7 +541,14 @@ pub(crate) fn spawn_reload_loop(tls_path: String, holder: Arc Ok(mut snapshot) => { if let Err(e) = enrich_outbound(&mut snapshot.outbound, &tls_dir).await { record_tls_reload_result("rustfs_server_reload_loop", "enrich_err", None, None); - warn!("TLS outbound enrichment failed (will retry): {}", e); + warn!( + component = LOG_COMPONENT_TLS, + subsystem = LOG_SUBSYSTEM_TLS, + event = "tls_reload_failed", + stage = "enrich_outbound", + error = %e, + "TLS reload failed" + ); continue; } @@ -448,28 +556,64 @@ pub(crate) fn spawn_reload_loop(tls_path: String, holder: Arc publish_global_outbound_tls_state(TlsGeneration(generation), &snapshot.outbound).await; record_tls_generation("rustfs_server_reload_loop", generation); if !snapshot.outbound.root_ca_pem.is_empty() { - info!("Configured custom root certificates for inter-node communication"); + info!( + component = LOG_COMPONENT_TLS, + subsystem = LOG_SUBSYSTEM_TLS, + event = "tls_root_ca_enriched", + source = "reload_snapshot", + generation, + "TLS root CA set updated" + ); } match build_acceptor_from_loaded(snapshot.server, &tls_dir).await { Ok(Some(new_holder)) => { - info!("TLS certificates reloaded successfully"); + info!( + component = LOG_COMPONENT_TLS, + subsystem = LOG_SUBSYSTEM_TLS, + event = "tls_reload_state", + state = "reloaded", + generation, + "TLS reload state changed" + ); holder.swap(&new_holder); record_tls_reload_result("rustfs_server_reload_loop", "ok", None, Some(generation)); } Ok(None) => { record_tls_reload_skipped("rustfs_server_reload_loop", "no_acceptor"); - warn!("TLS reload returned no acceptor despite configured TLS path; keeping previous acceptor") + warn!( + component = LOG_COMPONENT_TLS, + subsystem = LOG_SUBSYSTEM_TLS, + event = "tls_reload_failed", + stage = "build_acceptor", + reason = "no_acceptor", + "TLS reload returned no acceptor" + ) } Err(e) => { record_tls_reload_result("rustfs_server_reload_loop", "acceptor_err", None, Some(generation)); - warn!("TLS certificate reload failed (will retry): {}", e) + warn!( + component = LOG_COMPONENT_TLS, + subsystem = LOG_SUBSYSTEM_TLS, + event = "tls_reload_failed", + stage = "build_acceptor", + error = %e, + generation, + "TLS reload failed" + ) } } } Err(e) => { record_tls_reload_result("rustfs_server_reload_loop", "load_err", None, None); - warn!("TLS material reload failed (will retry): {}", e); + warn!( + component = LOG_COMPONENT_TLS, + subsystem = LOG_SUBSYSTEM_TLS, + event = "tls_reload_failed", + stage = "load_snapshot", + error = %e, + "TLS reload failed" + ); } } } diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index 74008988a..1836e839d 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -52,9 +52,14 @@ use rustfs_utils::http::headers::{ use s3s::{S3, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, dto::*, s3_error}; use std::fmt::Debug; use time::{OffsetDateTime, format_description::well_known::Rfc3339}; -use tracing::{debug, error, info, instrument, warn}; +use tracing::{debug, error, instrument, warn}; use uuid::Uuid; +const LOG_COMPONENT_STORAGE: &str = "storage"; +const LOG_SUBSYSTEM_OBJECT: &str = "object"; +const LOG_SUBSYSTEM_OBJECT_LOCK: &str = "object_lock"; +const LOG_SUBSYSTEM_TAGGING: &str = "tagging"; + #[derive(Debug, Clone)] pub struct FS { // pub store: ECStore, @@ -394,7 +399,13 @@ impl S3 for FS { validate_table_catalog_object_mutation(&bucket, &object).await?; let Some(store) = new_object_layer_fn() else { - error!("Store not initialized"); + error!( + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_TAGGING, + event = "object_tagging_store_uninitialized", + operation = "delete", + "Object tagging operation failed because storage is not initialized" + ); return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -407,7 +418,15 @@ impl S3 for FS { let delete_tags_result = store.delete_object_tags(&bucket, &object, &opts).await; Self::record_replication_tagging_metric(&bucket, &object, "DeleteObjectTagging", delete_tags_result.is_err()).await; delete_tags_result.map_err(|e| { - error!("Failed to delete object tags: {}", e); + error!( + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_TAGGING, + event = "object_tagging_delete_failed", + bucket = %bucket, + object = %object, + error = %e, + "Failed to delete object tags" + ); ApiError::from(e) })?; @@ -698,7 +717,15 @@ impl S3 for FS { .map_err(ApiError::from)?; let object_info = store.get_object_info(&bucket, &key, &opts).await.map_err(|e| { - error!("get_object_info failed, {}", e.to_string()); + error!( + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_OBJECT, + event = "object_info_load_failed", + bucket = %bucket, + object = %key, + error = %e, + "Failed to load object info" + ); s3_error!(InternalError, "{}", e.to_string()) })?; @@ -753,7 +780,14 @@ impl S3 for FS { "Object Lock configuration does not exist for this bucket".to_string(), )); } - warn!("get_object_lock_config err {:?}", err); + warn!( + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_OBJECT_LOCK, + event = "object_lock_config_load_failed", + bucket = %bucket, + error = ?err, + "Failed to load bucket object lock configuration" + ); return Err(S3Error::with_message( S3ErrorCode::InternalError, "Failed to load Object Lock configuration".to_string(), @@ -787,7 +821,15 @@ impl S3 for FS { .map_err(ApiError::from)?; let object_info = store.get_object_info(&bucket, &key, &opts).await.map_err(|e| { - error!("get_object_info failed, {}", e.to_string()); + error!( + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_OBJECT, + event = "object_info_load_failed", + bucket = %bucket, + object = %key, + error = %e, + "Failed to load object info" + ); s3_error!(InternalError, "{}", e.to_string()) })?; @@ -820,10 +862,16 @@ impl S3 for FS { let bucket = req.input.bucket.as_str(); let object = req.input.key.as_str(); - info!("Starting get_object_tagging for bucket: {}, object: {}", bucket, object); - let Some(store) = new_object_layer_fn() else { - error!("Store not initialized"); + error!( + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_TAGGING, + event = "object_tagging_store_uninitialized", + operation = "get", + bucket = %bucket, + object = %object, + "Object tagging operation failed because storage is not initialized" + ); return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -837,15 +885,39 @@ impl S3 for FS { Self::record_replication_tagging_metric(bucket, object, "GetObjectTagging", tags_result.is_err()).await; let tags = tags_result.map_err(|e| { if is_err_object_not_found(&e) { - error!("Object not found: {}", e); + debug!( + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_TAGGING, + event = "object_tagging_not_found", + bucket = %bucket, + object = %object, + error = %e, + "Object tags not found" + ); return s3_error!(NoSuchKey); } - error!("Failed to get object tags: {}", e); + error!( + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_TAGGING, + event = "object_tagging_get_failed", + bucket = %bucket, + object = %object, + error = %e, + "Failed to load object tags" + ); ApiError::from(e).into() })?; let tag_set = decode_tags(tags.as_str()); - debug!("Decoded tag set: {:?}", tag_set); + debug!( + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_TAGGING, + event = "object_tagging_decoded", + bucket = %bucket, + object = %object, + tag_count = tag_set.len(), + "Decoded object tags" + ); counter!("rustfs_get_object_tagging_success").increment(1); let duration = start_time.elapsed(); diff --git a/rustfs/src/storage/ecfs_extend.rs b/rustfs/src/storage/ecfs_extend.rs index de0b41fd6..d0f1f8e9a 100644 --- a/rustfs/src/storage/ecfs_extend.rs +++ b/rustfs/src/storage/ecfs_extend.rs @@ -48,6 +48,10 @@ use time::format_description::well_known::Rfc3339; use time::{format_description::FormatItem, macros::format_description}; use tracing::{debug, warn}; +const LOG_COMPONENT_STORAGE: &str = "storage"; +const LOG_SUBSYSTEM_OBJECT_LOCK: &str = "object_lock"; +const LOG_SUBSYSTEM_OBJECT_KEY: &str = "object_key"; + pub const RFC1123: &[FormatItem<'_>] = format_description!("[weekday repr:short], [day] [month repr:short] [year] [hour]:[minute]:[second] GMT"); @@ -149,7 +153,14 @@ pub(crate) async fn apply_bucket_default_lock_retention( if err == StorageError::ConfigNotFound { None } else { - warn!("get_object_lock_config err {:?}", err); + warn!( + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_OBJECT_LOCK, + event = "object_lock_config_load_failed", + bucket = %bucket, + error = ?err, + "Failed to load bucket object lock configuration" + ); return Err(S3Error::with_message( S3ErrorCode::InternalError, "Failed to load Object Lock configuration".to_string(), @@ -293,7 +304,14 @@ pub(crate) fn validate_object_key(key: &str, operation: &str) -> S3Result<()> { // Log debug info for keys with special characters to help diagnose encoding issues if key.contains([' ', '+', '%']) { - debug!("{} object with special characters in key: {:?}", operation, key); + debug!( + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_OBJECT_KEY, + event = "object_key_special_characters", + operation, + key = ?key, + "Object key contains special characters" + ); } Ok(()) @@ -423,7 +441,14 @@ pub(crate) async fn validate_bucket_object_lock_enabled(bucket: &str) -> S3Resul "Bucket is missing ObjectLockConfiguration".to_string(), )); } - warn!("get_object_lock_config err {:?}", err); + warn!( + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_OBJECT_LOCK, + event = "object_lock_config_load_failed", + bucket = %bucket, + error = ?err, + "Failed to load bucket object lock configuration" + ); return Err(S3Error::with_message( S3ErrorCode::InternalError, "Failed to get bucket ObjectLockConfiguration".to_string(), diff --git a/rustfs/src/storage/rpc/http_service.rs b/rustfs/src/storage/rpc/http_service.rs index 8a6205d64..2e230fb94 100644 --- a/rustfs/src/storage/rpc/http_service.rs +++ b/rustfs/src/storage/rpc/http_service.rs @@ -245,7 +245,7 @@ async fn handle_walk_dir(req: Request) -> Response { let (rd, mut wd) = tokio::io::duplex(DEFAULT_READ_BUFFER_SIZE); tokio::spawn(async move { if let Err(e) = disk.walk_dir(args, &mut wd).await { - warn!("walk dir err {}", e); + warn!(error = %e, "walk_dir failed"); } }); diff --git a/rustfs/src/storage/rpc/metrics.rs b/rustfs/src/storage/rpc/metrics.rs index e64d56938..9a4a58418 100644 --- a/rustfs/src/storage/rpc/metrics.rs +++ b/rustfs/src/storage/rpc/metrics.rs @@ -27,7 +27,7 @@ impl NodeService { let t: MetricType = match Deserialize::deserialize(&mut buf_t) { Ok(t) => t, Err(err) => { - error!("Failed to deserialize metric_type: {}", err); + error!(error = %err, "failed to deserialize metric_type"); return Ok(Response::new(GetMetricsResponse { success: false, realtime_metrics: Bytes::new(), @@ -41,7 +41,7 @@ impl NodeService { let opts: CollectMetricsOpts = match Deserialize::deserialize(&mut buf_o) { Ok(opts) => opts, Err(err) => { - error!("Failed to deserialize opts: {}", err); + error!(error = %err, "failed to deserialize opts"); return Ok(Response::new(GetMetricsResponse { success: false, realtime_metrics: Bytes::new(), diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index 02e287722..44e0a8232 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -58,6 +58,9 @@ use tokio_stream::wrappers::ReceiverStream; use tonic::{Request, Response, Status, Streaming}; use tracing::{debug, error, info, warn}; +const LOG_COMPONENT_STORAGE: &str = "storage"; +const LOG_SUBSYSTEM_RPC: &str = "rpc"; + type ResponseStream = Pin> + Send>>; fn unimplemented_rpc(method: &str) -> Status { @@ -110,17 +113,25 @@ impl NodeService { #[tonic::async_trait] impl Node for NodeService { async fn ping(&self, request: Request) -> Result, Status> { - debug!("PING"); - let ping_req = request.into_inner(); if ping_req.body.is_empty() { - debug!("ping_req received empty body; treating request as liveness probe"); + debug!( + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_RPC, + event = "ping_request", + request_type = "liveness_probe", + "RPC ping request received" + ); } else { let ping_body = flatbuffers::root::(&ping_req.body); if let Err(e) = ping_body { - warn!("invalid ping request body: {}", e); - } else { - info!("ping_req:body(flatbuffer): {:?}", ping_body); + warn!( + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_RPC, + event = "ping_request_decode_failed", + error = %e, + "Failed to decode RPC ping request body" + ); } } @@ -196,7 +207,6 @@ impl Node for NodeService { type WriteStreamStream = ResponseStream; async fn write_stream(&self, request: Request>) -> Result, Status> { - info!("write_stream"); let _ = request; Err(unimplemented_rpc("write_stream")) @@ -204,7 +214,6 @@ impl Node for NodeService { type ReadAtStream = ResponseStream; async fn read_at(&self, _request: Request>) -> Result, Status> { - info!("read_at"); Err(unimplemented_rpc("read_at")) } @@ -214,7 +223,6 @@ impl Node for NodeService { type WalkDirStream = ResponseStream; async fn walk_dir(&self, request: Request) -> Result, Status> { - info!("walk_dir"); let request = request.into_inner(); let (tx, rx) = mpsc::channel(128); if let Some(disk) = self.find_disk(&request.disk).await { @@ -229,7 +237,13 @@ impl Node for NodeService { let (rd, mut wr) = tokio::io::duplex(64); let job1 = spawn(async move { if let Err(err) = disk.walk_dir(opts, &mut wr).await { - error!("walk_dir failed: {err:?}"); + error!( + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_RPC, + event = "walk_dir_failed", + error = ?err, + "walk_dir RPC failed" + ); } }); let job2 = spawn(async move { @@ -250,7 +264,13 @@ impl Node for NodeService { .await .is_err() { - warn!("walk_dir stream receiver dropped while sending meta cache entry"); + warn!( + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_RPC, + event = "walk_dir_stream_closed", + stage = "entry_send", + "walk_dir stream receiver dropped" + ); break; } } @@ -264,7 +284,13 @@ impl Node for NodeService { .await .is_err() { - warn!("walk_dir stream receiver dropped while sending serialization error"); + warn!( + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_RPC, + event = "walk_dir_stream_closed", + stage = "serialization_error_send", + "walk_dir stream receiver dropped" + ); break; } } @@ -298,7 +324,13 @@ impl Node for NodeService { break; } - warn!("walk_dir metacache read error: {err:?}"); + warn!( + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_RPC, + event = "walk_dir_metacache_read_failed", + error = ?err, + "walk_dir metacache read failed" + ); let _ = tx .send(Ok(WalkDirResponse { @@ -903,7 +935,7 @@ impl Node for NodeService { })) } - #[tracing::instrument(skip_all)] + #[tracing::instrument(skip_all, fields(start_rebalance))] async fn load_rebalance_meta( &self, request: Request, @@ -917,21 +949,21 @@ impl Node for NodeService { let LoadRebalanceMetaRequest { start_rebalance } = request.into_inner(); - warn!("handle LoadRebalanceMetaRequest"); + info!("handling load_rebalance_meta request"); store.load_rebalance_meta().await.map_err(|err| { - error!("load_rebalance_meta err {:?}", err); + error!(error = ?err, "load_rebalance_meta failed"); Status::internal(err.to_string()) })?; - warn!("load_rebalance_meta success"); + info!("load_rebalance_meta completed"); if start_rebalance { - warn!("start rebalance"); + info!(start_rebalance, "spawning background rebalance task"); let store = store.clone(); spawn(async move { if let Some(message) = background_rebalance_start_error_message(store.start_rebalance().await) { - error!("{message}"); + error!(error = %message, "background rebalance start failed"); } }); } diff --git a/rustfs/src/storage/sse.rs b/rustfs/src/storage/sse.rs index aa1b73fcb..bcaca76a4 100644 --- a/rustfs/src/storage/sse.rs +++ b/rustfs/src/storage/sse.rs @@ -92,6 +92,9 @@ use std::collections::HashMap; use std::sync::{Arc, LazyLock, RwLock}; use tracing::{debug, error}; +const LOG_COMPONENT_STORAGE: &str = "storage"; +const LOG_SUBSYSTEM_SSE: &str = "sse"; + const INTERNAL_ENCRYPTION_KEY_ID_HEADER: &str = "x-rustfs-encryption-key-id"; const INTERNAL_ENCRYPTION_KEY_HEADER: &str = "x-rustfs-encryption-key"; const INTERNAL_ENCRYPTION_IV_HEADER: &str = "x-rustfs-encryption-iv"; @@ -211,14 +214,28 @@ async fn prepare_sse_configuration( // Get bucket default encryption configuration. let bucket_sse_config_result = metadata_sys::get_sse_config(bucket).await; - debug!("bucket_sse_config_result={:?}", bucket_sse_config_result); + debug!( + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_SSE, + event = "bucket_sse_config_lookup", + bucket = %bucket, + found = bucket_sse_config_result.is_ok(), + "Bucket SSE configuration lookup completed" + ); if let Ok((bucket_sse_config, _timestamp)) = bucket_sse_config_result { let effective_sse = server_side_encryption.clone().or_else(|| { bucket_sse_config.rules.first().and_then(|rule| { - debug!("Processing SSE rule: {:?}", rule); rule.apply_server_side_encryption_by_default.as_ref().map(|sse| { - debug!("Found SSE default: {:?}", sse); + debug!( + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_SSE, + event = "bucket_sse_default_applied", + bucket = %bucket, + algorithm = sse.sse_algorithm.as_str(), + has_kms_key_id = sse.kms_master_key_id.is_some(), + "Bucket SSE default resolved" + ); match sse.sse_algorithm.as_str() { "AES256" => ServerSideEncryption::from_static(ServerSideEncryption::AES256), "aws:kms" => ServerSideEncryption::from_static(ServerSideEncryption::AWS_KMS), @@ -231,7 +248,15 @@ async fn prepare_sse_configuration( return Ok(None); } - debug!("effective_sse={:?} (original={:?})", effective_sse, server_side_encryption); + debug!( + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_SSE, + event = "effective_sse_resolved", + bucket = %bucket, + requested = ?server_side_encryption, + effective = ?effective_sse, + "Resolved effective SSE configuration" + ); let effective_kms_key_id = resolve_effective_kms_key_id(effective_sse.as_ref(), ssekms_key_id, || { bucket_sse_config.rules.first().and_then(|rule| { diff --git a/scripts/check_logging_guardrails.sh b/scripts/check_logging_guardrails.sh index 2752120ca..1261b3b34 100755 --- a/scripts/check_logging_guardrails.sh +++ b/scripts/check_logging_guardrails.sh @@ -59,6 +59,58 @@ forbidden_patterns=( 'error!("Failed to initialize SFTP system: {}"' 'warn!("prewarm_local_disk_id_map: failed to load disk id for {}: {}"' 'info!("retrying get formats after {:?}"' + 'info!("📝 Release notes: {}"' + 'info!("🔗 Download URL: {}"' + 'debug!("✅ Version check: Current version is up to date: {}"' + 'warn!("get_notification_config err {:?}"' + 'error!("ecstore config::init_global_config_sys failed {:?}"' + 'info!(target: "rustfs::main::startup", "RustFS API: {api_endpoints} {localhost_endpoint}")' + 'info!("virtual-hosted-style requests are enabled use domain_name {:?}"' + 'info!("HTTP response compression enabled: extensions={:?}, mime_patterns={:?}, min_size={} bytes"' + 'warn!("handle ListCannedPolicies")' + 'warn!("handle AddCannedPolicy")' + 'warn!("handle InfoCannedPolicy")' + 'warn!("handle RemoveCannedPolicy")' + 'warn!("handle SetPolicyForUserOrGroup")' + 'warn!("handle Trace")' + 'warn!("handle AddServiceAccount ")' + 'warn!("handle UpdateServiceAccount")' + 'warn!("handle InfoServiceAccount")' + 'warn!("handle TemporaryAccountInfo")' + 'warn!("handle InfoAccessKey")' + 'warn!("handle ListServiceAccount")' + 'warn!("handle ListAccessKeysBulk")' + 'warn!("handle DeleteServiceAccount")' + 'debug!("{action}, e: {:?}")' + 'warn!("list policies failed, e: {:?}")' + 'error!("Invalid JSON in configure request: {}")' + 'warn!("get_sse_config err {:?}")' + 'warn!("get_cors_config err {:?}")' + 'warn!("get_notification_config err {:?}")' + 'warn!("get_tagging_config err {:?}")' + 'warn!("get_object_lock_config err {:?}")' + 'info!("Starting get_object_tagging for bucket: {}, object: {}")' + 'debug!("bucket_sse_config_result={:?}")' + 'debug!("TDD: bucket_sse_config={:?}")' + 'info!("Keystone authentication disabled")' + 'info!("Initializing Keystone authentication...")' + 'info!("Keystone authentication initialized successfully")' + 'info!("Created TLS acceptor with root single certificate")' + 'info!("TLS certificate hot reload enabled, checking every {}s")' + 'info!("Initializing capacity management system...")' + 'info!("Capacity management system initialized successfully")' + 'info!("CPU profile exported: {}")' + 'warn!("failed to reload current server config for object lambda request: {err}")' + 'warn!("failed to fetch live events from peer {}: {err}")' + 'warn!("timed out fetching live events from peer {}")' + 'warn!("failed to serialize remote listen notification event: {err}")' + 'warn!("failed to decode live events from peer {}: {err}")' + 'warn!("failed to serialize listen notification event: {err}")' + 'warn!("listen notification stream lagged and skipped {skipped} events")' + 'debug!("Ignoring crypto provider installation error: {err:?}")' + 'warn!("KMS initialization skipped: {e}")' + 'warn!("Audit system: {e}")' + 'warn!("notification system: {e}")' ) for pattern in "${forbidden_patterns[@]}"; do