kube_client/client/
builder.rs

1use bytes::Bytes;
2use http::{Request, Response, header::HeaderMap};
3use hyper::{
4    body::Incoming,
5    rt::{Read, Write},
6};
7use hyper_timeout::TimeoutConnector;
8
9use hyper_util::{
10    client::legacy::connect::{Connection, HttpConnector},
11    rt::TokioExecutor,
12};
13
14use jiff::Timestamp;
15use std::time::Duration;
16use tower::{BoxError, Layer, Service, ServiceBuilder, ServiceExt as _, retry::RetryLayer, util::BoxService};
17use tower_http::{ServiceExt as _, classify::ServerErrorsFailureClass, trace::TraceLayer};
18use tracing::Span;
19
20use super::body::Body;
21use crate::{Client, Config, Error, Result, client::{ConfigExt, retry::RetryPolicy}};
22
23/// HTTP body of a dynamic backing type.
24///
25/// The suggested implementation type is [`crate::client::Body`].
26pub type DynBody = dyn http_body::Body<Data = Bytes, Error = BoxError> + Send + Unpin;
27
28/// Builder for [`Client`] instances with customized [tower](`Service`) middleware.
29pub struct ClientBuilder<Svc> {
30    service: Svc,
31    default_ns: String,
32    valid_until: Option<Timestamp>,
33}
34
35impl<Svc> ClientBuilder<Svc> {
36    /// Construct a [`ClientBuilder`] from scratch with a fully custom [`Service`] stack.
37    ///
38    /// This method is only intended for advanced use cases, most users will want to use [`ClientBuilder::try_from`] instead,
39    /// which provides a default stack as a starting point.
40    pub fn new(service: Svc, default_namespace: impl Into<String>) -> Self
41    where
42        Svc: Service<Request<Body>>,
43    {
44        Self {
45            service,
46            default_ns: default_namespace.into(),
47            valid_until: None,
48        }
49    }
50
51    /// Add a [`Layer`] to the current [`Service`] stack.
52    pub fn with_layer<L: Layer<Svc>>(self, layer: &L) -> ClientBuilder<L::Service> {
53        let Self {
54            service: stack,
55            default_ns,
56            valid_until,
57        } = self;
58        ClientBuilder {
59            service: layer.layer(stack),
60            default_ns,
61            valid_until,
62        }
63    }
64
65    /// Sets an expiration timestamp for the client.
66    pub fn with_valid_until(self, valid_until: Option<Timestamp>) -> Self {
67        ClientBuilder {
68            service: self.service,
69            default_ns: self.default_ns,
70            valid_until,
71        }
72    }
73
74    /// Build a [`Client`] instance with the current [`Service`] stack.
75    pub fn build<B>(self) -> Client
76    where
77        Svc: Service<Request<Body>, Response = Response<B>> + Send + 'static,
78        Svc::Future: Send + 'static,
79        Svc::Error: Into<BoxError>,
80        B: http_body::Body<Data = bytes::Bytes> + Send + 'static,
81        B::Error: Into<BoxError>,
82    {
83        Client::new(self.service, self.default_ns).with_valid_until(self.valid_until)
84    }
85}
86
87pub type GenericService = BoxService<Request<Body>, Response<Box<DynBody>>, BoxError>;
88
89#[cfg(feature = "http-proxy")]
90fn with_proxy_basic_auth<C>(
91    proxy_url: &http::Uri,
92    mut connector: hyper_util::client::legacy::connect::proxy::Tunnel<C>,
93) -> hyper_util::client::legacy::connect::proxy::Tunnel<C> {
94    if let Some(authority) = proxy_url.authority()
95        && let Some((userinfo, _)) = authority.as_str().split_once('@')
96    {
97        use base64::Engine;
98        use http::HeaderValue;
99
100        let value = format!(
101            "Basic {}",
102            base64::engine::general_purpose::STANDARD.encode(userinfo)
103        );
104        if let Ok(header) = HeaderValue::from_str(&value) {
105            connector = connector.with_auth(header);
106        }
107    }
108
109    connector
110}
111
112impl TryFrom<Config> for ClientBuilder<GenericService> {
113    type Error = Error;
114
115    /// Builds a default [`ClientBuilder`] stack from a given configuration
116    fn try_from(config: Config) -> Result<Self> {
117        let mut connector = HttpConnector::new();
118        connector.enforce_http(false);
119
120        #[cfg(all(feature = "aws-lc-rs", feature = "rustls-tls"))]
121        {
122            if rustls::crypto::CryptoProvider::get_default().is_none() {
123                // the only error here is if it's been initialized in between: we can ignore it
124                // since our semantic is only to set the default value if it does not exist.
125                let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
126            }
127        }
128
129        match config.proxy_url.as_ref() {
130            Some(proxy_url) if proxy_url.scheme_str() == Some("socks5") => {
131                #[cfg(feature = "socks5")]
132                {
133                    let connector = hyper_util::client::legacy::connect::proxy::SocksV5::new(
134                        proxy_url.clone(),
135                        connector,
136                    );
137                    make_generic_builder(connector, config)
138                }
139
140                #[cfg(not(feature = "socks5"))]
141                Err(Error::ProxyProtocolDisabled {
142                    proxy_url: proxy_url.clone(),
143                    protocol_feature: "kube/socks5",
144                })
145            }
146
147            Some(proxy_url) if proxy_url.scheme_str() == Some("http") => {
148                #[cfg(feature = "http-proxy")]
149                {
150                    let connector =
151                        hyper_util::client::legacy::connect::proxy::Tunnel::new(proxy_url.clone(), connector);
152                    let connector = with_proxy_basic_auth(proxy_url, connector);
153
154                    make_generic_builder(connector, config)
155                }
156
157                #[cfg(not(feature = "http-proxy"))]
158                Err(Error::ProxyProtocolDisabled {
159                    proxy_url: proxy_url.clone(),
160                    protocol_feature: "kube/http-proxy",
161                })
162            }
163
164            Some(proxy_url) if proxy_url.scheme_str() == Some("https") => {
165                #[cfg(all(feature = "http-proxy", any(feature = "rustls-tls", feature = "openssl-tls")))]
166                {
167                    #[cfg(feature = "rustls-tls")]
168                    let proxy_connector = config.rustls_https_connector_with_connector(connector)?;
169                    #[cfg(all(not(feature = "rustls-tls"), feature = "openssl-tls"))]
170                    let proxy_connector = config.openssl_https_connector_with_connector(connector)?;
171
172                    let connector =
173                        hyper_util::client::legacy::connect::proxy::Tunnel::new(proxy_url.clone(), proxy_connector);
174                    let connector = with_proxy_basic_auth(proxy_url, connector);
175
176                    make_generic_builder(connector, config)
177                }
178
179                #[cfg(all(feature = "http-proxy", not(any(feature = "rustls-tls", feature = "openssl-tls"))))]
180                return Err(Error::TlsRequired);
181
182                #[cfg(not(feature = "http-proxy"))]
183                Err(Error::ProxyProtocolDisabled {
184                    proxy_url: proxy_url.clone(),
185                    protocol_feature: "kube/http-proxy",
186                })
187            }
188
189            Some(proxy_url) => Err(Error::ProxyProtocolUnsupported {
190                proxy_url: proxy_url.clone(),
191            }),
192
193            None => make_generic_builder(connector, config),
194        }
195    }
196}
197
198/// Helper function for implementation of [`TryFrom<Config>`] for [`ClientBuilder`].
199/// Ignores [`Config::proxy_url`], which at this point is already handled.
200fn make_generic_builder<H>(connector: H, config: Config) -> Result<ClientBuilder<GenericService>, Error>
201where
202    H: 'static + Clone + Send + Sync + Service<http::Uri>,
203    H::Response: 'static + Connection + Read + Write + Send + Unpin,
204    H::Future: 'static + Send,
205    H::Error: 'static + Send + Sync + std::error::Error,
206{
207    let default_ns = config.default_namespace.clone();
208    let auth_layer = config.auth_layer()?;
209
210    let client: hyper_util::client::legacy::Client<_, Body> = {
211        // Current TLS feature precedence when more than one are set:
212        // 1. rustls-tls
213        // 2. openssl-tls
214        // Create a custom client to use something else.
215        // If TLS features are not enabled, http connector will be used.
216        #[cfg(feature = "rustls-tls")]
217        let connector = config.rustls_https_connector_with_connector(connector)?;
218        #[cfg(all(not(feature = "rustls-tls"), feature = "openssl-tls"))]
219        let connector = config.openssl_https_connector_with_connector(connector)?;
220        #[cfg(all(not(feature = "rustls-tls"), not(feature = "openssl-tls")))]
221        if config.cluster_url.scheme() == Some(&http::uri::Scheme::HTTPS) {
222            // no tls stack situation only works with http scheme
223            return Err(Error::TlsRequired);
224        }
225
226        let mut connector = TimeoutConnector::new(connector);
227
228        // Set the timeouts for the client
229        connector.set_connect_timeout(config.connect_timeout);
230        connector.set_read_timeout(config.read_timeout);
231        connector.set_write_timeout(config.write_timeout);
232
233        hyper_util::client::legacy::Builder::new(TokioExecutor::new()).build(connector)
234    };
235
236    let stack = ServiceBuilder::new().layer(config.base_uri_layer()).into_inner();
237    #[cfg(feature = "gzip")]
238    let stack = ServiceBuilder::new()
239        .layer(stack)
240        .layer(
241            tower_http::decompression::DecompressionLayer::new()
242                .no_br()
243                .no_deflate()
244                .no_zstd()
245                .gzip(!config.disable_compression),
246        )
247        .into_inner();
248
249    let service = ServiceBuilder::new()
250        .layer(stack)
251        .option_layer(config.default_retry.then_some(RetryLayer::new(RetryPolicy::server_retry())))
252        .option_layer(auth_layer)
253        .layer(config.extra_headers_layer()?)
254        .layer(
255            // Attribute names follow [Semantic Conventions].
256            // [Semantic Conventions]: https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/trace/semantic_conventions/http.md
257            TraceLayer::new_for_http()
258                .make_span_with(|req: &Request<Body>| {
259                    tracing::debug_span!(
260                        "HTTP",
261                         http.method = %req.method(),
262                         http.url = %req.uri(),
263                         http.status_code = tracing::field::Empty,
264                         otel.name = req.extensions().get::<&'static str>().unwrap_or(&"HTTP"),
265                         otel.kind = "client",
266                         otel.status_code = tracing::field::Empty,
267                    )
268                })
269                .on_request(|_req: &Request<Body>, _span: &Span| {
270                    tracing::debug!("requesting");
271                })
272                .on_response(|res: &Response<Incoming>, _latency: Duration, span: &Span| {
273                    let status = res.status();
274                    span.record("http.status_code", status.as_u16());
275                    if status.is_client_error() || status.is_server_error() {
276                        span.record("otel.status_code", "ERROR");
277                    }
278                })
279                // Explicitly disable `on_body_chunk`. The default does nothing.
280                .on_body_chunk(())
281                .on_eos(|_: Option<&HeaderMap>, _duration: Duration, _span: &Span| {
282                    tracing::debug!("stream closed");
283                })
284                .on_failure(|ec: ServerErrorsFailureClass, _latency: Duration, span: &Span| {
285                    // Called when
286                    // - Calling the inner service errored
287                    // - Polling `Body` errored
288                    // - the response was classified as failure (5xx)
289                    // - End of stream was classified as failure
290                    span.record("otel.status_code", "ERROR");
291                    match ec {
292                        ServerErrorsFailureClass::StatusCode(status) => {
293                            span.record("http.status_code", status.as_u16());
294                            tracing::error!("failed with status {}", status)
295                        }
296                        ServerErrorsFailureClass::Error(err) => {
297                            tracing::error!("failed with error {}", err)
298                        }
299                    }
300                }),
301        )
302        .map_err(BoxError::from)
303        .service(client);
304
305    let (_, expiration) = config.exec_identity_pem();
306
307    let client = ClientBuilder::new(
308        service
309            .map_response_body(|body| {
310                Box::new(http_body_util::BodyExt::map_err(body, BoxError::from)) as Box<DynBody>
311            })
312            .boxed(),
313        default_ns,
314    )
315    .with_valid_until(expiration);
316
317    Ok(client)
318}
319
320#[cfg(test)]
321mod tests {
322    #[cfg(feature = "gzip")] use super::*;
323
324    #[cfg(feature = "gzip")]
325    #[tokio::test]
326    async fn test_no_accept_encoding_header_sent_when_compression_disabled()
327    -> Result<(), Box<dyn std::error::Error>> {
328        use http::Uri;
329        use std::net::SocketAddr;
330        use tokio::net::{TcpListener, TcpStream};
331
332        // setup a server that echoes back any encoding header value
333        let addr: SocketAddr = ([127, 0, 0, 1], 0).into();
334        let listener = TcpListener::bind(addr).await?;
335        let local_addr = listener.local_addr()?;
336        let uri: Uri = format!("http://{}", local_addr).parse()?;
337
338        tokio::spawn(async move {
339            use http_body_util::Full;
340            use hyper::{server::conn::http1, service::service_fn};
341            use hyper_util::rt::{TokioIo, TokioTimer};
342            use std::convert::Infallible;
343
344            loop {
345                let (tcp, _) = listener.accept().await.unwrap();
346                let io: TokioIo<TcpStream> = TokioIo::new(tcp);
347
348                tokio::spawn(async move {
349                    http1::Builder::new()
350                        .timer(TokioTimer::new())
351                        .serve_connection(
352                            io,
353                            service_fn(|req| async move {
354                                let response = req
355                                    .headers()
356                                    .get(http::header::ACCEPT_ENCODING)
357                                    .map(|b| Bytes::copy_from_slice(b.as_bytes()))
358                                    .unwrap_or_default();
359                                Ok::<_, Infallible>(Response::new(Full::new(response)))
360                            }),
361                        )
362                        .await
363                        .unwrap();
364                });
365            }
366        });
367
368        // confirm gzip echoed back with default config
369        let config = Config { ..Config::new(uri) };
370        let client = make_generic_builder(HttpConnector::new(), config.clone())?.build();
371        let response = client.request_text(http::Request::default()).await?;
372        assert_eq!(&response, "gzip");
373
374        // now disable and check empty string echoed back
375        let config = Config {
376            disable_compression: true,
377            ..config
378        };
379        let client = make_generic_builder(HttpConnector::new(), config)?.build();
380        let response = client.request_text(http::Request::default()).await?;
381        assert_eq!(&response, "");
382
383        Ok(())
384    }
385}
kube_client/client/builder.rs

kube_client/client/
builder.rs
