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;
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};
44use tokio::time::Instant;
45
46use super::Body;
47
48/// Backoff configuration validation error.
49pub use tower::retry::backoff::InvalidBackoff;
50
51/// A retry policy for Kubernetes API requests.
52///
53/// This policy retries requests that fail with:
54/// - 429 Too Many Requests
55/// - 503 Service Unavailable
56/// - 504 Gateway Timeout
57///
58/// Uses exponential backoff starting from `min_delay` up to `max_delay`,
59/// with a configurable maximum number of retries.
60#[derive(Clone)]
61pub struct RetryPolicy {
62    backoff: ExponentialBackoff,
63    current_attempt: u32,
64    max_retries: u32,
65    server_aware: bool,
66}
67
68impl RetryPolicy {
69    /// Create a new retry policy with custom parameters.
70    ///
71    /// # Arguments
72    ///
73    /// * `min_delay` - Initial delay between retries
74    /// * `max_delay` - Maximum delay between retries (cap for exponential growth)
75    /// * `max_retries` - Maximum number of retry attempts
76    /// * `server_retry` - Whether to respect server-provided `Retry-After` header for retry timing
77    ///
78    /// # Errors
79    ///
80    /// Returns [`InvalidBackoff`] if the backoff parameters are invalid.
81    pub fn new(
82        min_delay: Duration,
83        max_delay: Duration,
84        max_retries: u32,
85        server_retry: bool,
86    ) -> Result<Self, InvalidBackoff> {
87        let backoff =
88            ExponentialBackoffMaker::new(min_delay, max_delay, 2.0, HasherRng::new())?.make_backoff();
89
90        Ok(Self {
91            backoff,
92            current_attempt: 0,
93            max_retries,
94            server_aware: server_retry,
95        })
96    }
97
98    /// Create a new server-aware retry policy with parameters optimized for typical controller API usage.
99    ///
100    /// Default parameters:
101    /// - `min_delay`: 5ms
102    /// - `max_delay`: 1000s
103    /// - `max_retries`: 15
104    /// - `server_retry`: true (respects server-provided `Retry-After` header for retry timing)
105    ///
106    /// # Errors
107    ///
108    /// Returns [`InvalidBackoff`] if the backoff parameters are invalid.
109    pub fn server_retry() -> Self {
110        Self::new(Duration::from_millis(5), Duration::from_secs(1000), 15, true)
111            .expect("default server RetryPolicy parameters are valid")
112    }
113
114    /// Check if the status code is retryable.
115    fn is_retryable_status(status: StatusCode) -> bool {
116        matches!(
117            status,
118            StatusCode::TOO_MANY_REQUESTS | StatusCode::SERVICE_UNAVAILABLE | StatusCode::GATEWAY_TIMEOUT
119        )
120    }
121}
122
123impl Default for RetryPolicy {
124    /// Create a default retry policy.
125    ///
126    /// Default parameters:
127    /// - `min_delay`: 500ms
128    /// - `max_delay`: 5s
129    /// - `max_retries`: 3
130    fn default() -> Self {
131        Self::new(Duration::from_millis(500), Duration::from_secs(5), 3, false)
132            .expect("default RetryPolicy parameters are valid")
133    }
134}
135
136impl<Res> Policy<Request<Body>, Response<Res>, BoxError> for RetryPolicy {
137    type Future = tokio::time::Sleep;
138
139    fn retry(
140        &mut self,
141        _req: &mut Request<Body>,
142        result: &mut Result<Response<Res>, BoxError>,
143    ) -> Option<Self::Future> {
144        match result {
145            Ok(response)
146                if Self::is_retryable_status(response.status())
147                    && self.current_attempt < self.max_retries =>
148            {
149                self.current_attempt += 1;
150                // Tick the backoff retry anyways
151                let backoff = self.backoff.next_backoff();
152                if self.server_aware
153                    && let Some(retry_after) = response.headers().get("Retry-After")
154                    && let Some(retry_after) = retry_after.to_str().ok()
155                    && let Some(retry_after) = retry_after.parse::<u64>().ok()
156                {
157                    let server_delay = Duration::from_secs(retry_after);
158                    if Instant::now()
159                        .checked_add(server_delay)
160                        .is_some_and(|retry_after| backoff.deadline() <= retry_after)
161                    {
162                        return Some(tokio::time::sleep(server_delay));
163                    }
164                }
165                Some(backoff)
166            }
167            _ => None,
168        }
169    }
170
171    fn clone_request(&mut self, req: &Request<Body>) -> Option<Request<Body>> {
172        // Try to clone the body - only Kind::Once bodies can be cloned
173        let body = req.body().try_clone()?;
174
175        let mut builder = Request::builder()
176            .method(req.method().clone())
177            .uri(req.uri().clone())
178            .version(req.version());
179
180        // Copy headers
181        if let Some(headers) = builder.headers_mut() {
182            headers.extend(req.headers().clone());
183        }
184
185        // Copy extensions
186        builder.body(body).ok().map(|mut new_req| {
187            *new_req.extensions_mut() = req.extensions().clone();
188            new_req
189        })
190    }
191}
192
193#[cfg(test)]
194mod tests {
195    use super::*;
196
197    #[test]
198    fn test_default_policy() {
199        let policy = RetryPolicy::default();
200        assert_eq!(policy.max_retries, 3);
201    }
202
203    #[test]
204    fn test_retryable_status() {
205        assert!(RetryPolicy::is_retryable_status(StatusCode::TOO_MANY_REQUESTS));
206        assert!(RetryPolicy::is_retryable_status(StatusCode::SERVICE_UNAVAILABLE));
207        assert!(RetryPolicy::is_retryable_status(StatusCode::GATEWAY_TIMEOUT));
208
209        assert!(!RetryPolicy::is_retryable_status(StatusCode::OK));
210        assert!(!RetryPolicy::is_retryable_status(StatusCode::BAD_REQUEST));
211        assert!(!RetryPolicy::is_retryable_status(
212            StatusCode::INTERNAL_SERVER_ERROR
213        ));
214        assert!(!RetryPolicy::is_retryable_status(StatusCode::NOT_FOUND));
215    }
216
217    #[test]
218    fn test_invalid_backoff() {
219        let result = RetryPolicy::new(Duration::from_secs(10), Duration::from_secs(1), 3, false);
220        assert!(result.is_err());
221    }
222
223    #[test]
224    fn test_compile_server_retry() {
225        let policy = RetryPolicy::server_retry();
226        assert!(policy.server_aware);
227        assert_eq!(policy.max_retries, 15);
228    }
229
230    fn retry_after(policy: &mut RetryPolicy, secs: &str) -> Instant {
231        let mut req = Request::new(Body::empty());
232        let mut res: Result<_, BoxError> = Ok(Response::builder()
233            .status(StatusCode::TOO_MANY_REQUESTS)
234            .header("Retry-After", secs)
235            .body(())
236            .unwrap());
237        policy.retry(&mut req, &mut res).expect("429 is retried").deadline()
238    }
239
240    #[tokio::test(start_paused = true)]
241    async fn retry_after_overflow_falls_back_to_backoff() {
242        let now = Instant::now();
243        let mut policy = RetryPolicy::server_retry();
244        assert_eq!(retry_after(&mut policy, "30"), now + Duration::from_secs(30));
245        // `now + Retry-After` overflows an Instant; must not panic.
246        assert!(retry_after(&mut policy, &u64::MAX.to_string()) < now + Duration::from_secs(1));
247    }
248}
kube_client/client/retry.rs

kube_client/client/
retry.rs
