diff --git a/tonic-xds/src/xds/cluster_discovery.rs b/tonic-xds/src/xds/cluster_discovery.rs index 61251b9a9..a33a119cf 100644 --- a/tonic-xds/src/xds/cluster_discovery.rs +++ b/tonic-xds/src/xds/cluster_discovery.rs @@ -14,6 +14,7 @@ //! connector is kept and a warning is logged. use std::sync::Arc; +use std::time::Duration; use arc_swap::ArcSwap; use tokio::sync::mpsc; @@ -38,6 +39,33 @@ use crate::xds::resource::security::ClusterSecurityConfig; /// Tower's LB layer. const DISCOVER_CHANNEL_CAPACITY: usize = 64; +/// Timeout for establishing a connection to an endpoint. +const ENDPOINT_CONNECT_TIMEOUT: Duration = Duration::from_secs(10); + +/// HTTP/2 keepalive PING interval for endpoint connections. +const ENDPOINT_KEEP_ALIVE_INTERVAL: Duration = Duration::from_secs(30); + +/// Time to wait for a keepalive PING ack before closing the connection. +const ENDPOINT_KEEP_ALIVE_TIMEOUT: Duration = Duration::from_secs(10); + +/// Apply connection-liveness settings to a per-endpoint `Endpoint`. +/// +/// Endpoint channels are created with `connect_lazy`, so a tonic `Channel` +/// reports readiness independently of TCP connectivity. Without these +/// settings a request routed to a black-holed address (e.g. a deleted pod IP, +/// which Kubernetes drops without a RST) hangs on unanswered SYNs, and an +/// established connection to a dead peer is never torn down — either way the +/// endpoint stays "ready" to the LB and requests only die by the caller's +/// deadline. The connect timeout and keepalives turn both cases into prompt +/// transport errors instead. +fn with_liveness_settings(endpoint: Endpoint) -> Endpoint { + endpoint + .connect_timeout(ENDPOINT_CONNECT_TIMEOUT) + .http2_keep_alive_interval(ENDPOINT_KEEP_ALIVE_INTERVAL) + .keep_alive_timeout(ENDPOINT_KEEP_ALIVE_TIMEOUT) + .keep_alive_while_idle(true) +} + /// xDS-backed cluster discovery. /// /// Resolves cluster names into endpoint change streams by watching the @@ -189,9 +217,9 @@ impl Connector for PlaintextConnector { // EndpointAddress only holds validated Ipv4/Ipv6/Hostname + u16 port, // and its Display impl produces "ip:port" or "hostname:port". Prefixing // with "http://" always yields a valid URI, so from_shared cannot fail. - let channel = Endpoint::from_shared(format!("http://{addr}")) - .expect("EndpointAddress Display guarantees valid URI") - .connect_lazy(); + let endpoint = Endpoint::from_shared(format!("http://{addr}")) + .expect("EndpointAddress Display guarantees valid URI"); + let channel = with_liveness_settings(endpoint).connect_lazy(); let svc = EndpointChannel::new(channel); Box::pin(async move { svc }) } @@ -284,8 +312,10 @@ impl Connector for TlsConnector { } let uri = format!("https://{addr}"); - let endpoint = Endpoint::from_shared(uri.clone()) - .expect("EndpointAddress Display guarantees valid URI"); + let endpoint = with_liveness_settings( + Endpoint::from_shared(uri.clone()) + .expect("EndpointAddress Display guarantees valid URI"), + ); let channel = match endpoint.tls_config_with_verifier(tls_config, verifier) { Ok(ep) => ep.connect_lazy(), @@ -298,9 +328,11 @@ impl Connector for TlsConnector { error = %e, address = %addr, "tls_config_with_verifier failed; non-TLS lazy fallback", ); - Endpoint::from_shared(uri) - .expect("EndpointAddress Display guarantees valid URI") - .connect_lazy() + with_liveness_settings( + Endpoint::from_shared(uri) + .expect("EndpointAddress Display guarantees valid URI"), + ) + .connect_lazy() } }; let svc = EndpointChannel::new(channel); diff --git a/xds-client/src/transport/tonic.rs b/xds-client/src/transport/tonic.rs index e37e679ac..71e793e27 100644 --- a/xds-client/src/transport/tonic.rs +++ b/xds-client/src/transport/tonic.rs @@ -10,6 +10,7 @@ use crate::transport::{Transport, TransportBuilder, TransportStream}; use bytes::{Buf, BufMut, Bytes}; use http::uri::PathAndQuery; use std::sync::Arc; +use std::time::Duration; use tokio::sync::mpsc; use tokio_stream::StreamExt as _; use tonic::client::Grpc; @@ -42,6 +43,23 @@ const ADS_PATH: &str = const ADS_CHANNEL_BUFFER_SIZE: usize = 16; +/// Default timeout for establishing the TCP/TLS connection to the xDS server. +const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(10); + +/// Default HTTP/2 keepalive PING interval on the ADS channel. +/// +/// The ADS stream is mostly idle from the client's perspective (the server +/// only pushes on resource changes), so without keepalives a half-open +/// connection — e.g. after an xDS server restart where the RST/GOAWAY was +/// lost, or a dropped conntrack/NAT entry — is undetectable: `recv()` on the +/// stream pends forever and the client keeps serving its last known +/// resources. Keepalives surface such connections as transport errors, which +/// triggers the worker's reconnect + re-subscribe path. +const DEFAULT_KEEP_ALIVE_INTERVAL: Duration = Duration::from_secs(30); + +/// Default time to wait for a keepalive PING ack before closing the connection. +const DEFAULT_KEEP_ALIVE_TIMEOUT: Duration = Duration::from_secs(10); + /// A codec that passes bytes through without serialization. /// /// This allows us to handle serialization in the xDS client layer @@ -169,11 +187,9 @@ impl TonicTransport { /// let builder = TonicTransportBuilder::new() /// .with_tls_config(ClientTlsConfig::new().with_enabled_roots()); /// ``` -#[derive(Debug, Clone, Default)] +#[derive(Debug, Clone)] pub struct TonicTransportBuilder { // Future extensions: - // - Connection timeout settings - // - Keep-alive configuration // - Connection pooling settings // - Per-server credential overrides (via ServerConfig.extensions) #[cfg(any(feature = "tonic-tls-ring", feature = "tonic-tls-aws-lc"))] @@ -181,6 +197,28 @@ pub struct TonicTransportBuilder { /// Per-stream call credentials for the ADS stream. call_creds: Option>, + + /// Timeout for establishing the connection to the xDS server. + connect_timeout: Duration, + + /// HTTP/2 keepalive PING interval; `None` disables keepalives. + keep_alive_interval: Option, + + /// Time to wait for a keepalive PING ack before closing the connection. + keep_alive_timeout: Duration, +} + +impl Default for TonicTransportBuilder { + fn default() -> Self { + Self { + #[cfg(any(feature = "tonic-tls-ring", feature = "tonic-tls-aws-lc"))] + tls_config: None, + call_creds: None, + connect_timeout: DEFAULT_CONNECT_TIMEOUT, + keep_alive_interval: Some(DEFAULT_KEEP_ALIVE_INTERVAL), + keep_alive_timeout: DEFAULT_KEEP_ALIVE_TIMEOUT, + } + } } impl TonicTransportBuilder { @@ -189,6 +227,24 @@ impl TonicTransportBuilder { Self::default() } + /// Set the timeout for establishing the connection to the xDS server. + pub fn with_connect_timeout(mut self, timeout: Duration) -> Self { + self.connect_timeout = timeout; + self + } + + /// Configure HTTP/2 keepalives on the ADS channel. + /// + /// A PING is sent every `interval` (even while the stream is idle); if no + /// ack arrives within `timeout` the connection is closed, surfacing a + /// transport error that triggers reconnect + re-subscription. Pass + /// `interval = None` to disable keepalives. + pub fn with_keep_alive(mut self, interval: Option, timeout: Duration) -> Self { + self.keep_alive_interval = interval; + self.keep_alive_timeout = timeout; + self + } + /// Set the TLS configuration for connections to the xDS server. /// /// When set, all connections created by this builder will use TLS @@ -250,6 +306,14 @@ impl TransportBuilder for TonicTransportBuilder { let endpoint = Endpoint::from_shared(Self::ensure_secure_server_uri(server.uri(), secure)) .map_err(|e| Error::Connection(e.to_string()))?; + let mut endpoint = endpoint.connect_timeout(self.connect_timeout); + if let Some(interval) = self.keep_alive_interval { + endpoint = endpoint + .http2_keep_alive_interval(interval) + .keep_alive_timeout(self.keep_alive_timeout) + .keep_alive_while_idle(true); + } + #[cfg(any(feature = "tonic-tls-ring", feature = "tonic-tls-aws-lc"))] let endpoint = match &self.tls_config { Some(tls) => endpoint