diff --git a/crates/e2e_test/src/inline_fast_path_cluster_test.rs b/crates/e2e_test/src/inline_fast_path_cluster_test.rs index b35dbabbb..6e4c606c7 100644 --- a/crates/e2e_test/src/inline_fast_path_cluster_test.rs +++ b/crates/e2e_test/src/inline_fast_path_cluster_test.rs @@ -67,6 +67,7 @@ type MetricValues = Arc>>; const KIB: usize = 1024; const READER_PATH_COUNTER: &str = "rustfs_io_get_object_reader_path_by_size_total"; +const MSGPACK_JSON_DECODE_COUNTER: &str = "rustfs_system_network_internode_msgpack_json_decode_total"; const MSGPACK_JSON_FALLBACK_COUNTER: &str = "rustfs_system_network_internode_msgpack_json_fallback_total"; const MSGPACK_JSON_DECODE_ERROR_COUNTER: &str = "rustfs_system_network_internode_msgpack_json_decode_error_total"; const DIRECTION_LABEL: &str = "direction"; @@ -142,6 +143,7 @@ impl VersionState { struct OtlpMetricCollector { endpoint: String, values: MetricValues, + decode_values: MetricValues, fallback_values: MetricValues, decode_error_values: MetricValues, task: JoinHandle<()>, @@ -152,9 +154,11 @@ impl OtlpMetricCollector { let listener = TcpListener::bind("127.0.0.1:0").await?; let endpoint = format!("http://{}/v1/metrics", listener.local_addr()?); let values = Arc::new(Mutex::new(BTreeMap::new())); + let decode_values = Arc::new(Mutex::new(BTreeMap::new())); let fallback_values = Arc::new(Mutex::new(BTreeMap::new())); let decode_error_values = Arc::new(Mutex::new(BTreeMap::new())); let task_values = values.clone(); + let task_decode_values = decode_values.clone(); let task_fallback_values = fallback_values.clone(); let task_decode_error_values = decode_error_values.clone(); let task = tokio::spawn(async move { @@ -163,6 +167,7 @@ impl OtlpMetricCollector { break; }; let values = task_values.clone(); + let decode_values = task_decode_values.clone(); let fallback_values = task_fallback_values.clone(); let decode_error_values = task_decode_error_values.clone(); tokio::spawn(async move { @@ -173,6 +178,7 @@ impl OtlpMetricCollector { handle_metric_export( request, values.clone(), + decode_values.clone(), fallback_values.clone(), decode_error_values.clone(), ) @@ -185,6 +191,7 @@ impl OtlpMetricCollector { Ok(Self { endpoint, values, + decode_values, fallback_values, decode_error_values, task, @@ -225,6 +232,15 @@ impl OtlpMetricCollector { .collect() } + async fn msgpack_json_decode_totals(&self) -> BTreeMap { + self.decode_values + .lock() + .await + .iter() + .map(|(key, points)| (key.clone(), points.values().map(|(_, value)| value).sum())) + .collect() + } + async fn msgpack_json_decode_error_totals(&self) -> BTreeMap { self.decode_error_values .lock() @@ -302,6 +318,7 @@ impl Drop for OtlpMetricCollector { async fn handle_metric_export( request: Request, values: MetricValues, + decode_values: MetricValues, fallback_values: MetricValues, decode_error_values: MetricValues, ) -> Result>, Infallible> { @@ -334,9 +351,11 @@ async fn handle_metric_export( match ExportMetricsServiceRequest::decode(payload.as_slice()) { Ok(export) => { let mut values = values.lock().await; + let mut decode_values = decode_values.lock().await; let mut fallback_values = fallback_values.lock().await; let mut decode_error_values = decode_error_values.lock().await; record_reader_path_metrics(&export, &mut values); + record_msgpack_decode_metrics(&export, &mut decode_values); record_msgpack_fallback_metrics(&export, &mut fallback_values); record_msgpack_decode_error_metrics(&export, &mut decode_error_values); Ok(response(StatusCode::OK)) @@ -415,6 +434,19 @@ fn msgpack_fallback_metric_key(direction: &str, message: &str) -> String { format!("{direction}\u{1f}{message}") } +fn record_msgpack_decode_metrics(export: &ExportMetricsServiceRequest, values: &mut BTreeMap) { + record_msgpack_counter_metrics(export, MSGPACK_JSON_DECODE_COUNTER, values, |attributes| { + let direction = attribute_string(attributes, DIRECTION_LABEL)?; + let message = attribute_string(attributes, MESSAGE_LABEL)?; + let codec = attribute_string(attributes, CODEC_LABEL)?; + Some(msgpack_decode_metric_key(direction, message, codec)) + }); +} + +fn msgpack_decode_metric_key(direction: &str, message: &str, codec: &str) -> String { + format!("{direction}\u{1f}{message}\u{1f}{codec}") +} + fn record_msgpack_decode_error_metrics(export: &ExportMetricsServiceRequest, values: &mut BTreeMap) { record_msgpack_counter_metrics(export, MSGPACK_JSON_DECODE_ERROR_COUNTER, values, |attributes| { let direction = attribute_string(attributes, DIRECTION_LABEL)?; @@ -488,6 +520,33 @@ async fn assert_msgpack_fallback_unchanged( Ok(()) } +async fn assert_msgpack_decode_observed( + collector: &OtlpMetricCollector, + before: &BTreeMap, + series: &[(&str, &str)], +) -> TestResult { + let deadline = Instant::now() + Duration::from_secs(20); + loop { + let after = collector.msgpack_json_decode_totals().await; + let missing = series + .iter() + .filter_map(|(direction, message)| { + let key = msgpack_decode_metric_key(direction, message, MSGPACK_CODEC_MSGPACK); + let before_value = before.get(&key).copied().unwrap_or_default(); + let after_value = after.get(&key).copied().unwrap_or_default(); + (after_value <= before_value).then_some(format!("{direction}/{message}/{}", MSGPACK_CODEC_MSGPACK)) + }) + .collect::>(); + if missing.is_empty() { + return Ok(()); + } + if Instant::now() >= deadline { + return Err(format!("timed out waiting for msgpack decode traffic on {missing:?}; totals={after:?}").into()); + } + sleep(Duration::from_millis(100)).await; + } +} + async fn assert_msgpack_decode_errors_unchanged( collector: &OtlpMetricCollector, before: &BTreeMap, @@ -521,6 +580,42 @@ fn attribute_string<'a>(attributes: &'a [KeyValue], wanted_key: &str) -> Option< }) } +#[test] +fn records_msgpack_decode_metric_with_codec_label() { + let export = ExportMetricsServiceRequest { + resource_metrics: vec![ResourceMetrics { + scope_metrics: vec![ScopeMetrics { + metrics: vec![Metric { + name: MSGPACK_JSON_DECODE_COUNTER.to_string(), + data: Some(metric::Data::Sum(Sum { + data_points: vec![NumberDataPoint { + attributes: vec![ + metric_attribute(DIRECTION_LABEL, FALLBACK_RESPONSE_DIRECTION), + metric_attribute(MESSAGE_LABEL, "ReadMultipleResp"), + metric_attribute(CODEC_LABEL, MSGPACK_CODEC_MSGPACK), + ], + start_time_unix_nano: 7, + time_unix_nano: 11, + value: Some(number_data_point::Value::AsInt(5)), + ..Default::default() + }], + ..Default::default() + })), + ..Default::default() + }], + ..Default::default() + }], + ..Default::default() + }], + }; + let mut values = BTreeMap::new(); + + record_msgpack_decode_metrics(&export, &mut values); + + let key = msgpack_decode_metric_key(FALLBACK_RESPONSE_DIRECTION, "ReadMultipleResp", MSGPACK_CODEC_MSGPACK); + assert_eq!(values.get(&key).and_then(|points| points.get(&7)).copied(), Some((11, 5))); +} + #[test] fn records_msgpack_decode_error_metric_with_codec_label() { let export = ExportMetricsServiceRequest { @@ -1660,6 +1755,7 @@ async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls() -> Te configure_mixed_msgpack_cluster(&mut cluster, &collector)?; cluster.start().await?; + let decode_before = collector.msgpack_json_decode_totals().await; let fallback_before = collector.msgpack_json_fallback_totals().await; let decode_errors_before = collector.msgpack_json_decode_error_totals().await; @@ -1685,6 +1781,7 @@ async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls() -> Te PartNumberReaderPathExpectation::new(bucket, multipart_key, &second_part, multipart_body.len(), MULTIPART, LEGACY_DUPLEX), ) .await?; + assert_msgpack_decode_observed(&collector, &decode_before, &MSGPACK_FALLBACK_CONTROL_SERIES).await?; assert_msgpack_fallback_unchanged(&collector, &fallback_before, &MSGPACK_FALLBACK_CONTROL_SERIES).await?; assert_msgpack_decode_errors_unchanged(&collector, &decode_errors_before, &MSGPACK_FALLBACK_CONTROL_SERIES).await?; @@ -1862,6 +1959,7 @@ async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls_during_ hot.start().await?; let hot_client = hot.create_s3_client(0)?; + let decode_before = collector.msgpack_json_decode_totals().await; let fallback_before = collector.msgpack_json_fallback_totals().await; let decode_errors_before = collector.msgpack_json_decode_error_totals().await; @@ -1890,6 +1988,7 @@ async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls_during_ PartNumberReaderPathExpectation::new(bucket, key, &second_part, body.len(), REMOTE, REMOTE_TRANSITION), ) .await?; + assert_msgpack_decode_observed(&collector, &decode_before, &MSGPACK_FALLBACK_CONTROL_SERIES).await?; let encrypted_key = "transition/encrypted-sse.bin"; let encrypted_body = payload(16 * KIB, 0xAB); diff --git a/crates/ecstore/src/cluster/rpc/remote_disk.rs b/crates/ecstore/src/cluster/rpc/remote_disk.rs index 583947881..155f7069c 100644 --- a/crates/ecstore/src/cluster/rpc/remote_disk.rs +++ b/crates/ecstore/src/cluster/rpc/remote_disk.rs @@ -1131,19 +1131,31 @@ fn encode_msgpack_named(value: &T) -> Result> { fn decode_msgpack_or_json(binary: &[u8], json: &str, value_name: &'static str) -> Result { if !binary.is_empty() { let mut deserializer = rmp_serde::Deserializer::new(Cursor::new(binary)); - return T::deserialize(&mut deserializer).map_err(|err| { - crate::cluster::rpc::runtime_sources::record_response_msgpack_decode_error(value_name); - Error::from(err) - }); + return match T::deserialize(&mut deserializer) { + Ok(value) => { + crate::cluster::rpc::runtime_sources::record_response_msgpack_decode(value_name); + Ok(value) + } + Err(err) => { + crate::cluster::rpc::runtime_sources::record_response_msgpack_decode_error(value_name); + Err(Error::from(err)) + } + }; } // The msgpack payload was absent, so fall back to the JSON compatibility field. This branch // must read zero across a release window before the redundant JSON fields can be dropped (P2). crate::cluster::rpc::runtime_sources::record_response_json_fallback(value_name); - serde_json::from_str(json).map_err(|err| { - crate::cluster::rpc::runtime_sources::record_response_json_decode_error(value_name); - Error::from(err) - }) + match serde_json::from_str(json) { + Ok(value) => { + crate::cluster::rpc::runtime_sources::record_response_json_decode(value_name); + Ok(value) + } + Err(err) => { + crate::cluster::rpc::runtime_sources::record_response_json_decode_error(value_name); + Err(Error::from(err)) + } + } } /// Aggregate encoded size (bytes) of a `ReadMultiple` response, preferring the msgpack payloads @@ -1195,6 +1207,7 @@ fn decode_read_multiple_response_items(response: ReadMultipleResponse, endpoint: crate::cluster::rpc::runtime_sources::record_response_json_decode_error("ReadMultipleResp"); Error::other(format!("decode ReadMultipleResp json item {index} from {endpoint} failed: {err}")) })?; + crate::cluster::rpc::runtime_sources::record_response_json_decode("ReadMultipleResp"); read_multiple_resps.push(resp); } @@ -1245,6 +1258,7 @@ fn decode_batch_read_version_response_items( crate::cluster::rpc::runtime_sources::record_response_json_decode_error("BatchReadVersionResp"); Error::other(format!("decode BatchReadVersionResp json item {index} from {endpoint} failed: {err}")) })?; + crate::cluster::rpc::runtime_sources::record_response_json_decode("BatchReadVersionResp"); if resp.success { validate_decoded_file_info(&resp.file_info)?; } @@ -3120,12 +3134,15 @@ mod tests { read_multiple_resps_bin: vec![encode_msgpack(&msgpack_resp).expect("msgpack response should encode").into()], error: None, }; + let before = rustfs_io_metrics::internode_metrics::global_internode_metrics().msgpack_json_decode_total_for_test(); let decoded = decode_read_multiple_response_items(response, &endpoint).expect("msgpack response should decode"); + let after = rustfs_io_metrics::internode_metrics::global_internode_metrics().msgpack_json_decode_total_for_test(); assert_eq!(decoded.len(), 1); assert_eq!(decoded[0].file, "msgpack"); assert_eq!(decoded[0].data, b"binary"); + assert!(after > before, "successful response msgpack decode should increment traffic metrics"); } #[test] @@ -3138,12 +3155,15 @@ mod tests { read_multiple_resps_bin: Vec::new(), error: None, }; + let before = rustfs_io_metrics::internode_metrics::global_internode_metrics().msgpack_json_decode_total_for_test(); let decoded = decode_read_multiple_response_items(response, &endpoint).expect("json response should decode"); + let after = rustfs_io_metrics::internode_metrics::global_internode_metrics().msgpack_json_decode_total_for_test(); assert_eq!(decoded.len(), 1); assert_eq!(decoded[0].file, "json"); assert_eq!(decoded[0].data, b"fallback"); + assert!(after > before, "successful response JSON decode should increment traffic metrics"); } #[test] diff --git a/crates/ecstore/src/cluster/rpc/runtime_sources.rs b/crates/ecstore/src/cluster/rpc/runtime_sources.rs index 9b147422e..03d19f60f 100644 --- a/crates/ecstore/src/cluster/rpc/runtime_sources.rs +++ b/crates/ecstore/src/cluster/rpc/runtime_sources.rs @@ -127,6 +127,22 @@ pub(crate) fn record_response_json_fallback(message: &'static str) { global_internode_metrics().record_msgpack_json_fallback(INTERNODE_MSGPACK_DIRECTION_RESPONSE, message); } +pub(crate) fn record_response_msgpack_decode(message: &'static str) { + global_internode_metrics().record_msgpack_json_decode( + INTERNODE_MSGPACK_DIRECTION_RESPONSE, + message, + INTERNODE_MSGPACK_CODEC_MSGPACK, + ); +} + +pub(crate) fn record_response_json_decode(message: &'static str) { + global_internode_metrics().record_msgpack_json_decode( + INTERNODE_MSGPACK_DIRECTION_RESPONSE, + message, + INTERNODE_MSGPACK_CODEC_JSON, + ); +} + pub(crate) fn record_response_msgpack_decode_error(message: &'static str) { global_internode_metrics().record_msgpack_json_decode_error( INTERNODE_MSGPACK_DIRECTION_RESPONSE, diff --git a/crates/io-metrics/src/internode_metrics.rs b/crates/io-metrics/src/internode_metrics.rs index efd65d09c..0790c2c8b 100644 --- a/crates/io-metrics/src/internode_metrics.rs +++ b/crates/io-metrics/src/internode_metrics.rs @@ -62,6 +62,7 @@ const INTERNODE_OPERATION_WRITE_SHUTDOWN_ERRORS_TOTAL: &str = "rustfs_system_network_internode_operation_write_shutdown_errors_total"; const INTERNODE_OPERATION_PAYLOAD_BYTES: &str = "rustfs_system_network_internode_operation_payload_bytes"; const INTERNODE_OPERATION_LARGE_PAYLOADS_TOTAL: &str = "rustfs_system_network_internode_operation_large_payloads_total"; +const INTERNODE_MSGPACK_JSON_DECODE_TOTAL: &str = "rustfs_system_network_internode_msgpack_json_decode_total"; const INTERNODE_MSGPACK_JSON_FALLBACK_TOTAL: &str = "rustfs_system_network_internode_msgpack_json_fallback_total"; const INTERNODE_MSGPACK_JSON_DECODE_ERROR_TOTAL: &str = "rustfs_system_network_internode_msgpack_json_decode_error_total"; const INTERNODE_SIGNATURE_V1_FALLBACK_TOTAL: &str = "rustfs_system_network_internode_signature_v1_fallback_total"; @@ -175,6 +176,7 @@ pub struct InternodeMetrics { operation_http_versions_total: AtomicU64, operation_stall_timeouts_total: AtomicU64, operation_write_shutdown_errors_total: AtomicU64, + msgpack_json_decode_total: AtomicU64, msgpack_json_decode_error_total: AtomicU64, signature_v1_fallback_total: AtomicU64, body_digest_fallback_total: AtomicU64, @@ -375,6 +377,17 @@ impl InternodeMetrics { counter!(INTERNODE_MSGPACK_JSON_FALLBACK_TOTAL, DIRECTION_LABEL => direction, MESSAGE_LABEL => message).increment(1); } + pub fn record_msgpack_json_decode(&self, direction: &'static str, message: &'static str, codec: &'static str) { + self.msgpack_json_decode_total.fetch_add(1, Ordering::Relaxed); + counter!( + INTERNODE_MSGPACK_JSON_DECODE_TOTAL, + DIRECTION_LABEL => direction, + MESSAGE_LABEL => message, + CODEC_LABEL => codec + ) + .increment(1); + } + pub fn record_msgpack_json_decode_error(&self, direction: &'static str, message: &'static str, codec: &'static str) { self.msgpack_json_decode_error_total.fetch_add(1, Ordering::Relaxed); counter!( @@ -391,6 +404,11 @@ impl InternodeMetrics { self.msgpack_json_decode_error_total.load(Ordering::Relaxed) } + #[doc(hidden)] + pub fn msgpack_json_decode_total_for_test(&self) -> u64 { + self.msgpack_json_decode_total.load(Ordering::Relaxed) + } + /// Count an internode gRPC request that was accepted through the legacy constant-target /// signature because it carried no v2 auth headers (rolling-upgrade fallback, see /// ). Only accepted requests count: rejected @@ -488,6 +506,7 @@ impl InternodeMetrics { self.operation_http_versions_total.store(0, Ordering::Relaxed); self.operation_stall_timeouts_total.store(0, Ordering::Relaxed); self.operation_write_shutdown_errors_total.store(0, Ordering::Relaxed); + self.msgpack_json_decode_total.store(0, Ordering::Relaxed); self.msgpack_json_decode_error_total.store(0, Ordering::Relaxed); self.signature_v1_fallback_total.store(0, Ordering::Relaxed); self.body_digest_fallback_total.store(0, Ordering::Relaxed); @@ -790,6 +809,10 @@ mod tests { "rustfs_system_network_internode_operation_large_payloads_total" ); assert_eq!(INTERNODE_OPERATION_GRPC_READ_MULTIPLE, "grpc_read_multiple"); + assert_eq!( + INTERNODE_MSGPACK_JSON_DECODE_TOTAL, + "rustfs_system_network_internode_msgpack_json_decode_total" + ); assert_eq!( INTERNODE_MSGPACK_JSON_FALLBACK_TOTAL, "rustfs_system_network_internode_msgpack_json_fallback_total" @@ -816,6 +839,19 @@ mod tests { metrics.record_msgpack_json_fallback(INTERNODE_MSGPACK_DIRECTION_RESPONSE, "RawFileInfo"); } + #[test] + fn msgpack_json_decode_counter_tracks_codec_direction_and_message() { + let metrics = InternodeMetrics::default(); + assert_eq!(metrics.msgpack_json_decode_total_for_test(), 0); + + metrics.record_msgpack_json_decode(INTERNODE_MSGPACK_DIRECTION_REQUEST, "FileInfo", INTERNODE_MSGPACK_CODEC_MSGPACK); + metrics.record_msgpack_json_decode(INTERNODE_MSGPACK_DIRECTION_RESPONSE, "RawFileInfo", INTERNODE_MSGPACK_CODEC_JSON); + + assert_eq!(metrics.msgpack_json_decode_total_for_test(), 2); + metrics.reset_for_test(); + assert_eq!(metrics.msgpack_json_decode_total_for_test(), 0); + } + #[test] fn msgpack_json_decode_error_counter_tracks_codec_direction_and_message() { let metrics = InternodeMetrics::default(); diff --git a/docs/operations/internode-grpc-benchmark-runbook.md b/docs/operations/internode-grpc-benchmark-runbook.md index d73ea1230..5615bc1b7 100644 --- a/docs/operations/internode-grpc-benchmark-runbook.md +++ b/docs/operations/internode-grpc-benchmark-runbook.md @@ -51,6 +51,7 @@ gitignored — attach the paired directories to the PR / issue. | `rustfs_system_network_internode_operation_payload_bytes` | payload size distribution (P0/P1 sizing) | | `rustfs_system_network_internode_operation_large_payloads_total` | large unary RPCs sharing the channel (P1 target) | | `rustfs_system_network_internode_dial_avg_time_nanos`, `..._dial_errors_total` | connect cost (P3 prewarm) | +| `rustfs_system_network_internode_msgpack_json_decode_total{direction,message,codec}` | must be **>0** for each expected P2 series before zero fallback/error readings are meaningful | | `rustfs_system_network_internode_msgpack_json_fallback_total{direction,message}` | must be **0** before enabling both msgpack-only gates (P2) | | `rustfs_system_network_internode_msgpack_json_decode_error_total{direction,message,codec}` | must be **0** before enabling both msgpack-only gates (P2) | | `rustfs_cluster_servers_offline_total` | offline detection correctness (P3 bypass) | @@ -89,7 +90,7 @@ column, everything else at defaults. Roll a restart between runs. `>4 MiB` multi-version `xl.meta` no longer fails `out_of_range`. - **P1** — mixed workload (large `ReadAll` + high-frequency `Refresh`). Acceptance gate from the design doc: **lock p99 down ≥ 20%** with `RUSTFS_INTERNODE_CHANNEL_ISOLATION=true`. -- **P2** — observe `msgpack_json_fallback_total` and `msgpack_json_decode_error_total` across a release window; both must stay **0** before flipping both `RUSTFS_INTERNODE_RPC_MSGPACK_ONLY=true` and `RUSTFS_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED=true` (see the msgpack convergence runbook). Codec allocation via a `dhat`/`heaptrack` micro-run. +- **P2** — observe `msgpack_json_decode_total`, `msgpack_json_fallback_total`, and `msgpack_json_decode_error_total` across a release window; every expected `codec="msgpack"` decode series must be **>0**, while fallback and decode-error series must stay **0** before flipping both `RUSTFS_INTERNODE_RPC_MSGPACK_ONLY=true` and `RUSTFS_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED=true` (see the msgpack convergence runbook). Codec allocation via a `dhat`/`heaptrack` micro-run. - **P3** — cold-start: first cross-node op latency should drop ~one connect RTT with prewarm. Failover: `run_four_node_cluster_failover_bench.sh`, kill a node with `RUSTFS_INTERNODE_OFFLINE_BYPASS=true`; expect faster failover and a correct @@ -107,9 +108,7 @@ the gate verdict (pass/fail + measured delta) in the paired directory's `summary | **P3 cold-start** | `run_internode_transport_baseline.sh` (fresh cluster, first cross-node op) | first cross-node op latency **drops ~one connect RTT** with `RUSTFS_INTERNODE_PREWARM=true`. | `..._dial_avg_time_nanos`, first-op `..._operation_duration_ms` | | **P3 offline** | dedicated *sustained-offline + survivor cross-node access* experiment (the standard four-node failover bench is **not** sensitive to the bypass — quorum holds, `recovery_seconds=0`) | with `RUSTFS_INTERNODE_OFFLINE_BYPASS=true`, survivor cross-node op latency to the downed peer **fast-fails** instead of hanging the dial timeout; `rustfs_cluster_servers_offline_total` = **1** while down, back to **0** after recovery. | `rustfs_cluster_servers_offline_total`, survivor cross-node `..._operation_duration_ms`, `..._dial_errors_total` | -> **P2 is not a throughput gate.** Its acceptance is operational: `msgpack_json_fallback_total` -> must read **0** across a full release window before both msgpack-only env gates are enabled (see -> the msgpack convergence runbook). Do not benchmark P2 as before/after throughput. +> **P2 is not a throughput gate.** Its acceptance is operational: every expected `msgpack_json_decode_total{codec="msgpack"}` series must have non-zero traffic, while `msgpack_json_fallback_total` and `msgpack_json_decode_error_total` must read **0** across a full release window before both msgpack-only env gates are enabled (see the msgpack convergence runbook). Do not benchmark P2 as before/after throughput. Artifact layout per stage (attach both halves + the diff): diff --git a/docs/operations/internode-msgpack-json-convergence-runbook.md b/docs/operations/internode-msgpack-json-convergence-runbook.md index 461b10ef5..927475b23 100644 --- a/docs/operations/internode-msgpack-json-convergence-runbook.md +++ b/docs/operations/internode-msgpack-json-convergence-runbook.md @@ -20,11 +20,11 @@ otherwise a rolling upgrade with mixed node versions could read an emptied field ``` rustfs_system_network_internode_msgpack_json_fallback_total{direction, message} +rustfs_system_network_internode_msgpack_json_decode_total{direction, message, codec} rustfs_system_network_internode_msgpack_json_decode_error_total{direction, message, codec} ``` -Incremented whenever a decode falls back to the JSON field because the msgpack payload was -absent. +The decode counter increments after a successful msgpack or JSON compatibility decode. The fallback counter increments whenever a decode falls back to the JSON field because the msgpack payload was absent. The decode-error counter increments whenever either codec fails to decode. - `direction="request"` — a server decoding a peer's request (`node_service/disk.rs`). - `direction="response"` — a client decoding a peer's response (`cluster/rpc/remote_disk.rs`), @@ -34,8 +34,18 @@ absent. ## Stage 0 — Observe (current stage) -Ship the current release (which contains the counter) and let it run for **at least one -full release window** across the whole fleet. The counter must stay at **zero**. +Ship the current release (which contains these counters) and let it run for **at least one +full release window** across the whole fleet. The fallback and decode-error counters must stay at **zero**. + +First confirm each expected message/direction has real traffic in the observation window: + +```promql +sum by (direction, message, codec) ( + increase(rustfs_system_network_internode_msgpack_json_decode_total[30d]) +) +``` + +For every convergence-ready message/direction below, the `codec="msgpack"` series must be non-zero before a zero fallback result is meaningful. Missing series, zero traffic, counter reset, or scrape gaps make the gate inconclusive rather than passed. Confirm zero across the observation window (adjust `[30d]` to the window length): diff --git a/rustfs/src/storage/rpc/node_service/disk.rs b/rustfs/src/storage/rpc/node_service/disk.rs index a8ad7c603..fef403ed0 100644 --- a/rustfs/src/storage/rpc/node_service/disk.rs +++ b/rustfs/src/storage/rpc/node_service/disk.rs @@ -43,27 +43,47 @@ fn decode_msgpack_or_json( ) -> std::result::Result { if !binary.is_empty() { let mut deserializer = rmp_serde::Deserializer::new(Cursor::new(binary)); - return T::deserialize(&mut deserializer).map_err(|err| { - global_internode_metrics().record_msgpack_json_decode_error( - INTERNODE_MSGPACK_DIRECTION_REQUEST, - value_name, - INTERNODE_MSGPACK_CODEC_MSGPACK, - ); - DiskError::other(format!("decode {value_name} msgpack failed: {err}")) - }); + return match T::deserialize(&mut deserializer) { + Ok(value) => { + global_internode_metrics().record_msgpack_json_decode( + INTERNODE_MSGPACK_DIRECTION_REQUEST, + value_name, + INTERNODE_MSGPACK_CODEC_MSGPACK, + ); + Ok(value) + } + Err(err) => { + global_internode_metrics().record_msgpack_json_decode_error( + INTERNODE_MSGPACK_DIRECTION_REQUEST, + value_name, + INTERNODE_MSGPACK_CODEC_MSGPACK, + ); + Err(DiskError::other(format!("decode {value_name} msgpack failed: {err}"))) + } + }; } // The msgpack payload was absent, so fall back to the JSON compatibility field. This branch // must read zero across a release window before the redundant JSON fields can be dropped (P2). global_internode_metrics().record_msgpack_json_fallback(INTERNODE_MSGPACK_DIRECTION_REQUEST, value_name); - serde_json::from_str(json).map_err(|err| { - global_internode_metrics().record_msgpack_json_decode_error( - INTERNODE_MSGPACK_DIRECTION_REQUEST, - value_name, - INTERNODE_MSGPACK_CODEC_JSON, - ); - DiskError::other(format!("decode {value_name} failed: {err}")) - }) + match serde_json::from_str(json) { + Ok(value) => { + global_internode_metrics().record_msgpack_json_decode( + INTERNODE_MSGPACK_DIRECTION_REQUEST, + value_name, + INTERNODE_MSGPACK_CODEC_JSON, + ); + Ok(value) + } + Err(err) => { + global_internode_metrics().record_msgpack_json_decode_error( + INTERNODE_MSGPACK_DIRECTION_REQUEST, + value_name, + INTERNODE_MSGPACK_CODEC_JSON, + ); + Err(DiskError::other(format!("decode {value_name} failed: {err}"))) + } + } } fn encode_msgpack(value: &T, value_name: &str) -> std::result::Result, DiskError> { @@ -1298,15 +1318,20 @@ mod tests { }; let binary = encode_msgpack(&payload, "SamplePayload").unwrap(); + let before = global_internode_metrics().msgpack_json_decode_total_for_test(); let decoded = decode_msgpack_or_json::(&binary, r#"{"name":"ignored","count":1}"#, "SamplePayload").unwrap(); + let after = global_internode_metrics().msgpack_json_decode_total_for_test(); assert_eq!(decoded, payload); + assert!(after > before, "successful request msgpack decode should increment traffic metrics"); } #[test] fn decode_msgpack_or_json_falls_back_to_json() { + let before = global_internode_metrics().msgpack_json_decode_total_for_test(); let decoded = decode_msgpack_or_json::(&[], r#"{"name":"compat","count":7}"#, "SamplePayload").unwrap(); + let after = global_internode_metrics().msgpack_json_decode_total_for_test(); assert_eq!( decoded, @@ -1315,6 +1340,7 @@ mod tests { count: 7, } ); + assert!(after > before, "successful request JSON fallback decode should increment traffic metrics"); } fn with_internode_msgpack_env(vars: [(&'static str, Option<&'static str>); 2], f: impl FnOnce() -> R) -> R {