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 6efcf574b..72ed10517 100644 --- a/crates/e2e_test/src/inline_fast_path_cluster_test.rs +++ b/crates/e2e_test/src/inline_fast_path_cluster_test.rs @@ -633,6 +633,35 @@ struct ReaderPathExpectation<'a> { expected_size_bucket: Option<&'a str>, } +struct PartNumberReaderPathExpectation<'a> { + bucket: &'a str, + key: &'a str, + expected_part: &'a [u8], + full_object_size: usize, + object_class: &'a str, + expected_path: &'a str, +} + +impl<'a> PartNumberReaderPathExpectation<'a> { + fn new( + bucket: &'a str, + key: &'a str, + expected_part: &'a [u8], + full_object_size: usize, + object_class: &'a str, + expected_path: &'a str, + ) -> Self { + Self { + bucket, + key, + expected_part, + full_object_size, + object_class, + expected_path, + } + } +} + impl<'a> ReaderPathExpectation<'a> { fn plain(object: ReaderObject<'a>, expected_path: &'a str) -> Self { Self::for_class(object, expected_path, PLAIN_SINGLE_PART) @@ -804,17 +833,21 @@ async fn assert_reader_path( async fn assert_part_number_reader_path( collector: &OtlpMetricCollector, client: &Client, - bucket: &str, - key: &str, - expected_part: &[u8], - full_object_size: usize, - expected_path: &str, + expectation: PartNumberReaderPathExpectation<'_>, ) -> TestResult { + let PartNumberReaderPathExpectation { + bucket, + key, + expected_part, + full_object_size, + object_class, + expected_path, + } = expectation; let paths = [INLINE_DIRECT, LEGACY_DUPLEX, EMPTY, REMOTE_TRANSITION]; let size_bucket = size_bucket(full_object_size); let mut before = BTreeMap::<&str, u64>::new(); for path in paths { - before.insert(path, collector.reader_path_total(path, MULTIPART, size_bucket).await); + before.insert(path, collector.reader_path_total(path, object_class, size_bucket).await); } let response = client.get_object().bucket(bucket).key(key).part_number(2).send().await?; assert_eq!( @@ -825,17 +858,17 @@ async fn assert_part_number_reader_path( let body = response.body.collect().await?.into_bytes(); assert_eq!(body.as_ref(), expected_part, "partNumber GET body changed for {key}"); collector - .wait_for_reader_path_total(expected_path, MULTIPART, size_bucket, before[expected_path] + 1) + .wait_for_reader_path_total(expected_path, object_class, size_bucket, before[expected_path] + 1) .await?; - collector.wait_for_reader_paths_to_settle(MULTIPART, size_bucket).await?; + collector.wait_for_reader_paths_to_settle(object_class, size_bucket).await?; for path in paths { if path == expected_path { continue; } assert_eq!( - collector.reader_path_total(path, MULTIPART, size_bucket).await, + collector.reader_path_total(path, object_class, size_bucket).await, before[path], - "{READER_PATH_COUNTER}{{path={path}, object_class={MULTIPART}, size_bucket={size_bucket}}} must not advance for partNumber GET {key}; expected {expected_path} only" + "{READER_PATH_COUNTER}{{path={path}, object_class={object_class}, size_bucket={size_bucket}}} must not advance for partNumber GET {key}; expected {expected_path} only" ); } Ok(()) @@ -1286,11 +1319,7 @@ async fn four_node_inline_fallback_controls() -> TestResult { assert_part_number_reader_path( &collector, &client, - bucket, - multipart_key, - &second_part, - multipart_body.len(), - LEGACY_DUPLEX, + PartNumberReaderPathExpectation::new(bucket, multipart_key, &second_part, multipart_body.len(), MULTIPART, LEGACY_DUPLEX), ) .await?; @@ -1385,7 +1414,12 @@ async fn four_node_multipart_ignores_disk_compression_fallback() -> TestResult { ReaderPathExpectation::for_class(ReaderObject::new(bucket, key, &body, etag.as_deref(), None), LEGACY_DUPLEX, MULTIPART), ) .await?; - assert_part_number_reader_path(&collector, &client, bucket, key, &second_part, body.len(), LEGACY_DUPLEX).await?; + assert_part_number_reader_path( + &collector, + &client, + PartNumberReaderPathExpectation::new(bucket, key, &second_part, body.len(), MULTIPART, LEGACY_DUPLEX), + ) + .await?; Ok(()) } @@ -1424,11 +1458,7 @@ async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls() -> Te assert_part_number_reader_path( &collector, &client, - bucket, - multipart_key, - &second_part, - multipart_body.len(), - LEGACY_DUPLEX, + PartNumberReaderPathExpectation::new(bucket, multipart_key, &second_part, multipart_body.len(), MULTIPART, LEGACY_DUPLEX), ) .await?; assert_msgpack_fallback_unchanged(&collector, &fallback_before, &MSGPACK_FALLBACK_CONTROL_SERIES).await?; @@ -1517,7 +1547,7 @@ async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls_during_ put_lifecycle_with_transition_retry(&hot_client, bucket).await?; let key = "transition/mixed-multipart.bin"; - let (body, _, etag) = put_two_part_multipart(&hot_client, bucket, key).await?; + let (body, second_part, etag) = put_two_part_multipart(&hot_client, bucket, key).await?; wait_for_transition(&hot_client, bucket, key).await?; assert!( cold_tier_object_count(&cold_client).await? >= 1, @@ -1529,6 +1559,12 @@ async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls_during_ ReaderPathExpectation::for_class(ReaderObject::new(bucket, key, &body, etag.as_deref(), None), REMOTE_TRANSITION, REMOTE), ) .await?; + assert_part_number_reader_path( + &collector, + &hot_client, + PartNumberReaderPathExpectation::new(bucket, key, &second_part, body.len(), REMOTE, REMOTE_TRANSITION), + ) + .await?; let encrypted_key = "transition/encrypted-sse.bin"; let encrypted_body = payload(16 * KIB, 0xAB); diff --git a/crates/e2e_test/src/reliant/tiering.rs b/crates/e2e_test/src/reliant/tiering.rs index 6c7ec1371..3346e344e 100644 --- a/crates/e2e_test/src/reliant/tiering.rs +++ b/crates/e2e_test/src/reliant/tiering.rs @@ -70,11 +70,13 @@ const MANUAL_DUE_BUCKET: &str = "ilm7-manual-due"; const MANUAL_DRY_RUN_BUCKET: &str = "ilm7-manual-dry-run"; const MANUAL_NOT_DUE_BUCKET: &str = "ilm7-manual-not-due"; const MANUAL_QUEUE_PRESSURE_BUCKET: &str = "ilm7-manual-queue-pressure"; +const MANUAL_ASYNC_STATUS_BUCKET: &str = "ilm7-manual-async-status"; const MANUAL_QUEUE_PRESSURE_PREFIX: &str = "manual-queue-pressure/"; const OBJECT_KEY: &str = "tier/鲁A12345/report.bin"; const MANUAL_DUE_KEY: &str = "manual-due/report.bin"; const MANUAL_DRY_RUN_KEY: &str = "manual-dry-run/report.bin"; const MANUAL_NOT_DUE_KEY: &str = "manual-not-due/report.bin"; +const MANUAL_ASYNC_STATUS_KEY: &str = "manual-async-status/report.bin"; const CONTENT_TYPE: &str = "application/x-ilm7"; const USER_META_KEY: &str = "ilm7-origin"; const USER_META_VAL: &str = "hermetic-transition"; @@ -350,6 +352,16 @@ struct ManualTransitionRunReport { truncated_by_duration: bool, } +#[derive(Debug, Deserialize)] +struct ManualTransitionJobStatusResponse { + job_id: String, + status_endpoint: String, + status: String, + cancel_requested: bool, + failure_reason: Option, + report: ManualTransitionRunReport, +} + async fn manual_transition_run( hot: &RustFSTestEnvironment, bucket: &str, @@ -379,6 +391,66 @@ async fn manual_transition_run_with_max( Ok(serde_json::from_str(&body)?) } +async fn manual_transition_async_run( + hot: &RustFSTestEnvironment, + bucket: &str, + prefix: &str, + dry_run: bool, + max_objects: u64, +) -> Result> { + let bucket = urlencoding::encode(bucket); + let prefix = urlencoding::encode(prefix); + let tier = urlencoding::encode(TIER_NAME); + let path = format!( + "/rustfs/admin/v3/ilm/transition/run?bucket={bucket}&prefix={prefix}&tier={tier}&dryRun={dry_run}&maxObjects={max_objects}&mode=async" + ); + let (status, body) = signed_admin_request(&hot.url, Method::POST, &path, None, &hot.access_key, &hot.secret_key).await?; + assert_eq!(status, reqwest::StatusCode::ACCEPTED, "async manual transition response: {body}"); + Ok(serde_json::from_str(&body)?) +} + +async fn manual_transition_job_status( + hot: &RustFSTestEnvironment, + status_endpoint: &str, +) -> Result> { + let (status, body) = + signed_admin_request(&hot.url, Method::GET, status_endpoint, None, &hot.access_key, &hot.secret_key).await?; + assert_eq!(status, reqwest::StatusCode::OK, "manual transition job status response: {body}"); + Ok(serde_json::from_str(&body)?) +} + +async fn manual_transition_job_cancel( + hot: &RustFSTestEnvironment, + status_endpoint: &str, +) -> Result> { + let (status, body) = + signed_admin_request(&hot.url, Method::DELETE, status_endpoint, None, &hot.access_key, &hot.secret_key).await?; + assert_eq!(status, reqwest::StatusCode::OK, "manual transition job cancel response: {body}"); + Ok(serde_json::from_str(&body)?) +} + +async fn wait_for_manual_transition_job_terminal( + hot: &RustFSTestEnvironment, + status_endpoint: &str, + deadline: StdDuration, +) -> Result> { + let start = Instant::now(); + loop { + let status = manual_transition_job_status(hot, status_endpoint).await?; + if matches!(status.status.as_str(), "completed" | "partial" | "cancelled" | "failed" | "unknown") { + return Ok(status); + } + if start.elapsed() >= deadline { + return Err(format!( + "manual transition job at {status_endpoint} did not reach a terminal state within {}s; last={status:#?}", + deadline.as_secs() + ) + .into()); + } + tokio::time::sleep(StdDuration::from_millis(250)).await; + } +} + /// Number of objects currently stored in the cold-tier bucket. async fn cold_tier_object_count(cold_client: &Client) -> Result> { let resp = cold_client.list_objects_v2().bucket(TIER_BUCKET).send().await?; @@ -661,6 +733,86 @@ async fn test_manual_transition_run_black_box_semantics() -> TestResult { Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn test_manual_transition_async_job_status_polling() -> TestResult { + let mut cold = RustFSTestEnvironment::new().await?; + cold.access_key = "manualasynccoldtieradmin".to_string(); + cold.secret_key = "manualasynccoldtiersecret".to_string(); + cold.start_rustfs_server_without_cleanup(vec![]).await?; + let cold_client = cold.create_s3_client(); + cold_client.create_bucket().bucket(TIER_BUCKET).send().await?; + + let mut hot = RustFSTestEnvironment::new().await?; + hot.start_rustfs_server_with_env(vec![], &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")]) + .await?; + let hot_client = hot.create_s3_client(); + add_rustfs_tier(&hot, &cold).await?; + + hot_client.create_bucket().bucket(MANUAL_ASYNC_STATUS_BUCKET).send().await?; + put_lifecycle_transition_rule(&hot_client, MANUAL_ASYNC_STATUS_BUCKET, "manual-async-status", "manual-async-status/", 1) + .await?; + put_single_part_object( + &hot_client, + MANUAL_ASYNC_STATUS_BUCKET, + MANUAL_ASYNC_STATUS_KEY, + b"manual async status not-yet-due dry-run object", + ) + .await?; + + let before_dry_run_remote_count = cold_tier_object_count(&cold_client).await?; + let accepted = manual_transition_async_run(&hot, MANUAL_ASYNC_STATUS_BUCKET, "manual-async-status/", true, 10).await?; + assert_eq!(accepted.state, "accepted"); + assert_eq!(accepted.mode, "durable_job"); + assert_eq!(accepted.report.bucket, MANUAL_ASYNC_STATUS_BUCKET); + assert_eq!(accepted.report.prefix, "manual-async-status/"); + assert!(accepted.report.dry_run); + assert_eq!(accepted.report.scanned, 0); + assert_eq!(accepted.report.eligible, 0); + let job_id = accepted.job_id.as_deref().ok_or("async response must include job_id")?; + let status_endpoint = accepted + .status_endpoint + .as_deref() + .ok_or("async response must include status_endpoint")?; + assert!( + status_endpoint.ends_with(job_id), + "status endpoint must embed job id: endpoint={status_endpoint}, job_id={job_id}" + ); + + let terminal = wait_for_manual_transition_job_terminal(&hot, status_endpoint, StdDuration::from_secs(30)).await?; + assert_eq!(terminal.job_id, job_id); + assert_eq!(terminal.status_endpoint, status_endpoint); + assert_eq!(terminal.status, "completed", "terminal job response: {terminal:#?}"); + assert!(!terminal.cancel_requested); + assert_eq!(terminal.failure_reason, None); + assert_eq!(terminal.report.bucket, MANUAL_ASYNC_STATUS_BUCKET); + assert_eq!(terminal.report.prefix, "manual-async-status/"); + assert_eq!(terminal.report.tier.as_deref(), Some(TIER_NAME)); + assert!(terminal.report.dry_run); + assert!(terminal.report.lifecycle_config_found); + assert_eq!(terminal.report.scanned, 1, "terminal job response: {terminal:#?}"); + assert_eq!(terminal.report.eligible, 0, "terminal job response: {terminal:#?}"); + assert_eq!(terminal.report.dry_run_eligible, 0, "terminal job response: {terminal:#?}"); + assert_eq!(terminal.report.enqueued, 0, "terminal job response: {terminal:#?}"); + assert_eq!(terminal.report.skipped_not_transition, 1, "terminal job response: {terminal:#?}"); + assert_eq!(terminal.report.skipped_queue_full, 0); + assert_eq!(terminal.report.skipped_queue_closed, 0); + assert_eq!(terminal.report.skipped_queue_timeout, 0); + assert!(!terminal.report.truncated_by_limit); + assert!(!terminal.report.truncated_by_duration); + + let after_cancel = manual_transition_job_cancel(&hot, status_endpoint).await?; + assert_eq!(after_cancel.status, "completed"); + assert!(!after_cancel.cancel_requested); + assert_eq!( + cold_tier_object_count(&cold_client).await?, + before_dry_run_remote_count, + "not-yet-due async dry-run job must not write to cold tier" + ); + assert_not_transitioned(&hot_client, MANUAL_ASYNC_STATUS_BUCKET, MANUAL_ASYNC_STATUS_KEY).await?; + + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn test_manual_transition_run_contract_no_status_cancel_fields() -> TestResult { let mut cold = RustFSTestEnvironment::new().await?; diff --git a/crates/ecstore/src/cluster/rpc/remote_disk.rs b/crates/ecstore/src/cluster/rpc/remote_disk.rs index 1a924016b..55c9656ae 100644 --- a/crates/ecstore/src/cluster/rpc/remote_disk.rs +++ b/crates/ecstore/src/cluster/rpc/remote_disk.rs @@ -1118,13 +1118,19 @@ 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(Error::from); + return T::deserialize(&mut deserializer).map_err(|err| { + crate::cluster::rpc::runtime_sources::record_response_msgpack_decode_error(value_name); + 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(Error::from) + serde_json::from_str(json).map_err(|err| { + crate::cluster::rpc::runtime_sources::record_response_json_decode_error(value_name); + Error::from(err) + }) } /// Aggregate encoded size (bytes) of a `ReadMultiple` response, preferring the msgpack payloads @@ -1172,8 +1178,10 @@ fn decode_read_multiple_response_items(response: ReadMultipleResponse, endpoint: } let mut read_multiple_resps = Vec::with_capacity(response.read_multiple_resps.len()); for (index, json_str) in response.read_multiple_resps.iter().enumerate() { - let resp = serde_json::from_str::(json_str) - .map_err(|err| Error::other(format!("decode ReadMultipleResp json item {index} from {endpoint} failed: {err}")))?; + let resp = serde_json::from_str::(json_str).map_err(|err| { + crate::cluster::rpc::runtime_sources::record_response_json_decode_error("ReadMultipleResp"); + Error::other(format!("decode ReadMultipleResp json item {index} from {endpoint} failed: {err}")) + })?; read_multiple_resps.push(resp); } @@ -1221,6 +1229,7 @@ fn decode_batch_read_version_response_items( let mut batch_read_version_resps = Vec::with_capacity(response.batch_read_version_resps.len()); for (index, json_str) in response.batch_read_version_resps.iter().enumerate() { let resp = serde_json::from_str::(json_str).map_err(|err| { + crate::cluster::rpc::runtime_sources::record_response_json_decode_error("BatchReadVersionResp"); Error::other(format!("decode BatchReadVersionResp json item {index} from {endpoint} failed: {err}")) })?; if resp.success { @@ -3227,12 +3236,15 @@ mod tests { ], error: None, }; + let before = rustfs_io_metrics::internode_metrics::global_internode_metrics().msgpack_json_decode_error_total_for_test(); let err = decode_read_multiple_response_items(response, &endpoint).expect_err("corrupt msgpack item should fail"); + let after = rustfs_io_metrics::internode_metrics::global_internode_metrics().msgpack_json_decode_error_total_for_test(); let err = err.to_string(); assert!(err.contains("ReadMultipleResp msgpack item 1"), "unexpected error: {err}"); assert!(err.contains("server:9000"), "unexpected error: {err}"); + assert!(after > before, "corrupt response msgpack should increment decode-error metrics"); } #[test] @@ -3247,12 +3259,15 @@ mod tests { read_multiple_resps_bin: Vec::new(), error: None, }; + let before = rustfs_io_metrics::internode_metrics::global_internode_metrics().msgpack_json_decode_error_total_for_test(); let err = decode_read_multiple_response_items(response, &endpoint).expect_err("corrupt json item should fail"); + let after = rustfs_io_metrics::internode_metrics::global_internode_metrics().msgpack_json_decode_error_total_for_test(); let err = err.to_string(); assert!(err.contains("ReadMultipleResp json item 1"), "unexpected error: {err}"); assert!(err.contains("server:9000"), "unexpected error: {err}"); + assert!(after > before, "corrupt response JSON should increment decode-error metrics"); } fn sample_batch_read_version_resp(index: usize, path: &str, success: bool) -> BatchReadVersionResp { @@ -3325,13 +3340,16 @@ mod tests { ], error: None, }; + let before = rustfs_io_metrics::internode_metrics::global_internode_metrics().msgpack_json_decode_error_total_for_test(); let err = decode_batch_read_version_response_items(response, &endpoint) .expect_err("corrupt msgpack item should fail") .to_string(); + let after = rustfs_io_metrics::internode_metrics::global_internode_metrics().msgpack_json_decode_error_total_for_test(); assert!(err.contains("BatchReadVersionResp msgpack item 1"), "unexpected error: {err}"); assert!(err.contains("server:9000"), "unexpected error: {err}"); + assert!(after > before, "corrupt batch response msgpack should increment decode-error metrics"); } #[test] @@ -3346,13 +3364,16 @@ mod tests { batch_read_version_resps_bin: Vec::new(), error: None, }; + let before = rustfs_io_metrics::internode_metrics::global_internode_metrics().msgpack_json_decode_error_total_for_test(); let err = decode_batch_read_version_response_items(response, &endpoint) .expect_err("corrupt json item should fail") .to_string(); + let after = rustfs_io_metrics::internode_metrics::global_internode_metrics().msgpack_json_decode_error_total_for_test(); assert!(err.contains("BatchReadVersionResp json item 1"), "unexpected error: {err}"); assert!(err.contains("server:9000"), "unexpected error: {err}"); + assert!(after > before, "corrupt batch response JSON should increment decode-error metrics"); } #[test] diff --git a/crates/ecstore/src/cluster/rpc/runtime_sources.rs b/crates/ecstore/src/cluster/rpc/runtime_sources.rs index 026e7fa59..9b147422e 100644 --- a/crates/ecstore/src/cluster/rpc/runtime_sources.rs +++ b/crates/ecstore/src/cluster/rpc/runtime_sources.rs @@ -13,9 +13,10 @@ // limitations under the License. use rustfs_io_metrics::internode_metrics::{ - INTERNODE_MSGPACK_DIRECTION_RESPONSE, INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_READ_MULTIPLE, - 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_MSGPACK_CODEC_JSON, INTERNODE_MSGPACK_CODEC_MSGPACK, INTERNODE_MSGPACK_DIRECTION_RESPONSE, + INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_READ_MULTIPLE, 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, }; #[cfg(test)] @@ -126,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_error(message: &'static str) { + global_internode_metrics().record_msgpack_json_decode_error( + INTERNODE_MSGPACK_DIRECTION_RESPONSE, + message, + INTERNODE_MSGPACK_CODEC_MSGPACK, + ); +} + +pub(crate) fn record_response_json_decode_error(message: &'static str) { + global_internode_metrics().record_msgpack_json_decode_error( + INTERNODE_MSGPACK_DIRECTION_RESPONSE, + message, + INTERNODE_MSGPACK_CODEC_JSON, + ); +} + #[cfg(test)] pub(crate) fn reset_internode_metrics_for_test() { global_internode_metrics().reset_for_test(); @@ -135,3 +152,8 @@ pub(crate) fn reset_internode_metrics_for_test() { pub(crate) fn internode_metrics_snapshot_for_test() -> InternodeMetricsSnapshot { global_internode_metrics().snapshot() } + +#[cfg(test)] +pub(crate) fn internode_msgpack_json_decode_error_total_for_test() -> u64 { + global_internode_metrics().msgpack_json_decode_error_total_for_test() +} diff --git a/crates/io-metrics/src/internode_metrics.rs b/crates/io-metrics/src/internode_metrics.rs index 55e2ca99e..9fb51c32e 100644 --- a/crates/io-metrics/src/internode_metrics.rs +++ b/crates/io-metrics/src/internode_metrics.rs @@ -35,6 +35,8 @@ pub const INTERNODE_TRANSPORT_BACKEND_UNKNOWN: &str = "unknown"; /// peer's request vs a client decoding a peer's response (grpc-optimization P2). pub const INTERNODE_MSGPACK_DIRECTION_REQUEST: &str = "request"; pub const INTERNODE_MSGPACK_DIRECTION_RESPONSE: &str = "response"; +pub const INTERNODE_MSGPACK_CODEC_MSGPACK: &str = "msgpack"; +pub const INTERNODE_MSGPACK_CODEC_JSON: &str = "json"; const OPERATION_LABEL: &str = "operation"; const BACKEND_LABEL: &str = "backend"; @@ -44,6 +46,7 @@ const DOMINANT_ERROR_LABEL: &str = "dominant_error"; const HTTP_VERSION_LABEL: &str = "http_version"; const DIRECTION_LABEL: &str = "direction"; const MESSAGE_LABEL: &str = "message"; +const CODEC_LABEL: &str = "codec"; const INTERNODE_OPERATION_SENT_BYTES_TOTAL: &str = "rustfs_system_network_internode_operation_sent_bytes_total"; const INTERNODE_OPERATION_RECV_BYTES_TOTAL: &str = "rustfs_system_network_internode_operation_recv_bytes_total"; const INTERNODE_OPERATION_REQUESTS_OUTGOING_TOTAL: &str = "rustfs_system_network_internode_operation_requests_outgoing_total"; @@ -60,6 +63,7 @@ const INTERNODE_OPERATION_WRITE_SHUTDOWN_ERRORS_TOTAL: &str = 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_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"; const ERASURE_WRITE_QUORUM_FAILURES_TOTAL: &str = "rustfs_system_storage_erasure_write_quorum_failures_total"; @@ -167,6 +171,7 @@ pub struct InternodeMetrics { operation_http_versions_total: AtomicU64, operation_stall_timeouts_total: AtomicU64, operation_write_shutdown_errors_total: AtomicU64, + msgpack_json_decode_error_total: AtomicU64, signature_v1_fallback_total: AtomicU64, } @@ -364,6 +369,22 @@ impl InternodeMetrics { counter!(INTERNODE_MSGPACK_JSON_FALLBACK_TOTAL, DIRECTION_LABEL => direction, MESSAGE_LABEL => message).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!( + INTERNODE_MSGPACK_JSON_DECODE_ERROR_TOTAL, + DIRECTION_LABEL => direction, + MESSAGE_LABEL => message, + CODEC_LABEL => codec + ) + .increment(1); + } + + #[doc(hidden)] + pub fn msgpack_json_decode_error_total_for_test(&self) -> u64 { + self.msgpack_json_decode_error_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 @@ -439,6 +460,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_error_total.store(0, Ordering::Relaxed); self.signature_v1_fallback_total.store(0, Ordering::Relaxed); } } @@ -742,8 +764,14 @@ mod tests { INTERNODE_MSGPACK_JSON_FALLBACK_TOTAL, "rustfs_system_network_internode_msgpack_json_fallback_total" ); + assert_eq!( + INTERNODE_MSGPACK_JSON_DECODE_ERROR_TOTAL, + "rustfs_system_network_internode_msgpack_json_decode_error_total" + ); assert_eq!(INTERNODE_MSGPACK_DIRECTION_REQUEST, "request"); assert_eq!(INTERNODE_MSGPACK_DIRECTION_RESPONSE, "response"); + assert_eq!(INTERNODE_MSGPACK_CODEC_MSGPACK, "msgpack"); + assert_eq!(INTERNODE_MSGPACK_CODEC_JSON, "json"); assert_eq!( INTERNODE_SIGNATURE_V1_FALLBACK_TOTAL, "rustfs_system_network_internode_signature_v1_fallback_total" @@ -758,6 +786,27 @@ mod tests { metrics.record_msgpack_json_fallback(INTERNODE_MSGPACK_DIRECTION_RESPONSE, "RawFileInfo"); } + #[test] + fn msgpack_json_decode_error_counter_tracks_codec_direction_and_message() { + let metrics = InternodeMetrics::default(); + assert_eq!(metrics.msgpack_json_decode_error_total_for_test(), 0); + + metrics.record_msgpack_json_decode_error( + INTERNODE_MSGPACK_DIRECTION_REQUEST, + "FileInfo", + INTERNODE_MSGPACK_CODEC_MSGPACK, + ); + metrics.record_msgpack_json_decode_error( + INTERNODE_MSGPACK_DIRECTION_RESPONSE, + "RawFileInfo", + INTERNODE_MSGPACK_CODEC_JSON, + ); + + assert_eq!(metrics.msgpack_json_decode_error_total_for_test(), 2); + metrics.reset_for_test(); + assert_eq!(metrics.msgpack_json_decode_error_total_for_test(), 0); + } + #[test] fn signature_v1_fallback_counter_updates_snapshot_and_resets() { // Instance-local metrics keep this independent of the process-global registry. diff --git a/docs/operations/internode-grpc-benchmark-runbook.md b/docs/operations/internode-grpc-benchmark-runbook.md index 5e8aefffa..bee4d3d6d 100644 --- a/docs/operations/internode-grpc-benchmark-runbook.md +++ b/docs/operations/internode-grpc-benchmark-runbook.md @@ -50,6 +50,7 @@ gitignored — attach the paired directories to the PR / issue. | `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_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) | | lock p99 (lock metrics) | P1 head-of-line-blocking win | @@ -86,10 +87,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` across a release window; it 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_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. - **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 diff --git a/docs/operations/internode-msgpack-json-convergence-runbook.md b/docs/operations/internode-msgpack-json-convergence-runbook.md index 009b612f3..67e5dbca2 100644 --- a/docs/operations/internode-msgpack-json-convergence-runbook.md +++ b/docs/operations/internode-msgpack-json-convergence-runbook.md @@ -20,6 +20,7 @@ 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_error_total{direction, message, codec} ``` Incremented whenever a decode falls back to the JSON field because the msgpack payload was @@ -29,6 +30,7 @@ absent. - `direction="response"` — a client decoding a peer's response (`cluster/rpc/remote_disk.rs`), including the list-level `ReadMultiple` / `BatchReadVersion` fallbacks. - `message` — the value name, e.g. `FileInfo`, `RawFileInfo`, `ReadMultipleResp`. +- `codec` — the failed codec for decode errors: `msgpack` for corrupt non-empty `_bin`, or `json` for corrupt legacy fallback JSON. ## Stage 0 — Observe (current stage) @@ -47,6 +49,16 @@ Every series must be `0`. A non-zero value means some peer is still emitting an `_bin` (an old node, or a message whose sender does not fill `_bin`) — investigate the `{direction, message}` label before proceeding. +Decode errors must also stay at zero across the observation window: + +```promql +sum by (direction, message, codec) ( + increase(rustfs_system_network_internode_msgpack_json_decode_error_total[30d]) +) +``` + +A non-zero `codec="msgpack"` series means a peer sent corrupt or incompatible `_bin` bytes; it must fail closed and block convergence. A non-zero `codec="json"` series means the legacy fallback field was corrupt or semantically incompatible; it also blocks convergence and rollback confidence. + Standing alert (keep enabled through all stages): ```yaml @@ -57,6 +69,13 @@ Standing alert (keep enabled through all stages): annotations: summary: "Internode RPC fell back to JSON decode ({{ $labels.direction }}/{{ $labels.message }})" description: "A peer sent an empty msgpack _bin payload. Do NOT advance msgpack-only convergence while this fires." +- alert: InternodeMsgpackJsonDecodeError + expr: sum by (direction, message, codec) (increase(rustfs_system_network_internode_msgpack_json_decode_error_total[15m])) > 0 + for: 5m + labels: { severity: warning } + annotations: + summary: "Internode RPC msgpack/JSON decode failed ({{ $labels.direction }}/{{ $labels.message }}/{{ $labels.codec }})" + description: "A peer sent an undecodable msgpack or JSON compatibility payload. Do NOT advance msgpack-only convergence while this fires." ``` ## Field → peer-decoder audit diff --git a/rustfs/src/admin/handlers/ilm_transition.rs b/rustfs/src/admin/handlers/ilm_transition.rs index 2ea82fc7d..27497bcd1 100644 --- a/rustfs/src/admin/handlers/ilm_transition.rs +++ b/rustfs/src/admin/handlers/ilm_transition.rs @@ -537,6 +537,15 @@ fn request_manual_transition_job_cancel(job_id: &str) -> Option ManualTransitionJobRecord { + if !record.status.is_terminal() { + record.status = ManualTransitionJobStatus::Unknown; + record.finished_at = Some(manual_transition_timestamp(OffsetDateTime::now_utc())); + record.failure_reason = Some("manual transition job owner is not active on this node".to_string()); + } + record +} + fn remove_manual_transition_job(job_id: &str) { let mut jobs = lock_manual_transition_jobs(); jobs.remove(job_id); @@ -570,15 +579,10 @@ async fn load_manual_transition_job_record(store: Arc, job_id: &str) -> if let Some(record) = in_memory_manual_transition_job_record(job_id) { return Ok(record); } - let Some(mut record) = read_manual_transition_job_record(store, job_id).await? else { + let Some(record) = read_manual_transition_job_record(store, job_id).await? else { return Err(s3_error!(NoSuchKey, "manual transition job not found")); }; - if !record.status.is_terminal() { - record.status = ManualTransitionJobStatus::Unknown; - record.finished_at = Some(manual_transition_timestamp(OffsetDateTime::now_utc())); - record.failure_reason = Some("manual transition job owner is not active on this node".to_string()); - } - Ok(record) + Ok(mark_manual_transition_job_owner_unknown(record)) } async fn persist_manual_transition_job_update(store: Arc, record: &ManualTransitionJobRecord) -> bool { @@ -762,10 +766,8 @@ impl Operation for ManualTransitionJobCancelHandler { return Err(s3_error!(NoSuchKey, "manual transition job not found")); }; if !record.status.is_terminal() { - record.status = ManualTransitionJobStatus::Unknown; record.cancel_requested = true; - record.finished_at = Some(manual_transition_timestamp(OffsetDateTime::now_utc())); - record.failure_reason = Some("manual transition job owner is not active on this node".to_string()); + record = mark_manual_transition_job_owner_unknown(record); save_manual_transition_job_record(store, &record).await?; } record @@ -1065,6 +1067,47 @@ mod tests { assert!(cancel_token.is_cancelled()); } + #[test] + fn manual_transition_inactive_owner_marks_nonterminal_job_unknown() { + let (bucket, options, _run_mode) = + parse_manual_transition_query(Some("bucket=data&prefix=logs/&async=true")).expect("async query should parse"); + let record = new_manual_transition_job_record( + "11111111-1111-4111-8111-111111111111".to_string(), + &bucket, + &options, + OffsetDateTime::UNIX_EPOCH, + ); + + let unknown = mark_manual_transition_job_owner_unknown(record); + + assert_eq!(unknown.status, ManualTransitionJobStatus::Unknown); + assert!(unknown.finished_at.is_some()); + assert_eq!( + unknown.failure_reason.as_deref(), + Some("manual transition job owner is not active on this node") + ); + } + + #[test] + fn manual_transition_inactive_owner_preserves_terminal_job() { + let (bucket, options, _run_mode) = + parse_manual_transition_query(Some("bucket=data&prefix=logs/&async=true")).expect("async query should parse"); + let mut record = new_manual_transition_job_record( + "11111111-1111-4111-8111-111111111111".to_string(), + &bucket, + &options, + OffsetDateTime::UNIX_EPOCH, + ); + record.status = ManualTransitionJobStatus::Completed; + record.finished_at = Some("1970-01-01T00:00:00Z".to_string()); + + let completed = mark_manual_transition_job_owner_unknown(record); + + assert_eq!(completed.status, ManualTransitionJobStatus::Completed); + assert_eq!(completed.finished_at.as_deref(), Some("1970-01-01T00:00:00Z")); + assert_eq!(completed.failure_reason, None); + } + #[test] fn manual_transition_handler_requires_set_tier_action() { let src = include_str!("ilm_transition.rs"); diff --git a/rustfs/src/storage/rpc/node_service/disk.rs b/rustfs/src/storage/rpc/node_service/disk.rs index e6f54da76..c3b8fa55c 100644 --- a/rustfs/src/storage/rpc/node_service/disk.rs +++ b/rustfs/src/storage/rpc/node_service/disk.rs @@ -21,8 +21,9 @@ use crate::storage::storage_api::runtime_sources_consumer::runtime_sources; use bytes::Bytes; use rustfs_filemeta::FileInfo; use rustfs_io_metrics::internode_metrics::{ - INTERNODE_MSGPACK_DIRECTION_REQUEST, INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_WRITE_ALL, - INTERNODE_TRANSPORT_BACKEND_GRPC, global_internode_metrics, + INTERNODE_MSGPACK_CODEC_JSON, INTERNODE_MSGPACK_CODEC_MSGPACK, INTERNODE_MSGPACK_DIRECTION_REQUEST, + INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_TRANSPORT_BACKEND_GRPC, + global_internode_metrics, }; use rustfs_protos::proto_gen::node_service::*; use serde::de::DeserializeOwned; @@ -41,14 +42,27 @@ 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| DiskError::other(format!("decode {value_name} msgpack failed: {err}"))); + 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}")) + }); } // 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| DiskError::other(format!("decode {value_name} failed: {err}"))) + 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}")) + }) } fn encode_msgpack(value: &T, value_name: &str) -> std::result::Result, DiskError> { @@ -1191,6 +1205,7 @@ mod tests { }; use crate::storage::storage_api::ReadMultipleResp; use crate::storage::storage_api::rpc_consumer::node_service::BatchReadVersionResp; + use rustfs_io_metrics::internode_metrics::global_internode_metrics; use serde::{Deserialize, Serialize}; #[derive(Debug, PartialEq, Eq, Serialize, Deserialize)] @@ -1314,18 +1329,24 @@ mod tests { #[test] fn decode_msgpack_or_json_fails_closed_on_corrupt_non_empty_msgpack() { + let before = global_internode_metrics().msgpack_json_decode_error_total_for_test(); let err = decode_msgpack_or_json::(b"not-msgpack", r#"{"name":"json","count":1}"#, "SamplePayload") .expect_err("corrupt non-empty msgpack must not fall back to JSON"); + let after = global_internode_metrics().msgpack_json_decode_error_total_for_test(); assert!(err.to_string().contains("decode SamplePayload msgpack failed"), "unexpected error: {err}"); + assert!(after > before, "corrupt msgpack should increment decode-error metrics"); } #[test] fn decode_msgpack_or_json_reports_corrupt_json_item_when_msgpack_absent() { + let before = global_internode_metrics().msgpack_json_decode_error_total_for_test(); let err = decode_msgpack_or_json::(&[], "{not-json", "SamplePayload") .expect_err("corrupt json item should fail in fallback branch"); + let after = global_internode_metrics().msgpack_json_decode_error_total_for_test(); assert!(err.to_string().contains("decode SamplePayload failed"), "unexpected error: {err}"); + assert!(after > before, "corrupt fallback JSON should increment decode-error metrics"); } #[test]