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/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/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/diagnostics/batch_read_diagnostics.rs b/crates/ecstore/src/diagnostics/batch_read_diagnostics.rs new file mode 100644 index 000000000..64a170624 --- /dev/null +++ b/crates/ecstore/src/diagnostics/batch_read_diagnostics.rs @@ -0,0 +1,568 @@ +// 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. +//! +//! 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}; +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_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(); +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() +} + +/// 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, + 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.is_multiple_of(rate) { + 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()); + } + + #[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(); + 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", + "diagnostics::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 recorder = crate::test_metrics::CapturingRecorder::default(); + rustfs_io_metrics::set_get_stage_metrics_enabled(true); + let subscriber = tracing_subscriber::registry().with(Recorder(Arc::clone(&state))); + metrics::with_local_recorder(&recorder, || { + 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 counter = "rustfs_io_get_object_local_batch_shards_total"; + assert_eq!(recorder.counter_value(counter, &[("outcome", "success")]), 2); + assert_eq!(recorder.counter_value(counter, &[("outcome", "error")]), 2); + 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| { + event.get("stage") == Some(&json!("shard_result")) + && event.get("shard_index") == Some(&json!(index)) + && event.get("result") == Some(&json!(result)) + }), + "missing diagnostic result for shard {index}" + ); + } + 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/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 9fc17fe5f..a786c35dc 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -11328,10 +11328,39 @@ pub(crate) async fn batch_shard_pread(requests: Vec<(std::path::PathBuf, usize, let n = requests.len(); let metrics = rustfs_io_metrics::LocalBatchReadMetrics::capture(); let submitted = metrics.as_ref().map(|_| std::time::Instant::now()); + #[cfg(feature = "diagnostic-local-batch")] + let diagnostic = crate::diagnostics::batch_read_diagnostics::Sample::capture(); tokio::task::spawn_blocking(move || { use rustfs_io_metrics::LocalBatchReadStage; 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::diagnostics::batch_read_diagnostics::measure( + diagnostic.as_ref(), + crate::diagnostics::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 execution_started = metrics.as_ref().map(|_| std::time::Instant::now()); if let (Some(metrics), Some(submitted)) = (&metrics, submitted) { metrics.record_stage(LocalBatchReadStage::QueueWait, true, submitted.elapsed()); @@ -11339,41 +11368,64 @@ pub(crate) async fn batch_shard_pread(requests: Vec<(std::path::PathBuf, usize, } let mut results = Vec::with_capacity(n); for (file_path, offset, length) in requests { - let mut bytes_read = 0; + #[cfg(feature = "diagnostic-local-batch")] + let _index = results.len(); + let mut total = 0usize; let r = (|| -> Result { - let meta = measure_local_batch_stage(metrics.as_deref(), LocalBatchReadStage::Metadata, || { - std::fs::metadata(&file_path).map_err(DiskError::from) - })?; + let meta = diagnostic_stage!(Metadata, _index, || measure_local_batch_stage( + metrics.as_deref(), + LocalBatchReadStage::Metadata, + || 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 = measure_local_batch_stage(metrics.as_deref(), LocalBatchReadStage::Open, || { - std::fs::File::open(&file_path).map_err(DiskError::from) - })?; - let mut buf = vec![0u8; length]; + let file = diagnostic_stage!(Open, _index, || measure_local_batch_stage( + metrics.as_deref(), + LocalBatchReadStage::Open, + || 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 { - measure_local_batch_stage(metrics.as_deref(), LocalBatchReadStage::Read, || { - while bytes_read < length { - let nbytes = file - .read_at(&mut buf[bytes_read..], u64::try_from(offset + bytes_read).unwrap_or(u64::MAX)) - .map_err(DiskError::from)?; - if nbytes == 0 { - return Err(DiskError::FileCorrupt); + diagnostic_stage!(Read, _index, || measure_local_batch_stage( + metrics.as_deref(), + LocalBatchReadStage::Read, + || { + 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; } - bytes_read += nbytes; + Ok::<_, DiskError>(()) } - Ok(()) - })?; + ))?; } Ok(Bytes::from(buf)) })(); + #[cfg(feature = "diagnostic-local-batch")] + if let Some(sample) = &diagnostic { + sample.shard_result(_index, length, total, r.is_ok()); + } if let Some(metrics) = &metrics { - metrics.record_shard(r.is_ok(), length, bytes_read); + metrics.record_shard(r.is_ok(), length, total); } results.push(r); } + #[cfg(feature = "diagnostic-local-batch")] + if let (Some(sample), Some(started)) = (&diagnostic, execution) { + sample.finish( + crate::diagnostics::batch_read_diagnostics::Stage::Execution, + None, + started, + results.iter().all(Result::is_ok), + ); + } if let (Some(metrics), Some(started)) = (&metrics, execution_started) { metrics.record_stage(LocalBatchReadStage::Execution, results.iter().all(Result::is_ok), started.elapsed()); } diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 4442d22ac..47b548c23 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) } 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/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/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/docs/operations/local-batch-cpu-diagnostics.md b/docs/operations/local-batch-cpu-diagnostics.md new file mode 100644 index 000000000..0ce849f63 --- /dev/null +++ b/docs/operations/local-batch-cpu-diagnostics.md @@ -0,0 +1,69 @@ +# 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_local_batch_diagnostic=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. + +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. +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 +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. +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 +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 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 7338ed8b3..8f1628030 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", "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"] diff --git a/rustfs/src/server/layer.rs b/rustfs/src/server/layer.rs index b58098dda..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; @@ -73,6 +75,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 +291,12 @@ where } else { req.headers().get(REQUEST_ID_HEADER).cloned() }; + #[cfg(feature = "diagnostic-local-batch")] + let diagnostic_request = is_s3 + .then(|| batch_diagnostic::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 +312,12 @@ 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 = 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)), None => self.inner.call(req), @@ -309,6 +327,7 @@ where } ExternalRequestContextFuture { inner, + diagnostic_request, request_id, is_s3, metrics, @@ -358,6 +377,7 @@ pin_project! { pub struct ExternalRequestContextFuture { #[pin] inner: F, + diagnostic_request: BatchDiagnosticRequest, request_id: Option, is_s3: bool, metrics: Option, @@ -374,6 +394,12 @@ where fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { let this = self.project(); + #[cfg(feature = "diagnostic-local-batch")] + 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), + }); + #[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 +2634,83 @@ 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!(batch_diagnostic::current_request_id().as_deref(), Some(expected.as_str())); + async move { + assert_eq!(batch_diagnostic::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!(batch_diagnostic::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!(batch_diagnostic::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!(batch_diagnostic::current_request_id().is_none()); + } #[tokio::test] async fn foreground_feedback_unobserved_body_is_a_direct_passthrough() { struct EndProbe(Full); 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 3d315836e..e101b81cc 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/diagnostics/batch_read_diagnostics.rs" "rustfs/src/main.rs" "rustfs/src/startup_entrypoint.rs" "rustfs/src/init.rs"