mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-05 12:57:42 +00:00
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
This commit is contained in:
GatewayJ
@@ -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<String, HashSet<IpAddr>> = HashMap::new();
|
||||
let mut host_ip_cache = HashMap::new();
|
||||
let mut host_ip_cache: HashMap<Host<&str>, HashSet<IpAddr>> = 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<Duration> {
|
||||
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<T, Resolve, ResolveFut, Sleep, SleepFut, PermanentError, TimeoutError, RetryLog>(
|
||||
mut resolve: Resolve,
|
||||
mut sleep: Sleep,
|
||||
dns_retry_deadline: &DnsRetryDeadline,
|
||||
mut permanent_error: PermanentError,
|
||||
mut timeout_error: TimeoutError,
|
||||
mut retry_log: RetryLog,
|
||||
) -> Result<T>
|
||||
where
|
||||
Resolve: FnMut() -> ResolveFut,
|
||||
ResolveFut: Future<Output = Result<T>>,
|
||||
Sleep: FnMut(Duration) -> SleepFut,
|
||||
SleepFut: Future<Output = ()>,
|
||||
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<bool> {
|
||||
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<HashSet<IpAddr>> {
|
||||
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 = [
|
||||
|
||||
Reference in New Issue
Block a user