diff --git a/Cargo.lock b/Cargo.lock index c34e2e718..d72e32923 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9175,7 +9175,6 @@ dependencies = [ "hashbrown 0.17.1", "metrics", "rustfs-config", - "rustfs-ecstore", "rustfs-s3-types", "rustfs-targets", "serde", @@ -10057,7 +10056,6 @@ dependencies = [ "reqwest", "rumqttc-next", "rustfs-config", - "rustfs-ecstore", "rustfs-extension-schema", "rustfs-kafka-async", "rustfs-s3-types", diff --git a/crates/audit/Cargo.toml b/crates/audit/Cargo.toml index 8b27e25d7..a1135f3ee 100644 --- a/crates/audit/Cargo.toml +++ b/crates/audit/Cargo.toml @@ -28,7 +28,6 @@ categories = ["web-programming", "development-tools", "asynchronous", "api-bindi [dependencies] rustfs-targets = { workspace = true } rustfs-config = { workspace = true, features = ["audit", "constants", "server-config-model"] } -rustfs-ecstore = { workspace = true } rustfs-s3-types = { workspace = true } chrono = { workspace = true } const-str = { workspace = true } diff --git a/crates/audit/src/entity.rs b/crates/audit/src/entity.rs index 39931ecfb..c2a1f3923 100644 --- a/crates/audit/src/entity.rs +++ b/crates/audit/src/entity.rs @@ -160,6 +160,8 @@ pub struct AuditEntry { pub api: ApiDetails, #[serde(rename = "remotehost", skip_serializing_if = "Option::is_none")] pub remote_host: Option, + // Historical external audit contract: keep `requestID` instead of normalizing + // this field to `request_id` or `request-id`. #[serde(rename = "requestID", skip_serializing_if = "Option::is_none")] pub request_id: Option, #[serde(rename = "userAgent", skip_serializing_if = "Option::is_none")] @@ -315,3 +317,29 @@ impl AuditEntryBuilder { self.0 } } + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::Value; + + #[test] + fn audit_entry_serializes_historical_request_id_field_name() { + let entry = AuditEntryBuilder::new( + "1", + EventName::ObjectCreatedPut, + "s3", + ApiDetailsBuilder::new() + .name("PutObject") + .status("OK") + .status_code(200) + .build(), + ) + .request_id("req-audit-123") + .build(); + + let value = serde_json::to_value(entry).expect("audit entry should serialize"); + assert_eq!(value["requestID"], Value::String("req-audit-123".to_string())); + assert!(value.get("request_id").is_none(), "historical audit contract must not expose request_id"); + } +} diff --git a/crates/ecstore/src/bucket/quota/mod.rs b/crates/ecstore/src/bucket/quota/mod.rs index 1593e3d44..99a355348 100644 --- a/crates/ecstore/src/bucket/quota/mod.rs +++ b/crates/ecstore/src/bucket/quota/mod.rs @@ -124,6 +124,7 @@ pub struct QuotaErrorResponse { pub message: String, #[serde(rename = "Resource")] pub resource: String, + // External quota error contract follows the existing PascalCase error schema. #[serde(rename = "RequestId")] pub request_id: String, #[serde(rename = "HostId")] @@ -168,6 +169,7 @@ impl QuotaErrorResponse { #[cfg(test)] mod tests { use super::*; + use serde_json::Value; /// Legacy format: quota, created_at, updated_at (no quota_type) #[test] @@ -217,4 +219,19 @@ mod tests { assert_eq!(q.quota, Some(1073741824)); assert_eq!(q.quota_type, QuotaType::Hard); } + + #[test] + fn quota_error_response_serializes_request_id_as_pascal_case_contract() { + let response = QuotaErrorResponse::new( + &QuotaError::InvalidConfig { + reason: "bad quota".to_string(), + }, + "req-quota-123", + "host-quota-1", + ); + + let value = serde_json::to_value(response).expect("quota error response should serialize"); + assert_eq!(value["RequestId"], Value::String("req-quota-123".to_string())); + assert!(value.get("request_id").is_none(), "quota error contract must not expose request_id"); + } } diff --git a/crates/ecstore/src/client/api_error_response.rs b/crates/ecstore/src/client/api_error_response.rs index 9e2cce480..8cd576e90 100644 --- a/crates/ecstore/src/client/api_error_response.rs +++ b/crates/ecstore/src/client/api_error_response.rs @@ -36,6 +36,8 @@ pub struct ErrorResponse { pub bucket_name: String, pub key: String, pub resource: String, + // External S3-style error response contract: keep `RequestId`. + #[serde(rename = "RequestId")] pub request_id: String, pub host_id: String, pub region: String, @@ -302,3 +304,29 @@ pub fn err_api_not_supported(message: &str) -> ErrorResponse { ..Default::default() } } + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::Value; + + #[test] + fn error_response_serializes_request_id_as_pascal_case_contract() { + let response = ErrorResponse { + code: S3ErrorCode::InvalidArgument, + message: "bad request".to_string(), + bucket_name: "bucket".to_string(), + key: "key".to_string(), + resource: "/bucket/key".to_string(), + request_id: "req-xml-123".to_string(), + host_id: "host-1".to_string(), + region: "us-east-1".to_string(), + server: "rustfs".to_string(), + status_code: StatusCode::BAD_REQUEST, + }; + + let value = serde_json::to_value(response).expect("error response should serialize"); + assert_eq!(value["RequestId"], Value::String("req-xml-123".to_string())); + assert!(value.get("request_id").is_none(), "external error contract must not expose request_id"); + } +} diff --git a/crates/ecstore/src/rpc/client.rs b/crates/ecstore/src/rpc/client.rs index 2138345b6..be121b525 100644 --- a/crates/ecstore/src/rpc/client.rs +++ b/crates/ecstore/src/rpc/client.rs @@ -134,12 +134,42 @@ impl tonic::service::Interceptor for TonicInterceptor { #[cfg(test)] mod tests { use super::*; + use opentelemetry::global; + use opentelemetry::trace::{SpanContext, TraceContextExt, TraceFlags, TraceId, TraceState, TracerProvider as _}; + use opentelemetry_sdk::propagation::TraceContextPropagator; + use opentelemetry_sdk::trace::SdkTracerProvider; use tonic::service::Interceptor; + use tracing_opentelemetry::OpenTelemetrySpanExt; + use tracing_subscriber::{Registry, layer::SubscriberExt}; fn ensure_test_rpc_secret() { let _ = rustfs_credentials::GLOBAL_RUSTFS_RPC_SECRET.set("test-rpc-secret".to_string()); } + fn with_trace_parent(trace_id_hex: &str, f: F) + where + F: FnOnce(), + { + global::set_text_map_propagator(TraceContextPropagator::new()); + + let provider = SdkTracerProvider::builder().build(); + let tracer = provider.tracer("rpc-client-tests"); + let subscriber = Registry::default().with(tracing_opentelemetry::layer().with_tracer(tracer)); + + tracing::subscriber::with_default(subscriber, || { + let span = tracing::info_span!("rpc-client-test-span"); + let trace_id = TraceId::from_hex(trace_id_hex).expect("trace id should be valid hex"); + let span_id = opentelemetry::trace::SpanId::from_hex("0102030405060708").expect("span id should be valid hex"); + let parent = SpanContext::new(trace_id, span_id, TraceFlags::SAMPLED, true, TraceState::default()); + span.set_parent(opentelemetry::Context::new().with_remote_span_context(parent)) + .expect("failed to set parent context"); + let _guard = span.enter(); + + f(); + }); + let _ = provider.shutdown(); + } + #[test] fn test_signature_interceptor_keeps_auth_headers() { ensure_test_rpc_secret(); @@ -166,4 +196,21 @@ mod tests { assert!(!v.as_encoded_bytes().is_empty()); } } + + #[test] + fn test_signature_interceptor_injects_traceparent_metadata() { + ensure_test_rpc_secret(); + let mut interceptor = TonicSignatureInterceptor; + let req = tonic::Request::new(()); + + with_trace_parent("4bf92f3577b34da6a3ce929d0e0e4736", || { + let req = interceptor.call(req).expect("interceptor call should succeed"); + let traceparent = req + .metadata() + .get("traceparent") + .and_then(|v| v.to_str().ok()) + .expect("traceparent metadata should be injected"); + assert!(traceparent.starts_with("00-4bf92f3577b34da6a3ce929d0e0e4736-")); + }); + } } diff --git a/crates/ecstore/src/rpc/context_propagation.rs b/crates/ecstore/src/rpc/context_propagation.rs index 3ee1e806b..5005ea2df 100644 --- a/crates/ecstore/src/rpc/context_propagation.rs +++ b/crates/ecstore/src/rpc/context_propagation.rs @@ -14,11 +14,10 @@ use http::{HeaderMap, HeaderValue}; use opentelemetry::{global, propagation::Injector, trace::TraceContextExt}; +pub(crate) use rustfs_utils::http::headers::REQUEST_ID_HEADER; use tracing::Span; use tracing_opentelemetry::OpenTelemetrySpanExt; -pub(crate) const REQUEST_ID_HEADER: &str = "x-request-id"; - struct HttpHeaderInjector<'a> { headers: &'a mut HeaderMap, } diff --git a/crates/obs/src/telemetry/local.rs b/crates/obs/src/telemetry/local.rs index fe431008b..e1b7ea5c3 100644 --- a/crates/obs/src/telemetry/local.rs +++ b/crates/obs/src/telemetry/local.rs @@ -71,7 +71,8 @@ const LOG_SUBSYSTEM_LOCAL_LOGGING: &str = "local_logging"; const EVENT_LOCAL_LOGGING_STATE: &str = "local_logging_state"; const EVENT_LOG_CLEANER_STATE: &str = "log_cleaner_state"; const STDERR_WARNING_PREFIX: &str = "[WARN]"; -const REQUEST_ID: &str = "request-id"; +const REQUEST_ID_CANONICAL: &str = "request_id"; +const REQUEST_ID_COMPAT: &str = "request-id"; #[derive(Clone, Debug)] struct RequestIdJsonFormat { @@ -106,9 +107,15 @@ where let trimmed = buffer.trim_end(); let mut payload: JsonValue = serde_json::from_str(trimmed).map_err(|_| fmt::Error)?; if let Some(object) = payload.as_object_mut() { + let request_id = request_id.expect("checked is_some"); object - .entry(REQUEST_ID.to_string()) - .or_insert_with(|| JsonValue::String(request_id.expect("checked is_some"))); + .entry(REQUEST_ID_CANONICAL.to_string()) + .or_insert_with(|| JsonValue::String(request_id.clone())); + // Keep a top-level compatibility alias for operators or downstream + // queries that already rely on the earlier hyphenated field name. + object + .entry(REQUEST_ID_COMPAT.to_string()) + .or_insert_with(|| JsonValue::String(request_id)); } let serialized = serde_json::to_string(&payload).map_err(|_| fmt::Error)?; @@ -131,7 +138,11 @@ where let extensions = span.extensions(); let formatted_fields = extensions.get::>()?; let fields: BTreeMap = serde_json::from_str(&formatted_fields.fields).ok()?; - if let Some(value) = fields.get(REQUEST_ID).and_then(JsonValue::as_str) { + if let Some(value) = fields + .get(REQUEST_ID_CANONICAL) + .or_else(|| fields.get(REQUEST_ID_COMPAT)) + .and_then(JsonValue::as_str) + { request_id = Some(value.to_string()); } } @@ -842,6 +853,7 @@ mod tests { let parsed = render_json_log_with_request_span(); assert_eq!(parsed["request_id"], Value::String("req-123".to_string())); + assert_eq!(parsed["request-id"], Value::String("req-123".to_string())); assert_eq!(parsed["message"], Value::String("inside request span".to_string())); assert_eq!(parsed["span"]["request_id"], Value::String("req-123".to_string())); } @@ -851,6 +863,7 @@ mod tests { let parsed = render_json_log_with_recovery_monitor_child_span(); assert_eq!(parsed["request_id"], Value::String("req-parent".to_string())); + assert_eq!(parsed["request-id"], Value::String("req-parent".to_string())); assert_eq!(parsed["message"], Value::String("inside recovery monitor".to_string())); assert_eq!(parsed["span"]["name"], Value::String("recovery-monitor".to_string())); assert_eq!(parsed["span"]["kind"], Value::String("remote_disk".to_string())); diff --git a/crates/targets/Cargo.toml b/crates/targets/Cargo.toml index 191100667..7cad1657c 100644 --- a/crates/targets/Cargo.toml +++ b/crates/targets/Cargo.toml @@ -13,7 +13,6 @@ documentation = "https://docs.rs/rustfs-targets/latest/rustfs_targets/" [dependencies] rustfs-config = { workspace = true, features = ["notify", "constants", "audit", "server-config-model"] } -rustfs-ecstore = { workspace = true } rustfs-extension-schema = { workspace = true } rustfs-tls-runtime = { workspace = true } rustfs-s3-types = { workspace = true } diff --git a/crates/utils/src/http/headers.rs b/crates/utils/src/http/headers.rs index db9fead1e..a881692ac 100644 --- a/crates/utils/src/http/headers.rs +++ b/crates/utils/src/http/headers.rs @@ -151,7 +151,8 @@ pub const AMZ_ENCRYPTION_KMS: &str = "aws:kms"; pub const AMZ_SIGNATURE_V2: &str = "Signature"; pub const AMZ_ACCESS_KEY_ID: &str = "AWSAccessKeyId"; -// Response request id. +// Request id headers. +pub const REQUEST_ID_HEADER: &str = "x-request-id"; pub const AMZ_REQUEST_ID: &str = "x-amz-request-id"; pub const AMZ_REQUEST_HOST_ID: &str = "x-amz-id-2"; diff --git a/rustfs/src/server/layer.rs b/rustfs/src/server/layer.rs index 7380b407a..15bac6720 100644 --- a/rustfs/src/server/layer.rs +++ b/rustfs/src/server/layer.rs @@ -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> { - let headers = self - .0 - .get_all(key) - .iter() - .filter_map(|value| value.to_str().ok()) - .collect::>(); - - 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) -> 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::() + .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(); diff --git a/rustfs/src/storage/request_context.rs b/rustfs/src/storage/request_context.rs index 56d781245..8829e1b3e 100644 --- a/rustfs/src/storage/request_context.rs +++ b/rustfs/src/storage/request_context.rs @@ -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();