From 4287df08821c7472e111f17eb6b0d13b8d930160 Mon Sep 17 00:00:00 2001 From: Hauser Date: Wed, 28 Oct 2026 05:54:56 +0800 Subject: [PATCH 1/5] feat(ecstore): add sampled batch CPU diagnostics Keep timing and request correlation in a separate default-off feature. Preserve the existing positioned-read backend, byte and error contracts. Co-Authored-By: heihutu Co-Authored-By: zhi22915 --- crates/ecstore/Cargo.toml | 2 + crates/ecstore/src/batch_read_diagnostics.rs | 525 ++++++++++++++++++ crates/ecstore/src/disk/local.rs | 73 ++- crates/ecstore/src/lib.rs | 2 + docs/README.md | 2 + .../operations/local-batch-cpu-diagnostics.md | 60 ++ rustfs/Cargo.toml | 2 + rustfs/src/server/layer.rs | 114 ++++ scripts/check_logging_guardrails.sh | 1 + 9 files changed, 768 insertions(+), 13 deletions(-) create mode 100644 crates/ecstore/src/batch_read_diagnostics.rs create mode 100644 docs/operations/local-batch-cpu-diagnostics.md diff --git a/crates/ecstore/Cargo.toml b/crates/ecstore/Cargo.toml index b37b4b926..566917eea 100644 --- a/crates/ecstore/Cargo.toml +++ b/crates/ecstore/Cargo.toml @@ -32,6 +32,8 @@ workspace = true [features] default = [] diagnostic-read-version-caller = ["rustfs-io-metrics/diagnostic-read-version-caller"] +# Sampled wall/CPU diagnostics only; does not change the read backend. +diagnostic-local-batch = ["rustix/time"] gcs = ["dep:google-cloud-storage", "dep:google-cloud-auth"] rio-v2 = ["dep:rustfs-rio-v2"] hotpath = [ diff --git a/crates/ecstore/src/batch_read_diagnostics.rs b/crates/ecstore/src/batch_read_diagnostics.rs new file mode 100644 index 000000000..5be19daee --- /dev/null +++ b/crates/ecstore/src/batch_read_diagnostics.rs @@ -0,0 +1,525 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Sampled local batch diagnostics for a separate, default-off build. +//! +//! Events contain server-owned correlation IDs, sizes and timing, never paths +//! or I/O error strings. They are not metrics or performance acceptance. +//! Linux supplies thread CPU time; other targets report it absent. + +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, OnceLock}; +use std::time::{Instant, SystemTime, UNIX_EPOCH}; +use tracing::{Dispatch, Level}; + +const LOG_COMPONENT_ECSTORE: &str = "ecstore"; +const LOG_SUBSYSTEM_LOCAL_BATCH: &str = "local_batch_diagnostic"; +const EVENT_LOCAL_BATCH_DIAGNOSTIC: &str = "local_batch_diagnostic"; +const TARGET: &str = "rustfs_ecstore::batch_read_diagnostics"; +const ENV_SAMPLE_EVERY: &str = "RUSTFS_DIAGNOSTIC_LOCAL_BATCH_SAMPLE_EVERY"; +const MAX_SAMPLE_EVERY: u64 = 65_536; +static SAMPLE_EVERY: OnceLock = OnceLock::new(); +static NEXT_BATCH: AtomicU64 = AtomicU64::new(0); + +tokio::task_local! { + static REQUEST_ID: Option>; +} + +fn parse_rate(value: Option<&str>) -> u64 { + value + .and_then(|value| value.parse::().ok()) + .filter(|rate| (1..=MAX_SAMPLE_EVERY).contains(rate)) + .unwrap_or(0) +} + +fn enabled_rate() -> u64 { + *SAMPLE_EVERY.get_or_init(|| parse_rate(std::env::var(ENV_SAMPLE_EVERY).ok().as_deref())) +} + +fn recording_enabled() -> bool { + enabled_rate() != 0 && tracing::enabled!(target: TARGET, Level::TRACE) +} + +fn valid_request_id(id: &str) -> bool { + !id.is_empty() + && id.len() <= 128 + && id + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_')) +} + +/// Retain a bounded server-owned canonical ID only while diagnostics are enabled. +/// +/// The external S3 boundary must supply its canonical ID, not a client header. +pub fn request_context(id: &str) -> Option> { + (recording_enabled() && valid_request_id(id)).then(|| Arc::from(id)) +} + +/// Poll a request under its diagnostic ID without retaining request fields. +pub fn in_request_context(id: Option<&Arc>, operation: impl FnOnce() -> T) -> T { + REQUEST_ID.sync_scope(id.cloned(), operation) +} + +/// Return the current ID for explicit diagnostic propagation checks. +pub fn current_request_id() -> Option> { + REQUEST_ID.try_with(Clone::clone).ok().flatten() +} + +#[derive(Clone, Copy)] +pub(crate) enum Stage { + Execution, + Metadata, + Open, + Allocation, + Read, +} + +impl Stage { + fn as_str(self) -> &'static str { + match self { + Self::Execution => "execution", + Self::Metadata => "metadata", + Self::Open => "open", + Self::Allocation => "allocation", + Self::Read => "read", + } + } +} + +fn unix_ns() -> Option { + SystemTime::now().duration_since(UNIX_EPOCH).ok()?.as_nanos().try_into().ok() +} + +#[cfg(target_os = "linux")] +fn cpu_ns() -> Option { + use rustix::time::{ClockId, DynamicClockId, clock_gettime_dynamic}; + let time = clock_gettime_dynamic(DynamicClockId::Known(ClockId::ThreadCPUTime)).ok()?; + let seconds = u64::try_from(time.tv_sec).ok()?; + let nanos = u64::try_from(time.tv_nsec).ok().filter(|value| *value < 1_000_000_000)?; + seconds.checked_mul(1_000_000_000)?.checked_add(nanos) +} + +#[cfg(not(target_os = "linux"))] +fn cpu_ns() -> Option { + None +} + +fn thread_id() -> Option { + #[cfg(target_os = "linux")] + { + thread_local! { static TID: i32 = rustix::thread::gettid().as_raw_nonzero().get(); } + Some(TID.with(|id| *id)) + } + #[cfg(not(target_os = "linux"))] + { + None + } +} + +pub(crate) struct Started { + wall: Instant, + unix_ns: Option, + cpu_ns: Option, +} +struct Timing { + unix_ns: Option, + wall_ns: Option, + cpu_ns: Option, + cpu_status: &'static str, +} + +pub(crate) struct Sample { + batch_id: u64, + request_id: Option>, + submitted: Instant, + submitted_unix_ns: Option, + dispatch: Dispatch, +} + +impl Sample { + pub(crate) fn capture() -> Option { + Self::capture_with(enabled_rate(), tracing::enabled!(target: TARGET, Level::TRACE), &NEXT_BATCH) + } + + fn capture_with(rate: u64, enabled: bool, sequence: &AtomicU64) -> Option { + if rate == 0 || !enabled { + return None; + } + let batch_id = sequence.fetch_add(1, Ordering::Relaxed); + if batch_id % rate != 0 { + return None; + } + Some(Self { + batch_id, + request_id: current_request_id(), + submitted: Instant::now(), + submitted_unix_ns: unix_ns(), + dispatch: tracing::dispatcher::get_default(Clone::clone), + }) + } + + pub(crate) fn dispatch(&self) -> &Dispatch { + &self.dispatch + } + + pub(crate) fn start(&self) -> Started { + Started { + wall: Instant::now(), + unix_ns: unix_ns(), + cpu_ns: cpu_ns(), + } + } + + pub(crate) fn worker_started(&self) { + self.emit( + "queue_wait", + None, + true, + Timing { + unix_ns: self.submitted_unix_ns, + wall_ns: self.submitted.elapsed().as_nanos().try_into().ok(), + cpu_ns: None, + cpu_status: "not_applicable", + }, + ); + } + + pub(crate) fn finish(&self, stage: Stage, shard: Option, start: Started, success: bool) { + let cpu = start + .cpu_ns + .zip(cpu_ns()) + .and_then(|(before, after)| after.checked_sub(before)); + self.emit( + stage.as_str(), + shard, + success, + Timing { + unix_ns: start.unix_ns, + wall_ns: start.wall.elapsed().as_nanos().try_into().ok(), + cpu_ns: cpu, + cpu_status: if cpu.is_some() { "available" } else { "unavailable" }, + }, + ); + } + + fn emit(&self, stage: &'static str, shard: Option, success: bool, timing: Timing) { + tracing::trace!(target: TARGET, parent: None, + event = EVENT_LOCAL_BATCH_DIAGNOSTIC, component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LOCAL_BATCH, result = if success { "success" } else { "error" }, + stage, batch_id = self.batch_id, request_id = self.request_id.as_deref(), + request_context = if self.request_id.is_some() { "present" } else { "missing" }, + pid = std::process::id(), tid = thread_id(), shard_index = shard, + submitted_unix_ns = self.submitted_unix_ns, start_unix_ns = timing.unix_ns, + wall_ns = timing.wall_ns, cpu_ns = timing.cpu_ns, cpu_clock_status = timing.cpu_status, + "Sampled local batch stage"); + } + + pub(crate) fn shard_result(&self, shard: usize, requested: usize, read: usize, success: bool) { + tracing::trace!(target: TARGET, parent: None, + event = EVENT_LOCAL_BATCH_DIAGNOSTIC, component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LOCAL_BATCH, result = if success { "success" } else { "error" }, + stage = "shard_result", batch_id = self.batch_id, request_id = self.request_id.as_deref(), + request_context = if self.request_id.is_some() { "present" } else { "missing" }, + pid = std::process::id(), tid = thread_id(), submitted_unix_ns = self.submitted_unix_ns, + shard_index = shard, requested_bytes = requested, read_bytes = read, + "Sampled local batch result"); + } +} + +pub(crate) fn measure( + sample: Option<&Sample>, + stage: Stage, + shard: usize, + operation: impl FnOnce() -> Result, +) -> Result { + let Some(sample) = sample else { return operation() }; + let start = sample.start(); + let result = operation(); + sample.finish(stage, Some(shard), start, result.is_ok()); + result +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::{Value, json}; + use std::collections::BTreeMap; + use std::sync::{Barrier, Mutex, mpsc}; + use tracing::field::{Field, Visit}; + use tracing_subscriber::{Layer, layer::Context, prelude::*}; + + #[derive(Default)] + struct Values(BTreeMap); + impl Visit for Values { + fn record_u64(&mut self, field: &Field, value: u64) { + self.0.insert(field.name().into(), json!(value)); + } + fn record_i64(&mut self, field: &Field, value: i64) { + self.0.insert(field.name().into(), json!(value)); + } + fn record_bool(&mut self, field: &Field, value: bool) { + self.0.insert(field.name().into(), json!(value)); + } + fn record_str(&mut self, field: &Field, value: &str) { + self.0.insert(field.name().into(), json!(value)); + } + fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { + self.0.insert(field.name().into(), json!(format!("{value:?}"))); + } + } + + struct Records { + events: Mutex>>, + block_queue: std::sync::atomic::AtomicBool, + queue_started: Mutex>>, + release: Barrier, + } + + struct Recorder(Arc); + impl Layer for Recorder { + fn on_event(&self, event: &tracing::Event<'_>, _: Context<'_, S>) { + if event.metadata().target() != TARGET { + return; + } + assert!(event.is_root(), "diagnostic events must not inherit URI-bearing spans"); + let mut fields = Values::default(); + event.record(&mut fields); + let queue = fields.0.get("stage") == Some(&json!("queue_wait")); + self.0.events.lock().expect("record diagnostic event").push(fields.0); + if queue && self.0.block_queue.swap(false, Ordering::SeqCst) { + if let Some(sender) = self.0.queue_started.lock().expect("queue sender").take() { + sender.send(()).expect("signal worker start"); + } + self.0.release.wait(); + } + } + } + + fn records() -> Arc { + Arc::new(Records { + events: Mutex::new(Vec::new()), + block_queue: false.into(), + queue_started: Mutex::new(None), + release: Barrier::new(2), + }) + } + + #[test] + fn rate_is_default_off_and_bounded() { + for value in [None, Some("0"), Some("invalid"), Some("-1"), Some("65537")] { + assert_eq!(parse_rate(value), 0); + } + assert_eq!(parse_rate(Some("1")), 1); + assert_eq!(parse_rate(Some("64")), 64); + assert_eq!(parse_rate(Some("65536")), MAX_SAMPLE_EVERY); + } + + #[test] + fn disabled_or_filtered_capture_does_not_advance_sampling() { + let sequence = AtomicU64::new(0); + assert!(Sample::capture_with(0, true, &sequence).is_none()); + assert!(Sample::capture_with(64, false, &sequence).is_none()); + assert_eq!(sequence.load(Ordering::Relaxed), 0); + let result = measure::<(), _>(None, Stage::Read, 0, || Err("original error")); + assert_eq!(result, Err("original error")); + } + + #[test] + fn sampling_keeps_the_registered_interval() { + let sequence = AtomicU64::new(0); + let selected: Vec<_> = (0..12) + .filter_map(|_| Sample::capture_with(4, true, &sequence).map(|sample| sample.batch_id)) + .collect(); + assert_eq!(selected, [0, 4, 8]); + } + + #[test] + fn request_context_is_bounded_and_restored_on_unwind() { + assert!(valid_request_id("server-owned-uuid_123")); + assert!(valid_request_id(&"a".repeat(128))); + for value in ["", "secret/path", "line\nbreak", &"a".repeat(129)] { + assert!(!valid_request_id(value)); + } + let outer: Arc = Arc::from("server-outer"); + in_request_context(Some(&outer), || { + in_request_context(None, || assert!(current_request_id().is_none())); + assert_eq!(current_request_id().as_deref(), Some("server-outer")); + let nested: Arc = Arc::from("server-inner"); + let result = std::panic::catch_unwind(|| in_request_context(Some(&nested), || panic!("test unwind"))); + assert!(result.is_err()); + assert_eq!(current_request_id().as_deref(), Some("server-outer")); + }); + assert!(current_request_id().is_none()); + } + + #[test] + fn failed_partial_operation_preserves_error_and_initialized_byte_count() { + let _interest = crate::test_tracing::pin_callsite_interest_for_test(); + let state = records(); + let subscriber = tracing_subscriber::registry().with(Recorder(Arc::clone(&state))); + tracing::subscriber::with_default(subscriber, || { + let parent = tracing::info_span!("request", uri = "/must-not-copy-parent-path"); + let _entered = parent.enter(); + let sample = Sample::capture_with(1, true, &AtomicU64::new(0)).expect("sampled diagnostic"); + let mut read = 0; + let result = measure::<(), _>(Some(&sample), Stage::Read, 0, || { + read = 2; + Err("partial read failure") + }); + assert_eq!(result, Err("partial read failure")); + sample.shard_result(0, 4, read, false); + }); + let events = state.events.lock().expect("read events"); + assert!( + events + .iter() + .any(|e| e.get("stage") == Some(&json!("read")) && e.get("result") == Some(&json!("error"))) + ); + assert!( + events + .iter() + .any(|e| e.get("read_bytes") == Some(&json!(2)) && e.get("requested_bytes") == Some(&json!(4))) + ); + } + + #[test] + fn unavailable_cpu_is_absent_and_not_a_zero_measurement() { + let _interest = crate::test_tracing::pin_callsite_interest_for_test(); + let state = records(); + let subscriber = tracing_subscriber::registry().with(Recorder(Arc::clone(&state))); + tracing::subscriber::with_default(subscriber, || { + let sample = Sample::capture_with(1, true, &AtomicU64::new(0)).expect("diagnostic sample"); + sample.emit( + "metadata", + Some(0), + true, + Timing { + unix_ns: None, + wall_ns: Some(1), + cpu_ns: None, + cpu_status: "unavailable", + }, + ); + sample.emit( + "metadata", + Some(0), + true, + Timing { + unix_ns: None, + wall_ns: Some(1), + cpu_ns: Some(0), + cpu_status: "available", + }, + ); + }); + let events = state.events.lock().expect("CPU availability events"); + assert!(!events[0].contains_key("cpu_ns")); + assert_eq!(events[0]["cpu_clock_status"], "unavailable"); + assert_eq!(events[1]["cpu_ns"], 0); + assert_eq!(events[1]["cpu_clock_status"], "available"); + } + + #[cfg(target_os = "linux")] + #[test] + fn thread_cpu_clock_is_available_without_zero_substitution() { + let before = cpu_ns().expect("Linux thread CPU clock available"); + for value in 0..10_000 { + std::hint::black_box(value); + } + let after = cpu_ns().expect("Linux thread CPU clock remains available"); + assert!(after >= before); + } + + #[cfg(unix)] + #[test] + fn production_batch_diagnostic_survives_caller_cancellation() { + const CHILD: &str = "RUSTFS_TEST_BATCH_DIAGNOSTIC_CHILD"; + if std::env::var_os(CHILD).is_none() { + let status = std::process::Command::new(std::env::current_exe().expect("current test executable")) + .args([ + "--exact", + "batch_read_diagnostics::tests::production_batch_diagnostic_survives_caller_cancellation", + "--nocapture", + ]) + .env(CHILD, "1") + .env(ENV_SAMPLE_EVERY, "1") + .status() + .expect("run isolated diagnostic test"); + assert!(status.success(), "isolated production diagnostic failed"); + return; + } + let _interest = crate::test_tracing::pin_callsite_interest_for_test(); + let state = records(); + let subscriber = tracing_subscriber::registry().with(Recorder(Arc::clone(&state))); + tracing::subscriber::with_default(subscriber, || { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("test runtime"); + runtime.block_on(async { + let directory = tempfile::tempdir().expect("diagnostic test directory"); + let path = directory.path().join("must-not-appear-in-trace"); + std::fs::write(&path, b"abcdef").expect("write payload"); + let results = REQUEST_ID + .scope( + Some(Arc::from("server-original")), + crate::disk::local::batch_shard_pread(vec![ + (path.clone(), 1, 3), + (directory.path().join("missing-secret-path"), 0, 3), + (path.clone(), 0, 0), + (path.clone(), 7, 1), + ]), + ) + .await; + assert_eq!(results[0].as_ref().expect("positioned read").as_ref(), b"bcd"); + assert!(results[1].is_err()); + assert!(results[2].as_ref().expect("zero-length read").is_empty()); + assert!(results[3].is_err()); + { + let events = state.events.lock().expect("normal events"); + assert!( + !events + .iter() + .any(|e| e.get("stage") == Some(&json!("read")) && e.get("shard_index") == Some(&json!(2))) + ); + assert!(!format!("{events:?}").contains("must-not-appear")); + assert!(!format!("{events:?}").contains("missing-secret")); + } + let (sender, receiver) = mpsc::channel(); + *state.queue_started.lock().expect("install queue signal") = Some(sender); + state.block_queue.store(true, Ordering::SeqCst); + let mut future = Box::pin(REQUEST_ID.scope( + Some(Arc::from("server-cancelled")), + crate::disk::local::batch_shard_pread(vec![(path, 0, 6)]), + )); + assert!(futures::poll!(future.as_mut()).is_pending()); + receiver + .recv_timeout(std::time::Duration::from_secs(5)) + .expect("blocking worker started"); + drop(future); + state.release.wait(); + let deadline = Instant::now() + std::time::Duration::from_secs(5); + loop { + if state.events.lock().expect("completion events").iter().any(|e| { + e.get("stage") == Some(&json!("execution")) && e.get("request_id") == Some(&json!("server-cancelled")) + }) { + break; + } + assert!(Instant::now() < deadline, "detached worker must complete diagnostic"); + tokio::time::sleep(std::time::Duration::from_millis(1)).await; + } + }); + }); + } +} diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 27eda8bf0..2ebc4d769 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -11309,34 +11309,81 @@ impl LocalDisk { #[cfg(unix)] pub(crate) async fn batch_shard_pread(requests: Vec<(std::path::PathBuf, usize, usize)>) -> Vec> { let n = requests.len(); + #[cfg(feature = "diagnostic-local-batch")] + let diagnostic = crate::batch_read_diagnostics::Sample::capture(); tokio::task::spawn_blocking(move || { use std::os::unix::fs::FileExt; + // The disabled build expands directly to the original operation. + macro_rules! diagnostic_stage { + ($stage:ident, $index:expr, $operation:expr) => {{ + #[cfg(feature = "diagnostic-local-batch")] + { + crate::batch_read_diagnostics::measure( + diagnostic.as_ref(), + crate::batch_read_diagnostics::Stage::$stage, + $index, + $operation, + ) + } + #[cfg(not(feature = "diagnostic-local-batch"))] + { + $operation() + } + }}; + } + #[cfg(feature = "diagnostic-local-batch")] + let _dispatch = diagnostic + .as_ref() + .map(|sample| tracing::dispatcher::set_default(sample.dispatch())); + #[cfg(feature = "diagnostic-local-batch")] + let execution = diagnostic.as_ref().map(|sample| { + sample.worker_started(); + sample.start() + }); let mut results = Vec::with_capacity(n); - for (file_path, offset, length) in requests { + for (_index, (file_path, offset, length)) in requests.into_iter().enumerate() { + let mut total = 0usize; let r = (|| -> Result { - let meta = std::fs::metadata(&file_path).map_err(DiskError::from)?; + let meta = diagnostic_stage!(Metadata, _index, || std::fs::metadata(&file_path).map_err(DiskError::from))?; let end = offset.checked_add(length).ok_or(DiskError::FileCorrupt)?; if meta.len() < u64::try_from(end).unwrap_or(u64::MAX) { return Err(DiskError::FileCorrupt); } - let file = std::fs::File::open(&file_path).map_err(DiskError::from)?; - let mut buf = vec![0u8; length]; - let mut total = 0usize; - while total < length { - let nbytes = file - .read_at(&mut buf[total..], u64::try_from(offset + total).unwrap_or(u64::MAX)) - .map_err(DiskError::from)?; - if nbytes == 0 { - return Err(DiskError::FileCorrupt); - } - total += nbytes; + let file = diagnostic_stage!(Open, _index, || std::fs::File::open(&file_path).map_err(DiskError::from))?; + let mut buf = diagnostic_stage!(Allocation, _index, || Ok::<_, DiskError>(vec![0u8; length]))?; + if length > 0 { + diagnostic_stage!(Read, _index, || { + while total < length { + let nbytes = file + .read_at(&mut buf[total..], u64::try_from(offset + total).unwrap_or(u64::MAX)) + .map_err(DiskError::from)?; + if nbytes == 0 { + return Err(DiskError::FileCorrupt); + } + total += nbytes; + } + Ok::<_, DiskError>(()) + })?; } Ok(Bytes::from(buf)) })(); + #[cfg(feature = "diagnostic-local-batch")] + if let Some(sample) = &diagnostic { + sample.shard_result(_index, length, total, r.is_ok()); + } results.push(r); } + #[cfg(feature = "diagnostic-local-batch")] + if let (Some(sample), Some(started)) = (&diagnostic, execution) { + sample.finish( + crate::batch_read_diagnostics::Stage::Execution, + None, + started, + results.iter().all(Result::is_ok), + ); + } results }) .await diff --git a/crates/ecstore/src/lib.rs b/crates/ecstore/src/lib.rs index 2e44df108..e38368627 100644 --- a/crates/ecstore/src/lib.rs +++ b/crates/ecstore/src/lib.rs @@ -33,6 +33,8 @@ macro_rules! hp_guard { } pub mod api; +#[cfg(feature = "diagnostic-local-batch")] +pub mod batch_read_diagnostics; mod bucket; mod cache_value; mod cluster; diff --git a/docs/README.md b/docs/README.md index 6ce330707..b3b2a359d 100644 --- a/docs/README.md +++ b/docs/README.md @@ -8,6 +8,8 @@ collection: ## Operations +- [Sampled local batch CPU diagnostics](operations/local-batch-cpu-diagnostics.md) — default-off request-correlated allocation and worker wall/CPU diagnosis. + - [Retained MRF responsibilities](operations/mrf-responsibilities.md) — unknown bucket identities, bounded probes, and unreadable checkpoint recovery. - [Read-only BTF declaration inspection](operations/btf-prototype-inspection.md) — bounded prototypes and scalar member layouts, explicit void handling, and independent native references. diff --git a/docs/operations/local-batch-cpu-diagnostics.md b/docs/operations/local-batch-cpu-diagnostics.md new file mode 100644 index 000000000..4fcfcf858 --- /dev/null +++ b/docs/operations/local-batch-cpu-diagnostics.md @@ -0,0 +1,60 @@ +# Sampled local batch CPU diagnostics + +**Use this when:** separating sampled local EC batch allocation/initialization +and filesystem wall time from worker CPU time in a default-off diagnostic build. + +This diagnostic build separates allocation/initialization, metadata, open and +positioned-read elapsed time from Linux thread CPU time. It preserves the +single blocking dispatch, buffer initialization, offsets, integrity checks and +error handling. It does not implement a rustix optimization or qualify +observation overhead. + +Build separately from any frozen performance binary: + +```sh +RUSTUP_TOOLCHAIN=stable cargo build --release --locked -p rustfs --bin rustfs \ + --features diagnostic-local-batch +``` + +The feature is absent from default builds. Even this diagnostic build records +nothing until both the sample rate and the target filter are enabled: + +```sh +export RUSTFS_DIAGNOSTIC_LOCAL_BATCH_SAMPLE_EVERY=64 +export RUST_LOG=warn,rustfs_ecstore::batch_read_diagnostics=trace +``` + +The rate is an integer in `1..=65536`. Missing, zero or invalid values disable +capture. Rate 64 selects one in every 64 submitted batches; rate 1 is intended +only for small correctness probes. Restart to change the rate. Disabled or +unselected batches create no timing records and read no diagnostic clocks. +While recording is enabled, each external S3 request retains a bounded shared +ID, including requests whose batches are not selected. This is diagnostic work +and must be accounted for before any observation-cost claim. + +Events contain a process-local batch ID, the bounded server-owned canonical +request ID when propagated, worker PID/TID, shard index, sizes and timing. +They explicitly have no parent span, so URI-bearing HTTP span fields are not +copied. There are no variable-ID metric labels. Missing request context is +reported as `request_context=missing`, never treated as a successful join. + +Queue wait is elapsed time across the async/blocking boundary and has no +thread CPU delta. Execution and per-shard stages run synchronously on one +worker thread. CPU clock failure or unsupported platforms omit `cpu_ns` and +report `cpu_clock_status=unavailable`; this is different from an available +zero-duration measurement. Clock and logging work perturb diagnostic samples. +Elapsed and CPU are measured separately and must not be assumed equal or +subtracted without accounting for measurement overhead. Stage durations can +overlap and per-shard samples differ from per-batch samples. + +`shard_result` reports requested and actually read bytes, including partial +reads on error. Unattempted stages are absent; a zero-length shard has no read +sample. Captured correlation and dispatch stay with a blocking worker even +after its caller is cancelled. + +Keep the diagnostic binary, source revision, feature, rate, filters and raw +events distinct from baseline or acceptance artifacts. Do not log paths, +credentials, request bodies or arbitrary client-supplied identifiers to add +correlation. Request-level client association must use the server's canonical +response ID. Diagnostic results locate hypotheses; they do not regrade failed +A/A or ABBA runs. diff --git a/rustfs/Cargo.toml b/rustfs/Cargo.toml index 7338ed8b3..a7c6319bf 100644 --- a/rustfs/Cargo.toml +++ b/rustfs/Cargo.toml @@ -79,6 +79,8 @@ pyroscope = ["rustfs-obs/pyroscope", "dep:pyroscope", "dep:object", "dep:symboli dial9 = ["rustfs-obs/dial9"] # Diagnostic-only Rust GlobalAlloc live-byte accounting; adds per-allocation atomic updates. diagnostic-allocator-live = [] +# Default-off sampled local batch wall/CPU and request correlation diagnostics. +diagnostic-local-batch = ["rustfs-ecstore/diagnostic-local-batch"] # Diagnostic-only S3-operation attribution for internode ReadVersion and list-walker metrics. diagnostic-read-version-caller = ["rustfs-ecstore/diagnostic-read-version-caller", "rustfs-io-metrics/diagnostic-read-version-caller", "rustfs-scanner/diagnostic-read-version-caller"] http3 = ["dep:h3", "dep:h3-quinn", "dep:quinn"] diff --git a/rustfs/src/server/layer.rs b/rustfs/src/server/layer.rs index b58098dda..55553ae11 100644 --- a/rustfs/src/server/layer.rs +++ b/rustfs/src/server/layer.rs @@ -73,6 +73,10 @@ const HTTP_REQUEST_INFLIGHT_SLOW_EVENT: &str = "http_request_inflight_slow"; const LOG_COMPONENT_SERVER: &str = "server"; const LOG_SUBSYSTEM_HTTP: &str = "http"; const REDACTED_QUERY_VALUE: &str = "redacted"; +#[cfg(feature = "diagnostic-local-batch")] +type BatchDiagnosticRequest = Option>; +#[cfg(not(feature = "diagnostic-local-batch"))] +type BatchDiagnosticRequest = (); const OBJECT_ZIP_DOWNLOADS_PATH: &str = "/v3/object-zip-downloads/"; const HTTP_REQUEST_INFLIGHT_WARN_THRESHOLD: Duration = Duration::from_secs(5); static HTTP_SERVER_ERROR_LOGS: [rustfs_utils::LogThrottle; 100] = [const { rustfs_utils::LogThrottle::new(5_000) }; 100]; @@ -285,6 +289,12 @@ where } else { req.headers().get(REQUEST_ID_HEADER).cloned() }; + #[cfg(feature = "diagnostic-local-batch")] + let diagnostic_request = is_s3 + .then(|| rustfs_ecstore::batch_read_diagnostics::request_context(&request_context.request_id)) + .flatten(); + #[cfg(not(feature = "diagnostic-local-batch"))] + let diagnostic_request = (); req.extensions_mut().insert(request_context); // This outer boundary includes readiness, rate-limit and auth @@ -300,6 +310,13 @@ where .flatten(); let head_request = req.method() == Method::HEAD; let mut metrics = is_s3.then(|| s3_http_request_guard(req.method().as_str())); + #[cfg(feature = "diagnostic-local-batch")] + let inner = + rustfs_ecstore::batch_read_diagnostics::in_request_context(diagnostic_request.as_ref(), || match metrics.as_mut() { + Some(metrics) => metrics.in_scope(|| self.inner.call(req)), + None => self.inner.call(req), + }); + #[cfg(not(feature = "diagnostic-local-batch"))] let inner = match metrics.as_mut() { Some(metrics) => metrics.in_scope(|| self.inner.call(req)), None => self.inner.call(req), @@ -309,6 +326,7 @@ where } ExternalRequestContextFuture { inner, + diagnostic_request, request_id, is_s3, metrics, @@ -358,6 +376,7 @@ pin_project! { pub struct ExternalRequestContextFuture { #[pin] inner: F, + diagnostic_request: BatchDiagnosticRequest, request_id: Option, is_s3: bool, metrics: Option, @@ -374,6 +393,15 @@ where fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { let this = self.project(); + #[cfg(feature = "diagnostic-local-batch")] + let result = rustfs_ecstore::batch_read_diagnostics::in_request_context(this.diagnostic_request.as_ref(), || match this + .metrics + .as_mut() + { + Some(metrics) => metrics.in_scope(|| this.inner.poll(cx)), + None => this.inner.poll(cx), + }); + #[cfg(not(feature = "diagnostic-local-batch"))] let result = match this.metrics.as_mut() { Some(metrics) => metrics.in_scope(|| this.inner.poll(cx)), None => this.inner.poll(cx), @@ -2608,6 +2636,92 @@ fn rewrite_double_slash_root(uri: &Uri) -> Option { #[cfg(test)] mod tests { + #[cfg(feature = "diagnostic-local-batch")] + #[test] + fn diagnostic_external_request_uses_server_id_not_client_header() { + const CHILD: &str = "RUSTFS_TEST_BATCH_REQUEST_CHILD"; + if std::env::var_os(CHILD).is_none() { + let status = std::process::Command::new(std::env::current_exe().expect("current test executable")) + .args([ + "--exact", + "server::layer::tests::diagnostic_external_request_uses_server_id_not_client_header", + "--nocapture", + ]) + .env(CHILD, "1") + .env("RUSTFS_DIAGNOSTIC_LOCAL_BATCH_SAMPLE_EVERY", "1") + .status() + .expect("run isolated request correlation test"); + assert!(status.success(), "isolated request correlation test failed"); + return; + } + let subscriber = tracing_subscriber::fmt() + .with_max_level(tracing::Level::TRACE) + .with_writer(std::io::sink) + .finish(); + tracing::subscriber::with_default(subscriber, || { + let inner = tower::service_fn(|request: HttpRequest>| { + let context = request + .extensions() + .get::() + .expect("canonical request context"); + let expected = context.request_id.clone(); + assert_ne!(expected, "client-controlled-id"); + assert_eq!( + rustfs_ecstore::batch_read_diagnostics::current_request_id().as_deref(), + Some(expected.as_str()) + ); + async move { + assert_eq!( + rustfs_ecstore::batch_read_diagnostics::current_request_id().as_deref(), + Some(expected.as_str()) + ); + Ok::<_, std::convert::Infallible>(Response::new(Full::new(Bytes::new()))) + } + }); + let mut service = ExternalRequestContextLayer::new(false).layer(inner); + let request = HttpRequest::builder() + .method(Method::GET) + .uri("/diagnostic-bucket/object") + .header(REQUEST_ID_HEADER, "client-controlled-id") + .body(Full::new(Bytes::new())) + .expect("test request"); + let response = futures::executor::block_on(service.call(request)).expect("correlated response"); + assert_eq!(response.status(), StatusCode::OK); + assert!(rustfs_ecstore::batch_read_diagnostics::current_request_id().is_none()); + }); + } + + #[cfg(feature = "diagnostic-local-batch")] + #[test] + fn diagnostic_request_scope_is_present_on_every_poll_and_restored() { + let mut polls = 0; + let inner = std::future::poll_fn(|cx| { + assert_eq!( + rustfs_ecstore::batch_read_diagnostics::current_request_id().as_deref(), + Some("server-owned-test") + ); + polls += 1; + if polls == 1 { + cx.waker().wake_by_ref(); + Poll::Pending + } else { + Poll::Ready(Ok::<_, std::convert::Infallible>(Response::new(http_body_util::Full::new(Bytes::new())))) + } + }); + let future = ExternalRequestContextFuture { + inner, + diagnostic_request: Some(Arc::from("server-owned-test")), + request_id: None, + is_s3: false, + metrics: None, + feedback: None, + head_request: false, + }; + let response = futures::executor::block_on(future).expect("diagnostic request future"); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(polls, 2); + assert!(rustfs_ecstore::batch_read_diagnostics::current_request_id().is_none()); + } #[tokio::test] async fn foreground_feedback_unobserved_body_is_a_direct_passthrough() { struct EndProbe(Full); diff --git a/scripts/check_logging_guardrails.sh b/scripts/check_logging_guardrails.sh index 3d315836e..bb1cdd9be 100755 --- a/scripts/check_logging_guardrails.sh +++ b/scripts/check_logging_guardrails.sh @@ -5,6 +5,7 @@ repo_root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" cd "$repo_root" checked_files=( + "crates/ecstore/src/batch_read_diagnostics.rs" "rustfs/src/main.rs" "rustfs/src/startup_entrypoint.rs" "rustfs/src/init.rs" From f4daab72b0fde85f22d1f98dc4f935e628cce323 Mon Sep 17 00:00:00 2001 From: Hauser Date: Wed, 28 Oct 2026 06:14:44 +0800 Subject: [PATCH 2/5] fix(ecstore): route batch diagnostics through storage facade Keep the diagnostic implementation private to its owner and expose explicit feature-gated helpers through the existing storage boundary. Co-Authored-By: heihutu Co-Authored-By: zhi22915 --- crates/ecstore/src/api/mod.rs | 6 ++- .../batch_read_diagnostics.rs | 4 +- crates/ecstore/src/diagnostics/mod.rs | 2 + crates/ecstore/src/disk/local.rs | 8 ++-- crates/ecstore/src/lib.rs | 2 - .../ecstore-api-facade-inventory.md | 1 + rustfs/src/server/layer.rs | 37 +++++++------------ rustfs/src/storage/storage_api.rs | 9 ++++- rustfs/src/storage_api.rs | 7 ++++ scripts/check_logging_guardrails.sh | 2 +- 10 files changed, 42 insertions(+), 36 deletions(-) rename crates/ecstore/src/{ => diagnostics}/batch_read_diagnostics.rs (99%) diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 38b431516..b08db91a1 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -14,12 +14,16 @@ //! Explicit ECStore public facades for outer crate compatibility boundaries. -#[cfg(feature = "diagnostic-read-version-caller")] +#[cfg(any(feature = "diagnostic-read-version-caller", feature = "diagnostic-local-batch"))] pub mod diagnostic { + #[cfg(feature = "diagnostic-read-version-caller")] pub use crate::cluster::rpc::runtime_sources::{ current_read_version_entry_source, observe_read_version_context_boundary, with_captured_read_version_context, with_read_version_entry, with_read_version_spawn_context, }; + #[cfg(feature = "diagnostic-local-batch")] + pub use crate::diagnostics::batch_read_diagnostics::{current_request_id, in_request_context, request_context}; + #[cfg(feature = "diagnostic-read-version-caller")] pub use rustfs_io_metrics::internode_metrics::{ReadVersionContextBoundary, ReadVersionEntrySource}; } diff --git a/crates/ecstore/src/batch_read_diagnostics.rs b/crates/ecstore/src/diagnostics/batch_read_diagnostics.rs similarity index 99% rename from crates/ecstore/src/batch_read_diagnostics.rs rename to crates/ecstore/src/diagnostics/batch_read_diagnostics.rs index 5be19daee..208fdf35c 100644 --- a/crates/ecstore/src/batch_read_diagnostics.rs +++ b/crates/ecstore/src/diagnostics/batch_read_diagnostics.rs @@ -157,7 +157,7 @@ impl Sample { return None; } let batch_id = sequence.fetch_add(1, Ordering::Relaxed); - if batch_id % rate != 0 { + if !batch_id.is_multiple_of(rate) { return None; } Some(Self { @@ -449,7 +449,7 @@ mod tests { let status = std::process::Command::new(std::env::current_exe().expect("current test executable")) .args([ "--exact", - "batch_read_diagnostics::tests::production_batch_diagnostic_survives_caller_cancellation", + "diagnostics::batch_read_diagnostics::tests::production_batch_diagnostic_survives_caller_cancellation", "--nocapture", ]) .env(CHILD, "1") diff --git a/crates/ecstore/src/diagnostics/mod.rs b/crates/ecstore/src/diagnostics/mod.rs index 514d6f850..c51383f3e 100644 --- a/crates/ecstore/src/diagnostics/mod.rs +++ b/crates/ecstore/src/diagnostics/mod.rs @@ -15,5 +15,7 @@ // #730: diagnostics constants are staged for request-path telemetry migration. pub(crate) mod admin_server_info; +#[cfg(feature = "diagnostic-local-batch")] +pub(crate) mod batch_read_diagnostics; pub(crate) mod get; pub(crate) mod object_lock; diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 2ebc4d769..f884eb7c6 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -11310,7 +11310,7 @@ impl LocalDisk { pub(crate) async fn batch_shard_pread(requests: Vec<(std::path::PathBuf, usize, usize)>) -> Vec> { let n = requests.len(); #[cfg(feature = "diagnostic-local-batch")] - let diagnostic = crate::batch_read_diagnostics::Sample::capture(); + let diagnostic = crate::diagnostics::batch_read_diagnostics::Sample::capture(); tokio::task::spawn_blocking(move || { use std::os::unix::fs::FileExt; @@ -11319,9 +11319,9 @@ pub(crate) async fn batch_shard_pread(requests: Vec<(std::path::PathBuf, usize, ($stage:ident, $index:expr, $operation:expr) => {{ #[cfg(feature = "diagnostic-local-batch")] { - crate::batch_read_diagnostics::measure( + crate::diagnostics::batch_read_diagnostics::measure( diagnostic.as_ref(), - crate::batch_read_diagnostics::Stage::$stage, + crate::diagnostics::batch_read_diagnostics::Stage::$stage, $index, $operation, ) @@ -11378,7 +11378,7 @@ pub(crate) async fn batch_shard_pread(requests: Vec<(std::path::PathBuf, usize, #[cfg(feature = "diagnostic-local-batch")] if let (Some(sample), Some(started)) = (&diagnostic, execution) { sample.finish( - crate::batch_read_diagnostics::Stage::Execution, + crate::diagnostics::batch_read_diagnostics::Stage::Execution, None, started, results.iter().all(Result::is_ok), diff --git a/crates/ecstore/src/lib.rs b/crates/ecstore/src/lib.rs index e38368627..2e44df108 100644 --- a/crates/ecstore/src/lib.rs +++ b/crates/ecstore/src/lib.rs @@ -33,8 +33,6 @@ macro_rules! hp_guard { } pub mod api; -#[cfg(feature = "diagnostic-local-batch")] -pub mod batch_read_diagnostics; mod bucket; mod cache_value; mod cluster; diff --git a/docs/architecture/ecstore-api-facade-inventory.md b/docs/architecture/ecstore-api-facade-inventory.md index 3f14f4cea..4fb8cb7e1 100644 --- a/docs/architecture/ecstore-api-facade-inventory.md +++ b/docs/architecture/ecstore-api-facade-inventory.md @@ -15,6 +15,7 @@ The broad `rustfs_ecstore::api` facade is a compatibility boundary, not an archi | `data_usage`, `capacity`, `notification`, `metrics`, `rebalance`, `integrity` | Domain and service facades consumed through owner-local boundaries. | Narrow one group at a time after explicit aliases or wrappers exist. | | `set_disk`, `object`, `object_api_utils`, `rio`, `bitrot`, `erasure`, `compression`, `cache`, `store_list` | Low-level object IO, reader, erasure, cache, and migration helper compatibility. | Keep stable while `SetDisks` remains the shared state carrier. | | `admin`, `event`, `global` | Admin, event hook, and bootstrap-global compatibility. | `global` is limited to bootstrap writes and lifecycle controls; read-only runtime access goes through `runtime`. | +| `diagnostic` | Feature-gated read-version attribution and sampled local-batch request context. | Keep explicit exports and default-off gates; consumers use the storage-owner and root-local diagnostic boundaries. | The S3 client is no longer a facade group: it lives in `crates/s3-client` (`rustfs_s3_client`). Regenerate the group list with: diff --git a/rustfs/src/server/layer.rs b/rustfs/src/server/layer.rs index 55553ae11..9dd36e107 100644 --- a/rustfs/src/server/layer.rs +++ b/rustfs/src/server/layer.rs @@ -29,6 +29,8 @@ use crate::server::{ }; use crate::shared_types::ReadinessDegradedReason; use crate::storage_api::server::layer::apply_cors_headers; +#[cfg(feature = "diagnostic-local-batch")] +use crate::storage_api::server::layer::batch_diagnostic; use crate::storage_api::server::layer::request_context::{RequestContext, extract_request_id_from_headers, spawn_traced}; use bytes::{Bytes, BytesMut}; use futures::future::Either; @@ -291,7 +293,7 @@ where }; #[cfg(feature = "diagnostic-local-batch")] let diagnostic_request = is_s3 - .then(|| rustfs_ecstore::batch_read_diagnostics::request_context(&request_context.request_id)) + .then(|| batch_diagnostic::request_context(&request_context.request_id)) .flatten(); #[cfg(not(feature = "diagnostic-local-batch"))] let diagnostic_request = (); @@ -311,11 +313,10 @@ where let head_request = req.method() == Method::HEAD; let mut metrics = is_s3.then(|| s3_http_request_guard(req.method().as_str())); #[cfg(feature = "diagnostic-local-batch")] - let inner = - rustfs_ecstore::batch_read_diagnostics::in_request_context(diagnostic_request.as_ref(), || match metrics.as_mut() { - Some(metrics) => metrics.in_scope(|| self.inner.call(req)), - None => self.inner.call(req), - }); + let inner = batch_diagnostic::in_request_context(diagnostic_request.as_ref(), || match metrics.as_mut() { + Some(metrics) => metrics.in_scope(|| self.inner.call(req)), + None => self.inner.call(req), + }); #[cfg(not(feature = "diagnostic-local-batch"))] let inner = match metrics.as_mut() { Some(metrics) => metrics.in_scope(|| self.inner.call(req)), @@ -394,10 +395,7 @@ where fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { let this = self.project(); #[cfg(feature = "diagnostic-local-batch")] - let result = rustfs_ecstore::batch_read_diagnostics::in_request_context(this.diagnostic_request.as_ref(), || match this - .metrics - .as_mut() - { + let result = batch_diagnostic::in_request_context(this.diagnostic_request.as_ref(), || match this.metrics.as_mut() { Some(metrics) => metrics.in_scope(|| this.inner.poll(cx)), None => this.inner.poll(cx), }); @@ -2666,15 +2664,9 @@ mod tests { .expect("canonical request context"); let expected = context.request_id.clone(); assert_ne!(expected, "client-controlled-id"); - assert_eq!( - rustfs_ecstore::batch_read_diagnostics::current_request_id().as_deref(), - Some(expected.as_str()) - ); + assert_eq!(batch_diagnostic::current_request_id().as_deref(), Some(expected.as_str())); async move { - assert_eq!( - rustfs_ecstore::batch_read_diagnostics::current_request_id().as_deref(), - Some(expected.as_str()) - ); + assert_eq!(batch_diagnostic::current_request_id().as_deref(), Some(expected.as_str())); Ok::<_, std::convert::Infallible>(Response::new(Full::new(Bytes::new()))) } }); @@ -2687,7 +2679,7 @@ mod tests { .expect("test request"); let response = futures::executor::block_on(service.call(request)).expect("correlated response"); assert_eq!(response.status(), StatusCode::OK); - assert!(rustfs_ecstore::batch_read_diagnostics::current_request_id().is_none()); + assert!(batch_diagnostic::current_request_id().is_none()); }); } @@ -2696,10 +2688,7 @@ mod tests { fn diagnostic_request_scope_is_present_on_every_poll_and_restored() { let mut polls = 0; let inner = std::future::poll_fn(|cx| { - assert_eq!( - rustfs_ecstore::batch_read_diagnostics::current_request_id().as_deref(), - Some("server-owned-test") - ); + assert_eq!(batch_diagnostic::current_request_id().as_deref(), Some("server-owned-test")); polls += 1; if polls == 1 { cx.waker().wake_by_ref(); @@ -2720,7 +2709,7 @@ mod tests { let response = futures::executor::block_on(future).expect("diagnostic request future"); assert_eq!(response.status(), StatusCode::OK); assert_eq!(polls, 2); - assert!(rustfs_ecstore::batch_read_diagnostics::current_request_id().is_none()); + assert!(batch_diagnostic::current_request_id().is_none()); } #[tokio::test] async fn foreground_feedback_unobserved_body_is_a_direct_passthrough() { diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 90637c4c4..1a14826d6 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -28,13 +28,18 @@ pub(crate) mod ecstore_integrity { }; } -#[cfg(feature = "diagnostic-read-version-caller")] +#[cfg(any(feature = "diagnostic-read-version-caller", feature = "diagnostic-local-batch"))] pub(crate) mod ecstore_diagnostic { + #[cfg(all(test, feature = "diagnostic-local-batch"))] + pub(crate) use rustfs_ecstore::api::diagnostic::current_request_id; + #[cfg(feature = "diagnostic-read-version-caller")] pub(crate) use rustfs_ecstore::api::diagnostic::{ ReadVersionContextBoundary, ReadVersionEntrySource, with_captured_read_version_context, with_read_version_spawn_context, }; - #[cfg(test)] + #[cfg(all(test, feature = "diagnostic-read-version-caller"))] pub(crate) use rustfs_ecstore::api::diagnostic::{current_read_version_entry_source, with_read_version_entry}; + #[cfg(feature = "diagnostic-local-batch")] + pub(crate) use rustfs_ecstore::api::diagnostic::{in_request_context, request_context}; } use tokio::sync::{Mutex, OwnedMutexGuard}; use tokio_util::sync::CancellationToken; diff --git a/rustfs/src/storage_api.rs b/rustfs/src/storage_api.rs index 64328dad8..5a6a913c9 100644 --- a/rustfs/src/storage_api.rs +++ b/rustfs/src/storage_api.rs @@ -245,6 +245,13 @@ pub(crate) mod server { pub(crate) mod layer { pub(crate) use crate::storage::storage_api::apply_cors_headers; + #[cfg(feature = "diagnostic-local-batch")] + pub(crate) mod batch_diagnostic { + #[cfg(test)] + pub(crate) use crate::storage::storage_api::ecstore_diagnostic::current_request_id; + pub(crate) use crate::storage::storage_api::ecstore_diagnostic::{in_request_context, request_context}; + } + pub(crate) mod request_context { pub(crate) use crate::storage::storage_api::request_context_consumer::{ RequestContext, extract_request_id_from_headers, spawn_traced, diff --git a/scripts/check_logging_guardrails.sh b/scripts/check_logging_guardrails.sh index bb1cdd9be..e101b81cc 100755 --- a/scripts/check_logging_guardrails.sh +++ b/scripts/check_logging_guardrails.sh @@ -5,7 +5,7 @@ repo_root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" cd "$repo_root" checked_files=( - "crates/ecstore/src/batch_read_diagnostics.rs" + "crates/ecstore/src/diagnostics/batch_read_diagnostics.rs" "rustfs/src/main.rs" "rustfs/src/startup_entrypoint.rs" "rustfs/src/init.rs" From bb7b0dc5dbd6746c231f05443f12a67c4dbb0abf Mon Sep 17 00:00:00 2001 From: Hauser Date: Wed, 28 Oct 2026 06:27:39 +0800 Subject: [PATCH 3/5] fix(ecstore): retain diagnostic IDs across GET producer tasks Capture request context before spawning the legacy producer. Keep the existing producer, cancellation and read-lock lifetime intact. Co-Authored-By: heihutu Co-Authored-By: zhi22915 --- .../src/diagnostics/batch_read_diagnostics.rs | 19 +++++++++++++++++++ crates/ecstore/src/set_disk/ops/object.rs | 7 +++++-- 2 files changed, 24 insertions(+), 2 deletions(-) diff --git a/crates/ecstore/src/diagnostics/batch_read_diagnostics.rs b/crates/ecstore/src/diagnostics/batch_read_diagnostics.rs index 208fdf35c..392a4c84c 100644 --- a/crates/ecstore/src/diagnostics/batch_read_diagnostics.rs +++ b/crates/ecstore/src/diagnostics/batch_read_diagnostics.rs @@ -76,6 +76,11 @@ pub fn current_request_id() -> Option> { REQUEST_ID.try_with(Clone::clone).ok().flatten() } +/// Capture before spawning the legacy producer; task locals are not inherited. +pub(crate) fn with_captured_request_context(future: F) -> impl std::future::Future { + REQUEST_ID.scope(current_request_id(), future) +} + #[derive(Clone, Copy)] pub(crate) enum Stage { Execution, @@ -363,6 +368,20 @@ mod tests { assert!(current_request_id().is_none()); } + #[tokio::test] + async fn captured_request_context_crosses_spawn_without_leaking_to_caller() { + let original: Arc = Arc::from("server-detached-producer"); + let task = in_request_context(Some(&original), || { + tokio::spawn(with_captured_request_context(async { + tokio::task::yield_now().await; + current_request_id() + })) + }); + assert!(current_request_id().is_none()); + assert_eq!(task.await.expect("detached diagnostic context").as_deref(), Some(original.as_ref())); + assert!(current_request_id().is_none()); + } + #[test] fn failed_partial_operation_preserves_error_and_initialized_byte_count() { let _interest = crate::test_tracing::pin_callsite_interest_for_test(); diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index baede67c7..f054a93dc 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -3604,7 +3604,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { let read_policy = crate::set_disk::get_object_read_policy(); let erasure_cache = Arc::clone(&self.erasure_cache); let (fi, files, disks) = snapshot.into_owned(); - tokio::spawn(async move { + let producer = async move { let _guard = read_lock_guard; let mut writer = GetObjectDownstreamWriter::new(wd); // Do not wrap the entire read+write pipeline in `disk_read_timeout`. @@ -3687,7 +3687,10 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { } }; let _ = producer_terminal_tx.send(producer_result.map(|_| ())); - }); + }; + #[cfg(feature = "diagnostic-local-batch")] + let producer = crate::diagnostics::batch_read_diagnostics::with_captured_request_context(producer); + tokio::spawn(producer); Ok(reader) } From 0c25f15847f2be665d93641ca68a84b3ab26ef3e Mon Sep 17 00:00:00 2001 From: Hauser Date: Wed, 28 Oct 2026 07:36:39 +0800 Subject: [PATCH 4/5] fix(ecstore): avoid unused batch read index Co-Authored-By: heihutu Co-Authored-By: zhi22915 --- crates/ecstore/src/disk/local.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index f884eb7c6..617ea311e 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -11342,7 +11342,9 @@ pub(crate) async fn batch_shard_pread(requests: Vec<(std::path::PathBuf, usize, sample.start() }); let mut results = Vec::with_capacity(n); - for (_index, (file_path, offset, length)) in requests.into_iter().enumerate() { + for (file_path, offset, length) in requests { + #[cfg(feature = "diagnostic-local-batch")] + let _index = results.len(); let mut total = 0usize; let r = (|| -> Result { let meta = diagnostic_stage!(Metadata, _index, || std::fs::metadata(&file_path).map_err(DiskError::from))?; From 5e6c9b8a1ec11607b3120efd28732244f4bab0b3 Mon Sep 17 00:00:00 2001 From: Hauser Date: Wed, 28 Oct 2026 08:34:14 +0800 Subject: [PATCH 5/5] fix(obs): isolate local batch diagnostic JSON events Omit span metadata and inferred parent IDs for diagnostic root events. Verify per-shard index order and document existing production metric timing. Co-Authored-By: heihutu Co-Authored-By: zhi22915 --- crates/config/src/constants/app.rs | 3 + .../src/diagnostics/batch_read_diagnostics.rs | 12 ++- crates/obs/Cargo.toml | 2 + crates/obs/src/telemetry/local.rs | 81 ++++++++++++++++++- .../operations/local-batch-cpu-diagnostics.md | 23 ++++-- rustfs/Cargo.toml | 2 +- 6 files changed, 111 insertions(+), 12 deletions(-) diff --git a/crates/config/src/constants/app.rs b/crates/config/src/constants/app.rs index 7addb89bc..7b77036b0 100644 --- a/crates/config/src/constants/app.rs +++ b/crates/config/src/constants/app.rs @@ -374,6 +374,9 @@ pub const DEFAULT_OBS_PROFILING_EXPORT_ENABLED: bool = true; /// Environment variable: RUSTFS_OBS_LOG_STDOUT_ENABLED pub const DEFAULT_OBS_LOG_STDOUT_ENABLED: bool = false; +/// Fixed target for sampled local-batch diagnostics with isolated JSON output. +pub const LOCAL_BATCH_DIAGNOSTIC_LOG_TARGET: &str = "rustfs_local_batch_diagnostic"; + /// Constant representing 1 Kibibyte (1024 bytes) /// Default value: 1024 pub const KI_B: usize = 1024; diff --git a/crates/ecstore/src/diagnostics/batch_read_diagnostics.rs b/crates/ecstore/src/diagnostics/batch_read_diagnostics.rs index d4a478888..64a170624 100644 --- a/crates/ecstore/src/diagnostics/batch_read_diagnostics.rs +++ b/crates/ecstore/src/diagnostics/batch_read_diagnostics.rs @@ -14,8 +14,8 @@ //! Sampled local batch diagnostics for a separate, default-off build. //! -//! Events contain server-owned correlation IDs, sizes and timing, never paths -//! or I/O error strings. They are not metrics or performance acceptance. +//! Event fields contain server-owned correlation IDs, sizes and timing, never +//! object/disk paths or I/O error strings. They are not performance acceptance. //! Linux supplies thread CPU time; other targets report it absent. use std::sync::atomic::{AtomicU64, Ordering}; @@ -26,7 +26,7 @@ use tracing::{Dispatch, Level}; const LOG_COMPONENT_ECSTORE: &str = "ecstore"; const LOG_SUBSYSTEM_LOCAL_BATCH: &str = "local_batch_diagnostic"; const EVENT_LOCAL_BATCH_DIAGNOSTIC: &str = "local_batch_diagnostic"; -const TARGET: &str = "rustfs_ecstore::batch_read_diagnostics"; +const TARGET: &str = rustfs_config::LOCAL_BATCH_DIAGNOSTIC_LOG_TARGET; const ENV_SAMPLE_EVERY: &str = "RUSTFS_DIAGNOSTIC_LOCAL_BATCH_SAMPLE_EVERY"; const MAX_SAMPLE_EVERY: u64 = 65_536; static SAMPLE_EVERY: OnceLock = OnceLock::new(); @@ -514,6 +514,12 @@ mod tests { assert_eq!(recorder.counter_value("rustfs_io_get_object_local_batch_read_bytes_total", &[]), 3); { let events = state.events.lock().expect("normal events"); + let indices: Vec<_> = events + .iter() + .filter(|event| event.get("stage") == Some(&json!("shard_result"))) + .map(|event| event["shard_index"].as_u64().expect("shard index")) + .collect(); + assert_eq!(indices, [0, 1, 2, 3], "errors and empty reads retain their input index"); for (index, result) in [(0, "success"), (1, "error"), (2, "success"), (3, "error")] { assert!( events.iter().any(|event| { diff --git a/crates/obs/Cargo.toml b/crates/obs/Cargo.toml index 5819894b8..b48e22c64 100644 --- a/crates/obs/Cargo.toml +++ b/crates/obs/Cargo.toml @@ -27,6 +27,8 @@ documentation = "https://docs.rs/rustfs-obs/latest/rustfs_obs/" [features] default = [] +# Default-off JSON isolation for sampled local-batch diagnostic root events. +diagnostic-local-batch = [] hotpath = [ "hotpath/hotpath", "hotpath/tokio", diff --git a/crates/obs/src/telemetry/local.rs b/crates/obs/src/telemetry/local.rs index 49bf94f85..a0792031f 100644 --- a/crates/obs/src/telemetry/local.rs +++ b/crates/obs/src/telemetry/local.rs @@ -192,7 +192,7 @@ impl RequestIdJsonFormat { impl FormatEvent for RequestIdJsonFormat where S: Subscriber + for<'lookup> LookupSpan<'lookup>, - T: tracing_subscriber::fmt::time::FormatTime, + T: tracing_subscriber::fmt::time::FormatTime + Clone, { fn format_event( &self, @@ -200,6 +200,17 @@ where mut writer: Writer<'_>, event: &tracing::Event<'_>, ) -> fmt::Result { + #[cfg(feature = "diagnostic-local-batch")] + if event.is_root() && event.metadata().target() == rustfs_config::LOCAL_BATCH_DIAGNOSTIC_LOG_TARGET { + // Explicit root events must not acquire the formatter's current + // request span or an inferred request ID from that span. + return self + .inner + .clone() + .with_current_span(false) + .with_span_list(false) + .format_event(ctx, writer, event); + } let mut buffer = String::new(); self.inner.format_event(ctx, Writer::new(&mut buffer), event)?; @@ -813,6 +824,74 @@ mod tests { } } + #[cfg(feature = "diagnostic-local-batch")] + fn render_local_batch_diagnostic(request_id: Option<&str>) -> (Value, Value) { + let writer = SharedWriter::default(); + let layer = build_json_log_layer(writer.clone(), false, FmtSpan::FULL, true); + let subscriber = Registry::default().with(layer); + tracing::subscriber::with_default(subscriber, || { + let parent = tracing::info_span!("request", request_id = "parent-request", uri = "/must-not-copy-parent-path"); + let _entered = parent.enter(); + tracing::trace!(target: rustfs_config::LOCAL_BATCH_DIAGNOSTIC_LOG_TARGET, parent: None, + event = "local_batch_diagnostic", component = "ecstore", subsystem = "local_batch_diagnostic", + result = "success", request_id, + request_context = if request_id.is_some() { "present" } else { "missing" }, + "diagnostic root event"); + tracing::info!( + event = EVENT_LOCAL_LOGGING_STATE, + component = LOG_COMPONENT_OBS, + subsystem = LOG_SUBSYSTEM_LOCAL_LOGGING, + result = "test", + "ordinary context event" + ); + }); + let bytes = writer.inner.lock().expect("formatted diagnostic records"); + let records: Vec = std::str::from_utf8(&bytes) + .expect("UTF-8 JSON") + .lines() + .map(|line| serde_json::from_str(line).expect("JSON record")) + .collect(); + let diagnostic = records + .iter() + .find(|record| record["target"] == rustfs_config::LOCAL_BATCH_DIAGNOSTIC_LOG_TARGET) + .expect("diagnostic event") + .clone(); + let ordinary = records + .into_iter() + .find(|record| record["message"] == "ordinary context event") + .expect("ordinary event"); + (diagnostic, ordinary) + } + + #[cfg(feature = "diagnostic-local-batch")] + #[test] + fn local_batch_diagnostic_root_omits_parent_span_without_changing_ordinary_logs() { + let (diagnostic, ordinary) = render_local_batch_diagnostic(Some("server-owned")); + assert_eq!(diagnostic["request_id"], "server-owned"); + assert!(diagnostic.get("request-id").is_none()); + assert!(diagnostic.get("span").is_none()); + assert!(diagnostic.get("spans").is_none()); + assert!( + !serde_json::to_string(&diagnostic) + .expect("diagnostic JSON") + .contains("must-not-copy") + ); + assert_eq!(ordinary["span"]["uri"], "/must-not-copy-parent-path"); + assert_eq!(ordinary["request_id"], "parent-request"); + assert_eq!(ordinary["request-id"], "parent-request"); + } + + #[cfg(feature = "diagnostic-local-batch")] + #[test] + fn local_batch_diagnostic_missing_context_does_not_infer_parent_request_id() { + let (diagnostic, _) = render_local_batch_diagnostic(None); + assert_eq!(diagnostic["request_context"], "missing"); + assert!(diagnostic.get("request_id").is_none()); + assert!(diagnostic.get("request-id").is_none()); + assert!(diagnostic.get("span").is_none()); + assert!(diagnostic.get("spans").is_none()); + } + fn render_json_log(enable_ansi: bool) -> (String, Value) { let writer = SharedWriter::default(); let layer = build_json_log_layer(writer.clone(), enable_ansi, FmtSpan::NONE, false); diff --git a/docs/operations/local-batch-cpu-diagnostics.md b/docs/operations/local-batch-cpu-diagnostics.md index 4fcfcf858..0ce849f63 100644 --- a/docs/operations/local-batch-cpu-diagnostics.md +++ b/docs/operations/local-batch-cpu-diagnostics.md @@ -21,7 +21,7 @@ nothing until both the sample rate and the target filter are enabled: ```sh export RUSTFS_DIAGNOSTIC_LOCAL_BATCH_SAMPLE_EVERY=64 -export RUST_LOG=warn,rustfs_ecstore::batch_read_diagnostics=trace +export RUST_LOG=warn,rustfs_local_batch_diagnostic=trace ``` The rate is an integer in `1..=65536`. Missing, zero or invalid values disable @@ -32,11 +32,17 @@ While recording is enabled, each external S3 request retains a bounded shared ID, including requests whose batches are not selected. This is diagnostic work and must be accounted for before any observation-cost claim. -Events contain a process-local batch ID, the bounded server-owned canonical +Business fields contain a process-local batch ID, the bounded server-owned canonical request ID when propagated, worker PID/TID, shard index, sizes and timing. -They explicitly have no parent span, so URI-bearing HTTP span fields are not -copied. There are no variable-ID metric labels. Missing request context is -reported as `request_context=missing`, never treated as a successful join. +The RustFS feature also enables a matching JSON formatter policy: explicit +root events on this fixed diagnostic target omit current/ancestor spans and +parent-derived request ID aliases. No additional spans are allocated on the +sampled I/O path. There are no variable-ID metric labels. Missing request +context is reported as `request_context=missing`, never treated as a successful +join. +The standard logger may attach source-location and thread metadata. Its JSON +format flattens business fields; readers must distinguish these fields from +formatter metadata and reject inherited request spans. Queue wait is elapsed time across the async/blocking boundary and has no thread CPU delta. Execution and per-shard stages run synchronously on one @@ -46,6 +52,9 @@ zero-duration measurement. Clock and logging work perturb diagnostic samples. Elapsed and CPU are measured separately and must not be assumed equal or subtracted without accounting for measurement overhead. Stage durations can overlap and per-shard samples differ from per-batch samples. +When production batch-stage metrics are enabled, these timings also include +their recording work. Keep that state in the provenance; they are not bare +system-call durations. `shard_result` reports requested and actually read bytes, including partial reads on error. Unattempted stages are absent; a zero-length shard has no read @@ -53,8 +62,8 @@ sample. Captured correlation and dispatch stay with a blocking worker even after its caller is cancelled. Keep the diagnostic binary, source revision, feature, rate, filters and raw -events distinct from baseline or acceptance artifacts. Do not log paths, -credentials, request bodies or arbitrary client-supplied identifiers to add +events distinct from baseline or acceptance artifacts. Do not log credentials, +object/disk paths, request bodies or arbitrary client-supplied identifiers to add correlation. Request-level client association must use the server's canonical response ID. Diagnostic results locate hypotheses; they do not regrade failed A/A or ABBA runs. diff --git a/rustfs/Cargo.toml b/rustfs/Cargo.toml index a7c6319bf..8f1628030 100644 --- a/rustfs/Cargo.toml +++ b/rustfs/Cargo.toml @@ -80,7 +80,7 @@ dial9 = ["rustfs-obs/dial9"] # Diagnostic-only Rust GlobalAlloc live-byte accounting; adds per-allocation atomic updates. diagnostic-allocator-live = [] # Default-off sampled local batch wall/CPU and request correlation diagnostics. -diagnostic-local-batch = ["rustfs-ecstore/diagnostic-local-batch"] +diagnostic-local-batch = ["rustfs-ecstore/diagnostic-local-batch", "rustfs-obs/diagnostic-local-batch"] # Diagnostic-only S3-operation attribution for internode ReadVersion and list-walker metrics. diagnostic-read-version-caller = ["rustfs-ecstore/diagnostic-read-version-caller", "rustfs-io-metrics/diagnostic-read-version-caller", "rustfs-scanner/diagnostic-read-version-caller"] http3 = ["dep:h3", "dep:h3-quinn", "dep:quinn"]