Merge branch 'main' into houseme/rustix-e0-observability

Record local batch eligibility, blocking queue and execution time, shard
outcomes, and actual read bytes behind the existing GET-stage metrics gate.
Keep captured metric handles with the blocking worker after cancellation
and use scoped low-latency histogram buckets for the new duration series.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

crates/ecstore/src/disk/local.rs

@@ -11306,34 +11306,57 @@ impl LocalDisk {
 #[cfg(unix)]
 pub(crate) async fn batch_shard_pread(requests: Vec<(std::path::PathBuf, usize, usize)>) -> Vec<Result<Bytes>> {
     let n = requests.len();
+    let metrics = rustfs_io_metrics::LocalBatchReadMetrics::capture();
+    let submitted = metrics.as_ref().map(|_| std::time::Instant::now());
     tokio::task::spawn_blocking(move || {
+        use rustfs_io_metrics::LocalBatchReadStage;
         use std::os::unix::fs::FileExt;
 
+        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());
+            metrics.record_batch(n);
+        }
         let mut results = Vec::with_capacity(n);
         for (file_path, offset, length) in requests {
+            let mut bytes_read = 0;
             let r = (|| -> Result<Bytes> {
-                let meta = std::fs::metadata(&file_path).map_err(DiskError::from)?;
+                let meta = 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 = std::fs::File::open(&file_path).map_err(DiskError::from)?;
+                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 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;
+                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);
+                            }
+                            bytes_read += nbytes;
+                        }
+                        Ok(())
+                    })?;
                 }
                 Ok(Bytes::from(buf))
             })();
+            if let Some(metrics) = &metrics {
+                metrics.record_shard(r.is_ok(), length, bytes_read);
+            }
             results.push(r);
         }
+        if let (Some(metrics), Some(started)) = (&metrics, execution_started) {
+            metrics.record_stage(LocalBatchReadStage::Execution, results.iter().all(Result::is_ok), started.elapsed());
+        }
         results
     })
     .await
@@ -11343,6 +11366,20 @@ pub(crate) async fn batch_shard_pread(requests: Vec<(std::path::PathBuf, usize,
     })
 }
 
+#[cfg(unix)]
+fn measure_local_batch_stage<T>(
+    metrics: Option<&rustfs_io_metrics::LocalBatchReadMetrics>,
+    stage: rustfs_io_metrics::LocalBatchReadStage,
+    operation: impl FnOnce() -> Result<T>,
+) -> Result<T> {
+    let started = metrics.map(|_| std::time::Instant::now());
+    let result = operation();
+    if let (Some(metrics), Some(started)) = (metrics, started) {
+        metrics.record_stage(stage, result.is_ok(), started.elapsed());
+    }
+    result
+}
+
 #[async_trait::async_trait]
 impl DiskAPI for LocalDisk {
     async fn supports_guarded_delete_versions(&self) -> Result<()> {
@@ -30862,6 +30899,109 @@ mod test {
         );
     }
 
+    #[cfg(unix)]
+    #[test]
+    fn local_batch_read_metrics_cover_success_and_unattempted_failure_stages() {
+        use crate::test_metrics::CapturingRecorder;
+
+        let dir = tempfile::tempdir().expect("batch metrics directory");
+        let good = dir.path().join("good.bin");
+        std::fs::write(&good, b"good data").expect("write batch metrics payload");
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("batch metrics runtime");
+        let recorder = CapturingRecorder::default();
+        let previous = rustfs_io_metrics::get_stage_metrics_enabled();
+        rustfs_io_metrics::set_get_stage_metrics_enabled(true);
+        let results = metrics::with_local_recorder(&recorder, || {
+            runtime.block_on(batch_shard_pread(vec![
+                (good.clone(), 0, 9),
+                (dir.path().join("missing.bin"), 0, 4),
+                (good.clone(), 0, 1024),
+                (good, 0, 0),
+            ]))
+        });
+        rustfs_io_metrics::set_get_stage_metrics_enabled(previous);
+
+        assert_eq!(results[0].as_ref().expect("valid shard read").as_ref(), b"good data");
+        assert!(results[1].is_err(), "missing shard must remain an error");
+        assert!(matches!(results[2], Err(DiskError::FileCorrupt)), "range failure must preserve its error");
+        assert!(results[3].as_ref().expect("empty range succeeds").is_empty());
+        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", &[]), 9);
+        let stages = "rustfs_io_get_object_local_batch_stage_seconds";
+        for (stage, success, errors) in [("metadata", 3, 1), ("open", 2, 0), ("read", 1, 0), ("execution", 0, 1)] {
+            assert_eq!(
+                recorder
+                    .histogram_values(stages, &[("stage", stage), ("outcome", "success")])
+                    .len(),
+                success
+            );
+            assert_eq!(
+                recorder
+                    .histogram_values(stages, &[("stage", stage), ("outcome", "error")])
+                    .len(),
+                errors
+            );
+        }
+        assert_eq!(recorder.histogram_values(stages, &[("stage", "queue_wait")]).len(), 1);
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn local_batch_read_metrics_survive_caller_cancellation() {
+        use crate::test_metrics::CapturingRecorder;
+
+        let dir = tempfile::tempdir().expect("cancelled batch directory");
+        let path = dir.path().join("shard");
+        std::fs::write(&path, b"data").expect("write cancellation fixture");
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .max_blocking_threads(1)
+            .enable_all()
+            .build()
+            .expect("single blocking-worker runtime");
+        let recorder = CapturingRecorder::default();
+        let previous = rustfs_io_metrics::get_stage_metrics_enabled();
+        rustfs_io_metrics::set_get_stage_metrics_enabled(true);
+        metrics::with_local_recorder(&recorder, || {
+            runtime.block_on(async {
+                let (started_tx, started_rx) = tokio::sync::oneshot::channel();
+                let (release_tx, release_rx) = std::sync::mpsc::channel();
+                let blocker = tokio::task::spawn_blocking(move || {
+                    started_tx.send(()).expect("signal occupied blocking worker");
+                    release_rx.recv().expect("release occupied blocking worker");
+                });
+                started_rx.await.expect("blocking worker has started");
+                let mut batch = Box::pin(batch_shard_pread(vec![(path, 0, 4)]));
+                let pending = futures::FutureExt::now_or_never(&mut batch).is_none();
+                drop(batch);
+                rustfs_io_metrics::set_get_stage_metrics_enabled(false);
+                release_tx.send(()).expect("release worker after dropping caller");
+                blocker.await.expect("blocking worker finishes");
+                assert!(pending, "batch must have been submitted while the worker was occupied");
+                tokio::time::timeout(Duration::from_secs(5), async {
+                    while recorder
+                        .histogram_values("rustfs_io_get_object_local_batch_stage_seconds", &[("stage", "execution")])
+                        .is_empty()
+                    {
+                        tokio::task::yield_now().await;
+                    }
+                })
+                .await
+                .expect("detached batch must still record completion");
+            });
+        });
+        rustfs_io_metrics::set_get_stage_metrics_enabled(previous);
+        assert_eq!(
+            recorder.counter_value("rustfs_io_get_object_local_batch_shards_total", &[("outcome", "success")]),
+            1
+        );
+        assert_eq!(recorder.counter_value("rustfs_io_get_object_local_batch_read_bytes_total", &[]), 4);
+    }
+
     #[cfg(unix)]
     #[tokio::test]
     async fn test_batch_shard_pread_basic() {

crates/ecstore/src/set_disk/core/io_primitives.rs

@@ -1234,16 +1234,23 @@ async fn try_create_bitrot_readers_via_batch_pread(
     skip_verify_bitrot: bool,
 ) -> Option<BitrotReaderSetup> {
     use crate::disk::local::batch_shard_pread;
+    use rustfs_io_metrics::{LocalBatchReadDecision, record_local_batch_read_decision};
     use std::io::Cursor;
 
+    let prepare_started = rustfs_io_metrics::get_stage_metrics_enabled().then(std::time::Instant::now);
+    let record_decision = |decision| {
+        record_local_batch_read_decision(decision, prepare_started.map(|started| started.elapsed()));
+    };
     let (adj_off, adj_len) = adjust_shard_read_params(read_offset, read_length, shard_size, &checksum_algo);
     if adj_len > object_mmap_read_max_length() {
+        record_decision(LocalBatchReadDecision::ReadCapExceeded);
         return None;
     }
 
     let mut batch_items: Vec<(usize, std::path::PathBuf, usize, usize)> = Vec::new();
     for (idx, disk_op) in disks.iter().enumerate() {
         if files.get(idx).is_some_and(|fi| fi.data.is_some()) {
+            record_decision(LocalBatchReadDecision::InlineData);
             return None;
         }
         if let Some(disk) = disk_op.as_ref() {
@@ -1251,12 +1258,20 @@ async fn try_create_bitrot_readers_via_batch_pread(
             let path_str = format!("{object}/{data_dir}/part.{part_number}");
             match disk.get_object_path_for_io_if_local(bucket, &path_str) {
                 Some(Ok(p)) => batch_items.push((idx, p, adj_off, adj_len)),
-                _ => return None,
+                Some(Err(_)) => {
+                    record_decision(LocalBatchReadDecision::InvalidPath);
+                    return None;
+                }
+                None => {
+                    record_decision(LocalBatchReadDecision::NonLocalShard);
+                    return None;
+                }
             }
         }
     }
 
     if batch_items.is_empty() {
+        record_decision(LocalBatchReadDecision::NoLocalShards);
         return None;
     }
 
@@ -1265,6 +1280,7 @@ async fn try_create_bitrot_readers_via_batch_pread(
     for (disk_index, _, _, _) in &batch_items {
         disk_call_counters::record(object, disk_call_counters::KIND_DISK_SHARD_OPEN, *disk_index);
     }
+    record_decision(LocalBatchReadDecision::Selected);
     let batch_results = batch_shard_pread(requests).await;
 
     let mut setup = BitrotReaderSetup::new(disks.len());
@@ -1313,6 +1329,7 @@ async fn try_create_bitrot_readers_via_batch_pread(
     _checksum_algo: HashAlgorithm,
     _skip_verify_bitrot: bool,
 ) -> Option<BitrotReaderSetup> {
+    rustfs_io_metrics::record_local_batch_read_decision(rustfs_io_metrics::LocalBatchReadDecision::UnsupportedPlatform, None);
     None
 }
 
@@ -1476,11 +1493,19 @@ pub(in crate::set_disk) async fn create_bitrot_readers_until_quorum_with_prefere
 ) -> BitrotReaderSetup {
     let strategy = get_bitrot_reader_setup_strategy(mode, prefer_data_blocks_first);
 
-    if use_mmap_read
-        && !matches!(
-            crate::set_disk::get_object_read_policy(),
-            crate::set_disk::GetObjectReadPolicy::CopySource
-        )
+    let batch_read_enabled = if !use_mmap_read {
+        rustfs_io_metrics::record_local_batch_read_decision(rustfs_io_metrics::LocalBatchReadDecision::MmapDisabled, None);
+        false
+    } else if matches!(
+        crate::set_disk::get_object_read_policy(),
+        crate::set_disk::GetObjectReadPolicy::CopySource
+    ) {
+        rustfs_io_metrics::record_local_batch_read_decision(rustfs_io_metrics::LocalBatchReadDecision::CopySource, None);
+        false
+    } else {
+        true
+    };
+    if batch_read_enabled
         && let Some(mut setup) = try_create_bitrot_readers_via_batch_pread(
             files,
             disks,
@@ -7885,6 +7910,228 @@ mod tests {
         (dirs, disks)
     }
 
+    #[cfg(unix)]
+    #[test]
+    fn local_batch_read_decisions_follow_the_actual_reader_setup_entry() {
+        use crate::test_metrics::CapturingRecorder;
+
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("reader setup validation runtime");
+        let recorder = CapturingRecorder::default();
+        let previous = rustfs_io_metrics::get_stage_metrics_enabled();
+        rustfs_io_metrics::set_get_stage_metrics_enabled(true);
+        metrics::with_local_recorder(&recorder, || {
+            runtime.block_on(temp_env::async_with_vars(
+                [
+                    (rustfs_config::ENV_OBJECT_MMAP_READ_MAX_LENGTH, Some("4")),
+                    (rustfs_config::ENV_INTERNODE_PREWARM, Some("false")),
+                ],
+                async {
+                    let bucket = "batch-path-attribution";
+                    let object = "object";
+                    let (_dirs, disks) = call_counter_local_disks(bucket, 2).await;
+                    let data_dir = Uuid::new_v4();
+                    let mut files = vec![FileInfo::new(object, 1, 1); 2];
+                    for (file, disk) in files.iter_mut().zip(&disks) {
+                        file.data_dir = Some(data_dir);
+                        disk.as_ref()
+                            .expect("local disk")
+                            .write_all(bucket, &format!("{object}/{data_dir}/part.1"), Bytes::from_static(b"data"))
+                            .await
+                            .expect("write batch fixture");
+                    }
+                    assert_eq!(object_mmap_read_max_length(), 4, "cap must be controlled by this test");
+                    for length in [3, 4] {
+                        let setup = create_bitrot_readers_until_quorum_with_preference(
+                            &files,
+                            &disks,
+                            bucket,
+                            object,
+                            1,
+                            0,
+                            length,
+                            4,
+                            HashAlgorithm::None,
+                            false,
+                            true,
+                            1,
+                            1,
+                            BitrotReaderSetupMode::ReadQuorum,
+                            false,
+                            None,
+                            None,
+                        )
+                        .await;
+                        assert_eq!(setup.readers.iter().flatten().count(), 2);
+                    }
+                    assert!(
+                        try_create_bitrot_readers_via_batch_pread(
+                            &files,
+                            &disks,
+                            bucket,
+                            object,
+                            1,
+                            0,
+                            5,
+                            4,
+                            HashAlgorithm::None,
+                            false,
+                        )
+                        .await
+                        .is_none(),
+                        "a read above the adjusted-length cap must fall back"
+                    );
+                    assert!(
+                        try_create_bitrot_readers_via_batch_pread(
+                            &files,
+                            &disks,
+                            bucket,
+                            object,
+                            1,
+                            0,
+                            4,
+                            4,
+                            HashAlgorithm::HighwayHash256S,
+                            false,
+                        )
+                        .await
+                        .is_none(),
+                        "checksum overhead must be included in the read cap"
+                    );
+                    let mut inline_files = files.clone();
+                    inline_files[0].data = Some(Bytes::from_static(b"data"));
+                    assert!(
+                        try_create_bitrot_readers_via_batch_pread(
+                            &inline_files,
+                            &disks,
+                            bucket,
+                            object,
+                            1,
+                            0,
+                            4,
+                            4,
+                            HashAlgorithm::None,
+                            false,
+                        )
+                        .await
+                        .is_none()
+                    );
+                    let mut endpoint = Endpoint::try_from("http://127.0.0.1:1/disk").expect("remote placeholder endpoint");
+                    endpoint.is_local = false;
+                    let remote = new_disk(
+                        &endpoint,
+                        &DiskOption {
+                            cleanup: false,
+                            health_check: false,
+                        },
+                    )
+                    .await
+                    .expect("remote placeholder must initialize without a connection");
+                    assert!(
+                        try_create_bitrot_readers_via_batch_pread(
+                            &files,
+                            &[Some(remote), None],
+                            bucket,
+                            object,
+                            1,
+                            0,
+                            4,
+                            4,
+                            HashAlgorithm::None,
+                            false,
+                        )
+                        .await
+                        .is_none(),
+                        "any remote shard must keep the setup out of the local batch path"
+                    );
+                    assert!(
+                        try_create_bitrot_readers_via_batch_pread(
+                            &files,
+                            &disks,
+                            bucket,
+                            "../outside",
+                            1,
+                            0,
+                            4,
+                            4,
+                            HashAlgorithm::None,
+                            false,
+                        )
+                        .await
+                        .is_none()
+                    );
+                    let missing = vec![None, None];
+                    assert!(
+                        try_create_bitrot_readers_via_batch_pread(
+                            &files,
+                            &missing,
+                            bucket,
+                            object,
+                            1,
+                            0,
+                            4,
+                            4,
+                            HashAlgorithm::None,
+                            false,
+                        )
+                        .await
+                        .is_none()
+                    );
+                    for (use_mmap, policy) in [
+                        (false, crate::set_disk::GetObjectReadPolicy::Default),
+                        (true, crate::set_disk::GetObjectReadPolicy::CopySource),
+                    ] {
+                        crate::set_disk::with_get_object_read_policy(policy, async {
+                            create_bitrot_readers_until_quorum_with_preference(
+                                &files,
+                                &missing,
+                                bucket,
+                                object,
+                                1,
+                                0,
+                                4,
+                                4,
+                                HashAlgorithm::None,
+                                false,
+                                use_mmap,
+                                1,
+                                1,
+                                BitrotReaderSetupMode::ReadQuorum,
+                                false,
+                                None,
+                                None,
+                            )
+                            .await;
+                        })
+                        .await;
+                    }
+                },
+            ));
+        });
+        rustfs_io_metrics::set_get_stage_metrics_enabled(previous);
+        let metric = "rustfs_io_get_object_local_batch_decisions_total";
+        for (decision, expected) in [
+            ("selected", 2),
+            ("read_cap_exceeded", 2),
+            ("inline_data", 1),
+            ("invalid_path", 1),
+            ("non_local_shard", 1),
+            ("no_local_shards", 1),
+            ("mmap_disabled", 1),
+            ("copy_source", 1),
+        ] {
+            assert_eq!(recorder.counter_value(metric, &[("decision", decision)]), expected, "{decision}");
+        }
+        assert_eq!(
+            recorder.counter_value("rustfs_io_get_object_local_batch_shards_total", &[("outcome", "success")]),
+            4
+        );
+        assert_eq!(recorder.counter_value("rustfs_io_get_object_local_batch_read_bytes_total", &[]), 14);
+        assert_eq!(recorder.histogram_sample_count("rustfs_io_get_object_local_batch_prepare_seconds"), 8);
+    }
+
     async fn reopen_local_disk(dir: &TempDir) -> DiskStore {
         let endpoint =
             Endpoint::try_from(dir.path().to_str().expect("tempdir path should be utf8")).expect("endpoint should parse");

crates/io-metrics/benches/metrics_pipeline.rs

@@ -1,10 +1,67 @@
 use criterion::{Criterion, criterion_group, criterion_main};
+use metrics_util::debugging::DebuggingRecorder;
+use rustfs_io_metrics::{
+    LocalBatchReadDecision, LocalBatchReadMetrics, LocalBatchReadStage, record_local_batch_read_decision,
+    set_get_stage_metrics_enabled,
+};
 use rustfs_io_metrics::{MetricsCollector, PerformanceMetrics, record_get_object_timeout_operation_start};
 use rustfs_io_metrics::{record_s3_op, s3_http_metrics::S3HttpRequestGuard};
 use rustfs_s3_ops::S3Operation;
 use std::hint::black_box;
 use std::sync::Arc;
-use std::time::Duration;
+use std::time::{Duration, Instant};
+
+fn local_batch_attribution_sample() {
+    let prepare = rustfs_io_metrics::get_stage_metrics_enabled().then(Instant::now);
+    record_local_batch_read_decision(LocalBatchReadDecision::Selected, prepare.map(|started| started.elapsed()));
+    if let Some(metrics) = LocalBatchReadMetrics::capture() {
+        metrics.record_batch(4);
+        let execution = Instant::now();
+        metrics.record_stage(LocalBatchReadStage::QueueWait, true, execution.elapsed());
+        for _ in 0..4 {
+            for stage in [
+                LocalBatchReadStage::Metadata,
+                LocalBatchReadStage::Open,
+                LocalBatchReadStage::Read,
+            ] {
+                let started = Instant::now();
+                metrics.record_stage(stage, true, started.elapsed());
+            }
+            metrics.record_shard(true, 32 * 1024, 32 * 1024);
+        }
+        metrics.record_stage(LocalBatchReadStage::Execution, true, execution.elapsed());
+    }
+}
+
+fn bench_local_batch_attribution(c: &mut Criterion) {
+    let mut group = c.benchmark_group("local_batch_attribution_4_shards_debug_recorder");
+    for (name, enabled) in [("disabled", false), ("enabled", true)] {
+        set_get_stage_metrics_enabled(enabled);
+        group.bench_function(name, |b| {
+            b.iter_custom(|iterations| {
+                let mut remaining = iterations;
+                let mut elapsed = Duration::ZERO;
+                while remaining > 0 {
+                    // Bound captured histogram storage independently of Criterion's
+                    // iteration count. Recorder setup/teardown is not measured.
+                    let chunk = remaining.min(128);
+                    let recorder = DebuggingRecorder::new();
+                    metrics::with_local_recorder(&recorder, || {
+                        let started = Instant::now();
+                        for _ in 0..chunk {
+                            local_batch_attribution_sample();
+                        }
+                        elapsed += started.elapsed();
+                    });
+                    remaining -= chunk;
+                }
+                elapsed
+            });
+        });
+    }
+    group.finish();
+    set_get_stage_metrics_enabled(false);
+}
 
 fn bench_record_get_object_timeout_operation_start(c: &mut Criterion) {
     c.bench_function("record_get_object_timeout_operation_start", |b| {
@@ -54,6 +111,7 @@ criterion_group!(
     bench_record_get_object_timeout_operation_start,
     bench_s3_http_outcomes,
     bench_update_concurrent_requests,
-    bench_metrics_collector_record_io_operation
+    bench_metrics_collector_record_io_operation,
+    bench_local_batch_attribution
 );
 criterion_main!(benches);

crates/io-metrics/src/lib.rs

@@ -49,6 +49,11 @@
 #[macro_use]
 extern crate metrics;
 
+mod local_batch_read;
+pub use local_batch_read::{
+    LocalBatchReadDecision, LocalBatchReadMetrics, LocalBatchReadStage, record_local_batch_read_decision,
+};
+
 use std::collections::HashMap;
 use std::sync::{
     LazyLock, Mutex,

crates/io-metrics/src/local_batch_read.rs

@@ -0,0 +1,210 @@
+// 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.
+
+//! Opt-in attribution for the buffered local shard batch path.
+
+use crate::get_stage_metrics_enabled;
+use metrics::{Counter, Histogram};
+use std::time::Duration;
+
+/// One decision per reader-setup attempt, rather than per S3 request.
+#[derive(Clone, Copy, Debug)]
+pub enum LocalBatchReadDecision {
+    Selected,
+    MmapDisabled,
+    CopySource,
+    ReadCapExceeded,
+    InlineData,
+    NonLocalShard,
+    InvalidPath,
+    NoLocalShards,
+    UnsupportedPlatform,
+}
+
+impl LocalBatchReadDecision {
+    const fn as_str(self) -> &'static str {
+        match self {
+            Self::Selected => "selected",
+            Self::MmapDisabled => "mmap_disabled",
+            Self::CopySource => "copy_source",
+            Self::ReadCapExceeded => "read_cap_exceeded",
+            Self::InlineData => "inline_data",
+            Self::NonLocalShard => "non_local_shard",
+            Self::InvalidPath => "invalid_path",
+            Self::NoLocalShards => "no_local_shards",
+            Self::UnsupportedPlatform => "unsupported_platform",
+        }
+    }
+}
+
+/// Record eligibility, including path preparation only when it actually ran.
+pub fn record_local_batch_read_decision(decision: LocalBatchReadDecision, preparation: Option<Duration>) {
+    if !get_stage_metrics_enabled() {
+        return;
+    }
+    let decision = decision.as_str();
+    counter!("rustfs_io_get_object_local_batch_decisions_total", "decision" => decision).increment(1);
+    if let Some(duration) = preparation {
+        histogram!("rustfs_io_get_object_local_batch_prepare_seconds", "decision" => decision).record(duration.as_secs_f64());
+    }
+}
+
+/// Stages of the existing buffered positioned-read implementation.
+#[derive(Clone, Copy)]
+pub enum LocalBatchReadStage {
+    QueueWait,
+    Execution,
+    Metadata,
+    Open,
+    Read,
+}
+
+impl LocalBatchReadStage {
+    const fn index(self) -> usize {
+        match self {
+            Self::QueueWait => 0,
+            Self::Execution => 1,
+            Self::Metadata => 2,
+            Self::Open => 3,
+            Self::Read => 4,
+        }
+    }
+}
+
+/// Handles captured before dispatch, kept alive by the actual blocking worker.
+///
+/// Capturing the recorder once avoids per-I/O registration and also preserves
+/// completion samples when the caller stops waiting. No timers or handles are
+/// created with GET-stage metrics disabled. The snapshot applies to the whole
+/// submitted batch, including work that finishes after the switch is disabled.
+pub struct LocalBatchReadMetrics {
+    stages: [[Histogram; 2]; 5],
+    shards: [Counter; 2],
+    requested_bytes: [Counter; 2],
+    read_bytes: [Counter; 2],
+    batch_size: Histogram,
+}
+
+fn outcome(success: bool) -> &'static str {
+    if success { "success" } else { "error" }
+}
+
+impl LocalBatchReadMetrics {
+    /// Resolve bounded-label handles only for an explicitly instrumented batch.
+    /// Boxing the enabled state keeps the blocking task small when disabled.
+    pub fn capture() -> Option<Box<Self>> {
+        if !get_stage_metrics_enabled() {
+            return None;
+        }
+        let stages = ["queue_wait", "execution", "metadata", "open", "read"].map(|stage| {
+            [false, true].map(|success| {
+                histogram!("rustfs_io_get_object_local_batch_stage_seconds", "stage" => stage, "outcome" => outcome(success))
+            })
+        });
+        Some(Box::new(Self {
+            stages,
+            shards: [false, true]
+                .map(|success| counter!("rustfs_io_get_object_local_batch_shards_total", "outcome" => outcome(success))),
+            requested_bytes: [false, true]
+                .map(|success| counter!("rustfs_io_get_object_local_batch_requested_bytes_total", "outcome" => outcome(success))),
+            read_bytes: [false, true]
+                .map(|success| counter!("rustfs_io_get_object_local_batch_read_bytes_total", "outcome" => outcome(success))),
+            batch_size: histogram!("rustfs_io_get_object_local_batch_shards"),
+        }))
+    }
+
+    /// Count shards actually submitted to this blocking batch.
+    pub fn record_batch(&self, shards: usize) {
+        self.batch_size.record(crate::shard_read_fanout_to_f64(shards));
+    }
+
+    /// Record an executed stage; unattempted stages receive no sample.
+    ///
+    /// Queue wait succeeds when the worker starts. Execution covers the whole
+    /// batch, while metadata/open/read describe individual I/O operations.
+    /// Read timing excludes buffer allocation; execution timing includes it.
+    pub fn record_stage(&self, stage: LocalBatchReadStage, success: bool, duration: Duration) {
+        self.stages[stage.index()][usize::from(success)].record(duration.as_secs_f64());
+    }
+
+    /// Separate completed shard outcomes and bytes actually returned by read_at.
+    pub fn record_shard(&self, success: bool, requested: usize, read: usize) {
+        let index = usize::from(success);
+        self.shards[index].increment(1);
+        self.requested_bytes[index].increment(u64::try_from(requested).unwrap_or(u64::MAX));
+        self.read_bytes[index].increment(u64::try_from(read).unwrap_or(u64::MAX));
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::set_get_stage_metrics_enabled;
+    use crate::tests::{METRICS_FLAG_LOCK, counter_total, histogram_samples};
+    use metrics_util::debugging::DebuggingRecorder;
+    use std::time::Duration;
+
+    #[test]
+    fn local_batch_read_metrics_are_disabled_without_stage_opt_in() {
+        let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|error| error.into_inner());
+        let recorder = DebuggingRecorder::new();
+        let snapshotter = recorder.snapshotter();
+        metrics::with_local_recorder(&recorder, || {
+            set_get_stage_metrics_enabled(false);
+            assert!(LocalBatchReadMetrics::capture().is_none());
+            record_local_batch_read_decision(LocalBatchReadDecision::Selected, Some(Duration::from_millis(1)));
+        });
+        assert!(snapshotter.snapshot().into_vec().is_empty());
+    }
+
+    #[test]
+    fn local_batch_read_handles_record_worker_costs_and_actual_bytes() {
+        let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|error| error.into_inner());
+        let recorder = DebuggingRecorder::new();
+        let snapshotter = recorder.snapshotter();
+        let observation = metrics::with_local_recorder(&recorder, || {
+            set_get_stage_metrics_enabled(true);
+            record_local_batch_read_decision(LocalBatchReadDecision::Selected, Some(Duration::from_millis(2)));
+            LocalBatchReadMetrics::capture().expect("enabled batch metrics must capture recorder handles")
+        });
+        set_get_stage_metrics_enabled(false);
+        std::thread::spawn(move || {
+            observation.record_stage(LocalBatchReadStage::QueueWait, true, Duration::from_millis(3));
+            observation.record_stage(LocalBatchReadStage::Metadata, false, Duration::from_millis(4));
+            observation.record_shard(false, 1024, 0);
+            observation.record_shard(true, 16, 16);
+            observation.record_batch(2);
+        })
+        .join()
+        .expect("worker should record through the captured handles");
+        set_get_stage_metrics_enabled(false);
+
+        let rows = snapshotter.snapshot().into_vec();
+        assert_eq!(counter_total(&rows, "rustfs_io_get_object_local_batch_decisions_total"), Some(1));
+        assert_eq!(counter_total(&rows, "rustfs_io_get_object_local_batch_shards_total"), Some(2));
+        assert_eq!(counter_total(&rows, "rustfs_io_get_object_local_batch_requested_bytes_total"), Some(1040));
+        assert_eq!(counter_total(&rows, "rustfs_io_get_object_local_batch_read_bytes_total"), Some(16));
+        assert_eq!(histogram_samples(&rows, "rustfs_io_get_object_local_batch_prepare_seconds"), vec![0.002]);
+        assert_eq!(
+            histogram_samples(&rows, "rustfs_io_get_object_local_batch_stage_seconds"),
+            vec![0.003, 0.004]
+        );
+        assert_eq!(histogram_samples(&rows, "rustfs_io_get_object_local_batch_shards"), vec![2.0]);
+        for (key, _, _, _) in rows {
+            for label in key.key().labels() {
+                assert!(matches!(label.key(), "decision" | "stage" | "outcome"));
+            }
+        }
+    }
+}

crates/obs/src/telemetry/otel.rs

@@ -83,6 +83,16 @@ const GET_OBJECT_DURATION_HISTOGRAM_BUCKETS: &[f64] = &[
     0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0,
 ];
 
+const LOCAL_BATCH_DURATION_HISTOGRAM_METRICS: &[&str] = &[
+    "rustfs_io_get_object_local_batch_prepare_seconds",
+    "rustfs_io_get_object_local_batch_stage_seconds",
+];
+
+const LOCAL_BATCH_DURATION_HISTOGRAM_BUCKETS: &[f64] = &[
+    0.0000001, 0.00000025, 0.0000005, 0.000001, 0.0000025, 0.000005, 0.00001, 0.000025, 0.00005, 0.0001, 0.00025, 0.0005, 0.001,
+    0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0,
+];
+
 const DEFAULT_OTLP_HTTP_TIMEOUT: Duration = Duration::from_secs(10);
 const OTEL_EXPORTER_OTLP_TIMEOUT: &str = "OTEL_EXPORTER_OTLP_TIMEOUT";
 const OTEL_EXPORTER_OTLP_TRACES_TIMEOUT: &str = "OTEL_EXPORTER_OTLP_TRACES_TIMEOUT";
@@ -481,21 +491,25 @@ fn build_meter_provider(
 }
 
 fn get_object_duration_histogram_view(instrument: &Instrument) -> Option<Stream> {
-    if !is_get_object_duration_histogram_metric(instrument.name()) {
-        return None;
-    }
+    let boundaries = get_object_duration_histogram_buckets(instrument.name())?;
 
     Stream::builder()
         .with_aggregation(Aggregation::ExplicitBucketHistogram {
-            boundaries: GET_OBJECT_DURATION_HISTOGRAM_BUCKETS.to_vec(),
+            boundaries: boundaries.to_vec(),
             record_min_max: true,
         })
         .build()
         .ok()
 }
 
-fn is_get_object_duration_histogram_metric(name: &str) -> bool {
-    GET_OBJECT_DURATION_HISTOGRAM_METRICS.contains(&name)
+fn get_object_duration_histogram_buckets(name: &str) -> Option<&'static [f64]> {
+    if LOCAL_BATCH_DURATION_HISTOGRAM_METRICS.contains(&name) {
+        Some(LOCAL_BATCH_DURATION_HISTOGRAM_BUCKETS)
+    } else if GET_OBJECT_DURATION_HISTOGRAM_METRICS.contains(&name) {
+        Some(GET_OBJECT_DURATION_HISTOGRAM_BUCKETS)
+    } else {
+        None
+    }
 }
 
 /// Build an optional [`SdkLoggerProvider`] for the given log endpoint.
@@ -1098,10 +1112,22 @@ mod tests {
 
     #[test]
     fn test_get_object_duration_histogram_metric_match_is_scoped() {
-        assert!(is_get_object_duration_histogram_metric("rustfs_io_get_object_stage_duration_seconds"));
-        assert!(is_get_object_duration_histogram_metric("rustfs_io_get_object_request_duration_seconds"));
-        assert!(!is_get_object_duration_histogram_metric("rustfs_io_put_object_request_duration_seconds"));
-        assert!(!is_get_object_duration_histogram_metric("rustfs_io_get_object_response_size_bytes"));
+        for name in [
+            "rustfs_io_get_object_stage_duration_seconds",
+            "rustfs_io_get_object_request_duration_seconds",
+        ] {
+            assert_eq!(get_object_duration_histogram_buckets(name), Some(GET_OBJECT_DURATION_HISTOGRAM_BUCKETS));
+        }
+        for name in LOCAL_BATCH_DURATION_HISTOGRAM_METRICS {
+            assert_eq!(get_object_duration_histogram_buckets(name), Some(LOCAL_BATCH_DURATION_HISTOGRAM_BUCKETS));
+        }
+        for name in [
+            "rustfs_io_put_object_request_duration_seconds",
+            "rustfs_io_get_object_response_size_bytes",
+            "rustfs_io_get_object_local_batch_shards",
+        ] {
+            assert!(get_object_duration_histogram_buckets(name).is_none());
+        }
     }
 
     #[test]
@@ -1111,6 +1137,17 @@ mod tests {
         assert_eq!(GET_OBJECT_DURATION_HISTOGRAM_BUCKETS.last(), Some(&10.0));
     }
 
+    #[test]
+    fn local_batch_duration_buckets_resolve_sub_microsecond_operations() {
+        assert!(
+            LOCAL_BATCH_DURATION_HISTOGRAM_BUCKETS
+                .windows(2)
+                .all(|pair| pair[0] < pair[1])
+        );
+        assert_eq!(LOCAL_BATCH_DURATION_HISTOGRAM_BUCKETS.first(), Some(&0.0000001));
+        assert_eq!(LOCAL_BATCH_DURATION_HISTOGRAM_BUCKETS.last(), Some(&10.0));
+    }
+
     #[cfg(all(
         feature = "pyroscope",
         any(target_os = "macos", all(target_os = "linux", target_env = "gnu", target_arch = "x86_64"))

docs/operations/hotpath-warp-ab-runbook.md

@@ -143,6 +143,39 @@ Rules that override the table: a candidate result is actionable only when the `A
 
 ### CPU and memory evidence
 
+#### Local shard batch attribution
+
+The existing GET-stage metric gate also controls local batch attribution. This
+path performs buffered positioned reads, even if O_DIRECT or io_uring is enabled
+for another backend. Keep the gate, exporter, and build features identical
+between compared binaries; calibrate instrumentation overhead separately.
+
+| Metric suffix after `rustfs_io_get_object_local_batch_` | Meaning |
+| --- | --- |
+| `decisions_total` | Reader-setup decisions, labelled by a fixed decision enum; this is not an S3 request counter. |
+| `prepare_seconds` | Eligibility and path preparation, including pre-worker path checks; absent when preparation was skipped. |
+| `stage_seconds` | Queue wait, whole-worker execution, and per-shard metadata/open/read durations, labelled by stage and operation outcome. |
+| `shards` / `shards_total` | Batch size and completed shard outcomes. |
+| `requested_bytes_total` / `read_bytes_total` | Requested bytes and bytes actually returned by positioned reads, split by shard outcome. |
+
+Worker handles are captured before dispatch and remain owned by the worker, so
+completion samples can arrive after the caller stops waiting. A stage that did
+not execute contributes no sample. Queue wait ends when the worker starts;
+execution includes allocation and instrumentation, while read timing excludes
+buffer allocation. These nested and per-shard durations cannot be added to
+request latency, and stage p99 values cannot be summed. Match request-boundary
+samples and successful-request counters to the same workload window when
+normalizing costs. Keep failed requests and partial-shard failures visible.
+Inline GET and HEAD can bypass this reader-setup entry entirely; absence of a
+decision sample does not imply a failed request. The two local duration metrics
+use dedicated 0.1-microsecond-to-10-second histogram boundaries in the OTLP view;
+request-duration histogram boundaries are separate.
+
+For a bounded, four-shard instrumentation microbenchmark, run
+`cargo bench -p rustfs-io-metrics --bench metrics_pipeline -- local_batch_attribution_4_shards_debug_recorder`.
+It performs no disk I/O and uses a debugging recorder, so its result calibrates
+the sampling code rather than production exporter overhead or S3 latency.
+
 Warp output says whether throughput or latency changed; host telemetry says why. Collect it for the whole run and stop the collectors after the script exits.
 
 | Tool | Command | Answers |