From 3e7e39a59c5d131251248dc5c39794076c45a44b Mon Sep 17 00:00:00 2001 From: GatewayJ <835269233@qq.com> Date: Sat, 20 Jun 2026 23:53:11 +0800 Subject: [PATCH] fix(ecstore): retry transient endpoint DNS failures (#3652) * fix(ecstore): retry transient endpoint DNS failures * fix(ecstore): satisfy DNS retry clippy check * fix(ecstore): harden endpoint DNS retry * test(ecstore): stabilize DNS retry error test --- crates/ecstore/src/layout/endpoints.rs | 418 +++++++++++++++++++++---- 1 file changed, 358 insertions(+), 60 deletions(-) diff --git a/crates/ecstore/src/layout/endpoints.rs b/crates/ecstore/src/layout/endpoints.rs index 3c979f0fd..4a349a980 100644 --- a/crates/ecstore/src/layout/endpoints.rs +++ b/crates/ecstore/src/layout/endpoints.rs @@ -23,10 +23,14 @@ use rustfs_config::{DEFAULT_UNSAFE_BYPASS_DISK_CHECK, ENV_MINIO_CI, ENV_UNSAFE_B use rustfs_utils::{XHost, check_local_server_addr, get_host_ip, is_local_host}; use std::{ collections::{BTreeMap, BTreeSet, HashMap, HashSet, hash_map::Entry}, + future::Future, io::{Error, ErrorKind, Result}, net::IpAddr, + time::{Duration, Instant, SystemTime, UNIX_EPOCH}, }; +use tokio::time::sleep as async_sleep; use tracing::{error, info, instrument, warn}; +use url::Host; /// enum for setup type. #[derive(PartialEq, Eq, Debug, Clone)] @@ -264,18 +268,20 @@ impl PoolEndpointList { setup_type: SetupType::Unknown, }; - pool_endpoint_list.update_is_local(server_addr.port())?; + let dns_retry_deadline = DnsRetryDeadline::new(DNS_RETRY_TOTAL_TIMEOUT); + pool_endpoint_list + .update_is_local(server_addr.port(), &dns_retry_deadline) + .await?; for endpoints in pool_endpoint_list.inner.iter_mut() { // Check whether same path is not used in endpoints of a host on different port. let mut path_ip_map: HashMap> = HashMap::new(); - let mut host_ip_cache = HashMap::new(); + let mut host_ip_cache: HashMap, HashSet> = HashMap::new(); for ep in endpoints.as_ref() { - if !ep.url.has_host() { + let Some(host) = ep.url.host() else { continue; - } + }; - let host = ep.url.host().unwrap(); let host_ip_set = if let Some(set) = host_ip_cache.get(&host) { info!( target: "rustfs::ecstore::endpoints", @@ -284,13 +290,13 @@ impl PoolEndpointList { from = "cache", "Create pool endpoints host '{}' found in cache for endpoint '{}'", host, ep.to_string() ); - set + set.clone() } else { - let ips = match get_host_ip(host.clone()).await { + let ips = match resolve_host_ips_with_retry(host.clone(), &ep.to_string(), &dns_retry_deadline).await { Ok(ips) => ips, Err(e) => { error!("Create pool endpoints host {} not found, error:{}", host, e); - return Err(Error::other(format!("host '{host}' cannot resolve: {e}"))); + return Err(e); } }; info!( @@ -303,14 +309,14 @@ impl PoolEndpointList { ips, ep.to_string() ); - host_ip_cache.insert(host.clone(), ips); - host_ip_cache.get(&host).unwrap() + host_ip_cache.insert(host.clone(), ips.clone()); + ips }; let path = ep.get_file_path(); match path_ip_map.entry(path) { Entry::Occupied(mut e) => { - if e.get().intersection(host_ip_set).count() > 0 { + if e.get().intersection(&host_ip_set).count() > 0 { let path_key = e.key().clone(); return Err(Error::other(format!( "same path '{path_key}' can not be served by different port on same address" @@ -401,7 +407,12 @@ impl PoolEndpointList { } /// resolves all hosts and discovers which are local - fn update_is_local(&mut self, local_port: u16) -> Result<()> { + async fn update_is_local(&mut self, local_port: u16, dns_retry_deadline: &DnsRetryDeadline) -> Result<()> { + self._update_is_local(local_port, dns_retry_deadline).await + } + + /// resolves all hosts and discovers which are local + async fn _update_is_local(&mut self, local_port: u16, dns_retry_deadline: &DnsRetryDeadline) -> Result<()> { for endpoints in self.inner.iter_mut() { for ep in endpoints.as_mut() { match ep.url.host() { @@ -409,7 +420,14 @@ impl PoolEndpointList { ep.is_local = true; } Some(host) => { - ep.is_local = is_local_host(host, ep.url.port().unwrap_or_default(), local_port)?; + ep.is_local = resolve_local_host_with_retry( + host, + ep.url.port().unwrap_or_default(), + local_port, + &ep.to_string(), + dns_retry_deadline, + ) + .await?; } } } @@ -417,57 +435,198 @@ impl PoolEndpointList { Ok(()) } +} - /// resolves all hosts and discovers which are local - fn _update_is_local(&mut self, local_port: u16) -> Result<()> { - let mut eps_resolved = 0; - let mut found_local = false; - let mut resolved_set: HashSet<(usize, usize)> = HashSet::new(); - let ep_count: usize = self.inner.iter().map(|v| v.as_ref().len()).sum(); +const DNS_RETRY_TOTAL_TIMEOUT: Duration = Duration::from_secs(90); +const DNS_RETRY_BASE_DELAY: Duration = Duration::from_millis(500); +const DNS_RETRY_MAX_DELAY: Duration = Duration::from_secs(8); +const DNS_RETRY_JITTER_PERCENT: u64 = 20; - loop { - // Break if the local endpoint is found already Or all the endpoints are resolved. - if found_local || eps_resolved == ep_count { - break; - } +struct DnsRetryDeadline { + started: Instant, + timeout: Duration, +} - for (i, endpoints) in self.inner.iter_mut().enumerate() { - for (j, ep) in endpoints.as_mut().iter_mut().enumerate() { - if resolved_set.contains(&(i, j)) { - // Continue if host is already resolved. - continue; - } +impl DnsRetryDeadline { + fn new(timeout: Duration) -> Self { + Self { + started: Instant::now(), + timeout, + } + } - match ep.url.host() { - None => { - if !found_local { - found_local = true; - } - ep.is_local = true; - eps_resolved += 1; - resolved_set.insert((i, j)); - continue; - } - Some(host) => match is_local_host(host, ep.url.port().unwrap_or_default(), local_port) { - Ok(is_local) => { - if !found_local { - found_local = is_local; - } - ep.is_local = is_local; - eps_resolved += 1; - resolved_set.insert((i, j)); - } - Err(err) => { - // TODO Retry infinitely on Kubernetes and Docker swarm? - return Err(err); - } - }, - } - } - } + fn timeout(&self) -> Duration { + self.timeout + } + + fn bounded_delay(&self, delay: Duration) -> Option { + let remaining = self.timeout.saturating_sub(self.started.elapsed()); + if remaining.is_zero() { + return None; } - Ok(()) + Some(delay.min(remaining)) + } +} + +async fn retry_dns_operation( + mut resolve: Resolve, + mut sleep: Sleep, + dns_retry_deadline: &DnsRetryDeadline, + mut permanent_error: PermanentError, + mut timeout_error: TimeoutError, + mut retry_log: RetryLog, +) -> Result +where + Resolve: FnMut() -> ResolveFut, + ResolveFut: Future>, + Sleep: FnMut(Duration) -> SleepFut, + SleepFut: Future, + PermanentError: FnMut(Error) -> Error, + TimeoutError: FnMut(u32, Duration, Error) -> Error, + RetryLog: FnMut(u32, Duration, &Error), +{ + let mut attempts: u32 = 0; + + loop { + match resolve().await { + Ok(value) => return Ok(value), + Err(err) => { + if !is_retryable_dns_error(&err) { + return Err(permanent_error(err)); + } + + attempts += 1; + let Some(delay) = dns_retry_deadline.bounded_delay(dns_retry_delay(attempts)) else { + return Err(timeout_error(attempts, dns_retry_deadline.timeout(), err)); + }; + + retry_log(attempts, delay, &err); + sleep(delay).await; + } + } + } +} + +async fn resolve_local_host_with_retry( + host: Host<&str>, + port: u16, + local_port: u16, + context: &str, + dns_retry_deadline: &DnsRetryDeadline, +) -> Result { + retry_dns_operation( + || { + let host = host.clone(); + async move { is_local_host(host, port, local_port) } + }, + async_sleep, + dns_retry_deadline, + |err| Error::other(format!("endpoint '{context}' local-host detection failed for host '{host}': {err}")), + |attempts, timeout, err| { + Error::other(format!( + "endpoint '{context}' local-host detection timed out after {attempts} attempts and {timeout:?}: {err}" + )) + }, + |attempts, delay, err| { + warn!( + target = "rustfs::ecstore::endpoints", + context = %context, + host = %host, + endpoint = %context, + attempt = attempts, + delay_ms = delay.as_millis(), + error = %err, + "retrying endpoint local-host detection after temporary DNS error" + ); + }, + ) + .await +} + +async fn resolve_host_ips_with_retry( + host: Host<&str>, + context: &str, + dns_retry_deadline: &DnsRetryDeadline, +) -> Result> { + retry_dns_operation( + || { + let host = host.clone(); + async move { get_host_ip(host).await } + }, + async_sleep, + dns_retry_deadline, + |err| Error::other(format!("endpoint '{context}' host '{host}' cannot resolve: {err}")), + |attempts, timeout, err| { + Error::other(format!( + "endpoint '{context}' host '{host}' DNS resolution timed out after {attempts} attempts and {timeout:?}: {err}" + )) + }, + |attempts, delay, err| { + warn!( + target = "rustfs::ecstore::endpoints", + context = %context, + host = %host, + attempt = attempts, + delay_ms = delay.as_millis(), + error = %err, + "retrying endpoint DNS resolution after temporary error" + ); + }, + ) + .await +} + +fn is_retryable_dns_error(err: &Error) -> bool { + if matches!(err.kind(), ErrorKind::Interrupted | ErrorKind::WouldBlock | ErrorKind::TimedOut) { + return true; + } + + if matches!(err.raw_os_error(), Some(-3) | Some(-2)) { + return true; + } + + let message = err.to_string().to_ascii_lowercase(); + // Kubernetes and Docker DNS records can be observed as negative lookups + // while headless service records are still propagating during startup. + message.contains("temporary failure in name resolution") + || message.contains("try again") + || message.contains("name or service not known") + || message.contains("no such host") + || message.contains("nodename nor servname provided") +} + +fn dns_retry_delay(attempt: u32) -> Duration { + let capped_attempt = attempt.saturating_sub(1).min(10); + let raw_delay = DNS_RETRY_BASE_DELAY.saturating_mul(1_u32 << capped_attempt); + let bounded_delay = if raw_delay > DNS_RETRY_MAX_DELAY { + DNS_RETRY_MAX_DELAY + } else { + raw_delay + }; + apply_jitter(bounded_delay) +} + +fn apply_jitter(delay: Duration) -> Duration { + let delay_ms = u64::try_from(delay.as_millis()).unwrap_or(u64::MAX); + if delay_ms == 0 { + return delay; + } + + let jitter_window_ms = delay_ms.saturating_mul(DNS_RETRY_JITTER_PERCENT) / 100; + if jitter_window_ms == 0 { + return delay; + } + + let jitter_ms = SystemTime::now() + .duration_since(UNIX_EPOCH) + .ok() + .map_or(0, |ts| u64::from(ts.subsec_nanos()) % (2 * jitter_window_ms + 1)); + + if jitter_ms >= jitter_window_ms { + delay + Duration::from_millis(jitter_ms - jitter_window_ms) + } else { + delay.saturating_sub(Duration::from_millis(jitter_window_ms - jitter_ms)) } } @@ -593,6 +752,10 @@ impl EndpointServerPools { for pool in self.0.iter() { for ep in pool.endpoints.as_ref() { + let Ok(pool_idx) = usize::try_from(ep.pool_idx) else { + continue; + }; + let n = node_map.entry(ep.host_port()).or_insert_with(|| Node { url: ep.url.clone(), pools: vec![], @@ -600,8 +763,8 @@ impl EndpointServerPools { grid_host: ep.grid_host(), }); - if !n.pools.contains(&(ep.pool_idx as usize)) { - n.pools.push(ep.pool_idx as usize); + if !n.pools.contains(&pool_idx) { + n.pools.push(pool_idx); } } } @@ -866,6 +1029,141 @@ mod test { #[cfg(target_os = "linux")] use tempfile::tempdir; + #[test] + fn retryable_dns_error_accepts_startup_dns_transients() { + assert!(is_retryable_dns_error(&Error::new(ErrorKind::TimedOut, "resolver timeout"))); + assert!(is_retryable_dns_error(&Error::other( + "failed to lookup address information: Name or service not known" + ))); + assert!(is_retryable_dns_error(&Error::other("no such host"))); + assert!(is_retryable_dns_error(&Error::other("nodename nor servname provided, or not known"))); + } + + #[test] + fn retryable_dns_error_rejects_configuration_errors() { + assert!(!is_retryable_dns_error(&Error::new( + ErrorKind::InvalidInput, + "invalid URL endpoint format" + ))); + assert!(!is_retryable_dns_error(&Error::other("mixed scheme is not supported"))); + } + + #[test] + fn retryable_dns_error_rejects_non_dns_transport_errors() { + assert!(!is_retryable_dns_error(&Error::new(ErrorKind::ConnectionRefused, "connection refused"))); + assert!(!is_retryable_dns_error(&Error::new(ErrorKind::ConnectionReset, "connection reset"))); + assert!(!is_retryable_dns_error(&Error::new(ErrorKind::UnexpectedEof, "unexpected eof"))); + assert!(!is_retryable_dns_error(&Error::new(ErrorKind::PermissionDenied, "permission denied"))); + } + + #[test] + fn dns_retry_delay_starts_from_base_and_caps_at_max() { + let first = dns_retry_delay(1); + assert!(first >= DNS_RETRY_BASE_DELAY.saturating_sub(Duration::from_millis(100))); + assert!(first <= DNS_RETRY_BASE_DELAY + Duration::from_millis(100)); + + let capped = dns_retry_delay(20); + assert!(capped >= DNS_RETRY_MAX_DELAY.saturating_sub(Duration::from_millis(1600))); + assert!(capped <= DNS_RETRY_MAX_DELAY + Duration::from_millis(1600)); + } + + #[tokio::test] + async fn retry_dns_operation_retries_with_backoff_without_real_sleep() { + let deadline = DnsRetryDeadline::new(Duration::from_secs(1)); + let mut calls = 0_u32; + let mut sleeps = Vec::new(); + + let result = retry_dns_operation( + || { + calls += 1; + let call = calls; + async move { + if call < 3 { + Err(Error::new(ErrorKind::TimedOut, "resolver timeout")) + } else { + Ok(call) + } + } + }, + |delay| { + sleeps.push(delay); + async {} + }, + &deadline, + |err| Error::other(format!("permanent: {err}")), + |attempts, timeout, err| Error::other(format!("timed out after {attempts} attempts and {timeout:?}: {err}")), + |_, _, _| {}, + ) + .await + .unwrap(); + + assert_eq!(result, 3); + assert_eq!(calls, 3); + assert_eq!(sleeps.len(), 2); + assert!(sleeps.iter().all(|delay| *delay <= Duration::from_secs(1))); + } + + #[tokio::test] + async fn retry_dns_operation_does_not_retry_configuration_errors() { + let deadline = DnsRetryDeadline::new(Duration::from_secs(1)); + let mut calls = 0_u32; + let mut sleeps = 0_u32; + + let err = retry_dns_operation( + || { + calls += 1; + async { Err::<(), Error>(Error::new(ErrorKind::InvalidInput, "invalid URL endpoint format")) } + }, + |_delay| { + sleeps += 1; + async {} + }, + &deadline, + |err| Error::other(format!("permanent: {err}")), + |attempts, timeout, err| Error::other(format!("timed out after {attempts} attempts and {timeout:?}: {err}")), + |_, _, _| {}, + ) + .await + .unwrap_err(); + + assert_eq!(calls, 1); + assert_eq!(sleeps, 0); + assert!(err.to_string().contains("permanent")); + } + + #[tokio::test] + async fn retry_dns_operation_times_out_with_context() { + let deadline = DnsRetryDeadline::new(Duration::ZERO); + let mut calls = 0_u32; + let mut sleeps = 0_u32; + + let err = retry_dns_operation( + || { + calls += 1; + async { Err::<(), Error>(Error::new(ErrorKind::TimedOut, "resolver timeout")) } + }, + |_delay| { + sleeps += 1; + async {} + }, + &deadline, + |err| Error::other(format!("permanent: {err}")), + |attempts, timeout, err| { + Error::other(format!( + "endpoint 'endpoint-a' DNS resolution timed out after {attempts} attempts and {timeout:?}: {err}" + )) + }, + |_, _, _| {}, + ) + .await + .unwrap_err(); + + assert_eq!(calls, 1); + assert_eq!(sleeps, 0); + assert!(err.to_string().contains("endpoint-a")); + assert!(err.to_string().contains("timed out after 1 attempts")); + } + #[test] fn test_new_endpoints() { let test_cases = [