From a9c8cb8bf4f4c6537e0876a8bad23931d4781b5e Mon Sep 17 00:00:00 2001 From: aryehlev Date: Wed, 22 Jul 2026 13:14:47 +0300 Subject: [PATCH 1/2] add timeout for connections. --- tonic-xds/src/xds/cluster_discovery.rs | 48 +++++++-- xds-client/src/transport/tonic.rs | 131 ++++++++++++++++++++----- 2 files changed, 149 insertions(+), 30 deletions(-) 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..194b6aab4 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 @@ -208,17 +264,30 @@ impl TonicTransportBuilder { self } - /// Prepend `https://` to a scheme-less `server_uri` on the secure path. + /// Reconcile the `server_uri` scheme with the channel's security mode. /// - /// Bootstrap URIs like `trafficdirector.googleapis.com:443` parse with no scheme, - /// so `Endpoint` won't negotiate TLS. A scheme lets it, and tonic derive SNI from - /// `uri.host()`. Non-`http::Uri` inputs (`unix://`) and plaintext are left as-is. - fn ensure_secure_server_uri(raw: &str, secure: bool) -> String { - if secure - && let Ok(uri) = raw.parse::() - && uri.scheme().is_none() - { - return format!("https://{raw}"); + /// Secure path: bootstrap URIs like `trafficdirector.googleapis.com:443` + /// parse with no scheme, so `Endpoint` won't negotiate TLS. A `https://` + /// scheme lets it, and tonic derive SNI from `uri.host()`. + /// + /// Insecure path: bootstraps in the wild carry `https://` URIs alongside + /// `insecure` channel creds (e.g. `https://istiod.istio-system.svc:15010`, + /// istiod's plaintext port). With no TLS configured, passing the `https` + /// scheme through hands tonic a contradictory endpoint, so it is rewritten + /// to `http://` to match the creds — per xDS bootstrap semantics, + /// `channel_creds` decides transport security, not the URI scheme. + /// + /// Non-`http::Uri` inputs (`unix://`) are left as-is. + fn reconcile_server_uri(raw: &str, secure: bool) -> String { + if let Ok(uri) = raw.parse::() { + if secure && uri.scheme().is_none() { + return format!("https://{raw}"); + } + if !secure && uri.scheme_str() == Some("https") { + // Slice by scheme length: `http::Uri` normalizes the scheme's + // case, so don't assume the raw string spells it lowercase. + return format!("http://{}", &raw["https".len() + "://".len()..]); + } } raw.to_string() } @@ -247,9 +316,17 @@ impl TransportBuilder for TonicTransportBuilder { // `Endpoint::from_shared` routes `unix://` URIs to tonic's UDS connector. // Required for control planes like Istio's grpc-agent that ship `unix:///etc/istio/proxy/XDS`. - let endpoint = Endpoint::from_shared(Self::ensure_secure_server_uri(server.uri(), secure)) + let endpoint = Endpoint::from_shared(Self::reconcile_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 @@ -513,26 +590,36 @@ mod tests { } #[test] - fn ensure_secure_server_uri_adds_scheme_only_when_needed() { + fn reconcile_server_uri_matches_scheme_to_security_mode() { assert_eq!( - TonicTransportBuilder::ensure_secure_server_uri( - "trafficdirector.googleapis.com:443", - true - ), + TonicTransportBuilder::reconcile_server_uri("trafficdirector.googleapis.com:443", true), "https://trafficdirector.googleapis.com:443", ); assert_eq!( - TonicTransportBuilder::ensure_secure_server_uri("https://xds.example.com:443", true), + TonicTransportBuilder::reconcile_server_uri("https://xds.example.com:443", true), "https://xds.example.com:443" ); assert_eq!( - TonicTransportBuilder::ensure_secure_server_uri("unix:///etc/istio/proxy/XDS", true), + TonicTransportBuilder::reconcile_server_uri("unix:///etc/istio/proxy/XDS", true), "unix:///etc/istio/proxy/XDS" ); assert_eq!( - TonicTransportBuilder::ensure_secure_server_uri("127.0.0.1:18000", false), + TonicTransportBuilder::reconcile_server_uri("127.0.0.1:18000", false), "127.0.0.1:18000" ); + // `https://` with insecure creds is rewritten to plaintext: the + // channel_creds decide security, not the URI scheme. + assert_eq!( + TonicTransportBuilder::reconcile_server_uri( + "https://istiod.istio-system.svc:15010", + false + ), + "http://istiod.istio-system.svc:15010" + ); + assert_eq!( + TonicTransportBuilder::reconcile_server_uri("http://127.0.0.1:18000", false), + "http://127.0.0.1:18000" + ); } #[tokio::test] From 83c531780235d10d0818d2c31a75b036ea5a1df3 Mon Sep 17 00:00:00 2001 From: aryehlev Date: Wed, 22 Jul 2026 19:02:56 +0000 Subject: [PATCH 2/2] keep server_uri scheme handling unchanged; keepalives only --- xds-client/src/transport/tonic.rs | 61 ++++++++++--------------------- 1 file changed, 19 insertions(+), 42 deletions(-) diff --git a/xds-client/src/transport/tonic.rs b/xds-client/src/transport/tonic.rs index 194b6aab4..71e793e27 100644 --- a/xds-client/src/transport/tonic.rs +++ b/xds-client/src/transport/tonic.rs @@ -264,30 +264,17 @@ impl TonicTransportBuilder { self } - /// Reconcile the `server_uri` scheme with the channel's security mode. + /// Prepend `https://` to a scheme-less `server_uri` on the secure path. /// - /// Secure path: bootstrap URIs like `trafficdirector.googleapis.com:443` - /// parse with no scheme, so `Endpoint` won't negotiate TLS. A `https://` - /// scheme lets it, and tonic derive SNI from `uri.host()`. - /// - /// Insecure path: bootstraps in the wild carry `https://` URIs alongside - /// `insecure` channel creds (e.g. `https://istiod.istio-system.svc:15010`, - /// istiod's plaintext port). With no TLS configured, passing the `https` - /// scheme through hands tonic a contradictory endpoint, so it is rewritten - /// to `http://` to match the creds — per xDS bootstrap semantics, - /// `channel_creds` decides transport security, not the URI scheme. - /// - /// Non-`http::Uri` inputs (`unix://`) are left as-is. - fn reconcile_server_uri(raw: &str, secure: bool) -> String { - if let Ok(uri) = raw.parse::() { - if secure && uri.scheme().is_none() { - return format!("https://{raw}"); - } - if !secure && uri.scheme_str() == Some("https") { - // Slice by scheme length: `http::Uri` normalizes the scheme's - // case, so don't assume the raw string spells it lowercase. - return format!("http://{}", &raw["https".len() + "://".len()..]); - } + /// Bootstrap URIs like `trafficdirector.googleapis.com:443` parse with no scheme, + /// so `Endpoint` won't negotiate TLS. A scheme lets it, and tonic derive SNI from + /// `uri.host()`. Non-`http::Uri` inputs (`unix://`) and plaintext are left as-is. + fn ensure_secure_server_uri(raw: &str, secure: bool) -> String { + if secure + && let Ok(uri) = raw.parse::() + && uri.scheme().is_none() + { + return format!("https://{raw}"); } raw.to_string() } @@ -316,7 +303,7 @@ impl TransportBuilder for TonicTransportBuilder { // `Endpoint::from_shared` routes `unix://` URIs to tonic's UDS connector. // Required for control planes like Istio's grpc-agent that ship `unix:///etc/istio/proxy/XDS`. - let endpoint = Endpoint::from_shared(Self::reconcile_server_uri(server.uri(), secure)) + 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); @@ -590,36 +577,26 @@ mod tests { } #[test] - fn reconcile_server_uri_matches_scheme_to_security_mode() { + fn ensure_secure_server_uri_adds_scheme_only_when_needed() { assert_eq!( - TonicTransportBuilder::reconcile_server_uri("trafficdirector.googleapis.com:443", true), + TonicTransportBuilder::ensure_secure_server_uri( + "trafficdirector.googleapis.com:443", + true + ), "https://trafficdirector.googleapis.com:443", ); assert_eq!( - TonicTransportBuilder::reconcile_server_uri("https://xds.example.com:443", true), + TonicTransportBuilder::ensure_secure_server_uri("https://xds.example.com:443", true), "https://xds.example.com:443" ); assert_eq!( - TonicTransportBuilder::reconcile_server_uri("unix:///etc/istio/proxy/XDS", true), + TonicTransportBuilder::ensure_secure_server_uri("unix:///etc/istio/proxy/XDS", true), "unix:///etc/istio/proxy/XDS" ); assert_eq!( - TonicTransportBuilder::reconcile_server_uri("127.0.0.1:18000", false), + TonicTransportBuilder::ensure_secure_server_uri("127.0.0.1:18000", false), "127.0.0.1:18000" ); - // `https://` with insecure creds is rewritten to plaintext: the - // channel_creds decide security, not the URI scheme. - assert_eq!( - TonicTransportBuilder::reconcile_server_uri( - "https://istiod.istio-system.svc:15010", - false - ), - "http://istiod.istio-system.svc:15010" - ); - assert_eq!( - TonicTransportBuilder::reconcile_server_uri("http://127.0.0.1:18000", false), - "http://127.0.0.1:18000" - ); } #[tokio::test]