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(server): consolidate request transport logs (#3354)

Browse Source
This commit is contained in:
houseme
2026-06-11 17:07:02 +08:00
committed by GitHub
parent ca58d7f0ec
commit 704c95a22d
3 changed files with 569 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/server/http.rs
+184 -79
View File
@@ -23,7 +23,8 @@ use crate::server::{
hybrid::hybrid,
layer::{
BodylessStatusFixLayer, ConditionalCorsLayer, EmptyBodyContentLengthCompatLayer, HeadRequestBodyFixLayer,
ObjectAttributesEtagFixLayer, PublicHealthEndpointLayer, RedirectLayer, RequestContextLayer, S3ErrorMessageCompatLayer,
ObjectAttributesEtagFixLayer, PublicHealthEndpointLayer, RedirectLayer, RequestContextLayer, RequestLoggingLayer,
S3ErrorMessageCompatLayer,
},
tls_material::{
TlsAcceptorHolder, TlsHandshakeFailureKind, build_acceptor_from_loaded, load_tls_material, spawn_reload_loop,
@@ -66,7 +67,7 @@ use tower_http::catch_panic::CatchPanicLayer;
use tower_http::compression::CompressionLayer;
use tower_http::request_id::{MakeRequestUuid, PropagateRequestIdLayer, SetRequestIdLayer};
use tower_http::trace::TraceLayer;
use tracing::{Span, debug, error, info, instrument, warn};
use tracing::{Span, debug, error, info, instrument, trace, warn};
use tracing_opentelemetry::OpenTelemetrySpanExt;
const LABEL_HTTP_METHOD: &str = "method";
@@ -79,6 +80,13 @@ const METRIC_HTTP_SERVER_REQUEST_BODY_BYTES_TOTAL: &str = "rustfs_http_server_re
const METRIC_HTTP_SERVER_REQUEST_BODY_SIZE_BYTES: &str = "rustfs_http_server_request_body_size_bytes";
const METRIC_HTTP_SERVER_RESPONSE_BODY_BYTES_TOTAL: &str = "rustfs_http_server_response_body_bytes_total";
const METRIC_HTTP_SERVER_RESPONSE_BODY_SIZE_BYTES: &str = "rustfs_http_server_response_body_size_bytes";
const LOG_COMPONENT_SERVER: &str = "server";
const LOG_SUBSYSTEM_HTTP: &str = "http";
const LOG_SUBSYSTEM_TRANSPORT: &str = "transport";
const LOG_SUBSYSTEM_TLS: &str = "tls";
const EVENT_TLS_HANDSHAKE_FAILED: &str = "tls_handshake_failed";
const EVENT_HTTP_TRANSPORT_CLOSED: &str = "http_transport_closed";
const EVENT_HTTP_TRANSPORT_FAILED: &str = "http_transport_failed";
static ACTIVE_HTTP_REQUESTS: AtomicU64 = AtomicU64::new(0);
@@ -110,6 +118,78 @@ fn status_class_label(status: http::StatusCode) -> &'static str {
}
}
#[inline]
fn duration_ms(duration: Duration) -> u64 {
duration.as_millis().try_into().unwrap_or(u64::MAX)
}
fn log_tls_handshake_failure(peer_addr: &str, kind: TlsHandshakeFailureKind, err: &dyn std::fmt::Display) {
match kind {
TlsHandshakeFailureKind::UnexpectedEof => {
debug!(
event = EVENT_TLS_HANDSHAKE_FAILED,
component = LOG_COMPONENT_SERVER,
subsystem = LOG_SUBSYSTEM_TLS,
peer_addr = %peer_addr,
failure_type = kind.as_str(),
error = %err,
result = "client_disconnect",
"TLS handshake failed"
);
}
TlsHandshakeFailureKind::ProtocolVersion | TlsHandshakeFailureKind::Certificate | TlsHandshakeFailureKind::Alert => {
warn!(
event = EVENT_TLS_HANDSHAKE_FAILED,
component = LOG_COMPONENT_SERVER,
subsystem = LOG_SUBSYSTEM_TLS,
peer_addr = %peer_addr,
failure_type = kind.as_str(),
error = %err,
result = "client_error",
"TLS handshake failed"
);
}
TlsHandshakeFailureKind::Unknown => {
error!(
event = EVENT_TLS_HANDSHAKE_FAILED,
component = LOG_COMPONENT_SERVER,
subsystem = LOG_SUBSYSTEM_TLS,
peer_addr = %peer_addr,
failure_type = kind.as_str(),
error = %err,
result = "transport_error",
"TLS handshake failed"
);
}
}
}
fn log_transport_closed(peer_addr: &str, error_kind: &str, error_message: &str, result: &str) {
debug!(
event = EVENT_HTTP_TRANSPORT_CLOSED,
component = LOG_COMPONENT_SERVER,
subsystem = LOG_SUBSYSTEM_TRANSPORT,
peer_addr = %peer_addr,
error_kind,
error = %error_message,
result,
"HTTP transport closed"
);
}
fn log_transport_failed(peer_addr: &str, error_kind: &str, error_message: &str) {
warn!(
event = EVENT_HTTP_TRANSPORT_FAILED,
component = LOG_COMPONENT_SERVER,
subsystem = LOG_SUBSYSTEM_TRANSPORT,
peer_addr = %peer_addr,
error_kind,
error = %error_message,
result = "transport_error",
"HTTP transport failed"
);
}
#[inline]
fn record_active_http_requests(delta: i64) {
let next = if delta >= 0 {
@@ -423,7 +503,7 @@ pub async fn start_http_server(config: &config::Config, readiness: Arc<GlobalRea
debug!("graceful initiated");
loop {
debug!("Waiting for new connection...");
trace!("Waiting for new connection");
let (socket, _) = {
#[cfg(unix)]
{
@@ -665,17 +745,18 @@ fn process_connection(
// 7. CatchPanicLayer — panic → 500
// 8. ReadinessGateLayer — blocks until ready
// 9. KeystoneAuthLayer — X-Auth-Token validation
// 10. TraceLayer — request/response tracing + metrics
// 11. PropagateRequestIdLayer — X-Request-ID → response
// 12. CompressionLayer — response compression (whitelist, path-aware)
// 13. PathCategoryInjectionLayer — injects path category for compression predicate
// 14. S3ErrorMessageCompatLayer — missing S3 error message compatibility
// 15. ObjectAttributesEtagFixLayer — ETag fix for GetObjectAttributes
// 16. ConditionalCorsLayer — S3 API CORS
// 17. RedirectLayer — console redirect (conditional)
// 18. BodylessStatusFixLayer — clears body for 1xx/204/205/304 responses
// 19. HeadRequestBodyFixLayer — strips actual body bytes from HEAD responses
// 20. PublicHealthEndpointLayer — handles public health before s3s host parsing
// 10. TraceLayer — request span creation + metrics
// 11. RequestLoggingLayer — single completion event per request
// 12. PropagateRequestIdLayer — X-Request-ID → response
// 13. CompressionLayer — response compression (whitelist, path-aware)
// 14. PathCategoryInjectionLayer — injects path category for compression predicate
// 15. S3ErrorMessageCompatLayer — missing S3 error message compatibility
// 16. ObjectAttributesEtagFixLayer — ETag fix for GetObjectAttributes
// 17. ConditionalCorsLayer — S3 API CORS
// 18. RedirectLayer — console redirect (conditional)
// 19. BodylessStatusFixLayer — clears body for 1xx/204/205/304 responses
// 20. HeadRequestBodyFixLayer — strips actual body bytes from HEAD responses
// 21. PublicHealthEndpointLayer — handles public health before s3s host parsing
// ─────────────────────────────────────────────────────────────
let hybrid_service = ServiceBuilder::new()
// NOTE: Both extension types are intentionally inserted to maintain compatibility:
@@ -705,10 +786,15 @@ fn process_connection(
.layer(
TraceLayer::new_for_http()
.make_span_with(|request: &HttpRequest<_>| {
let request_id = request
.headers()
.get(http::header::HeaderName::from_static("x-request-id"))
.and_then(|v| v.to_str().ok())
let request_context = request.extensions().get::<crate::storage::request_context::RequestContext>();
let request_id = request_context
.map(|ctx| ctx.request_id.as_str())
.unwrap_or("unknown");
let trace_id = request_context
.and_then(|ctx| ctx.trace_id.as_deref())
.unwrap_or("unknown");
let span_id = request_context
.and_then(|ctx| ctx.span_id.as_deref())
.unwrap_or("unknown");
let parent_context = global::get_text_map_propagator(|propagator| {
@@ -718,38 +804,49 @@ fn process_connection(
// Log trace context extraction for debugging distributed tracing
if parent_context.has_active_span() {
let span_ref = parent_context.span();
debug!(
trace!(
otel_trace_id = %span_ref.span_context().trace_id(),
otel_parent_span_id = %span_ref.span_context().span_id(),
sampled = span_ref.span_context().is_sampled(),
"Extracted trace context from incoming request headers"
);
} else {
debug!("No trace context found in request headers, will create root span");
trace!("No trace context found in request headers, will create root span");
}
// Extract real client IP from trusted proxy middleware if available
let client_info = request.extensions().get::<ClientInfo>();
let real_ip = client_info
let peer_addr = client_info
.map(|info| info.real_ip.to_string())
.or_else(|| request.extensions().get::<RemoteAddr>().map(|addr| addr.0.to_string()))
.unwrap_or_else(|| "unknown".to_string());
let span = tracing::info_span!("http-request",
request_id = %request_id,
trace_id = %trace_id,
span_id = %span_id,
status_code = tracing::field::Empty,
method = %request.method(),
real_ip = %real_ip,
peer_addr = %peer_addr,
uri = %request.uri(),
version = ?request.version(),
user_agent = tracing::field::Empty,
content_type = tracing::field::Empty,
content_length = tracing::field::Empty,
);
if span.is_disabled() {
return span;
}
if let Err(e) = span.set_parent(parent_context) {
warn!("Failed to propagate tracing context: `{:?}`", e);
debug!(component = LOG_COMPONENT_SERVER, subsystem = LOG_SUBSYSTEM_HTTP, error = ?e, "Failed to propagate tracing context");
}
for (header_name, header_value) in request.headers() {
if header_name == "user-agent" || header_name == "content-type" || header_name == "content-length" {
span.record(header_name.as_str(), header_value.to_str().unwrap_or("invalid"));
let value = header_value.to_str().unwrap_or("invalid");
if header_name == "user-agent" {
span.record("user_agent", value);
} else if header_name == "content-type" {
span.record("content_type", value);
} else if header_name == "content-length" {
span.record("content_length", value);
}
}
@@ -757,7 +854,7 @@ fn process_connection(
})
.on_request(|request: &HttpRequest<_>, span: &Span| {
let _enter = span.enter();
debug!("http started method: {}, url path: {}", request.method(), request.uri().path());
trace!("HTTP request started");
let method = request_method_label(request.method());
record_active_http_requests(1);
counter!(
@@ -803,14 +900,13 @@ fn process_connection(
)
.record(len as f64);
}
debug!("http response generated in {:?}", latency)
})
.on_body_chunk(|chunk: &Bytes, latency: Duration, span: &Span| {
counter!(METRIC_HTTP_SERVER_RESPONSE_BODY_BYTES_TOTAL).increment(chunk.len() as u64);
#[cfg(feature = "tracing-chunk-debug")]
{
let _enter = span.enter();
debug!("http body sending {} bytes in {:?}", chunk.len(), latency);
debug!(chunk_bytes = chunk.len(), duration_ms = duration_ms(latency), "HTTP response body chunk sent");
}
#[cfg(not(feature = "tracing-chunk-debug"))]
{
@@ -821,7 +917,7 @@ fn process_connection(
#[cfg(feature = "tracing-chunk-debug")]
{
let _enter = span.enter();
debug!("http stream closed after {:?}", stream_duration);
debug!(duration_ms = duration_ms(stream_duration), "HTTP response stream closed");
}
#[cfg(not(feature = "tracing-chunk-debug"))]
{
@@ -836,9 +932,10 @@ fn process_connection(
LABEL_HTTP_STATUS_CLASS => "transport"
)
.increment(1);
debug!("http request failure error: {:?} in {:?}", _error, latency)
trace!(error = ?_error, duration_ms = duration_ms(latency), "HTTP request failure captured by trace layer");
}),
)
.layer(RequestLoggingLayer)
.layer(PropagateRequestIdLayer::x_request_id())
// Compress responses based on whitelist configuration
// Only compresses when enabled and matches configured extensions/MIME types
@@ -872,7 +969,7 @@ fn process_connection(
// Decide whether to handle HTTPS or HTTP connections based on the existence of TLS Acceptor
if let Some(holder) = tls_acceptor {
debug!("TLS handshake start");
trace!("TLS handshake start");
let peer_addr = socket
.peer_addr()
.ok()
@@ -880,7 +977,7 @@ fn process_connection(
let acceptor = holder.get();
match acceptor.accept(socket).await {
Ok(tls_socket) => {
debug!("TLS handshake successful");
trace!("TLS handshake successful");
let stream = TokioIo::new(tls_socket);
let conn = http_server.serve_connection(stream, hybrid_service);
if let Err(err) = graceful.watch(conn).await {
@@ -890,25 +987,9 @@ fn process_connection(
Err(err) => {
let err_str = err.to_string();
let kind = TlsHandshakeFailureKind::classify(&err_str);
match kind {
TlsHandshakeFailureKind::UnexpectedEof => {
warn!(peer_addr = %peer_addr, "TLS handshake failed (unexpected EOF). If this client needs HTTP, it should connect to the HTTP port instead");
}
TlsHandshakeFailureKind::ProtocolVersion => {
error!(peer_addr = %peer_addr, "TLS handshake failed (protocol version mismatch): {}", err);
}
TlsHandshakeFailureKind::Certificate => {
error!(peer_addr = %peer_addr, "TLS handshake failed (certificate issue): {}", err);
}
TlsHandshakeFailureKind::Alert => {
error!(peer_addr = %peer_addr, "TLS handshake failed (alert): {}", err);
}
TlsHandshakeFailureKind::Unknown => {
error!(peer_addr = %peer_addr, "TLS handshake failed: {}", err);
}
}
log_tls_handshake_failure(&peer_addr, kind, &err);
counter!("rustfs_tls_handshake_failures", &[("failure_type", kind.as_str())]).increment(1);
debug!(
trace!(
peer_addr = %peer_addr,
error_type = %std::any::type_name_of_val(&err),
error_details = %err,
@@ -918,61 +999,85 @@ fn process_connection(
return;
}
}
debug!("TLS handshake success");
trace!("TLS handshake success");
} else {
debug!("Http handshake start");
trace!("HTTP connection handling start");
let peer_addr = socket.peer_addr().ok().map(|addr| addr.to_string());
let stream = TokioIo::new(socket);
let conn = http_server.serve_connection(stream, hybrid_service);
if let Err(err) = graceful.watch(conn).await {
handle_connection_error(peer_addr.as_deref(), &*err);
}
debug!("Http handshake success");
trace!("HTTP connection handling finished");
};
});
}
/// Handles connection errors by logging them with appropriate severity
fn handle_connection_error(peer_addr: Option<&str>, err: &(dyn std::error::Error + 'static)) {
let peer_addr = peer_addr.unwrap_or("unknown");
let s = err.to_string();
if s.contains("connection reset") || s.contains("broken pipe") {
warn!(
peer_addr = %peer_addr.unwrap_or("unknown"),
"The connection was reset by the peer or broken pipe: {}", s
);
// Ignore common non-fatal errors
log_transport_closed(peer_addr, "connection_reset", &s, "client_disconnect");
return;
}
if let Some(hyper_err) = err.downcast_ref::<hyper::Error>() {
if hyper_err.is_incomplete_message() {
warn!(
peer_addr = %peer_addr.unwrap_or("unknown"),
"The HTTP connection is closed prematurely and the message is not completed:{}", hyper_err
);
log_transport_closed(peer_addr, "incomplete_message", &hyper_err.to_string(), "client_disconnect");
} else if hyper_err.is_closed() {
warn!(peer_addr = %peer_addr.unwrap_or("unknown"), "The HTTP connection is closed:{}", hyper_err);
log_transport_closed(peer_addr, "connection_closed", &hyper_err.to_string(), "client_disconnect");
} else if hyper_err.is_parse() {
error!(peer_addr = %peer_addr.unwrap_or("unknown"), "HTTP message parsing failed:{}", hyper_err);
log_transport_failed(peer_addr, "parse_failure", &hyper_err.to_string());
} else if hyper_err.is_user() {
error!(peer_addr = %peer_addr.unwrap_or("unknown"), "HTTP user-custom error:{}", hyper_err);
error!(
event = EVENT_HTTP_TRANSPORT_FAILED,
component = LOG_COMPONENT_SERVER,
subsystem = LOG_SUBSYSTEM_TRANSPORT,
peer_addr = %peer_addr,
error_kind = "service_error",
error = %hyper_err,
result = "transport_error",
"HTTP transport failed"
);
} else if hyper_err.is_canceled() {
warn!(
peer_addr = %peer_addr.unwrap_or("unknown"),
"The HTTP connection is canceled:{}", hyper_err
);
log_transport_closed(peer_addr, "canceled", &hyper_err.to_string(), "client_disconnect");
} else if format!("{:?}", hyper_err).contains("HeaderTimeout") {
info!(
peer_addr = %peer_addr.unwrap_or("unknown"),
"The HTTP connection timed out while reading request headers (HeaderTimeout): {}", hyper_err
);
log_transport_closed(peer_addr, "header_timeout", &hyper_err.to_string(), "client_timeout");
} else {
error!(peer_addr = %peer_addr.unwrap_or("unknown"), "Unknown hyper error:{:?}", hyper_err);
error!(
event = EVENT_HTTP_TRANSPORT_FAILED,
component = LOG_COMPONENT_SERVER,
subsystem = LOG_SUBSYSTEM_TRANSPORT,
peer_addr = %peer_addr,
error_kind = "hyper_error",
error = ?hyper_err,
result = "transport_error",
"HTTP transport failed"
);
}
} else if let Some(io_err) = err.downcast_ref::<Error>() {
error!(peer_addr = %peer_addr.unwrap_or("unknown"), "Unknown connection IO error:{}", io_err);
error!(
event = EVENT_HTTP_TRANSPORT_FAILED,
component = LOG_COMPONENT_SERVER,
subsystem = LOG_SUBSYSTEM_TRANSPORT,
peer_addr = %peer_addr,
error_kind = "io_error",
error = %io_err,
result = "transport_error",
"HTTP transport failed"
);
} else {
error!(peer_addr = %peer_addr.unwrap_or("unknown"), "Unknown connection error type:{:?}", err);
error!(
event = EVENT_HTTP_TRANSPORT_FAILED,
component = LOG_COMPONENT_SERVER,
subsystem = LOG_SUBSYSTEM_TRANSPORT,
peer_addr = %peer_addr,
error_kind = "unknown_error",
error = ?err,
result = "transport_error",
"HTTP transport failed"
);
}
}
@@ -1097,8 +1202,8 @@ mod tests {
};
/// Number of middleware layers in the canonical stack order (see http.rs).
/// Layers 1-2 are per-connection (AddExtension), 3-15 are stateless.
pub const MIDDLEWARE_LAYER_COUNT: usize = 15;
/// Layers 1-2 are per-connection (AddExtension), 3-21 are stateless.
pub const MIDDLEWARE_LAYER_COUNT: usize = 21;
/// Current HTTP/2 defaults (from rustfs_config).
pub const H2_INITIAL_STREAM_WINDOW_SIZE: u32 = DEFAULT_H2_INITIAL_STREAM_WINDOW_SIZE;
@@ -1139,7 +1244,7 @@ mod tests {
#[test]
fn test_baseline_middleware_count() {
assert_eq!(baseline::MIDDLEWARE_LAYER_COUNT, 15);
assert_eq!(baseline::MIDDLEWARE_LAYER_COUNT, 21);
}
#[test]
rustfs/src/server/layer.rs
+374 -2
View File
@@ -15,6 +15,7 @@
use crate::admin::console::is_console_path;
use crate::admin::handlers::health::{HealthProbe, build_health_response_parts};
use crate::error::ApiError;
use crate::server::RemoteAddr;
use crate::server::cors;
use crate::server::hybrid::HybridBody;
use crate::server::{
@@ -31,6 +32,7 @@ use http_body_util::BodyExt;
use hyper::body::Incoming;
use opentelemetry::global;
use opentelemetry::trace::TraceContextExt;
use rustfs_trusted_proxies::ClientInfo;
use rustfs_utils::get_env_opt_str;
use rustfs_utils::http::headers::AMZ_REQUEST_ID;
use s3s::S3ErrorCode;
@@ -40,7 +42,12 @@ use std::sync::Arc;
use std::task::{Context, Poll};
use std::time::Instant;
use tower::{Layer, Service};
use tracing::debug;
use tracing::{debug, error, info};
const HTTP_REQUEST_COMPLETED_EVENT: &str = "http_request_completed";
const HTTP_REQUEST_FAILED_EVENT: &str = "http_request_failed";
const LOG_COMPONENT_SERVER: &str = "server";
const LOG_SUBSYSTEM_HTTP: &str = "http";
/// A carrier that adapts [`HeaderMap`] for OpenTelemetry trace context propagation.
struct HeaderMapCarrier<'a>(&'a HeaderMap);
@@ -152,6 +159,174 @@ where
}
}
#[derive(Clone, Default)]
pub struct RequestLoggingLayer;
impl<S> Layer<S> for RequestLoggingLayer {
type Service = RequestLoggingService<S>;
fn layer(&self, inner: S) -> Self::Service {
RequestLoggingService { inner }
}
}
#[derive(Clone)]
pub struct RequestLoggingService<S> {
inner: S,
}
#[derive(Clone, Debug)]
struct RequestLogContext {
request_id: String,
trace_id: Option<String>,
span_id: Option<String>,
peer_addr: String,
method: String,
uri: String,
request_started_at: Option<RequestContext>,
fallback_start: Instant,
}
impl RequestLogContext {
fn from_request<B>(req: &HttpRequest<B>) -> Self {
let request_context = req.extensions().get::<RequestContext>().cloned();
let request_id = request_context
.as_ref()
.map(|ctx| ctx.request_id.clone())
.unwrap_or_else(|| extract_request_id_from_headers(req.headers()));
let peer_addr = req
.extensions()
.get::<ClientInfo>()
.map(|info| info.real_ip.to_string())
.or_else(|| req.extensions().get::<RemoteAddr>().map(|addr| addr.0.to_string()))
.unwrap_or_else(|| "unknown".to_string());
Self {
request_id,
trace_id: request_context.as_ref().and_then(|ctx| ctx.trace_id.clone()),
span_id: request_context.as_ref().and_then(|ctx| ctx.span_id.clone()),
peer_addr,
method: req.method().to_string(),
uri: req.uri().to_string(),
request_started_at: request_context,
fallback_start: Instant::now(),
}
}
fn duration_ms(&self) -> u64 {
self.request_started_at
.as_ref()
.map(RequestContext::duration_ms)
.unwrap_or_else(|| self.fallback_start.elapsed().as_millis().try_into().unwrap_or(u64::MAX))
}
fn result_label(status: StatusCode) -> &'static str {
if status.is_server_error() {
"server_error"
} else if status.is_client_error() {
"client_error"
} else if status.is_redirection() {
"redirect"
} else {
"success"
}
}
fn log_response<ResBody>(&self, response: &Response<ResBody>) {
let duration_ms = self.duration_ms();
let status = response.status();
let status_code = status.as_u16();
let result = Self::result_label(status);
let trace_id = self.trace_id.as_deref().unwrap_or("unknown");
let span_id = self.span_id.as_deref().unwrap_or("unknown");
if status.is_server_error() {
error!(
event = HTTP_REQUEST_COMPLETED_EVENT,
component = LOG_COMPONENT_SERVER,
subsystem = LOG_SUBSYSTEM_HTTP,
request_id = %self.request_id,
trace_id = %trace_id,
span_id = %span_id,
peer_addr = %self.peer_addr,
method = %self.method,
uri = %self.uri,
status_code,
duration_ms,
result,
"HTTP request completed"
);
} else {
info!(
event = HTTP_REQUEST_COMPLETED_EVENT,
component = LOG_COMPONENT_SERVER,
subsystem = LOG_SUBSYSTEM_HTTP,
request_id = %self.request_id,
trace_id = %trace_id,
span_id = %span_id,
peer_addr = %self.peer_addr,
method = %self.method,
uri = %self.uri,
status_code,
duration_ms,
result,
"HTTP request completed"
);
}
}
fn log_failure<E>(&self, error: &E)
where
E: std::fmt::Display,
{
error!(
event = HTTP_REQUEST_FAILED_EVENT,
component = LOG_COMPONENT_SERVER,
subsystem = LOG_SUBSYSTEM_HTTP,
request_id = %self.request_id,
trace_id = %self.trace_id.as_deref().unwrap_or("unknown"),
span_id = %self.span_id.as_deref().unwrap_or("unknown"),
peer_addr = %self.peer_addr,
method = %self.method,
uri = %self.uri,
duration_ms = self.duration_ms(),
result = "service_error",
error = %error,
"HTTP request failed before a response was produced"
);
}
}
impl<S, B, ResBody> Service<HttpRequest<B>> for RequestLoggingService<S>
where
S: Service<HttpRequest<B>, Response = Response<ResBody>> + Clone + Send + 'static,
S::Future: Send + 'static,
S::Error: std::fmt::Display + Send + 'static,
B: Send + 'static,
{
type Response = Response<ResBody>;
type Error = S::Error;
type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.inner.poll_ready(cx)
}
fn call(&mut self, req: HttpRequest<B>) -> Self::Future {
let context = RequestLogContext::from_request(&req);
let mut inner = self.inner.clone();
Box::pin(async move {
let result = inner.call(req).await;
match &result {
Ok(response) => context.log_response(response),
Err(error) => context.log_failure(error),
}
result
})
}
}
/// Redirect layer that redirects browser requests to the console
#[derive(Clone)]
pub struct RedirectLayer;
@@ -1145,16 +1320,18 @@ where
#[cfg(test)]
mod tests {
use super::*;
use crate::server::{FAVICON_PATH, LICENSE, VERSION};
use crate::server::{FAVICON_PATH, LICENSE, RemoteAddr, VERSION};
use futures::future::{Ready, ready};
use http::Request;
use http_body_util::BodyExt;
use http_body_util::Full;
use serial_test::serial;
use std::convert::Infallible;
use std::io::{self, Write};
use std::sync::Mutex;
use std::sync::atomic::{AtomicUsize, Ordering};
use temp_env::{async_with_vars, with_var};
use tracing_subscriber::{Registry, fmt::MakeWriter, layer::SubscriberExt};
#[derive(Clone, Debug)]
struct CaptureService;
@@ -2379,4 +2556,199 @@ mod tests {
);
assert_eq!(response_headers.get(cors::response::ACCESS_CONTROL_ALLOW_METHODS).unwrap(), "GET");
}
#[derive(Clone)]
struct StatusService {
status: StatusCode,
}
impl StatusService {
fn new(status: StatusCode) -> Self {
Self { status }
}
}
impl<B> Service<Request<B>> for StatusService {
type Response = Response<Full<Bytes>>;
type Error = Infallible;
type Future = Ready<Result<Self::Response, Self::Error>>;
fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
fn call(&mut self, _req: Request<B>) -> Self::Future {
ready(Ok(Response::builder()
.status(self.status)
.body(Full::from(Bytes::new()))
.expect("response")))
}
}
#[derive(Clone, Default)]
struct SharedWriter {
buffer: Arc<Mutex<Vec<u8>>>,
}
struct SharedWriterGuard {
buffer: Arc<Mutex<Vec<u8>>>,
}
impl Write for SharedWriterGuard {
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
self.buffer.lock().expect("log buffer").extend_from_slice(buf);
Ok(buf.len())
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}
impl<'writer> MakeWriter<'writer> for SharedWriter {
type Writer = SharedWriterGuard;
fn make_writer(&'writer self) -> Self::Writer {
SharedWriterGuard {
buffer: self.buffer.clone(),
}
}
}
#[test]
fn request_log_context_classifies_statuses() {
assert_eq!(RequestLogContext::result_label(StatusCode::OK), "success");
assert_eq!(RequestLogContext::result_label(StatusCode::TEMPORARY_REDIRECT), "redirect");
assert_eq!(RequestLogContext::result_label(StatusCode::BAD_REQUEST), "client_error");
assert_eq!(RequestLogContext::result_label(StatusCode::INTERNAL_SERVER_ERROR), "server_error");
}
#[test]
fn request_log_context_prefers_request_context_and_remote_addr_extensions() {
let mut request = Request::builder()
.method(Method::PUT)
.uri("/bucket/object.txt")
.body(())
.expect("request");
request.extensions_mut().insert(RequestContext {
request_id: "req-ctx".to_string(),
x_amz_request_id: "amz-ctx".to_string(),
trace_id: Some("trace-123".to_string()),
span_id: Some("span-456".to_string()),
start_time: Instant::now(),
});
request
.extensions_mut()
.insert(RemoteAddr("127.0.0.1:9000".parse().expect("socket addr")));
let context = RequestLogContext::from_request(&request);
assert_eq!(context.request_id, "req-ctx");
assert_eq!(context.trace_id.as_deref(), Some("trace-123"));
assert_eq!(context.span_id.as_deref(), Some("span-456"));
assert_eq!(context.peer_addr, "127.0.0.1:9000");
assert_eq!(context.method, "PUT");
assert_eq!(context.uri, "/bucket/object.txt");
}
#[test]
fn request_logging_layer_emits_single_completion_event_with_standard_fields() {
let writer = SharedWriter::default();
let captured = writer.buffer.clone();
let subscriber = Registry::default().with(
tracing_subscriber::fmt::layer()
.without_time()
.with_target(false)
.with_level(false)
.with_ansi(false)
.with_writer(writer),
);
tracing::subscriber::with_default(subscriber, || {
let mut service = tower::ServiceBuilder::new()
.layer(RequestContextLayer)
.layer(RequestLoggingLayer)
.service(StatusService::new(StatusCode::OK));
let mut request: Request<Full<Bytes>> = Request::builder()
.method(Method::GET)
.uri("/bucket/object.txt")
.header("x-request-id", "req-123")
.body(Full::from(Bytes::new()))
.expect("request");
request
.extensions_mut()
.insert(RemoteAddr("127.0.0.1:9000".parse().expect("socket addr")));
let response = futures::executor::block_on(service.call(request)).expect("response");
assert_eq!(response.status(), StatusCode::OK);
});
let output = String::from_utf8(captured.lock().expect("captured logs").clone()).expect("utf8 logs");
assert_eq!(output.matches("HTTP request completed").count(), 1, "{output}");
assert!(output.contains("event"), "{output}");
assert!(output.contains("http_request_completed"), "{output}");
assert!(output.contains("component"), "{output}");
assert!(output.contains("server"), "{output}");
assert!(output.contains("subsystem"), "{output}");
assert!(output.contains("http"), "{output}");
assert!(output.contains("request_id"), "{output}");
assert!(output.contains("req-123"), "{output}");
assert!(output.contains("peer_addr"), "{output}");
assert!(output.contains("127.0.0.1:9000"), "{output}");
assert!(output.contains("method"), "{output}");
assert!(output.contains("GET"), "{output}");
assert!(output.contains("uri"), "{output}");
assert!(output.contains("/bucket/object.txt"), "{output}");
assert!(output.contains("status_code"), "{output}");
assert!(output.contains("200"), "{output}");
assert!(output.contains("result"), "{output}");
assert!(output.contains("success"), "{output}");
assert!(output.contains("duration_ms"), "{output}");
}
#[test]
fn request_logging_layer_uses_request_context_trace_fields() {
let writer = SharedWriter::default();
let captured = writer.buffer.clone();
let subscriber = Registry::default().with(
tracing_subscriber::fmt::layer()
.without_time()
.with_target(false)
.with_level(false)
.with_ansi(false)
.with_writer(writer),
);
tracing::subscriber::with_default(subscriber, || {
let mut service = RequestLoggingLayer.layer(StatusService::new(StatusCode::INTERNAL_SERVER_ERROR));
let mut request = Request::builder()
.method(Method::GET)
.uri("/bucket/object.txt")
.body(())
.expect("request");
request.extensions_mut().insert(RequestContext {
request_id: "req-ctx".to_string(),
x_amz_request_id: "amz-ctx".to_string(),
trace_id: Some("trace-ctx".to_string()),
span_id: Some("span-ctx".to_string()),
start_time: Instant::now(),
});
request
.extensions_mut()
.insert(RemoteAddr("127.0.0.1:9000".parse().expect("socket addr")));
let response = futures::executor::block_on(service.call(request)).expect("response");
assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR);
});
let output = String::from_utf8(captured.lock().expect("captured logs").clone()).expect("utf8 logs");
assert!(output.contains("http_request_completed"), "{output}");
assert!(output.contains("req-ctx"), "{output}");
assert!(output.contains("trace-ctx"), "{output}");
assert!(output.contains("span-ctx"), "{output}");
assert!(output.contains("500"), "{output}");
assert!(output.contains("server_error"), "{output}");
}
}
rustfs/src/storage/request_context.rs
+11
View File
@@ -99,6 +99,11 @@ impl RequestContext {
start_time: Instant::now(),
}
}
/// Return the elapsed request lifetime in whole milliseconds.
pub fn duration_ms(&self) -> u64 {
self.start_time.elapsed().as_millis().try_into().unwrap_or(u64::MAX)
}
}
fn current_trace_context_ids() -> Option<(String, String)> {
@@ -217,6 +222,12 @@ mod tests {
});
}
#[test]
fn test_request_context_duration_ms_is_non_negative() {
let ctx = RequestContext::fallback();
assert!(ctx.duration_ms() <= 10);
}
#[test]
fn test_extract_request_id_from_x_request_id() {
let mut headers = HeaderMap::new();