From 704c95a22d9106b75cde26bf6fcc3988a175a19b Mon Sep 17 00:00:00 2001 From: houseme Date: Thu, 11 Jun 2026 17:07:02 +0800 Subject: [PATCH] refactor(server): consolidate request transport logs (#3354) --- rustfs/src/server/http.rs | 263 ++++++++++++------ rustfs/src/server/layer.rs | 376 +++++++++++++++++++++++++- rustfs/src/storage/request_context.rs | 11 + 3 files changed, 569 insertions(+), 81 deletions(-) diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index 1ce05a4f6..e33e77315 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -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| { - 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::(); + 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::(); - let real_ip = client_info + let peer_addr = client_info .map(|info| info.real_ip.to_string()) + .or_else(|| request.extensions().get::().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::() { 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!(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] diff --git a/rustfs/src/server/layer.rs b/rustfs/src/server/layer.rs index c5ca00252..cf68f5910 100644 --- a/rustfs/src/server/layer.rs +++ b/rustfs/src/server/layer.rs @@ -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 Layer for RequestLoggingLayer { + type Service = RequestLoggingService; + + fn layer(&self, inner: S) -> Self::Service { + RequestLoggingService { inner } + } +} + +#[derive(Clone)] +pub struct RequestLoggingService { + inner: S, +} + +#[derive(Clone, Debug)] +struct RequestLogContext { + request_id: String, + trace_id: Option, + span_id: Option, + peer_addr: String, + method: String, + uri: String, + request_started_at: Option, + fallback_start: Instant, +} + +impl RequestLogContext { + fn from_request(req: &HttpRequest) -> Self { + let request_context = req.extensions().get::().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::() + .map(|info| info.real_ip.to_string()) + .or_else(|| req.extensions().get::().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(&self, response: &Response) { + 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(&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 Service> for RequestLoggingService +where + S: Service, Response = Response> + Clone + Send + 'static, + S::Future: Send + 'static, + S::Error: std::fmt::Display + Send + 'static, + B: Send + 'static, +{ + type Response = Response; + type Error = S::Error; + type Future = Pin> + Send>>; + + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + self.inner.poll_ready(cx) + } + + fn call(&mut self, req: HttpRequest) -> 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 Service> for StatusService { + type Response = Response>; + type Error = Infallible; + type Future = Ready>; + + fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn call(&mut self, _req: Request) -> Self::Future { + ready(Ok(Response::builder() + .status(self.status) + .body(Full::from(Bytes::new())) + .expect("response"))) + } + } + + #[derive(Clone, Default)] + struct SharedWriter { + buffer: Arc>>, + } + + struct SharedWriterGuard { + buffer: Arc>>, + } + + impl Write for SharedWriterGuard { + fn write(&mut self, buf: &[u8]) -> io::Result { + 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> = 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}"); + } } diff --git a/rustfs/src/storage/request_context.rs b/rustfs/src/storage/request_context.rs index b37a35fc4..1ab33edc3 100644 --- a/rustfs/src/storage/request_context.rs +++ b/rustfs/src/storage/request_context.rs @@ -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();