gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-11 07:36:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(http): reduce tracing/logging hotpath overhead (#5893)

Browse Source 
perf(http): reduce disabled tracing overhead

Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
2026-08-09 19:35:55 +08:00
committed by GitHub
parent 73e4ef4dd4
commit ec7f5f7b7d
2 changed files with 172 additions and 195 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/server/http.rs
+88 -137
View File
@@ -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<ResBody>(response: &Response<ResBody>, latency: Duration, s
}
}
fn make_http_trace_span<ReqBody>(request: &HttpRequest<ReqBody>) -> Span {
if !tracing::enabled!(Level::INFO) {
return Span::none();
}
let request_context = request
.extensions()
.get::<crate::storage_api::server::http::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| 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::<ClientInfo>();
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(),
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<GlobalReadiness>,
@@ -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::<crate::storage_api::server::http::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| {
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::<ClientInfo>();
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(),
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::<crate::storage_api::server::http::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| {
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::<ClientInfo>();
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(),
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);
rustfs/src/server/layer.rs
+84 -58
View File
@@ -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<String>,
span_id: Option<String>,
peer_addr: String,
method: String,
uri: String,
client_ip: Option<IpAddr>,
peer_addr: Option<SocketAddr>,
method: Method,
uri: Uri,
request_started_at: Option<RequestContext>,
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::<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: redact_sensitive_uri_query(req.uri()),
client_ip: req.extensions().get::<ClientInfo>().map(|info| info.real_ip),
peer_addr: req.extensions().get::<RemoteAddr>().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<ResBody>(&self, response: &Response<ResBody>) {
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<B>) -> 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]