Compare commits

..

1 Commits

Author SHA1 Message Date
唐小鸭 9ed8d38a81 fix(admin): expose per-target disableProxy through remote target admin API
The read-proxy selector already honors a target's disable_proxy flag
(PR #6172), but the admin API still rejected the field, so the only way
to set it was importing a MinIO-written bucket-targets.json.

- move disableProxy from REMOTE_TARGET_UNSUPPORTED_FIELDS to
  REMOTE_TARGET_WRITABLE_FIELDS (set-remote-target create accepts it)
- add TargetUpdateOp::Proxy so set-remote-target?update=true&proxy=true
  overlays only the proxy group (MinIO TargetUpdateType parity)
- bump REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION 1 -> 2 and update the
  runtime capability pin tests
- keep edge/edgeSyncBeforeExpiry rejected (no implementation behind them)
- pin that a published TargetClient carries disable_proxy, the field the
  proxy-target selector consults

Refs rustfs/backlog#1950
2026-08-21 19:18:48 +08:00
5 changed files with 121 additions and 152 deletions
@@ -3425,6 +3425,44 @@ mod tests {
assert!(mutexes.contains_key("second"));
}
#[tokio::test]
async fn update_all_targets_publishes_disable_proxy_on_target_client() {
// The read-proxy selector (replication_proxy::get_proxy_targets) skips
// targets whose TargetClient carries disable_proxy — the persisted
// per-target opt-out must survive client publication.
let sys = BucketTargetSys::default();
let target = |arn: &str, disable_proxy: bool| BucketTarget {
arn: arn.to_string(),
endpoint: "192.168.1.10:9000".to_string(),
target_bucket: "target-bucket".to_string(),
region: "us-east-1".to_string(),
disable_proxy,
credentials: Some(Credentials {
access_key: "access".to_string(),
secret_key: "secret".to_string(),
session_token: None,
expiration: None,
}),
..Default::default()
};
let targets = BucketTargets {
targets: vec![target("arn:proxied", false), target("arn:opted-out", true)],
};
sys.update_all_targets("bucket", Some(&targets)).await;
let proxied = sys
.get_remote_target_client("bucket", "arn:proxied")
.await
.expect("client should be published");
assert!(!proxied.disable_proxy);
let opted_out = sys
.get_remote_target_client("bucket", "arn:opted-out")
.await
.expect("client should be published");
assert!(opted_out.disable_proxy, "disable_proxy must reach the published TargetClient");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn target_updates_serialize_client_build_through_publication_per_bucket() {
let sys = Arc::new(BucketTargetSys::default());
@@ -32,10 +32,6 @@ pub struct Credentials {
pub access_key: String,
#[serde(rename = "secretKey")]
pub secret_key: String,
// The aliases accept madmin's JSON tags (MinIO-written bucket-targets
// metadata and mc request bodies) without changing the snake_case
// persisted/peer wire format this struct serializes to.
#[serde(alias = "sessionToken")]
pub session_token: Option<String>,
pub expiration: Option<Timestamp>,
}
@@ -211,7 +207,7 @@ pub struct BucketTarget {
#[serde(rename = "replicationSync", default)]
pub replication_sync: bool,
#[serde(alias = "storageclass", default)]
#[serde(default)]
pub storage_class: String,
#[serde(rename = "skipTlsVerify", default)]
pub skip_tls_verify: bool,
@@ -224,7 +220,7 @@ pub struct BucketTarget {
#[serde(rename = "resetBeforeDate", with = "time::serde::rfc3339::option", default)]
pub reset_before_date: Option<OffsetDateTime>,
#[serde(alias = "resetID", default)]
#[serde(default)]
pub reset_id: String,
#[serde(rename = "totalDowntime", with = "duration_seconds", default)]
pub total_downtime: Duration,
@@ -237,7 +233,7 @@ pub struct BucketTarget {
#[serde(default)]
pub latency: LatencyStat,
#[serde(alias = "deploymentID", default)]
#[serde(default)]
pub deployment_id: String,
#[serde(default)]
@@ -535,84 +531,6 @@ mod tests {
assert_eq!(value["totalDowntime"], 90);
}
#[test]
fn bucket_target_persisted_wire_keys_stay_snake_case() {
// bucket-targets.json (persisted via `serde_json::to_vec(&BucketTargets)`
// in the admin set/remove handlers) and the msgpack struct-map form
// (`BucketTargets::marshal_msg`) both come straight from this struct's
// serde field names. madmin naming is applied only in the admin
// response layer (`remote_target_admin_json`); renaming here would
// silently break every existing deployment's persisted metadata.
let targets = BucketTargets {
targets: vec![BucketTarget {
credentials: Some(Credentials {
access_key: "ak".to_string(),
secret_key: "sk".to_string(),
session_token: Some("token".to_string()),
expiration: None,
}),
bandwidth_limit: 5,
storage_class: "STANDARD".to_string(),
reset_id: "reset-1".to_string(),
deployment_id: "deploy-1".to_string(),
..Default::default()
}],
};
let json = serde_json::to_value(&targets).expect("targets should serialize to JSON");
let msgpack: serde_json::Value =
rmp_serde::from_slice(&targets.marshal_msg().expect("targets should marshal to msgpack"))
.expect("msgpack struct map should decode into a JSON value");
for (wire, entry) in [("JSON", &json["targets"][0]), ("msgpack", &msgpack["targets"][0])] {
assert_eq!(entry["bandwidth_limit"], 5, "{wire} key `bandwidth_limit` must stay");
assert_eq!(entry["storage_class"], "STANDARD", "{wire} key `storage_class` must stay");
assert_eq!(entry["reset_id"], "reset-1", "{wire} key `reset_id` must stay");
assert_eq!(entry["deployment_id"], "deploy-1", "{wire} key `deployment_id` must stay");
assert_eq!(entry["credentials"]["session_token"], "token", "{wire} key `session_token` must stay");
}
}
#[test]
fn minio_written_bucket_targets_json_populates_madmin_named_fields() {
// A MinIO-written bucket-targets.json carries madmin's JSON tags
// (`bandwidth`, `storageclass`, `resetID`, `deploymentID`,
// `credentials.sessionToken`). On migration these must land in the
// matching fields instead of silently defaulting (backlog#1946).
let targets: BucketTargets = serde_json::from_value(serde_json::json!({
"targets": [{
"sourcebucket": "src",
"endpoint": "minio.example:9000",
"credentials": {
"accessKey": "ak",
"secretKey": "sk",
"sessionToken": "minio-session-token"
},
"targetbucket": "dst",
"type": "replication",
"replicationSync": true,
"bandwidth": 107374182400i64,
"storageclass": "STANDARD",
"resetID": "reset-789",
"deploymentID": "deploy-123"
}]
}))
.expect("MinIO-written bucket-targets.json must deserialize");
let target = &targets.targets[0];
assert_eq!(target.bandwidth_limit, 107374182400);
assert_eq!(target.storage_class, "STANDARD");
assert_eq!(target.reset_id, "reset-789");
assert_eq!(target.deployment_id, "deploy-123");
assert_eq!(
target
.credentials
.as_ref()
.and_then(|credentials| credentials.session_token.as_deref()),
Some("minio-session-token")
);
}
#[test]
fn test_bucket_target_debug_redacts_credentials() {
let target = BucketTarget {
+7 -2
View File
@@ -60,7 +60,9 @@ pub const REPLICATION_READ_ONLY_HISTORICAL_FIELDS: &[&str] = &[
"Destination.ReplicationTime",
];
pub const REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION: u32 = 1;
// v2: disableProxy moved from unsupported to writable (per-target read-proxy
// opt-out is accepted by set-remote-target and the `proxy` update op).
pub const REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION: u32 = 2;
pub const REMOTE_TARGET_WRITABLE_FIELDS: &[&str] = &[
"sourcebucket",
@@ -83,9 +85,12 @@ pub const REMOTE_TARGET_WRITABLE_FIELDS: &[&str] = &[
// madmin default of 60s); the per-target health-check interval is not
// yet applied — the heartbeat keeps its global env-configured interval.
"healthCheckDuration",
// Per-target read-proxy opt-out, consumed by the proxy-target selector
// (contract v2; previously only importable via MinIO bucket-targets.json).
"disableProxy",
];
pub const REMOTE_TARGET_UNSUPPORTED_FIELDS: &[&str] = &["disableProxy", "edge", "edgeSyncBeforeExpiry"];
pub const REMOTE_TARGET_UNSUPPORTED_FIELDS: &[&str] = &["edge", "edgeSyncBeforeExpiry"];
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct ObjectOpts {
+52 -61
View File
@@ -73,6 +73,8 @@ enum TargetUpdateOp {
/// Connection group: credentials plus endpoint, target bucket, and TLS settings.
Credentials,
Sync,
/// Per-target read-proxy opt-out (`disableProxy`).
Proxy,
Bandwidth,
Path,
}
@@ -81,12 +83,13 @@ fn parse_remote_target_update_ops(queries: &HashMap<String, String>) -> S3Result
const SUPPORTED_OPS: &[(&str, TargetUpdateOp)] = &[
("creds", TargetUpdateOp::Credentials),
("sync", TargetUpdateOp::Sync),
("proxy", TargetUpdateOp::Proxy),
("bandwidth", TargetUpdateOp::Bandwidth),
("path", TargetUpdateOp::Path),
];
// Present in the MinIO wire contract, but they drive target fields this
// version rejects as unsupported — fail loudly instead of silently ignoring.
const UNSUPPORTED_OPS: &[&str] = &["proxy", "healthcheck", "edge", "edgeSyncBeforeExpiry"];
const UNSUPPORTED_OPS: &[&str] = &["healthcheck", "edge", "edgeSyncBeforeExpiry"];
for key in UNSUPPORTED_OPS {
if queries.get(*key).is_some_and(|value| value == "true") {
@@ -312,11 +315,10 @@ impl RemoteTargetRequest {
));
}
for (unsupported, configured) in
REMOTE_TARGET_UNSUPPORTED_FIELDS
.iter()
.copied()
.zip([self.disable_proxy, self.edge, self.edge_sync_before_expiry])
for (unsupported, configured) in REMOTE_TARGET_UNSUPPORTED_FIELDS
.iter()
.copied()
.zip([self.edge, self.edge_sync_before_expiry])
{
if configured {
return Err(s3_error!(
@@ -374,10 +376,8 @@ impl RemoteTargetRequest {
/// Admin-response encoding of a remote target: the persisted bucket-targets
/// format keeps `healthCheckDuration`/`totalDowntime` in seconds and the
/// `latency` stats in milliseconds, but madmin decodes all of them as Go
/// `time.Duration` (nanoseconds) — and it looks the fields up under its own
/// JSON tags (`bandwidth`, `storageclass`, `resetID`, `deploymentID`,
/// `credentials.sessionToken`), not the persisted snake_case keys. Re-encode
/// just those fields here without touching the persistence wire format.
/// `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)
@@ -385,12 +385,6 @@ fn remote_target_admin_json(target: &BucketTarget) -> Result<serde_json::Value,
u64::try_from(duration.as_nanos()).unwrap_or(u64::MAX).into()
}
fn rename_key(value: &mut serde_json::Value, from: &str, to: &str) {
if let Some(moved) = value.as_object_mut().and_then(|object| object.remove(from)) {
value[to] = moved;
}
}
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);
@@ -399,11 +393,6 @@ fn remote_target_admin_json(target: &BucketTarget) -> Result<serde_json::Value,
"avg": go_duration_nanos(target.latency.avg),
"max": go_duration_nanos(target.latency.max),
});
rename_key(&mut value, "bandwidth_limit", "bandwidth");
rename_key(&mut value, "storage_class", "storageclass");
rename_key(&mut value, "reset_id", "resetID");
rename_key(&mut value, "deployment_id", "deploymentID");
rename_key(&mut value["credentials"], "session_token", "sessionToken");
Ok(value)
}
@@ -715,6 +704,7 @@ impl Operation for SetRemoteTargetHandler {
target.deployment_id = remote_target.deployment_id.clone();
}
TargetUpdateOp::Sync => target.replication_sync = remote_target.replication_sync,
TargetUpdateOp::Proxy => target.disable_proxy = remote_target.disable_proxy,
TargetUpdateOp::Bandwidth => target.bandwidth_limit = remote_target.bandwidth_limit,
TargetUpdateOp::Path => target.path = remote_target.path.clone(),
}
@@ -1486,7 +1476,7 @@ mod tests {
parse_remote_target_update_ops, render_mrf_backlog, render_replication_diff, unique_replication_peers,
validate_remote_target_tls_settings,
};
use crate::admin::storage_api::bucket::target::{BucketTarget, Credentials as TargetCredentials, LatencyStat};
use crate::admin::storage_api::bucket::target::{BucketTarget, LatencyStat};
use crate::admin::storage_api::replication::{BucketStats, DurableMrfBacklog, MrfOpKind, MrfReplicateEntry};
use http::Uri;
@@ -1533,6 +1523,7 @@ mod tests {
("update", "true"),
("creds", "true"),
("sync", "true"),
("proxy", "true"),
("bandwidth", "true"),
("path", "true"),
]))
@@ -1542,6 +1533,7 @@ mod tests {
vec![
TargetUpdateOp::Credentials,
TargetUpdateOp::Sync,
TargetUpdateOp::Proxy,
TargetUpdateOp::Bandwidth,
TargetUpdateOp::Path
]
@@ -2083,7 +2075,6 @@ mod tests {
("credentials.session_token", serde_json::json!("session-token")),
("credentials.expiration", serde_json::json!("2026-01-01T00:00:00Z")),
("api", serde_json::json!("s3v2")),
("disableProxy", serde_json::json!(true)),
("edge", serde_json::json!(true)),
("edgeSyncBeforeExpiry", serde_json::json!(true)),
] {
@@ -2281,44 +2272,6 @@ mod tests {
assert_eq!(persisted["latency"]["max"], 250);
}
#[test]
fn list_remote_targets_response_uses_madmin_key_names() {
// madmin's BucketTarget JSON tags are `bandwidth`, `storageclass`,
// `resetID`, `deploymentID`, and `credentials.sessionToken`
// (backlog#1946); the persisted snake_case keys decode to zero values
// in mc, blanking the bandwidth and reset-id columns of
// `mc replicate ls`.
let target = BucketTarget {
endpoint: "192.168.1.10:9000".to_string(),
target_bucket: "target".to_string(),
credentials: Some(TargetCredentials {
access_key: "access".to_string(),
secret_key: String::new(),
session_token: Some("session-token".to_string()),
expiration: None,
}),
bandwidth_limit: 107_374_182_400,
storage_class: "STANDARD".to_string(),
reset_id: "reset-123".to_string(),
deployment_id: "deploy-456".to_string(),
..Default::default()
};
let value = super::remote_target_admin_json(&target).expect("admin response should serialize");
assert_eq!(value["bandwidth"], 107_374_182_400i64);
assert_eq!(value["storageclass"], "STANDARD");
assert_eq!(value["resetID"], "reset-123");
assert_eq!(value["deploymentID"], "deploy-456");
assert_eq!(value["credentials"]["sessionToken"], "session-token");
// The madmin keys replace the snake_case ones rather than duplicating
// them next to each other.
for stale in ["bandwidth_limit", "storage_class", "reset_id", "deployment_id"] {
assert!(value.get(stale).is_none(), "admin response must not carry `{stale}`");
}
assert!(value["credentials"].get("session_token").is_none());
}
#[test]
fn remote_target_admin_json_latency_round_trips_through_go_duration() {
// Round trip: a madmin reader decodes the latency values as Go
@@ -2351,6 +2304,44 @@ mod tests {
assert!(!REMOTE_TARGET_UNSUPPORTED_FIELDS.contains(&"healthCheckDuration"));
}
#[test]
fn remote_target_disable_proxy_is_declared_writable_edge_stays_unsupported() {
assert!(REMOTE_TARGET_WRITABLE_FIELDS.contains(&"disableProxy"));
assert!(!REMOTE_TARGET_UNSUPPORTED_FIELDS.contains(&"disableProxy"));
// edge sync has no implementation behind it — it must stay rejected.
assert!(REMOTE_TARGET_UNSUPPORTED_FIELDS.contains(&"edge"));
assert!(REMOTE_TARGET_UNSUPPORTED_FIELDS.contains(&"edgeSyncBeforeExpiry"));
}
#[test]
fn remote_target_create_accepts_disable_proxy() {
let mut request = valid_remote_target_request();
request["disableProxy"] = serde_json::json!(true);
let target = serde_json::from_value::<RemoteTargetRequest>(request)
.expect("request should deserialize")
.into_bucket_target()
.expect("disableProxy is a supported per-target read-proxy opt-out");
assert!(target.disable_proxy);
}
#[test]
fn update_body_with_proxy_op_toggles_disable_proxy_without_credentials() {
// Mirrors the other partial-update groups: a proxy-only update body may
// omit the connection fields entirely.
let body = serde_json::json!({
"arn": "arn:rustfs:replication:us-east-1:dep:target",
"type": "replication",
"disableProxy": true
});
let request: RemoteTargetRequest = serde_json::from_value(body).expect("partial update body should deserialize");
let target = request
.into_update_bucket_target(&[TargetUpdateOp::Proxy])
.expect("proxy-only update must not require credentials");
assert!(target.disable_proxy);
}
#[test]
fn remote_target_capability_fields_do_not_overlap() {
for field in REMOTE_TARGET_UNSUPPORTED_FIELDS {
+21 -4
View File
@@ -1262,7 +1262,9 @@ mod tests {
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);
// v2: disableProxy moved from unsupported to writable (per-target
// read-proxy opt-out reached the admin API).
assert_eq!(response.replication.remote_targets.contract_version, 2);
assert_eq!(response.replication.bucket_replication.status.state, CapabilityState::Supported);
assert_eq!(response.replication.remote_targets.status.state, CapabilityState::Supported);
assert_eq!(
@@ -1293,7 +1295,15 @@ mod tests {
.remote_targets
.fields
.iter()
.any(|field| field.name == "disableProxy" && field.state == super::ReplicationFieldState::Unsupported)
.any(|field| field.name == "disableProxy" && field.state == super::ReplicationFieldState::Supported)
);
assert!(
response
.replication
.remote_targets
.fields
.iter()
.any(|field| field.name == "edge" && field.state == super::ReplicationFieldState::Unsupported)
);
assert!(
response
@@ -1364,7 +1374,7 @@ mod tests {
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"]["remote_targets"]["contract_version"], 2);
assert_eq!(value["replication"]["bucket_replication"]["status"]["state"], "supported");
assert_eq!(value["replication"]["remote_targets"]["status"]["state"], "supported");
assert_eq!(
@@ -1383,7 +1393,14 @@ mod tests {
.as_array()
.expect("remote target fields should be an array")
.iter()
.any(|field| field["name"] == "disableProxy" && field["state"] == "unsupported")
.any(|field| field["name"] == "disableProxy" && field["state"] == "supported")
);
assert!(
value["replication"]["remote_targets"]["fields"]
.as_array()
.expect("remote target fields should be an array")
.iter()
.any(|field| field["name"] == "edge" && field["state"] == "unsupported")
);
assert!(
value["replication"]["remote_targets"]["fields"]