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

perf(server): lighten internode data-plane stack (#3735)

Browse Source 
* refactor(server): split internode dispatch scaffold

* test(server): cover internode dispatch prefix split

* refactor(server): name internode stack boundaries

* perf(server): skip internode request logging layer

* perf(server): skip internode trace layer

* perf(server): use lite internode request context

* feat(metrics): track internode rpc duration

* feat(ecstore): add put object stage summary logs

* test(metrics): update internode descriptor expectations

* fix(server): tighten internode path matching

* fix(pr): address review follow-up comments

* style(ecstore): simplify commit tail duration field

* refactor(ecstore): group put stage summary fields

* refactor(ecstore): inline put stage summary log

* fix(s3): return storage class for object attributes

* merge: sync latest main and resolve object attributes conflict

* fmt

* fix(server): remove duplicate rpc imports

* build(deps): bump memmap2 for RUSTSEC-2026-0186

* fix(s3select): align object_store with datafusion

* chore(deps): prune workspace dependencies

* perf(fuzz): optimize CI runtime with build/run split and matrix parallelization

  Separate fuzz harness compilation from execution to eliminate redundant
  builds across targets. Introduce matrix-based parallel execution for
  PR smoke and nightly fuzz jobs.

  Changes:

  - Split CI workflow into `fuzz-build` (compile once) and matrix run jobs
    (`pr-fuzz-smoke`, `nightly-fuzz-corpus`) that execute targets in parallel
  - Add `BUILD_ONLY` mode to run_ci_targets.sh / run_nightly_targets.sh
  - Add run_single_target.sh for matrix jobs (no build phase)
  - Optimize `local_metadata` fuzz target: reduce prefix iterations from
    8-10 (4 functions each) to 5 critical prefixes (parser-only), cutting
    per-iteration cost by ~3-5x
  - Move archive path validation (`validate_extract_relative_path`,
    `normalize_extract_entry_key`) from `rustfs` to `rustfs-utils::path`,
    eliminating `rustfs` binary crate dependency from fuzz harness
  - Remove `rustfs` from fuzz/Cargo.toml (drops significant transitive deps)
  - Add unit tests for archive path validation in rustfs-utils
  - Update fuzz/README.md with new workflow and script documentation

  Expected CI improvement: PR smoke wall-clock from ~120min (frequent
  timeout) to ~40min; nightly from ~180min to ~60min.

* refactor(fuzz): consolidate scripts and fix prefix test alignment

Replace three duplicated shell scripts (run_ci_targets.sh,
run_nightly_targets.sh, run_single_target.sh) with a single
parameterized run.sh that supports BUILD_ONLY, SKIP_BUILD, and
MAX_TOTAL_TIME environment variables.

Fix local_metadata fuzz target prefix testing: replace always-true
'len > 0' guard with lengths aligned to xl.meta binary layout
(4/5/8/12 bytes for magic+version+header fields). Remove redundant
empty-slice test.

Hoist RUSTFLAGS to workflow top-level env to eliminate per-job
duplication. Update README with unified script documentation.

Net: -118 lines, zero functionality loss.

* perf(fuzz): optimize CI runtime with build/run split and matrix parallelization

Restructure fuzz CI workflow to eliminate redundant compilation and
run targets in parallel via matrix strategy.

Workflow changes:
- Split into fuzz-build (compile once) and matrix run jobs
- PR smoke: 3 targets parallel, 60s each, timeout 30min (was 120min)
- Nightly: 3 targets parallel, 300s each, timeout 60min (was 180min)
- Pass compiled harness via actions/artifact between jobs
- Hoist RUSTFLAGS to workflow top-level env

Script consolidation:
- Replace 3 duplicated scripts with single parameterized run.sh
- Supports BUILD_ONLY, SKIP_BUILD, MAX_TOTAL_TIME env vars

Target optimizations:
- Remove rustfs binary crate from fuzz dependencies (was pulling
  979 transitive deps); move archive path validation to rustfs-utils
- Optimize local_metadata: reduce prefix iterations from 8-10x4
  calls to 5 prefixes with parser-only (no decompress), aligned
  with xl.meta binary layout (4/5/8/12 bytes)
- Add unit tests for archive path validation in rustfs-utils
- Update fuzz/README.md with unified script documentation

Expected: PR smoke wall-clock from ~120min (frequent timeout)
to ~40min; nightly from ~180min to ~60min.

* fix(rpc): resolve internode metrics via app context

* fmt

* ci: speed up fuzz smoke artifact restore

---------

Signed-off-by: houseme <housemecn@gmail.com>
This commit is contained in:
houseme
2026-06-23 21:36:39 +08:00
committed by GitHub
parent 7bdb25ae9d
commit 0a00d8d500
23 changed files with 1103 additions and 5816 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/app/object_usecase.rs
+53 -77
View File
@@ -14,12 +14,43 @@
//! Object application use-case contracts.
use super::object_api_utils::to_s3s_etag;
use super::quota::checker::QuotaChecker;
use super::storageclass;
// Performance metrics recording (with zero-copy-metrics integration)
use super::ECStore;
use super::{AppReplicationConfigExt as _, AppVersioningConfigExt as _, predict_lifecycle_expiration, validate_restore_request};
use super::{DiskError, is_all_buckets_not_found};
use super::{DynReader, HashReader, WritePlan, wrap_reader};
use super::{Error as EcstoreError, StorageError, is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found};
use super::{MIN_DISK_COMPRESSIBLE_SIZE, is_disk_compressible};
use super::{get_lock_acquire_timeout, is_valid_storage_class};
use super::{
lifecycle::{
bucket_lifecycle_audit::LcEventSrc,
bucket_lifecycle_ops::{enqueue_transition_immediate, post_restore_opts},
lifecycle::{self, TransitionOptions},
},
metadata_sys,
object_lock::{
objectlock::{get_object_legalhold_meta, get_object_retention_meta},
objectlock_sys::{BucketObjectLockSys, check_object_lock_for_deletion, is_retention_active},
},
quota::QuotaOperation,
replication::{
DeletedObjectReplicationInfo, ObjectOpts as ReplicationObjectOpts, check_replicate_delete, get_must_replicate_options,
must_replicate, schedule_replication, schedule_replication_delete,
},
tagging::decode_tags,
versioning_sys::BucketVersioningSys,
};
use crate::app::context::{
AppContext, get_global_app_context, resolve_notify_interface_for_context, resolve_object_store_handle_for_context,
};
use crate::config::RustFSBufferConfig;
use crate::delete_tail_activity::{DeleteTailActivityGuard, DeleteTailStage};
use crate::error::ApiError;
use crate::server::convert_ecstore_object_info;
use crate::storage::access::{PostObjectRequestMarker, authorize_request, has_bypass_governance_header, req_info_mut};
use crate::storage::concurrency::{
ConcurrencyManager, GetObjectGuard, get_concurrency_aware_buffer_size, get_concurrency_manager,
@@ -46,38 +77,6 @@ use http::{HeaderMap, HeaderValue, StatusCode};
use md5::Context as Md5Context;
use metrics::{counter, histogram};
use pin_project_lite::pin_project;
use rustfs_object_capacity::capacity_manager::get_capacity_manager;
// Performance metrics recording (with zero-copy-metrics integration)
use super::ECStore;
use super::object_api_utils::to_s3s_etag;
use super::quota::checker::QuotaChecker;
use super::storageclass;
use super::{AppReplicationConfigExt as _, AppVersioningConfigExt as _, predict_lifecycle_expiration, validate_restore_request};
use super::{DiskError, is_all_buckets_not_found};
use super::{DynReader, HashReader, WritePlan, wrap_reader};
use super::{Error as EcstoreError, StorageError, is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found};
use super::{MIN_DISK_COMPRESSIBLE_SIZE, is_disk_compressible};
use super::{get_lock_acquire_timeout, is_valid_storage_class};
use super::{
lifecycle::{
bucket_lifecycle_audit::LcEventSrc,
bucket_lifecycle_ops::{enqueue_transition_immediate, post_restore_opts},
lifecycle::{self, TransitionOptions},
},
metadata_sys,
object_lock::{
objectlock::{get_object_legalhold_meta, get_object_retention_meta},
objectlock_sys::{BucketObjectLockSys, check_object_lock_for_deletion, is_retention_active},
},
quota::QuotaOperation,
replication::{
DeletedObjectReplicationInfo, ObjectOpts as ReplicationObjectOpts, check_replicate_delete, get_must_replicate_options,
must_replicate, schedule_replication, schedule_replication_delete,
},
tagging::decode_tags,
versioning_sys::BucketVersioningSys,
};
use crate::server::convert_ecstore_object_info;
use rustfs_concurrency::GetObjectQueueSnapshot;
use rustfs_filemeta::{
REPLICATE_INCOMING_DELETE, ReplicateDecision, ReplicateTargetDecision, ReplicationState, ReplicationStatusType,
@@ -88,6 +87,7 @@ use rustfs_io_core::{BytesPool, PooledBuffer};
use rustfs_io_metrics;
use rustfs_lock::NamespaceLockGuard;
use rustfs_notify::EventArgsBuilder;
use rustfs_object_capacity::capacity_manager::get_capacity_manager;
use rustfs_policy::policy::action::{Action, S3Action};
use rustfs_s3_ops::{S3Operation, delete_event_name_for_marker, put_event_name_for_post_object};
use rustfs_s3select_api::object_store::bytes_stream;
@@ -119,7 +119,7 @@ use s3s::header::{X_AMZ_RESTORE, X_AMZ_RESTORE_OUTPUT_PATH};
use s3s::{S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error};
use std::collections::HashMap;
use std::ops::Add;
use std::path::{Component, Path};
use std::path::Path;
use std::pin::Pin;
use std::task::{Context, Poll};
@@ -337,14 +337,14 @@ impl MemoryTrackedBytesStream {
impl futures::Stream for MemoryTrackedBytesStream {
type Item = std::io::Result<Bytes>;
fn poll_next(self: std::pin::Pin<&mut Self>, _cx: &mut std::task::Context<'_>) -> std::task::Poll<Option<Self::Item>> {
fn poll_next(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let this = self.project();
if *this.emitted {
return std::task::Poll::Ready(None);
return Poll::Ready(None);
}
*this.emitted = true;
std::task::Poll::Ready(Some(Ok(this.bytes.clone())))
Poll::Ready(Some(Ok(this.bytes.clone())))
}
}
@@ -360,16 +360,12 @@ impl<R> ExtractArchiveEtagReader<R> {
}
impl<R: AsyncRead> AsyncRead for ExtractArchiveEtagReader<R> {
fn poll_read(
self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
buf: &mut ReadBuf<'_>,
) -> std::task::Poll<std::io::Result<()>> {
fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
let this = self.project();
let before = buf.filled().len();
match this.inner.poll_read(cx, buf) {
std::task::Poll::Pending => std::task::Poll::Pending,
std::task::Poll::Ready(Ok(())) => {
Poll::Pending => Poll::Pending,
Poll::Ready(Ok(())) => {
let filled = &buf.filled()[before..];
if !filled.is_empty() {
this.md5.consume(filled);
@@ -379,9 +375,9 @@ impl<R: AsyncRead> AsyncRead for ExtractArchiveEtagReader<R> {
*etag = Some(format!("{:x}", this.md5.clone().finalize()));
}
}
std::task::Poll::Ready(Ok(()))
Poll::Ready(Ok(()))
}
std::task::Poll::Ready(Err(err)) => std::task::Poll::Ready(Err(err)),
Poll::Ready(Err(err)) => Poll::Ready(Err(err)),
}
}
}
@@ -988,21 +984,12 @@ fn snowball_meta_flag(headers: &HeaderMap, exact_keys: &[&str], suffix_lower: &s
snowball_meta_value(headers, exact_keys, suffix_lower).is_some_and(|value| value.eq_ignore_ascii_case("true"))
}
fn contains_parent_dir_component(path: &str) -> bool {
path.split(['/', '\\']).any(|component| component == "..")
}
/// Validates that an archive entry path does not escape the target bucket.
///
/// Delegates to [`rustfs_utils::path::validate_extract_relative_path`] and wraps
/// the result as an S3 error on failure.
pub fn validate_extract_relative_path(path: &str) -> S3Result<()> {
let path = Path::new(path);
if path
.components()
.any(|component| matches!(component, Component::Prefix(_) | Component::RootDir | Component::ParentDir))
|| contains_parent_dir_component(path.to_string_lossy().as_ref())
{
return Err(s3_error!(InvalidArgument, "archive entry path must stay within the target bucket"));
}
Ok(())
rustfs_utils::path::validate_extract_relative_path(path).map_err(|msg| s3_error!(InvalidArgument, "{msg}"))
}
fn normalize_snowball_prefix(prefix: &str) -> S3Result<Option<String>> {
@@ -1016,22 +1003,13 @@ fn normalize_snowball_prefix(prefix: &str) -> S3Result<Option<String>> {
Ok(Some(normalized.to_string()))
}
/// Normalizes an archive entry key by applying a prefix, trimming slashes,
/// and ensuring directory entries end with `/`.
///
/// Delegates to [`rustfs_utils::path::normalize_extract_entry_key`] and wraps
/// the result as an S3 error on failure.
pub fn normalize_extract_entry_key(path: &str, prefix: Option<&str>, is_dir: bool) -> S3Result<String> {
validate_extract_relative_path(path)?;
let path = path.trim_matches('/');
let mut key = match prefix {
Some(prefix) if !path.is_empty() => format!("{prefix}/{path}"),
Some(prefix) => prefix.to_string(),
None => path.to_string(),
};
if is_dir && !key.ends_with('/') {
key.push('/');
}
validate_extract_relative_path(&key)?;
Ok(key)
rustfs_utils::path::normalize_extract_entry_key(path, prefix, is_dir).map_err(|msg| s3_error!(InvalidArgument, "{msg}"))
}
fn map_extract_archive_error(err: impl std::fmt::Display) -> S3Error {
@@ -1614,7 +1592,7 @@ impl DefaultObjectUsecase {
#[allow(clippy::too_many_arguments)]
async fn prepare_get_object_read(
req: &S3Request<GetObjectInput>,
store: &super::ECStore,
store: &ECStore,
manager: &ConcurrencyManager,
bucket: &str,
key: &str,
@@ -2895,7 +2873,6 @@ impl DefaultObjectUsecase {
validate_ssec_for_read(&info.user_defined, sse_customer_key.as_ref(), sse_customer_key_md5.as_ref())?;
let metadata_map = info.user_defined.clone();
debug!(
"GetObjectAttributes raw object_attributes={:?}",
object_attributes.iter().map(|value| value.as_str()).collect::<Vec<_>>()
@@ -2952,7 +2929,6 @@ impl DefaultObjectUsecase {
} else {
None
};
let object_parts = if requested(ObjectAttributes::OBJECT_PARTS) && info.is_multipart() {
let params = parse_list_parts_params(part_number_marker, max_parts)?;
let mut parts = Vec::new();
rustfs/src/server/http.rs
+475 -202
View File
@@ -31,11 +31,13 @@ use crate::server::{
},
};
use crate::storage;
use crate::storage::request_context::{RequestContext, extract_request_id_from_headers};
use crate::storage::rpc::InternodeRpcService;
use crate::storage::tonic_service::make_server;
use crate::storage::{TONIC_RPC_PREFIX, verify_rpc_signature};
use bytes::Bytes;
use http::{HeaderMap, Method, Request as HttpRequest, Response};
use hyper::body::Incoming;
use hyper_util::{
rt::{TokioExecutor, TokioIo, TokioTimer},
server::conn::auto::Builder as ConnBuilder,
@@ -56,12 +58,14 @@ use s3s::{host::MultiDomain, service::S3Service, service::S3ServiceBuilder};
use socket2::{SockRef, TcpKeepalive};
use std::io::{Error, Result};
use std::net::SocketAddr;
use std::pin::Pin;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::task::{Context, Poll};
use std::time::Duration;
use tokio::net::{TcpListener, TcpStream};
use tonic::{Request, Status};
use tower::ServiceBuilder;
use tower::{Service, ServiceBuilder};
use tower_http::add_extension::AddExtensionLayer;
use tower_http::catch_panic::CatchPanicLayer;
use tower_http::compression::CompressionLayer;
@@ -220,6 +224,34 @@ pub(crate) fn active_http_requests() -> u64 {
ACTIVE_HTTP_REQUESTS.load(Ordering::Relaxed)
}
fn trace_on_response<ResBody>(response: &Response<ResBody>, latency: Duration, span: &Span) {
span.record("status_code", tracing::field::display(response.status()));
let _enter = span.enter();
let status_class = status_class_label(response.status());
record_active_http_requests(-1);
histogram!(
METRIC_HTTP_SERVER_REQUEST_DURATION_SECONDS,
LABEL_HTTP_STATUS_CLASS => status_class
)
.record(latency.as_secs_f64());
if response.status().is_client_error() || response.status().is_server_error() {
counter!(
METRIC_HTTP_SERVER_FAILURES_TOTAL,
LABEL_HTTP_STATUS_CLASS => status_class
)
.increment(1);
}
if let Some(cl) = response.headers().get("content-length")
&& let Some(len) = cl.to_str().ok().and_then(|s| s.parse::<u64>().ok())
{
histogram!(
METRIC_HTTP_SERVER_RESPONSE_BODY_SIZE_BYTES,
LABEL_HTTP_STATUS_CLASS => status_class
)
.record(len as f64);
}
}
pub async fn start_http_server(config: &config::Config, readiness: Arc<GlobalReadiness>) -> Result<(ShutdownHandle, SocketAddr)> {
let server_addr = parse_and_resolve_address(config.address.as_str()).map_err(Error::other)?;
@@ -800,6 +832,90 @@ struct ConnectionContext {
trusted_proxy_layer: Option<rustfs_trusted_proxies::TrustedProxyLayer>,
}
#[derive(Clone)]
struct PathDispatchService<A, B> {
external: A,
internode: B,
}
#[derive(Clone, Default)]
struct InternodeRequestContextLiteLayer;
impl<S> tower::Layer<S> for InternodeRequestContextLiteLayer {
type Service = InternodeRequestContextLiteService<S>;
fn layer(&self, inner: S) -> Self::Service {
InternodeRequestContextLiteService { inner }
}
}
#[derive(Clone)]
struct InternodeRequestContextLiteService<S> {
inner: S,
}
impl<S, B> Service<HttpRequest<B>> for InternodeRequestContextLiteService<S>
where
S: Service<HttpRequest<B>> + Clone,
{
type Response = S::Response;
type Error = S::Error;
type Future = S::Future;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<std::result::Result<(), Self::Error>> {
self.inner.poll_ready(cx)
}
fn call(&mut self, mut req: HttpRequest<B>) -> Self::Future {
let request_id = extract_request_id_from_headers(req.headers());
req.extensions_mut().insert(RequestContext {
x_amz_request_id: request_id.clone(),
request_id,
trace_id: None,
span_id: None,
start_time: std::time::Instant::now(),
});
self.inner.call(req)
}
}
impl<A, B> PathDispatchService<A, B> {
fn new(external: A, internode: B) -> Self {
Self { external, internode }
}
fn is_internode_path(path: &str) -> bool {
crate::server::has_path_prefix(path, crate::server::RPC_PREFIX)
}
}
impl<A, B> Service<HttpRequest<Incoming>> for PathDispatchService<A, B>
where
A: Service<HttpRequest<Incoming>> + Clone + Send + 'static,
A::Future: Send + 'static,
B: Service<HttpRequest<Incoming>, Response = A::Response, Error = A::Error> + Clone + Send + 'static,
B::Future: Send + 'static,
{
type Response = A::Response;
type Error = A::Error;
type Future = Pin<Box<dyn Future<Output = std::result::Result<Self::Response, Self::Error>> + Send>>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<std::result::Result<(), Self::Error>> {
match self.external.poll_ready(cx)? {
Poll::Ready(()) => self.internode.poll_ready(cx),
Poll::Pending => Poll::Pending,
}
}
fn call(&mut self, req: HttpRequest<Incoming>) -> Self::Future {
if Self::is_internode_path(req.uri().path()) {
Box::pin(self.internode.call(req))
} else {
Box::pin(self.external.call(req))
}
}
}
/// Adapter that implements the OpenTelemetry [`Extractor`] trait for Hyper's
/// [`HeaderMap`], enabling trace context propagation by extracting
/// OpenTelemetry headers from incoming HTTP requests.
@@ -912,7 +1028,8 @@ fn process_connection(
let http_service = s3_service;
let http_service = InternodeRpcService::new(http_service);
let service = hybrid(http_service, rpc_service);
let external_service = hybrid(http_service.clone(), rpc_service.clone());
let internode_service = hybrid(http_service, rpc_service);
let remote_addr = match socket.peer_addr() {
Ok(addr) => Some(RemoteAddr(addr)),
@@ -954,217 +1071,325 @@ fn process_connection(
// 21. PublicHealthEndpointLayer — handles public health before s3s host parsing
// 22. VirtualHostStyleHintLayer — actionable error for unroutable virtual-hosted-style (conditional)
// ─────────────────────────────────────────────────────────────
let hybrid_service = ServiceBuilder::new()
// NOTE: Both extension types are intentionally inserted to maintain compatibility:
// 1. `Option<RemoteAddr>` - Used by existing admin/storage handlers throughout the codebase
// 2. `std::net::SocketAddr` - Required by TrustedProxyMiddleware for proxy validation
// This dual insertion is necessary because the middleware expects the raw SocketAddr type
// while our application code uses the RemoteAddr wrapper. Consolidating these would
// require either modifying the third-party middleware or refactoring all existing handlers.
.layer(AddExtensionLayer::new(remote_addr))
.option_layer(remote_addr.map(|ra| AddExtensionLayer::new(ra.0)))
// Add TrustedProxyLayer to handle X-Forwarded-For and other proxy headers
// This should be placed before TraceLayer so that logs reflect the real client IP
// Pre-computed in ConnectionContext to avoid per-connection is_enabled() check.
.option_layer(trusted_proxy_layer)
.layer(SetRequestIdLayer::x_request_id(MakeRequestUuid))
.layer(RequestContextLayer)
.layer(EmptyBodyContentLengthCompatLayer)
.layer(CatchPanicLayer::new())
// CRITICAL: Insert ReadinessGateLayer before business logic
// This stops requests from hitting IAMAuth or Storage if they are not ready.
.layer(ReadinessGateLayer::new(readiness))
// Add Keystone authentication middleware
// This validates X-Auth-Token headers and stores credentials in task-local storage
// Must be placed AFTER ReadinessGateLayer but BEFORE business logic
// Pre-computed in ConnectionContext to avoid per-connection OnceLock read.
.layer(KeystoneAuthLayer::new(keystone_auth))
.layer(
TraceLayer::new_for_http()
.make_span_with(|request: &HttpRequest<_>| {
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");
// Batch 1 intentionally keeps the external and internode stacks behaviorally
// identical while giving each path family a named construction boundary.
// Later batches will trim internode-only middleware without risking drift in
// the public HTTP stack.
let build_external_stack = |service| {
ServiceBuilder::new()
// NOTE: Both extension types are intentionally inserted to maintain compatibility:
// 1. `Option<RemoteAddr>` - Used by existing admin/storage handlers throughout the codebase
// 2. `std::net::SocketAddr` - Required by TrustedProxyMiddleware for proxy validation
// This dual insertion is necessary because the middleware expects the raw SocketAddr type
// while our application code uses the RemoteAddr wrapper. Consolidating these would
// require either modifying the third-party middleware or refactoring all existing handlers.
.layer(AddExtensionLayer::new(remote_addr))
.option_layer(remote_addr.map(|ra| AddExtensionLayer::new(ra.0)))
// Add TrustedProxyLayer to handle X-Forwarded-For and other proxy headers
// This should be placed before TraceLayer so that logs reflect the real client IP
// Pre-computed in ConnectionContext to avoid per-connection is_enabled() check.
.option_layer(trusted_proxy_layer.clone())
.layer(SetRequestIdLayer::x_request_id(MakeRequestUuid))
.layer(InternodeRequestContextLiteLayer)
.layer(EmptyBodyContentLengthCompatLayer)
.layer(CatchPanicLayer::new())
// CRITICAL: Insert ReadinessGateLayer before business logic
// This stops requests from hitting IAMAuth or Storage if they are not ready.
.layer(ReadinessGateLayer::new(readiness.clone()))
// Add Keystone authentication middleware
// This validates X-Auth-Token headers and stores credentials in task-local storage
// Must be placed AFTER ReadinessGateLayer but BEFORE business logic
// Pre-computed in ConnectionContext to avoid per-connection OnceLock read.
.layer(KeystoneAuthLayer::new(keystone_auth.clone()))
.layer(
TraceLayer::new_for_http()
.make_span_with(|request: &HttpRequest<_>| {
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| {
propagator.extract(&HeaderMapCarrier::new(request.headers()))
});
let parent_context = global::get_text_map_propagator(|propagator| {
propagator.extract(&HeaderMapCarrier::new(request.headers()))
});
// Log trace context extraction for debugging distributed tracing
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");
}
// Extract real client IP from trusted proxy middleware if available
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);
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());
span
})
.on_request(|request: &HttpRequest<_>, span: &Span| {
let _enter = span.enter();
trace!("HTTP request started");
let method = request_method_label(request.method());
record_active_http_requests(1);
counter!(
METRIC_HTTP_SERVER_REQUESTS_TOTAL,
LABEL_HTTP_METHOD => method
)
.increment(1);
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);
}
}
if let Some(cl) = request.headers().get("content-length")
&& let Some(len) = cl.to_str().ok().and_then(|s| s.parse::<u64>().ok())
{
counter!(METRIC_HTTP_SERVER_REQUEST_BODY_BYTES_TOTAL).increment(len);
histogram!(
METRIC_HTTP_SERVER_REQUEST_BODY_SIZE_BYTES,
span
})
.on_request(|request: &HttpRequest<_>, span: &Span| {
let _enter = span.enter();
trace!("HTTP request started");
let method = request_method_label(request.method());
record_active_http_requests(1);
counter!(
METRIC_HTTP_SERVER_REQUESTS_TOTAL,
LABEL_HTTP_METHOD => method
)
.record(len as f64);
}
})
.on_response(|response: &Response<_>, latency: Duration, span: &Span| {
span.record("status_code", tracing::field::display(response.status()));
let _enter = span.enter();
let status_class = status_class_label(response.status());
record_active_http_requests(-1);
histogram!(
METRIC_HTTP_SERVER_REQUEST_DURATION_SECONDS,
LABEL_HTTP_STATUS_CLASS => status_class
)
.record(latency.as_secs_f64());
if response.status().is_client_error() || response.status().is_server_error() {
.increment(1);
if let Some(cl) = request.headers().get("content-length")
&& let Some(len) = cl.to_str().ok().and_then(|s| s.parse::<u64>().ok())
{
counter!(METRIC_HTTP_SERVER_REQUEST_BODY_BYTES_TOTAL).increment(len);
histogram!(
METRIC_HTTP_SERVER_REQUEST_BODY_SIZE_BYTES,
LABEL_HTTP_METHOD => method
)
.record(len as f64);
}
})
.on_response(trace_on_response)
.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!(chunk_bytes = chunk.len(), duration_ms = duration_ms(latency), "HTTP response body chunk sent");
}
#[cfg(not(feature = "tracing-chunk-debug"))]
{
let _ = (latency, span);
}
})
.on_eos(|_trailers: Option<&HeaderMap>, stream_duration: Duration, span: &Span| {
#[cfg(feature = "tracing-chunk-debug")]
{
let _enter = span.enter();
debug!(duration_ms = duration_ms(stream_duration), "HTTP response stream closed");
}
#[cfg(not(feature = "tracing-chunk-debug"))]
{
let _ = (_trailers, stream_duration, span);
}
})
.on_failure(|error, latency: Duration, span: &Span| {
let _enter = span.enter();
record_active_http_requests(-1);
counter!(
METRIC_HTTP_SERVER_FAILURES_TOTAL,
LABEL_HTTP_STATUS_CLASS => status_class
LABEL_HTTP_STATUS_CLASS => "transport"
)
.increment(1);
}
if let Some(cl) = response.headers().get("content-length")
&& let Some(len) = cl.to_str().ok().and_then(|s| s.parse::<u64>().ok())
{
histogram!(
METRIC_HTTP_SERVER_RESPONSE_BODY_SIZE_BYTES,
LABEL_HTTP_STATUS_CLASS => status_class
trace!(error = ?error, duration_ms = duration_ms(latency), "HTTP request failure captured by trace layer");
}),
)
.layer(RequestLoggingLayer)
.layer(PropagateRequestIdLayer::x_request_id())
.layer(CompressionLayer::new().compress_when(PathAwareHttpCompressionPredicate::new(compression_config.clone())))
.layer(PathCategoryInjectionLayer)
.layer(S3ErrorMessageCompatLayer)
.layer(ObjectAttributesEtagFixLayer)
.layer(ConditionalCorsLayer::new())
.option_layer(if is_console { Some(RedirectLayer) } else { None })
.layer(BodylessStatusFixLayer)
.layer(HeadRequestBodyFixLayer)
.layer(PublicHealthEndpointLayer)
.option_layer((!server_domains_configured && !is_console).then_some(VirtualHostStyleHintLayer))
.service(service)
};
let build_internode_stack = |service| {
ServiceBuilder::new()
.layer(AddExtensionLayer::new(remote_addr))
.option_layer(remote_addr.map(|ra| AddExtensionLayer::new(ra.0)))
.option_layer(trusted_proxy_layer.clone())
.layer(SetRequestIdLayer::x_request_id(MakeRequestUuid))
.layer(RequestContextLayer)
.layer(EmptyBodyContentLengthCompatLayer)
.layer(CatchPanicLayer::new())
.layer(ReadinessGateLayer::new(readiness.clone()))
.layer(KeystoneAuthLayer::new(keystone_auth.clone()))
.layer(
TraceLayer::new_for_http()
.make_span_with(|request: &HttpRequest<_>| {
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| {
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
})
.on_request(|request: &HttpRequest<_>, span: &Span| {
let _enter = span.enter();
trace!("HTTP request started");
let method = request_method_label(request.method());
record_active_http_requests(1);
counter!(
METRIC_HTTP_SERVER_REQUESTS_TOTAL,
LABEL_HTTP_METHOD => method
)
.record(len as f64);
}
})
.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")]
{
.increment(1);
if let Some(cl) = request.headers().get("content-length")
&& let Some(len) = cl.to_str().ok().and_then(|s| s.parse::<u64>().ok())
{
counter!(METRIC_HTTP_SERVER_REQUEST_BODY_BYTES_TOTAL).increment(len);
histogram!(
METRIC_HTTP_SERVER_REQUEST_BODY_SIZE_BYTES,
LABEL_HTTP_METHOD => method
)
.record(len as f64);
}
})
.on_response(trace_on_response)
.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!(chunk_bytes = chunk.len(), duration_ms = duration_ms(latency), "HTTP response body chunk sent");
}
#[cfg(not(feature = "tracing-chunk-debug"))]
{
let _ = (latency, span);
}
})
.on_eos(|_trailers: Option<&HeaderMap>, stream_duration: Duration, span: &Span| {
#[cfg(feature = "tracing-chunk-debug")]
{
let _enter = span.enter();
debug!(duration_ms = duration_ms(stream_duration), "HTTP response stream closed");
}
#[cfg(not(feature = "tracing-chunk-debug"))]
{
let _ = (_trailers, stream_duration, span);
}
})
.on_failure(|error, latency: Duration, span: &Span| {
let _enter = span.enter();
debug!(chunk_bytes = chunk.len(), duration_ms = duration_ms(latency), "HTTP response body chunk sent");
}
#[cfg(not(feature = "tracing-chunk-debug"))]
{
let _ = (latency, span);
}
})
.on_eos(|_trailers: Option<&HeaderMap>, stream_duration: Duration, span: &Span| {
#[cfg(feature = "tracing-chunk-debug")]
{
let _enter = span.enter();
debug!(duration_ms = duration_ms(stream_duration), "HTTP response stream closed");
}
#[cfg(not(feature = "tracing-chunk-debug"))]
{
let _ = (_trailers, stream_duration, span);
}
})
.on_failure(|_error, latency: Duration, span: &Span| {
let _enter = span.enter();
record_active_http_requests(-1);
counter!(
METRIC_HTTP_SERVER_FAILURES_TOTAL,
LABEL_HTTP_STATUS_CLASS => "transport"
)
.increment(1);
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
.layer(CompressionLayer::new().compress_when(PathAwareHttpCompressionPredicate::new(compression_config)))
.layer(PathCategoryInjectionLayer)
.layer(S3ErrorMessageCompatLayer)
.layer(ObjectAttributesEtagFixLayer)
// Conditional CORS layer: only applies to S3 API requests (not Admin, not Console)
// Admin has its own CORS handling in router.rs
// Console has its own CORS layer in setup_console_middleware_stack()
// S3 API uses this system default CORS (RUSTFS_CORS_ALLOWED_ORIGINS)
// Bucket-level CORS takes precedence when configured (handled in router.rs for OPTIONS, and in ecfs.rs for actual requests)
.layer(ConditionalCorsLayer::new())
.option_layer(if is_console { Some(RedirectLayer) } else { None })
// Must run before outer response-transforming layers: clear the body and remove
// Content-Length, Content-Type, and Transfer-Encoding for statuses
// that MUST NOT carry a body (1xx/204/304). Placed inside those
// layers so they see the already-bodyless
// response and so no layer (e.g. CORS) re-adds body headers afterward.
.layer(BodylessStatusFixLayer)
// HEAD responses must not send body bytes even when the inner S3 layer
// serializes an XML error payload.
.layer(HeadRequestBodyFixLayer)
// Health probes are public admin routes, but s3s parses virtual-host
// buckets before custom routes. Handle them here so SERVER_DOMAINS
// cannot turn /health into an S3 bucket request.
.layer(PublicHealthEndpointLayer)
// Virtual-hosted-style S3 requests (the AWS SDK / Terraform default) cannot be
// routed when no server domain is configured: s3s parses them path-style and
// returns an opaque 501. When RUSTFS_SERVER_DOMAINS is unset, return an actionable
// error pointing at the fix. Inert (not installed) once domains are configured.
.option_layer((!server_domains_configured && !is_console).then_some(VirtualHostStyleHintLayer))
.service(service);
record_active_http_requests(-1);
counter!(
METRIC_HTTP_SERVER_FAILURES_TOTAL,
LABEL_HTTP_STATUS_CLASS => "transport"
)
.increment(1);
trace!(error = ?error, duration_ms = duration_ms(latency), "HTTP request failure captured by trace layer");
}),
)
.layer(PropagateRequestIdLayer::x_request_id())
.layer(CompressionLayer::new().compress_when(PathAwareHttpCompressionPredicate::new(compression_config.clone())))
.layer(PathCategoryInjectionLayer)
.layer(S3ErrorMessageCompatLayer)
.layer(ObjectAttributesEtagFixLayer)
.layer(ConditionalCorsLayer::new())
.option_layer(if is_console { Some(RedirectLayer) } else { None })
.layer(BodylessStatusFixLayer)
.layer(HeadRequestBodyFixLayer)
.layer(PublicHealthEndpointLayer)
.option_layer((!server_domains_configured && !is_console).then_some(VirtualHostStyleHintLayer))
.service(service)
};
let external_stack_service = build_external_stack(external_service);
let internode_stack_service = build_internode_stack(internode_service);
let hybrid_service = PathDispatchService::new(external_stack_service, internode_stack_service);
let hybrid_service = TowerToHyperService::new(hybrid_service);
@@ -1397,12 +1622,13 @@ mod tests {
use super::*;
use crate::server::compress::RequestPathCategory;
use bytes::Bytes;
use http::HeaderMap;
use http::Request as HttpRequest;
use http_body_util::Empty;
use http::{HeaderMap, StatusCode};
use http_body_util::{Empty, Full};
use opentelemetry::propagation::Extractor;
use std::convert::Infallible;
use std::future::Ready;
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll};
use tower::{Layer, Service, ServiceBuilder};
@@ -1620,6 +1846,36 @@ mod tests {
}
}
#[derive(Clone)]
struct MarkerService {
name: &'static str,
hits: Arc<Mutex<Vec<&'static str>>>,
}
impl MarkerService {
fn new(name: &'static str, hits: Arc<Mutex<Vec<&'static str>>>) -> Self {
Self { name, hits }
}
}
impl<ReqBody> Service<HttpRequest<ReqBody>> for MarkerService {
type Response = Response<Full<Bytes>>;
type Error = Infallible;
type Future = Ready<std::result::Result<Response<Full<Bytes>>, Infallible>>;
fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<std::result::Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
fn call(&mut self, _req: HttpRequest<ReqBody>) -> Self::Future {
self.hits.lock().expect("hits").push(self.name);
std::future::ready(Ok(Response::builder()
.status(StatusCode::OK)
.body(Full::from(Bytes::from_static(self.name.as_bytes())))
.expect("response")))
}
}
#[test]
fn test_service_builder_order_regression_for_response_extensions() {
let request = HttpRequest::builder().uri("/bucket/archive.zip").body(()).expect("request");
@@ -1648,4 +1904,21 @@ mod tests {
Some("true")
);
}
#[test]
fn path_dispatch_service_identifies_rpc_prefix() {
let hits = Arc::new(Mutex::new(Vec::new()));
let service =
PathDispatchService::new(MarkerService::new("external", Arc::clone(&hits)), MarkerService::new("internode", hits));
assert!(PathDispatchService::<MarkerService, MarkerService>::is_internode_path(&format!(
"{}/put_file_stream",
crate::server::RPC_PREFIX
)));
assert!(!PathDispatchService::<MarkerService, MarkerService>::is_internode_path(
"/bucket/object.txt"
));
assert!(!PathDispatchService::<MarkerService, MarkerService>::is_internode_path("/rustfs/rpcx"));
let _ = service;
}
}
rustfs/src/storage/rpc/http_service.rs
+10
View File
@@ -38,6 +38,7 @@ use serde_urlencoded::from_bytes;
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Instant;
use tokio::io::{self, AsyncWriteExt};
use tokio_util::io::ReaderStream;
use tower::Service;
@@ -280,6 +281,7 @@ fn is_internode_rpc_path(path: &str) -> bool {
async fn handle_internode_rpc(req: Request<Incoming>) -> Response<Body> {
let operation = internode_http_operation(req.uri().path());
let started_at = Instant::now();
if let Err(response) = verify_internode_rpc_signature(req.uri(), req.method(), req.headers()) {
record_internode_rpc_error(operation);
return *response;
@@ -299,6 +301,14 @@ async fn handle_internode_rpc(req: Request<Incoming>) -> Response<Body> {
record_internode_rpc_error(operation);
}
if let Some(operation) = operation {
resolve_internode_metrics().record_duration_for_operation_and_backend(
operation,
INTERNODE_TRANSPORT_BACKEND_TCP_HTTP,
started_at.elapsed(),
);
}
response
}