diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index c40e8a83a..b2d2f79c7 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -3599,7 +3599,6 @@ impl ECStore { if should_save_pool_meta { self.ctx.advance_data_movement_operation_epoch(); } - drop(_movement_guard); if should_reload_pool_meta && let Some(notification_sys) = runtime_sources::notification_sys() { let stage = format!("decommission_cancel for pool {idx}"); diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index a7310b44f..0c743a4cb 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -71,6 +71,9 @@ use crate::set_disk::shard_source::ShardReadCost; use futures::FutureExt as _; use futures::stream::{FuturesUnordered, StreamExt}; use metrics::counter; +use rustfs_io_metrics::internode_metrics::{ + INTERNODE_STAGE_BATCH_READ_VERSION_COALESCER_WAIT, INTERNODE_STAGE_BATCH_READ_VERSION_RESPONSE_MAP, +}; use std::{ collections::{HashMap, HashSet, VecDeque}, future::Future, @@ -193,6 +196,16 @@ fn record_read_version_coalescer_event(event: &'static str, item_count: usize) { .increment(1); } +fn batch_read_version_stage_timer() -> Option { + rustfs_io_metrics::get_stage_metrics_enabled().then(Instant::now) +} + +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()); + } +} + async fn read_version_via_coalescer( disk: DiskStore, org_bucket: &str, @@ -242,8 +255,12 @@ async fn read_version_via_coalescer( 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"))) + let wait_started = batch_read_version_stage_timer(); + let response = rx + .await + .unwrap_or_else(|_| Err(DiskError::other("coalesced read_version response channel closed"))); + record_batch_read_version_stage(INTERNODE_STAGE_BATCH_READ_VERSION_COALESCER_WAIT, wait_started); + response } async fn flush_read_version_coalescer_lane(lane_key: ReadVersionCoalescerKey, disk: DiskStore, opts: ReadOptions) { @@ -292,7 +309,9 @@ async fn flush_read_version_coalescer_pending( }; match result { Ok(responses) => { + let map_started = batch_read_version_stage_timer(); let results = map_batch_read_version_responses(&expected_items, responses); + record_batch_read_version_stage(INTERNODE_STAGE_BATCH_READ_VERSION_RESPONSE_MAP, map_started); for (tx, result) in senders.into_iter().zip(results) { let _ = tx.send(result); } diff --git a/crates/io-metrics/src/internode_metrics.rs b/crates/io-metrics/src/internode_metrics.rs index 6b1ea382c..24709b82a 100644 --- a/crates/io-metrics/src/internode_metrics.rs +++ b/crates/io-metrics/src/internode_metrics.rs @@ -61,6 +61,8 @@ pub const INTERNODE_STAGE_BATCH_READ_VERSION_RESPONSE_JSON_ENCODE: &str = "batch 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"; +pub const INTERNODE_STAGE_BATCH_READ_VERSION_COALESCER_WAIT: &str = "batch_read_version_coalescer_wait"; +pub const INTERNODE_STAGE_BATCH_READ_VERSION_RESPONSE_MAP: &str = "batch_read_version_response_map"; const OPERATION_LABEL: &str = "operation"; const BACKEND_LABEL: &str = "backend"; @@ -1431,6 +1433,8 @@ mod tests { ); assert_eq!(INTERNODE_STAGE_READ_VERSION_RPC_ROUNDTRIP, "read_version_rpc_roundtrip"); assert_eq!(INTERNODE_STAGE_READ_VERSION_RESPONSE_DECODE, "read_version_response_decode"); + assert_eq!(INTERNODE_STAGE_BATCH_READ_VERSION_COALESCER_WAIT, "batch_read_version_coalescer_wait"); + assert_eq!(INTERNODE_STAGE_BATCH_READ_VERSION_RESPONSE_MAP, "batch_read_version_response_map"); assert_eq!( INTERNODE_SIGNATURE_V1_FALLBACK_TOTAL, "rustfs_system_network_internode_signature_v1_fallback_total"