From ec7f5f7b7d1464ac5474402d7a1a2718aec8b5fc Mon Sep 17 00:00:00 2001 From: houseme Date: Sun, 9 Aug 2026 19:35:55 +0800 Subject: [PATCH] perf(http): reduce tracing/logging hotpath overhead (#5893) perf(http): reduce disabled tracing overhead Co-authored-by: heihutu --- rustfs/src/server/http.rs | 225 +++++++++++++++---------------------- rustfs/src/server/layer.rs | 142 +++++++++++++---------- 2 files changed, 172 insertions(+), 195 deletions(-) diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index 62ddb3736..6739e895a 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -91,7 +91,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, trace, warn}; +use tracing::{Level, Span, debug, error, info, instrument, trace, warn}; use tracing_opentelemetry::OpenTelemetrySpanExt; const LABEL_HTTP_METHOD: &str = "method"; @@ -429,6 +429,72 @@ fn trace_on_response(response: &Response, latency: Duration, s } } +fn make_http_trace_span(request: &HttpRequest) -> Span { + if !tracing::enabled!(Level::INFO) { + return Span::none(); + } + + 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| propagator.extract(&HeaderMapCarrier::new(request.headers()))); + + if parent_context.has_active_span() { + let span_ref = parent_context.span(); + 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 { + trace!("No trace context found in request headers, will create root span"); + } + + let client_info = request.extensions().get::(); + 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(), + peer_addr = %peer_addr, + uri = %redact_sensitive_uri_query(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) { + debug!(component = LOG_COMPONENT_SERVER, subsystem = LOG_SUBSYSTEM_HTTP, error = ?e, "Failed to propagate tracing context"); + } + for (header_name, header_value) in request.headers() { + 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); + } + } + + span +} + pub async fn start_http_server( config: &config::Config, readiness: Arc, @@ -1379,8 +1445,8 @@ fn process_connection( // 10. KeystoneAuthLayer — X-Auth-Token validation // 11. TraceLayer — request span creation + metrics // 12. RequestLoggingLayer — single completion event per request - // 13. CompressionLayer — response compression (whitelist, path-aware) - // 14. PathCategoryInjectionLayer — injects path category for compression predicate + // 13. CompressionLayer — response compression predicate (whitelist, path-aware) + // 14. PathCategoryInjectionLayer — injects path category when compression is enabled // 15. S3ErrorMessageCompatLayer — missing S3 error message compatibility // 16. IcebergRestErrorCompatLayer — Iceberg REST JSON error compatibility // 17. ObjectAttributesEtagFixLayer — ETag fix for GetObjectAttributes @@ -1431,72 +1497,7 @@ fn process_connection( .layer(InFlightLayer) .layer( TraceLayer::new_for_http() - .make_span_with(|request: &HttpRequest<_>| { - 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| { - propagator.extract(&HeaderMapCarrier::new(request.headers())) - }); - - if parent_context.has_active_span() { - let span_ref = parent_context.span(); - 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 { - trace!("No trace context found in request headers, will create root span"); - } - let client_info = request.extensions().get::(); - 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(), - peer_addr = %peer_addr, - uri = %redact_sensitive_uri_query(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) { - debug!(component = LOG_COMPONENT_SERVER, subsystem = LOG_SUBSYSTEM_HTTP, error = ?e, "Failed to propagate tracing context"); - } - for (header_name, header_value) in request.headers() { - 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); - } - } - - span - }) + .make_span_with(make_http_trace_span) .on_request(|request: &HttpRequest<_>, span: &Span| { let _enter = span.enter(); trace!("HTTP request started"); @@ -1560,7 +1561,7 @@ fn process_connection( ) .layer(RequestLoggingLayer) .layer(CompressionLayer::new().compress_when(PathAwareHttpCompressionPredicate::new(compression_config.clone()))) - .layer(PathCategoryInjectionLayer) + .option_layer(compression_config.enabled.then_some(PathCategoryInjectionLayer)) .layer(S3ErrorMessageCompatLayer) .layer(IcebergRestErrorCompatLayer) .layer(ObjectAttributesEtagFixLayer) @@ -1591,72 +1592,7 @@ fn process_connection( .layer(InFlightLayer) .layer( TraceLayer::new_for_http() - .make_span_with(|request: &HttpRequest<_>| { - 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| { - propagator.extract(&HeaderMapCarrier::new(request.headers())) - }); - - if parent_context.has_active_span() { - let span_ref = parent_context.span(); - 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 { - trace!("No trace context found in request headers, will create root span"); - } - let client_info = request.extensions().get::(); - 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(), - peer_addr = %peer_addr, - uri = %redact_sensitive_uri_query(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) { - debug!(component = LOG_COMPONENT_SERVER, subsystem = LOG_SUBSYSTEM_HTTP, error = ?e, "Failed to propagate tracing context"); - } - for (header_name, header_value) in request.headers() { - 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); - } - } - - span - }) + .make_span_with(make_http_trace_span) .on_request(|request: &HttpRequest<_>, span: &Span| { let _enter = span.enter(); trace!("HTTP request started"); @@ -1720,7 +1656,7 @@ fn process_connection( ) .layer(PropagateRequestIdLayer::x_request_id()) .layer(CompressionLayer::new().compress_when(PathAwareHttpCompressionPredicate::new(compression_config.clone()))) - .layer(PathCategoryInjectionLayer) + .option_layer(compression_config.enabled.then_some(PathCategoryInjectionLayer)) .layer(S3ErrorMessageCompatLayer) .layer(IcebergRestErrorCompatLayer) .layer(ObjectAttributesEtagFixLayer) @@ -2102,6 +2038,21 @@ mod tests { assert_eq!(baseline::HTTP1_MAX_BUF_SIZE, DEFAULT_HTTP1_MAX_BUF_SIZE); } + #[test] + fn http_trace_span_is_empty_when_info_is_disabled() { + let subscriber = tracing_subscriber::fmt().with_max_level(Level::ERROR).finish(); + let _guard = tracing::subscriber::set_default(subscriber); + let request = HttpRequest::builder() + .method(Method::GET) + .uri("/bucket/object.txt") + .body(()) + .expect("request"); + + let span = make_http_trace_span(&request); + + assert!(span.is_disabled()); + } + #[test] fn test_baseline_middleware_count() { assert_eq!(baseline::MIDDLEWARE_LAYER_COUNT, 22); diff --git a/rustfs/src/server/layer.rs b/rustfs/src/server/layer.rs index fcd2c8923..c6c889e53 100644 --- a/rustfs/src/server/layer.rs +++ b/rustfs/src/server/layer.rs @@ -45,13 +45,14 @@ use s3s::S3ErrorCode; use serde::{Deserialize, Serialize}; use std::borrow::Cow; use std::future::Future; +use std::net::{IpAddr, SocketAddr}; use std::pin::Pin; use std::sync::Arc; use std::task::{Context, Poll}; use std::time::{Duration, Instant}; use tokio_util::sync::CancellationToken; use tower::{Layer, Service}; -use tracing::{debug, error, info, warn}; +use tracing::{Level, debug, error, info, warn}; use url::form_urlencoded; const HTTP_REQUEST_COMPLETED_EVENT: &str = "http_request_completed"; @@ -330,9 +331,10 @@ struct RequestLogContext { request_id: String, trace_id: Option, span_id: Option, - peer_addr: String, - method: String, - uri: String, + client_ip: Option, + peer_addr: Option, + method: Method, + uri: Uri, request_started_at: Option, fallback_start: Instant, } @@ -344,20 +346,14 @@ impl RequestLogContext { .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: redact_sensitive_uri_query(req.uri()), + client_ip: req.extensions().get::().map(|info| info.real_ip), + peer_addr: req.extensions().get::().map(|addr| addr.0), + method: req.method().clone(), + uri: req.uri().clone(), request_started_at: request_context, fallback_start: Instant::now(), } @@ -382,6 +378,40 @@ impl RequestLogContext { } } + fn peer_addr(&self) -> String { + self.client_ip + .map(|addr| addr.to_string()) + .or_else(|| self.peer_addr.map(|addr| addr.to_string())) + .unwrap_or_else(|| "unknown".to_string()) + } + + fn redacted_uri(&self) -> String { + redact_sensitive_uri_query(&self.uri) + } + + fn log_slow_inflight(&self) { + if !tracing::enabled!(target: HTTP_SERVER_LOG_TARGET, Level::WARN) { + return; + } + warn!( + target: HTTP_SERVER_LOG_TARGET, + event = HTTP_REQUEST_INFLIGHT_SLOW_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.as_str(), + uri = %self.redacted_uri(), + duration_ms = self.duration_ms(), + active_requests = active_http_requests(), + threshold_ms = HTTP_REQUEST_INFLIGHT_WARN_THRESHOLD.as_millis() as u64, + state = "response_pending", + "HTTP request remains in flight" + ); + } + fn log_response(&self, response: &Response) { let duration_ms = self.duration_ms(); let status = response.status(); @@ -391,6 +421,9 @@ impl RequestLogContext { let span_id = self.span_id.as_deref().unwrap_or("unknown"); if status.is_server_error() { + if !tracing::enabled!(target: HTTP_SERVER_LOG_TARGET, Level::ERROR) { + return; + } error!( target: HTTP_SERVER_LOG_TARGET, event = HTTP_REQUEST_COMPLETED_EVENT, @@ -399,15 +432,18 @@ impl RequestLogContext { request_id = %self.request_id, trace_id = %trace_id, span_id = %span_id, - peer_addr = %self.peer_addr, - method = %self.method, - uri = %self.uri, + peer_addr = %self.peer_addr(), + method = %self.method.as_str(), + uri = %self.redacted_uri(), status_code, duration_ms, result, "HTTP request completed" ); } else { + if !tracing::enabled!(target: HTTP_SERVER_LOG_TARGET, Level::INFO) { + return; + } info!( target: HTTP_SERVER_LOG_TARGET, event = HTTP_REQUEST_COMPLETED_EVENT, @@ -416,9 +452,9 @@ impl RequestLogContext { request_id = %self.request_id, trace_id = %trace_id, span_id = %span_id, - peer_addr = %self.peer_addr, - method = %self.method, - uri = %self.uri, + peer_addr = %self.peer_addr(), + method = %self.method.as_str(), + uri = %self.redacted_uri(), status_code, duration_ms, result, @@ -439,9 +475,9 @@ impl RequestLogContext { 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, + peer_addr = %self.peer_addr(), + method = %self.method.as_str(), + uri = %self.redacted_uri(), duration_ms = self.duration_ms(), result = "service_error", error = %error, @@ -468,39 +504,27 @@ where fn call(&mut self, req: HttpRequest) -> Self::Future { let context = RequestLogContext::from_request(&req); let mut inner = self.inner.clone(); - let watchdog = CancellationToken::new(); - let watchdog_context = context.clone(); - - spawn_traced({ - let watchdog = watchdog.clone(); - async move { - tokio::select! { - _ = watchdog.cancelled() => {} - _ = tokio::time::sleep(HTTP_REQUEST_INFLIGHT_WARN_THRESHOLD) => { - warn!( - event = HTTP_REQUEST_INFLIGHT_SLOW_EVENT, - component = LOG_COMPONENT_SERVER, - subsystem = LOG_SUBSYSTEM_HTTP, - request_id = %watchdog_context.request_id, - trace_id = %watchdog_context.trace_id.as_deref().unwrap_or("unknown"), - span_id = %watchdog_context.span_id.as_deref().unwrap_or("unknown"), - peer_addr = %watchdog_context.peer_addr, - method = %watchdog_context.method, - uri = %watchdog_context.uri, - duration_ms = watchdog_context.duration_ms(), - active_requests = active_http_requests(), - threshold_ms = HTTP_REQUEST_INFLIGHT_WARN_THRESHOLD.as_millis() as u64, - state = "response_pending", - "HTTP request remains in flight" - ); + let watchdog = tracing::enabled!(target: HTTP_SERVER_LOG_TARGET, Level::WARN).then(CancellationToken::new); + if let Some(watchdog) = watchdog.as_ref() { + spawn_traced({ + let watchdog = watchdog.clone(); + let watchdog_context = context.clone(); + async move { + tokio::select! { + _ = watchdog.cancelled() => {} + _ = tokio::time::sleep(HTTP_REQUEST_INFLIGHT_WARN_THRESHOLD) => { + watchdog_context.log_slow_inflight(); + } } } - } - }); + }); + } Box::pin(async move { let result = inner.call(req).await; - watchdog.cancel(); + if let Some(watchdog) = watchdog { + watchdog.cancel(); + } match &result { Ok(response) => context.log_response(response), Err(error) => context.log_failure(error), @@ -4583,9 +4607,9 @@ mod tests { 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"); + assert_eq!(context.peer_addr(), "127.0.0.1:9000"); + assert_eq!(context.method.as_str(), "PUT"); + assert_eq!(context.redacted_uri(), "/bucket/object.txt"); } #[test] @@ -4598,8 +4622,9 @@ mod tests { let context = RequestLogContext::from_request(&request); - assert_eq!(context.uri, "/rustfs/admin/v3/object-zip-downloads/download-id.zip?token=redacted&part=1"); - assert!(!context.uri.contains("secret-token")); + let uri = context.redacted_uri(); + assert_eq!(uri, "/rustfs/admin/v3/object-zip-downloads/download-id.zip?token=redacted&part=1"); + assert!(!uri.contains("secret-token")); } #[test] @@ -4612,8 +4637,9 @@ mod tests { let context = RequestLogContext::from_request(&request); - assert_eq!(context.uri, "/minio/admin/v3/object-zip-downloads/download-id.zip?token=redacted&part=1"); - assert!(!context.uri.contains("secret-token")); + let uri = context.redacted_uri(); + assert_eq!(uri, "/minio/admin/v3/object-zip-downloads/download-id.zip?token=redacted&part=1"); + assert!(!uri.contains("secret-token")); } #[test]