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

refactor(logging): align API response boundary events (#3466)

Browse Source 
* refactor(logging): align api response boundary events

* refactor(logging): align heal admin boundary events

* refactor(logging): restore admin callsite line numbers

* refactor(logging): simplify audit target error logging

* refactor(logging): restore rpc callsite line numbers
This commit is contained in:
houseme
2026-06-15 14:07:54 +08:00
committed by GitHub
parent 218f473fb5
commit 73a48c06e5
8 changed files with 1078 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/admin/handlers/audit.rs
+154 -9
View File
@@ -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<AdminOperation>) -> 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<String> {
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")))
}
rustfs/src/admin/handlers/event.rs
+171 -21
View File
@@ -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<AdminOperation>) -> 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<String> {
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")))
}
rustfs/src/admin/handlers/heal.rs
+125 -13
View File
@@ -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<Vec<u8>>
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<rustfs_ecstore::error::Error>) -> 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<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
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<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
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))
}
rustfs/src/admin/handlers/pools.rs
+110 -14
View File
@@ -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<rustfs_ecstore::endpoints::EndpointServerPools> {
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<usize> {
let idx = pool.parse::<usize>().ok()?;
(idx < endpoint_count).then_some(idx)
@@ -148,8 +241,6 @@ impl Operation for ListPools {
// GET <endpoint>/<admin-API>/pools/list
#[tracing::instrument(skip_all)]
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
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 <endpoint>/<admin-API>/pools/status?pool=http://server{1...4}/disk{1...4}
#[tracing::instrument(skip_all)]
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
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 <endpoint>/<admin-API>/pools/decommission?pool=http://server{1...4}/disk{1...4}
#[tracing::instrument(skip_all)]
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
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 <endpoint>/<admin-API>/pools/cancel?pool=http://server{1...4}/disk{1...4}
#[tracing::instrument(skip_all)]
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
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())))
}
}
rustfs/src/admin/handlers/system.rs
+69 -16
View File
@@ -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<AdminOperation>) -> std::io::Result<()> {
r.insert(
@@ -71,14 +119,12 @@ pub fn register_system_route(r: &mut S3Router<AdminOperation>) -> std::io::Resul
Ok(())
}
pub struct ServiceHandle {}
#[async_trait::async_trait]
impl Operation for ServiceHandle {
async fn call(&self, _req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
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<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
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<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
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<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
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<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
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))
}
rustfs/src/storage/rpc/http_service.rs
+344 -14
View File
@@ -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<dyn std::error::Error + Send + Sync + 'static>;
type RpcErrorResponse = Box<Response<Body>>;
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<S> {
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<Incoming>) -> Response<Body> {
};
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<Incoming>) -> Response<Body> {
.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<Incoming>) -> Response<Body> {
};
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<Incoming>) -> Response<Body> {
}
async fn handle_put_file(req: Request<Incoming>) -> Response<Body> {
let method = req.method().clone();
let path = req.uri().path().to_string();
let query = match parse_query::<PutFileQuery>(&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<Incoming>) -> Response<Body> {
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<Incoming>) -> Response<Body> {
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<Incoming>) -> Response<Body> {
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<Incoming>) -> Response<Body> {
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<String>) -> 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);
rustfs/src/storage/rpc/node_service.rs
+80 -9
View File
@@ -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<T> = Pin<Box<dyn Stream<Item = Result<T, Status>> + Send>>;
@@ -940,30 +1005,36 @@ impl Node for NodeService {
&self,
request: Request<LoadRebalanceMetaRequest>,
) -> Result<Response<LoadRebalanceMetaResponse>, 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"
);
}
});
}
scripts/check_logging_guardrails.sh
+25 -4
View File
@@ -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"'