From c0ddc14bb8df68045a4129ed89a646b15d4d4db8 Mon Sep 17 00:00:00 2001 From: houseme Date: Fri, 26 Jun 2026 23:50:37 +0800 Subject: [PATCH] feat(get): observe shard locality cost (#3922) * feat(get): observe shard locality cost * fix(storage): avoid request counter underflow panic * fix(storage): import put guard in concurrency tests --- crates/ecstore/src/erasure_coding/bitrot.rs | 10 + crates/ecstore/src/erasure_coding/decode.rs | 189 ++++++++++++++++-- crates/ecstore/src/get_diagnostics.rs | 12 ++ crates/ecstore/src/set_disk/read.rs | 20 +- crates/ecstore/src/set_disk/shard_source.rs | 90 ++++++++- crates/io-metrics/src/lib.rs | 117 +++++++++++ rustfs/src/storage/concurrency/manager.rs | 3 +- .../src/storage/concurrency/request_guard.rs | 6 +- 8 files changed, 414 insertions(+), 33 deletions(-) diff --git a/crates/ecstore/src/erasure_coding/bitrot.rs b/crates/ecstore/src/erasure_coding/bitrot.rs index 01e2f4c3c..874ef052f 100644 --- a/crates/ecstore/src/erasure_coding/bitrot.rs +++ b/crates/ecstore/src/erasure_coding/bitrot.rs @@ -16,6 +16,7 @@ use bytes::Bytes; use pin_project_lite::pin_project; use rustfs_utils::HashAlgorithm; use std::io::IoSlice; +use std::time::Duration; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; use tracing::error; use uuid::Uuid; @@ -30,6 +31,7 @@ pin_project! { buf: Vec, hash_buf: Vec, skip_verify: bool, + last_verify_duration: Duration, id: Uuid, } } @@ -48,13 +50,19 @@ where buf: Vec::new(), hash_buf: vec![0u8; hash_size], skip_verify, + last_verify_duration: Duration::ZERO, id: Uuid::new_v4(), } } + pub(crate) fn last_verify_duration(&self) -> Duration { + self.last_verify_duration + } + /// Read a single (hash+data) block, verify hash, and return the number of bytes read into `out`. /// Returns an error if hash verification fails or data exceeds shard_size. pub async fn read(&mut self, out: &mut [u8]) -> std::io::Result { + self.last_verify_duration = Duration::ZERO; if out.len() > self.shard_size { return Err(std::io::Error::new( std::io::ErrorKind::InvalidInput, @@ -86,7 +94,9 @@ where } if hash_size > 0 && !self.skip_verify { + let verify_start = std::time::Instant::now(); let actual_hash = self.hash_algo.hash_encode(&out[..data_len]); + self.last_verify_duration = verify_start.elapsed(); if actual_hash.as_ref() != self.hash_buf.as_slice() { error!("bitrot reader hash mismatch, id={} data_len={}, out_len={}", self.id, data_len, out.len()); return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "bitrot hash mismatch")); diff --git a/crates/ecstore/src/erasure_coding/decode.rs b/crates/ecstore/src/erasure_coding/decode.rs index 5b0198df8..5c464a775 100644 --- a/crates/ecstore/src/erasure_coding/decode.rs +++ b/crates/ecstore/src/erasure_coding/decode.rs @@ -17,12 +17,12 @@ use crate::disk::error_reduce::reduce_errs; use crate::erasure_codec::workspace::ShardBufferPool; use crate::erasure_coding::{BitrotReader, Erasure}; use crate::get_diagnostics::{ - GET_OBJECT_PATH_LEGACY_DUPLEX, GET_SHARD_READ_OUTCOME_ERROR, GET_SHARD_READ_OUTCOME_MISSING, GET_SHARD_READ_OUTCOME_SUCCESS, - GET_SHARD_ROLE_DATA, GET_SHARD_ROLE_PARITY, GET_STAGE_EMIT, GET_STAGE_RANGE, GET_STAGE_RECONSTRUCT, GET_STAGE_STRIPE_READ, - GET_STAGE_STRIPE_READ_FIRST_SHARD, GET_STAGE_STRIPE_READ_QUORUM, GetObjectFailureReason, classify_io_error, - record_get_object_pipeline_failure, + GET_OBJECT_PATH_LEGACY_DUPLEX, GET_SHARD_READ_ERROR_MISSING, GET_SHARD_READ_ERROR_NONE, GET_SHARD_READ_OUTCOME_ERROR, + GET_SHARD_READ_OUTCOME_MISSING, GET_SHARD_READ_OUTCOME_SUCCESS, GET_SHARD_ROLE_DATA, GET_SHARD_ROLE_PARITY, GET_STAGE_EMIT, + GET_STAGE_RANGE, GET_STAGE_RECONSTRUCT, GET_STAGE_STRIPE_READ, GET_STAGE_STRIPE_READ_FIRST_SHARD, + GET_STAGE_STRIPE_READ_QUORUM, GetObjectFailureReason, classify_io_error, record_get_object_pipeline_failure, }; -use crate::set_disk::shard_source::{ShardStripeSource, StripeReadState}; +use crate::set_disk::shard_source::{ShardReadCost, ShardStripeSource, StripeReadState}; use futures::stream::{FuturesUnordered, StreamExt}; use pin_project_lite::pin_project; use std::future::Future; @@ -35,7 +35,30 @@ use tokio::io::AsyncWrite; use tokio::io::AsyncWriteExt; use tracing::error; -type ShardReadFuture<'a> = Pin, Error>)> + Send + 'a>>; +type ShardReadFuture<'a> = Pin, Error>)> + Send + 'a>>; + +#[derive(Default)] +struct ShardReadCostCounts { + local: usize, + same_node: usize, + remote: usize, + unknown: usize, +} + +impl ShardReadCostCounts { + fn record(&mut self, cost: ShardReadCost) { + match cost { + ShardReadCost::Local => self.local += 1, + ShardReadCost::SameNode => self.same_node += 1, + ShardReadCost::Remote => self.remote += 1, + ShardReadCost::Unknown => self.unknown += 1, + } + } + + fn low_cost(&self) -> usize { + self.local + self.same_node + } +} fn shard_role(index: usize, data_shards: usize) -> &'static str { if index < data_shards { @@ -47,6 +70,7 @@ fn shard_role(index: usize, data_shards: usize) -> &'static str { fn read_shard<'a, R>( index: usize, + read_cost: ShardReadCost, reader: &'a mut Option>, recycled_buf: Option>, shard_size: usize, @@ -66,36 +90,56 @@ where Ok(n) => { buf.truncate(n); if let Some(path) = metrics_path { - rustfs_io_metrics::record_get_object_shard_read( + rustfs_io_metrics::record_get_object_shard_read_observation( path, + index, role, + read_cost.as_str(), GET_SHARD_READ_OUTCOME_SUCCESS, + GET_SHARD_READ_ERROR_NONE, n, read_start.elapsed().as_secs_f64(), + reader.last_verify_duration().as_secs_f64(), ); } - (index, Ok(buf)) + (index, read_cost, Ok(buf)) } Err(e) => { + let verify_duration_secs = reader.last_verify_duration().as_secs_f64(); + let error_class = classify_io_error(&e).as_str(); if let Some(path) = metrics_path { - rustfs_io_metrics::record_get_object_shard_read( + rustfs_io_metrics::record_get_object_shard_read_observation( path, + index, role, + read_cost.as_str(), GET_SHARD_READ_OUTCOME_ERROR, + error_class, 0, read_start.elapsed().as_secs_f64(), + verify_duration_secs, ); } - (index, Err(Error::from(e))) + (index, read_cost, Err(Error::from(e))) } } }) } else { Box::pin(async move { if let Some(path) = metrics_path { - rustfs_io_metrics::record_get_object_shard_read(path, role, GET_SHARD_READ_OUTCOME_MISSING, 0, 0.0); + rustfs_io_metrics::record_get_object_shard_read_observation( + path, + index, + role, + read_cost.as_str(), + GET_SHARD_READ_OUTCOME_MISSING, + GET_SHARD_READ_ERROR_MISSING, + 0, + 0.0, + 0.0, + ); } - (index, Err(Error::FileNotFound)) + (index, read_cost, Err(Error::FileNotFound)) }) } } @@ -110,6 +154,7 @@ pub(crate) struct ParallelReader { data_shards: usize, total_shards: usize, metrics_path: Option<&'static str>, + read_costs: Vec, // Request-scoped shard buffers keyed by shard index. Keeping ownership in // `ParallelReader` avoids dropping unused parity/backup slot buffers between stripes. buffers: ShardBufferPool, @@ -131,11 +176,25 @@ where offset: usize, total_length: usize, metrics_path: Option<&'static str>, + ) -> Self { + let read_costs = vec![ShardReadCost::Unknown; readers.len()]; + Self::new_with_metrics_path_and_read_costs(readers, e, offset, total_length, metrics_path, read_costs) + } + + pub fn new_with_metrics_path_and_read_costs( + readers: Vec>>, + e: Erasure, + offset: usize, + total_length: usize, + metrics_path: Option<&'static str>, + mut read_costs: Vec, ) -> Self { let shard_size = e.shard_size(); let shard_file_size = e.shard_file_size(total_length as i64) as usize; let offset = (offset / e.block_size) * shard_size; + read_costs.resize(readers.len(), ShardReadCost::Unknown); + read_costs.truncate(readers.len()); // Ensure offset does not exceed shard_file_size @@ -147,6 +206,7 @@ where data_shards: e.data_shards, total_shards: e.data_shards + e.parity_shards, metrics_path, + read_costs, buffers: ShardBufferPool::new(e.data_shards + e.parity_shards), } } @@ -174,6 +234,21 @@ where let mut shards: Vec>> = vec![None; num_readers]; let mut errs = vec![None; num_readers]; + let low_cost_available = self + .readers + .iter() + .enumerate() + .filter(|(index, reader)| { + reader.is_some() + && self + .read_costs + .get(*index) + .copied() + .unwrap_or(ShardReadCost::Unknown) + .is_low_cost() + }) + .count(); + let mut successful_costs = ShardReadCostCounts::default(); self.buffers.ensure_slots(num_readers); @@ -191,8 +266,17 @@ where } else { None }; + let read_cost = self.read_costs.get(i).copied().unwrap_or(ShardReadCost::Unknown); scheduled += 1; - sets.push(read_shard(i, reader, recycled_buf, shard_size, self.data_shards, self.metrics_path)); + sets.push(read_shard( + i, + read_cost, + reader, + recycled_buf, + shard_size, + self.data_shards, + self.metrics_path, + )); } } @@ -201,7 +285,7 @@ where let mut failed = 0usize; let mut first_shard_recorded = false; let mut quorum_recorded = false; - while let Some((i, result)) = sets.next().await { + while let Some((i, read_cost, result)) = sets.next().await { completed += 1; if !first_shard_recorded { if let Some(path) = self.metrics_path { @@ -217,6 +301,7 @@ where match result { Ok(v) => { shards[i] = Some(v); + successful_costs.record(read_cost); success += 1; } Err(e) => { @@ -229,9 +314,11 @@ where } else { None }; + let next_read_cost = self.read_costs.get(next_i).copied().unwrap_or(ShardReadCost::Unknown); scheduled += 1; sets.push(read_shard( next_i, + next_read_cost, next_reader, recycled_buf, shard_size, @@ -266,6 +353,20 @@ where } } + if let Some(path) = self.metrics_path { + rustfs_io_metrics::record_get_object_shard_read_cost_summary( + path, + successful_costs.local, + successful_costs.same_node, + successful_costs.remote, + successful_costs.unknown, + low_cost_available, + successful_costs.low_cost(), + self.data_shards, + low_cost_available >= self.data_shards, + ); + } + (shards, errs) } @@ -292,7 +393,7 @@ where async fn read_next_stripe(&mut self) -> StripeReadState { let read_quorum = self.data_shards; let (shards, errors) = ParallelReader::read(self).await; - StripeReadState::from_parts(shards, errors, read_quorum) + StripeReadState::from_parts_with_read_costs(shards, errors, &self.read_costs, read_quorum) } } @@ -438,6 +539,39 @@ impl Erasure { length: usize, total_length: usize, ) -> (usize, Option) + where + W: AsyncWrite + Send + Sync + Unpin, + R: AsyncRead + Unpin + Send + Sync, + { + self.decode_inner(writer, readers, offset, length, total_length, None).await + } + + pub(crate) async fn decode_with_read_costs( + &self, + writer: &mut W, + readers: Vec>>, + offset: usize, + length: usize, + total_length: usize, + read_costs: Vec, + ) -> (usize, Option) + where + W: AsyncWrite + Send + Sync + Unpin, + R: AsyncRead + Unpin + Send + Sync, + { + self.decode_inner(writer, readers, offset, length, total_length, Some(read_costs)) + .await + } + + async fn decode_inner( + &self, + writer: &mut W, + readers: Vec>>, + offset: usize, + length: usize, + total_length: usize, + read_costs: Option>, + ) -> (usize, Option) where W: AsyncWrite + Send + Sync + Unpin, R: AsyncRead + Unpin + Send + Sync, @@ -464,13 +598,24 @@ impl Erasure { let mut written = 0; - let mut reader = ParallelReader::new_with_metrics_path( - readers, - self.clone(), - offset, - total_length, - Some(GET_OBJECT_PATH_LEGACY_DUPLEX), - ); + let mut reader = if let Some(read_costs) = read_costs { + ParallelReader::new_with_metrics_path_and_read_costs( + readers, + self.clone(), + offset, + total_length, + Some(GET_OBJECT_PATH_LEGACY_DUPLEX), + read_costs, + ) + } else { + ParallelReader::new_with_metrics_path( + readers, + self.clone(), + offset, + total_length, + Some(GET_OBJECT_PATH_LEGACY_DUPLEX), + ) + }; let start = offset / self.block_size; let end = end_offset.saturating_sub(1) / self.block_size; diff --git a/crates/ecstore/src/get_diagnostics.rs b/crates/ecstore/src/get_diagnostics.rs index ca2f88a5e..7df1edf95 100644 --- a/crates/ecstore/src/get_diagnostics.rs +++ b/crates/ecstore/src/get_diagnostics.rs @@ -58,6 +58,12 @@ pub(crate) const GET_READER_POLL_READY_ERROR: &str = "ready_error"; pub(crate) const GET_SHARD_READ_OUTCOME_ERROR: &str = "error"; pub(crate) const GET_SHARD_READ_OUTCOME_MISSING: &str = "missing"; pub(crate) const GET_SHARD_READ_OUTCOME_SUCCESS: &str = "success"; +pub(crate) const GET_SHARD_READ_COST_LOCAL: &str = "local"; +pub(crate) const GET_SHARD_READ_COST_REMOTE: &str = "remote"; +pub(crate) const GET_SHARD_READ_COST_SAME_NODE: &str = "same_node"; +pub(crate) const GET_SHARD_READ_COST_UNKNOWN: &str = "unknown"; +pub(crate) const GET_SHARD_READ_ERROR_MISSING: &str = "missing"; +pub(crate) const GET_SHARD_READ_ERROR_NONE: &str = "none"; pub(crate) const GET_SHARD_ROLE_DATA: &str = "data"; pub(crate) const GET_SHARD_ROLE_PARITY: &str = "parity"; @@ -244,6 +250,12 @@ mod tests { assert_eq!(GET_SHARD_READ_OUTCOME_ERROR, "error"); assert_eq!(GET_SHARD_READ_OUTCOME_MISSING, "missing"); assert_eq!(GET_SHARD_READ_OUTCOME_SUCCESS, "success"); + assert_eq!(GET_SHARD_READ_COST_LOCAL, "local"); + assert_eq!(GET_SHARD_READ_COST_REMOTE, "remote"); + assert_eq!(GET_SHARD_READ_COST_SAME_NODE, "same_node"); + assert_eq!(GET_SHARD_READ_COST_UNKNOWN, "unknown"); + assert_eq!(GET_SHARD_READ_ERROR_MISSING, "missing"); + assert_eq!(GET_SHARD_READ_ERROR_NONE, "none"); assert_eq!(GET_SHARD_ROLE_DATA, "data"); assert_eq!(GET_SHARD_ROLE_PARITY, "parity"); assert_eq!(GET_METADATA_RESPONSE_CORRUPT, "corrupt"); diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index f37c81702..1694f3b09 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -24,6 +24,7 @@ use crate::get_diagnostics::{ GET_STAGE_RANGE, GET_STAGE_READER_SETUP, GetObjectFailureReason, classify_disk_error, record_get_object_pipeline_failure, record_get_object_pipeline_failure_for_path, }; +use crate::set_disk::shard_source::ShardReadCost; use metrics::counter; use rustfs_config::{DEFAULT_OBJECT_ZERO_COPY_ENABLE, ENV_OBJECT_ZERO_COPY_ENABLE}; use std::{ @@ -361,6 +362,14 @@ fn is_metadata_fanout_ignored_error(err: &DiskError) -> bool { OBJECT_OP_IGNORED_ERRS.iter().any(|ignored| ignored == err) } +fn shard_read_cost_for_disk(disk: Option<&DiskStore>) -> ShardReadCost { + match disk { + Some(disk) if disk.is_local() => ShardReadCost::Local, + Some(_) => ShardReadCost::Remote, + None => ShardReadCost::Unknown, + } +} + #[derive(Clone, Debug, Eq, Hash, PartialEq)] struct ReadRepairHealCacheKey { bucket: String, @@ -1713,8 +1722,10 @@ impl SetDisks { let reader_setup_stage_start = rustfs_io_metrics::get_stage_metrics_enabled().then(Instant::now); let mut readers = Vec::with_capacity(disks.len()); + let mut read_costs = Vec::with_capacity(disks.len()); let mut errors = Vec::with_capacity(disks.len()); for (idx, disk_op) in disks.iter().enumerate() { + read_costs.push(shard_read_cost_for_disk(disk_op.as_ref())); match create_bitrot_reader( files[idx].data.as_deref(), disk_op.as_ref(), @@ -1843,7 +1854,9 @@ impl SetDisks { // part_number, part_offset, part_length, part_size // ); let decode_stage_start = rustfs_io_metrics::get_stage_metrics_enabled().then(Instant::now); - let (written, err) = erasure.decode(writer, readers, part_offset, part_length, part_size).await; + let (written, err) = erasure + .decode_with_read_costs(writer, readers, part_offset, part_length, part_size, read_costs) + .await; if let Some(decode_stage_start) = decode_stage_start { rustfs_io_metrics::record_get_object_decode_duration(decode_stage_start.elapsed().as_secs_f64()); } @@ -1961,8 +1974,10 @@ impl SetDisks { let reader_setup_stage_start = rustfs_io_metrics::get_stage_metrics_enabled().then(Instant::now); let mut readers = Vec::with_capacity(disks.len()); + let mut read_costs = Vec::with_capacity(disks.len()); let mut errors = Vec::with_capacity(disks.len()); for (idx, disk_op) in disks.iter().enumerate() { + read_costs.push(shard_read_cost_for_disk(disk_op.as_ref())); match create_bitrot_reader( files[idx].data.as_deref(), disk_op.as_ref(), @@ -2030,12 +2045,13 @@ impl SetDisks { .await; } - let source = erasure_coding::decode::ParallelReader::new_with_metrics_path( + let source = erasure_coding::decode::ParallelReader::new_with_metrics_path_and_read_costs( readers, erasure.clone(), 0, part_size, Some(GET_OBJECT_PATH_CODEC_STREAMING), + read_costs, ); let engine = crate::erasure_codec::bridge::LegacyEcDecodeEngine::new(erasure); let reader = erasure_coding::decode_reader::ErasureDecodeReader::new(source, engine, part_length)?; diff --git a/crates/ecstore/src/set_disk/shard_source.rs b/crates/ecstore/src/set_disk/shard_source.rs index 156416db0..3bd3cfa47 100644 --- a/crates/ecstore/src/set_disk/shard_source.rs +++ b/crates/ecstore/src/set_disk/shard_source.rs @@ -13,31 +13,79 @@ // limitations under the License. use crate::disk::error::Error; +use crate::get_diagnostics::{ + GET_SHARD_READ_COST_LOCAL, GET_SHARD_READ_COST_REMOTE, GET_SHARD_READ_COST_SAME_NODE, GET_SHARD_READ_COST_UNKNOWN, +}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum ShardReadCost { + Local, + SameNode, + Remote, + Unknown, +} + +impl ShardReadCost { + pub(crate) const fn as_str(self) -> &'static str { + match self { + Self::Local => GET_SHARD_READ_COST_LOCAL, + Self::SameNode => GET_SHARD_READ_COST_SAME_NODE, + Self::Remote => GET_SHARD_READ_COST_REMOTE, + Self::Unknown => GET_SHARD_READ_COST_UNKNOWN, + } + } + + pub(crate) const fn is_low_cost(self) -> bool { + matches!(self, Self::Local | Self::SameNode) + } +} #[derive(Debug, Clone, PartialEq, Eq)] pub(crate) struct ShardSlot { index: usize, + read_cost: ShardReadCost, data: Option>, error: Option, } impl ShardSlot { pub(crate) fn new(index: usize, data: Option>, error: Option) -> Self { - Self { index, data, error } + Self::with_read_cost(index, ShardReadCost::Unknown, data, error) + } + + pub(crate) fn with_read_cost(index: usize, read_cost: ShardReadCost, data: Option>, error: Option) -> Self { + Self { + index, + read_cost, + data, + error, + } } pub(crate) fn data(index: usize, data: Vec) -> Self { Self::new(index, Some(data), None) } + pub(crate) fn data_with_read_cost(index: usize, read_cost: ShardReadCost, data: Vec) -> Self { + Self::with_read_cost(index, read_cost, Some(data), None) + } + pub(crate) fn missing(index: usize, error: Error) -> Self { Self::new(index, None, Some(error)) } + pub(crate) fn missing_with_read_cost(index: usize, read_cost: ShardReadCost, error: Error) -> Self { + Self::with_read_cost(index, read_cost, None, Some(error)) + } + pub(crate) fn index(&self) -> usize { self.index } + pub(crate) fn read_cost(&self) -> ShardReadCost { + self.read_cost + } + pub(crate) fn has_data(&self) -> bool { self.data.is_some() } @@ -63,12 +111,27 @@ impl StripeReadState { } pub(crate) fn from_parts(shards: Vec>>, errors: Vec>, read_quorum: usize) -> Self { + Self::from_parts_with_read_costs(shards, errors, &[], read_quorum) + } + + pub(crate) fn from_parts_with_read_costs( + shards: Vec>>, + errors: Vec>, + read_costs: &[ShardReadCost], + read_quorum: usize, + ) -> Self { let slot_count = shards.len().max(errors.len()); let mut slots = Vec::with_capacity(slot_count); let mut shards = shards.into_iter(); let mut errors = errors.into_iter(); for index in 0..slot_count { - slots.push(ShardSlot::new(index, shards.next().flatten(), errors.next().flatten())); + let read_cost = read_costs.get(index).copied().unwrap_or(ShardReadCost::Unknown); + slots.push(ShardSlot::with_read_cost( + index, + read_cost, + shards.next().flatten(), + errors.next().flatten(), + )); } Self::new(slots, read_quorum) } @@ -125,9 +188,9 @@ mod tests { fn stripe_read_state_tracks_decode_quorum() { let state = StripeReadState::new( vec![ - ShardSlot::data(0, vec![1]), - ShardSlot::missing(1, Error::FileNotFound), - ShardSlot::data(2, vec![2]), + ShardSlot::data_with_read_cost(0, ShardReadCost::Local, vec![1]), + ShardSlot::missing_with_read_cost(1, ShardReadCost::Remote, Error::FileNotFound), + ShardSlot::data_with_read_cost(2, ShardReadCost::SameNode, vec![2]), ], 2, ); @@ -135,6 +198,8 @@ mod tests { assert_eq!(state.available_shards(), 2); assert!(state.can_decode()); assert_eq!(state.slots()[1].index(), 1); + assert_eq!(state.slots()[0].read_cost(), ShardReadCost::Local); + assert!(state.slots()[2].read_cost().is_low_cost()); } #[test] @@ -157,6 +222,21 @@ mod tests { assert_eq!(state.slots()[1].error(), Some(&Error::FileNotFound)); } + #[test] + fn stripe_read_state_preserves_read_cost_hints() { + let state = StripeReadState::from_parts_with_read_costs( + vec![Some(vec![1]), None, Some(vec![3])], + vec![None, Some(Error::FileNotFound)], + &[ShardReadCost::Local, ShardReadCost::Remote, ShardReadCost::Unknown], + 2, + ); + + assert_eq!(state.slots()[0].read_cost(), ShardReadCost::Local); + assert_eq!(state.slots()[1].read_cost(), ShardReadCost::Remote); + assert_eq!(state.slots()[2].read_cost(), ShardReadCost::Unknown); + assert_eq!(ShardReadCost::SameNode.as_str(), GET_SHARD_READ_COST_SAME_NODE); + } + #[test] fn stripe_read_state_reports_complete_data_shards_without_parity() { let state = StripeReadState::from_parts(vec![Some(vec![1]), Some(vec![2]), None], Vec::new(), 2); diff --git a/crates/io-metrics/src/lib.rs b/crates/io-metrics/src/lib.rs index 8716be2f5..f791ee1a0 100644 --- a/crates/io-metrics/src/lib.rs +++ b/crates/io-metrics/src/lib.rs @@ -180,6 +180,12 @@ pub use performance::PerformanceMetrics; static EC_ENCODE_INFLIGHT_BYTES: AtomicU64 = AtomicU64::new(0); static GET_OBJECT_BUFFERED_BYTES: AtomicU64 = AtomicU64::new(0); +const SHARD_READ_COST_LOCAL: &str = "local"; +const SHARD_READ_COST_REMOTE: &str = "remote"; +const SHARD_READ_COST_SAME_NODE: &str = "same_node"; +const SHARD_READ_COST_UNKNOWN: &str = "unknown"; +const LOW_COST_QUORUM_CANDIDATE_FALSE: &str = "false"; +const LOW_COST_QUORUM_CANDIDATE_TRUE: &str = "true"; fn saturating_sub_atomic(counter: &AtomicU64, bytes: u64) -> u64 { let mut current = counter.load(Ordering::Relaxed); @@ -638,6 +644,69 @@ pub fn record_get_object_shard_read( .record(duration_secs); } +/// Record one underlying shard read attempt with bounded locality and error attribution. +#[inline(always)] +#[allow(clippy::too_many_arguments)] +pub fn record_get_object_shard_read_observation( + path: &'static str, + shard_index: usize, + role: &'static str, + cost_class: &'static str, + outcome: &'static str, + error_class: &'static str, + bytes: usize, + duration_secs: f64, + verify_duration_secs: f64, +) { + if !get_stage_metrics_enabled() { + return; + } + record_get_object_shard_read(path, role, outcome, bytes, duration_secs); + + let bytes = u64::try_from(bytes).unwrap_or(u64::MAX); + let shard_index = shard_index.to_string(); + counter!( + "rustfs_io_get_object_shard_read_observed_total", + "path" => path, + "shard_index" => shard_index.clone(), + "role" => role, + "cost_class" => cost_class, + "outcome" => outcome, + "error_class" => error_class + ) + .increment(1); + counter!( + "rustfs_io_get_object_shard_read_observed_bytes_total", + "path" => path, + "shard_index" => shard_index.clone(), + "role" => role, + "cost_class" => cost_class, + "outcome" => outcome, + "error_class" => error_class + ) + .increment(bytes); + histogram!( + "rustfs_io_get_object_shard_read_observed_duration_seconds", + "path" => path, + "shard_index" => shard_index.clone(), + "role" => role, + "cost_class" => cost_class, + "outcome" => outcome, + "error_class" => error_class + ) + .record(duration_secs); + histogram!( + "rustfs_io_get_object_shard_bitrot_verify_duration_seconds", + "path" => path, + "shard_index" => shard_index, + "role" => role, + "cost_class" => cost_class, + "outcome" => outcome, + "error_class" => error_class + ) + .record(verify_duration_secs); +} + #[inline(always)] fn shard_read_fanout_to_f64(value: usize) -> f64 { u32::try_from(value).map(f64::from).unwrap_or(f64::from(u32::MAX)) @@ -648,6 +717,48 @@ fn metadata_fanout_count_to_f64(value: usize) -> f64 { u32::try_from(value).map(f64::from).unwrap_or(f64::from(u32::MAX)) } +/// Record per-stripe shard locality shape for GetObject read-path attribution. +#[inline(always)] +#[allow(clippy::too_many_arguments)] +pub fn record_get_object_shard_read_cost_summary( + path: &'static str, + local: usize, + same_node: usize, + remote: usize, + unknown: usize, + low_cost_available: usize, + low_cost_successful: usize, + read_quorum: usize, + low_cost_quorum_candidate: bool, +) { + if !get_stage_metrics_enabled() { + return; + } + histogram!("rustfs_io_get_object_shard_read_cost_class_count", "path" => path, "cost_class" => SHARD_READ_COST_LOCAL) + .record(shard_read_fanout_to_f64(local)); + histogram!("rustfs_io_get_object_shard_read_cost_class_count", "path" => path, "cost_class" => SHARD_READ_COST_SAME_NODE) + .record(shard_read_fanout_to_f64(same_node)); + histogram!("rustfs_io_get_object_shard_read_cost_class_count", "path" => path, "cost_class" => SHARD_READ_COST_REMOTE) + .record(shard_read_fanout_to_f64(remote)); + histogram!("rustfs_io_get_object_shard_read_cost_class_count", "path" => path, "cost_class" => SHARD_READ_COST_UNKNOWN) + .record(shard_read_fanout_to_f64(unknown)); + histogram!("rustfs_io_get_object_shard_read_low_cost_available", "path" => path) + .record(shard_read_fanout_to_f64(low_cost_available)); + histogram!("rustfs_io_get_object_shard_read_low_cost_successful", "path" => path) + .record(shard_read_fanout_to_f64(low_cost_successful)); + histogram!("rustfs_io_get_object_shard_read_quorum", "path" => path).record(shard_read_fanout_to_f64(read_quorum)); + counter!( + "rustfs_io_get_object_shard_read_low_cost_quorum_candidate_total", + "path" => path, + "candidate" => if low_cost_quorum_candidate { + LOW_COST_QUORUM_CANDIDATE_TRUE + } else { + LOW_COST_QUORUM_CANDIDATE_FALSE + } + ) + .increment(1); +} + /// Record per-stripe shard-read fanout shape for GetObject read-path attribution. #[inline(always)] pub fn record_get_object_shard_read_fanout( @@ -1340,6 +1451,8 @@ mod tests { record_get_object_duplex_backpressure_duration(0.005); record_get_object_pipeline_failure("decode", "read_quorum"); record_get_object_pipeline_failure_for_path("codec_streaming", "decode", "read_quorum"); + record_get_object_shard_read_observation("codec_streaming", 0, "data", "local", "success", "none", 1024, 0.004, 0.001); + record_get_object_shard_read_cost_summary("codec_streaming", 3, 1, 2, 0, 4, 4, 4, true); assert!(0.005_f64.is_sign_positive()); } @@ -1417,6 +1530,8 @@ mod tests { record_get_object_duplex_backpressure_duration(0.005); record_get_object_pipeline_failure("decode", "read_quorum"); record_get_object_pipeline_failure_for_path("codec_streaming", "decode", "read_quorum"); + record_get_object_shard_read_observation("codec_streaming", 0, "data", "local", "success", "none", 1024, 0.004, 0.001); + record_get_object_shard_read_cost_summary("codec_streaming", 3, 1, 2, 0, 4, 4, 4, true); set_get_stage_metrics_enabled(false); } @@ -1458,6 +1573,8 @@ mod tests { record_get_object_duplex_backpressure_duration(0.005); record_get_object_pipeline_failure("decode", "read_quorum"); record_get_object_pipeline_failure_for_path("codec_streaming", "decode", "read_quorum"); + record_get_object_shard_read_observation("codec_streaming", 0, "data", "local", "success", "none", 1024, 0.004, 0.001); + record_get_object_shard_read_cost_summary("codec_streaming", 3, 1, 2, 0, 4, 4, 4, true); assert!(!get_stage_metrics_enabled()); } diff --git a/rustfs/src/storage/concurrency/manager.rs b/rustfs/src/storage/concurrency/manager.rs index cd3b929ab..b535fdc7c 100644 --- a/rustfs/src/storage/concurrency/manager.rs +++ b/rustfs/src/storage/concurrency/manager.rs @@ -667,8 +667,9 @@ impl Default for ConcurrencyManager { #[allow(unused_imports)] mod integration_tests { use super::super::io_schedule::{IoLoadLevel, IoPriority}; - use super::super::request_guard::{GetObjectGuard, PutObjectGuard}; + use super::super::request_guard::GetObjectGuard; use super::ConcurrencyManager; + use crate::storage::storage_api::concurrency_consumer::PutObjectGuard; use rustfs_concurrency::{AdmissionState, WorkloadAdmissionSnapshotProvider, WorkloadClass}; use rustfs_config::MI_B; use rustfs_io_core::io_profile::{AccessPattern, StorageMedia}; diff --git a/rustfs/src/storage/concurrency/request_guard.rs b/rustfs/src/storage/concurrency/request_guard.rs index 7737192c9..70a873aa0 100644 --- a/rustfs/src/storage/concurrency/request_guard.rs +++ b/rustfs/src/storage/concurrency/request_guard.rs @@ -95,12 +95,12 @@ impl Drop for GetObjectGuard { // `elapsed()` method remains meaningful for tests and callers. let duration_secs = self.start_time.elapsed().as_secs_f64(); // Use the caller-set status, or "unknown" if the result was never set - // (e.g., the future was cancelled or the guard dropped without explicit completion). + // (e.g., the future was canceled or the guard dropped without explicit completion). let status = self.result.unwrap_or("unknown"); record_get_object_request_result(status, duration_secs); if let Err(previous) = - ACTIVE_GET_REQUESTS.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| current.checked_sub(1)) + ACTIVE_GET_REQUESTS.try_update(Ordering::Relaxed, Ordering::Relaxed, |current| current.checked_sub(1)) { debug_assert_eq!( previous, 0, @@ -160,7 +160,7 @@ impl Drop for PutObjectGuard { record_put_object_request_result(status, duration_secs); if let Err(previous) = - ACTIVE_PUT_REQUESTS.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| current.checked_sub(1)) + ACTIVE_PUT_REQUESTS.try_update(Ordering::Relaxed, Ordering::Relaxed, |current| current.checked_sub(1)) { debug_assert_eq!( previous, 0,