diff --git a/Cargo.lock b/Cargo.lock index c2eec7a0a..988712641 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8344,6 +8344,7 @@ dependencies = [ "chrono", "humantime", "hyper", + "rmp-serde", "serde", "serde_json", "time", diff --git a/crates/madmin/Cargo.toml b/crates/madmin/Cargo.toml index c8c772b27..192656800 100644 --- a/crates/madmin/Cargo.toml +++ b/crates/madmin/Cargo.toml @@ -38,3 +38,6 @@ time.workspace = true [lib] doctest = false + +[dev-dependencies] +rmp-serde.workspace = true diff --git a/crates/madmin/src/info_commands.rs b/crates/madmin/src/info_commands.rs index 1c326e01c..eaa2d1c41 100644 --- a/crates/madmin/src/info_commands.rs +++ b/crates/madmin/src/info_commands.rs @@ -86,10 +86,6 @@ pub struct Disk { pub write_latency: f64, pub utilization: f64, pub metrics: Option, - #[serde(rename = "runtimeState", default, skip_serializing_if = "Option::is_none")] - pub runtime_state: Option, - #[serde(rename = "offlineDurationSeconds", default, skip_serializing_if = "Option::is_none")] - pub offline_duration_seconds: Option, pub heal_info: Option, pub used_inodes: u64, pub free_inodes: u64, @@ -97,6 +93,10 @@ pub struct Disk { pub pool_index: i32, pub set_index: i32, pub disk_index: i32, + #[serde(rename = "runtimeState", default, skip_serializing_if = "Option::is_none")] + pub runtime_state: Option, + #[serde(rename = "offlineDurationSeconds", default, skip_serializing_if = "Option::is_none")] + pub offline_duration_seconds: Option, } #[derive(Clone, Debug, Default, Serialize, Deserialize)] @@ -353,10 +353,50 @@ pub struct InfoMessage { #[cfg(test)] mod tests { use super::*; + use rmp_serde::{Deserializer, Serializer}; use serde_json; - use std::collections::HashMap; + use std::{collections::HashMap, io::Cursor}; use time::OffsetDateTime; + #[derive(Clone, Debug, Default, serde::Serialize, serde::Deserialize)] + struct LegacyDiskCompat { + endpoint: String, + #[serde(rename = "rootDisk")] + root_disk: bool, + #[serde(rename = "path")] + drive_path: String, + healing: bool, + scanning: bool, + state: String, + uuid: String, + major: u32, + minor: u32, + model: Option, + #[serde(rename = "totalspace")] + total_space: u64, + #[serde(rename = "usedspace")] + used_space: u64, + #[serde(rename = "availspace")] + available_space: u64, + #[serde(rename = "readthroughput")] + read_throughput: f64, + #[serde(rename = "writethroughput")] + write_throughput: f64, + #[serde(rename = "readlatency")] + read_latency: f64, + #[serde(rename = "writelatency")] + write_latency: f64, + utilization: f64, + metrics: Option, + heal_info: Option, + used_inodes: u64, + free_inodes: u64, + local: bool, + pool_index: i32, + set_index: i32, + disk_index: i32, + } + #[test] fn test_item_state_to_string() { assert_eq!(ItemState::Offline.to_string(), ITEM_OFFLINE); @@ -496,6 +536,92 @@ mod tests { assert!(disk.local); } + #[test] + fn test_disk_msgpack_backward_compat_from_legacy_layout() { + let legacy = LegacyDiskCompat { + endpoint: "http://legacy-node:9000".to_string(), + root_disk: false, + drive_path: "/data/legacy".to_string(), + healing: false, + scanning: false, + state: ITEM_ONLINE.to_string(), + uuid: "legacy-uuid".to_string(), + major: 8, + minor: 2, + model: Some("legacy".to_string()), + total_space: 42, + used_space: 12, + available_space: 30, + read_throughput: 1.0, + write_throughput: 2.0, + read_latency: 3.0, + write_latency: 4.0, + utilization: 5.0, + metrics: None, + heal_info: None, + used_inodes: 11_125, + free_inodes: 98_000, + local: true, + pool_index: 1, + set_index: 2, + disk_index: 3, + }; + + let mut encoded = Vec::new(); + legacy.serialize(&mut Serializer::new(&mut encoded)).unwrap(); + + let mut decoder = Deserializer::new(Cursor::new(encoded)); + let decoded: Disk = serde::Deserialize::deserialize(&mut decoder).unwrap(); + assert_eq!(decoded.used_inodes, 11_125); + assert_eq!(decoded.runtime_state, None); + assert_eq!(decoded.offline_duration_seconds, None); + } + + #[test] + fn test_disk_msgpack_forward_compat_to_legacy_layout() { + let current = Disk { + endpoint: "http://current-node:9000".to_string(), + root_disk: false, + drive_path: "/data/current".to_string(), + healing: false, + scanning: false, + state: ITEM_ONLINE.to_string(), + uuid: "current-uuid".to_string(), + major: 8, + minor: 3, + model: Some("current".to_string()), + total_space: 64, + used_space: 20, + available_space: 44, + read_throughput: 1.5, + write_throughput: 2.5, + read_latency: 3.5, + write_latency: 4.5, + utilization: 6.5, + metrics: None, + heal_info: None, + used_inodes: 22_250, + free_inodes: 97_000, + local: true, + pool_index: 1, + set_index: 2, + disk_index: 3, + runtime_state: Some("online".to_string()), + offline_duration_seconds: Some(0), + }; + + let mut encoded = Vec::new(); + current + .serialize(&mut Serializer::new(&mut encoded).with_struct_map()) + .unwrap(); + + let mut decoder = Deserializer::new(Cursor::new(encoded)); + let decoded: LegacyDiskCompat = serde::Deserialize::deserialize(&mut decoder).unwrap(); + assert_eq!(decoded.used_inodes, 22_250); + assert_eq!(decoded.disk_index, 3); + assert_eq!(decoded.endpoint, "http://current-node:9000"); + } + #[test] fn test_healing_disk_default() { let healing_disk = HealingDisk::default(); diff --git a/crates/targets/src/target/mqtt.rs b/crates/targets/src/target/mqtt.rs index 2c2ce7759..f5bfed891 100644 --- a/crates/targets/src/target/mqtt.rs +++ b/crates/targets/src/target/mqtt.rs @@ -467,12 +467,12 @@ impl MQTTArgs { if !self.queue_dir.is_empty() { let path = std::path::Path::new(&self.queue_dir); if !path.is_absolute() { - return Err(TargetError::Configuration("mqtt queueDir path should be absolute".to_string())); + return Err(TargetError::Configuration("mqtt queue_dir path should be absolute".to_string())); } if self.qos == QoS::AtMostOnce { return Err(TargetError::Configuration( - "QoS should be AtLeastOnce (1) or ExactlyOnce (2) if queueDir is set".to_string(), + "QoS should be AtLeastOnce (1) or ExactlyOnce (2) if queue_dir is set".to_string(), )); } } diff --git a/crates/targets/src/target/webhook.rs b/crates/targets/src/target/webhook.rs index 79b6ba520..2c76ffcb5 100644 --- a/crates/targets/src/target/webhook.rs +++ b/crates/targets/src/target/webhook.rs @@ -79,7 +79,7 @@ impl WebhookArgs { if !self.queue_dir.is_empty() { let path = std::path::Path::new(&self.queue_dir); if !path.is_absolute() { - return Err(TargetError::Configuration("webhook queueDir path should be absolute".to_string())); + return Err(TargetError::Configuration("webhook queue_dir path should be absolute".to_string())); } } diff --git a/rustfs/src/storage/rpc/health.rs b/rustfs/src/storage/rpc/health.rs index 1f7e6eb98..3109299f7 100644 --- a/rustfs/src/storage/rpc/health.rs +++ b/rustfs/src/storage/rpc/health.rs @@ -204,7 +204,9 @@ impl NodeService { ) -> Result, Status> { let info = get_local_server_property().await; let mut buf = Vec::new(); - if let Err(err) = info.serialize(&mut Serializer::new(&mut buf)) { + // Use map encoding for forward/backward compatibility across mixed versions: + // unknown fields can be ignored by older nodes during deserialization. + if let Err(err) = info.serialize(&mut Serializer::new(&mut buf).with_struct_map()) { return Ok(Response::new(ServerInfoResponse { success: false, server_properties: Bytes::new(),