From f1b92af4a37267e01b944129ab1f41e2154073bf Mon Sep 17 00:00:00 2001 From: houseme Date: Sun, 23 Aug 2026 01:15:38 +0800 Subject: [PATCH] feat(ecstore): coalesce GET ReadVersion RPCs (#6395) --- crates/ecstore/src/cluster/rpc/remote_disk.rs | 97 ++- .../src/cluster/rpc/runtime_sources.rs | 60 +- crates/ecstore/src/disk/mod.rs | 94 ++- crates/ecstore/src/lib.rs | 8 + crates/ecstore/src/runtime/global.rs | 15 +- .../src/set_disk/core/io_primitives.rs | 572 +++++++++++++++++- crates/ecstore/src/set_disk/ops/object.rs | 2 +- crates/ecstore/src/set_disk/read.rs | 80 ++- crates/io-metrics/src/internode_metrics.rs | 7 + rustfs/src/server/readiness.rs | 5 + rustfs/src/storage/rpc/node_service/disk.rs | 114 +++- rustfs/src/storage/storage_api.rs | 4 + rustfs/src/storage_api.rs | 3 +- 13 files changed, 987 insertions(+), 74 deletions(-) diff --git a/crates/ecstore/src/cluster/rpc/remote_disk.rs b/crates/ecstore/src/cluster/rpc/remote_disk.rs index 32bf3ae57..a47533a81 100644 --- a/crates/ecstore/src/cluster/rpc/remote_disk.rs +++ b/crates/ecstore/src/cluster/rpc/remote_disk.rs @@ -42,8 +42,9 @@ use futures::lock::Mutex; use metrics::counter; use rustfs_filemeta::{FileInfo, ObjectPartInfo, RawFileInfo}; use rustfs_io_metrics::internode_metrics::{ - INTERNODE_STAGE_READ_VERSION_REQUEST_ENCODE, INTERNODE_STAGE_READ_VERSION_RESPONSE_DECODE, - INTERNODE_STAGE_READ_VERSION_RPC_ROUNDTRIP, + INTERNODE_STAGE_BATCH_READ_VERSION_REQUEST_ENCODE, INTERNODE_STAGE_BATCH_READ_VERSION_RESPONSE_DECODE, + INTERNODE_STAGE_BATCH_READ_VERSION_RPC_ROUNDTRIP, INTERNODE_STAGE_READ_VERSION_REQUEST_ENCODE, + INTERNODE_STAGE_READ_VERSION_RESPONSE_DECODE, INTERNODE_STAGE_READ_VERSION_RPC_ROUNDTRIP, }; use rustfs_protos::ChannelClass; use rustfs_protos::evict_failed_connection; @@ -98,6 +99,7 @@ const NS_SCANNER_CAPABILITY_PROBE_TIMEOUT: Duration = Duration::from_secs(5); const REMOTE_DISK_READ_RETRY_BASE_BACKOFF: Duration = Duration::from_millis(50); const ENV_RUSTFS_METADATA_BATCH_READ: &str = "RUSTFS_METADATA_BATCH_READ"; const LEGACY_ENV_RUSTFS_BATCH_METADATA_RPC: &str = "RUSTFS_BATCH_METADATA_RPC"; +const ENV_RUSTFS_GET_METADATA_READ_VERSION_COALESCE: &str = "RUSTFS_GET_METADATA_READ_VERSION_COALESCE"; const BATCH_METADATA_RPC_OFF: &str = "off"; const BATCH_METADATA_RPC_AUTO: &str = "auto"; const BATCH_METADATA_RPC_ON: &str = "on"; @@ -202,7 +204,8 @@ fn parse_batch_metadata_rpc_mode(raw: &str) -> BatchMetadataRpcMode { } fn batch_metadata_rpc_mode_from_env() -> BatchMetadataRpcMode { - rustfs_utils::get_env_opt_str(ENV_RUSTFS_METADATA_BATCH_READ) + rustfs_utils::get_env_opt_str(ENV_RUSTFS_GET_METADATA_READ_VERSION_COALESCE) + .or_else(|| rustfs_utils::get_env_opt_str(ENV_RUSTFS_METADATA_BATCH_READ)) .or_else(|| rustfs_utils::get_env_opt_str(LEGACY_ENV_RUSTFS_BATCH_METADATA_RPC)) .as_deref() .map(parse_batch_metadata_rpc_mode) @@ -1826,6 +1829,12 @@ fn record_read_version_stage(stage: &'static str, started_at: Option) { } } +fn record_batch_read_version_stage(stage: &'static str, started_at: Option) { + if let Some(started_at) = started_at { + crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_batch_read_version_stage(stage, started_at.elapsed()); + } +} + /// Aggregate encoded size (bytes) of a `ReadMultiple` response, preferring the msgpack payloads /// and falling back to the JSON compatibility strings. Used to size the RPC for the payload /// histogram / large-payload alerting (grpc-optimization P0 instrumentation). @@ -1936,6 +1945,27 @@ fn decode_batch_read_version_response_items( Ok(batch_read_version_resps) } +fn batch_read_version_request_payload_len(req: &BatchReadVersionReq, req_json: &str, req_bin: &[u8]) -> usize { + req.items + .iter() + .fold(req_json.len().saturating_add(req_bin.len()), |total, item| { + total + .saturating_add(item.org_volume.len()) + .saturating_add(item.volume.len()) + .saturating_add(item.path.len()) + .saturating_add(item.version_id.len()) + }) +} + +fn batch_read_version_response_payload_len(response: &BatchReadVersionResponse) -> usize { + response + .batch_read_version_resps + .iter() + .map(String::len) + .sum::() + .saturating_add(response.batch_read_version_resps_bin.iter().map(Bytes::len).sum::()) +} + fn validate_decoded_file_info(file_info: &FileInfo) -> Result<()> { file_info.validate_for_metadata_read().map_err(Into::into) } @@ -2837,14 +2867,19 @@ impl DiskAPI for RemoteDisk { state = "started", "Remote disk RPC started" ); + let batch_read_version_attribution_enabled = rustfs_io_metrics::get_stage_metrics_enabled(); + let encode_started = read_version_stage_timer(batch_read_version_attribution_enabled); let batch_read_version_req = compat_json(&req)?; let batch_read_version_req_bin = encode_msgpack(&req)?; - + record_batch_read_version_stage(INTERNODE_STAGE_BATCH_READ_VERSION_REQUEST_ENCODE, encode_started); + let request_payload_bytes = batch_read_version_attribution_enabled + .then(|| batch_read_version_request_payload_len(&req, &batch_read_version_req, &batch_read_version_req_bin)); let batch_result = self .execute_with_timeout_for_op( "batch_read_version", move || async move { let disk = self.disk_ref().await; + let disk_len = disk.len(); let mut client = self .get_bulk_client() .await @@ -2855,9 +2890,20 @@ impl DiskAPI for RemoteDisk { batch_read_version_req_bin: batch_read_version_req_bin.into(), }); + crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_batch_read_version_request(); + if let Some(request_payload_bytes) = request_payload_bytes { + crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_batch_read_version_sent_bytes( + request_payload_bytes.saturating_add(disk_len), + ); + } + let rpc_started = read_version_stage_timer(batch_read_version_attribution_enabled); let response = match client.batch_read_version(request).await { - Ok(response) => response.into_inner(), + Ok(response) => { + record_batch_read_version_stage(INTERNODE_STAGE_BATCH_READ_VERSION_RPC_ROUNDTRIP, rpc_started); + response.into_inner() + } Err(status) if status.code() == Code::Unimplemented => { + record_batch_read_version_stage(INTERNODE_STAGE_BATCH_READ_VERSION_RPC_ROUNDTRIP, rpc_started); if mode.should_fallback_on_unimplemented() { record_batch_read_version_gate_decision(mode, BATCH_READ_VERSION_GATE_FALLBACK_UNIMPLEMENTED); warn!( @@ -2874,6 +2920,7 @@ impl DiskAPI for RemoteDisk { } record_batch_read_version_gate_decision(mode, BATCH_READ_VERSION_GATE_UNSUPPORTED_NO_FALLBACK); + crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_batch_read_version_error(); warn!( event = EVENT_REMOTE_DISK_RPC, component = LOG_COMPONENT_ECSTORE, @@ -2886,14 +2933,33 @@ impl DiskAPI for RemoteDisk { ); return Err(Error::from(status)); } - Err(status) => return Err(Error::from(status)), + Err(status) => { + record_batch_read_version_stage(INTERNODE_STAGE_BATCH_READ_VERSION_RPC_ROUNDTRIP, rpc_started); + crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_batch_read_version_error(); + return Err(Error::from(status)); + } }; if !response.success { + crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_batch_read_version_error(); return Err(response.error.unwrap_or_default().into()); } - decode_batch_read_version_response_items(response, &self.endpoint).map(Some) + crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_batch_read_version_recv_bytes( + batch_read_version_response_payload_len(&response), + ); + let decode_started = read_version_stage_timer(batch_read_version_attribution_enabled); + match decode_batch_read_version_response_items(response, &self.endpoint) { + Ok(batch_read_version_resps) => { + record_batch_read_version_stage(INTERNODE_STAGE_BATCH_READ_VERSION_RESPONSE_DECODE, decode_started); + Ok(Some(batch_read_version_resps)) + } + Err(err) => { + record_batch_read_version_stage(INTERNODE_STAGE_BATCH_READ_VERSION_RESPONSE_DECODE, decode_started); + crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_batch_read_version_error(); + Err(err) + } + } }, get_max_timeout_duration(), ) @@ -4621,6 +4687,7 @@ mod tests { } else { "file version not found".to_string() }, + error_code: if success { 0 } else { DiskError::FileVersionNotFound.to_u32() }, } } @@ -4740,6 +4807,7 @@ mod tests { fn batch_metadata_rpc_mode_uses_documented_env_before_legacy_alias() { temp_env::with_vars( [ + (ENV_RUSTFS_GET_METADATA_READ_VERSION_COALESCE, None::<&str>), (ENV_RUSTFS_METADATA_BATCH_READ, Some("auto")), (LEGACY_ENV_RUSTFS_BATCH_METADATA_RPC, Some("on")), ], @@ -4749,10 +4817,25 @@ mod tests { ); } + #[test] + fn batch_metadata_rpc_mode_uses_get_coalescer_env_before_batch_env() { + temp_env::with_vars( + [ + (ENV_RUSTFS_GET_METADATA_READ_VERSION_COALESCE, Some("on")), + (ENV_RUSTFS_METADATA_BATCH_READ, Some("off")), + (LEGACY_ENV_RUSTFS_BATCH_METADATA_RPC, Some("off")), + ], + || { + assert_eq!(batch_metadata_rpc_mode_from_env(), BatchMetadataRpcMode::On); + }, + ); + } + #[test] fn batch_metadata_rpc_mode_falls_back_to_legacy_env_alias() { temp_env::with_vars( [ + (ENV_RUSTFS_GET_METADATA_READ_VERSION_COALESCE, None::<&str>), (ENV_RUSTFS_METADATA_BATCH_READ, None::<&str>), (LEGACY_ENV_RUSTFS_BATCH_METADATA_RPC, Some("on")), ], diff --git a/crates/ecstore/src/cluster/rpc/runtime_sources.rs b/crates/ecstore/src/cluster/rpc/runtime_sources.rs index 0c8393a2e..ab9a34fe9 100644 --- a/crates/ecstore/src/cluster/rpc/runtime_sources.rs +++ b/crates/ecstore/src/cluster/rpc/runtime_sources.rs @@ -14,9 +14,10 @@ use rustfs_io_metrics::internode_metrics::{ INTERNODE_MSGPACK_CODEC_JSON, INTERNODE_MSGPACK_CODEC_MSGPACK, INTERNODE_MSGPACK_DIRECTION_RESPONSE, - INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_READ_MULTIPLE, INTERNODE_OPERATION_GRPC_READ_VERSION, - INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_OPERATION_PUT_FILE_STREAM, INTERNODE_OPERATION_READ_FILE_STREAM, - INTERNODE_TRANSPORT_BACKEND_GRPC, INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, global_internode_metrics, + INTERNODE_OPERATION_GRPC_BATCH_READ_VERSION, INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_READ_MULTIPLE, + INTERNODE_OPERATION_GRPC_READ_VERSION, INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_OPERATION_PUT_FILE_STREAM, + INTERNODE_OPERATION_READ_FILE_STREAM, INTERNODE_TRANSPORT_BACKEND_GRPC, INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, + global_internode_metrics, }; use std::time::Duration; @@ -93,6 +94,59 @@ pub(crate) fn record_remote_disk_grpc_read_version_request() { ); } +pub(crate) fn record_remote_disk_grpc_batch_read_version_request() { + if !rustfs_io_metrics::get_stage_metrics_enabled() { + return; + } + global_internode_metrics().record_outgoing_request_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_BATCH_READ_VERSION, + INTERNODE_TRANSPORT_BACKEND_GRPC, + ); +} + +pub(crate) fn record_remote_disk_grpc_batch_read_version_stage(stage: &'static str, duration: Duration) { + if !rustfs_io_metrics::get_stage_metrics_enabled() { + return; + } + global_internode_metrics().record_stage_duration_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_BATCH_READ_VERSION, + INTERNODE_TRANSPORT_BACKEND_GRPC, + stage, + duration, + ); +} + +pub(crate) fn record_remote_disk_grpc_batch_read_version_error() { + if !rustfs_io_metrics::get_stage_metrics_enabled() { + return; + } + global_internode_metrics() + .record_error_for_operation_and_backend(INTERNODE_OPERATION_GRPC_BATCH_READ_VERSION, INTERNODE_TRANSPORT_BACKEND_GRPC); +} + +pub(crate) fn record_remote_disk_grpc_batch_read_version_sent_bytes(bytes: usize) { + if !rustfs_io_metrics::get_stage_metrics_enabled() { + return; + } + global_internode_metrics().record_sent_bytes_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_BATCH_READ_VERSION, + INTERNODE_TRANSPORT_BACKEND_GRPC, + bytes, + ); +} + +pub(crate) fn record_remote_disk_grpc_batch_read_version_recv_bytes(bytes: usize) { + if !rustfs_io_metrics::get_stage_metrics_enabled() { + return; + } + global_internode_metrics().record_recv_bytes_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_BATCH_READ_VERSION, + INTERNODE_TRANSPORT_BACKEND_GRPC, + bytes, + ); + record_grpc_payload_size(INTERNODE_OPERATION_GRPC_BATCH_READ_VERSION, bytes); +} + pub(crate) fn record_remote_disk_grpc_read_version_error() { if !rustfs_io_metrics::get_stage_metrics_enabled() { return; diff --git a/crates/ecstore/src/disk/mod.rs b/crates/ecstore/src/disk/mod.rs index ecd0179e0..fb8c8de5a 100644 --- a/crates/ecstore/src/disk/mod.rs +++ b/crates/ecstore/src/disk/mod.rs @@ -44,6 +44,8 @@ pub const PART_TRANSACTION_ROLLBACK: &str = "rollback"; const LOG_COMPONENT_ECSTORE: &str = "ecstore"; const LOG_SUBSYSTEM_DISK: &str = "disk"; const EVENT_DISK_PART_ERR_UNCLASSIFIED: &str = "disk_part_err_unclassified"; +const ENV_BATCH_READ_VERSION_SERVER_PARALLELISM: &str = "RUSTFS_BATCH_READ_VERSION_SERVER_PARALLELISM"; +const BATCH_READ_VERSION_SERVER_PARALLELISM: usize = 4; pub fn part_transaction_path(part_path: &str) -> String { match part_path.rsplit_once('/') { @@ -62,6 +64,7 @@ use bytes::Bytes; use endpoint::Endpoint; use error::DiskError; use error::{Error, Result}; +use futures::stream::{self, StreamExt}; use local::LocalDisk; use rustfs_filemeta::{FileInfo, ObjectPartInfo, RawFileInfo}; use rustfs_madmin::info_commands::DiskMetrics; @@ -417,6 +420,14 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(level = "trace", skip_all)] + async fn batch_read_version(&self, req: BatchReadVersionReq) -> Result> { + match self { + Disk::Local(local_disk) => local_disk.batch_read_version(req).await, + Disk::Remote(remote_disk) => remote_disk.batch_read_version(req).await, + } + } + #[tracing::instrument(level = "trace", skip_all)] async fn read_xl(&self, volume: &str, path: &str, read_data: bool) -> Result { match self { @@ -1028,36 +1039,47 @@ where D: DiskAPI + ?Sized, { validate_batch_read_version_item_count(req.items.len())?; + let parallelism = batch_read_version_server_parallelism(); - let mut responses = Vec::with_capacity(req.items.len()); - for (index, item) in req.items.iter().enumerate() { - let response = match disk - .read_version(&item.org_volume, &item.volume, &item.path, &item.version_id, &req.opts) - .await - { - Ok(file_info) => BatchReadVersionResp { - index, - path: item.path.clone(), - version_id: item.version_id.clone(), - success: true, - file_info, - error: String::new(), - }, - Err(err) => BatchReadVersionResp { - index, - path: item.path.clone(), - version_id: item.version_id.clone(), - success: false, - file_info: FileInfo::default(), - error: err.to_string(), - }, - }; - responses.push(response); - } + let mut responses = stream::iter(req.items.into_iter().enumerate()) + .map(|(index, item)| async move { + match disk + .read_version(&item.org_volume, &item.volume, &item.path, &item.version_id, &req.opts) + .await + { + Ok(file_info) => BatchReadVersionResp { + index, + path: item.path, + version_id: item.version_id, + success: true, + file_info, + error: String::new(), + error_code: 0, + }, + Err(err) => BatchReadVersionResp { + index, + path: item.path, + version_id: item.version_id, + success: false, + file_info: FileInfo::default(), + error: err.to_string(), + error_code: err.to_u32(), + }, + } + }) + .buffer_unordered(parallelism) + .collect::>() + .await; + responses.sort_unstable_by_key(|response| response.index); Ok(responses) } +fn batch_read_version_server_parallelism() -> usize { + rustfs_utils::get_env_usize(ENV_BATCH_READ_VERSION_SERVER_PARALLELISM, BATCH_READ_VERSION_SERVER_PARALLELISM) + .clamp(1, BATCH_READ_VERSION_MAX_ITEMS) +} + #[derive(Debug, Default, Serialize, Deserialize)] pub struct CheckPartsResp { pub results: Vec, @@ -1322,6 +1344,8 @@ pub struct BatchReadVersionResp { pub success: bool, pub file_info: FileInfo, pub error: String, + #[serde(default)] + pub error_code: u32, } pub fn validate_batch_read_version_item_count(item_count: usize) -> Result<()> { @@ -1417,6 +1441,26 @@ mod tests { assert!(!partial_valid_location.valid()); } + #[test] + fn batch_read_version_server_parallelism_defaults_to_conservative_four() { + temp_env::with_var(ENV_BATCH_READ_VERSION_SERVER_PARALLELISM, None::<&str>, || { + assert_eq!(batch_read_version_server_parallelism(), 4); + }); + } + + #[test] + fn batch_read_version_server_parallelism_honors_env_with_bounds() { + temp_env::with_var(ENV_BATCH_READ_VERSION_SERVER_PARALLELISM, Some("8"), || { + assert_eq!(batch_read_version_server_parallelism(), 8); + }); + temp_env::with_var(ENV_BATCH_READ_VERSION_SERVER_PARALLELISM, Some("0"), || { + assert_eq!(batch_read_version_server_parallelism(), 1); + }); + temp_env::with_var(ENV_BATCH_READ_VERSION_SERVER_PARALLELISM, Some("9999"), || { + assert_eq!(batch_read_version_server_parallelism(), BATCH_READ_VERSION_MAX_ITEMS); + }); + } + /// Test FileInfoVersions find_version_index #[test] fn test_file_info_versions_find_version_index() { diff --git a/crates/ecstore/src/lib.rs b/crates/ecstore/src/lib.rs index 2ed643ead..da4abbd39 100644 --- a/crates/ecstore/src/lib.rs +++ b/crates/ecstore/src/lib.rs @@ -81,6 +81,14 @@ pub fn shutdown_background_monitors() { cluster::rpc::shutdown_background_monitors(); } +/// Publish that the process is ready to serve user-object GET traffic. +/// +/// Experimental metadata coalescing is allowed to run only after this point so +/// startup and internal metadata reads keep the original per-disk path. +pub fn mark_get_metadata_read_version_coalescing_service_ready() { + runtime::global::mark_get_metadata_read_version_coalescing_service_ready(); +} + #[cfg(test)] mod rio_tests { #[test] diff --git a/crates/ecstore/src/runtime/global.rs b/crates/ecstore/src/runtime/global.rs index fcc4411f0..56ba22a0a 100644 --- a/crates/ecstore/src/runtime/global.rs +++ b/crates/ecstore/src/runtime/global.rs @@ -25,7 +25,10 @@ use lazy_static::lazy_static; use rustfs_lock::client::LockClient; use std::{ collections::HashMap, - sync::{Arc, OnceLock}, + sync::{ + Arc, OnceLock, + atomic::{AtomicBool, Ordering}, + }, time::SystemTime, }; use tokio::sync::{OnceCell, RwLock}; @@ -37,6 +40,16 @@ pub const DISK_MIN_INODES: u64 = 1000; pub const DISK_FILL_FRACTION: f64 = 0.99; pub const DISK_RESERVE_FRACTION: f64 = 0.15; +static GET_METADATA_READ_VERSION_COALESCING_SERVICE_READY: AtomicBool = AtomicBool::new(false); + +pub(crate) fn mark_get_metadata_read_version_coalescing_service_ready() { + GET_METADATA_READ_VERSION_COALESCING_SERVICE_READY.store(true, Ordering::Release); +} + +pub(crate) fn get_metadata_read_version_coalescing_service_ready() -> bool { + GET_METADATA_READ_VERSION_COALESCING_SERVICE_READY.load(Ordering::Acquire) +} + // Global singletons for backward compatibility with MinIO port. // These should be migrated to AppContext over time. // See issue #730 for migration plan. diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index d17c1c751..1a77ba267 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -53,11 +53,12 @@ use crate::diagnostics::get::{ GetObjectFailureReason, classify_disk_error, get_stage_timer_if_enabled, record_get_object_pipeline_failure, record_get_object_pipeline_failure_for_path, record_get_stage_duration_if_enabled, }; -use crate::disk::disk_store::DiskStoreRenameDataExt; +use crate::disk::disk_store::{DiskStoreRenameDataExt, get_drive_metadata_timeout}; use crate::disk::local::DELETE_DATA_DIR_MARKER_PREFIX; use crate::disk::{ - DataDirDeleteStatus, OldCurrentSize, PART_TRANSACTION_NEW_META, PART_TRANSACTION_OLD_META, PART_TRANSACTION_ROLLBACK, - PartTransactionAction, STORAGE_FORMAT_FILE_BACKUP, part_transaction_path, + BATCH_READ_VERSION_MAX_ITEMS, BatchReadVersionItem, BatchReadVersionReq, BatchReadVersionResp, DataDirDeleteStatus, Disk, + OldCurrentSize, PART_TRANSACTION_NEW_META, PART_TRANSACTION_OLD_META, PART_TRANSACTION_ROLLBACK, PartTransactionAction, + STORAGE_FORMAT_FILE_BACKUP, part_transaction_path, }; use crate::erasure::coding::BitrotReader; use crate::io_support::bitrot::ShardReader; @@ -75,7 +76,7 @@ use std::{ future::Future, pin::Pin, sync::{ - OnceLock, + Arc, OnceLock, atomic::{AtomicUsize, Ordering}, }, task::{Context, Poll}, @@ -94,6 +95,242 @@ fn metadata_distribution_key(bucket: &str, object: &str) -> String { [bucket, object].join("/") } +fn read_version_coalescing_enabled() -> bool { + let enabled = || { + rustfs_utils::get_env_opt_str(ENV_RUSTFS_GET_METADATA_READ_VERSION_COALESCE) + .is_some_and(|value| value.eq_ignore_ascii_case("auto") || value.eq_ignore_ascii_case("on")) + }; + + #[cfg(test)] + { + enabled() + } + + #[cfg(not(test))] + { + static ENABLED: OnceLock = OnceLock::new(); + *ENABLED.get_or_init(enabled) + } +} + +fn read_version_coalescing_delay() -> Duration { + #[cfg(test)] + { + let micros = rustfs_utils::get_env_u64( + ENV_RUSTFS_GET_METADATA_READ_VERSION_COALESCE_DELAY_MICROS, + DEFAULT_GET_METADATA_READ_VERSION_COALESCE_DELAY_MICROS, + ); + Duration::from_micros(micros) + } + + #[cfg(not(test))] + { + static DELAY: OnceLock = OnceLock::new(); + *DELAY.get_or_init(|| { + Duration::from_micros(rustfs_utils::get_env_u64( + ENV_RUSTFS_GET_METADATA_READ_VERSION_COALESCE_DELAY_MICROS, + DEFAULT_GET_METADATA_READ_VERSION_COALESCE_DELAY_MICROS, + )) + }) + } +} + +struct CoalescedReadVersionRequest { + item: BatchReadVersionItem, + tx: oneshot::Sender>, +} + +#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] +struct ReadVersionCoalescerKey { + disk: usize, + incl_free_versions: bool, + read_data: bool, + healing: bool, +} + +impl ReadVersionCoalescerKey { + fn new(disk: &DiskStore, opts: &ReadOptions) -> Self { + Self { + disk: Arc::as_ptr(disk) as usize, + incl_free_versions: opts.incl_free_versions, + read_data: opts.read_data, + healing: opts.healing, + } + } +} + +#[derive(Default)] +struct ReadVersionCoalescer { + lanes: HashMap>, +} + +fn read_version_coalescer() -> &'static Mutex { + static COALESCER: OnceLock> = OnceLock::new(); + COALESCER.get_or_init(|| Mutex::new(ReadVersionCoalescer::default())) +} + +fn record_read_version_coalescer_event(event: &'static str, item_count: usize) { + counter!( + METRIC_GET_METADATA_READ_VERSION_COALESCER_TOTAL, + "event" => event, + "item_count" => item_count.to_string() + ) + .increment(1); +} + +async fn read_version_via_coalescer( + disk: DiskStore, + org_bucket: &str, + bucket: &str, + object: &str, + version_id: &str, + opts: &ReadOptions, + allow_coalescing: bool, +) -> disk::error::Result { + if !allow_coalescing || !read_version_coalescing_enabled() { + return disk.read_version(org_bucket, bucket, object, version_id, opts).await; + } + if !matches!(disk.as_ref(), Disk::Remote(_)) { + record_read_version_coalescer_event("bypass_non_remote", 1); + return disk.read_version(org_bucket, bucket, object, version_id, opts).await; + } + + let (tx, rx) = oneshot::channel(); + let item = BatchReadVersionItem { + org_volume: org_bucket.to_string(), + volume: bucket.to_string(), + path: object.to_string(), + version_id: version_id.to_string(), + }; + let lane_key = ReadVersionCoalescerKey::new(&disk, opts); + let pending = { + let mut coalescer = read_version_coalescer().lock().await; + let lane = coalescer.lanes.entry(lane_key).or_default(); + let schedule_delayed_flush = lane.is_empty(); + lane.push(CoalescedReadVersionRequest { item, tx }); + if lane.len() >= BATCH_READ_VERSION_MAX_ITEMS { + coalescer.lanes.remove(&lane_key) + } else if schedule_delayed_flush { + let disk = disk.clone(); + let task_opts = *opts; + tokio::spawn(async move { + tokio::time::sleep(read_version_coalescing_delay()).await; + flush_read_version_coalescer_lane(lane_key, disk, task_opts).await; + }); + None + } else { + None + } + }; + + if let Some(pending) = pending { + flush_read_version_coalescer_pending(lane_key, disk, *opts, pending).await; + } + + rx.await + .unwrap_or_else(|_| Err(DiskError::other("coalesced read_version response channel closed"))) +} + +async fn flush_read_version_coalescer_lane(lane_key: ReadVersionCoalescerKey, disk: DiskStore, opts: ReadOptions) { + let pending = { + let mut coalescer = read_version_coalescer().lock().await; + coalescer.lanes.remove(&lane_key).unwrap_or_default() + }; + flush_read_version_coalescer_pending(lane_key, disk, opts, pending).await; +} + +async fn flush_read_version_coalescer_pending( + lane_key: ReadVersionCoalescerKey, + disk: DiskStore, + opts: ReadOptions, + pending: Vec, +) { + if pending.is_empty() { + return; + } + + #[cfg(test)] + { + let mut observed_paths = HashSet::new(); + for request in &pending { + if observed_paths.insert(request.item.path.as_str()) { + disk_call_counters::record(&request.item.path, disk_call_counters::KIND_BATCH_READ_VERSION, lane_key.disk); + } + } + } + + let mut senders = Vec::with_capacity(pending.len()); + let mut items = Vec::with_capacity(pending.len()); + for request in pending { + senders.push(request.tx); + items.push(request.item); + } + + let expected_items = items.clone(); + record_read_version_coalescer_event("attempted_batch", items.len()); + let result = + match tokio::time::timeout(get_drive_metadata_timeout(), disk.batch_read_version(BatchReadVersionReq { items, opts })) + .await + { + Ok(result) => result, + Err(_) => Err(DiskError::Timeout), + }; + match result { + Ok(responses) => { + let results = map_batch_read_version_responses(&expected_items, responses); + for (tx, result) in senders.into_iter().zip(results) { + let _ = tx.send(result); + } + } + Err(err) => { + let message = err.to_string(); + for tx in senders { + let _ = tx.send(Err(DiskError::other(message.clone()))); + } + } + } +} + +fn map_batch_read_version_responses( + expected_items: &[BatchReadVersionItem], + responses: Vec, +) -> Vec> { + let mut results = (0..expected_items.len()) + .map(|_| Err(DiskError::other("coalesced read_version response missing"))) + .collect::>(); + let mut seen = vec![false; expected_items.len()]; + for response in responses { + let Some(expected) = expected_items.get(response.index) else { + continue; + }; + let Some(slot) = results.get_mut(response.index) else { + continue; + }; + if seen[response.index] { + *slot = Err(DiskError::other("coalesced read_version response duplicate index")); + continue; + } + seen[response.index] = true; + if response.path != expected.path || response.version_id != expected.version_id { + *slot = Err(DiskError::other("coalesced read_version response identity mismatch")); + } else { + *slot = if response.success { + Ok(response.file_info) + } else { + Err(batch_read_version_response_error(response.error_code, response.error)) + }; + } + } + results +} + +fn batch_read_version_response_error(error_code: u32, error: String) -> DiskError { + match DiskError::from_u32(error_code) { + Some(DiskError::Io(_)) | None => DiskError::other(error), + Some(error) => error, + } +} + pub(in crate::set_disk) fn bounded_metadata_fanout_order( bucket: &str, object: &str, @@ -133,11 +370,15 @@ pub(in crate::set_disk) fn bounded_metadata_fanout_order( order } use tokio::io::{AsyncRead, ReadBuf}; -use tokio::sync::RwLock; +use tokio::sync::{Mutex, RwLock, oneshot}; use tokio::task::JoinSet; pub(in crate::set_disk) const EVENT_SET_DISK_READ: &str = "set_disk_read"; pub(in crate::set_disk) const ENV_RUSTFS_GET_DATA_BLOCKS_FIRST_READER_SETUP: &str = "RUSTFS_GET_DATA_BLOCKS_FIRST_READER_SETUP"; +const ENV_RUSTFS_GET_METADATA_READ_VERSION_COALESCE: &str = "RUSTFS_GET_METADATA_READ_VERSION_COALESCE"; +const ENV_RUSTFS_GET_METADATA_READ_VERSION_COALESCE_DELAY_MICROS: &str = "RUSTFS_GET_METADATA_READ_VERSION_COALESCE_DELAY_MICROS"; +const DEFAULT_GET_METADATA_READ_VERSION_COALESCE_DELAY_MICROS: u64 = 200; +const METRIC_GET_METADATA_READ_VERSION_COALESCER_TOTAL: &str = "rustfs_get_metadata_read_version_coalescer_total"; pub(in crate::set_disk) const ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE: &str = "RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE"; /// Default reader-setup strategy for the GET read path (rustfs/backlog#1215, /// #1159, #923). @@ -2356,6 +2597,7 @@ impl SetDisks { false, true, 0, + false, ) .await?; Ok((ress, errors)) @@ -2386,6 +2628,36 @@ impl SetDisks { true, caller_allows_early_stop, default_parity_count, + false, + ) + .await + } + + #[allow(clippy::too_many_arguments)] + pub(in crate::set_disk) async fn read_all_fileinfo_observed_for_get_object( + disks: &[Option], + org_bucket: &str, + bucket: &str, + object: &str, + version_id: &str, + read_data: bool, + incl_free_versions: bool, + caller_allows_early_stop: bool, + default_parity_count: usize, + ) -> disk::error::Result<(Vec, Vec>, MetadataFanoutDiagnostics)> { + Self::read_all_fileinfo_inner( + disks, + org_bucket, + bucket, + object, + version_id, + read_data, + false, + incl_free_versions, + true, + caller_allows_early_stop, + default_parity_count, + true, ) .await } @@ -2408,6 +2680,7 @@ impl SetDisks { // subset would fail write quorum (backlog#872 regression). caller_allows_early_stop: bool, default_parity_count: usize, + allow_coalescing: bool, ) -> disk::error::Result<(Vec, Vec>, MetadataFanoutDiagnostics)> { let early_stop_enabled = caller_allows_early_stop && observe && (is_get_metadata_early_stop_enabled() || is_version_early_stop_enabled()); @@ -2424,6 +2697,7 @@ impl SetDisks { healing, incl_free_versions, default_parity_count, + allow_coalescing, ) .await; } @@ -2446,6 +2720,7 @@ impl SetDisks { healing, incl_free_versions, observe, + allow_coalescing, ) .await } @@ -2461,6 +2736,7 @@ impl SetDisks { healing: bool, incl_free_versions: bool, observe: bool, + allow_coalescing: bool, ) -> disk::error::Result<(Vec, Vec>, MetadataFanoutDiagnostics)> { let fanout_start = observe.then(Instant::now); let mut ress = Vec::with_capacity(disks.len()); @@ -2492,7 +2768,7 @@ impl SetDisks { if let Some(delay) = slowtail_fault.as_ref().and_then(|fault| fault.delay_for_disk(disk_index)) { tokio::time::sleep(delay).await; } - disk.read_version(&org_bucket, &bucket, &object, &version_id, &task_opts) + read_version_via_coalescer(disk, &org_bucket, &bucket, &object, &version_id, &task_opts, allow_coalescing) .await } else { Err(DiskError::DiskNotFound) @@ -2559,6 +2835,7 @@ impl SetDisks { healing: bool, incl_free_versions: bool, default_parity_count: usize, + allow_coalescing: bool, ) -> disk::error::Result<(Vec, Vec>, MetadataFanoutDiagnostics)> { let fanout_start = Instant::now(); let mut ress = vec![FileInfo::default(); disks.len()]; @@ -2607,7 +2884,7 @@ impl SetDisks { if let Some(delay) = slowtail_fault.as_ref().and_then(|fault| fault.delay_for_disk(index)) { tokio::time::sleep(delay).await; } - disk.read_version(&org_bucket, &bucket, &object, &version_id, &task_opts) + read_version_via_coalescer(disk, &org_bucket, &bucket, &object, &version_id, &task_opts, allow_coalescing) .await } else { Err(DiskError::DiskNotFound) @@ -5737,6 +6014,7 @@ pub(crate) mod disk_call_counters { /// Kind label for the per-disk `read_version` metadata RPC. pub const KIND_READ_VERSION: &str = "read_version"; + pub const KIND_BATCH_READ_VERSION: &str = "batch_read_version"; /// Registry key: (object, kind, disk_index). type CountKey = (String, String, usize); @@ -6460,6 +6738,286 @@ mod tests { drop(dirs); } + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn metadata_read_version_coalescer_bypasses_local_disks() { + const DISKS: usize = 4; + let bucket = "coalesced-read-version-local-bypass-bucket"; + let object_a = "coalesced-local-object-a"; + let object_b = "coalesced-local-object-b"; + let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await; + install_metadata_fanout_fileinfo(&disks, bucket, object_a, None).await; + install_metadata_fanout_fileinfo(&disks, bucket, object_b, None).await; + + temp_env::async_with_vars( + [ + (ENV_RUSTFS_GET_METADATA_READ_VERSION_COALESCE, Some("auto")), + (ENV_RUSTFS_GET_METADATA_READ_VERSION_COALESCE_DELAY_MICROS, Some("5000")), + ], + async { + let calls = disk_call_counters::observe(object_a); + let disks_a = disks.clone(); + let disks_b = disks.clone(); + let read_a = tokio::spawn(async move { + SetDisks::read_all_fileinfo_observed_for_get_object( + &disks_a, "", bucket, object_a, "", false, false, false, 2, + ) + .await + .map(|(file_infos, errors, _)| (file_infos, errors)) + }); + tokio::task::yield_now().await; + let read_b = tokio::spawn(async move { + SetDisks::read_all_fileinfo_observed_for_get_object( + &disks_b, "", bucket, object_b, "", false, false, false, 2, + ) + .await + .map(|(file_infos, errors, _)| (file_infos, errors)) + }); + + let (metadata_a, errs_a) = read_a + .await + .expect("first read task should not panic") + .expect("first coalesced read should resolve"); + let (metadata_b, errs_b) = read_b + .await + .expect("second read task should not panic") + .expect("second coalesced read should resolve"); + + assert_eq!(metadata_a.iter().filter(|fi| fi.name == object_a).count(), DISKS); + assert_eq!(metadata_b.iter().filter(|fi| fi.name == object_b).count(), DISKS); + assert!(errs_a.iter().all(Option::is_none)); + assert!(errs_b.iter().all(Option::is_none)); + assert_eq!( + calls.total(disk_call_counters::KIND_READ_VERSION), + DISKS as u64, + "local disks still execute the ordinary per-disk read_version path" + ); + assert_eq!( + calls.total(disk_call_counters::KIND_BATCH_READ_VERSION), + 0, + "GET coalescing targets internode RPC count only and must not batch local disk reads" + ); + }, + ) + .await; + + drop(dirs); + } + + #[tokio::test] + async fn metadata_read_version_coalescer_requires_get_object_intent() { + const DISKS: usize = 4; + let bucket = "coalesced-read-version-default-bypass-bucket"; + let object = "default-bypass-object"; + let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await; + install_metadata_fanout_fileinfo(&disks, bucket, object, None).await; + + temp_env::async_with_vars([(ENV_RUSTFS_GET_METADATA_READ_VERSION_COALESCE, Some("auto"))], async { + let calls = disk_call_counters::observe(object); + let (metadata, errs) = SetDisks::read_all_fileinfo(&disks, "", bucket, object, "", false, false, false) + .await + .expect("default metadata read should resolve"); + + assert_eq!(metadata.iter().filter(|fi| fi.name == object).count(), DISKS); + assert!(errs.iter().all(Option::is_none)); + assert_eq!(calls.total(disk_call_counters::KIND_READ_VERSION), DISKS as u64); + assert_eq!( + calls.total(disk_call_counters::KIND_BATCH_READ_VERSION), + 0, + "non-GET metadata paths must bypass coalescer even when the env gate is enabled" + ); + }) + .await; + + drop(dirs); + } + + #[test] + fn batch_read_version_response_mapping_preserves_index_and_errors() { + let expected_items = vec![ + BatchReadVersionItem { + org_volume: String::new(), + volume: "bucket".to_string(), + path: "object-a".to_string(), + version_id: "v-a".to_string(), + }, + BatchReadVersionItem { + org_volume: String::new(), + volume: "bucket".to_string(), + path: "object-b".to_string(), + version_id: "v-b".to_string(), + }, + BatchReadVersionItem { + org_volume: String::new(), + volume: "bucket".to_string(), + path: "object-c".to_string(), + version_id: "v-c".to_string(), + }, + ]; + let ok_file_info = FileInfo { + name: "object-a".to_string(), + ..Default::default() + }; + let responses = vec![ + BatchReadVersionResp { + index: 2, + path: "object-c".to_string(), + version_id: "v-c".to_string(), + success: false, + file_info: FileInfo::default(), + error: "disk read failed".to_string(), + error_code: 0, + }, + BatchReadVersionResp { + index: 0, + path: "object-a".to_string(), + version_id: "v-a".to_string(), + success: true, + file_info: ok_file_info, + error: String::new(), + error_code: 0, + }, + ]; + + let mut results = map_batch_read_version_responses(&expected_items, responses).into_iter(); + let first = results + .next() + .expect("slot 0 should exist") + .expect("slot 0 should map the success response by index"); + assert_eq!(first.name, "object-a"); + + let missing = results + .next() + .expect("slot 1 should exist") + .expect_err("slot 1 should stay missing"); + assert!( + missing.to_string().contains("response missing"), + "unexpected missing response error: {missing}" + ); + + let failed = results + .next() + .expect("slot 2 should exist") + .expect_err("slot 2 should map the response error"); + assert!(failed.to_string().contains("disk read failed"), "unexpected per-item error: {failed}"); + assert!(results.next().is_none()); + } + + #[test] + fn batch_read_version_response_mapping_preserves_typed_not_found_errors() { + let expected_items = vec![ + BatchReadVersionItem { + org_volume: String::new(), + volume: "bucket".to_string(), + path: "object-a".to_string(), + version_id: "v-a".to_string(), + }, + BatchReadVersionItem { + org_volume: String::new(), + volume: "bucket".to_string(), + path: "object-b".to_string(), + version_id: "v-b".to_string(), + }, + ]; + let results = map_batch_read_version_responses( + &expected_items, + vec![ + BatchReadVersionResp { + index: 0, + path: "object-a".to_string(), + version_id: "v-a".to_string(), + success: false, + file_info: FileInfo::default(), + error: DiskError::FileNotFound.to_string(), + error_code: DiskError::FileNotFound.to_u32(), + }, + BatchReadVersionResp { + index: 1, + path: "object-b".to_string(), + version_id: "v-b".to_string(), + success: false, + file_info: FileInfo::default(), + error: DiskError::FileVersionNotFound.to_string(), + error_code: DiskError::FileVersionNotFound.to_u32(), + }, + ], + ); + + assert!(matches!(results.first().expect("slot 0 should exist"), Err(DiskError::FileNotFound))); + assert!(matches!( + results.get(1).expect("slot 1 should exist"), + Err(DiskError::FileVersionNotFound) + )); + } + + #[test] + fn batch_read_version_response_mapping_rejects_identity_mismatch_and_duplicate_index() { + let expected_items = vec![BatchReadVersionItem { + org_volume: String::new(), + volume: "bucket".to_string(), + path: "object-a".to_string(), + version_id: "v-a".to_string(), + }]; + let mismatched = map_batch_read_version_responses( + &expected_items, + vec![BatchReadVersionResp { + index: 0, + path: "object-b".to_string(), + version_id: "v-a".to_string(), + success: true, + file_info: FileInfo { + name: "object-b".to_string(), + ..Default::default() + }, + error: String::new(), + error_code: 0, + }], + ) + .pop() + .expect("slot 0 should exist") + .expect_err("identity mismatch should fail closed"); + assert!( + mismatched.to_string().contains("identity mismatch"), + "unexpected mismatch error: {mismatched}" + ); + + let duplicate = map_batch_read_version_responses( + &expected_items, + vec![ + BatchReadVersionResp { + index: 0, + path: "object-a".to_string(), + version_id: "v-a".to_string(), + success: true, + file_info: FileInfo { + name: "object-a".to_string(), + ..Default::default() + }, + error: String::new(), + error_code: 0, + }, + BatchReadVersionResp { + index: 0, + path: "object-a".to_string(), + version_id: "v-a".to_string(), + success: true, + file_info: FileInfo { + name: "object-a".to_string(), + ..Default::default() + }, + error: String::new(), + error_code: 0, + }, + ], + ) + .pop() + .expect("slot 0 should exist") + .expect_err("duplicate response index should fail closed"); + assert!( + duplicate.to_string().contains("duplicate index"), + "unexpected duplicate error: {duplicate}" + ); + } + /// Isolation guard: unobserved objects record nothing (so parallel tests do /// not inflate one another), and a scope clears its own counts on drop. #[tokio::test] diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 0d429e3b3..8f07a7ff7 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -1294,7 +1294,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { (prepared.snapshot, prepared.object_info) } else { match self - .get_object_fileinfo( + .get_object_fileinfo_for_get_object_reader( bucket, object, opts, diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index 2b556b143..15e7e4e2a 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -259,10 +259,33 @@ impl SetDisks { read_data: bool, caller_allows_early_stop: bool, ) -> Result { - self.get_object_fileinfo_gated(bucket, object, opts, read_data, caller_allows_early_stop) + self.get_object_fileinfo_gated_inner(bucket, object, opts, read_data, caller_allows_early_stop, false) .await } + #[tracing::instrument(level = "debug", skip(self))] + #[hotpath::measure(impl_type = "SetDisks")] + pub(super) async fn get_object_fileinfo_for_get_object_reader( + &self, + bucket: &str, + object: &str, + opts: &ObjectOptions, + read_data: bool, + caller_allows_early_stop: bool, + ) -> Result { + let allow_read_version_coalescing = !crate::bucket::utils::is_meta_bucketname(bucket) + && crate::runtime::global::get_metadata_read_version_coalescing_service_ready(); + self.get_object_fileinfo_gated_inner( + bucket, + object, + opts, + read_data, + caller_allows_early_stop, + allow_read_version_coalescing, + ) + .await + } + /// Like `get_object_fileinfo`, but `allow_early_stop=false` forces the full /// quorum fanout. Read-before-write callers (object tagging) must use this: /// the returned online-disk set is the write target, and the early-stop @@ -275,6 +298,20 @@ impl SetDisks { opts: &ObjectOptions, read_data: bool, allow_early_stop: bool, + ) -> Result { + self.get_object_fileinfo_gated_inner(bucket, object, opts, read_data, allow_early_stop, false) + .await + } + + #[allow(clippy::too_many_arguments)] + async fn get_object_fileinfo_gated_inner( + &self, + bucket: &str, + object: &str, + opts: &ObjectOptions, + read_data: bool, + allow_early_stop: bool, + allow_read_version_coalescing: bool, ) -> Result { let vid = opts.version_id.clone().unwrap_or_default(); let stage_metrics_enabled = rustfs_io_metrics::get_stage_metrics_enabled(); @@ -337,19 +374,34 @@ impl SetDisks { // read_all_fileinfo_observed (see read_all_fileinfo_early_stop in // core/io_primitives.rs); unsafe requests and callers that opt out // (allow_early_stop=false) fall back to full-wait. - let (mut parts_metadata, errs, metadata_fanout_diagnostics) = Self::read_all_fileinfo_observed( - &disks, - "", - bucket, - object, - vid.as_str(), - read_data, - false, - opts.incl_free_versions, - allow_early_stop, - self.default_parity_count, - ) - .await?; + let (mut parts_metadata, errs, metadata_fanout_diagnostics) = if allow_read_version_coalescing { + Self::read_all_fileinfo_observed_for_get_object( + &disks, + "", + bucket, + object, + vid.as_str(), + read_data, + opts.incl_free_versions, + allow_early_stop, + self.default_parity_count, + ) + .await? + } else { + Self::read_all_fileinfo_observed( + &disks, + "", + bucket, + object, + vid.as_str(), + read_data, + false, + opts.incl_free_versions, + allow_early_stop, + self.default_parity_count, + ) + .await? + }; let metadata_metrics_path = if crate::bucket::utils::is_meta_bucketname(bucket) { GET_OBJECT_PATH_INTERNAL_META } else { diff --git a/crates/io-metrics/src/internode_metrics.rs b/crates/io-metrics/src/internode_metrics.rs index 65f93c2a1..6b1ea382c 100644 --- a/crates/io-metrics/src/internode_metrics.rs +++ b/crates/io-metrics/src/internode_metrics.rs @@ -54,6 +54,13 @@ pub const INTERNODE_STAGE_READ_VERSION_RESPONSE_JSON_ENCODE: &str = "read_versio pub const INTERNODE_STAGE_READ_VERSION_RESPONSE_MSGPACK_ENCODE: &str = "read_version_response_msgpack_encode"; pub const INTERNODE_STAGE_READ_VERSION_RPC_ROUNDTRIP: &str = "read_version_rpc_roundtrip"; pub const INTERNODE_STAGE_READ_VERSION_RESPONSE_DECODE: &str = "read_version_response_decode"; +pub const INTERNODE_STAGE_BATCH_READ_VERSION_REQUEST_ENCODE: &str = "batch_read_version_request_encode"; +pub const INTERNODE_STAGE_BATCH_READ_VERSION_REQUEST_DECODE: &str = "batch_read_version_request_decode"; +pub const INTERNODE_STAGE_BATCH_READ_VERSION_DISK_READ: &str = "batch_read_version_disk_read"; +pub const INTERNODE_STAGE_BATCH_READ_VERSION_RESPONSE_JSON_ENCODE: &str = "batch_read_version_response_json_encode"; +pub const INTERNODE_STAGE_BATCH_READ_VERSION_RESPONSE_MSGPACK_ENCODE: &str = "batch_read_version_response_msgpack_encode"; +pub const INTERNODE_STAGE_BATCH_READ_VERSION_RPC_ROUNDTRIP: &str = "batch_read_version_rpc_roundtrip"; +pub const INTERNODE_STAGE_BATCH_READ_VERSION_RESPONSE_DECODE: &str = "batch_read_version_response_decode"; const OPERATION_LABEL: &str = "operation"; const BACKEND_LABEL: &str = "backend"; diff --git a/rustfs/src/server/readiness.rs b/rustfs/src/server/readiness.rs index d3954b7a7..8a934e06e 100644 --- a/rustfs/src/server/readiness.rs +++ b/rustfs/src/server/readiness.rs @@ -20,6 +20,7 @@ use crate::storage_api::server::readiness::contract::admin::StorageAdminApi; use crate::storage_api::server::readiness::{Endpoint, EndpointServerPools, is_dist_erasure}; #[cfg(test)] use crate::storage_api::server::readiness::{Endpoints, PoolEndpoints}; +use crate::storage_api::startup::shutdown::mark_get_metadata_read_version_coalescing_service_ready; use bytes::Bytes; use http::HeaderValue; use http::{Request as HttpRequest, Response, StatusCode}; @@ -212,6 +213,9 @@ where if readiness_gate_blocks_path(path, &readiness) { return Ok(service_not_ready_response(readiness.current_stage())); } + if !is_probe_path(path) && readiness.is_ready() { + mark_get_metadata_read_version_coalescing_service_ready(); + } let resp = inner.call(req).await?; // System is ready, forward to the actual S3/RPC handlers // Transparently converts any response body into a BoxBody, and then Trace/Cors/Compression continues to work @@ -232,6 +236,7 @@ pub async fn publish_ready_when_runtime_ready( collect_node_readiness, |dependency_readiness| { readiness.mark_stage(rustfs_common::SystemStage::FullReady); + mark_get_metadata_read_version_coalescing_service_ready(); if let Some(state_manager) = state_manager { state_manager.update(ServiceState::Ready); } diff --git a/rustfs/src/storage/rpc/node_service/disk.rs b/rustfs/src/storage/rpc/node_service/disk.rs index de80a10f0..617d0b80d 100644 --- a/rustfs/src/storage/rpc/node_service/disk.rs +++ b/rustfs/src/storage/rpc/node_service/disk.rs @@ -23,10 +23,12 @@ use bytes::Bytes; use rustfs_filemeta::FileInfo; use rustfs_io_metrics::internode_metrics::{ INTERNODE_MSGPACK_CODEC_JSON, INTERNODE_MSGPACK_CODEC_MSGPACK, INTERNODE_MSGPACK_DIRECTION_REQUEST, - INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_READ_VERSION, INTERNODE_OPERATION_GRPC_WRITE_ALL, - INTERNODE_STAGE_READ_VERSION_DISK_READ, INTERNODE_STAGE_READ_VERSION_REQUEST_DECODE, - INTERNODE_STAGE_READ_VERSION_RESPONSE_JSON_ENCODE, INTERNODE_STAGE_READ_VERSION_RESPONSE_MSGPACK_ENCODE, - INTERNODE_TRANSPORT_BACKEND_GRPC, global_internode_metrics, + INTERNODE_OPERATION_GRPC_BATCH_READ_VERSION, INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_READ_VERSION, + INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_STAGE_BATCH_READ_VERSION_DISK_READ, + INTERNODE_STAGE_BATCH_READ_VERSION_REQUEST_DECODE, INTERNODE_STAGE_BATCH_READ_VERSION_RESPONSE_JSON_ENCODE, + INTERNODE_STAGE_BATCH_READ_VERSION_RESPONSE_MSGPACK_ENCODE, INTERNODE_STAGE_READ_VERSION_DISK_READ, + INTERNODE_STAGE_READ_VERSION_REQUEST_DECODE, INTERNODE_STAGE_READ_VERSION_RESPONSE_JSON_ENCODE, + INTERNODE_STAGE_READ_VERSION_RESPONSE_MSGPACK_ENCODE, INTERNODE_TRANSPORT_BACKEND_GRPC, global_internode_metrics, }; use rustfs_protos::proto_gen::node_service::*; use serde::de::DeserializeOwned; @@ -242,24 +244,42 @@ fn record_read_version_stage(stage: &'static str, started_at: Option) { } } +fn record_batch_read_version_stage(stage: &'static str, started_at: Option) { + if let Some(started_at) = started_at { + global_internode_metrics().record_stage_duration_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_BATCH_READ_VERSION, + INTERNODE_TRANSPORT_BACKEND_GRPC, + stage, + started_at.elapsed(), + ); + } +} + fn encode_batch_read_version_response_payloads( batch_read_version_resps: &[BatchReadVersionResp], request_decoded_from_msgpack: bool, ) -> std::result::Result<(Vec, Vec), DiskError> { + let attribution_enabled = rustfs_io_metrics::get_stage_metrics_enabled(); let mut batch_read_version_resps_json = Vec::with_capacity(batch_read_version_resps.len()); - let mut batch_read_version_resps_bin = Vec::with_capacity(batch_read_version_resps.len()); - + let json_encode_started = internode_stage_timer(attribution_enabled); for batch_read_version_resp in batch_read_version_resps { batch_read_version_resps_json.push( compat_response_json(batch_read_version_resp, request_decoded_from_msgpack) .map_err(|err| DiskError::other(format!("encode BatchReadVersionResp json failed: {err}")))?, ); + } + record_batch_read_version_stage(INTERNODE_STAGE_BATCH_READ_VERSION_RESPONSE_JSON_ENCODE, json_encode_started); + + let mut batch_read_version_resps_bin = Vec::with_capacity(batch_read_version_resps.len()); + let msgpack_encode_started = internode_stage_timer(attribution_enabled); + for batch_read_version_resp in batch_read_version_resps { batch_read_version_resps_bin.push(Bytes::from(encode_msgpack_with_capacity( batch_read_version_resp, "BatchReadVersionResp", FILE_INFO_MSGPACK_ENCODE_CAPACITY_HINT, )?)); } + record_batch_read_version_stage(INTERNODE_STAGE_BATCH_READ_VERSION_RESPONSE_MSGPACK_ENCODE, msgpack_encode_started); Ok((batch_read_version_resps_json, batch_read_version_resps_bin)) } @@ -485,15 +505,37 @@ impl NodeService { &self, request: Request, ) -> Result, Status> { + let attribution_enabled = rustfs_io_metrics::get_stage_metrics_enabled(); let request = request.into_inner(); + if attribution_enabled { + let metrics = global_internode_metrics(); + metrics.record_incoming_request_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_BATCH_READ_VERSION, + INTERNODE_TRANSPORT_BACKEND_GRPC, + ); + metrics.record_recv_bytes_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_BATCH_READ_VERSION, + INTERNODE_TRANSPORT_BACKEND_GRPC, + request + .disk + .len() + .saturating_add(request.batch_read_version_req.len()) + .saturating_add(request.batch_read_version_req_bin.len()), + ); + } if let Some(disk) = self.find_disk(&request.disk).await { + let decode_started = internode_stage_timer(attribution_enabled); let decoded_batch_read_version_req: DecodedRpcPayload = match decode_msgpack_or_json_with_source( &request.batch_read_version_req_bin, &request.batch_read_version_req, "BatchReadVersionReq", ) { - Ok(batch_read_version_req) => batch_read_version_req, + Ok(batch_read_version_req) => { + record_batch_read_version_stage(INTERNODE_STAGE_BATCH_READ_VERSION_REQUEST_DECODE, decode_started); + batch_read_version_req + } Err(err) => { + record_batch_read_version_stage(INTERNODE_STAGE_BATCH_READ_VERSION_REQUEST_DECODE, decode_started); return Ok(Response::new(BatchReadVersionResponse { success: false, batch_read_version_resps: Vec::new(), @@ -514,8 +556,10 @@ impl NodeService { })); } + let disk_read_started = internode_stage_timer(attribution_enabled); match disk.batch_read_version(batch_read_version_req).await { Ok(batch_read_version_resps) => { + record_batch_read_version_stage(INTERNODE_STAGE_BATCH_READ_VERSION_DISK_READ, disk_read_started); let (batch_read_version_resps, batch_read_version_resps_bin) = match encode_batch_read_version_response_payloads(&batch_read_version_resps, request_decoded_from_msgpack) { @@ -537,12 +581,15 @@ impl NodeService { error: None, })) } - Err(err) => Ok(Response::new(BatchReadVersionResponse { - success: false, - batch_read_version_resps: Vec::new(), - batch_read_version_resps_bin: Vec::new(), - error: Some(err.into()), - })), + Err(err) => { + record_batch_read_version_stage(INTERNODE_STAGE_BATCH_READ_VERSION_DISK_READ, disk_read_started); + Ok(Response::new(BatchReadVersionResponse { + success: false, + batch_read_version_resps: Vec::new(), + batch_read_version_resps_bin: Vec::new(), + error: Some(err.into()), + })) + } } } else { Ok(Response::new(BatchReadVersionResponse { @@ -722,9 +769,9 @@ impl NodeService { &self, request: Request, ) -> Result, Status> { - let request = request.into_inner(); let metrics = global_internode_metrics(); let read_version_attribution_enabled = rustfs_io_metrics::get_stage_metrics_enabled(); + let request = request.into_inner(); if read_version_attribution_enabled { metrics.record_incoming_request_for_operation_and_backend( INTERNODE_OPERATION_GRPC_READ_VERSION, @@ -1635,6 +1682,7 @@ mod tests { encode_batch_read_version_response_payloads, encode_delete_versions_errors, encode_file_info_msgpack, encode_msgpack, encode_msgpack_named, encode_read_multiple_response_payloads, encode_rename_data_response_payloads, }; + use crate::storage::DiskError; use crate::storage::rpc::node_service::make_server; use crate::storage::storage_api::ReadMultipleResp; use crate::storage::storage_api::RenameDataResp; @@ -2028,7 +2076,8 @@ mod tests { path: "object-a".to_string(), version_id: "version-a".to_string(), success: false, - error: "file version not found".to_string(), + error: DiskError::FileVersionNotFound.to_string(), + error_code: DiskError::FileVersionNotFound.to_u32(), ..Default::default() }]; @@ -2044,8 +2093,43 @@ mod tests { .expect("msgpack batch read version response should decode"); assert_eq!(json_decoded.index, responses[0].index); + assert_eq!(json_decoded.error_code, responses[0].error_code); assert_eq!(msgpack_decoded.path, responses[0].path); assert_eq!(msgpack_decoded.error, responses[0].error); + assert_eq!(msgpack_decoded.error_code, responses[0].error_code); + } + + #[test] + fn batch_read_version_response_decode_accepts_legacy_payload_without_error_code() { + #[derive(Serialize)] + struct LegacyBatchReadVersionResp { + index: usize, + path: String, + version_id: String, + success: bool, + file_info: FileInfo, + error: String, + } + + let legacy = LegacyBatchReadVersionResp { + index: 2, + path: "object-legacy".to_string(), + version_id: "version-legacy".to_string(), + success: false, + file_info: FileInfo::default(), + error: "legacy error".to_string(), + }; + let legacy_json = serde_json::to_string(&legacy).expect("legacy json should encode"); + let legacy_msgpack = encode_msgpack(&legacy, "LegacyBatchReadVersionResp").expect("legacy msgpack should encode"); + + let json_decoded: BatchReadVersionResp = + decode_msgpack_or_json(&[], &legacy_json, "BatchReadVersionResp").expect("legacy json should decode"); + let msgpack_decoded: BatchReadVersionResp = + decode_msgpack_or_json(&legacy_msgpack, "", "BatchReadVersionResp").expect("legacy msgpack should decode"); + + assert_eq!(json_decoded.error_code, 0); + assert_eq!(msgpack_decoded.error_code, 0); + assert_eq!(msgpack_decoded.error, legacy.error); } #[test] diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index e5277c0dd..512e255ee 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -1100,6 +1100,10 @@ pub(crate) fn shutdown_background_monitors() { rustfs_ecstore::shutdown_background_monitors(); } +pub(crate) fn mark_get_metadata_read_version_coalescing_service_ready() { + rustfs_ecstore::mark_get_metadata_read_version_coalescing_service_ready(); +} + pub(crate) fn set_global_rustfs_port(value: u16) { ecstore_global::set_global_rustfs_port(value); } diff --git a/rustfs/src/storage_api.rs b/rustfs/src/storage_api.rs index e90c1f528..dea2923ac 100644 --- a/rustfs/src/storage_api.rs +++ b/rustfs/src/storage_api.rs @@ -284,7 +284,8 @@ pub(crate) mod startup { pub(crate) mod shutdown { pub(crate) use crate::storage::storage_api::{ - shutdown_background_monitors, shutdown_background_services, store_compression_total_in_backend, + mark_get_metadata_read_version_coalescing_service_ready, shutdown_background_monitors, shutdown_background_services, + store_compression_total_in_backend, }; }