Split client into HTTP/2 and HTTP/1.1 transports

The previous commit on this branch enabled HTTP/2 on the rustls
connector by switching ALPN from `enable_http1()` alone to
`enable_http1().enable_http2()`.  REST traffic, watch streams and
log streaming benefit from HTTP/2 multiplexing, but exec, attach
and port-forward use HTTP/1.1 connection upgrades, which are
unrepresentable on an HTTP/2 connection.  After ALPN negotiates
`h2`, the apiserver answered upgrade requests with 4xx/5xx and the
integration suite started failing with
`UpgradeConnection(ProtocolSwitch(...))`.

Fix this by giving `Client` two transports with separate connection
pools and ALPN policies:

* a primary, h2-capable transport for normal traffic, configured
  with `TokioTimer` and HTTP/2 keep-alive PINGs (interval 30s,
  while-idle on) so watch streams survive idle-killing
  intermediaries such as HAProxy;
* an HTTP/1.1-only transport used by `Client::connect()`, the only
  caller of `hyper::upgrade::on` in the workspace, hard-routed
  there via a new `upgrade_inner` field on `Client`.

A subtle point governs the h1-only path on rustls.  hyper-rustls'
builder asserts that `alpn_protocols.is_empty()` when accepting a
rustls config, and only `enable_http2()` populates the list.  The
`enable_http1()`-only builder path therefore leaves ALPN empty,
which means no ALPN extension is sent on the wire and a modern
apiserver may still negotiate HTTP/2 -- the very condition we are
trying to avoid.  The h1-only sibling must advertise `http/1.1`
*explicitly*.

The natural workaround -- `HttpsConnector::from((http,
Arc::new(rustls_config)))` with `alpn_protocols =
vec![b"http/1.1".to_vec()]` -- bypasses the builder's assertion but
also drops `Config::tls_server_name`, because the `From` impl
constructs the connector with the default server-name resolver and
the resolver field is private.  Rather than depend on an upstream
hyper-rustls change, ship a small `H1OnlyHttpsConnector<H>` in
`kube-client/src/client/tls.rs` that mirrors the public TCP-then-TLS
dance from `hyper_rustls::HttpsConnector::call`, sets the explicit
ALPN advertisement, and resolves SNI via `Config::tls_server_name`
when set or from the URI host otherwise.  This adds `tokio-rustls`
as a direct dep gated on `rustls-tls` (already in the dep tree
transitively via hyper-rustls).

Other invariants worth preserving:

* `Config::auth_layer()` and `Config::extra_headers_layer()` are
  computed once and *cloned* into both transport stacks.  Calling
  `auth_layer()` twice would mint independent `RefreshableToken`
  state, so each transport would refresh tokens on its own and
  could diverge under exec-plugin or token-file auth.  `AuthLayer`
  gains `#[derive(Clone)]` for this; the inner `Either` was
  already trivially cloneable.

* `Config` gains a public `disable_http2: bool` field as a runtime
  escape hatch.  When set, the primary transport falls back to the
  h1-only client; the two-client shape stays the same so the upgrade
  path is unaffected.  HTTP/2 stays on by default; the field
  matches the existing `disable_compression` style.

* `ClientBuilder` grows `with_upgrade_service`, paired with a new
  `Client::new_with_upgrade` constructor for custom-service users
  who supply their own service stack and need to opt in to the
  split.  The single-service `Client::new` keeps its signature; it
  internally clones the buffered handle into `upgrade_inner` so
  back-compat is preserved.

Verified against k3d v1.34.1 with the integration suite, the
`pod_exec`, `pod_attach`, `pod_portforward*`, `pod_shell*` and
`log_stream` examples, and a full 1200-combo `cargo hack` feature
powerset.

Signed-off-by: Andrew McDermott <aim@frobware.com>

Author: frobware

kube-client/Cargo.toml

@@ -13,7 +13,7 @@ categories = ["web-programming::http-client", "network-programming", "api-bindin
 
 [features]
 default = ["client", "ring"]
-rustls-tls = ["rustls", "hyper-rustls"]
+rustls-tls = ["rustls", "hyper-rustls", "tokio-rustls"]
 webpki-roots = ["hyper-rustls/webpki-roots"]
 aws-lc-rs = ["hyper-rustls?/aws-lc-rs"]
 ring = ["hyper-rustls?/ring"]
@@ -57,6 +57,7 @@ jiff = { workspace = true, optional = true, features = ["std", "serde"] }
 pem = { workspace = true, optional = true }
 openssl = { workspace = true, optional = true }
 rustls = { workspace = true, optional = true }
+tokio-rustls = { version = "0.26", default-features = false, optional = true }
 bytes = { workspace = true, optional = true }
 tokio = { workspace = true, features = ["time", "signal", "sync", "rt"], optional = true }
 kube-core = { path = "../kube-core", version = "=3.1.0" }

kube-client/src/client/builder.rs

@@ -8,7 +8,7 @@ use hyper_timeout::TimeoutConnector;
 
 use hyper_util::{
     client::legacy::connect::{Connection, HttpConnector},
-    rt::TokioExecutor,
+    rt::{TokioExecutor, TokioTimer},
 };
 
 use jiff::Timestamp;
@@ -28,6 +28,7 @@ pub type DynBody = dyn http_body::Body<Data = Bytes, Error = BoxError> + Send +
 /// Builder for [`Client`] instances with customized [tower](`Service`) middleware.
 pub struct ClientBuilder<Svc> {
     service: Svc,
+    upgrade_service: Option<GenericService>,
     default_ns: String,
     valid_until: Option<Timestamp>,
 }
@@ -43,20 +44,29 @@ impl<Svc> ClientBuilder<Svc> {
     {
         Self {
             service,
+            upgrade_service: None,
             default_ns: default_namespace.into(),
             valid_until: None,
         }
     }
 
     /// Add a [`Layer`] to the current [`Service`] stack.
+    ///
+    /// The layer is applied to the primary [`Service`] only. If an upgrade
+    /// service has been set via [`with_upgrade_service`](Self::with_upgrade_service)
+    /// it is left untouched; users wanting to layer both must apply the
+    /// layer to each service themselves before calling
+    /// [`with_upgrade_service`].
     pub fn with_layer<L: Layer<Svc>>(self, layer: &L) -> ClientBuilder<L::Service> {
         let Self {
             service: stack,
+            upgrade_service,
             default_ns,
             valid_until,
         } = self;
         ClientBuilder {
             service: layer.layer(stack),
+            upgrade_service,
             default_ns,
             valid_until,
         }
@@ -66,11 +76,33 @@ impl<Svc> ClientBuilder<Svc> {
     pub fn with_valid_until(self, valid_until: Option<Timestamp>) -> Self {
         ClientBuilder {
             service: self.service,
+            upgrade_service: self.upgrade_service,
             default_ns: self.default_ns,
             valid_until,
         }
     }
 
+    /// Provide a separate [`Service`] used by the upgrade transport that
+    /// backs exec, attach, and port-forward.
+    ///
+    /// The supplied service is the same shape as [`GenericService`], the
+    /// boxed service produced by the default builder stack. Custom-service
+    /// users that do not naturally arrive at this shape can instead call
+    /// [`Client::new_with_upgrade`] directly.
+    ///
+    /// This is required only if the primary service may negotiate HTTP/2
+    /// *and* the application also uses upgrade subresources. HTTP/1.1
+    /// upgrades are unrepresentable on an HTTP/2 connection, so the
+    /// upgrade transport must offer only HTTP/1.1.
+    pub fn with_upgrade_service(self, upgrade_service: GenericService) -> Self {
+        ClientBuilder {
+            service: self.service,
+            upgrade_service: Some(upgrade_service),
+            default_ns: self.default_ns,
+            valid_until: self.valid_until,
+        }
+    }
+
     /// Build a [`Client`] instance with the current [`Service`] stack.
     pub fn build<B>(self) -> Client
     where
@@ -80,7 +112,17 @@ impl<Svc> ClientBuilder<Svc> {
         B: http_body::Body<Data = bytes::Bytes> + Send + 'static,
         B::Error: Into<BoxError>,
     {
-        Client::new(self.service, self.default_ns).with_valid_until(self.valid_until)
+        let Self {
+            service,
+            upgrade_service,
+            default_ns,
+            valid_until,
+        } = self;
+        match upgrade_service {
+            Some(upgrade) => Client::new_with_upgrade(service, upgrade, default_ns),
+            None => Client::new(service, default_ns),
+        }
+        .with_valid_until(valid_until)
     }
 }
 
@@ -132,10 +174,13 @@ impl TryFrom<Config> for ClientBuilder<GenericService> {
                             use base64::Engine;
                             use http::HeaderValue;
 
-                            let value = format!("Basic {}", base64::engine::general_purpose::STANDARD.encode(userinfo));
+                            let value = format!(
+                                "Basic {}",
+                                base64::engine::general_purpose::STANDARD.encode(userinfo)
+                            );
                             let header = HeaderValue::from_str(&value).unwrap();
                             connector = connector.with_auth(header);
-                        } 
+                        }
                     }
 
                     make_generic_builder(connector, config)
@@ -167,34 +212,152 @@ where
     H::Error: 'static + Send + Sync + std::error::Error,
 {
     let default_ns = config.default_namespace.clone();
+
+    // Build two hyper clients with separate connection pools and ALPN policies:
+    // - the primary, h2-capable transport for normal REST/watch/log traffic
+    //   (subject to `Config::disable_http2`)
+    // - an HTTP/1.1-only transport for the upgrade path used by exec, attach,
+    //   and port-forward, regardless of `disable_http2`.
+    //
+    // Current TLS feature precedence when more than one is set:
+    //   1. rustls-tls
+    //   2. openssl-tls
+    // If neither TLS feature is enabled, the http connector is used; only the
+    // http scheme is supported in that case.
+    // Compute auth and extra-headers layers once and share across both
+    // transports. Calling `auth_layer()` twice would mint independent
+    // `RefreshableToken` state per transport, so each path would refresh
+    // tokens on its own and they'd diverge under exec-plugin or token-file
+    // auth.
     let auth_layer = config.auth_layer()?;
+    let extra_headers_layer = config.extra_headers_layer()?;
+
+    // The two transports use connectors with different concrete types after
+    // TLS wrapping (h2-capable vs explicit-h1 ALPN), so each path is built
+    // and wrapped independently and erased to a `GenericService` here rather
+    // than threading the connector type through.
+    let main_service = build_main_service(
+        connector.clone(),
+        &config,
+        auth_layer.clone(),
+        extra_headers_layer.clone(),
+    )?;
+    let upgrade_service = build_upgrade_service(connector, &config, auth_layer, extra_headers_layer)?;
+
+    let (_, expiration) = config.exec_identity_pem();
+
+    let client = ClientBuilder::new(main_service, default_ns)
+        .with_upgrade_service(upgrade_service)
+        .with_valid_until(expiration);
+
+    Ok(client)
+}
+
+/// Build the primary, h2-capable transport service.
+///
+/// Uses the dual-protocol TLS connector (rustls advertises `h2,http/1.1`
+/// in ALPN; openssl currently advertises nothing -- parity work is a
+/// follow-up). The hyper client carries `TokioTimer` and HTTP/2
+/// keep-alive PINGs so watch streams survive idle-killing intermediaries
+/// such as HAProxy.
+///
+/// If `Config::disable_http2` is set, falls back to building a structurally
+/// identical service via [`build_upgrade_service`] -- both clients then
+/// carry HTTP/1.1-only connectors but the two-client shape stays the same.
+fn build_main_service<H>(
+    connector: H,
+    config: &Config,
+    auth_layer: Option<crate::client::middleware::AuthLayer>,
+    extra_headers_layer: crate::client::middleware::ExtraHeadersLayer,
+) -> Result<GenericService, Error>
+where
+    H: 'static + Clone + Send + Sync + Service<http::Uri>,
+    H::Response: 'static + Connection + Read + Write + Send + Unpin,
+    H::Future: 'static + Send,
+    H::Error: 'static + Send + Sync + std::error::Error,
+{
+    if config.disable_http2 {
+        return build_upgrade_service(connector, config, auth_layer, extra_headers_layer);
+    }
 
-    let client: hyper_util::client::legacy::Client<_, Body> = {
-        // Current TLS feature precedence when more than one are set:
-        // 1. rustls-tls
-        // 2. openssl-tls
-        // Create a custom client to use something else.
-        // If TLS features are not enabled, http connector will be used.
-        #[cfg(feature = "rustls-tls")]
-        let connector = config.rustls_https_connector_with_connector(connector)?;
-        #[cfg(all(not(feature = "rustls-tls"), feature = "openssl-tls"))]
-        let connector = config.openssl_https_connector_with_connector(connector)?;
-        #[cfg(all(not(feature = "rustls-tls"), not(feature = "openssl-tls")))]
+    #[cfg(feature = "rustls-tls")]
+    let connector = config.rustls_https_connector_with_connector(connector)?;
+    #[cfg(all(not(feature = "rustls-tls"), feature = "openssl-tls"))]
+    let connector = config.openssl_https_connector_with_connector(connector)?;
+    #[cfg(all(not(feature = "rustls-tls"), not(feature = "openssl-tls")))]
+    {
         if config.cluster_url.scheme() == Some(&http::uri::Scheme::HTTPS) {
-            // no tls stack situation only works with http scheme
             return Err(Error::TlsRequired);
         }
+    }
 
-        let mut connector = TimeoutConnector::new(connector);
+    let mut connector = TimeoutConnector::new(connector);
+    connector.set_connect_timeout(config.connect_timeout);
+    connector.set_read_timeout(config.read_timeout);
+    connector.set_write_timeout(config.write_timeout);
+
+    let mut builder = hyper_util::client::legacy::Builder::new(TokioExecutor::new());
+    builder
+        .timer(TokioTimer::new())
+        .http2_keep_alive_interval(Duration::from_secs(30))
+        .http2_keep_alive_while_idle(true);
+    let client = builder.build(connector);
+    wrap_with_layers(client, config, auth_layer, extra_headers_layer)
+}
+
+/// Build the HTTP/1.1-only upgrade transport service.
+///
+/// Used by exec, attach, and port-forward; HTTP/1.1 upgrades are
+/// unrepresentable on an HTTP/2 connection. The connector explicitly
+/// advertises `http/1.1` in ALPN (rustls) so the server cannot pick
+/// HTTP/2 at the TLS handshake.
+fn build_upgrade_service<H>(
+    connector: H,
+    config: &Config,
+    auth_layer: Option<crate::client::middleware::AuthLayer>,
+    extra_headers_layer: crate::client::middleware::ExtraHeadersLayer,
+) -> Result<GenericService, Error>
+where
+    H: 'static + Clone + Send + Sync + Service<http::Uri>,
+    H::Response: 'static + Connection + Read + Write + Send + Unpin,
+    H::Future: 'static + Send,
+    H::Error: 'static + Send + Sync + std::error::Error,
+{
+    #[cfg(feature = "rustls-tls")]
+    let connector = config.rustls_https_connector_http1_only_with_connector(connector)?;
+    #[cfg(all(not(feature = "rustls-tls"), feature = "openssl-tls"))]
+    let connector = config.openssl_https_connector_with_connector(connector)?;
+    #[cfg(all(not(feature = "rustls-tls"), not(feature = "openssl-tls")))]
+    {
+        if config.cluster_url.scheme() == Some(&http::uri::Scheme::HTTPS) {
+            return Err(Error::TlsRequired);
+        }
+    }
 
-        // Set the timeouts for the client
-        connector.set_connect_timeout(config.connect_timeout);
-        connector.set_read_timeout(config.read_timeout);
-        connector.set_write_timeout(config.write_timeout);
+    let mut connector = TimeoutConnector::new(connector);
+    connector.set_connect_timeout(config.connect_timeout);
+    connector.set_read_timeout(config.read_timeout);
+    connector.set_write_timeout(config.write_timeout);
 
-        hyper_util::client::legacy::Builder::new(TokioExecutor::new()).build(connector)
-    };
+    let builder = hyper_util::client::legacy::Builder::new(TokioExecutor::new());
+    let client = builder.build(connector);
+    wrap_with_layers(client, config, auth_layer, extra_headers_layer)
+}
 
+/// Wrap a hyper client with the standard tower layer stack (base URI, gzip,
+/// auth, extra headers, tracing) and erase to a [`GenericService`].
+fn wrap_with_layers<C>(
+    client: hyper_util::client::legacy::Client<C, Body>,
+    config: &Config,
+    auth_layer: Option<crate::client::middleware::AuthLayer>,
+    extra_headers_layer: crate::client::middleware::ExtraHeadersLayer,
+) -> Result<GenericService, Error>
+where
+    C: 'static + Clone + Send + Sync + Service<http::Uri>,
+    C::Response: 'static + Connection + Read + Write + Send + Unpin,
+    C::Future: 'static + Send + Unpin,
+    C::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
+{
     let stack = ServiceBuilder::new().layer(config.base_uri_layer()).into_inner();
     #[cfg(feature = "gzip")]
     let stack = ServiceBuilder::new()
@@ -211,7 +374,7 @@ where
     let service = ServiceBuilder::new()
         .layer(stack)
         .option_layer(auth_layer)
-        .layer(config.extra_headers_layer()?)
+        .layer(extra_headers_layer)
         .layer(
             // Attribute names follow [Semantic Conventions].
             // [Semantic Conventions]: https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/trace/semantic_conventions/http.md
@@ -263,19 +426,11 @@ where
         .map_err(BoxError::from)
         .service(client);
 
-    let (_, expiration) = config.exec_identity_pem();
-
-    let client = ClientBuilder::new(
-        service
-            .map_response_body(|body| {
-                Box::new(http_body_util::BodyExt::map_err(body, BoxError::from)) as Box<DynBody>
-            })
-            .boxed(),
-        default_ns,
-    )
-    .with_valid_until(expiration);
-
-    Ok(client)
+    Ok(service
+        .map_response_body(|body| {
+            Box::new(http_body_util::BodyExt::map_err(body, BoxError::from)) as Box<DynBody>
+        })
+        .boxed())
 }
 
 #[cfg(test)]