kube_client/client/
retry.rs

1//! Retry policy for Kubernetes API requests.
2//!
3//! This module provides a [`RetryPolicy`] that implements [`tower::retry::Policy`]
4//! for retrying failed Kubernetes API requests with exponential backoff.
5//!
6//! # Example
7//!
8//! ```no_run
9//! use kube::{Client, Config, client::ConfigExt};
10//! use kube::client::retry::RetryPolicy;
11//! use tower::{ServiceBuilder, BoxError};
12//! use tower::retry::RetryLayer;
13//! use tower::buffer::BufferLayer;
14//! use hyper_util::rt::TokioExecutor;
15//!
16//! # async fn example() -> Result<(), Box<dyn std::error::Error>> {
17//! let config = Config::infer().await?;
18//! let https = config.rustls_https_connector()?;
19//!
20//! let service = ServiceBuilder::new()
21//!     .layer(config.base_uri_layer())
22//!     .option_layer(config.auth_layer()?)
23//!     .layer(BufferLayer::new(1024))
24//!     .layer(RetryLayer::new(RetryPolicy::default()))
25//!     .map_err(BoxError::from)
26//!     .service(hyper_util::client::legacy::Client::builder(TokioExecutor::new()).build(https));
27//!
28//! let client = Client::new(service, config.default_namespace);
29//! # Ok(())
30//! # }
31//! ```
32
33use std::time::{Duration, Instant};
34
35use http::{Request, Response, StatusCode};
36use tower::{
37    BoxError,
38    retry::{
39        Policy,
40        backoff::{Backoff, ExponentialBackoff, ExponentialBackoffMaker, MakeBackoff},
41    },
42    util::rng::HasherRng,
43};
44
45use super::Body;
46
47/// Backoff configuration validation error.
48pub use tower::retry::backoff::InvalidBackoff;
49
50/// A retry policy for Kubernetes API requests.
51///
52/// This policy retries requests that fail with:
53/// - 429 Too Many Requests
54/// - 503 Service Unavailable
55/// - 504 Gateway Timeout
56///
57/// Uses exponential backoff starting from `min_delay` up to `max_delay`,
58/// with a configurable maximum number of retries.
59#[derive(Clone)]
60pub struct RetryPolicy {
61    backoff: ExponentialBackoff,
62    current_attempt: u32,
63    max_retries: u32,
64    server_aware: bool,
65}
66
67impl RetryPolicy {
68    /// Create a new retry policy with custom parameters.
69    ///
70    /// # Arguments
71    ///
72    /// * `min_delay` - Initial delay between retries
73    /// * `max_delay` - Maximum delay between retries (cap for exponential growth)
74    /// * `max_retries` - Maximum number of retry attempts
75    /// * `server_retry` - Whether to respect server-provided `Retry-After` header for retry timing
76    ///
77    /// # Errors
78    ///
79    /// Returns [`InvalidBackoff`] if the backoff parameters are invalid.
80    pub fn new(
81        min_delay: Duration,
82        max_delay: Duration,
83        max_retries: u32,
84        server_retry: bool,
85    ) -> Result<Self, InvalidBackoff> {
86        let backoff =
87            ExponentialBackoffMaker::new(min_delay, max_delay, 2.0, HasherRng::new())?.make_backoff();
88
89        Ok(Self {
90            backoff,
91            current_attempt: 0,
92            max_retries,
93            server_aware: server_retry,
94        })
95    }
96
97    /// Create a new server-aware retry policy with parameters optimized for typical controller API usage.
98    ///
99    /// Default parameters:
100    /// - `min_delay`: 5ms
101    /// - `max_delay`: 1000s
102    /// - `max_retries`: 15
103    /// - `server_retry`: true (respects server-provided `Retry-After` header for retry timing)
104    ///
105    /// # Errors
106    ///
107    /// Returns [`InvalidBackoff`] if the backoff parameters are invalid.
108    pub fn server_retry() -> Self {
109        Self::new(Duration::from_millis(5), Duration::from_secs(1000), 15, true)
110            .expect("default server RetryPolicy parameters are valid")
111    }
112
113    /// Check if the status code is retryable.
114    fn is_retryable_status(status: StatusCode) -> bool {
115        matches!(
116            status,
117            StatusCode::TOO_MANY_REQUESTS | StatusCode::SERVICE_UNAVAILABLE | StatusCode::GATEWAY_TIMEOUT
118        )
119    }
120}
121
122impl Default for RetryPolicy {
123    /// Create a default retry policy.
124    ///
125    /// Default parameters:
126    /// - `min_delay`: 500ms
127    /// - `max_delay`: 5s
128    /// - `max_retries`: 3
129    fn default() -> Self {
130        Self::new(Duration::from_millis(500), Duration::from_secs(5), 3, false)
131            .expect("default RetryPolicy parameters are valid")
132    }
133}
134
135impl<Res> Policy<Request<Body>, Response<Res>, BoxError> for RetryPolicy {
136    type Future = tokio::time::Sleep;
137
138    fn retry(
139        &mut self,
140        _req: &mut Request<Body>,
141        result: &mut Result<Response<Res>, BoxError>,
142    ) -> Option<Self::Future> {
143        match result {
144            Ok(response)
145                if Self::is_retryable_status(response.status())
146                    && self.current_attempt < self.max_retries =>
147            {
148                self.current_attempt += 1;
149                // Tick the backoff retry anyways
150                let backoff = self.backoff.next_backoff();
151                if self.server_aware
152                    && let Some(retry_after) = response.headers().get("Retry-After")
153                    && let Some(retry_after) = retry_after.to_str().ok()
154                    && let Some(retry_after) = retry_after.parse::<u64>().ok()
155                {
156                    let server_delay = Duration::from_secs(retry_after);
157                    let retry_after = Instant::now() + server_delay;
158                    if backoff.deadline().le(&retry_after.into()) {
159                        return Some(tokio::time::sleep(server_delay));
160                    }
161                }
162                Some(backoff)
163            }
164            _ => None,
165        }
166    }
167
168    fn clone_request(&mut self, req: &Request<Body>) -> Option<Request<Body>> {
169        // Try to clone the body - only Kind::Once bodies can be cloned
170        let body = req.body().try_clone()?;
171
172        let mut builder = Request::builder()
173            .method(req.method().clone())
174            .uri(req.uri().clone())
175            .version(req.version());
176
177        // Copy headers
178        if let Some(headers) = builder.headers_mut() {
179            headers.extend(req.headers().clone());
180        }
181
182        // Copy extensions
183        builder.body(body).ok().map(|mut new_req| {
184            *new_req.extensions_mut() = req.extensions().clone();
185            new_req
186        })
187    }
188}
189
190#[cfg(test)]
191mod tests {
192    use super::*;
193
194    #[test]
195    fn test_default_policy() {
196        let policy = RetryPolicy::default();
197        assert_eq!(policy.max_retries, 3);
198    }
199
200    #[test]
201    fn test_retryable_status() {
202        assert!(RetryPolicy::is_retryable_status(StatusCode::TOO_MANY_REQUESTS));
203        assert!(RetryPolicy::is_retryable_status(StatusCode::SERVICE_UNAVAILABLE));
204        assert!(RetryPolicy::is_retryable_status(StatusCode::GATEWAY_TIMEOUT));
205
206        assert!(!RetryPolicy::is_retryable_status(StatusCode::OK));
207        assert!(!RetryPolicy::is_retryable_status(StatusCode::BAD_REQUEST));
208        assert!(!RetryPolicy::is_retryable_status(
209            StatusCode::INTERNAL_SERVER_ERROR
210        ));
211        assert!(!RetryPolicy::is_retryable_status(StatusCode::NOT_FOUND));
212    }
213
214    #[test]
215    fn test_invalid_backoff() {
216        let result = RetryPolicy::new(Duration::from_secs(10), Duration::from_secs(1), 3, false);
217        assert!(result.is_err());
218    }
219
220    #[test]
221    fn test_compile_server_retry() {
222        let policy = RetryPolicy::server_retry();
223        assert!(policy.server_aware);
224        assert_eq!(policy.max_retries, 15);
225    }
226}
kube_client/client/retry.rs

kube_client/client/
retry.rs
