mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-24 05:06:28 +00:00
fix(replication): accept remote target healthCheckDuration nanoseconds (#5754)
* test(replication): accept madmin nanosecond healthCheckDuration payloads Red-phase TDD tests for P0-7: mc 'replicate add' sends the madmin default healthCheckDuration=60s as a Go time.Duration nanosecond integer (60000000000), which RustFS currently rejects as an unsupported field and would misread as seconds. Also pins the defensive seconds-or-nanos read for persisted bucket-targets metadata and the capability contract listing healthCheckDuration as writable. Currently failing (red): - remote_target_request_accepts_go_duration_wire_values - remote_target_request_accepts_legacy_seconds_health_check - remote_target_health_check_duration_is_declared_writable - bucket_target_reads_go_nanosecond_durations_defensively - runtime_capabilities_response_reports_missing_topology_before_storage_init * fix(replication): accept remote target healthCheckDuration nanoseconds mc 'replicate add' always sends the madmin default healthcheck-seconds=60 serialized as a Go time.Duration nanosecond integer (60000000000), so the default mc link-creation path (and 'mc replicate update') failed with InvalidRequest. Move healthCheckDuration from the unsupported to the writable remote-target field list; the capability contract in the runtime capabilities response follows the constants automatically. Fix the unit mismatch in both directions: - Request parsing and persisted bucket-targets reads decode the value defensively: below 10^7 it is legacy RustFS seconds, otherwise Go time.Duration nanoseconds (also covers MinIO-written metadata). totalDowntime shares the same wire shape and gets the same handling. - The list-remote-targets admin response re-encodes only these two fields as nanoseconds via a dedicated serialization path, leaving the persisted seconds-based wire format untouched for existing readers. The per-target health-check interval is accepted for mc compatibility but not yet applied; the heartbeat keeps its global env-configured interval, and the explicit 'healthcheck' update op stays rejected. disableProxy, edge, and edgeSyncBeforeExpiry remain explicitly rejected.
This commit is contained in:
@@ -206,7 +206,9 @@ pub mod bucket {
|
|||||||
}
|
}
|
||||||
|
|
||||||
pub mod target {
|
pub mod target {
|
||||||
pub use crate::bucket::target::{ARN, BucketTarget, BucketTargetType, BucketTargets, Credentials, LatencyStat};
|
pub use crate::bucket::target::{
|
||||||
|
ARN, BucketTarget, BucketTargetType, BucketTargets, Credentials, LatencyStat, duration_from_secs_or_nanos,
|
||||||
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
pub mod utils {
|
pub mod utils {
|
||||||
|
|||||||
@@ -94,6 +94,21 @@ mod duration_milliseconds {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Defensive decode for the two integer wire encodings of these duration
|
||||||
|
/// fields: RustFS persists (and legacy RustFS clients sent) plain seconds,
|
||||||
|
/// while Go `time.Duration` JSON — madmin/mc requests and MinIO-written
|
||||||
|
/// bucket-targets metadata — is nanoseconds. No meaningful interval lies
|
||||||
|
/// between 10^7 seconds (~115 days) and 10^7 nanoseconds (10ms), so the
|
||||||
|
/// magnitude disambiguates the unit.
|
||||||
|
pub fn duration_from_secs_or_nanos(value: u64) -> Duration {
|
||||||
|
const NANOS_THRESHOLD: u64 = 10_000_000;
|
||||||
|
if value < NANOS_THRESHOLD {
|
||||||
|
Duration::from_secs(value)
|
||||||
|
} else {
|
||||||
|
Duration::from_nanos(value)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
mod duration_seconds {
|
mod duration_seconds {
|
||||||
use serde::{Deserialize, Deserializer, Serializer};
|
use serde::{Deserialize, Deserializer, Serializer};
|
||||||
use std::time::Duration;
|
use std::time::Duration;
|
||||||
@@ -109,8 +124,8 @@ mod duration_seconds {
|
|||||||
where
|
where
|
||||||
D: Deserializer<'de>,
|
D: Deserializer<'de>,
|
||||||
{
|
{
|
||||||
let secs = u64::deserialize(deserializer)?;
|
let value = u64::deserialize(deserializer)?;
|
||||||
Ok(Duration::from_secs(secs))
|
Ok(super::duration_from_secs_or_nanos(value))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -489,6 +504,29 @@ mod tests {
|
|||||||
assert_eq!(original.offline_count, deserialized.offline_count);
|
assert_eq!(original.offline_count, deserialized.offline_count);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn bucket_target_reads_go_nanosecond_durations_defensively() {
|
||||||
|
// MinIO-written bucket-targets metadata and madmin clients encode
|
||||||
|
// these fields as Go `time.Duration` nanoseconds; RustFS has always
|
||||||
|
// persisted seconds. Both encodings must decode to the same interval.
|
||||||
|
let target: BucketTarget = serde_json::from_value(serde_json::json!({
|
||||||
|
"endpoint": "localhost:9000",
|
||||||
|
"targetbucket": "target",
|
||||||
|
"type": "replication",
|
||||||
|
"healthCheckDuration": 60_000_000_000u64,
|
||||||
|
"totalDowntime": 90_000_000_000u64
|
||||||
|
}))
|
||||||
|
.expect("nanosecond durations should deserialize");
|
||||||
|
|
||||||
|
assert_eq!(target.health_check_duration, Duration::from_secs(60));
|
||||||
|
assert_eq!(target.total_downtime, Duration::from_secs(90));
|
||||||
|
|
||||||
|
// The persisted wire format stays seconds for existing RustFS readers.
|
||||||
|
let value = serde_json::to_value(&target).expect("target should serialize");
|
||||||
|
assert_eq!(value["healthCheckDuration"], 60);
|
||||||
|
assert_eq!(value["totalDowntime"], 90);
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_bucket_target_debug_redacts_credentials() {
|
fn test_bucket_target_debug_redacts_credentials() {
|
||||||
let target = BucketTarget {
|
let target = BucketTarget {
|
||||||
|
|||||||
@@ -73,9 +73,13 @@ pub const REMOTE_TARGET_WRITABLE_FIELDS: &[&str] = &[
|
|||||||
"storage_class",
|
"storage_class",
|
||||||
"skipTlsVerify",
|
"skipTlsVerify",
|
||||||
"caCertPem",
|
"caCertPem",
|
||||||
|
// Accepted for mc compatibility (mc `replicate add` always sends the
|
||||||
|
// madmin default of 60s); the per-target health-check interval is not
|
||||||
|
// yet applied — the heartbeat keeps its global env-configured interval.
|
||||||
|
"healthCheckDuration",
|
||||||
];
|
];
|
||||||
|
|
||||||
pub const REMOTE_TARGET_UNSUPPORTED_FIELDS: &[&str] = &["disableProxy", "healthCheckDuration", "edge", "edgeSyncBeforeExpiry"];
|
pub const REMOTE_TARGET_UNSUPPORTED_FIELDS: &[&str] = &["disableProxy", "edge", "edgeSyncBeforeExpiry"];
|
||||||
|
|
||||||
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
|
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
|
||||||
pub struct ObjectOpts {
|
pub struct ObjectOpts {
|
||||||
|
|||||||
@@ -26,7 +26,9 @@ use crate::admin::storage_api::bucket::replication::REMOTE_TARGET_UNSUPPORTED_FI
|
|||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
use crate::admin::storage_api::bucket::replication::REMOTE_TARGET_WRITABLE_FIELDS;
|
use crate::admin::storage_api::bucket::replication::REMOTE_TARGET_WRITABLE_FIELDS;
|
||||||
use crate::admin::storage_api::bucket::replication::{BucketStats, ReplicationStatusType};
|
use crate::admin::storage_api::bucket::replication::{BucketStats, ReplicationStatusType};
|
||||||
use crate::admin::storage_api::bucket::target::{BucketTarget, BucketTargetType, Credentials as TargetCredentials, LatencyStat};
|
use crate::admin::storage_api::bucket::target::{
|
||||||
|
BucketTarget, BucketTargetType, Credentials as TargetCredentials, LatencyStat, duration_from_secs_or_nanos,
|
||||||
|
};
|
||||||
use crate::admin::storage_api::bucket::target_sys::{BucketTargetError, BucketTargetSys};
|
use crate::admin::storage_api::bucket::target_sys::{BucketTargetError, BucketTargetSys};
|
||||||
use crate::admin::storage_api::contract::bucket::{BucketOperations, BucketOptions};
|
use crate::admin::storage_api::contract::bucket::{BucketOperations, BucketOptions};
|
||||||
use crate::admin::storage_api::contract::list::ListOperations as _;
|
use crate::admin::storage_api::contract::list::ListOperations as _;
|
||||||
@@ -295,12 +297,12 @@ impl RemoteTargetRequest {
|
|||||||
));
|
));
|
||||||
}
|
}
|
||||||
|
|
||||||
for (unsupported, configured) in REMOTE_TARGET_UNSUPPORTED_FIELDS.iter().copied().zip([
|
for (unsupported, configured) in
|
||||||
self.disable_proxy,
|
REMOTE_TARGET_UNSUPPORTED_FIELDS
|
||||||
self.health_check_duration != 0,
|
.iter()
|
||||||
self.edge,
|
.copied()
|
||||||
self.edge_sync_before_expiry,
|
.zip([self.disable_proxy, self.edge, self.edge_sync_before_expiry])
|
||||||
]) {
|
{
|
||||||
if configured {
|
if configured {
|
||||||
return Err(s3_error!(
|
return Err(s3_error!(
|
||||||
InvalidRequest,
|
InvalidRequest,
|
||||||
@@ -325,11 +327,15 @@ impl RemoteTargetRequest {
|
|||||||
storage_class: self.storage_class,
|
storage_class: self.storage_class,
|
||||||
skip_tls_verify: self.skip_tls_verify,
|
skip_tls_verify: self.skip_tls_verify,
|
||||||
ca_cert_pem: self.ca_cert_pem,
|
ca_cert_pem: self.ca_cert_pem,
|
||||||
health_check_duration: Duration::from_secs(self.health_check_duration),
|
// madmin/mc encode these Go `time.Duration` fields as nanoseconds;
|
||||||
|
// legacy RustFS clients sent seconds. Accepted for mc compatibility;
|
||||||
|
// the per-target health-check interval is not yet applied — the
|
||||||
|
// heartbeat keeps its global env-configured interval.
|
||||||
|
health_check_duration: duration_from_secs_or_nanos(self.health_check_duration),
|
||||||
disable_proxy: self.disable_proxy,
|
disable_proxy: self.disable_proxy,
|
||||||
reset_before_date: self.reset_before_date,
|
reset_before_date: self.reset_before_date,
|
||||||
reset_id: self.reset_id,
|
reset_id: self.reset_id,
|
||||||
total_downtime: Duration::from_secs(self.total_downtime),
|
total_downtime: duration_from_secs_or_nanos(self.total_downtime),
|
||||||
last_online: self.last_online,
|
last_online: self.last_online,
|
||||||
online: self.online,
|
online: self.online,
|
||||||
latency: self.latency,
|
latency: self.latency,
|
||||||
@@ -341,6 +347,23 @@ impl RemoteTargetRequest {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Admin-response encoding of a remote target: the persisted bucket-targets
|
||||||
|
/// format keeps `healthCheckDuration`/`totalDowntime` in seconds, but madmin
|
||||||
|
/// decodes them as Go `time.Duration` (nanoseconds) — re-encode just those
|
||||||
|
/// fields without touching the persistence wire format.
|
||||||
|
fn remote_target_admin_json(target: &BucketTarget) -> Result<serde_json::Value, serde_json::Error> {
|
||||||
|
fn go_duration_nanos(duration: Duration) -> serde_json::Value {
|
||||||
|
// Saturate instead of truncating: >u64::MAX nanoseconds (~584 years)
|
||||||
|
// is unrepresentable for a Go time.Duration reader anyway.
|
||||||
|
u64::try_from(duration.as_nanos()).unwrap_or(u64::MAX).into()
|
||||||
|
}
|
||||||
|
|
||||||
|
let mut value = serde_json::to_value(target)?;
|
||||||
|
value["healthCheckDuration"] = go_duration_nanos(target.health_check_duration);
|
||||||
|
value["totalDowntime"] = go_duration_nanos(target.total_downtime);
|
||||||
|
Ok(value)
|
||||||
|
}
|
||||||
|
|
||||||
fn validate_remote_target_tls_settings(remote_target: &BucketTarget) -> S3Result<()> {
|
fn validate_remote_target_tls_settings(remote_target: &BucketTarget) -> S3Result<()> {
|
||||||
let has_custom_ca = !remote_target.ca_cert_pem.trim().is_empty();
|
let has_custom_ca = !remote_target.ca_cert_pem.trim().is_empty();
|
||||||
|
|
||||||
@@ -732,7 +755,14 @@ impl Operation for ListRemoteTargetHandler {
|
|||||||
let sys = BucketTargetSys::get();
|
let sys = BucketTargetSys::get();
|
||||||
let targets = sys.list_targets(bucket, "").await;
|
let targets = sys.list_targets(bucket, "").await;
|
||||||
|
|
||||||
let targets: Vec<_> = targets.iter().map(|target| target.redacted_credentials()).collect();
|
let targets: Vec<_> = targets
|
||||||
|
.iter()
|
||||||
|
.map(|target| remote_target_admin_json(&target.redacted_credentials()))
|
||||||
|
.collect::<Result<_, _>>()
|
||||||
|
.map_err(|e| {
|
||||||
|
error!("Serialization error: {}", e);
|
||||||
|
S3Error::with_message(S3ErrorCode::InternalError, "Failed to serialize targets".to_string())
|
||||||
|
})?;
|
||||||
let json_targets = serde_json::to_vec(&targets).map_err(|e| {
|
let json_targets = serde_json::to_vec(&targets).map_err(|e| {
|
||||||
error!("Serialization error: {}", e);
|
error!("Serialization error: {}", e);
|
||||||
S3Error::with_message(S3ErrorCode::InternalError, "Failed to serialize targets".to_string())
|
S3Error::with_message(S3ErrorCode::InternalError, "Failed to serialize targets".to_string())
|
||||||
@@ -1618,7 +1648,6 @@ mod tests {
|
|||||||
("credentials.expiration", serde_json::json!("2026-01-01T00:00:00Z")),
|
("credentials.expiration", serde_json::json!("2026-01-01T00:00:00Z")),
|
||||||
("api", serde_json::json!("s3v2")),
|
("api", serde_json::json!("s3v2")),
|
||||||
("disableProxy", serde_json::json!(true)),
|
("disableProxy", serde_json::json!(true)),
|
||||||
("healthCheckDuration", serde_json::json!(5)),
|
|
||||||
("edge", serde_json::json!(true)),
|
("edge", serde_json::json!(true)),
|
||||||
("edgeSyncBeforeExpiry", serde_json::json!(true)),
|
("edgeSyncBeforeExpiry", serde_json::json!(true)),
|
||||||
] {
|
] {
|
||||||
@@ -1679,6 +1708,65 @@ mod tests {
|
|||||||
assert!(!message.contains("secret"));
|
assert!(!message.contains("secret"));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn remote_target_request_accepts_go_duration_wire_values() {
|
||||||
|
// `mc replicate add` defaults `--healthcheck-seconds` to 60; madmin
|
||||||
|
// serializes Go `time.Duration` fields as nanosecond integers.
|
||||||
|
let mut request = valid_remote_target_request();
|
||||||
|
request["healthCheckDuration"] = serde_json::json!(60_000_000_000u64);
|
||||||
|
request["totalDowntime"] = serde_json::json!(90_000_000_000u64);
|
||||||
|
|
||||||
|
let target = serde_json::from_value::<RemoteTargetRequest>(request)
|
||||||
|
.expect("madmin-shaped request should deserialize")
|
||||||
|
.into_bucket_target()
|
||||||
|
.expect("mc default healthCheckDuration must be accepted");
|
||||||
|
|
||||||
|
assert_eq!(target.health_check_duration, std::time::Duration::from_secs(60));
|
||||||
|
assert_eq!(target.total_downtime, std::time::Duration::from_secs(90));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn remote_target_request_accepts_legacy_seconds_health_check() {
|
||||||
|
// Older RustFS clients sent these duration fields as plain seconds.
|
||||||
|
let mut request = valid_remote_target_request();
|
||||||
|
request["healthCheckDuration"] = serde_json::json!(60);
|
||||||
|
|
||||||
|
let target = serde_json::from_value::<RemoteTargetRequest>(request)
|
||||||
|
.expect("request should deserialize")
|
||||||
|
.into_bucket_target()
|
||||||
|
.expect("legacy seconds healthCheckDuration must be accepted");
|
||||||
|
|
||||||
|
assert_eq!(target.health_check_duration, std::time::Duration::from_secs(60));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn list_remote_targets_response_encodes_go_durations_as_nanoseconds() {
|
||||||
|
// madmin (and therefore mc) decode healthCheckDuration/totalDowntime
|
||||||
|
// as Go time.Duration nanoseconds; the persisted format stays seconds.
|
||||||
|
let target = BucketTarget {
|
||||||
|
endpoint: "192.168.1.10:9000".to_string(),
|
||||||
|
target_bucket: "target".to_string(),
|
||||||
|
health_check_duration: std::time::Duration::from_secs(60),
|
||||||
|
total_downtime: std::time::Duration::from_secs(90),
|
||||||
|
..Default::default()
|
||||||
|
};
|
||||||
|
|
||||||
|
let value = super::remote_target_admin_json(&target).expect("admin response should serialize");
|
||||||
|
|
||||||
|
assert_eq!(value["healthCheckDuration"], 60_000_000_000u64);
|
||||||
|
assert_eq!(value["totalDowntime"], 90_000_000_000u64);
|
||||||
|
// Persistence keeps seconds: the response path must not leak into it.
|
||||||
|
let persisted = serde_json::to_value(&target).expect("persisted form should serialize");
|
||||||
|
assert_eq!(persisted["healthCheckDuration"], 60);
|
||||||
|
assert_eq!(persisted["totalDowntime"], 90);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn remote_target_health_check_duration_is_declared_writable() {
|
||||||
|
assert!(REMOTE_TARGET_WRITABLE_FIELDS.contains(&"healthCheckDuration"));
|
||||||
|
assert!(!REMOTE_TARGET_UNSUPPORTED_FIELDS.contains(&"healthCheckDuration"));
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn remote_target_capability_fields_do_not_overlap() {
|
fn remote_target_capability_fields_do_not_overlap() {
|
||||||
for field in REMOTE_TARGET_UNSUPPORTED_FIELDS {
|
for field in REMOTE_TARGET_UNSUPPORTED_FIELDS {
|
||||||
|
|||||||
@@ -1279,6 +1279,14 @@ mod tests {
|
|||||||
.iter()
|
.iter()
|
||||||
.any(|field| field.name == "disableProxy" && field.state == super::ReplicationFieldState::Unsupported)
|
.any(|field| field.name == "disableProxy" && field.state == super::ReplicationFieldState::Unsupported)
|
||||||
);
|
);
|
||||||
|
assert!(
|
||||||
|
response
|
||||||
|
.replication
|
||||||
|
.remote_targets
|
||||||
|
.fields
|
||||||
|
.iter()
|
||||||
|
.any(|field| field.name == "healthCheckDuration" && field.state == super::ReplicationFieldState::Supported)
|
||||||
|
);
|
||||||
assert_eq!(response.manual_transition_jobs.contract_version, 1);
|
assert_eq!(response.manual_transition_jobs.contract_version, 1);
|
||||||
assert_eq!(response.manual_transition_jobs.status.state, CapabilityState::Supported);
|
assert_eq!(response.manual_transition_jobs.status.state, CapabilityState::Supported);
|
||||||
assert_eq!(response.manual_transition_jobs.modes, ["enqueue_only", "async"]);
|
assert_eq!(response.manual_transition_jobs.modes, ["enqueue_only", "async"]);
|
||||||
@@ -1361,6 +1369,13 @@ mod tests {
|
|||||||
.iter()
|
.iter()
|
||||||
.any(|field| field["name"] == "disableProxy" && field["state"] == "unsupported")
|
.any(|field| field["name"] == "disableProxy" && field["state"] == "unsupported")
|
||||||
);
|
);
|
||||||
|
assert!(
|
||||||
|
value["replication"]["remote_targets"]["fields"]
|
||||||
|
.as_array()
|
||||||
|
.expect("remote target fields should be an array")
|
||||||
|
.iter()
|
||||||
|
.any(|field| field["name"] == "healthCheckDuration" && field["state"] == "supported")
|
||||||
|
);
|
||||||
assert_eq!(value["manual_transition_jobs"]["contract_version"], 1);
|
assert_eq!(value["manual_transition_jobs"]["contract_version"], 1);
|
||||||
assert_eq!(value["manual_transition_jobs"]["status"]["state"], "supported");
|
assert_eq!(value["manual_transition_jobs"]["status"]["state"], "supported");
|
||||||
assert_eq!(value["manual_transition_jobs"]["modes"], json!(["enqueue_only", "async"]));
|
assert_eq!(value["manual_transition_jobs"]["modes"], json!(["enqueue_only", "async"]));
|
||||||
|
|||||||
@@ -599,6 +599,7 @@ pub(crate) mod replication {
|
|||||||
}
|
}
|
||||||
|
|
||||||
pub(crate) mod target {
|
pub(crate) mod target {
|
||||||
|
pub(crate) use super::ecstore_bucket::target::duration_from_secs_or_nanos;
|
||||||
#[allow(clippy::upper_case_acronyms)]
|
#[allow(clippy::upper_case_acronyms)]
|
||||||
pub(crate) type ARN = super::ecstore_bucket::target::ARN;
|
pub(crate) type ARN = super::ecstore_bucket::target::ARN;
|
||||||
pub(crate) type BucketTarget = super::ecstore_bucket::target::BucketTarget;
|
pub(crate) type BucketTarget = super::ecstore_bucket::target::BucketTarget;
|
||||||
|
|||||||
Reference in New Issue
Block a user