From cee0d5cf9bcf072fbc82ad184681317e48a772c4 Mon Sep 17 00:00:00 2001 From: houseme Date: Fri, 21 Aug 2026 00:39:26 +0800 Subject: [PATCH] perf(io-metrics): cache read version metric handles (#6317) Co-authored-by: heihutu --- crates/io-metrics/src/internode_metrics.rs | 202 ++++++++++++++++ .../tests/internode_cached_handles.rs | 216 ++++++++++++++++++ 2 files changed, 418 insertions(+) create mode 100644 crates/io-metrics/tests/internode_cached_handles.rs diff --git a/crates/io-metrics/src/internode_metrics.rs b/crates/io-metrics/src/internode_metrics.rs index fb12a9911..65f93c2a1 100644 --- a/crates/io-metrics/src/internode_metrics.rs +++ b/crates/io-metrics/src/internode_metrics.rs @@ -211,6 +211,146 @@ pub const INTERNODE_OPERATION_METRICS: &[InternodeOperationMetricDescriptor] = & static STABLE_SERVER_LABEL: OnceLock = OnceLock::new(); +#[cfg(not(test))] +struct InternodeServerMetricHandles { + sent_bytes: metrics::Counter, + recv_bytes: metrics::Counter, + outgoing_requests: metrics::Counter, + incoming_requests: metrics::Counter, + errors: metrics::Counter, +} + +#[cfg(not(test))] +impl InternodeServerMetricHandles { + fn new(server: &'static str) -> Self { + Self { + sent_bytes: counter!("rustfs_system_network_internode_sent_bytes_total", SERVER_LABEL => server), + recv_bytes: counter!("rustfs_system_network_internode_recv_bytes_total", SERVER_LABEL => server), + outgoing_requests: counter!("rustfs_system_network_internode_requests_outgoing_total", SERVER_LABEL => server), + incoming_requests: counter!("rustfs_system_network_internode_requests_incoming_total", SERVER_LABEL => server), + errors: counter!("rustfs_system_network_internode_errors_total", SERVER_LABEL => server), + } + } +} + +#[cfg(not(test))] +static INTERNODE_SERVER_METRIC_HANDLES: LazyLock = + LazyLock::new(|| InternodeServerMetricHandles::new(current_server_label())); + +#[cfg(not(test))] +struct GrpcReadVersionMetricHandles { + sent_bytes: metrics::Counter, + recv_bytes: metrics::Counter, + outgoing_requests: metrics::Counter, + incoming_requests: metrics::Counter, + errors: metrics::Counter, + duration: metrics::Histogram, + request_encode: metrics::Histogram, + request_decode: metrics::Histogram, + disk_read: metrics::Histogram, + response_json_encode: metrics::Histogram, + response_msgpack_encode: metrics::Histogram, + rpc_roundtrip: metrics::Histogram, + response_decode: metrics::Histogram, +} + +#[cfg(not(test))] +impl GrpcReadVersionMetricHandles { + fn new(server: &'static str) -> Self { + Self { + sent_bytes: counter!( + INTERNODE_OPERATION_SENT_BYTES_TOTAL, + SERVER_LABEL => server, + OPERATION_LABEL => INTERNODE_OPERATION_GRPC_READ_VERSION, + BACKEND_LABEL => INTERNODE_TRANSPORT_BACKEND_GRPC + ), + recv_bytes: counter!( + INTERNODE_OPERATION_RECV_BYTES_TOTAL, + SERVER_LABEL => server, + OPERATION_LABEL => INTERNODE_OPERATION_GRPC_READ_VERSION, + BACKEND_LABEL => INTERNODE_TRANSPORT_BACKEND_GRPC + ), + outgoing_requests: counter!( + INTERNODE_OPERATION_REQUESTS_OUTGOING_TOTAL, + SERVER_LABEL => server, + OPERATION_LABEL => INTERNODE_OPERATION_GRPC_READ_VERSION, + BACKEND_LABEL => INTERNODE_TRANSPORT_BACKEND_GRPC + ), + incoming_requests: counter!( + INTERNODE_OPERATION_REQUESTS_INCOMING_TOTAL, + SERVER_LABEL => server, + OPERATION_LABEL => INTERNODE_OPERATION_GRPC_READ_VERSION, + BACKEND_LABEL => INTERNODE_TRANSPORT_BACKEND_GRPC + ), + errors: counter!( + INTERNODE_OPERATION_ERRORS_TOTAL, + SERVER_LABEL => server, + OPERATION_LABEL => INTERNODE_OPERATION_GRPC_READ_VERSION, + BACKEND_LABEL => INTERNODE_TRANSPORT_BACKEND_GRPC + ), + duration: metrics::histogram!( + INTERNODE_OPERATION_DURATION_MS, + SERVER_LABEL => server, + OPERATION_LABEL => INTERNODE_OPERATION_GRPC_READ_VERSION, + BACKEND_LABEL => INTERNODE_TRANSPORT_BACKEND_GRPC + ), + request_encode: Self::stage_duration(server, INTERNODE_STAGE_READ_VERSION_REQUEST_ENCODE), + request_decode: Self::stage_duration(server, INTERNODE_STAGE_READ_VERSION_REQUEST_DECODE), + disk_read: Self::stage_duration(server, INTERNODE_STAGE_READ_VERSION_DISK_READ), + response_json_encode: Self::stage_duration(server, INTERNODE_STAGE_READ_VERSION_RESPONSE_JSON_ENCODE), + response_msgpack_encode: Self::stage_duration(server, INTERNODE_STAGE_READ_VERSION_RESPONSE_MSGPACK_ENCODE), + rpc_roundtrip: Self::stage_duration(server, INTERNODE_STAGE_READ_VERSION_RPC_ROUNDTRIP), + response_decode: Self::stage_duration(server, INTERNODE_STAGE_READ_VERSION_RESPONSE_DECODE), + } + } + + fn stage_duration(server: &'static str, stage: &'static str) -> metrics::Histogram { + metrics::histogram!( + INTERNODE_OPERATION_STAGE_DURATION_MS, + SERVER_LABEL => server, + OPERATION_LABEL => INTERNODE_OPERATION_GRPC_READ_VERSION, + BACKEND_LABEL => INTERNODE_TRANSPORT_BACKEND_GRPC, + STAGE_LABEL => stage + ) + } + + fn stage_duration_for(&self, stage: &'static str) -> Option<&metrics::Histogram> { + match stage { + INTERNODE_STAGE_READ_VERSION_REQUEST_ENCODE => Some(&self.request_encode), + INTERNODE_STAGE_READ_VERSION_REQUEST_DECODE => Some(&self.request_decode), + INTERNODE_STAGE_READ_VERSION_DISK_READ => Some(&self.disk_read), + INTERNODE_STAGE_READ_VERSION_RESPONSE_JSON_ENCODE => Some(&self.response_json_encode), + INTERNODE_STAGE_READ_VERSION_RESPONSE_MSGPACK_ENCODE => Some(&self.response_msgpack_encode), + INTERNODE_STAGE_READ_VERSION_RPC_ROUNDTRIP => Some(&self.rpc_roundtrip), + INTERNODE_STAGE_READ_VERSION_RESPONSE_DECODE => Some(&self.response_decode), + _ => None, + } + } +} + +#[cfg(not(test))] +static GRPC_READ_VERSION_METRIC_HANDLES: LazyLock = + LazyLock::new(|| GrpcReadVersionMetricHandles::new(current_server_label())); + +#[cfg(not(test))] +fn server_metric_handles_if_ready() -> Option<&'static InternodeServerMetricHandles> { + STABLE_SERVER_LABEL.get()?; + Some(&INTERNODE_SERVER_METRIC_HANDLES) +} + +#[cfg(not(test))] +fn grpc_read_version_metric_handles_if_ready( + operation: &'static str, + backend: &'static str, +) -> Option<&'static GrpcReadVersionMetricHandles> { + STABLE_SERVER_LABEL.get()?; + if operation == INTERNODE_OPERATION_GRPC_READ_VERSION && backend == INTERNODE_TRANSPORT_BACKEND_GRPC { + Some(&GRPC_READ_VERSION_METRIC_HANDLES) + } else { + None + } +} + /// Injects the stable server label (node name or address) stamped on /// internode metrics. The runtime calls this when the local node name is /// published (see ecstore's `set_local_node_name`); the first write wins. @@ -284,6 +424,11 @@ impl InternodeMetrics { return; } self.sent_bytes_total.fetch_add(bytes, Ordering::Relaxed); + #[cfg(not(test))] + if let Some(handles) = server_metric_handles_if_ready() { + handles.sent_bytes.increment(bytes); + return; + } counter!("rustfs_system_network_internode_sent_bytes_total", SERVER_LABEL => current_server_label()).increment(bytes); } @@ -298,6 +443,11 @@ impl InternodeMetrics { if bytes == 0 { return; } + #[cfg(not(test))] + if let Some(handles) = grpc_read_version_metric_handles_if_ready(operation, backend) { + handles.sent_bytes.increment(bytes); + return; + } counter!( INTERNODE_OPERATION_SENT_BYTES_TOTAL, SERVER_LABEL => current_server_label(), @@ -313,6 +463,11 @@ impl InternodeMetrics { return; } self.recv_bytes_total.fetch_add(bytes, Ordering::Relaxed); + #[cfg(not(test))] + if let Some(handles) = server_metric_handles_if_ready() { + handles.recv_bytes.increment(bytes); + return; + } counter!("rustfs_system_network_internode_recv_bytes_total", SERVER_LABEL => current_server_label()).increment(bytes); } @@ -327,6 +482,11 @@ impl InternodeMetrics { if bytes == 0 { return; } + #[cfg(not(test))] + if let Some(handles) = grpc_read_version_metric_handles_if_ready(operation, backend) { + handles.recv_bytes.increment(bytes); + return; + } counter!( INTERNODE_OPERATION_RECV_BYTES_TOTAL, SERVER_LABEL => current_server_label(), @@ -338,6 +498,11 @@ impl InternodeMetrics { pub fn record_outgoing_request(&self) { self.outgoing_requests_total.fetch_add(1, Ordering::Relaxed); + #[cfg(not(test))] + if let Some(handles) = server_metric_handles_if_ready() { + handles.outgoing_requests.increment(1); + return; + } counter!("rustfs_system_network_internode_requests_outgoing_total", SERVER_LABEL => current_server_label()).increment(1); } @@ -347,6 +512,11 @@ impl InternodeMetrics { pub fn record_outgoing_request_for_operation_and_backend(&self, operation: &'static str, backend: &'static str) { self.record_outgoing_request(); + #[cfg(not(test))] + if let Some(handles) = grpc_read_version_metric_handles_if_ready(operation, backend) { + handles.outgoing_requests.increment(1); + return; + } counter!( INTERNODE_OPERATION_REQUESTS_OUTGOING_TOTAL, SERVER_LABEL => current_server_label(), @@ -358,6 +528,11 @@ impl InternodeMetrics { pub fn record_incoming_request(&self) { self.incoming_requests_total.fetch_add(1, Ordering::Relaxed); + #[cfg(not(test))] + if let Some(handles) = server_metric_handles_if_ready() { + handles.incoming_requests.increment(1); + return; + } counter!("rustfs_system_network_internode_requests_incoming_total", SERVER_LABEL => current_server_label()).increment(1); } @@ -367,6 +542,11 @@ impl InternodeMetrics { pub fn record_incoming_request_for_operation_and_backend(&self, operation: &'static str, backend: &'static str) { self.record_incoming_request(); + #[cfg(not(test))] + if let Some(handles) = grpc_read_version_metric_handles_if_ready(operation, backend) { + handles.incoming_requests.increment(1); + return; + } counter!( INTERNODE_OPERATION_REQUESTS_INCOMING_TOTAL, SERVER_LABEL => current_server_label(), @@ -378,6 +558,11 @@ impl InternodeMetrics { pub fn record_error(&self) { self.errors_total.fetch_add(1, Ordering::Relaxed); + #[cfg(not(test))] + if let Some(handles) = server_metric_handles_if_ready() { + handles.errors.increment(1); + return; + } counter!("rustfs_system_network_internode_errors_total", SERVER_LABEL => current_server_label()).increment(1); } @@ -387,6 +572,11 @@ impl InternodeMetrics { pub fn record_error_for_operation_and_backend(&self, operation: &'static str, backend: &'static str) { self.record_error(); + #[cfg(not(test))] + if let Some(handles) = grpc_read_version_metric_handles_if_ready(operation, backend) { + handles.errors.increment(1); + return; + } counter!( INTERNODE_OPERATION_ERRORS_TOTAL, SERVER_LABEL => current_server_label(), @@ -398,6 +588,11 @@ impl InternodeMetrics { pub fn record_duration_for_operation_and_backend(&self, operation: &'static str, backend: &'static str, duration: Duration) { let duration_ms = duration.as_secs_f64() * 1000.0; + #[cfg(not(test))] + if let Some(handles) = grpc_read_version_metric_handles_if_ready(operation, backend) { + handles.duration.record(duration_ms); + return; + } metrics::histogram!( INTERNODE_OPERATION_DURATION_MS, SERVER_LABEL => current_server_label(), @@ -415,6 +610,13 @@ impl InternodeMetrics { duration: Duration, ) { let duration_ms = duration.as_secs_f64() * 1000.0; + #[cfg(not(test))] + if let Some(handles) = grpc_read_version_metric_handles_if_ready(operation, backend) + && let Some(histogram) = handles.stage_duration_for(stage) + { + histogram.record(duration_ms); + return; + } metrics::histogram!( INTERNODE_OPERATION_STAGE_DURATION_MS, SERVER_LABEL => current_server_label(), diff --git a/crates/io-metrics/tests/internode_cached_handles.rs b/crates/io-metrics/tests/internode_cached_handles.rs new file mode 100644 index 000000000..da7796d9b --- /dev/null +++ b/crates/io-metrics/tests/internode_cached_handles.rs @@ -0,0 +1,216 @@ +// 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. + +use metrics::with_local_recorder; +use metrics_util::debugging::{DebugValue, DebuggingRecorder}; +use rustfs_io_metrics::internode_metrics::{ + INTERNODE_OPERATION_GRPC_READ_VERSION, INTERNODE_STAGE_READ_VERSION_DISK_READ, INTERNODE_STAGE_READ_VERSION_RPC_ROUNDTRIP, + INTERNODE_TRANSPORT_BACKEND_GRPC, InternodeMetrics, set_internode_server_label, +}; +use std::time::Duration; + +type MetricRow = ( + metrics_util::CompositeKey, + Option, + Option, + DebugValue, +); + +const SERVER_LABEL: &str = "server"; +const OPERATION_LABEL: &str = "operation"; +const BACKEND_LABEL: &str = "backend"; +const STAGE_LABEL: &str = "stage"; +const SENT_BYTES_TOTAL: &str = "rustfs_system_network_internode_sent_bytes_total"; +const RECV_BYTES_TOTAL: &str = "rustfs_system_network_internode_recv_bytes_total"; +const REQUESTS_OUTGOING_TOTAL: &str = "rustfs_system_network_internode_requests_outgoing_total"; +const REQUESTS_INCOMING_TOTAL: &str = "rustfs_system_network_internode_requests_incoming_total"; +const ERRORS_TOTAL: &str = "rustfs_system_network_internode_errors_total"; +const OPERATION_SENT_BYTES_TOTAL: &str = "rustfs_system_network_internode_operation_sent_bytes_total"; +const OPERATION_RECV_BYTES_TOTAL: &str = "rustfs_system_network_internode_operation_recv_bytes_total"; +const OPERATION_REQUESTS_OUTGOING_TOTAL: &str = "rustfs_system_network_internode_operation_requests_outgoing_total"; +const OPERATION_REQUESTS_INCOMING_TOTAL: &str = "rustfs_system_network_internode_operation_requests_incoming_total"; +const OPERATION_ERRORS_TOTAL: &str = "rustfs_system_network_internode_operation_errors_total"; +const OPERATION_DURATION_MS: &str = "rustfs_system_network_internode_operation_duration_ms"; +const OPERATION_STAGE_DURATION_MS: &str = "rustfs_system_network_internode_operation_stage_duration_ms"; + +#[test] +fn cached_grpc_read_version_metric_handles_preserve_labels_and_values() { + set_internode_server_label("cached-grpc-read-version-test"); + + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let metrics = InternodeMetrics::default(); + + with_local_recorder(&recorder, || { + metrics.record_sent_bytes_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_READ_VERSION, + INTERNODE_TRANSPORT_BACKEND_GRPC, + 17, + ); + metrics.record_recv_bytes_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_READ_VERSION, + INTERNODE_TRANSPORT_BACKEND_GRPC, + 23, + ); + metrics.record_outgoing_request_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_READ_VERSION, + INTERNODE_TRANSPORT_BACKEND_GRPC, + ); + metrics.record_incoming_request_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_READ_VERSION, + INTERNODE_TRANSPORT_BACKEND_GRPC, + ); + metrics.record_error_for_operation_and_backend(INTERNODE_OPERATION_GRPC_READ_VERSION, INTERNODE_TRANSPORT_BACKEND_GRPC); + metrics.record_duration_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_READ_VERSION, + INTERNODE_TRANSPORT_BACKEND_GRPC, + Duration::from_micros(250), + ); + metrics.record_stage_duration_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_READ_VERSION, + INTERNODE_TRANSPORT_BACKEND_GRPC, + INTERNODE_STAGE_READ_VERSION_RPC_ROUNDTRIP, + Duration::from_micros(125), + ); + metrics.record_stage_duration_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_READ_VERSION, + INTERNODE_TRANSPORT_BACKEND_GRPC, + INTERNODE_STAGE_READ_VERSION_DISK_READ, + Duration::from_micros(75), + ); + }); + + let rows = snapshotter.snapshot().into_vec(); + assert_counter(&rows, SENT_BYTES_TOTAL, &[(SERVER_LABEL, "cached-grpc-read-version-test")], 17); + assert_counter(&rows, RECV_BYTES_TOTAL, &[(SERVER_LABEL, "cached-grpc-read-version-test")], 23); + assert_counter(&rows, REQUESTS_OUTGOING_TOTAL, &[(SERVER_LABEL, "cached-grpc-read-version-test")], 1); + assert_counter(&rows, REQUESTS_INCOMING_TOTAL, &[(SERVER_LABEL, "cached-grpc-read-version-test")], 1); + assert_counter(&rows, ERRORS_TOTAL, &[(SERVER_LABEL, "cached-grpc-read-version-test")], 1); + assert_counter( + &rows, + OPERATION_SENT_BYTES_TOTAL, + &[ + (SERVER_LABEL, "cached-grpc-read-version-test"), + (OPERATION_LABEL, INTERNODE_OPERATION_GRPC_READ_VERSION), + (BACKEND_LABEL, INTERNODE_TRANSPORT_BACKEND_GRPC), + ], + 17, + ); + assert_counter( + &rows, + OPERATION_RECV_BYTES_TOTAL, + &[ + (SERVER_LABEL, "cached-grpc-read-version-test"), + (OPERATION_LABEL, INTERNODE_OPERATION_GRPC_READ_VERSION), + (BACKEND_LABEL, INTERNODE_TRANSPORT_BACKEND_GRPC), + ], + 23, + ); + assert_counter( + &rows, + OPERATION_REQUESTS_OUTGOING_TOTAL, + &[ + (SERVER_LABEL, "cached-grpc-read-version-test"), + (OPERATION_LABEL, INTERNODE_OPERATION_GRPC_READ_VERSION), + (BACKEND_LABEL, INTERNODE_TRANSPORT_BACKEND_GRPC), + ], + 1, + ); + assert_counter( + &rows, + OPERATION_REQUESTS_INCOMING_TOTAL, + &[ + (SERVER_LABEL, "cached-grpc-read-version-test"), + (OPERATION_LABEL, INTERNODE_OPERATION_GRPC_READ_VERSION), + (BACKEND_LABEL, INTERNODE_TRANSPORT_BACKEND_GRPC), + ], + 1, + ); + assert_counter( + &rows, + OPERATION_ERRORS_TOTAL, + &[ + (SERVER_LABEL, "cached-grpc-read-version-test"), + (OPERATION_LABEL, INTERNODE_OPERATION_GRPC_READ_VERSION), + (BACKEND_LABEL, INTERNODE_TRANSPORT_BACKEND_GRPC), + ], + 1, + ); + assert_histogram( + &rows, + OPERATION_DURATION_MS, + &[ + (SERVER_LABEL, "cached-grpc-read-version-test"), + (OPERATION_LABEL, INTERNODE_OPERATION_GRPC_READ_VERSION), + (BACKEND_LABEL, INTERNODE_TRANSPORT_BACKEND_GRPC), + ], + &[0.25], + ); + assert_histogram( + &rows, + OPERATION_STAGE_DURATION_MS, + &[ + (SERVER_LABEL, "cached-grpc-read-version-test"), + (OPERATION_LABEL, INTERNODE_OPERATION_GRPC_READ_VERSION), + (BACKEND_LABEL, INTERNODE_TRANSPORT_BACKEND_GRPC), + (STAGE_LABEL, INTERNODE_STAGE_READ_VERSION_RPC_ROUNDTRIP), + ], + &[0.125], + ); + assert_histogram( + &rows, + OPERATION_STAGE_DURATION_MS, + &[ + (SERVER_LABEL, "cached-grpc-read-version-test"), + (OPERATION_LABEL, INTERNODE_OPERATION_GRPC_READ_VERSION), + (BACKEND_LABEL, INTERNODE_TRANSPORT_BACKEND_GRPC), + (STAGE_LABEL, INTERNODE_STAGE_READ_VERSION_DISK_READ), + ], + &[0.075], + ); +} + +fn assert_counter(rows: &[MetricRow], name: &str, labels: &[(&str, &str)], expected: u64) { + match metric_value(rows, name, labels) { + DebugValue::Counter(value) => assert_eq!(*value, expected), + other => panic!("{name} should be a counter, got {other:?}"), + } +} + +fn assert_histogram(rows: &[MetricRow], name: &str, labels: &[(&str, &str)], expected: &[f64]) { + match metric_value(rows, name, labels) { + DebugValue::Histogram(samples) => { + let actual: Vec<_> = samples.iter().map(|sample| sample.0).collect(); + assert_eq!(actual, expected); + } + other => panic!("{name} should be a histogram, got {other:?}"), + } +} + +fn metric_value<'a>(rows: &'a [MetricRow], name: &str, labels: &[(&str, &str)]) -> &'a DebugValue { + let mut matches = rows.iter().filter(|(composite, _, _, _)| { + composite.key().name() == name + && labels.iter().all(|(key, value)| { + composite + .key() + .labels() + .any(|label| label.key() == *key && label.value() == *value) + }) + }); + let Some((_, _, _, value)) = matches.next() else { + panic!("{name} with labels {labels:?} was not recorded; rows={rows:?}"); + }; + assert!(matches.next().is_none(), "{name} with labels {labels:?} must be unique; rows={rows:?}"); + value +}