diff --git a/rustfs/src/admin/handlers/audit.rs b/rustfs/src/admin/handlers/audit.rs index 6c218afa9..bf5e656c2 100644 --- a/rustfs/src/admin/handlers/audit.rs +++ b/rustfs/src/admin/handlers/audit.rs @@ -40,7 +40,119 @@ use s3s::{Body, S3Request, S3Response, S3Result, s3_error}; use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::sync::LazyLock; -use tracing::{Span, warn}; +use tracing::{Span, error, info, warn}; + +const LOG_COMPONENT_ADMIN_API: &str = "admin_api"; +const LOG_SUBSYSTEM_AUDIT_TARGET: &str = "audit_target"; +const EVENT_ADMIN_REQUEST_REJECTED: &str = "admin_request_rejected"; +const EVENT_ADMIN_REQUEST_FAILED: &str = "admin_request_failed"; +const EVENT_ADMIN_RESPONSE_EMITTED: &str = "admin_response_emitted"; +const EVENT_ADMIN_OPERATION_BLOCKED: &str = "admin_operation_blocked"; + +macro_rules! log_audit_target_operation_blocked { + ($operation:expr, Some($target_type:expr), Some($target_name:expr), $reason:expr) => { + warn!( + event = EVENT_ADMIN_OPERATION_BLOCKED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_AUDIT_TARGET, + operation = $operation, + result = "blocked", + target_type = $target_type, + target_name = $target_name, + reason = $reason, + "admin request rejected" + ); + }; + ($operation:expr, None, None, $reason:expr) => { + warn!( + event = EVENT_ADMIN_OPERATION_BLOCKED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_AUDIT_TARGET, + operation = $operation, + result = "blocked", + reason = $reason, + "admin request rejected" + ); + }; +} + +macro_rules! log_audit_target_body_read_failed { + ($target_type:expr, $target_name:expr, $err:expr) => { + warn!( + event = EVENT_ADMIN_REQUEST_REJECTED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_AUDIT_TARGET, + operation = "set_audit_target_config", + result = "rejected", + reason = "request_body_read_failed", + target_type = $target_type, + target_name = $target_name, + error = %$err, + "admin request rejected" + ); + }; +} + +macro_rules! log_audit_target_body_decode_failed { + ($target_type:expr, $target_name:expr, $err:expr) => { + warn!( + event = EVENT_ADMIN_REQUEST_REJECTED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_AUDIT_TARGET, + operation = "set_audit_target_config", + result = "rejected", + reason = "request_body_decode_failed", + target_type = $target_type, + target_name = $target_name, + error = %$err, + "admin request rejected" + ); + }; +} + +macro_rules! log_audit_target_config_updated { + ($operation:expr, $target_type:expr, $target_name:expr) => { + info!( + event = EVENT_ADMIN_RESPONSE_EMITTED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_AUDIT_TARGET, + operation = $operation, + result = "success", + target_type = $target_type, + target_name = $target_name, + "admin response emitted" + ); + }; +} + +macro_rules! log_audit_target_request_failed { + ($operation:expr, $reason:expr, Some($target_type:expr), Some($target_name:expr), $err:expr) => { + error!( + event = EVENT_ADMIN_REQUEST_FAILED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_AUDIT_TARGET, + operation = $operation, + result = "failed", + reason = $reason, + target_type = $target_type, + target_name = $target_name, + error = %$err, + "admin request failed" + ); + }; + ($operation:expr, $reason:expr, None, None, $err:expr) => { + error!( + event = EVENT_ADMIN_REQUEST_FAILED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_AUDIT_TARGET, + operation = $operation, + result = "failed", + reason = $reason, + error = %$err, + "admin request failed" + ); + }; +} pub fn register_audit_target_route(r: &mut S3Router) -> std::io::Result<()> { r.insert( @@ -123,8 +235,13 @@ fn audit_target_mutation_block_reason(config: &Config, target_type: &str, target async fn audit_target_operation_block_reason(action: &str) -> Option { if let Err(err) = refresh_persisted_module_switches_from_store().await { warn!( + event = EVENT_ADMIN_REQUEST_REJECTED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_AUDIT_TARGET, + operation = "reload_persisted_module_switches", + result = "failed", error = %err, - "failed to reload persisted module switches before checking audit target operation gating" + "admin request rejected" ); } refresh_audit_module_enabled(); @@ -158,21 +275,25 @@ impl Operation for AuditTargetConfig { authorize_audit_admin_request(&req, AdminAction::SetBucketTargetAction).await?; if let Some(reason) = audit_target_operation_block_reason("managing audit targets from the console").await { + log_audit_target_operation_blocked!("set_audit_target_config", Some(target_type), Some(target_name), &reason); return Err(s3_error!(InvalidRequest, "{reason}")); } let config_snapshot = load_server_config_from_store().await?; if let Some(reason) = audit_target_mutation_block_reason(&config_snapshot, target_type, target_name) { + log_audit_target_operation_blocked!("set_audit_target_config", Some(target_type), Some(target_name), &reason); return Err(s3_error!(InvalidRequest, "{reason}")); } let mut input = req.input; let body_bytes = input.store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE).await.map_err(|e| { - warn!("failed to read request body: {:?}", e); + log_audit_target_body_read_failed!(target_type, target_name, e); s3_error!(InvalidRequest, "failed to read request body") })?; - let audit_body: AuditTargetBody = serde_json::from_slice(&body_bytes) - .map_err(|e| s3_error!(InvalidArgument, "invalid json body for audit target config: {}", e))?; + let audit_body: AuditTargetBody = serde_json::from_slice(&body_bytes).map_err(|e| { + log_audit_target_body_decode_failed!(target_type, target_name, e); + s3_error!(InvalidArgument, "invalid json body for audit target config: {}", e) + })?; let specs = audit_target_specs(); let kvs = build_enabled_target_kvs( @@ -192,7 +313,17 @@ impl Operation for AuditTargetConfig { .insert(target_name.to_lowercase(), kvs.clone()); true }) - .await?; + .await + .inspect_err(|e| { + log_audit_target_request_failed!( + "set_audit_target_config", + "update_audit_config_failed", + Some(target_type), + Some(target_name), + e + ); + })?; + log_audit_target_config_updated!("set_audit_target_config", target_type, target_name); Ok(build_json_response(StatusCode::OK, Body::empty(), req.headers.get("x-request-id"))) } @@ -214,8 +345,10 @@ impl Operation for ListAuditTargets { let config = load_server_config_from_store().await?; let audit_endpoints = merge_audit_endpoints(&config, runtime_statuses); - let data = serde_json::to_vec(&AuditEndpointsResponse { audit_endpoints }) - .map_err(|e| s3_error!(InternalError, "failed to serialize audit targets: {}", e))?; + let data = serde_json::to_vec(&AuditEndpointsResponse { audit_endpoints }).map_err(|e| { + log_audit_target_request_failed!("list_audit_targets", "serialize_audit_targets_failed", None, None, e); + s3_error!(InternalError, "failed to serialize audit targets: {}", e) + })?; Ok(build_json_response(StatusCode::OK, Body::from(data), req.headers.get("x-request-id"))) } @@ -232,10 +365,12 @@ impl Operation for RemoveAuditTarget { authorize_audit_admin_request(&req, AdminAction::SetBucketTargetAction).await?; if let Some(reason) = audit_target_operation_block_reason("managing audit targets from the console").await { + log_audit_target_operation_blocked!("remove_audit_target_config", Some(target_type), Some(target_name), &reason); return Err(s3_error!(InvalidRequest, "{reason}")); } let config_snapshot = load_server_config_from_store().await?; if let Some(reason) = audit_target_mutation_block_reason(&config_snapshot, target_type, target_name) { + log_audit_target_operation_blocked!("remove_audit_target_config", Some(target_type), Some(target_name), &reason); return Err(s3_error!(InvalidRequest, "{reason}")); } @@ -251,7 +386,17 @@ impl Operation for RemoveAuditTarget { } changed }) - .await?; + .await + .inspect_err(|e| { + log_audit_target_request_failed!( + "remove_audit_target_config", + "update_audit_config_failed", + Some(target_type), + Some(target_name), + e + ); + })?; + log_audit_target_config_updated!("remove_audit_target_config", target_type, target_name); Ok(build_json_response(StatusCode::OK, Body::empty(), req.headers.get("x-request-id"))) } diff --git a/rustfs/src/admin/handlers/event.rs b/rustfs/src/admin/handlers/event.rs index 6a882fe33..799337f3e 100644 --- a/rustfs/src/admin/handlers/event.rs +++ b/rustfs/src/admin/handlers/event.rs @@ -40,7 +40,119 @@ use s3s::{Body, S3Request, S3Response, S3Result, s3_error}; use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::sync::LazyLock; -use tracing::{Span, info, warn}; +use tracing::{Span, error, info, warn}; + +const LOG_COMPONENT_ADMIN_API: &str = "admin_api"; +const LOG_SUBSYSTEM_NOTIFICATION_TARGET: &str = "notification_target"; +const EVENT_ADMIN_REQUEST_REJECTED: &str = "admin_request_rejected"; +const EVENT_ADMIN_REQUEST_FAILED: &str = "admin_request_failed"; +const EVENT_ADMIN_RESPONSE_EMITTED: &str = "admin_response_emitted"; +const EVENT_ADMIN_OPERATION_BLOCKED: &str = "admin_operation_blocked"; + +macro_rules! log_notification_target_operation_blocked { + ($operation:expr, Some($target_type:expr), Some($target_name:expr), $reason:expr) => { + warn!( + event = EVENT_ADMIN_OPERATION_BLOCKED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_NOTIFICATION_TARGET, + operation = $operation, + result = "blocked", + target_type = $target_type, + target_name = $target_name, + reason = $reason, + "admin request rejected" + ); + }; + ($operation:expr, None, None, $reason:expr) => { + warn!( + event = EVENT_ADMIN_OPERATION_BLOCKED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_NOTIFICATION_TARGET, + operation = $operation, + result = "blocked", + reason = $reason, + "admin request rejected" + ); + }; +} + +macro_rules! log_notification_target_body_read_failed { + ($target_type:expr, $target_name:expr, $err:expr) => { + warn!( + event = EVENT_ADMIN_REQUEST_REJECTED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_NOTIFICATION_TARGET, + operation = "set_target_config", + result = "rejected", + reason = "request_body_read_failed", + target_type = $target_type, + target_name = $target_name, + error = %$err, + "admin request rejected" + ); + }; +} + +macro_rules! log_notification_target_body_decode_failed { + ($target_type:expr, $target_name:expr, $err:expr) => { + warn!( + event = EVENT_ADMIN_REQUEST_REJECTED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_NOTIFICATION_TARGET, + operation = "set_target_config", + result = "rejected", + reason = "request_body_decode_failed", + target_type = $target_type, + target_name = $target_name, + error = %$err, + "admin request rejected" + ); + }; +} + +macro_rules! log_notification_target_config_updated { + ($operation:expr, $target_type:expr, $target_name:expr) => { + info!( + event = EVENT_ADMIN_RESPONSE_EMITTED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_NOTIFICATION_TARGET, + operation = $operation, + result = "success", + target_type = $target_type, + target_name = $target_name, + "admin response emitted" + ); + }; +} + +macro_rules! log_notification_target_request_failed { + ($operation:expr, $reason:expr, Some($target_type:expr), Some($target_name:expr), $err:expr) => { + error!( + event = EVENT_ADMIN_REQUEST_FAILED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_NOTIFICATION_TARGET, + operation = $operation, + result = "failed", + reason = $reason, + target_type = $target_type, + target_name = $target_name, + error = %$err, + "admin request failed" + ); + }; + ($operation:expr, $reason:expr, None, None, $err:expr) => { + error!( + event = EVENT_ADMIN_REQUEST_FAILED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_NOTIFICATION_TARGET, + operation = $operation, + result = "failed", + reason = $reason, + error = %$err, + "admin request failed" + ); + }; +} pub fn register_notification_target_route(r: &mut S3Router) -> std::io::Result<()> { r.insert( @@ -131,8 +243,13 @@ fn target_mutation_block_reason(config: &Config, target_type: &str, target_name: async fn notification_target_operation_block_reason(action: &str) -> Option { if let Err(err) = refresh_persisted_module_switches_from_store().await { warn!( + event = EVENT_ADMIN_REQUEST_REJECTED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_NOTIFICATION_TARGET, + operation = "reload_persisted_module_switches", + result = "failed", error = %err, - "failed to reload persisted module switches before checking notification target operation gating" + "admin request rejected" ); } refresh_notify_module_enabled(); @@ -170,21 +287,25 @@ impl Operation for NotificationTarget { authorize_notification_admin_request(&req, AdminAction::SetBucketTargetAction).await?; if let Some(reason) = notification_target_operation_block_reason("managing notification targets from the console").await { + log_notification_target_operation_blocked!("set_target_config", Some(target_type), Some(target_name), &reason); return Err(s3_error!(InvalidRequest, "{reason}")); } let (ns, config_snapshot) = load_notification_config_snapshot().await?; if let Some(reason) = target_mutation_block_reason(&config_snapshot, target_type, target_name) { + log_notification_target_operation_blocked!("set_target_config", Some(target_type), Some(target_name), &reason); return Err(s3_error!(InvalidRequest, "{reason}")); } let mut input = req.input; let body_bytes = input.store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE).await.map_err(|e| { - warn!("failed to read request body: {:?}", e); + log_notification_target_body_read_failed!(target_type, target_name, e); s3_error!(InvalidRequest, "failed to read request body") })?; - let notification_body: NotificationTargetBody = serde_json::from_slice(&body_bytes) - .map_err(|e| s3_error!(InvalidArgument, "invalid json body for target config: {}", e))?; + let notification_body: NotificationTargetBody = serde_json::from_slice(&body_bytes).map_err(|e| { + log_notification_target_body_decode_failed!(target_type, target_name, e); + s3_error!(InvalidArgument, "invalid json body for target config: {}", e) + })?; let specs = notification_target_specs(); let kvs = build_enabled_target_kvs( @@ -199,10 +320,17 @@ impl Operation for NotificationTarget { ) .await?; - info!("Setting target config for type '{}', name '{}'", target_type, target_name); - ns.set_target_config(target_type, target_name, kvs) - .await - .map_err(|e| s3_error!(InternalError, "failed to set target config: {}", e))?; + ns.set_target_config(target_type, target_name, kvs).await.map_err(|e| { + log_notification_target_request_failed!( + "set_target_config", + "set_target_config_failed", + Some(target_type), + Some(target_name), + e + ); + s3_error!(InternalError, "failed to set target config: {}", e) + })?; + log_notification_target_config_updated!("set_target_config", target_type, target_name); Ok(build_json_response(StatusCode::OK, Body::empty(), req.headers.get("x-request-id"))) } @@ -219,8 +347,10 @@ impl Operation for ListNotificationTargets { let runtime_statuses = collect_runtime_statuses(ns.get_target_values().await).await; let notification_endpoints = merge_notification_endpoints(&config, runtime_statuses); - let data = serde_json::to_vec(&NotificationEndpointsResponse { notification_endpoints }) - .map_err(|e| s3_error!(InternalError, "failed to serialize targets: {}", e))?; + let data = serde_json::to_vec(&NotificationEndpointsResponse { notification_endpoints }).map_err(|e| { + log_notification_target_request_failed!("list_targets", "serialize_targets_failed", None, None, e); + s3_error!(InternalError, "failed to serialize targets: {}", e) + })?; Ok(build_json_response(StatusCode::OK, Body::from(data), req.headers.get("x-request-id"))) } @@ -238,14 +368,23 @@ impl Operation for ListTargetsArns { ) .await { + log_notification_target_operation_blocked!("list_target_arns", None, None, &reason); return Err(s3_error!(InvalidRequest, "{reason}")); } let ns = get_notification_system()?; - let region = req - .region - .clone() - .ok_or_else(|| s3_error!(InvalidRequest, "region not found"))?; + let region = req.region.clone().ok_or_else(|| { + warn!( + event = EVENT_ADMIN_REQUEST_REJECTED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_NOTIFICATION_TARGET, + operation = "list_target_arns", + result = "rejected", + reason = "region_not_found", + "admin request rejected" + ); + s3_error!(InvalidRequest, "region not found") + })?; let target_statuses = collect_runtime_statuses(ns.get_target_values().await) .await .into_iter() @@ -254,8 +393,10 @@ impl Operation for ListTargetsArns { let data_target_arn_list = collect_online_target_arns(region.as_str(), target_statuses); - let data = serde_json::to_vec(&data_target_arn_list) - .map_err(|e| s3_error!(InternalError, "failed to serialize targets: {}", e))?; + let data = serde_json::to_vec(&data_target_arn_list).map_err(|e| { + log_notification_target_request_failed!("list_target_arns", "serialize_targets_failed", None, None, e); + s3_error!(InternalError, "failed to serialize targets: {}", e) + })?; Ok(build_json_response(StatusCode::OK, Body::from(data), req.headers.get("x-request-id"))) } @@ -271,17 +412,26 @@ impl Operation for RemoveNotificationTarget { authorize_notification_admin_request(&req, AdminAction::SetBucketTargetAction).await?; if let Some(reason) = notification_target_operation_block_reason("managing notification targets from the console").await { + log_notification_target_operation_blocked!("remove_target_config", Some(target_type), Some(target_name), &reason); return Err(s3_error!(InvalidRequest, "{reason}")); } let (ns, config_snapshot) = load_notification_config_snapshot().await?; if let Some(reason) = target_mutation_block_reason(&config_snapshot, target_type, target_name) { + log_notification_target_operation_blocked!("remove_target_config", Some(target_type), Some(target_name), &reason); return Err(s3_error!(InvalidRequest, "{reason}")); } - info!("Removing target config for type '{}', name '{}'", target_type, target_name); - ns.remove_target_config(target_type, target_name) - .await - .map_err(|e| s3_error!(InternalError, "failed to remove target config: {}", e))?; + ns.remove_target_config(target_type, target_name).await.map_err(|e| { + log_notification_target_request_failed!( + "remove_target_config", + "remove_target_config_failed", + Some(target_type), + Some(target_name), + e + ); + s3_error!(InternalError, "failed to remove target config: {}", e) + })?; + log_notification_target_config_updated!("remove_target_config", target_type, target_name); Ok(build_json_response(StatusCode::OK, Body::empty(), req.headers.get("x-request-id"))) } diff --git a/rustfs/src/admin/handlers/heal.rs b/rustfs/src/admin/handlers/heal.rs index 9d0054721..bb2a371ad 100644 --- a/rustfs/src/admin/handlers/heal.rs +++ b/rustfs/src/admin/handlers/heal.rs @@ -38,6 +38,12 @@ use time::{OffsetDateTime, format_description::well_known::Rfc3339}; use tokio::sync::mpsc; use tracing::{info, warn}; +const LOG_COMPONENT_ADMIN_API: &str = "admin_api"; +const LOG_SUBSYSTEM_HEAL_ADMIN: &str = "heal_admin"; +const EVENT_ADMIN_REQUEST_REJECTED: &str = "admin_request_rejected"; +const EVENT_ADMIN_REQUEST_FAILED: &str = "admin_request_failed"; +const EVENT_ADMIN_RESPONSE_EMITTED: &str = "admin_response_emitted"; + #[derive(Debug, Default, Serialize, Deserialize)] struct HealInitParams { bucket: String, @@ -85,7 +91,16 @@ fn extract_heal_init_params(body: &Bytes, uri: &Uri, params: Params<'_, '_>) -> if hip.client_token.is_empty() { hip.hs = serde_json::from_slice(body).map_err(|e| { - info!("err request body parse, err: {:?}", e); + warn!( + event = EVENT_ADMIN_REQUEST_REJECTED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_HEAL_ADMIN, + operation = "heal_request_init", + result = "rejected", + reason = "request_body_parse_failed", + error = ?e, + "admin request rejected" + ); s3_error!(InvalidRequest, "err request body parse") })?; } @@ -292,7 +307,19 @@ fn encode_background_heal_status(info: &BackgroundHealInfo) -> S3Result> heal_queue_length: rustfs_heal::current_heal_queue_length(), heal_active_tasks: rustfs_heal::current_heal_active_tasks(), }; - serde_json::to_vec(&status).map_err(|e| s3_error!(InternalError, "failed to serialize background heal status: {e}")) + serde_json::to_vec(&status).map_err(|e| { + warn!( + event = EVENT_ADMIN_REQUEST_FAILED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_HEAL_ADMIN, + operation = "background_heal_status", + result = "failed", + reason = "serialize_background_heal_status_failed", + error = %e, + "admin request failed" + ); + s3_error!(InternalError, "failed to serialize background heal status: {e}") + }) } fn validate_heal_request_mode(hip: &HealInitParams) -> S3Result<()> { @@ -311,10 +338,30 @@ fn map_root_heal_status(heal_err: Option) -> S3Res match heal_err { None => Ok(()), Some(rustfs_ecstore::error::StorageError::NoHealRequired) => { - warn!("root heal completed with non-fatal status: no heal required"); + info!( + event = EVENT_ADMIN_RESPONSE_EMITTED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_HEAL_ADMIN, + operation = "root_heal", + result = "success", + state = "no_heal_required", + "admin response emitted" + ); Ok(()) } - Some(err) => Err(s3_error!(InternalError, "root heal failed: {err}")), + Some(err) => { + warn!( + event = EVENT_ADMIN_REQUEST_FAILED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_HEAL_ADMIN, + operation = "root_heal", + result = "failed", + reason = "root_heal_failed", + error = %err, + "admin request failed" + ); + Err(s3_error!(InternalError, "root heal failed: {err}")) + } } } @@ -350,7 +397,6 @@ pub struct HealHandler {} #[async_trait::async_trait] impl Operation for HealHandler { async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { - warn!("handle HealHandler, req: {:?}, params: {:?}", req, params); validate_heal_admin_request(&req).await?; let client_address = req .extensions @@ -361,31 +407,72 @@ impl Operation for HealHandler { let bytes = match input.store_all_limited(MAX_HEAL_REQUEST_SIZE).await { Ok(b) => b, Err(e) => { - warn!("get body failed, e: {:?}", e); + warn!( + event = EVENT_ADMIN_REQUEST_REJECTED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_HEAL_ADMIN, + operation = "heal_request_body", + result = "rejected", + reason = "request_body_read_failed", + error = ?e, + "admin request rejected" + ); return Err(s3_error!(InvalidRequest, "heal request body too large or failed to read")); } }; - info!("bytes: {:?}", bytes); let hip = extract_heal_init_params(&bytes, &req.uri, params)?; // The heal channel currently models bucket/object work. Root heal reuses the // existing format-heal path directly so `/v3/heal/` is accepted intentionally. if should_handle_root_heal_directly(&hip) { let Some(store) = resolve_object_store_handle() else { + warn!( + event = EVENT_ADMIN_REQUEST_FAILED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_HEAL_ADMIN, + operation = "root_heal", + result = "failed", + reason = "server_not_initialized", + "admin request failed" + ); return Err(s3_error!(InternalError, "server not initialized")); }; - let (_, heal_err) = store - .heal_format(hip.hs.dry_run) - .await - .map_err(|e| s3_error!(InternalError, "root heal failed: {e}"))?; + let (_, heal_err) = store.heal_format(hip.hs.dry_run).await.map_err(|e| { + warn!( + event = EVENT_ADMIN_REQUEST_FAILED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_HEAL_ADMIN, + operation = "root_heal", + result = "failed", + reason = "heal_format_failed", + error = %e, + "admin request failed" + ); + s3_error!(InternalError, "root heal failed: {e}") + })?; map_root_heal_status(heal_err)?; let body = encode_heal_start_success("root-heal".to_string(), client_address)?; + info!( + event = EVENT_ADMIN_RESPONSE_EMITTED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_HEAL_ADMIN, + operation = "root_heal", + result = "success", + state = "started", + "admin response emitted" + ); return Ok(json_response(StatusCode::OK, body)); } validate_heal_request_mode(&hip)?; - info!("body: {:?}", hip); + let response_operation = if hip.force_stop { + "cancel_heal" + } else if !hip.client_token.is_empty() && !hip.force_start { + "query_heal_status" + } else { + "start_heal" + }; let heal_path = path_join(&[PathBuf::from(hip.bucket.clone()), PathBuf::from(hip.obj_prefix.clone())]); let (tx, mut rx) = mpsc::channel(1); @@ -548,6 +635,15 @@ impl Operation for HealHandler { } let (status, body) = map_heal_response(rx.recv().await)?; + info!( + event = EVENT_ADMIN_RESPONSE_EMITTED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_HEAL_ADMIN, + operation = response_operation, + result = "success", + status_code = status.as_u16(), + "admin response emitted" + ); Ok(json_response(status, body)) } } @@ -557,15 +653,31 @@ pub struct BackgroundHealStatusHandler {} #[async_trait::async_trait] impl Operation for BackgroundHealStatusHandler { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { - warn!("handle BackgroundHealStatusHandler"); validate_heal_admin_request(&req).await?; let Some(store) = resolve_object_store_handle() else { + warn!( + event = EVENT_ADMIN_REQUEST_FAILED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_HEAL_ADMIN, + operation = "background_heal_status", + result = "failed", + reason = "server_not_initialized", + "admin request failed" + ); return Err(s3_error!(InternalError, "server not initialized")); }; let info = read_background_heal_info(store).await; let body = encode_background_heal_status(&info)?; + info!( + event = EVENT_ADMIN_RESPONSE_EMITTED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_HEAL_ADMIN, + operation = "background_heal_status", + result = "success", + "admin response emitted" + ); Ok(json_response(StatusCode::OK, body)) } diff --git a/rustfs/src/admin/handlers/pools.rs b/rustfs/src/admin/handlers/pools.rs index 15689ea21..533e7fd18 100644 --- a/rustfs/src/admin/handlers/pools.rs +++ b/rustfs/src/admin/handlers/pools.rs @@ -19,7 +19,7 @@ use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, header::C use serde::Deserialize; use serde_urlencoded::from_bytes; use tokio_util::sync::CancellationToken; -use tracing::warn; +use tracing::{error, info, warn}; use crate::{ admin::{ @@ -36,6 +36,69 @@ use hyper::Method; use std::collections::HashSet; +const LOG_COMPONENT_ADMIN_API: &str = "admin_api"; +const LOG_SUBSYSTEM_POOL_ADMIN: &str = "pool_admin"; +const EVENT_ADMIN_REQUEST_REJECTED: &str = "admin_request_rejected"; +const EVENT_ADMIN_REQUEST_FAILED: &str = "admin_request_failed"; +const EVENT_ADMIN_RESPONSE_EMITTED: &str = "admin_response_emitted"; + +macro_rules! log_pool_request_rejected { + ($operation:expr, $reason:expr) => { + warn!( + event = EVENT_ADMIN_REQUEST_REJECTED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_POOL_ADMIN, + operation = $operation, + result = "rejected", + reason = $reason, + "admin request rejected" + ); + }; +} + +macro_rules! log_pool_request_rejected_with_pool { + ($operation:expr, $reason:expr, $pool:expr) => { + warn!( + event = EVENT_ADMIN_REQUEST_REJECTED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_POOL_ADMIN, + operation = $operation, + result = "rejected", + reason = $reason, + pool = $pool, + "admin request rejected" + ); + }; +} + +macro_rules! log_pool_request_failed { + ($operation:expr, $reason:expr, $err:expr) => { + error!( + event = EVENT_ADMIN_REQUEST_FAILED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_POOL_ADMIN, + operation = $operation, + result = "failed", + reason = $reason, + error = %$err, + "admin request failed" + ); + }; +} + +macro_rules! log_pool_response_emitted { + ($operation:expr) => { + info!( + event = EVENT_ADMIN_RESPONSE_EMITTED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_POOL_ADMIN, + operation = $operation, + result = "success", + "admin response emitted" + ); + }; +} + fn endpoints_from_context() -> Option { resolve_endpoints_handle() } @@ -68,22 +131,31 @@ fn contextualize_admin_pool_api_error( } fn decommission_admin_not_initialized_error(operation: &str) -> S3Error { + log_pool_request_failed!( + operation_to_event(operation), + "object_layer_not_initialized", + "object layer not initialized" + ); S3Error::with_message(S3ErrorCode::InternalError, format!("Failed to {operation}: object layer not initialized")) } fn pool_admin_missing_credentials_error(operation: &str) -> S3Error { + log_pool_request_rejected!(operation_to_event(operation), "missing_credentials"); S3Error::with_message(S3ErrorCode::InvalidRequest, format!("Failed to {operation}: missing credentials")) } fn pool_admin_query_parse_error(operation: &str) -> S3Error { + log_pool_request_rejected!(operation_to_event(operation), "invalid_query_parameters"); S3Error::with_message(S3ErrorCode::InvalidArgument, format!("Failed to {operation}: invalid query parameters")) } fn pool_admin_pool_parse_error(operation: &str, pool: &str) -> S3Error { + log_pool_request_rejected_with_pool!(operation_to_event(operation), "invalid_pool", pool); S3Error::with_message(S3ErrorCode::InvalidArgument, format!("Failed to {operation}: invalid pool `{pool}`")) } fn pool_admin_pool_not_found_error(operation: &str, pool: &str) -> S3Error { + log_pool_request_rejected_with_pool!(operation_to_event(operation), "pool_not_found", pool); S3Error::with_message( S3ErrorCode::InvalidArgument, format!("Failed to {operation}: pool `{pool}` was not found"), @@ -91,12 +163,33 @@ fn pool_admin_pool_not_found_error(operation: &str, pool: &str) -> S3Error { } fn pool_admin_pool_index_error(operation: &str, idx: usize, pool_count: usize) -> S3Error { + warn!( + event = EVENT_ADMIN_REQUEST_REJECTED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_POOL_ADMIN, + operation = operation_to_event(operation), + result = "rejected", + reason = "pool_index_out_of_range", + pool_index = idx, + pool_count, + "admin request rejected" + ); S3Error::with_message( S3ErrorCode::InvalidArgument, format!("Failed to {operation}: pool index {idx} is out of range for {pool_count} pools"), ) } +fn operation_to_event(operation: &str) -> &'static str { + match operation { + "list pools" => "list_pools", + "load pool status" => "query_pool_status", + "start decommission" => "start_decommission", + "cancel decommission" => "cancel_decommission", + _ => "pool_admin", + } +} + fn parse_pool_idx_by_id(pool: &str, endpoint_count: usize) -> Option { let idx = pool.parse::().ok()?; (idx < endpoint_count).then_some(idx) @@ -148,8 +241,6 @@ impl Operation for ListPools { // GET //pools/list #[tracing::instrument(skip_all)] async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { - warn!("handle ListPools"); - let Some(input_cred) = req.credentials else { return Err(pool_admin_missing_credentials_error("list pools")); }; @@ -173,11 +264,14 @@ impl Operation for ListPools { let usecase = DefaultAdminUsecase::from_global(); let pool_items = usecase.execute_list_pools().await.map_err(S3Error::from)?; - let data = serde_json::to_vec(&pool_items) - .map_err(|_e| S3Error::with_message(S3ErrorCode::InternalError, "serialize pools list failed"))?; + let data = serde_json::to_vec(&pool_items).map_err(|e| { + log_pool_request_failed!("list_pools", "serialize_pools_list_failed", e); + S3Error::with_message(S3ErrorCode::InternalError, "serialize pools list failed") + })?; let mut header = HeaderMap::new(); header.insert(CONTENT_TYPE, "application/json".parse().unwrap()); + log_pool_response_emitted!("list_pools"); Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), header)) } @@ -198,8 +292,6 @@ impl Operation for StatusPool { // GET //pools/status?pool=http://server{1...4}/disk{1...4} #[tracing::instrument(skip_all)] async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { - warn!("handle StatusPool"); - let Some(input_cred) = req.credentials else { return Err(pool_admin_missing_credentials_error("load pool status")); }; @@ -239,11 +331,14 @@ impl Operation for StatusPool { .await .map_err(S3Error::from)?; - let data = serde_json::to_vec(&pools_status) - .map_err(|_e| S3Error::with_message(S3ErrorCode::InternalError, "parse accountInfo failed"))?; + let data = serde_json::to_vec(&pools_status).map_err(|e| { + log_pool_request_failed!("query_pool_status", "serialize_pool_status_failed", e); + S3Error::with_message(S3ErrorCode::InternalError, "parse accountInfo failed") + })?; let mut header = HeaderMap::new(); header.insert(CONTENT_TYPE, "application/json".parse().unwrap()); + log_pool_response_emitted!("query_pool_status"); Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), header)) } @@ -256,8 +351,6 @@ impl Operation for StartDecommission { // POST //pools/decommission?pool=http://server{1...4}/disk{1...4} #[tracing::instrument(skip_all)] async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { - warn!("handle StartDecommission"); - let Some(input_cred) = req.credentials else { return Err(pool_admin_missing_credentials_error("start decommission")); }; @@ -276,10 +369,12 @@ impl Operation for StartDecommission { .await?; let Some(endpoints) = endpoints_from_context() else { + log_pool_request_rejected!("start_decommission", "not_implemented"); return Err(s3_error!(NotImplemented)); }; if endpoints.legacy() { + log_pool_request_rejected!("start_decommission", "legacy_endpoints_not_supported"); return Err(s3_error!(NotImplemented)); } @@ -335,6 +430,7 @@ impl Operation for StartDecommission { .map_err(|err| contextualize_admin_pool_api_error(err, "start decommission", &pool_context))?; } + log_pool_response_emitted!("start_decommission"); Ok(S3Response::new((StatusCode::OK, Body::default()))) } } @@ -346,8 +442,6 @@ impl Operation for CancelDecommission { // POST //pools/cancel?pool=http://server{1...4}/disk{1...4} #[tracing::instrument(skip_all)] async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { - warn!("handle CancelDecommission"); - let Some(input_cred) = req.credentials else { return Err(pool_admin_missing_credentials_error("cancel decommission")); }; @@ -366,10 +460,12 @@ impl Operation for CancelDecommission { .await?; let Some(endpoints) = endpoints_from_context() else { + log_pool_request_rejected!("cancel_decommission", "not_implemented"); return Err(s3_error!(NotImplemented)); }; if endpoints.legacy() { + log_pool_request_rejected!("cancel_decommission", "legacy_endpoints_not_supported"); return Err(s3_error!(NotImplemented)); } @@ -394,7 +490,6 @@ impl Operation for CancelDecommission { }; let Some(idx) = has_idx else { - warn!("specified pool {} not found, please specify a valid pool", &query.pool); return Err(pool_admin_pool_not_found_error("cancel decommission", &query.pool)); }; @@ -408,6 +503,7 @@ impl Operation for CancelDecommission { .map_err(ApiError::from) .map_err(|err| contextualize_admin_pool_api_error(err, "cancel decommission", format!("pool {idx}")))?; + log_pool_response_emitted!("cancel_decommission"); Ok(S3Response::new((StatusCode::OK, Body::default()))) } } diff --git a/rustfs/src/admin/handlers/system.rs b/rustfs/src/admin/handlers/system.rs index ae40483ad..d3ce440ce 100644 --- a/rustfs/src/admin/handlers/system.rs +++ b/rustfs/src/admin/handlers/system.rs @@ -24,7 +24,55 @@ use matchit::Params; use rustfs_policy::policy::action::{Action, AdminAction, S3Action}; use s3s::header::CONTENT_TYPE; use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error}; -use tracing::warn; +use tracing::{error, info, warn}; + +const LOG_COMPONENT_ADMIN_API: &str = "admin_api"; +const LOG_SUBSYSTEM_SYSTEM_ADMIN: &str = "system_admin"; +const EVENT_ADMIN_REQUEST_REJECTED: &str = "admin_request_rejected"; +const EVENT_ADMIN_REQUEST_FAILED: &str = "admin_request_failed"; +const EVENT_ADMIN_RESPONSE_EMITTED: &str = "admin_response_emitted"; + +macro_rules! log_system_request_rejected { + ($operation:expr, $reason:expr) => { + warn!( + event = EVENT_ADMIN_REQUEST_REJECTED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_SYSTEM_ADMIN, + operation = $operation, + result = "rejected", + reason = $reason, + "admin request rejected" + ); + }; +} + +macro_rules! log_system_request_failed { + ($operation:expr, $reason:expr, $err:expr) => { + error!( + event = EVENT_ADMIN_REQUEST_FAILED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_SYSTEM_ADMIN, + operation = $operation, + result = "failed", + reason = $reason, + error = %$err, + "admin request failed" + ); + }; +} + +macro_rules! log_system_response_emitted { + ($operation:expr) => { + info!( + event = EVENT_ADMIN_RESPONSE_EMITTED, + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_SYSTEM_ADMIN, + operation = $operation, + result = "success", + "admin response emitted" + ); + }; +} pub fn register_system_route(r: &mut S3Router) -> std::io::Result<()> { r.insert( @@ -71,14 +119,12 @@ pub fn register_system_route(r: &mut S3Router) -> std::io::Resul Ok(()) } - pub struct ServiceHandle {} #[async_trait::async_trait] impl Operation for ServiceHandle { async fn call(&self, _req: S3Request, _params: Params<'_, '_>) -> S3Result> { - warn!("handle ServiceHandle"); - + log_system_request_rejected!("service_handle", "not_implemented"); Err(s3_error!(NotImplemented)) } } @@ -89,6 +135,7 @@ pub struct ServerInfoHandler {} impl Operation for ServerInfoHandler { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { let Some(input_cred) = req.credentials else { + log_system_request_rejected!("query_server_info", "missing_credentials"); return Err(s3_error!(InvalidRequest, "get cred failed")); }; @@ -113,11 +160,14 @@ impl Operation for ServerInfoHandler { .map_err(S3Error::from)? .info; - let data = serde_json::to_vec(&info) - .map_err(|_e| S3Error::with_message(S3ErrorCode::InternalError, "parse serverInfo failed"))?; + let data = serde_json::to_vec(&info).map_err(|e| { + log_system_request_failed!("query_server_info", "serialize_server_info_failed", e); + S3Error::with_message(S3ErrorCode::InternalError, "parse serverInfo failed") + })?; let mut header = HeaderMap::new(); header.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); + log_system_response_emitted!("query_server_info"); Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), header)) } @@ -128,8 +178,7 @@ pub struct InspectDataHandler {} #[async_trait::async_trait] impl Operation for InspectDataHandler { async fn call(&self, _req: S3Request, _params: Params<'_, '_>) -> S3Result> { - warn!("handle InspectDataHandler"); - + log_system_request_rejected!("inspect_data", "not_implemented"); Err(s3_error!(NotImplemented)) } } @@ -139,9 +188,8 @@ pub struct StorageInfoHandler {} #[async_trait::async_trait] impl Operation for StorageInfoHandler { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { - warn!("handle StorageInfoHandler"); - let Some(input_cred) = req.credentials else { + log_system_request_rejected!("query_storage_info", "missing_credentials"); return Err(s3_error!(InvalidRequest, "get cred failed")); }; @@ -162,11 +210,14 @@ impl Operation for StorageInfoHandler { let usecase = DefaultAdminUsecase::from_global(); let info = usecase.execute_query_storage_info().await.map_err(S3Error::from)?; - let data = serde_json::to_vec(&info) - .map_err(|_e| S3Error::with_message(S3ErrorCode::InternalError, "failed to serialize storage info"))?; + let data = serde_json::to_vec(&info).map_err(|e| { + log_system_request_failed!("query_storage_info", "serialize_storage_info_failed", e); + S3Error::with_message(S3ErrorCode::InternalError, "failed to serialize storage info") + })?; let mut header = HeaderMap::new(); header.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); + log_system_response_emitted!("query_storage_info"); Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), header)) } @@ -177,9 +228,8 @@ pub struct DataUsageInfoHandler {} #[async_trait::async_trait] impl Operation for DataUsageInfoHandler { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { - warn!("handle DataUsageInfoHandler"); - let Some(input_cred) = req.credentials else { + log_system_request_rejected!("query_data_usage_info", "missing_credentials"); return Err(s3_error!(InvalidRequest, "get cred failed")); }; @@ -203,11 +253,14 @@ impl Operation for DataUsageInfoHandler { let usecase = DefaultAdminUsecase::from_global(); let info = usecase.execute_query_data_usage_info().await.map_err(S3Error::from)?; - let data = serde_json::to_vec(&info) - .map_err(|_e| S3Error::with_message(S3ErrorCode::InternalError, "parse DataUsageInfo failed"))?; + let data = serde_json::to_vec(&info).map_err(|e| { + log_system_request_failed!("query_data_usage_info", "serialize_data_usage_info_failed", e); + S3Error::with_message(S3ErrorCode::InternalError, "parse DataUsageInfo failed") + })?; let mut header = HeaderMap::new(); header.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); + log_system_response_emitted!("query_data_usage_info"); Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), header)) } diff --git a/rustfs/src/storage/rpc/http_service.rs b/rustfs/src/storage/rpc/http_service.rs index ddb58f4f2..b57191bb4 100644 --- a/rustfs/src/storage/rpc/http_service.rs +++ b/rustfs/src/storage/rpc/http_service.rs @@ -39,14 +39,181 @@ use std::task::{Context, Poll}; use tokio::io::{self, AsyncWriteExt}; use tokio_util::io::ReaderStream; use tower::Service; -use tracing::warn; +use tracing::{error, warn}; type BoxError = Box; type RpcErrorResponse = Box>; +const LOG_COMPONENT_INTERNODE_RPC: &str = "internode_rpc"; +const LOG_SUBSYSTEM_FILE_TRANSFER: &str = "file_transfer"; +const LOG_SUBSYSTEM_DIRECTORY_WALK: &str = "directory_walk"; +const LOG_SUBSYSTEM_ROUTING: &str = "routing"; +const EVENT_RPC_REQUEST_REJECTED: &str = "rpc_request_rejected"; +const EVENT_RPC_REQUEST_FAILED: &str = "rpc_request_failed"; +const EVENT_RPC_BACKGROUND_TASK_FAILED: &str = "rpc_background_task_failed"; +const RPC_OPERATION_UNKNOWN: &str = "unknown"; const READ_FILE_STREAM_PATH: &str = "/rustfs/rpc/read_file_stream"; const PUT_FILE_STREAM_PATH: &str = "/rustfs/rpc/put_file_stream"; const WALK_DIR_PATH: &str = "/rustfs/rpc/walk_dir"; +macro_rules! log_internode_rpc_response_failure { + ($status:expr, $rpc_path:expr, $method:expr, $operation:expr, $reason:expr, $result:expr, Some(($context_key:expr, $context_value:expr)), Some($error_text:expr)) => {{ + let operation = $operation.unwrap_or(RPC_OPERATION_UNKNOWN); + let subsystem = internode_rpc_subsystem(Some(operation)); + if $status.is_server_error() { + error!( + event = EVENT_RPC_REQUEST_FAILED, + component = LOG_COMPONENT_INTERNODE_RPC, + subsystem, + operation, + result = $result, + status_code = $status.as_u16(), + rpc_path = $rpc_path, + method = %$method, + reason = $reason, + $context_key = $context_value, + error = %$error_text, + "internode rpc request failed" + ); + } else { + warn!( + event = EVENT_RPC_REQUEST_REJECTED, + component = LOG_COMPONENT_INTERNODE_RPC, + subsystem, + operation, + result = $result, + status_code = $status.as_u16(), + rpc_path = $rpc_path, + method = %$method, + reason = $reason, + $context_key = $context_value, + error = %$error_text, + "internode rpc request rejected" + ); + } + }}; + ($status:expr, $rpc_path:expr, $method:expr, $operation:expr, $reason:expr, $result:expr, Some(($context_key:expr, $context_value:expr)), None) => {{ + let operation = $operation.unwrap_or(RPC_OPERATION_UNKNOWN); + let subsystem = internode_rpc_subsystem(Some(operation)); + if $status.is_server_error() { + error!( + event = EVENT_RPC_REQUEST_FAILED, + component = LOG_COMPONENT_INTERNODE_RPC, + subsystem, + operation, + result = $result, + status_code = $status.as_u16(), + rpc_path = $rpc_path, + method = %$method, + reason = $reason, + $context_key = $context_value, + "internode rpc request failed" + ); + } else { + warn!( + event = EVENT_RPC_REQUEST_REJECTED, + component = LOG_COMPONENT_INTERNODE_RPC, + subsystem, + operation, + result = $result, + status_code = $status.as_u16(), + rpc_path = $rpc_path, + method = %$method, + reason = $reason, + $context_key = $context_value, + "internode rpc request rejected" + ); + } + }}; + ($status:expr, $rpc_path:expr, $method:expr, $operation:expr, $reason:expr, $result:expr, None, Some($error_text:expr)) => {{ + let operation = $operation.unwrap_or(RPC_OPERATION_UNKNOWN); + let subsystem = internode_rpc_subsystem(Some(operation)); + if $status.is_server_error() { + error!( + event = EVENT_RPC_REQUEST_FAILED, + component = LOG_COMPONENT_INTERNODE_RPC, + subsystem, + operation, + result = $result, + status_code = $status.as_u16(), + rpc_path = $rpc_path, + method = %$method, + reason = $reason, + error = %$error_text, + "internode rpc request failed" + ); + } else { + warn!( + event = EVENT_RPC_REQUEST_REJECTED, + component = LOG_COMPONENT_INTERNODE_RPC, + subsystem, + operation, + result = $result, + status_code = $status.as_u16(), + rpc_path = $rpc_path, + method = %$method, + reason = $reason, + error = %$error_text, + "internode rpc request rejected" + ); + } + }}; + ($status:expr, $rpc_path:expr, $method:expr, $operation:expr, $reason:expr, $result:expr, None, None) => {{ + let operation = $operation.unwrap_or(RPC_OPERATION_UNKNOWN); + let subsystem = internode_rpc_subsystem(Some(operation)); + if $status.is_server_error() { + error!( + event = EVENT_RPC_REQUEST_FAILED, + component = LOG_COMPONENT_INTERNODE_RPC, + subsystem, + operation, + result = $result, + status_code = $status.as_u16(), + rpc_path = $rpc_path, + method = %$method, + reason = $reason, + "internode rpc request failed" + ); + } else { + warn!( + event = EVENT_RPC_REQUEST_REJECTED, + component = LOG_COMPONENT_INTERNODE_RPC, + subsystem, + operation, + result = $result, + status_code = $status.as_u16(), + rpc_path = $rpc_path, + method = %$method, + reason = $reason, + "internode rpc request rejected" + ); + } + }}; +} + +macro_rules! log_internode_put_file_stage_failure { + ($stage:expr, $query:expr, $err:expr) => { + error!( + event = EVENT_RPC_REQUEST_FAILED, + component = LOG_COMPONENT_INTERNODE_RPC, + subsystem = LOG_SUBSYSTEM_FILE_TRANSFER, + operation = INTERNODE_OPERATION_PUT_FILE_STREAM, + result = "failed", + status_code = StatusCode::INTERNAL_SERVER_ERROR.as_u16(), + rpc_path = PUT_FILE_STREAM_PATH, + method = %Method::PUT, + reason = "put_file_stage_failed", + stage = $stage, + disk = %$query.disk, + volume = %$query.volume, + path = %$query.path, + append = $query.append, + size = $query.size, + error = %$err, + "internode rpc request failed" + ); + }; +} + #[derive(Clone)] pub struct InternodeRpcService { inner: S, @@ -157,10 +324,18 @@ fn verify_internode_rpc_signature(uri: &Uri, method: &Method, headers: &HeaderMa } verify_rpc_signature(&uri.to_string(), method, headers).map_err(|e| { - Box::new(response_with_status( + let message = format!("rpc signature verification failed: {e}"); + log_internode_rpc_response_failure!( StatusCode::FORBIDDEN, - format!("rpc signature verification failed: {e}"), - )) + uri.path(), + method, + internode_http_operation(uri.path()), + "signature_verification_failed", + "rejected", + None, + Some(&e) + ); + Box::new(response_with_status(StatusCode::FORBIDDEN, message)) }) } @@ -175,6 +350,23 @@ async fn handle_read_file(req: Request) -> Response { }; let Some(disk) = find_local_disk_by_ref(&query.disk).await else { + warn!( + event = EVENT_RPC_REQUEST_REJECTED, + component = LOG_COMPONENT_INTERNODE_RPC, + subsystem = LOG_SUBSYSTEM_FILE_TRANSFER, + operation = INTERNODE_OPERATION_READ_FILE_STREAM, + result = "rejected", + status_code = StatusCode::BAD_REQUEST.as_u16(), + rpc_path = req.uri().path(), + method = %req.method(), + reason = "disk_not_found", + disk = %query.disk, + volume = %query.volume, + path = %query.path, + offset = query.offset, + length = query.length, + "internode rpc request rejected" + ); return response_with_status(StatusCode::BAD_REQUEST, "disk not found"); }; @@ -183,7 +375,28 @@ async fn handle_read_file(req: Request) -> Response { .await { Ok(file) => file, - Err(e) => return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, format!("read file err {e}")), + Err(e) => { + let message = format!("read file err {e}"); + error!( + event = EVENT_RPC_REQUEST_FAILED, + component = LOG_COMPONENT_INTERNODE_RPC, + subsystem = LOG_SUBSYSTEM_FILE_TRANSFER, + operation = INTERNODE_OPERATION_READ_FILE_STREAM, + result = "failed", + status_code = StatusCode::INTERNAL_SERVER_ERROR.as_u16(), + rpc_path = req.uri().path(), + method = %req.method(), + reason = "read_file_failed", + disk = %query.disk, + volume = %query.volume, + path = %query.path, + offset = query.offset, + length = query.length, + error = %e, + "internode rpc request failed" + ); + return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, message); + } }; global_internode_metrics().record_incoming_request_for_operation_and_backend( @@ -230,23 +443,94 @@ async fn handle_walk_dir(req: Request) -> Response { }; let Some(disk) = find_local_disk_by_ref(&query.disk).await else { + warn!( + event = EVENT_RPC_REQUEST_REJECTED, + component = LOG_COMPONENT_INTERNODE_RPC, + subsystem = LOG_SUBSYSTEM_DIRECTORY_WALK, + operation = INTERNODE_OPERATION_WALK_DIR, + result = "rejected", + status_code = StatusCode::BAD_REQUEST.as_u16(), + rpc_path = req.uri().path(), + method = %req.method(), + reason = "disk_not_found", + disk = %query.disk, + "internode rpc request rejected" + ); return response_with_status(StatusCode::BAD_REQUEST, "disk not found"); }; let body = match Limited::new(req.into_body(), MAX_ADMIN_REQUEST_BODY_SIZE).collect().await { Ok(body) => body.to_bytes(), - Err(e) => return response_with_status(StatusCode::PAYLOAD_TOO_LARGE, format!("read body err {e}")), + Err(e) => { + let message = format!("read body err {e}"); + log_internode_rpc_response_failure!( + StatusCode::PAYLOAD_TOO_LARGE, + WALK_DIR_PATH, + &Method::GET, + Some(INTERNODE_OPERATION_WALK_DIR), + "request_body_read_failed", + "rejected", + Some(("disk", query.disk.as_str())), + Some(&e) + ); + return response_with_status(StatusCode::PAYLOAD_TOO_LARGE, message); + } }; let args: WalkDirOptions = match serde_json::from_slice(&body) { Ok(args) => args, - Err(e) => return response_with_status(StatusCode::BAD_REQUEST, format!("unmarshal body err {e}")), + Err(e) => { + let message = format!("unmarshal body err {e}"); + warn!( + event = EVENT_RPC_REQUEST_REJECTED, + component = LOG_COMPONENT_INTERNODE_RPC, + subsystem = LOG_SUBSYSTEM_DIRECTORY_WALK, + operation = INTERNODE_OPERATION_WALK_DIR, + result = "rejected", + status_code = StatusCode::BAD_REQUEST.as_u16(), + rpc_path = WALK_DIR_PATH, + method = %Method::GET, + reason = "request_body_decode_failed", + disk = %query.disk, + error = %e, + "internode rpc request rejected" + ); + return response_with_status(StatusCode::BAD_REQUEST, message); + } }; + let log_disk = query.disk.clone(); + let log_bucket = args.bucket.clone(); + let log_base_dir = args.base_dir.clone(); + let log_recursive = args.recursive; + let log_report_notfound = args.report_notfound; + let log_filter_prefix = args.filter_prefix.clone(); + let log_forward_to = args.forward_to.clone(); + let log_limit = args.limit; + let log_disk_id = args.disk_id.clone(); + let log_skip_total_timeout = args.skip_total_timeout; let (rd, mut wd) = tokio::io::duplex(DEFAULT_READ_BUFFER_SIZE); spawn_traced(async move { if let Err(e) = disk.walk_dir(args, &mut wd).await { - warn!(error = %e, "walk_dir failed"); + warn!( + event = EVENT_RPC_BACKGROUND_TASK_FAILED, + component = LOG_COMPONENT_INTERNODE_RPC, + subsystem = LOG_SUBSYSTEM_DIRECTORY_WALK, + operation = INTERNODE_OPERATION_WALK_DIR, + result = "failed", + disk = %log_disk, + bucket = %log_bucket, + base_dir = %log_base_dir, + recursive = log_recursive, + report_notfound = log_report_notfound, + filter_prefix = ?log_filter_prefix, + forward_to = ?log_forward_to, + limit = log_limit, + disk_id = %log_disk_id, + skip_total_timeout = log_skip_total_timeout, + error = %e, + "internode rpc background task failed" + ); } }); @@ -269,12 +553,24 @@ async fn handle_walk_dir(req: Request) -> Response { } async fn handle_put_file(req: Request) -> Response { + let method = req.method().clone(); + let path = req.uri().path().to_string(); let query = match parse_query::(&req) { Ok(query) => query, Err(response) => return *response, }; let Some(disk) = find_local_disk_by_ref(&query.disk).await else { + log_internode_rpc_response_failure!( + StatusCode::BAD_REQUEST, + &path, + &method, + Some(INTERNODE_OPERATION_PUT_FILE_STREAM), + "disk_not_found", + "rejected", + Some(("disk", query.disk.as_str())), + None + ); return response_with_status(StatusCode::BAD_REQUEST, "disk not found"); }; @@ -283,7 +579,7 @@ async fn handle_put_file(req: Request) -> Response { Ok(file) => file, Err(e) => { let message = put_file_stage_error_message("append", &query, &e); - warn!("{message}"); + log_internode_put_file_stage_failure!("append", query, e); return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, message); } } @@ -292,7 +588,7 @@ async fn handle_put_file(req: Request) -> Response { Ok(file) => file, Err(e) => { let message = put_file_stage_error_message("create", &query, &e); - warn!("{message}"); + log_internode_put_file_stage_failure!("create", query, e); return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, message); } } @@ -302,7 +598,7 @@ async fn handle_put_file(req: Request) -> Response { Ok(copied) => copied, Err(e) => { let message = put_file_stage_error_message("write_body", &query, &e); - warn!("{message}"); + log_internode_put_file_stage_failure!("write_body", query, e); return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, message); } }; @@ -319,7 +615,7 @@ async fn handle_put_file(req: Request) -> Response { if let Err(e) = file.flush().await { let message = put_file_stage_error_message("flush", &query, &e); - warn!("{message}"); + log_internode_put_file_stage_failure!("flush", query, e); return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, message); } @@ -358,8 +654,20 @@ where T: DeserializeOwned + Default, { match req.uri().query() { - Some(query) => from_bytes(query.as_bytes()) - .map_err(|e| Box::new(response_with_status(StatusCode::BAD_REQUEST, format!("get query failed {e}")))), + Some(query) => from_bytes(query.as_bytes()).map_err(|e| { + let message = format!("get query failed {e}"); + log_internode_rpc_response_failure!( + StatusCode::BAD_REQUEST, + req.uri().path(), + req.method(), + internode_http_operation(req.uri().path()), + "query_parse_failed", + "rejected", + None, + Some(&e) + ); + Box::new(response_with_status(StatusCode::BAD_REQUEST, message)) + }), None => Ok(T::default()), } } @@ -379,6 +687,14 @@ fn response_with_status(status: StatusCode, message: impl Into) -> Respo .expect("failed to build rpc error response") } +fn internode_rpc_subsystem(operation: Option<&'static str>) -> &'static str { + match operation { + Some(INTERNODE_OPERATION_WALK_DIR) => LOG_SUBSYSTEM_DIRECTORY_WALK, + Some(INTERNODE_OPERATION_READ_FILE_STREAM | INTERNODE_OPERATION_PUT_FILE_STREAM) => LOG_SUBSYSTEM_FILE_TRANSFER, + _ => LOG_SUBSYSTEM_ROUTING, + } +} + fn put_file_stage_error_message(stage: &str, query: &PutFileQuery, err: &dyn std::fmt::Display) -> String { format!( "{stage} file err {err} [disk={}, volume={}, path={}, append={}, size={}]", @@ -445,6 +761,20 @@ mod tests { assert!(msg.contains("size=1024")); } + #[test] + fn internode_rpc_subsystem_matches_known_operations() { + assert_eq!( + internode_rpc_subsystem(Some(INTERNODE_OPERATION_READ_FILE_STREAM)), + LOG_SUBSYSTEM_FILE_TRANSFER + ); + assert_eq!( + internode_rpc_subsystem(Some(INTERNODE_OPERATION_PUT_FILE_STREAM)), + LOG_SUBSYSTEM_FILE_TRANSFER + ); + assert_eq!(internode_rpc_subsystem(Some(INTERNODE_OPERATION_WALK_DIR)), LOG_SUBSYSTEM_DIRECTORY_WALK); + assert_eq!(internode_rpc_subsystem(None), LOG_SUBSYSTEM_ROUTING); + } + #[tokio::test] async fn write_body_chunks_to_writer_streams_all_chunks() { let (mut reader, mut writer) = tokio::io::duplex(64); diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index ad05d667b..c24d7448f 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -60,6 +60,71 @@ use tracing::{debug, error, info, warn}; const LOG_COMPONENT_STORAGE: &str = "storage"; const LOG_SUBSYSTEM_RPC: &str = "rpc"; +const LOG_SUBSYSTEM_REBALANCE: &str = "rebalance"; +const EVENT_RPC_REQUEST_REJECTED: &str = "rpc_request_rejected"; +const EVENT_RPC_REQUEST_FAILED: &str = "rpc_request_failed"; +const EVENT_RPC_RESPONSE_EMITTED: &str = "rpc_response_emitted"; +const EVENT_RPC_BACKGROUND_TASK_SPAWNED: &str = "rpc_background_task_spawned"; +const EVENT_RPC_BACKGROUND_TASK_FAILED: &str = "rpc_background_task_failed"; + +macro_rules! log_load_rebalance_meta_rejected { + ($reason:expr, $start_rebalance:expr) => { + warn!( + event = EVENT_RPC_REQUEST_REJECTED, + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + operation = "load_rebalance_meta", + result = "rejected", + reason = $reason, + start_rebalance = $start_rebalance, + "node rpc request rejected" + ); + }; +} + +macro_rules! log_load_rebalance_meta_failed { + ($reason:expr, $start_rebalance:expr, $err:expr) => { + error!( + event = EVENT_RPC_REQUEST_FAILED, + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + operation = "load_rebalance_meta", + result = "failed", + reason = $reason, + start_rebalance = $start_rebalance, + error = %$err, + "node rpc request failed" + ); + }; +} + +macro_rules! log_load_rebalance_meta_response_emitted { + ($start_rebalance:expr) => { + info!( + event = EVENT_RPC_RESPONSE_EMITTED, + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + operation = "load_rebalance_meta", + result = "success", + start_rebalance = $start_rebalance, + "node rpc response emitted" + ); + }; +} + +macro_rules! log_background_rebalance_task_spawned { + ($start_rebalance:expr) => { + info!( + event = EVENT_RPC_BACKGROUND_TASK_SPAWNED, + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + operation = "start_rebalance", + state = "spawned", + start_rebalance = $start_rebalance, + "node rpc background task spawned" + ); + }; +} type ResponseStream = Pin> + Send>>; @@ -940,30 +1005,36 @@ impl Node for NodeService { &self, request: Request, ) -> Result, Status> { + let LoadRebalanceMetaRequest { start_rebalance } = request.into_inner(); let Some(store) = resolve_object_store_handle() else { + log_load_rebalance_meta_rejected!("server_not_initialized", start_rebalance); return Ok(Response::new(LoadRebalanceMetaResponse { success: false, error_info: Some("errServerNotInitialized".to_string()), })); }; - let LoadRebalanceMetaRequest { start_rebalance } = request.into_inner(); - - info!("handling load_rebalance_meta request"); - store.load_rebalance_meta().await.map_err(|err| { - error!(error = ?err, "load_rebalance_meta failed"); + log_load_rebalance_meta_failed!("load_rebalance_meta_failed", start_rebalance, err); Status::internal(err.to_string()) })?; - - info!("load_rebalance_meta completed"); + log_load_rebalance_meta_response_emitted!(start_rebalance); if start_rebalance { - info!(start_rebalance, "spawning background rebalance task"); + log_background_rebalance_task_spawned!(start_rebalance); let store = store.clone(); spawn(async move { if let Some(message) = background_rebalance_start_error_message(store.start_rebalance().await) { - error!(error = %message, "background rebalance start failed"); + error!( + event = EVENT_RPC_BACKGROUND_TASK_FAILED, + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_REBALANCE, + operation = "start_rebalance", + state = "failed", + start_rebalance, + error = %message, + "node rpc background task failed" + ); } }); } diff --git a/scripts/check_logging_guardrails.sh b/scripts/check_logging_guardrails.sh index b4ff5afe7..d3a3fb9db 100755 --- a/scripts/check_logging_guardrails.sh +++ b/scripts/check_logging_guardrails.sh @@ -20,6 +20,12 @@ checked_files=( "rustfs/src/admin/handlers/quota.rs" "rustfs/src/admin/handlers/rebalance.rs" "rustfs/src/admin/handlers/tier.rs" + "rustfs/src/admin/handlers/event.rs" + "rustfs/src/admin/handlers/audit.rs" + "rustfs/src/admin/handlers/pools.rs" + "rustfs/src/admin/handlers/system.rs" + "rustfs/src/storage/rpc/http_service.rs" + "rustfs/src/storage/rpc/node_service.rs" "crates/audit/src/pipeline.rs" "crates/audit/src/system.rs" "crates/audit/src/global.rs" @@ -223,6 +229,24 @@ forbidden_patterns=( 'warn!("handle RebalanceStop notification_sys load_rebalance_meta")' 'warn!("rebalance stop propagation failed after local state update: {err}")' 'warn!("handle RebalanceStop notification_sys load_rebalance_meta done")' + 'warn!("handle ListPools")' + 'warn!("handle StatusPool")' + 'warn!("handle StartDecommission")' + 'warn!("handle CancelDecommission")' + 'warn!("specified pool {} not found, please specify a valid pool", &query.pool)' + 'warn!("handle ServiceHandle")' + 'warn!("handle InspectDataHandler")' + 'warn!("handle StorageInfoHandler")' + 'warn!("handle DataUsageInfoHandler")' + 'warn!("failed to read request body: {:?}")' + 'info!("Setting target config for type '\''{}'\'', name '\''{}'\''", target_type, target_name)' + 'info!("Removing target config for type '\''{}'\'', name '\''{}'\''", target_type, target_name)' + 'warn!("{message}")' + 'warn!(error = %e, "walk_dir failed")' + 'info!("handling load_rebalance_meta request")' + 'info!("load_rebalance_meta completed")' + 'info!(start_rebalance, "spawning background rebalance task")' + 'error!(error = %message, "background rebalance start failed")' 'warn!("get body failed, e: {:?}")' 'debug!("add tier args {:?}")' 'warn!("parse force failed, e: {:?}")' @@ -235,10 +259,7 @@ forbidden_patterns=( 'warn!("tier_config_mgr rand: {}")' 'warn!("tier_config_mgr clear failed, e: {:?}")' 'warn!("tier_config_mgr save failed, e: {:?}")' - 'warn!( - component = LOG_COMPONENT_ADMIN, - subsystem = LOG_SUBSYSTEM_OBJECT_LAMBDA, - event = "object_lambda_tls_verification_disabled"' + 'event = "object_lambda_tls_verification_disabled"' 'info!("FTP system initialized successfully"' 'debug!("FTP system is disabled")' 'info!("FTP system disabled"'