From 114b2420a224dcb7fd4530d252347a281e218b65 Mon Sep 17 00:00:00 2001 From: cxymds Date: Sun, 2 Aug 2026 22:54:38 +0800 Subject: [PATCH] feat(admin): expose versioned replication capabilities (#5631) * feat(admin): expose replication capabilities * fix(admin): route replication capabilities through facades --- Cargo.lock | 4 +- crates/ecstore/src/api/mod.rs | 4 +- crates/ecstore/src/bucket/replication/mod.rs | 4 +- .../replication_config_boundary.rs | 8 +- crates/replication/src/config.rs | 75 +++++++++ crates/replication/src/lib.rs | 4 +- rustfs/src/admin/handlers/replication.rs | 29 +++- rustfs/src/admin/handlers/system.rs | 158 ++++++++++++++++++ rustfs/src/admin/storage_api.rs | 4 + 9 files changed, 274 insertions(+), 16 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index cd8dee77a..94bf7383d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1740,9 +1740,9 @@ dependencies = [ [[package]] name = "bytesize" -version = "2.6.0" +version = "2.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "351a3e803ee3c6eaeee6b00076b767514b37c32a73d326c3ec7abddb7d6c3493" +checksum = "7354288c522e7e980fafd2075d63d1285794c3a6a16cdd492f189ea406e5f18b" [[package]] name = "bytestring" diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 342c10b56..ce1d5ec3e 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -180,7 +180,9 @@ pub mod bucket { pub use crate::bucket::replication::{ BucketReplicationResyncStatus, BucketStats, DeleteReplicationConfigSnapshot, DeletedObjectReplicationInfo, DurableMrfBacklog, DynReplicationPool, MrfOpKind, MrfReplicateEntry, MustReplicateOptions, ObjectOpts, - REPLICATE_INCOMING_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicationBatchAdmission, ReplicationConfig, + REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, + REPLICATE_INCOMING_DELETE, REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, + REPLICATION_WRITABLE_FIELDS, ReplicateDecision, ReplicateObjectInfo, ReplicationBatchAdmission, ReplicationConfig, ReplicationConfigurationExt, ReplicationDeleteScheduleInput, ReplicationDeleteStateSource, ReplicationHealQueueResult, ReplicationObjectBridge, ReplicationObjectIO, ReplicationOperation, ReplicationPoolTrait, ReplicationPriority, ReplicationQueueAdmission, ReplicationScannerBridge, ReplicationState, ReplicationStats, diff --git a/crates/ecstore/src/bucket/replication/mod.rs b/crates/ecstore/src/bucket/replication/mod.rs index 986f2b050..18c6345a0 100644 --- a/crates/ecstore/src/bucket/replication/mod.rs +++ b/crates/ecstore/src/bucket/replication/mod.rs @@ -45,7 +45,9 @@ mod runtime_boundary; pub use datatypes::ResyncStatusType; pub use replication_config_boundary::{ - ObjectOpts, ReplicationConfigurationExt, ReplicationTargetValidationError, invalid_replication_config_status_field, + ObjectOpts, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, + REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, + ReplicationConfigurationExt, ReplicationTargetValidationError, invalid_replication_config_status_field, replication_target_arns, should_remove_replication_target, unsupported_replication_config_field, validate_replication_config_target_arns, }; diff --git a/crates/ecstore/src/bucket/replication/replication_config_boundary.rs b/crates/ecstore/src/bucket/replication/replication_config_boundary.rs index 0b9f5e0ed..04d53883f 100644 --- a/crates/ecstore/src/bucket/replication/replication_config_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_config_boundary.rs @@ -13,7 +13,9 @@ // limitations under the License. pub use rustfs_replication::{ - ObjectOpts, ReplicationConfigurationExt, ReplicationRuleExt, ReplicationTargetValidationError, - invalid_replication_config_status_field, replication_target_arns, should_remove_replication_target, - unsupported_replication_config_field, validate_replication_config_target_arns, + ObjectOpts, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, + REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, + ReplicationConfigurationExt, ReplicationRuleExt, ReplicationTargetValidationError, invalid_replication_config_status_field, + replication_target_arns, should_remove_replication_target, unsupported_replication_config_field, + validate_replication_config_target_arns, }; diff --git a/crates/replication/src/config.rs b/crates/replication/src/config.rs index 42e4f3067..cdfd32eb6 100644 --- a/crates/replication/src/config.rs +++ b/crates/replication/src/config.rs @@ -26,6 +26,53 @@ use serde::{Deserialize, Serialize}; use std::collections::{HashMap, HashSet}; use uuid::Uuid; +pub const REPLICATION_CAPABILITY_CONTRACT_VERSION: u32 = 1; + +pub const REPLICATION_WRITABLE_FIELDS: &[&str] = &[ + "Role", + "Rule.ID", + "Rule.Status", + "Rule.Priority", + "Rule.Filter.Prefix", + "Rule.Filter.Tag", + "Rule.Filter.And", + "Rule.Destination.Bucket", + "Rule.ExistingObjectReplication.Status", + "Rule.DeleteMarkerReplication.Status", + "Rule.DeleteReplication.Status", + "Rule.SourceSelectionCriteria.ReplicaModifications.Status", +]; + +pub const REPLICATION_READ_ONLY_HISTORICAL_FIELDS: &[&str] = &[ + "SourceSelectionCriteria.SseKmsEncryptedObjects", + "Destination.EncryptionConfiguration", + "Destination.Metrics", + "Destination.ReplicationTime", +]; + +pub const REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION: u32 = 1; + +pub const REMOTE_TARGET_WRITABLE_FIELDS: &[&str] = &[ + "sourcebucket", + "endpoint", + "credentials.accessKey", + "credentials.secretKey", + "targetbucket", + "secure", + "path", + "api", + "arn", + "type", + "region", + "bandwidth", + "replicationSync", + "storage_class", + "skipTlsVerify", + "caCertPem", +]; + +pub const REMOTE_TARGET_UNSUPPORTED_FIELDS: &[&str] = &["disableProxy", "healthCheckDuration", "edge", "edgeSyncBeforeExpiry"]; + #[derive(Debug, Clone, Serialize, Deserialize, Default)] pub struct ObjectOpts { pub name: String, @@ -920,6 +967,34 @@ mod tests { assert_eq!(unsupported_replication_config_field(&config), None); } + #[test] + fn capability_fields_match_validator_rejections() { + let rejected_fields = [ + "SourceSelectionCriteria.SseKmsEncryptedObjects", + "Destination.EncryptionConfiguration", + "Destination.Metrics", + "Destination.ReplicationTime", + ]; + + for field in rejected_fields { + assert!( + REPLICATION_READ_ONLY_HISTORICAL_FIELDS.contains(&field), + "rejected field {field} must be advertised as readable historical data" + ); + assert!( + !REPLICATION_WRITABLE_FIELDS.contains(&field), + "rejected field {field} must not be advertised as writable" + ); + } + + for field in REPLICATION_WRITABLE_FIELDS { + assert!( + !REPLICATION_READ_ONLY_HISTORICAL_FIELDS.contains(field), + "field {field} cannot be both writable and historical-only" + ); + } + } + #[test] fn invalid_replication_status_fields_are_reported_before_persistence() { let arn = "arn:rustfs:replication:us-east-1:target:bucket"; diff --git a/crates/replication/src/lib.rs b/crates/replication/src/lib.rs index 871059692..b60c6aab7 100644 --- a/crates/replication/src/lib.rs +++ b/crates/replication/src/lib.rs @@ -29,7 +29,9 @@ mod storage_api; pub mod tagging; pub use config::{ - ObjectOpts, ReplicationConfigurationExt, ReplicationTargetValidationError, active_replication_rule_destination_arns, + ObjectOpts, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, + REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, + ReplicationConfigurationExt, ReplicationTargetValidationError, active_replication_rule_destination_arns, invalid_replication_config_status_field, replication_target_arns, should_remove_replication_target, unsupported_replication_config_field, validate_replication_config_target_arns, }; diff --git a/rustfs/src/admin/handlers/replication.rs b/rustfs/src/admin/handlers/replication.rs index 27f85d764..3cc890b91 100644 --- a/rustfs/src/admin/handlers/replication.rs +++ b/rustfs/src/admin/handlers/replication.rs @@ -22,6 +22,9 @@ use crate::admin::runtime_sources::{ use crate::admin::storage_api::bucket::metadata::BUCKET_TARGETS_FILE; use crate::admin::storage_api::bucket::metadata_sys; use crate::admin::storage_api::bucket::metadata_sys::get_replication_config; +use crate::admin::storage_api::bucket::replication::REMOTE_TARGET_UNSUPPORTED_FIELDS; +#[cfg(test)] +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::target::{BucketTarget, BucketTargetType, Credentials as TargetCredentials, LatencyStat}; use crate::admin::storage_api::bucket::target_sys::{BucketTargetError, BucketTargetSys}; @@ -207,12 +210,12 @@ impl RemoteTargetRequest { )); } - for (unsupported, configured) in [ - ("disableProxy", self.disable_proxy), - ("healthCheckDuration", self.health_check_duration != 0), - ("edge", self.edge), - ("edgeSyncBeforeExpiry", self.edge_sync_before_expiry), - ] { + for (unsupported, configured) in REMOTE_TARGET_UNSUPPORTED_FIELDS.iter().copied().zip([ + self.disable_proxy, + self.health_check_duration != 0, + self.edge, + self.edge_sync_before_expiry, + ]) { if configured { return Err(s3_error!( InvalidRequest, @@ -1049,8 +1052,8 @@ impl Operation for ReplicationMrfHandler { #[cfg(test)] mod tests { use super::{ - RemoteTargetRequest, SUPPORTED_REMOTE_TARGET_API, build_mrf_response, extract_query_params, unique_replication_peers, - validate_remote_target_tls_settings, + REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, RemoteTargetRequest, SUPPORTED_REMOTE_TARGET_API, + build_mrf_response, extract_query_params, unique_replication_peers, validate_remote_target_tls_settings, }; use crate::admin::storage_api::bucket::target::BucketTarget; use crate::admin::storage_api::replication::{BucketStats, DurableMrfBacklog, MrfOpKind, MrfReplicateEntry}; @@ -1411,6 +1414,16 @@ mod tests { assert!(!message.contains("secret")); } + #[test] + fn remote_target_capability_fields_do_not_overlap() { + for field in REMOTE_TARGET_UNSUPPORTED_FIELDS { + assert!( + !REMOTE_TARGET_WRITABLE_FIELDS.contains(field), + "remote target field {field} cannot be both writable and unsupported" + ); + } + } + #[test] fn remote_target_request_converts_to_bucket_target() { let target = serde_json::from_value::(valid_remote_target_request()) diff --git a/rustfs/src/admin/handlers/system.rs b/rustfs/src/admin/handlers/system.rs index 010bc5eb0..4c5ebe4cb 100644 --- a/rustfs/src/admin/handlers/system.rs +++ b/rustfs/src/admin/handlers/system.rs @@ -21,6 +21,10 @@ use crate::admin::router::{AdminOperation, Operation, S3Router}; use crate::admin::runtime_sources::{ DefaultAdminUsecase, QueryServerInfoRequest, current_endpoints_handle, default_admin_usecase, object_store_from_req, }; +use crate::admin::storage_api::bucket::replication::{ + REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, + REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, +}; use crate::admin::storage_api::cluster::{ CapabilityState, CapabilityStatus, ObservabilitySnapshotProvider, TopologySnapshot, TopologySnapshotProvider, }; @@ -644,6 +648,7 @@ pub struct RuntimeCapabilitiesSummary { #[derive(Debug, Clone, PartialEq, Eq, Serialize)] pub struct RuntimeCapabilitiesResponse { pub summary: RuntimeCapabilitiesSummary, + pub replication: ReplicationCapabilities, pub manual_transition_jobs: ManualTransitionJobCapabilities, pub diagnostic_probes: DiagnosticProbeCapabilities, pub inspect_archive: super::inspect_archive::InspectArchiveCapability, @@ -657,6 +662,100 @@ pub struct RuntimeCapabilitiesResponse { pub topology_status: CapabilityStatus, } +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum ReplicationFieldState { + Supported, + ReadOnlyHistorical, + Unsupported, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +pub struct ReplicationFieldCapability { + pub name: &'static str, + pub state: ReplicationFieldState, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub struct ReplicationFeatureCapabilities { + pub contract_version: u32, + pub status: CapabilityStatus, + pub fields: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub struct ReplicationCapabilities { + pub contract_version: u32, + pub bucket_replication: ReplicationFeatureCapabilities, + pub remote_targets: ReplicationFeatureCapabilities, + pub mixed_version_policy: &'static str, +} + +impl ReplicationCapabilities { + fn current() -> Self { + let remote_target_routes = [ + admin_route_capability(HttpMethod::Get, "/rustfs/admin/v3/list-remote-targets"), + admin_route_capability(HttpMethod::Put, "/rustfs/admin/v3/set-remote-target"), + admin_route_capability(HttpMethod::Delete, "/rustfs/admin/v3/remove-remote-target"), + ]; + let remote_targets_supported = remote_target_routes + .iter() + .all(|status| status.state == CapabilityState::Supported); + + Self { + contract_version: REPLICATION_CAPABILITY_CONTRACT_VERSION, + bucket_replication: ReplicationFeatureCapabilities { + contract_version: REPLICATION_CAPABILITY_CONTRACT_VERSION, + status: CapabilityStatus::supported() + .with_reason("bucket replication validation and execution contracts are registered"), + fields: REPLICATION_WRITABLE_FIELDS + .iter() + .copied() + .map(|name| ReplicationFieldCapability { + name, + state: ReplicationFieldState::Supported, + }) + .chain( + REPLICATION_READ_ONLY_HISTORICAL_FIELDS + .iter() + .copied() + .map(|name| ReplicationFieldCapability { + name, + state: ReplicationFieldState::ReadOnlyHistorical, + }), + ) + .collect(), + }, + remote_targets: ReplicationFeatureCapabilities { + contract_version: REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, + status: if remote_targets_supported { + CapabilityStatus::supported().with_reason("remote target routes and field validation are registered") + } else { + CapabilityStatus::unsupported().with_reason("one or more remote target routes are unavailable") + }, + fields: REMOTE_TARGET_WRITABLE_FIELDS + .iter() + .copied() + .map(|name| ReplicationFieldCapability { + name, + state: ReplicationFieldState::Supported, + }) + .chain( + REMOTE_TARGET_UNSUPPORTED_FIELDS + .iter() + .copied() + .map(|name| ReplicationFieldCapability { + name, + state: ReplicationFieldState::Unsupported, + }), + ) + .collect(), + }, + mixed_version_policy: "fail_closed_when_capability_unknown_or_unsupported", + } + } +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize)] pub struct ManualTransitionJobCapabilities { pub contract_version: u32, @@ -871,6 +970,7 @@ pub(crate) async fn build_runtime_capabilities_response() Ok(RuntimeCapabilitiesResponse { summary, + replication: ReplicationCapabilities::current(), manual_transition_jobs: ManualTransitionJobCapabilities::current(), diagnostic_probes: DiagnosticProbeCapabilities::current(), inspect_archive: super::inspect_archive::InspectArchiveCapability::current(local_drive_count), @@ -1144,6 +1244,41 @@ mod tests { assert_eq!(response.summary.site_replication_resync.state, CapabilityState::Supported); assert_eq!(response.summary.site_replication_repair.state, CapabilityState::Supported); assert_eq!(response.summary.manual_transition_jobs.state, CapabilityState::Supported); + assert_eq!(response.replication.contract_version, 1); + assert_eq!(response.replication.bucket_replication.contract_version, 1); + assert_eq!(response.replication.remote_targets.contract_version, 1); + assert_eq!(response.replication.bucket_replication.status.state, CapabilityState::Supported); + assert_eq!(response.replication.remote_targets.status.state, CapabilityState::Supported); + assert_eq!( + response.replication.mixed_version_policy, + "fail_closed_when_capability_unknown_or_unsupported" + ); + assert!( + response + .replication + .bucket_replication + .fields + .iter() + .any(|field| field.name == "Rule.DeleteReplication.Status" + && field.state == super::ReplicationFieldState::Supported) + ); + assert!( + response + .replication + .bucket_replication + .fields + .iter() + .any(|field| field.name == "Destination.EncryptionConfiguration" + && field.state == super::ReplicationFieldState::ReadOnlyHistorical) + ); + assert!( + response + .replication + .remote_targets + .fields + .iter() + .any(|field| field.name == "disableProxy" && field.state == super::ReplicationFieldState::Unsupported) + ); 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.modes, ["enqueue_only", "async"]); @@ -1203,6 +1338,29 @@ mod tests { assert_eq!(value["summary"]["site_replication_resync"]["state"], "supported"); assert_eq!(value["summary"]["site_replication_repair"]["state"], "supported"); assert_eq!(value["summary"]["manual_transition_jobs"]["state"], "supported"); + assert_eq!(value["replication"]["contract_version"], 1); + assert_eq!(value["replication"]["bucket_replication"]["contract_version"], 1); + assert_eq!(value["replication"]["remote_targets"]["contract_version"], 1); + assert_eq!(value["replication"]["bucket_replication"]["status"]["state"], "supported"); + assert_eq!(value["replication"]["remote_targets"]["status"]["state"], "supported"); + assert_eq!( + value["replication"]["mixed_version_policy"], + "fail_closed_when_capability_unknown_or_unsupported" + ); + assert!( + value["replication"]["bucket_replication"]["fields"] + .as_array() + .expect("bucket replication fields should be an array") + .iter() + .any(|field| field["name"] == "Destination.EncryptionConfiguration" && field["state"] == "read_only_historical") + ); + assert!( + value["replication"]["remote_targets"]["fields"] + .as_array() + .expect("remote target fields should be an array") + .iter() + .any(|field| field["name"] == "disableProxy" && field["state"] == "unsupported") + ); assert_eq!(value["manual_transition_jobs"]["contract_version"], 1); assert_eq!(value["manual_transition_jobs"]["status"]["state"], "supported"); assert_eq!(value["manual_transition_jobs"]["modes"], json!(["enqueue_only", "async"])); diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index cd437819e..3feba88a6 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -372,6 +372,10 @@ pub(crate) mod quota { } pub(crate) mod replication { + pub(crate) use super::ecstore_bucket::replication::{ + REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, + REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, + }; pub(crate) type BucketReplicationResyncStatus = super::ecstore_bucket::replication::BucketReplicationResyncStatus; pub(crate) type BucketStats = super::ecstore_bucket::replication::BucketStats; pub(crate) type ReplicationStatusType = super::ecstore_bucket::replication::ReplicationStatusType;