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

refactor(request-id): align contracts and lock field names (#3454)

Browse Source 
* refactor(request-id): align header log and trace contracts

* test(contract): lock external request-id field names

* chore(deps): drop unused rustfs-ecstore links

Co-Authored-By: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
2026-06-15 01:59:11 +08:00
committed by GitHub
parent efaf07d323
commit 036741cb1c
12 changed files with 226 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/server/layer.rs
+33 -45
View File
@@ -24,14 +24,12 @@ use crate::server::{
has_path_prefix, is_admin_path, is_table_catalog_path,
};
use crate::storage::apply_cors_headers;
use crate::storage::request_context::{RequestContext, extract_request_id_from_headers};
use crate::storage::request_context::{RequestContext, extract_request_id_from_headers, extract_trace_context_ids_from_headers};
use bytes::Bytes;
use http::{HeaderMap, HeaderValue, Method, Request as HttpRequest, Response, StatusCode};
use http_body::Body;
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;
@@ -52,30 +50,6 @@ const LOG_SUBSYSTEM_HTTP: &str = "http";
const REDACTED_QUERY_VALUE: &str = "redacted";
const OBJECT_ZIP_DOWNLOADS_PATH: &str = "/v3/object-zip-downloads/";
/// A carrier that adapts [`HeaderMap`] for OpenTelemetry trace context propagation.
struct HeaderMapCarrier<'a>(&'a HeaderMap);
impl<'a> opentelemetry::propagation::Extractor for HeaderMapCarrier<'a> {
fn get(&self, key: &str) -> Option<&str> {
self.0.get(key).and_then(|v| v.to_str().ok())
}
fn keys(&self) -> Vec<&str> {
self.0.keys().map(|k| k.as_str()).collect()
}
fn get_all(&self, key: &str) -> Option<Vec<&str>> {
let headers = self
.0
.get_all(key)
.iter()
.filter_map(|value| value.to_str().ok())
.collect::<Vec<_>>();
if headers.is_empty() { None } else { Some(headers) }
}
}
pub(crate) fn redact_sensitive_uri_query(uri: &http::Uri) -> String {
let path = uri.path();
if !is_object_zip_download_path(path) {
@@ -131,8 +105,8 @@ fn is_object_zip_download_path(path: &str) -> bool {
/// This layer must be placed after `SetRequestIdLayer` in the middleware stack,
/// as it reads the `x-request-id` header that `SetRequestIdLayer` generates.
///
/// Additionally, it stores the S3-compatible request ID alias in the request context
/// without mutating signed request headers.
/// Additionally, it preserves any upstream `x-amz-request-id` in the separate
/// `RequestContext.x_amz_request_id` field without mutating signed request headers.
#[derive(Clone, Default)]
pub struct RequestContextLayer;
@@ -165,23 +139,12 @@ where
fn call(&mut self, mut req: HttpRequest<B>) -> Self::Future {
let request_id = extract_request_id_from_headers(req.headers());
// Extract OpenTelemetry trace/span context from incoming headers
let parent_cx = global::get_text_map_propagator(|propagator| propagator.extract(&HeaderMapCarrier(req.headers())));
let span_ref = parent_cx.span();
let span_context = span_ref.span_context();
let trace_id = if span_context.is_valid() {
Some(span_context.trace_id().to_string())
} else {
None
};
let span_id = if span_context.is_valid() {
Some(span_context.span_id().to_string())
} else {
None
};
let (trace_id, span_id) = extract_trace_context_ids_from_headers(req.headers())
.map(|(trace_id, span_id)| (Some(trace_id), Some(span_id)))
.unwrap_or((None, None));
// Preserve the upstream x-amz-request-id if present (S3 client forwarding),
// otherwise fall back to the canonical request_id.
// Preserve the upstream x-amz-request-id if present as the S3 compatibility alias;
// otherwise mirror the canonical internal request_id.
let x_amz_request_id = req
.headers()
.get(AMZ_REQUEST_ID)
@@ -1371,6 +1334,8 @@ mod tests {
use http::Request;
use http_body_util::BodyExt;
use http_body_util::Full;
use opentelemetry::global;
use opentelemetry_sdk::propagation::TraceContextPropagator;
use serial_test::serial;
use std::convert::Infallible;
use std::io::{self, Write};
@@ -2320,6 +2285,29 @@ mod tests {
assert_eq!(request.headers().get(AMZ_REQUEST_ID).unwrap(), "amz-456");
}
#[test]
fn request_context_layer_extracts_trace_context_from_traceparent_header() {
global::set_text_map_propagator(TraceContextPropagator::new());
let mut service = RequestContextLayer.layer(CaptureService);
let request = Request::builder()
.uri("/bucket/object")
.header("x-request-id", "req-trace-123")
.header("traceparent", "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01")
.body(())
.expect("request");
let request = service.call(request).into_inner().expect("service call should succeed");
let context = request
.extensions()
.get::<RequestContext>()
.expect("request context should be present");
assert_eq!(context.request_id, "req-trace-123");
assert_eq!(context.trace_id.as_deref(), Some("4bf92f3577b34da6a3ce929d0e0e4736"));
assert_eq!(context.span_id.as_deref(), Some("00f067aa0ba902b7"));
}
#[tokio::test]
async fn test_resolve_s3_options_cors_headers_no_headers_without_match() {
let mut req_headers = HeaderMap::new();
rustfs/src/storage/request_context.rs
+53 -9
View File
@@ -39,10 +39,12 @@
//!
//! # Frozen Rules (T00 Guardrails)
//!
//! ## request-id
//! - Canonical source: HTTP ingress `x-request-id` header (set by `SetRequestIdLayer`)
//! - `x-amz-request_id` is an alias for S3 compatibility, always equal to `request_id`
//! - Internal modules MUST NOT generate a second request-id under the name `request_id`
//! ## request-id contract
//! - Canonical wire header: `x-request-id` (set by `SetRequestIdLayer`)
//! - Compatibility wire header: `x-amz-request-id`
//! - Canonical internal field: `RequestContext.request_id`
//! - S3 compatibility internal alias field: `RequestContext.x_amz_request_id`
//! - Internal modules MUST NOT generate a second request id under the field name `request_id`
//! except for orphan/non-ingress fallback paths where no canonical request-id exists.
//! - Internal identifiers for sub-operations should use `operation_id` or `subtask_id`
//!
@@ -58,14 +60,13 @@
use http::HeaderMap;
use metrics::counter;
use opentelemetry::global;
use opentelemetry::trace::TraceContextExt;
use rustfs_utils::http::headers::AMZ_REQUEST_ID;
use rustfs_utils::http::headers::{AMZ_REQUEST_ID, REQUEST_ID_HEADER};
use std::time::Instant;
use tracing::Span;
use tracing_opentelemetry::OpenTelemetrySpanExt;
const REQUEST_ID_HEADER: &str = "x-request-id";
/// Canonical request context carried through the entire request lifecycle.
///
/// Created exactly once at HTTP ingress. Cloned by value; never mutated after creation.
@@ -86,7 +87,7 @@ pub struct RequestContext {
impl RequestContext {
/// Create a fallback `RequestContext` for paths that bypass HTTP ingress.
/// Generates a `trace-{trace_id}` or `req-{uuid}` format request-id.
/// Generates a canonical internal `request_id` in `trace-{trace_id}` or `req-{uuid}` format.
pub fn fallback() -> Self {
let trace_ctx = current_trace_context_ids();
let id = build_fallback_request_id(trace_ctx.as_ref());
@@ -117,6 +118,18 @@ fn current_trace_context_ids() -> Option<(String, String)> {
Some((span_context.trace_id().to_string(), span_context.span_id().to_string()))
}
struct HeaderMapExtractor<'a>(&'a HeaderMap);
impl opentelemetry::propagation::Extractor for HeaderMapExtractor<'_> {
fn get(&self, key: &str) -> Option<&str> {
self.0.get(key).and_then(|v| v.to_str().ok())
}
fn keys(&self) -> Vec<&str> {
self.0.keys().map(|k| k.as_str()).collect()
}
}
fn build_fallback_request_id(trace_ctx: Option<&(String, String)>) -> String {
trace_ctx
.map(|(trace_id, _)| format!("trace-{trace_id}"))
@@ -128,7 +141,20 @@ fn generate_fallback_request_id() -> String {
build_fallback_request_id(trace_ctx.as_ref())
}
/// Extract the canonical request ID from HTTP headers.
/// Extract remote trace/span IDs from HTTP headers using the configured
/// OpenTelemetry text map propagator (for example W3C `traceparent`).
pub fn extract_trace_context_ids_from_headers(headers: &HeaderMap) -> Option<(String, String)> {
let parent_context = global::get_text_map_propagator(|propagator| propagator.extract(&HeaderMapExtractor(headers)));
let span_ref = parent_context.span();
let span_context = span_ref.span_context();
if !span_context.is_valid() {
return None;
}
Some((span_context.trace_id().to_string(), span_context.span_id().to_string()))
}
/// Extract the canonical internal `request_id` from HTTP request headers.
///
/// Priority:
/// 1. `x-request-id` (primary, set by `SetRequestIdLayer`)
@@ -169,6 +195,7 @@ where
mod tests {
use super::*;
use opentelemetry::trace::{SpanContext, TraceContextExt, TraceFlags, TraceId, TraceState, TracerProvider as _};
use opentelemetry_sdk::propagation::TraceContextPropagator;
use opentelemetry_sdk::trace::SdkTracerProvider;
use tracing_opentelemetry::OpenTelemetrySpanExt;
use tracing_subscriber::{Registry, layer::SubscriberExt};
@@ -253,6 +280,23 @@ mod tests {
assert_eq!(id, "x-req-789");
}
#[test]
fn test_extract_trace_context_ids_from_traceparent_header() {
global::set_text_map_propagator(TraceContextPropagator::new());
let mut headers = HeaderMap::new();
headers.insert(
"traceparent",
"00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"
.parse()
.expect("traceparent header"),
);
let trace_ctx = extract_trace_context_ids_from_headers(&headers).expect("trace context should be extracted");
assert_eq!(trace_ctx.0, "4bf92f3577b34da6a3ce929d0e0e4736");
assert_eq!(trace_ctx.1, "00f067aa0ba902b7");
}
#[test]
fn test_extract_request_id_no_headers() {
let headers = HeaderMap::new();