gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-02 19:39:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(admin): expose replication capabilities

Browse Source
This commit is contained in:
马登山
2026-08-02 18:51:29 +08:00
parent c104ba23d4
commit ce9f04f28e
6 changed files with 263 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1
View File
@@ -9080,6 +9080,7 @@ dependencies = [
"rustfs-policy",
"rustfs-protocols",
"rustfs-protos",
"rustfs-replication",
"rustfs-rio",
"rustfs-s3-ops",
"rustfs-s3-types",
crates/replication/src/config.rs
+75
View File
@@ -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,
@@ -831,6 +878,34 @@ mod tests {
assert_eq!(unsupported_replication_config_field(&config), Some("Destination.ReplicationTime"));
}
#[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";
crates/replication/src/lib.rs
+3 -1
View File
@@ -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,
};
rustfs/Cargo.toml
+1
View File
@@ -217,6 +217,7 @@ rustfs-notify = { workspace = true }
rustfs-obs = { workspace = true }
rustfs-policy = { workspace = true }
rustfs-protocols = { workspace = true }
rustfs-replication = { workspace = true }
rustfs-protos = { workspace = true }
rustfs-rio = { workspace = true }
rustfs-s3-types = { workspace = true }
rustfs/src/admin/handlers/replication.rs
+25 -14
View File
@@ -40,6 +40,7 @@ use matchit::Params;
use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE;
use rustfs_credentials::Credentials;
use rustfs_policy::policy::action::{Action, AdminAction};
use rustfs_replication::{REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS};
use s3s::header::CONTENT_TYPE;
use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error};
use serde::{Deserialize, Serialize};
@@ -179,12 +180,12 @@ impl RemoteTargetRequest {
return Err(s3_error!(InvalidRequest, "credentials.secretKey is required"));
}
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,
@@ -1021,8 +1022,8 @@ impl Operation for ReplicationMrfHandler {
#[cfg(test)]
mod tests {
use super::{
RemoteTargetRequest, build_mrf_response, extract_query_params, unique_replication_peers,
validate_remote_target_tls_settings,
REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, RemoteTargetRequest, 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};
@@ -1313,12 +1314,12 @@ mod tests {
#[test]
fn remote_target_request_rejects_unimplemented_fields() {
for (field, value) in [
("disableProxy", serde_json::json!(true)),
("healthCheckDuration", serde_json::json!(5)),
("edge", serde_json::json!(true)),
("edgeSyncBeforeExpiry", serde_json::json!(true)),
] {
for (field, value) in REMOTE_TARGET_UNSUPPORTED_FIELDS.iter().copied().zip([
serde_json::json!(true),
serde_json::json!(5),
serde_json::json!(true),
serde_json::json!(true),
]) {
let mut request = valid_remote_target_request();
request[field] = value;
let request: RemoteTargetRequest =
@@ -1332,6 +1333,16 @@ mod tests {
}
}
#[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::<RemoteTargetRequest>(valid_remote_target_request())
rustfs/src/admin/handlers/system.rs
+158
View File
@@ -35,6 +35,10 @@ use matchit::Params;
use rustfs_concurrency::WorkloadAdmissionRegistrySnapshot;
use rustfs_madmin::{InfoMessage, StorageInfo};
use rustfs_policy::policy::action::{Action, AdminAction, S3Action};
use rustfs_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 rustfs_security_governance::{AdminRouteSpec, HttpMethod};
use s3s::header::CONTENT_TYPE;
use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error};
@@ -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<ReplicationFieldCapability>,
}
#[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"]));