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

refactor(logging): standardize rustfs runtime events (#3396)

Browse Source 
* 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
This commit is contained in:
houseme
2026-06-13 08:39:47 +08:00
committed by GitHub
parent b38e71fac3
commit 0d68851f7a
25 changed files with 1980 additions and 346 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+26 -26
View File
@@ -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]]
Cargo.toml
+6 -6
View File
@@ -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"
crates/e2e_test/Cargo.toml
+1 -1
View File
@@ -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"] }
rustfs/src/admin/handlers/kms_dynamic.rs
+251 -30
View File
@@ -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<rustfs_kms::KmsServiceManager> {
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<KmsConfig> {
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::<KmsConfig>(&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)
}
rustfs/src/admin/handlers/policies.rs
+129 -25
View File
@@ -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<AdminOperation>) -> 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<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
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<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
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<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
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<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
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<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
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<Body>) -> S3Result<S3Resp
let all_group_policy_mappings = collect_group_policy_mappings(&iam_store, &[]).await?;
let users = iam_store.list_users().await.map_err(|e| {
warn!("list users failed, e: {:?}", e);
warn!(
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_POLICY,
event = "policy_user_list_failed",
error = ?e,
"Failed to list users for policy entities"
);
S3Error::with_message(S3ErrorCode::InternalError, e.to_string())
})?;
@@ -847,7 +920,15 @@ async fn handle_builtin_policy_association(req: S3Request<Body>, 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<Body>, 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<Body>, 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())
})?;
rustfs/src/admin/handlers/service_account.rs
+50 -20
View File
@@ -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<SRSessionPolicy> {
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<Policy> {
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<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
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<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
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<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
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<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
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<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
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<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
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<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
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<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
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")
})?;
rustfs/src/admin/handlers/trace.rs
-3
View File
@@ -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<ServiceTraceOpts> {
@@ -38,8 +37,6 @@ pub struct Trace {}
#[async_trait::async_trait]
impl Operation for Trace {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
warn!("handle Trace");
let _trace_opts = extract_trace_options(&req.uri)?;
// let (tx, rx) = mpsc::channel(10000);
rustfs/src/admin/router.rs
+64 -9
View File
@@ -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<Config> {
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,
}
rustfs/src/app/bucket_usecase.rs
+51 -6
View File
@@ -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<T: xml::Serialize>(value: &T) -> S3Result<Vec<u8>> {
@@ -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());
}
}
rustfs/src/app/object_usecase.rs
+19 -7
View File
@@ -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)| {
rustfs/src/auth_keystone.rs
+70 -16
View File
@@ -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<Arc<KeystoneAuthProvider>> = OnceLock::new();
static KEYSTONE_MAPPER: OnceLock<Arc<KeystoneIdentityMapper>> = OnceLock::new();
static KEYSTONE_CONFIG: OnceLock<KeystoneConfig> = 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<dyn std::error::Error>> {
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<dyn st
.set(config.clone())
.map_err(|_| "Keystone config already initialized")?;
info!("Keystone authentication initialized successfully");
info!(" Auth URL: {}", config.auth_url);
info!(" Version: {}", config.version);
info!(" Tenant prefix enabled: {}", config.enable_tenant_prefix);
info!(" Token caching enabled: {}", config.enable_cache);
info!(
component = LOG_COMPONENT_AUTH,
subsystem = LOG_SUBSYSTEM_KEYSTONE,
event = "keystone_state",
state = "initialized",
auth_url = %config.auth_url,
version = %config.version,
enable_tenant_prefix = config.enable_tenant_prefix,
enable_cache = config.enable_cache,
"Keystone authentication state changed"
);
Ok(())
}
@@ -125,15 +151,29 @@ pub async fn authenticate_keystone(headers: &HeaderMap) -> S3Result<Option<Crede
// Check for X-Auth-Token header (Keystone v3)
if let Some(token) = headers.get("X-Auth-Token").and_then(|v| v.to_str().ok()) {
debug!("Found X-Auth-Token header, validating with Keystone");
return match auth_provider.authenticate_with_token(token).await {
Ok(cred) => {
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<Option<Crede
// Check for X-Storage-Token header (Swift compatibility)
if let Some(token) = headers.get("X-Storage-Token").and_then(|v| v.to_str().ok()) {
debug!("Found X-Storage-Token header, validating with Keystone");
return match auth_provider.authenticate_with_token(token).await {
Ok(cred) => {
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))
}
};
rustfs/src/capacity/service.rs
+108 -17
View File
@@ -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<String>, drive_path: impl Into<String>) -> 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)> {
rustfs/src/embedded.rs
+89 -10
View File
@@ -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"
);
}
}
rustfs/src/init.rs
+334 -57
View File
@@ -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<String>) {
.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"
);
}
}
rustfs/src/main.rs
+61 -5
View File
@@ -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(())
}
rustfs/src/profiling.rs
+122 -14
View File
@@ -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<Arc<Mutex<Option<pprof::ProfilerGuard<'static>>>>> = OnceLock::new();
static PROFILING_CANCEL_TOKEN: OnceLock<CancellationToken> = 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),
rustfs/src/server/http.rs
+187 -32
View File
@@ -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<GlobalRea
) {
Ok(s) => 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<GlobalRea
if server_addr.is_ipv6()
&& let Err(e) = socket.set_only_v6(false)
{
warn!("Failed to set IPV6_V6ONLY=false, attempting IPv4 fallback: {}", e);
warn!(
event = EVENT_SOCKET_FALLBACK,
component = LOG_COMPONENT_SERVER,
subsystem = LOG_SUBSYSTEM_STARTUP,
from_addr = %server_addr,
fallback = "ipv4",
error = %e,
"Dual-stack socket setup fell back to IPv4"
);
let ipv4_addr = SocketAddr::new(std::net::Ipv4Addr::UNSPECIFIED.into(), server_addr.port());
server_addr = ipv4_addr;
socket = socket2::Socket::new(socket2::Domain::IPV4, socket2::Type::STREAM, Some(socket2::Protocol::TCP))?;
@@ -258,7 +286,14 @@ pub async fn start_http_server(config: &config::Config, readiness: Arc<GlobalRea
// 2. Enable SO_REUSEPORT for better multi-core scalability on supported platforms
#[cfg(all(unix, not(target_os = "solaris"), not(target_os = "illumos")))]
if let Err(e) = socket.set_reuse_port(true) {
debug!("Failed to set SO_REUSEPORT: {}", e);
debug!(
event = "socket_option_unavailable",
component = LOG_COMPONENT_SERVER,
subsystem = LOG_SUBSYSTEM_STARTUP,
option = "SO_REUSEPORT",
error = %e,
"Socket option is unavailable"
);
}
// 3. Set system-level TCP KeepAlive to protect long connections
@@ -275,7 +310,14 @@ pub async fn start_http_server(config: &config::Config, readiness: Arc<GlobalRea
// Attempt bind; if bind fails for IPv6, try IPv4 fallback once more.
if let Err(bind_err) = socket.bind(&server_addr.into()) {
warn!("Failed to bind to {}: {}.", server_addr, bind_err);
warn!(
event = EVENT_HTTP_BIND_FAILED,
component = LOG_COMPONENT_SERVER,
subsystem = LOG_SUBSYSTEM_STARTUP,
bind_addr = %server_addr,
error = %bind_err,
"HTTP listener bind failed"
);
if server_addr.is_ipv6() {
// Try IPv4 fallback
let ipv4_addr = SocketAddr::new(std::net::Ipv4Addr::UNSPECIFIED.into(), server_addr.port());
@@ -336,7 +378,13 @@ pub async fn start_http_server(config: &config::Config, readiness: Arc<GlobalRea
let local_ip = match rustfs_utils::get_local_ip() {
Some(ip) => 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<GlobalRea
info!(
target: "rustfs::console::startup",
"Console WebUI available at: {protocol}://{local_ip_str}:{local_port}/rustfs/console/index.html"
event = EVENT_HTTP_STARTUP_ENDPOINTS,
component = LOG_COMPONENT_SERVER,
subsystem = LOG_SUBSYSTEM_STARTUP,
service = "console",
endpoint = %format!("{protocol}://{local_ip_str}:{local_port}/rustfs/console/index.html"),
"Startup endpoint available"
);
info!(
target: "rustfs::console::startup",
"Console WebUI (localhost): {protocol}://127.0.0.1:{local_port}/rustfs/console/index.html",
event = EVENT_HTTP_STARTUP_ENDPOINTS,
component = LOG_COMPONENT_SERVER,
subsystem = LOG_SUBSYSTEM_STARTUP,
service = "console_localhost",
endpoint = %format!("{protocol}://127.0.0.1:{local_port}/rustfs/console/index.html"),
"Startup endpoint available"
);
} else {
info!(target: "rustfs::main::startup", "RustFS API: {api_endpoints} {localhost_endpoint}");
info!(target: "rustfs::main::startup", "RustFS Start Time: {now_time}");
info!(target: "rustfs::main::startup","For more information, visit https://rustfs.com/docs/");
info!(target: "rustfs::main::startup", "To enable the console, restart the server with --console-enable and a valid --console-address.");
info!(
target: "rustfs::main::startup",
event = EVENT_HTTP_STARTUP_ENDPOINTS,
component = LOG_COMPONENT_SERVER,
subsystem = LOG_SUBSYSTEM_STARTUP,
service = "s3_api",
api_endpoint = %api_endpoints,
localhost_endpoint = %localhost_endpoint,
started_at = %now_time,
console_enabled = false,
docs_url = "https://rustfs.com/docs/",
"Startup endpoints ready"
);
}
// Setup S3 service
@@ -399,7 +465,15 @@ pub async fn start_http_server(config: &config::Config, readiness: Arc<GlobalRea
}
}
info!("virtual-hosted-style requests are enabled use domain_name {:?}", &domain_sets);
info!(
event = EVENT_HTTP_HOST_ROUTING,
component = LOG_COMPONENT_SERVER,
subsystem = LOG_SUBSYSTEM_HTTP,
state = "enabled",
domain_count = domain_sets.len(),
domains = ?domain_sets,
"Virtual-hosted-style routing configured"
);
b.set_host(MultiDomain::new(domain_sets).map_err(Error::other)?);
}
@@ -412,11 +486,23 @@ pub async fn start_http_server(config: &config::Config, readiness: Arc<GlobalRea
let compression_config = HttpCompressionConfig::from_env();
if compression_config.enabled {
info!(
"HTTP response compression enabled: extensions={:?}, mime_patterns={:?}, min_size={} bytes",
compression_config.extensions, compression_config.mime_patterns, compression_config.min_size
event = EVENT_HTTP_COMPRESSION_STATE,
component = LOG_COMPONENT_SERVER,
subsystem = LOG_SUBSYSTEM_HTTP,
state = "enabled",
extensions = ?compression_config.extensions,
mime_patterns = ?compression_config.mime_patterns,
min_size_bytes = compression_config.min_size,
"HTTP response compression state changed"
);
} else {
debug!("HTTP response compression is disabled");
debug!(
event = EVENT_HTTP_COMPRESSION_STATE,
component = LOG_COMPONENT_SERVER,
subsystem = LOG_SUBSYSTEM_HTTP,
state = "disabled",
"HTTP response compression state changed"
);
}
let is_console = config.console_enable;
@@ -460,18 +546,19 @@ pub async fn start_http_server(config: &config::Config, readiness: Arc<GlobalRea
rustfs_utils::get_env_usize(rustfs_config::ENV_HTTP1_MAX_BUF_SIZE, rustfs_config::DEFAULT_HTTP1_MAX_BUF_SIZE);
info!(
"HTTP transport parameters: h2_stream_window={}, h2_conn_window={}, h2_max_frame={}, \
h2_max_header_list={}, h2_max_concurrent_streams={}, h2_keepalive_interval={}s, \
h2_keepalive_timeout={}s, http1_header_timeout={}s, http1_max_buf={}",
event = EVENT_HTTP_TRANSPORT_PARAMETERS,
component = LOG_COMPONENT_SERVER,
subsystem = LOG_SUBSYSTEM_TRANSPORT,
h2_stream_window,
h2_conn_window,
h2_max_frame_size,
h2_max_header_list_size,
h2_max_concurrent_streams,
h2_keep_alive_interval,
h2_keep_alive_timeout,
http1_header_read_timeout,
h2_keep_alive_interval_secs = h2_keep_alive_interval,
h2_keep_alive_timeout_secs = h2_keep_alive_timeout,
http1_header_read_timeout_secs = http1_header_read_timeout,
http1_max_buf_size,
"HTTP transport parameters configured"
);
let mut conn_builder = ConnBuilder::new(TokioExecutor::new());
@@ -500,7 +587,13 @@ pub async fn start_http_server(config: &config::Config, readiness: Arc<GlobalRea
let http_server = Arc::new(conn_builder);
let graceful = GracefulShutdown::new();
debug!("graceful initiated");
debug!(
event = EVENT_HTTP_ACCEPT_LOOP_STATE,
component = LOG_COMPONENT_SERVER,
subsystem = LOG_SUBSYSTEM_TRANSPORT,
state = "started",
"HTTP accept loop started"
);
loop {
trace!("Waiting for new connection");
@@ -511,12 +604,25 @@ pub async fn start_http_server(config: &config::Config, readiness: Arc<GlobalRea
res = listener.accept() => 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<GlobalRea
res = listener.accept() => 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<GlobalRea
let active_connections = graceful.count();
if active_connections > 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<Request<()>, 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<Request<()>, 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)
rustfs/src/server/tls_material.rs
+165 -21
View File
@@ -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<u8>, 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<u8>, 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<u8>) -> 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<TlsAcceptorHolder>) {
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<TlsAcceptorHolder>
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<TlsAcceptorHolder>
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"
);
}
}
}
rustfs/src/storage/ecfs.rs
+84 -12
View File
@@ -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();
rustfs/src/storage/ecfs_extend.rs
+28 -3
View File
@@ -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(),
rustfs/src/storage/rpc/http_service.rs
+1 -1
View File
@@ -245,7 +245,7 @@ async fn handle_walk_dir(req: Request<Incoming>) -> Response<Body> {
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");
}
});
rustfs/src/storage/rpc/metrics.rs
+2 -2
View File
@@ -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(),
rustfs/src/storage/rpc/node_service.rs
+51 -19
View File
@@ -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<T> = Pin<Box<dyn Stream<Item = Result<T, Status>> + 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<PingRequest>) -> Result<Response<PingResponse>, 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::<PingBody>(&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<WriteResponse>;
async fn write_stream(&self, request: Request<Streaming<WriteRequest>>) -> Result<Response<Self::WriteStreamStream>, Status> {
info!("write_stream");
let _ = request;
Err(unimplemented_rpc("write_stream"))
@@ -204,7 +214,6 @@ impl Node for NodeService {
type ReadAtStream = ResponseStream<ReadAtResponse>;
async fn read_at(&self, _request: Request<Streaming<ReadAtRequest>>) -> Result<Response<Self::ReadAtStream>, Status> {
info!("read_at");
Err(unimplemented_rpc("read_at"))
}
@@ -214,7 +223,6 @@ impl Node for NodeService {
type WalkDirStream = ResponseStream<WalkDirResponse>;
async fn walk_dir(&self, request: Request<WalkDirRequest>) -> Result<Response<Self::WalkDirStream>, 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<LoadRebalanceMetaRequest>,
@@ -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");
}
});
}
rustfs/src/storage/sse.rs
+29 -4
View File
@@ -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| {
scripts/check_logging_guardrails.sh
+52
View File
@@ -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