Feature: Add config setting to disable gzip compression #1627 (#1628)

- rounds out all of the https://kubernetes.io/docs/reference/config-api/kubeconfig.v1/#Cluster options

Signed-off-by: Mark Ingram <mark@lincs.dev>
Co-authored-by: Eirik A <sszynrae@gmail.com>

kube-client/Cargo.toml

@@ -81,6 +81,7 @@ form_urlencoded = { workspace = true, optional = true }
 k8s-openapi= { workspace = true, features = [] }
 
 [dev-dependencies]
+hyper = { workspace = true, features = ["server"] }
 kube = { path = "../kube", features = ["derive", "client", "ws"], version = "<1.0.0, >=0.61.0" }
 tempfile.workspace = true
 futures = { workspace = true, features = ["async-await"] }

kube-client/src/client/builder.rs

@@ -178,7 +178,13 @@ where
     #[cfg(feature = "gzip")]
     let stack = ServiceBuilder::new()
         .layer(stack)
-        .layer(tower_http::decompression::DecompressionLayer::new())
+        .layer(
+            tower_http::decompression::DecompressionLayer::new()
+                .no_br()
+                .no_deflate()
+                .no_zstd()
+                .gzip(!config.disable_compression),
+        )
         .into_inner();
 
     let service = ServiceBuilder::new()
@@ -246,3 +252,70 @@ where
         default_ns,
     ))
 }
+
+#[cfg(test)]
+mod tests {
+    #[cfg(feature = "gzip")] use super::*;
+
+    #[cfg(feature = "gzip")]
+    #[tokio::test]
+    async fn test_no_accept_encoding_header_sent_when_compression_disabled(
+    ) -> Result<(), Box<dyn std::error::Error>> {
+        use http::Uri;
+        use std::net::SocketAddr;
+        use tokio::net::{TcpListener, TcpStream};
+
+        // setup a server that echoes back any encoding header value
+        let addr: SocketAddr = ([127, 0, 0, 1], 0).into();
+        let listener = TcpListener::bind(addr).await?;
+        let local_addr = listener.local_addr()?;
+        let uri: Uri = format!("http://{}", local_addr).parse()?;
+
+        tokio::spawn(async move {
+            use http_body_util::Full;
+            use hyper::{server::conn::http1, service::service_fn};
+            use hyper_util::rt::{TokioIo, TokioTimer};
+            use std::convert::Infallible;
+
+            loop {
+                let (tcp, _) = listener.accept().await.unwrap();
+                let io: TokioIo<TcpStream> = TokioIo::new(tcp);
+
+                tokio::spawn(async move {
+                    let _ = http1::Builder::new()
+                        .timer(TokioTimer::new())
+                        .serve_connection(
+                            io,
+                            service_fn(|req| async move {
+                                let response = req
+                                    .headers()
+                                    .get(http::header::ACCEPT_ENCODING)
+                                    .map(|b| Bytes::copy_from_slice(b.as_bytes()))
+                                    .unwrap_or_default();
+                                Ok::<_, Infallible>(Response::new(Full::new(response)))
+                            }),
+                        )
+                        .await
+                        .unwrap();
+                });
+            }
+        });
+
+        // confirm gzip echoed back with default config
+        let config = Config { ..Config::new(uri) };
+        let client = make_generic_builder(HttpConnector::new(), config.clone())?.build();
+        let response = client.request_text(http::Request::default()).await?;
+        assert_eq!(&response, "gzip");
+
+        // now disable and check empty string echoed back
+        let config = Config {
+            disable_compression: true,
+            ..config
+        };
+        let client = make_generic_builder(HttpConnector::new(), config)?.build();
+        let response = client.request_text(http::Request::default()).await?;
+        assert_eq!(&response, "");
+
+        Ok(())
+    }
+}

kube-client/src/config/file_config.rs

@@ -111,6 +111,14 @@ pub struct Cluster {
     #[serde(rename = "proxy-url")]
     #[serde(skip_serializing_if = "Option::is_none")]
     pub proxy_url: Option<String>,
+    /// Compression is enabled by default with the `gzip` feature.
+    /// `disable_compression` allows client to opt-out of response compression for all requests to the server.
+    /// This is useful to speed up requests (specifically lists) when client-server network bandwidth is ample,
+    /// by saving time on compression (server-side) and decompression (client-side):
+    /// https://github.com/kubernetes/kubernetes/issues/112296
+    #[serde(rename = "disable-compression")]
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub disable_compression: Option<bool>,
     /// Name used to check server certificate.
     ///
     /// If `tls_server_name` is `None`, the hostname used to contact the server is used.

kube-client/src/config/mod.rs

@@ -151,6 +151,8 @@ pub struct Config {
     pub accept_invalid_certs: bool,
     /// Stores information to tell the cluster who you are.
     pub auth_info: AuthInfo,
+    /// Whether to disable compression (would only have an effect when the `gzip` feature is enabled)
+    pub disable_compression: bool,
     /// Optional proxy URL. Proxy support requires the `socks5` feature.
     pub proxy_url: Option<http::Uri>,
     /// If set, apiserver certificate will be validated to contain this string
@@ -177,6 +179,7 @@ impl Config {
             write_timeout: Some(DEFAULT_WRITE_TIMEOUT),
             accept_invalid_certs: false,
             auth_info: AuthInfo::default(),
+            disable_compression: false,
             proxy_url: None,
             tls_server_name: None,
             headers: Vec::new(),
@@ -259,6 +262,7 @@ impl Config {
                 token_file: Some(incluster_config::token_file()),
                 ..Default::default()
             },
+            disable_compression: false,
             proxy_url: None,
             tls_server_name: None,
             headers: Vec::new(),
@@ -302,6 +306,8 @@ impl Config {
             .unwrap_or_else(|| String::from("default"));
 
         let accept_invalid_certs = loader.cluster.insecure_skip_tls_verify.unwrap_or(false);
+        let disable_compression = loader.cluster.disable_compression.unwrap_or(false);
+
         let mut root_cert = None;
 
         if let Some(ca_bundle) = loader.ca_bundle()? {
@@ -316,6 +322,7 @@ impl Config {
             read_timeout: Some(DEFAULT_READ_TIMEOUT),
             write_timeout: Some(DEFAULT_WRITE_TIMEOUT),
             accept_invalid_certs,
+            disable_compression,
             proxy_url: loader.proxy_url()?,
             auth_info: loader.user,
             tls_server_name: loader.cluster.tls_server_name,