gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-29 17:48:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(replication): normalize local site endpoint port (#3249)

Browse Source 
* fix(replication): normalize local site endpoint port

* fix(replication): respect runtime console endpoint

* fix(replication): allow valid peer endpoint ports

---------

Co-authored-by: loverustfs <hello@rustfs.com>
Co-authored-by: houseme <housemecn@gmail.com>
This commit is contained in:
cxymds
2026-06-07 12:19:04 +08:00
committed by GitHub
parent 316540aaf8
commit 04b5e8f988
2 changed files with 141 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/admin/handlers/site_replication.rs
+140 -22
View File
@@ -20,6 +20,7 @@ use crate::admin::site_replication_identity::{
};
use crate::admin::utils::{encode_compatible_admin_payload, read_compatible_admin_body};
use crate::auth::{check_key_valid, get_session_token};
use crate::config::get_config_snapshot;
use crate::error::ApiError;
use crate::server::{ADMIN_PREFIX, RemoteAddr};
use base64::Engine;
@@ -29,7 +30,10 @@ use http::header::{CONTENT_TYPE, HOST};
use http::{HeaderMap, HeaderValue, Uri};
use hyper::{Method, StatusCode};
use matchit::Params;
use rustfs_config::{DEFAULT_DELIMITER, DEFAULT_RUSTFS_TLS_PATH, ENV_RUSTFS_TLS_PATH, MAX_ADMIN_REQUEST_BODY_SIZE};
use rustfs_config::{
DEFAULT_CONSOLE_ADDRESS, DEFAULT_DELIMITER, DEFAULT_RUSTFS_TLS_PATH, ENV_RUSTFS_CONSOLE_ADDRESS, ENV_RUSTFS_TLS_PATH,
MAX_ADMIN_REQUEST_BODY_SIZE,
};
use rustfs_ecstore::bucket::bucket_target_sys::BucketTargetSys;
use rustfs_ecstore::bucket::metadata::{
BUCKET_CORS_CONFIG, BUCKET_LIFECYCLE_CONFIG, BUCKET_POLICY_CONFIG, BUCKET_QUOTA_CONFIG_FILE, BUCKET_REPLICATION_CONFIG,
@@ -721,8 +725,38 @@ fn request_endpoint(uri: &Uri, headers: &HeaderMap) -> String {
format!("{scheme}://{host}")
}
fn runtime_console_port() -> Option<u16> {
let console_address = get_config_snapshot()
.map(|snapshot| snapshot.console_address.clone())
.unwrap_or_else(|| rustfs_utils::get_env_str(ENV_RUSTFS_CONSOLE_ADDRESS, DEFAULT_CONSOLE_ADDRESS));
let parse_target = if console_address.starts_with(':') {
format!("127.0.0.1{console_address}")
} else {
console_address
};
Url::parse(&format!("http://{parse_target}"))
.ok()
.and_then(|parsed| parsed.port_or_known_default())
}
fn site_replication_local_endpoint(uri: &Uri, headers: &HeaderMap) -> String {
let endpoint = request_endpoint(uri, headers);
match Url::parse(&endpoint) {
Ok(mut parsed) => {
if parsed.port_or_known_default() == runtime_console_port() && parsed.set_port(Some(global_rustfs_port())).is_ok() {
parsed.to_string().trim_end_matches('/').to_string()
} else {
endpoint
}
}
Err(_) => endpoint,
}
}
fn current_local_runtime_endpoint() -> String {
request_endpoint(&Uri::from_static("/"), &HeaderMap::new())
site_replication_local_endpoint(&Uri::from_static("/"), &HeaderMap::new())
}
fn infer_site_name(endpoint: &str) -> String {
@@ -750,7 +784,7 @@ fn non_negative_u64(value: i64) -> u64 {
}
fn current_local_peer(req: &S3Request<Body>, state: &SiteReplicationState) -> PeerInfo {
let endpoint = request_endpoint(&req.uri, &req.headers);
let endpoint = site_replication_local_endpoint(&req.uri, &req.headers);
let deployment_id = get_global_deployment_id().unwrap_or_else(|| deployment_id_for_endpoint(&endpoint));
let stored_peer = state.peers.get(&deployment_id);
@@ -1747,16 +1781,21 @@ fn site_replication_bucket_target_for_peer(
state: &SiteReplicationState,
peer: &PeerInfo,
arn_override: Option<String>,
) -> Option<BucketTarget> {
) -> S3Result<Option<BucketTarget>> {
if state.service_account_access_key.is_empty() || state.service_account_secret_key.is_empty() {
return None;
return Ok(None);
}
let parsed = Url::parse(&peer.endpoint)
.ok()
.or_else(|| Url::parse(&format!("http://{}", peer.endpoint.trim())).ok())?;
let host = parsed.host_str()?;
let port = parsed.port_or_known_default()?;
.or_else(|| Url::parse(&format!("http://{}", peer.endpoint.trim())).ok())
.ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, format!("invalid peer endpoint: {}", peer.endpoint)))?;
let host = parsed.host_str().ok_or_else(|| {
S3Error::with_message(S3ErrorCode::InvalidRequest, format!("peer endpoint missing host: {}", peer.endpoint))
})?;
let port = parsed.port_or_known_default().ok_or_else(|| {
S3Error::with_message(S3ErrorCode::InvalidRequest, format!("peer endpoint missing port: {}", peer.endpoint))
})?;
let arn = arn_override.unwrap_or_else(|| {
ARN::new(
BucketTargetType::ReplicationService,
@@ -1767,7 +1806,7 @@ fn site_replication_bucket_target_for_peer(
.to_string()
});
Some(BucketTarget {
Ok(Some(BucketTarget {
source_bucket: bucket.to_string(),
endpoint: format!("{host}:{port}"),
credentials: Some(Credentials {
@@ -1782,7 +1821,7 @@ fn site_replication_bucket_target_for_peer(
target_type: BucketTargetType::ReplicationService,
deployment_id: peer.deployment_id.clone(),
..Default::default()
})
}))
}
fn reconcile_site_replication_bucket_targets(
@@ -1791,9 +1830,9 @@ fn reconcile_site_replication_bucket_targets(
state: &SiteReplicationState,
local_peer: &PeerInfo,
config: Option<&s3s::dto::ReplicationConfiguration>,
) -> BucketTargets {
) -> S3Result<BucketTargets> {
if !state.enabled() || state.service_account_access_key.is_empty() || state.service_account_secret_key.is_empty() {
return existing;
return Ok(existing);
}
let configured_arns = site_replication_target_arns_by_peer(config);
@@ -1805,7 +1844,7 @@ fn reconcile_site_replication_bucket_targets(
}
let Some(mut target) =
site_replication_bucket_target_for_peer(bucket, state, peer, configured_arns.get(&peer.deployment_id).cloned())
site_replication_bucket_target_for_peer(bucket, state, peer, configured_arns.get(&peer.deployment_id).cloned())?
else {
continue;
};
@@ -1837,7 +1876,7 @@ fn reconcile_site_replication_bucket_targets(
}
}
BucketTargets { targets }
Ok(BucketTargets { targets })
}
fn build_site_replication_rule(arn: &str, priority: i32, rule_id: &str) -> ReplicationRule {
@@ -1868,14 +1907,14 @@ fn build_site_replication_config(
bucket: &str,
state: &SiteReplicationState,
local_peer: &PeerInfo,
) -> Option<ReplicationConfiguration> {
) -> S3Result<Option<ReplicationConfiguration>> {
let mut rules = Vec::new();
for peer in state.peers.values() {
if peer.deployment_id == local_peer.deployment_id || same_identity_endpoint(&peer.endpoint, &local_peer.endpoint) {
continue;
}
let Some(target) = site_replication_bucket_target_for_peer(bucket, state, peer, None) else {
let Some(target) = site_replication_bucket_target_for_peer(bucket, state, peer, None)? else {
continue;
};
rules.push(build_site_replication_rule(
@@ -1886,12 +1925,12 @@ fn build_site_replication_config(
}
if rules.is_empty() {
None
Ok(None)
} else {
Some(ReplicationConfiguration {
Ok(Some(ReplicationConfiguration {
role: String::new(),
rules,
})
}))
}
}
@@ -1907,7 +1946,7 @@ async fn ensure_site_replication_bucket_targets(
Err(err) => return Err(ApiError::from(err).into()),
};
let updated = reconcile_site_replication_bucket_targets(existing, bucket, state, local_peer, config);
let updated = reconcile_site_replication_bucket_targets(existing, bucket, state, local_peer, config)?;
if updated.targets.is_empty() {
return Ok(());
}
@@ -1933,7 +1972,7 @@ async fn ensure_site_replication_bucket_replication_config(
Err(err) => return Err(ApiError::from(err).into()),
}
let Some(config) = build_site_replication_config(bucket, state, local_peer) else {
let Some(config) = build_site_replication_config(bucket, state, local_peer)? else {
return Ok(());
};
@@ -3240,6 +3279,42 @@ mod tests {
});
}
#[test]
fn test_site_replication_local_endpoint_uses_api_port_for_console_host_header() {
let uri: Uri = "/rustfs/admin/v3/site-replication/status".parse().unwrap();
let mut headers = HeaderMap::new();
headers.insert("x-forwarded-proto", HeaderValue::from_static("https"));
headers.insert("host", HeaderValue::from_static("node-a.example.com:9001"));
let endpoint = site_replication_local_endpoint(&uri, &headers);
assert_eq!(endpoint, "https://node-a.example.com:9000");
}
#[test]
fn test_site_replication_local_endpoint_preserves_ipv6_host() {
let uri: Uri = "/rustfs/admin/v3/site-replication/status".parse().unwrap();
let mut headers = HeaderMap::new();
headers.insert("x-forwarded-proto", HeaderValue::from_static("https"));
headers.insert("host", HeaderValue::from_static("[::1]:9001"));
let endpoint = site_replication_local_endpoint(&uri, &headers);
assert_eq!(endpoint, "https://[::1]:9000");
}
#[test]
fn test_site_replication_local_endpoint_preserves_non_console_port() {
let uri: Uri = "/rustfs/admin/v3/site-replication/status".parse().unwrap();
let mut headers = HeaderMap::new();
headers.insert("x-forwarded-proto", HeaderValue::from_static("https"));
headers.insert("host", HeaderValue::from_static("lb.example.com:9443"));
let endpoint = site_replication_local_endpoint(&uri, &headers);
assert_eq!(endpoint, "https://lb.example.com:9443");
}
#[test]
fn test_runtime_tls_enabled_prefers_explicit_tls_over_http_runtime_endpoint() {
let endpoints = EndpointServerPools::from(vec![PoolEndpoints {
@@ -3661,7 +3736,8 @@ mod tests {
..peer("local", "https://local.example.com")
},
None,
);
)
.expect("reconcile bucket targets");
assert_eq!(targets.targets.len(), 1);
let target = &targets.targets[0];
@@ -3679,6 +3755,48 @@ mod tests {
assert_eq!(credentials.secret_key, "secret");
}
#[test]
fn test_reconcile_site_replication_bucket_targets_allows_peer_on_same_port_as_local_console() {
with_var("RUSTFS_CONSOLE_ADDRESS", Some(":9001"), || {
let mut state = SiteReplicationState {
service_account_access_key: "site-replicator-0".to_string(),
service_account_secret_key: "secret".to_string(),
..Default::default()
};
state.peers.insert(
"local".to_string(),
PeerInfo {
deployment_id: "local".to_string(),
..peer("local", "https://local.example.com:9000")
},
);
state.peers.insert(
"remote".to_string(),
PeerInfo {
deployment_id: "remote".to_string(),
..peer("remote", "https://remote.example.com:9001")
},
);
let targets = reconcile_site_replication_bucket_targets(
BucketTargets::default(),
"photos",
&state,
&PeerInfo {
deployment_id: "local".to_string(),
..peer("local", "https://local.example.com:9000")
},
None,
)
.expect("peer using same numeric port as local console should remain valid");
assert_eq!(targets.targets.len(), 1);
let target = &targets.targets[0];
assert_eq!(target.endpoint, "remote.example.com:9001");
assert!(target.secure);
});
}
#[test]
fn test_apply_state_edit_req_only_updates_ilm_expiry_flags() {
let mut state = SiteReplicationState::default();
rustfs/src/config/mod.rs
+1
View File
@@ -55,6 +55,7 @@ pub use cli::{TlsCommands, TlsInspectOpts, TlsOpts};
pub use config_struct::Config;
pub use info::execute_info;
pub use opt::Opt;
pub(crate) use snapshot::get_config_snapshot;
pub use snapshot::{get_config_snapshot_for_display, init_config_snapshot};
// Re-export workload profiles