kube_runtime/
watcher.rs

1//! Watches a Kubernetes Resource for changes, with error recovery
2//!
3//! See [`watcher`] for the primary entry point.
4
5use crate::utils::{Backoff, ResetTimerBackoff};
6
7use backon::BackoffBuilder;
8use educe::Educe;
9use futures::{Stream, StreamExt, stream::BoxStream};
10use kube_client::{
11    Api, Error as ClientErr,
12    api::{ListParams, Resource, ResourceExt, VersionMatch, WatchEvent, WatchParams},
13    core::{ObjectList, Selector, metadata::PartialObjectMeta},
14    error::Status,
15};
16use serde::de::DeserializeOwned;
17#[cfg(test)]
18#[path = "stub_watcher.rs"]
19mod stub_watcher;
20
21use std::{
22    clone::Clone,
23    collections::VecDeque,
24    fmt::{Debug, Display},
25    future::{self, Future},
26    time::Duration,
27};
28use thiserror::Error;
29use tracing::{debug, error, warn};
30
31/// Errors that a watcher can emit
32///
33/// These are all considered retryable from a watcher's point of view,
34/// even though they may require patching of rbac/netpols in the background to fix.
35///
36/// To avoid constantly looping errors, make sure backoff is applied.
37#[derive(Debug, Error)]
38pub enum Error {
39    /// Received a raw error while performing an api.list
40    #[error("failed to perform initial object list: {0}")]
41    InitialListFailed(#[source] kube_client::Error),
42
43    /// Received a raw error while starting an api.watch
44    #[error("failed to start watching object: {0}")]
45    WatchStartFailed(#[source] kube_client::Error),
46
47    /// Received a `WatchEvent::Error` from the apiserver
48    #[error("error returned by apiserver during watch: {0}")]
49    WatchError(#[source] Box<Status>),
50
51    /// Received a raw error while watching
52    #[error("watch stream failed: {0}")]
53    WatchFailed(#[source] kube_client::Error),
54
55    /// Missing resource version field from api server
56    #[error("no metadata.resourceVersion in watch result (does resource support watch?)")]
57    NoResourceVersion,
58}
59
60/// Type alias for Result with a `watcher::Error` as default.
61pub type Result<T, E = Error> = std::result::Result<T, E>;
62
63#[derive(Debug, Clone)]
64/// Watch events returned from the [`watcher`]
65pub enum Event<K> {
66    /// An object was added or modified
67    Apply(K),
68    /// An object was deleted
69    ///
70    /// The object need not have `metadata.deletionTimestamp` set or a changed
71    /// `metadata.generation`. After decoding with
72    /// [`touched_objects`](crate::WatchStreamExt::touched_objects), predicates based on
73    /// unchanged fields can suppress it. See [predicate deletion handling](crate::predicates).
74    ///
75    /// NOTE: This should not be used for managing persistent state elsewhere, since
76    /// events may be lost if the watcher is unavailable. Use
77    /// [finalizers](crate::finalizer::finalizer) instead.
78    Delete(K),
79    /// The watch stream was restarted.
80    ///
81    /// A series of `InitApply` events are expected to follow until all matching objects
82    /// have been listed. This event can be used to prepare a buffer for `InitApply` events.
83    Init,
84    /// Received an object during `Init`.
85    ///
86    /// Objects returned here are either from the initial stream using the `StreamingList` strategy,
87    /// or from pages using the `ListWatch` strategy.
88    ///
89    /// These events can be passed up if having a complete set of objects is not a concern.
90    /// If you need to wait for a complete set, please buffer these events until an `InitDone`.
91    InitApply(K),
92    /// The initialisation is complete.
93    ///
94    /// This can be used as a signal to replace buffered store contents atomically.
95    /// No more `InitApply` events will happen until the next `Init` event.
96    ///
97    /// Any objects that were previously [`Applied`](Event::Applied) but are not listed in any of
98    /// the `InitApply` events should be assumed to have been [`Deleted`](Event::Deleted).
99    InitDone,
100}
101
102impl<K> Display for Event<K>
103where
104    K: ResourceExt,
105{
106    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
107        let desc = match self {
108            Event::Init => "Init".into(),
109            Event::InitApply(obj) => format!("InitApply({})", obj.name_any()),
110            Event::InitDone => "InitDone".into(),
111            Event::Apply(obj) => format!("Apply({})", obj.name_any()),
112            Event::Delete(obj) => format!("Delete({})", obj.name_any()),
113        };
114        write!(f, "{desc}")
115    }
116}
117
118impl<K> Event<K> {
119    /// Map each object in an event through a mutator fn
120    ///
121    /// This allows for memory optimizations in watch streams.
122    /// If you are chaining a watch stream into a reflector as an in memory state store,
123    /// you can control the space used by each object by dropping fields.
124    ///
125    /// ```no_run
126    /// use k8s_openapi::api::core::v1::Pod;
127    /// use kube::ResourceExt;
128    /// # use kube::runtime::watcher::Event;
129    /// # let event: Event<Pod> = todo!();
130    /// event.modify(|pod| {
131    ///     pod.managed_fields_mut().clear();
132    ///     pod.annotations_mut().clear();
133    ///     pod.status = None;
134    /// });
135    /// ```
136    #[must_use]
137    pub fn modify(mut self, mut f: impl FnMut(&mut K)) -> Self {
138        match &mut self {
139            Self::Apply(obj) | Self::Delete(obj) | Self::InitApply(obj) => (f)(obj),
140            Self::Init | Self::InitDone => {} // markers, nothing to modify
141        }
142        self
143    }
144}
145
146#[derive(Educe, Default)]
147#[educe(Debug)]
148/// The internal finite state machine driving the [`watcher`]
149enum State<K> {
150    /// The Watcher is empty, and the next [`poll`](Stream::poll_next) will start the initial LIST to get all existing objects
151    #[default]
152    Empty,
153    /// The Watcher is in the process of paginating through the initial LIST
154    InitPage {
155        continue_token: Option<String>,
156        objects: VecDeque<K>,
157        last_bookmark: Option<String>,
158    },
159    /// Kubernetes 1.27 Streaming Lists
160    /// The initial watch is in progress
161    InitialWatch {
162        #[educe(Debug(ignore))]
163        stream: BoxStream<'static, kube_client::Result<WatchEvent<K>>>,
164    },
165    /// The initial LIST was successful, so we should move on to starting the actual watch.
166    InitListed { resource_version: String },
167    /// The watch is in progress, from this point we just return events from the server.
168    ///
169    /// If the connection is disrupted then we propagate the error but try to restart the watch stream by
170    /// returning to the `InitListed` state.
171    /// If we fall out of the K8s watch window then we propagate the error and fall back doing a re-list
172    /// with `Empty`.
173    Watching {
174        resource_version: String,
175        #[educe(Debug(ignore))]
176        stream: BoxStream<'static, kube_client::Result<WatchEvent<K>>>,
177    },
178}
179
180/// Used to control whether the watcher receives the full object, or only the
181/// metadata
182trait ApiMode: Send + Sync {
183    type Value: Clone + Send + 'static;
184
185    fn list(
186        &self,
187        lp: &ListParams,
188    ) -> impl Future<Output = kube_client::Result<ObjectList<Self::Value>>> + Send;
189    fn watch(
190        &self,
191        wp: &WatchParams,
192        version: &str,
193    ) -> impl Future<
194        Output = kube_client::Result<BoxStream<'static, kube_client::Result<WatchEvent<Self::Value>>>>,
195    > + Send;
196}
197
198/// A wrapper around the `Api` of a `Resource` type that when used by the
199/// watcher will return the entire (full) object
200struct FullObject<K> {
201    api: Api<K>,
202}
203
204/// Configurable list semantics for `watcher` relists
205#[derive(Clone, Default, Debug, PartialEq)]
206pub enum ListSemantic {
207    /// List calls perform a full quorum read for most recent results
208    ///
209    /// Prefer this if you have strong consistency requirements. Note that this
210    /// is more taxing for the apiserver and can be less scalable for the cluster.
211    ///
212    /// If you are observing large resource sets (such as congested `Controller` cases),
213    /// you typically have a delay between the list call completing, and all the events
214    /// getting processed. In such cases, it is probably worth picking `Any` over `MostRecent`,
215    /// as your events are not guaranteed to be up-to-date by the time you get to them anyway.
216    #[default]
217    MostRecent,
218
219    /// List calls returns cached results from apiserver
220    ///
221    /// This is faster and much less taxing on the apiserver, but can result
222    /// in much older results than has previously observed for `Restarted` events,
223    /// particularly in HA configurations, due to partitions or stale caches.
224    ///
225    /// This option makes the most sense for controller usage where events have
226    /// some delay between being seen by the runtime, and it being sent to the reconciler.
227    Any,
228}
229
230/// Configurable watcher listwatch semantics
231
232#[derive(Clone, Default, Debug, PartialEq)]
233pub enum InitialListStrategy {
234    /// List first, then watch from given resource version
235    ///
236    /// This is the old and default way of watching. The watcher will do a paginated list call first before watching.
237    /// When using this mode, you can configure the `page_size` on the watcher.
238    #[default]
239    ListWatch,
240    /// Kubernetes 1.27 Streaming Lists
241    ///
242    /// See [upstream documentation on streaming lists](https://kubernetes.io/docs/reference/using-api/api-concepts/#streaming-lists),
243    /// and the [KEP](https://github.com/kubernetes/enhancements/tree/master/keps/sig-api-machinery/3157-watch-list#design-details).
244    StreamingList,
245}
246
247/// Accumulates all options that can be used on the watcher invocation.
248#[derive(Clone, Debug, PartialEq)]
249pub struct Config {
250    /// A selector to restrict the list of returned objects by their labels.
251    ///
252    /// Defaults to everything if `None`.
253    pub label_selector: Option<String>,
254
255    /// A selector to restrict the list of returned objects by their fields.
256    ///
257    /// Defaults to everything if `None`.
258    pub field_selector: Option<String>,
259
260    /// Timeout for the watch call.
261    ///
262    /// This limits the duration of the call, regardless of any activity or inactivity.
263    /// If unset, we will use 290s.
264    /// The watcher's dead-connection detection window is this value plus 5s,
265    /// so larger values delay noticing a silently dropped connection.
266    pub timeout: Option<u32>,
267
268    /// Semantics for list calls.
269    ///
270    /// Configures re-list for performance vs. consistency.
271    ///
272    /// NB: This option only has an effect for [`InitialListStrategy::ListWatch`].
273    pub list_semantic: ListSemantic,
274
275    /// Control how the watcher fetches the initial list of objects.
276    ///
277    /// - `ListWatch`: The watcher will fetch the initial list of objects using a list call.
278    /// - `StreamingList`: The watcher will fetch the initial list of objects using a watch call.
279    ///
280    /// `StreamingList` is more efficient than `ListWatch`, but it requires the server to support
281    /// streaming list bookmarks (opt-in feature gate in Kubernetes 1.27).
282    ///
283    /// See [upstream documentation on streaming lists](https://kubernetes.io/docs/reference/using-api/api-concepts/#streaming-lists),
284    /// and the [KEP](https://github.com/kubernetes/enhancements/tree/master/keps/sig-api-machinery/3157-watch-list#design-details).
285    pub initial_list_strategy: InitialListStrategy,
286
287    /// Maximum number of objects retrieved per list operation resyncs.
288    ///
289    /// This can reduce the memory consumption during resyncs, at the cost of requiring more
290    /// API roundtrips to complete.
291    ///
292    /// Defaults to 500. Note that `None` represents unbounded.
293    ///
294    /// NB: This option only has an effect for [`InitialListStrategy::ListWatch`].
295    pub page_size: Option<u32>,
296
297    /// Enables watch events with type "BOOKMARK".
298    ///
299    /// Requests watch bookmarks from the apiserver when enabled for improved watch precision and reduced list calls.
300    /// This is default enabled and should generally not be turned off.
301    pub bookmarks: bool,
302}
303
304impl Default for Config {
305    fn default() -> Self {
306        Self {
307            bookmarks: true,
308            label_selector: None,
309            field_selector: None,
310            timeout: None,
311            list_semantic: ListSemantic::default(),
312            // same default page size limit as client-go
313            // https://github.com/kubernetes/client-go/blob/aed71fa5cf054e1c196d67b2e21f66fd967b8ab1/tools/pager/pager.go#L31
314            page_size: Some(500),
315            initial_list_strategy: InitialListStrategy::ListWatch,
316        }
317    }
318}
319
320/// Builder interface to Config
321///
322/// Usage:
323/// ```
324/// use kube::runtime::watcher::Config;
325/// let wc = Config::default()
326///     .timeout(60)
327///     .labels("kubernetes.io/lifecycle=spot");
328/// ```
329impl Config {
330    /// Configure the timeout for watch calls
331    ///
332    /// This limits the duration of the call, regardless of any activity or inactivity.
333    /// Defaults to 290s
334    #[must_use]
335    pub fn timeout(mut self, timeout_secs: u32) -> Self {
336        self.timeout = Some(timeout_secs);
337        self
338    }
339
340    /// Configure the selector to restrict the list of returned objects by their fields.
341    ///
342    /// Defaults to everything.
343    /// Supports `=`, `==`, `!=`, and can be comma separated: `key1=value1,key2=value2`.
344    /// The server only supports a limited number of field queries per type.
345    #[must_use]
346    pub fn fields(mut self, field_selector: &str) -> Self {
347        self.field_selector = Some(field_selector.to_string());
348        self
349    }
350
351    /// Configure the selector to restrict the list of returned objects by their labels.
352    ///
353    /// Defaults to everything.
354    /// Supports `=`, `==`, `!=`, and can be comma separated: `key1=value1,key2=value2`.
355    #[must_use]
356    pub fn labels(mut self, label_selector: &str) -> Self {
357        self.label_selector = Some(label_selector.to_string());
358        self
359    }
360
361    /// Configure typed label selectors
362    ///
363    /// Configure typed selectors from [`Selector`](kube_client::core::Selector) and [`Expression`](kube_client::core::Expression) lists.
364    ///
365    /// ```
366    /// use kube_runtime::watcher::Config;
367    /// use kube_client::core::{Expression, Selector, ParseExpressionError};
368    /// use k8s_openapi::apimachinery::pkg::apis::meta::v1::LabelSelector;
369    /// let selector: Selector = Expression::In("env".into(), ["development".into(), "sandbox".into()].into()).into();
370    /// let cfg = Config::default().labels_from(&selector);
371    /// let cfg = Config::default().labels_from(&Expression::Exists("foo".into()).into());
372    /// let selector: Selector = LabelSelector::default().try_into()?;
373    /// let cfg = Config::default().labels_from(&selector);
374    /// # Ok::<(), ParseExpressionError>(())
375    ///```
376    #[must_use]
377    pub fn labels_from(mut self, selector: &Selector) -> Self {
378        self.label_selector = Some(selector.to_string());
379        self
380    }
381
382    /// Sets list semantic to configure re-list performance and consistency
383    ///
384    /// NB: This option only has an effect for [`InitialListStrategy::ListWatch`].
385    #[must_use]
386    pub fn list_semantic(mut self, semantic: ListSemantic) -> Self {
387        self.list_semantic = semantic;
388        self
389    }
390
391    /// Sets list semantic to `Any` to improve list performance
392    ///
393    /// NB: This option only has an effect for [`InitialListStrategy::ListWatch`].
394    #[must_use]
395    pub fn any_semantic(self) -> Self {
396        self.list_semantic(ListSemantic::Any)
397    }
398
399    /// Disables watch bookmarks to simplify watch handling
400    ///
401    /// This is not recommended to use with production watchers as it can cause desyncs.
402    /// See [#219](https://github.com/kube-rs/kube/issues/219) for details.
403    #[must_use]
404    pub fn disable_bookmarks(mut self) -> Self {
405        self.bookmarks = false;
406        self
407    }
408
409    /// Limits the number of objects retrieved in each list operation during resync.
410    ///
411    /// This can reduce the memory consumption during resyncs, at the cost of requiring more
412    /// API roundtrips to complete.
413    ///
414    /// NB: This option only has an effect for [`InitialListStrategy::ListWatch`].
415    #[must_use]
416    pub fn page_size(mut self, page_size: u32) -> Self {
417        self.page_size = Some(page_size);
418        self
419    }
420
421    /// Kubernetes 1.27 Streaming Lists
422    /// Sets list semantic to `Stream` to make use of watch bookmarks
423    #[must_use]
424    pub fn streaming_lists(mut self) -> Self {
425        self.initial_list_strategy = InitialListStrategy::StreamingList;
426        self
427    }
428
429    /// Converts generic `watcher::Config` structure to the instance of `ListParams` used for list requests.
430    fn to_list_params(&self) -> ListParams {
431        let (resource_version, version_match) = match self.list_semantic {
432            ListSemantic::Any => (Some("0".into()), Some(VersionMatch::NotOlderThan)),
433            ListSemantic::MostRecent => (None, None),
434        };
435        ListParams {
436            label_selector: self.label_selector.clone(),
437            field_selector: self.field_selector.clone(),
438            version_match,
439            resource_version,
440            // The watcher handles pagination internally.
441            limit: self.page_size,
442            continue_token: None,
443            ..Default::default()
444        }
445    }
446
447    /// Converts generic `watcher::Config` structure to the instance of `WatchParams` used for watch requests.
448    fn to_watch_params(&self, phase: WatchPhase) -> WatchParams {
449        WatchParams {
450            label_selector: self.label_selector.clone(),
451            field_selector: self.field_selector.clone(),
452            timeout: self.timeout,
453            bookmarks: self.bookmarks,
454            send_initial_events: phase == WatchPhase::Initial
455                && self.initial_list_strategy == InitialListStrategy::StreamingList,
456        }
457    }
458}
459
460/// Distinguishes between initial watch and resumed watch for streaming lists.
461///
462/// This is used to determine whether to set `sendInitialEvents=true` in watch requests.
463/// Only initial watches should request initial events; reconnections should not.
464#[derive(Clone, Copy, Debug, PartialEq, Eq)]
465enum WatchPhase {
466    /// Initial watch from `State::Empty` - requests initial events for streaming lists
467    Initial,
468    /// Resumed watch from `State::InitListed` - does not request initial events
469    Resumed,
470}
471
472impl<K> ApiMode for FullObject<K>
473where
474    K: Clone + Debug + DeserializeOwned + Send + 'static,
475{
476    type Value = K;
477
478    async fn list(&self, lp: &ListParams) -> kube_client::Result<ObjectList<Self::Value>> {
479        self.api.list(lp).await
480    }
481
482    async fn watch(
483        &self,
484        wp: &WatchParams,
485        version: &str,
486    ) -> kube_client::Result<BoxStream<'static, kube_client::Result<WatchEvent<Self::Value>>>> {
487        self.api.watch(wp, version).await.map(StreamExt::boxed)
488    }
489}
490
491/// A wrapper around the `Api` of a `Resource` type that when used by the
492/// watcher will return only the metadata associated with an object
493struct MetaOnly<K> {
494    api: Api<K>,
495}
496
497impl<K> ApiMode for MetaOnly<K>
498where
499    K: Clone + Debug + DeserializeOwned + Send + 'static,
500{
501    type Value = PartialObjectMeta<K>;
502
503    async fn list(&self, lp: &ListParams) -> kube_client::Result<ObjectList<Self::Value>> {
504        self.api.list_metadata(lp).await
505    }
506
507    async fn watch(
508        &self,
509        wp: &WatchParams,
510        version: &str,
511    ) -> kube_client::Result<BoxStream<'static, kube_client::Result<WatchEvent<Self::Value>>>> {
512        self.api.watch_metadata(wp, version).await.map(StreamExt::boxed)
513    }
514}
515
516/// Client-side idle timeout margin added on top of the server-side watch timeout.
517///
518/// The server closes the watch stream after the configured timeout (default 290s).
519/// We add a small margin so the client detects dead connections where the
520/// server's close never arrives (e.g. network failure).
521const WATCH_IDLE_TIMEOUT_MARGIN: Duration = Duration::from_secs(5);
522
523/// Poll the next item from a watch stream with an idle timeout.
524///
525/// Returns `None` when the stream ends **or** when no item arrives within
526/// `timeout + WATCH_IDLE_TIMEOUT_MARGIN`, causing the watcher to
527/// treat the connection as dead and reconnect. An explicit zero timeout
528/// disables this client-side idle timeout.
529async fn next_with_idle_timeout<S, T>(stream: &mut S, timeout: Option<u32>) -> Option<T>
530where
531    S: Stream<Item = T> + Unpin,
532{
533    if timeout == Some(0) {
534        return stream.next().await;
535    }
536
537    let idle_timeout = Duration::from_secs(u64::from(timeout.unwrap_or(290))) + WATCH_IDLE_TIMEOUT_MARGIN;
538    match tokio::time::timeout(idle_timeout, stream.next()).await {
539        Ok(item) => item,
540        Err(_elapsed) => {
541            debug!(
542                timeout_secs = idle_timeout.as_secs(),
543                "watch stream idle timeout, reconnecting"
544            );
545            None
546        }
547    }
548}
549
550/// Progresses the watcher a single step, returning (event, state)
551///
552/// This function should be trampolined: if event == `None`
553/// then the function should be called again until it returns a Some.
554#[allow(clippy::too_many_lines)] // for now
555async fn step_trampolined<A>(
556    api: &A,
557    wc: &Config,
558    state: State<A::Value>,
559) -> (Option<Result<Event<A::Value>>>, State<A::Value>)
560where
561    A: ApiMode,
562    A::Value: Resource + 'static,
563{
564    match state {
565        State::Empty => match wc.initial_list_strategy {
566            InitialListStrategy::ListWatch => (Some(Ok(Event::Init)), State::InitPage {
567                continue_token: None,
568                objects: VecDeque::default(),
569                last_bookmark: None,
570            }),
571            InitialListStrategy::StreamingList => {
572                match api.watch(&wc.to_watch_params(WatchPhase::Initial), "0").await {
573                    Ok(stream) => (Some(Ok(Event::Init)), State::InitialWatch { stream }),
574                    Err(err) => {
575                        if std::matches!(err, ClientErr::Api(ref status) if status.is_forbidden()) {
576                            warn!("watch initlist error with 403: {err:?}");
577                        } else {
578                            debug!("watch initlist error: {err:?}");
579                        }
580                        (Some(Err(Error::WatchStartFailed(err))), State::default())
581                    }
582                }
583            }
584        },
585        State::InitPage {
586            continue_token,
587            mut objects,
588            last_bookmark,
589        } => {
590            if let Some(next) = objects.pop_front() {
591                return (Some(Ok(Event::InitApply(next))), State::InitPage {
592                    continue_token,
593                    objects,
594                    last_bookmark,
595                });
596            }
597            // check if we need to perform more pages
598            if continue_token.is_none()
599                && let Some(resource_version) = last_bookmark
600            {
601                // we have drained the last page - move on to next stage
602                return (Some(Ok(Event::InitDone)), State::InitListed { resource_version });
603            }
604            let mut lp = wc.to_list_params();
605            lp.continue_token = continue_token;
606            match api.list(&lp).await {
607                Ok(list) => {
608                    let last_bookmark = list.metadata.resource_version.filter(|s| !s.is_empty());
609                    let continue_token = list.metadata.continue_.filter(|s| !s.is_empty());
610                    if last_bookmark.is_none() && continue_token.is_none() {
611                        return (Some(Err(Error::NoResourceVersion)), State::Empty);
612                    }
613                    // Buffer page here, causing us to return to this enum branch (State::InitPage)
614                    // until the objects buffer has drained
615                    (None, State::InitPage {
616                        continue_token,
617                        objects: list.items.into_iter().collect(),
618                        last_bookmark,
619                    })
620                }
621                Err(err) => {
622                    if std::matches!(err, ClientErr::Api(ref status) if status.is_forbidden()) {
623                        warn!("watch list error with 403: {err:?}");
624                    } else {
625                        debug!("watch list error: {err:?}");
626                    }
627                    (Some(Err(Error::InitialListFailed(err))), State::Empty)
628                }
629            }
630        }
631        State::InitialWatch { mut stream } => {
632            match next_with_idle_timeout(&mut stream, wc.timeout).await {
633                Some(Ok(WatchEvent::Added(obj) | WatchEvent::Modified(obj))) => {
634                    (Some(Ok(Event::InitApply(obj))), State::InitialWatch { stream })
635                }
636                Some(Ok(WatchEvent::Deleted(_obj))) => {
637                    // Kubernetes claims these events are impossible
638                    // https://kubernetes.io/docs/reference/using-api/api-concepts/#streaming-lists
639                    error!("got deleted event during initial watch. this is a bug");
640                    (None, State::InitialWatch { stream })
641                }
642                Some(Ok(WatchEvent::Bookmark(bm))) => {
643                    let marks_initial_end = bm.metadata.annotations.contains_key("k8s.io/initial-events-end");
644                    if marks_initial_end {
645                        (Some(Ok(Event::InitDone)), State::Watching {
646                            resource_version: bm.metadata.resource_version,
647                            stream,
648                        })
649                    } else {
650                        (None, State::InitialWatch { stream })
651                    }
652                }
653                Some(Ok(WatchEvent::Error(err))) => {
654                    // HTTP GONE, means we have desynced and need to start over and re-list :(
655                    let new_state = if err.code == 410 {
656                        State::default()
657                    } else {
658                        State::InitialWatch { stream }
659                    };
660                    if err.code == 403 {
661                        warn!("watcher watchevent error 403: {err:?}");
662                    } else {
663                        debug!("error watchevent error: {err:?}");
664                    }
665                    (Some(Err(Error::WatchError(err.boxed()))), new_state)
666                }
667                Some(Err(err)) => {
668                    if std::matches!(err, ClientErr::Api(ref status) if status.is_forbidden()) {
669                        warn!("watcher error 403: {err:?}");
670                    } else {
671                        debug!("watcher error: {err:?}");
672                    }
673                    (Some(Err(Error::WatchFailed(err))), State::InitialWatch { stream })
674                }
675                None => (None, State::default()),
676            }
677        }
678        State::InitListed { resource_version } => {
679            match api
680                .watch(&wc.to_watch_params(WatchPhase::Resumed), &resource_version)
681                .await
682            {
683                Ok(stream) => (None, State::Watching {
684                    resource_version,
685                    stream,
686                }),
687                Err(err) => {
688                    if std::matches!(err, ClientErr::Api(ref status) if status.is_forbidden()) {
689                        warn!("watch initlist error with 403: {err:?}");
690                    } else {
691                        debug!("watch initlist error: {err:?}");
692                    }
693                    (Some(Err(Error::WatchStartFailed(err))), State::InitListed {
694                        resource_version,
695                    })
696                }
697            }
698        }
699        State::Watching {
700            resource_version,
701            mut stream,
702        } => match next_with_idle_timeout(&mut stream, wc.timeout).await {
703            Some(Ok(WatchEvent::Added(obj) | WatchEvent::Modified(obj))) => {
704                let resource_version = obj.resource_version().unwrap_or_default();
705                if resource_version.is_empty() {
706                    (Some(Err(Error::NoResourceVersion)), State::default())
707                } else {
708                    (Some(Ok(Event::Apply(obj))), State::Watching {
709                        resource_version,
710                        stream,
711                    })
712                }
713            }
714            Some(Ok(WatchEvent::Deleted(obj))) => {
715                let resource_version = obj.resource_version().unwrap_or_default();
716                if resource_version.is_empty() {
717                    (Some(Err(Error::NoResourceVersion)), State::default())
718                } else {
719                    (Some(Ok(Event::Delete(obj))), State::Watching {
720                        resource_version,
721                        stream,
722                    })
723                }
724            }
725            Some(Ok(WatchEvent::Bookmark(bm))) => (None, State::Watching {
726                resource_version: bm.metadata.resource_version,
727                stream,
728            }),
729            Some(Ok(WatchEvent::Error(err))) => {
730                // HTTP GONE, means we have desynced and need to start over and re-list :(
731                let new_state = if err.code == 410 {
732                    State::default()
733                } else {
734                    State::Watching {
735                        resource_version,
736                        stream,
737                    }
738                };
739                if err.code == 403 {
740                    warn!("watcher watchevent error 403: {err:?}");
741                } else {
742                    debug!("error watchevent error: {err:?}");
743                }
744                (Some(Err(Error::WatchError(err.boxed()))), new_state)
745            }
746            Some(Err(err)) => {
747                if std::matches!(err, ClientErr::Api(ref status) if status.is_forbidden()) {
748                    warn!("watcher error 403: {err:?}");
749                } else {
750                    debug!("watcher error: {err:?}");
751                }
752                (Some(Err(Error::WatchFailed(err))), State::Watching {
753                    resource_version,
754                    stream,
755                })
756            }
757            None => (None, State::InitListed { resource_version }),
758        },
759    }
760}
761
762/// Trampoline helper for `step_trampolined`
763async fn step<A>(
764    api: &A,
765    config: &Config,
766    mut state: State<A::Value>,
767) -> (Result<Event<A::Value>>, State<A::Value>)
768where
769    A: ApiMode,
770    A::Value: Resource + 'static,
771{
772    loop {
773        match step_trampolined(api, config, state).await {
774            (Some(result), new_state) => return (result, new_state),
775            (None, new_state) => state = new_state,
776        }
777    }
778}
779
780/// Watches a Kubernetes Resource for changes continuously
781///
782/// Compared to [`Api::watch`], this automatically tries to recover the stream upon errors.
783///
784/// Errors from the underlying watch are propagated, after which the stream will go into recovery mode on the next poll.
785/// You can apply your own backoff by not polling the stream for a duration after errors.
786/// Keep in mind that some [`TryStream`](futures::TryStream) combinators (such as
787/// [`try_for_each`](futures::TryStreamExt::try_for_each) and [`try_concat`](futures::TryStreamExt::try_concat))
788/// will terminate eagerly as soon as they receive an [`Err`].
789///
790/// The events are intended to provide a safe input interface for a state store like a [`reflector`].
791/// Direct users may want to use [`WatchStreamExt`] for higher-level constructs.
792///
793/// ```
794/// use kube::{
795///   api::{Api, ResourceExt}, Client,
796///   runtime::{watcher, WatchStreamExt}
797/// };
798/// use k8s_openapi::api::core::v1::Pod;
799/// use futures::TryStreamExt;
800///
801/// # async fn wrapper() -> Result<(), watcher::Error> {
802/// #   let client: Client = todo!();
803/// let pods: Api<Pod> = Api::namespaced(client, "apps");
804///
805/// watcher(pods, watcher::Config::default()).applied_objects()
806///     .try_for_each(|p| async move {
807///         println!("Applied: {}", p.name_any());
808///        Ok(())
809///     })
810///     .await?;
811/// # Ok(())
812/// # }
813/// ```
814/// [`WatchStreamExt`]: super::WatchStreamExt
815/// [`reflector`]: super::reflector::reflector
816/// [`Api::watch`]: kube_client::Api::watch
817///
818/// # Recovery
819///
820/// The stream will attempt to be recovered on the next poll after an [`Err`] is returned.
821/// This will normally happen immediately, but you can use [`StreamBackoff`](crate::utils::StreamBackoff)
822/// to introduce an artificial delay. [`default_backoff`] returns a suitable default set of parameters.
823///
824/// If the watch connection is interrupted, then `watcher` will attempt to restart the watch using the last
825/// [resource version](https://kubernetes.io/docs/reference/using-api/api-concepts/#efficient-detection-of-changes)
826/// that we have seen on the stream. If this is successful then the stream is simply resumed from where it left off.
827/// If this fails because the resource version is no longer valid then we start over with a new stream, starting with
828/// an [`Event::Init`]. The internals mechanics of recovery should be considered an implementation detail.
829#[doc(alias = "informer")]
830pub fn watcher<K: Resource + Clone + DeserializeOwned + Debug + Send + 'static>(
831    api: Api<K>,
832    watcher_config: Config,
833) -> impl Stream<Item = Result<Event<K>>> + Send {
834    watcher_inner(FullObject { api }, watcher_config)
835}
836
837/// [`ApiMode`] compatibility helper. Introduced later so that the core logic of [`watcher`] can be tested without
838/// having to modify its pre-existing signature.
839fn watcher_inner<A>(api: A, watcher_config: Config) -> impl Stream<Item = Result<Event<A::Value>>> + Send
840where
841    A: ApiMode,
842    A::Value: Resource,
843{
844    futures::stream::unfold(
845        (api, watcher_config, State::default()),
846        |(api, watcher_config, state)| async {
847            let (event, state) = step(&api, &watcher_config, state).await;
848            Some((event, (api, watcher_config, state)))
849        },
850    )
851}
852
853/// Watches a Kubernetes Resource for changes continuously and receives only the
854/// metadata
855///
856/// Compared to [`Api::watch_metadata`], this automatically tries to recover the stream upon errors.
857///
858/// Errors from the underlying watch are propagated, after which the stream will go into recovery mode on the next poll.
859/// You can apply your own backoff by not polling the stream for a duration after errors.
860/// Keep in mind that some [`TryStream`](futures::TryStream) combinators (such as
861/// [`try_for_each`](futures::TryStreamExt::try_for_each) and [`try_concat`](futures::TryStreamExt::try_concat))
862/// will terminate eagerly as soon as they receive an [`Err`].
863///
864/// The events are intended to provide a safe input interface for a state store like a [`reflector`].
865/// Direct users may want to use [`WatchStreamExt`] for higher-level constructs.
866///
867/// ```
868/// use kube::{
869///   api::{Api, ResourceExt}, Client,
870///   runtime::{watcher, metadata_watcher, WatchStreamExt}
871/// };
872/// use k8s_openapi::api::core::v1::Pod;
873/// use futures::TryStreamExt;
874///
875/// # async fn wrapper() -> Result<(), watcher::Error> {
876/// #   let client: Client = todo!();
877/// let pods: Api<Pod> = Api::namespaced(client, "apps");
878///
879/// metadata_watcher(pods, watcher::Config::default()).applied_objects()
880///         .try_for_each(|p| async move {
881///          println!("Applied: {}", p.name_any());
882///             Ok(())
883///         })
884///         .await?;
885/// #   Ok(())
886/// # }
887/// ```
888/// [`WatchStreamExt`]: super::WatchStreamExt
889/// [`reflector`]: super::reflector::reflector
890/// [`Api::watch`]: kube_client::Api::watch
891///
892/// # Recovery
893///
894/// The stream will attempt to be recovered on the next poll after an [`Err`] is returned.
895/// This will normally happen immediately, but you can use [`StreamBackoff`](crate::utils::StreamBackoff)
896/// to introduce an artificial delay. [`default_backoff`] returns a suitable default set of parameters.
897///
898/// If the watch connection is interrupted, then `watcher` will attempt to restart the watch using the last
899/// [resource version](https://kubernetes.io/docs/reference/using-api/api-concepts/#efficient-detection-of-changes)
900/// that we have seen on the stream. If this is successful then the stream is simply resumed from where it left off.
901/// If this fails because the resource version is no longer valid then we start over with a new stream, starting with
902/// an [`Event::Init`]. The internals mechanics of recovery should be considered an implementation detail.
903#[deprecated(
904    since = "3.1.0",
905    note = "Use `watcher(Api::<PartialObjectMeta<K>>::all(client), config)` instead. \
906            `Api<PartialObjectMeta<K>>` now automatically uses metadata-only requests."
907)]
908pub fn metadata_watcher<K: Resource + Clone + DeserializeOwned + Debug + Send + 'static>(
909    api: Api<K>,
910    watcher_config: Config,
911) -> impl Stream<Item = Result<Event<PartialObjectMeta<K>>>> + Send {
912    watcher_inner(MetaOnly { api }, watcher_config)
913}
914
915/// Watch a single named object for updates
916///
917/// Emits `None` if the object is deleted (or not found), and `Some` if an object is updated (or created/found).
918///
919/// Often invoked indirectly via [`await_condition`](crate::wait::await_condition()).
920///
921/// ## Scope Warning
922///
923/// When using this with an `Api::all` on namespaced resources there is a chance of duplicated names.
924/// To avoid getting confusing / wrong answers for this, use `Api::namespaced` bound to a specific namespace
925/// when watching for transitions to namespaced objects.
926pub fn watch_object<K: Resource + Clone + DeserializeOwned + Debug + Send + 'static>(
927    api: Api<K>,
928    name: &str,
929) -> impl Stream<Item = Result<Option<K>>> + Send + use<K> {
930    // filtering by object name in given scope, so there's at most one matching object
931    // footgun: Api::all may generate events from namespaced objects with the same name in different namespaces
932    let fields = format!("metadata.name={name}");
933    watcher(api, Config::default().fields(&fields))
934        // The `obj_seen` state is used to track whether the object exists in each Init / InitApply / InitDone
935        // sequence of events. If the object wasn't seen in any particular sequence it is treated as deleted and
936        // `None` is emitted when the InitDone event is received.
937        //
938        // The first check ensures `None` is emitted if the object was already gone (or not found), subsequent
939        // checks ensure `None` is emitted even if for some reason the Delete event wasn't received, which
940        // could happen given K8S events aren't guaranteed delivery.
941        .scan(false, |obj_seen, event| {
942            if matches!(event, Ok(Event::Init)) {
943                *obj_seen = false;
944            } else if matches!(event, Ok(Event::InitApply(_))) {
945                *obj_seen = true;
946            }
947            future::ready(Some((*obj_seen, event)))
948        })
949        .filter_map(|(obj_seen, event)| async move {
950            match event {
951                // Pass up `Some` for Found / Updated
952                Ok(Event::Apply(obj) | Event::InitApply(obj)) => Some(Ok(Some(obj))),
953                // Pass up `None` for Deleted
954                Ok(Event::Delete(_)) => Some(Ok(None)),
955                // Pass up `None` if the object wasn't seen in the initial list
956                Ok(Event::InitDone) if !obj_seen => Some(Ok(None)),
957                // Ignore marker events
958                Ok(Event::Init | Event::InitDone) => None,
959                // Bubble up errors
960                Err(err) => Some(Err(err)),
961            }
962        })
963}
964
965/// A struct with a manually configured exponential backoff
966pub struct ExponentialBackoff {
967    inner: backon::ExponentialBackoff,
968    builder: backon::ExponentialBuilder,
969}
970
971impl ExponentialBackoff {
972    fn new(min_delay: Duration, max_delay: Duration, factor: f32, enable_jitter: bool) -> Self {
973        let builder = backon::ExponentialBuilder::default()
974            .with_min_delay(min_delay)
975            .with_max_delay(max_delay)
976            .with_factor(factor)
977            .without_max_times();
978
979        let builder = if enable_jitter {
980            builder.with_jitter()
981        } else {
982            builder
983        };
984
985        Self {
986            inner: builder.build(),
987            builder,
988        }
989    }
990}
991
992impl Backoff for ExponentialBackoff {
993    fn reset(&mut self) {
994        self.inner = self.builder.build();
995    }
996}
997
998impl Iterator for ExponentialBackoff {
999    type Item = Duration;
1000
1001    fn next(&mut self) -> Option<Self::Item> {
1002        self.inner.next()
1003    }
1004}
1005
1006impl From<backon::ExponentialBuilder> for ExponentialBackoff {
1007    fn from(builder: backon::ExponentialBuilder) -> Self {
1008        Self {
1009            inner: builder.build(),
1010            builder,
1011        }
1012    }
1013}
1014
1015/// Default watcher backoff inspired by Kubernetes' client-go.
1016///
1017/// The parameters currently optimize for being kind to struggling apiservers.
1018/// The exact parameters are taken from
1019/// [client-go's reflector source](https://github.com/kubernetes/client-go/blob/980663e185ab6fc79163b1c2565034f6d58368db/tools/cache/reflector.go#L177-L181)
1020/// and should not be considered stable.
1021///
1022/// This struct implements [`Backoff`] and is the default strategy used
1023/// when calling `WatchStreamExt::default_backoff`. If you need to create
1024/// this manually then [`DefaultBackoff::default`] can be used.
1025pub struct DefaultBackoff(Strategy);
1026type Strategy = ResetTimerBackoff<ExponentialBackoff>;
1027
1028impl Default for DefaultBackoff {
1029    fn default() -> Self {
1030        Self(ResetTimerBackoff::new(
1031            ExponentialBackoff::new(Duration::from_millis(800), Duration::from_secs(30), 2.0, true),
1032            Duration::from_secs(120),
1033        ))
1034    }
1035}
1036
1037impl Iterator for DefaultBackoff {
1038    type Item = Duration;
1039
1040    fn next(&mut self) -> Option<Self::Item> {
1041        self.0.next()
1042    }
1043}
1044
1045impl Backoff for DefaultBackoff {
1046    fn reset(&mut self) {
1047        self.0.reset();
1048    }
1049}
1050
1051#[cfg(test)]
1052mod tests {
1053    use std::{collections::BTreeMap, time::Duration};
1054
1055    use futures::StreamExt;
1056    use k8s_openapi::{api::core::v1::ConfigMap, apimachinery::pkg::apis::meta::v1::ObjectMeta};
1057    use kube_client::{
1058        api::{TypeMeta, WatchEvent, WatchParams},
1059        core::{
1060            Status,
1061            watch::{Bookmark, BookmarkMeta},
1062        },
1063    };
1064
1065    use crate::watcher::{
1066        ApiMode, Config, ExponentialBackoff, State, WatchPhase, next_with_idle_timeout, step,
1067        stub_watcher::{
1068            Recording, Sequence, SequenceStep, TestMode, error_indicates_graceful_watch_seq_exhaustion,
1069        },
1070        watcher_inner,
1071    };
1072
1073    use super::stub_watcher::ResultPage;
1074
1075    fn config_map(name: &str, resource_version: &str) -> ConfigMap {
1076        ConfigMap {
1077            metadata: ObjectMeta {
1078                name: Some(name.to_string()),
1079                resource_version: Some(resource_version.to_string()),
1080                ..ObjectMeta::default()
1081            },
1082            ..ConfigMap::default()
1083        }
1084    }
1085
1086    #[tokio::test(start_paused = true)]
1087    async fn test_mode_watch_waits_before_returning_next_event() {
1088        let api = TestMode::new(vec![], vec![Sequence::new(vec![
1089            SequenceStep::Wait(Duration::from_millis(50)),
1090            SequenceStep::List(vec![Ok(WatchEvent::Added(config_map("a", "1")))].into()),
1091        ])]);
1092        let mut stream = api.watch(&WatchParams::default(), "0").await.unwrap();
1093
1094        assert!(futures::poll!(stream.next()).is_pending());
1095        tokio::time::advance(Duration::from_millis(49)).await;
1096        assert!(futures::poll!(stream.next()).is_pending());
1097        tokio::time::advance(Duration::from_millis(1)).await;
1098
1099        assert!(matches!(
1100            stream.next().await,
1101            Some(Ok(WatchEvent::Added(config_map))) if config_map.metadata.name.as_deref() == Some("a")
1102        ));
1103    }
1104
1105    #[allow(clippy::too_many_lines)]
1106    #[tokio::test]
1107    async fn streaming_list_resync() {
1108        const LAST_VALID_RESOURCE_VERSION: &str = "5";
1109        let mut recoverable_err = Box::new(Status::default());
1110        recoverable_err.code = 502;
1111        recoverable_err.message = "some err".to_string();
1112        recoverable_err.reason = "Something went wrong".to_string();
1113        let api = TestMode::new(vec![], vec![
1114            Sequence::new(vec![
1115                SequenceStep::List(
1116                    vec![
1117                        Ok(WatchEvent::Added(config_map("a", "2"))),
1118                        Ok(WatchEvent::Added(config_map("b", "3"))),
1119                        Ok(WatchEvent::Added(config_map("c", "4"))),
1120                        // Details here: https://kubernetes.io/docs/reference/using-api/api-concepts/#streaming-lists
1121                        // Should only be sent when requested via allowWatchBookmarks=true
1122                        // Since TestMode does not contain any real logic and  is not dynamically
1123                        // progammable, this is up to the test
1124                        // author to return or not for now.
1125                        Ok(WatchEvent::Bookmark(Bookmark {
1126                            types: TypeMeta::resource::<ConfigMap>(),
1127                            metadata: BookmarkMeta {
1128                                resource_version: "4".to_string(),
1129                                annotations: BTreeMap::from_iter(vec![(
1130                                    "k8s.io/initial-events-end".into(),
1131                                    "true".into(),
1132                                )]),
1133                            },
1134                        })),
1135                    ]
1136                    .into(),
1137                ),
1138                SequenceStep::List(
1139                    vec![
1140                        Ok(WatchEvent::Modified(config_map("a", LAST_VALID_RESOURCE_VERSION))),
1141                        Ok(WatchEvent::Error(recoverable_err)),
1142                    ]
1143                    .into(),
1144                ),
1145                // After an error we expect the watcher to request its last(latest) known
1146                // resource version to be queried again.
1147                SequenceStep::List(
1148                    vec![Ok(WatchEvent::Modified(config_map(
1149                        "a",
1150                        LAST_VALID_RESOURCE_VERSION,
1151                    )))]
1152                    .into(),
1153                ),
1154            ]),
1155            // After an error we expect the watcher to request its last(latest) known
1156            // resource version to be queried again.
1157            Sequence::new(vec![SequenceStep::List(
1158                vec![
1159                    Ok(WatchEvent::Added(config_map("a", "5"))),
1160                    Ok(WatchEvent::Bookmark(Bookmark {
1161                        types: TypeMeta::resource::<ConfigMap>(),
1162                        metadata: BookmarkMeta {
1163                            resource_version: "5".to_string(),
1164                            annotations: BTreeMap::from_iter(vec![(
1165                                "k8s.io/initial-events-end".into(),
1166                                "true".into(),
1167                            )]),
1168                        },
1169                    })),
1170                ]
1171                .into(),
1172            )]),
1173        ]);
1174
1175        let config = Config::default()
1176            .timeout(1)
1177            .streaming_lists()
1178            .labels("app=test")
1179            .fields("metadata.name!=ignored");
1180        // NOTE(juf): The stream which is returned here is not the same stream that an individual `watch` result produces.
1181        // That stream is wrapped and hidden behind the inner state [`kube_runtime::watcher::State`] of the watcher implementation.
1182        let mut stream = std::pin::pin!(watcher_inner(api.clone(), config));
1183
1184        for expected in [
1185            "Init",
1186            "InitApply(a)",
1187            "InitApply(b)",
1188            "InitApply(c)",
1189            "InitDone",
1190            "Apply(a)",
1191            "error returned by apiserver during watch: some err: Something went wrong",
1192            "Apply(a)",
1193            "Apply(a)",
1194        ] {
1195            let event = stream.next().await.expect("watcher stream should not end");
1196            let repr = match event {
1197                Ok(event) => event.to_string(),
1198                Err(err) => err.to_string(),
1199            };
1200            assert_eq!(repr, expected);
1201        }
1202
1203        let event = tokio::time::timeout(Duration::from_millis(100), stream.next())
1204            .await
1205            .expect("watcher should return when the TestMode sequence is exhausted")
1206            .expect("watcher stream should not end");
1207        // See function docs for more context, we expect this error, when we want to check whether
1208        // the whole watch sequence has been consumed and as breaker to avoid consuming an infinite
1209        // async stream that returns None.
1210        dbg!(&event);
1211        assert!(event.is_err_and(|err| error_indicates_graceful_watch_seq_exhaustion(&err)),);
1212
1213        let records = api.get_recordings();
1214        match &records[..] {
1215            [
1216                Recording::Watch(initial_params, initial_version),
1217                Recording::Watch(resumed_params, resumed_version),
1218                Recording::Watch(exhausted_params, exhausted_version),
1219            ] => {
1220                assert_eq!(initial_version, "0");
1221                assert_eq!(initial_params.label_selector.as_deref(), Some("app=test"));
1222                assert_eq!(
1223                    initial_params.field_selector.as_deref(),
1224                    Some("metadata.name!=ignored")
1225                );
1226                assert!(initial_params.send_initial_events);
1227
1228                for (watch_params, watch_version) in [
1229                    (resumed_params, resumed_version),
1230                    (exhausted_params, exhausted_version),
1231                ] {
1232                    assert_eq!(watch_version, LAST_VALID_RESOURCE_VERSION);
1233                    assert_eq!(watch_params.label_selector.as_deref(), Some("app=test"));
1234                    assert_eq!(
1235                        watch_params.field_selector.as_deref(),
1236                        Some("metadata.name!=ignored")
1237                    );
1238                    assert!(!watch_params.send_initial_events);
1239                }
1240            }
1241            _ => panic!("unexpected API call sequence {records:?}"),
1242        }
1243    }
1244
1245    #[tokio::test]
1246    async fn streaming_list_init() {
1247        let api = TestMode::new(vec![], vec![Sequence::new(vec![
1248            SequenceStep::List(
1249                vec![
1250                    Ok(WatchEvent::Added(config_map("a", "2"))),
1251                    Ok(WatchEvent::Added(config_map("b", "3"))),
1252                    Ok(WatchEvent::Added(config_map("c", "4"))),
1253                    // Details here: https://kubernetes.io/docs/reference/using-api/api-concepts/#streaming-lists
1254                    // Should only be sent when requested via allowWatchBookmarks=true
1255                    // Since TestMode does not contain any real logic and  is not dynamically
1256                    // progammable, this is up to the test
1257                    // author to return or not for now.
1258                    Ok(WatchEvent::Bookmark(Bookmark {
1259                        types: TypeMeta::resource::<ConfigMap>(),
1260                        metadata: BookmarkMeta {
1261                            resource_version: "145".to_string(),
1262                            annotations: BTreeMap::from_iter(vec![(
1263                                "k8s.io/initial-events-end".into(),
1264                                "true".into(),
1265                            )]),
1266                        },
1267                    })),
1268                ]
1269                .into(),
1270            ),
1271            SequenceStep::List(
1272                vec![
1273                    Ok(WatchEvent::Modified(config_map("a", "5"))),
1274                    Ok(WatchEvent::Deleted(config_map("b", "7"))),
1275                    Ok(WatchEvent::Modified(config_map("c", "9"))),
1276                ]
1277                .into(),
1278            ),
1279        ])]);
1280
1281        let config = Config::default()
1282            .timeout(1)
1283            .streaming_lists()
1284            .labels("app=test")
1285            .fields("metadata.name!=ignored");
1286        let mut state = State::default();
1287        for expected in [
1288            "Init",
1289            "InitApply(a)",
1290            "InitApply(b)",
1291            "InitApply(c)",
1292            "InitDone",
1293            "Apply(a)",
1294            "Delete(b)",
1295            "Apply(c)",
1296        ] {
1297            let (event, next) = step(&api, &config, state).await;
1298            state = next;
1299            assert_eq!(event.unwrap().to_string(), expected);
1300        }
1301
1302        let (event, _) = tokio::time::timeout(Duration::from_millis(100), step(&api, &config, state))
1303            .await
1304            .expect("step should return when the TestMode sequence is exhausted");
1305        // See function docs for more context, we expect this error, when we want to check whether
1306        // the whole watch sequence has been consumed and as breaker to avoid consuming an infinite
1307        // async stream that returns None.
1308        assert!(event.is_err_and(|err| error_indicates_graceful_watch_seq_exhaustion(&err)));
1309
1310        let records = api.get_recordings();
1311        match &records[..] {
1312            [
1313                Recording::Watch(initial_params, initial_version),
1314                Recording::Watch(resumed_params, resumed_version),
1315            ] => {
1316                assert_eq!(initial_version, "0");
1317                assert_eq!(initial_params.label_selector.as_deref(), Some("app=test"));
1318                assert_eq!(
1319                    initial_params.field_selector.as_deref(),
1320                    Some("metadata.name!=ignored")
1321                );
1322                assert!(initial_params.send_initial_events);
1323
1324                assert_eq!(resumed_version, "9");
1325                assert_eq!(resumed_params.label_selector.as_deref(), Some("app=test"));
1326                assert_eq!(
1327                    resumed_params.field_selector.as_deref(),
1328                    Some("metadata.name!=ignored")
1329                );
1330                assert!(!resumed_params.send_initial_events);
1331            }
1332            _ => panic!("unexpected API call sequence {records:?}"),
1333        }
1334    }
1335
1336    #[tokio::test]
1337    async fn streaming_list_initial_stream_eof_before_bookmark_drops_partial_objects() {
1338        let api = TestMode::new(vec![], vec![
1339            Sequence::new(vec![SequenceStep::List(
1340                vec![
1341                    Ok(WatchEvent::Added(config_map("a", "1"))),
1342                    Ok(WatchEvent::Added(config_map("b", "2"))),
1343                ]
1344                .into(),
1345            )]),
1346            Sequence::new(vec![SequenceStep::List(
1347                vec![
1348                    Ok(WatchEvent::Added(config_map("a", "1"))),
1349                    Ok(WatchEvent::Bookmark(Bookmark {
1350                        types: TypeMeta::resource::<ConfigMap>(),
1351                        metadata: BookmarkMeta {
1352                            resource_version: "1".to_string(),
1353                            annotations: BTreeMap::from_iter(vec![(
1354                                "k8s.io/initial-events-end".into(),
1355                                "true".into(),
1356                            )]),
1357                        },
1358                    })),
1359                ]
1360                .into(),
1361            )]),
1362        ]);
1363
1364        let config = Config::default().streaming_lists();
1365        let (reader, mut writer) = crate::reflector::store::<ConfigMap>();
1366        let mut state = State::default();
1367        for expected in [
1368            "Init",
1369            "InitApply(a)",
1370            "InitApply(b)",
1371            "Init",
1372            "InitApply(a)",
1373            "InitDone",
1374        ] {
1375            let (event, next) = step(&api, &config, state).await;
1376            state = next;
1377            let event = event.unwrap();
1378            assert_eq!(event.to_string(), expected);
1379            writer.apply_watcher_event(&event);
1380        }
1381
1382        let mut names: Vec<_> = reader
1383            .state()
1384            .iter()
1385            .filter_map(|cm| cm.metadata.name.clone())
1386            .collect();
1387        names.sort();
1388        assert_eq!(names, ["a".to_string()]);
1389
1390        let records = api.get_recordings();
1391        match &records[..] {
1392            [
1393                Recording::Watch(first, first_version),
1394                Recording::Watch(second, second_version),
1395            ] => {
1396                assert_eq!(first_version, "0");
1397                assert!(first.send_initial_events);
1398                assert_eq!(second_version, "0");
1399                assert!(second.send_initial_events);
1400            }
1401            _ => panic!("unexpected API call sequence {records:?}"),
1402        }
1403    }
1404
1405    #[tokio::test]
1406    async fn streaming_list_initial_watch_410_before_bookmark_drops_partial_objects() {
1407        let mut gone = Box::new(Status::default());
1408        gone.code = 410;
1409        gone.message = "expired".to_string();
1410        gone.reason = "Expired".to_string();
1411        let api = TestMode::new(vec![], vec![
1412            Sequence::new(vec![SequenceStep::List(
1413                vec![
1414                    Ok(WatchEvent::Added(config_map("a", "1"))),
1415                    Ok(WatchEvent::Added(config_map("b", "2"))),
1416                    Ok(WatchEvent::Error(gone)),
1417                ]
1418                .into(),
1419            )]),
1420            Sequence::new(vec![SequenceStep::List(
1421                vec![
1422                    Ok(WatchEvent::Added(config_map("a", "1"))),
1423                    Ok(WatchEvent::Bookmark(Bookmark {
1424                        types: TypeMeta::resource::<ConfigMap>(),
1425                        metadata: BookmarkMeta {
1426                            resource_version: "1".to_string(),
1427                            annotations: BTreeMap::from_iter(vec![(
1428                                "k8s.io/initial-events-end".into(),
1429                                "true".into(),
1430                            )]),
1431                        },
1432                    })),
1433                ]
1434                .into(),
1435            )]),
1436        ]);
1437
1438        let config = Config::default().streaming_lists();
1439        let (reader, mut writer) = crate::reflector::store::<ConfigMap>();
1440        let mut state = State::default();
1441        for expected in [
1442            "Init",
1443            "InitApply(a)",
1444            "InitApply(b)",
1445            "error returned by apiserver during watch: expired: Expired",
1446            "Init",
1447            "InitApply(a)",
1448            "InitDone",
1449        ] {
1450            let (event, next) = step(&api, &config, state).await;
1451            state = next;
1452            let repr = match &event {
1453                Ok(event) => event.to_string(),
1454                Err(err) => err.to_string(),
1455            };
1456            assert_eq!(repr, expected);
1457            if let Ok(event) = &event {
1458                writer.apply_watcher_event(event);
1459            }
1460        }
1461
1462        let mut names: Vec<_> = reader
1463            .state()
1464            .iter()
1465            .filter_map(|cm| cm.metadata.name.clone())
1466            .collect();
1467        names.sort();
1468        assert_eq!(names, ["a".to_string()]);
1469
1470        let records = api.get_recordings();
1471        match &records[..] {
1472            [
1473                Recording::Watch(first, first_version),
1474                Recording::Watch(second, second_version),
1475            ] => {
1476                assert_eq!(first_version, "0");
1477                assert!(first.send_initial_events);
1478                assert_eq!(second_version, "0");
1479                assert!(second.send_initial_events);
1480            }
1481            _ => panic!("unexpected API call sequence {records:?}"),
1482        }
1483    }
1484
1485    #[tokio::test]
1486    async fn listwatch_paginates_initial_list_before_starting_watch() {
1487        let api = TestMode::new(
1488            vec![
1489                Ok(ResultPage::empty()
1490                    .items(vec![config_map("a", "1")])
1491                    .continue_token("first".into())
1492                    .into()),
1493                Ok(ResultPage::empty()
1494                    .items(vec![config_map("b", "2")])
1495                    .resource_version("2".into())
1496                    .continue_token("second".into())
1497                    .into()),
1498                Ok(ResultPage::empty()
1499                    .items(vec![config_map("c", "3")])
1500                    .resource_version("3".into())
1501                    .into()),
1502            ],
1503            vec![Sequence::new(vec![SequenceStep::List(
1504                vec![Ok(WatchEvent::Added(config_map("d", "4")))].into(),
1505            )])],
1506        );
1507
1508        let config = Config::default()
1509            .page_size(1)
1510            .labels("app=test")
1511            .fields("metadata.name!=ignored");
1512        let mut state = State::default();
1513
1514        let (event, next) = step(&api, &config, state).await;
1515        state = next;
1516        assert_eq!(event.unwrap().to_string(), "Init");
1517
1518        let (event, next) = step(&api, &config, state).await;
1519        state = next;
1520        assert_eq!(event.unwrap().to_string(), "InitApply(a)");
1521
1522        let (event, next) = step(&api, &config, state).await;
1523        state = next;
1524        assert_eq!(event.unwrap().to_string(), "InitApply(b)");
1525
1526        let (event, next) = step(&api, &config, state).await;
1527        state = next;
1528        assert_eq!(event.unwrap().to_string(), "InitApply(c)");
1529
1530        let (event, next) = step(&api, &config, state).await;
1531        state = next;
1532        assert_eq!(event.unwrap().to_string(), "InitDone");
1533
1534        let (event, _) = step(&api, &config, state).await;
1535        assert_eq!(event.unwrap().to_string(), "Apply(d)");
1536
1537        let records = api.get_recordings();
1538        match &records[..] {
1539            [
1540                Recording::List(first),
1541                Recording::List(second),
1542                Recording::List(third),
1543                Recording::Watch(watch_params, watch_version),
1544            ] => {
1545                assert_eq!(first.continue_token.as_deref(), None);
1546                assert_eq!(second.continue_token.as_deref(), Some("first"));
1547                assert_eq!(third.continue_token.as_deref(), Some("second"));
1548
1549                assert_eq!(first.limit, Some(1));
1550                assert_eq!(first.label_selector.as_deref(), Some("app=test"));
1551                assert_eq!(first.field_selector.as_deref(), Some("metadata.name!=ignored"));
1552
1553                assert_eq!(watch_version, "3");
1554                assert_eq!(watch_params.label_selector.as_deref(), Some("app=test"));
1555                assert_eq!(
1556                    watch_params.field_selector.as_deref(),
1557                    Some("metadata.name!=ignored")
1558                );
1559                assert!(!watch_params.send_initial_events);
1560            }
1561            _ => panic!("unexpected API call sequence"),
1562        }
1563    }
1564
1565    #[test]
1566    fn to_watch_params_initial_phase_with_streaming_list_sets_send_initial_events() {
1567        let config = Config::default().streaming_lists();
1568        let params = config.to_watch_params(WatchPhase::Initial);
1569        assert!(params.send_initial_events);
1570    }
1571
1572    #[test]
1573    fn to_watch_params_resumed_phase_with_streaming_list_does_not_set_send_initial_events() {
1574        let config = Config::default().streaming_lists();
1575        let params = config.to_watch_params(WatchPhase::Resumed);
1576        assert!(!params.send_initial_events);
1577    }
1578
1579    #[test]
1580    fn to_watch_params_listwatch_mode_does_not_set_send_initial_events() {
1581        let config = Config::default(); // ListWatch mode
1582        let params_initial = config.to_watch_params(WatchPhase::Initial);
1583        let params_resumed = config.to_watch_params(WatchPhase::Resumed);
1584        assert!(!params_initial.send_initial_events);
1585        assert!(!params_resumed.send_initial_events);
1586    }
1587
1588    fn approx_eq(a: Duration, b: Duration) -> bool {
1589        a.abs_diff(b) < Duration::from_micros(100)
1590    }
1591
1592    #[test]
1593    fn exponential_backoff_without_jitter() {
1594        let mut backoff =
1595            ExponentialBackoff::new(Duration::from_millis(100), Duration::from_secs(1), 2.0, false);
1596
1597        assert!(approx_eq(backoff.next().unwrap(), Duration::from_millis(100)));
1598        assert!(approx_eq(backoff.next().unwrap(), Duration::from_millis(200)));
1599        assert!(approx_eq(backoff.next().unwrap(), Duration::from_millis(400)));
1600    }
1601
1602    #[test]
1603    fn exponential_backoff_with_jitter_applies_randomness() {
1604        let mut backoff =
1605            ExponentialBackoff::new(Duration::from_millis(100), Duration::from_secs(1), 2.0, true);
1606
1607        let delays: Vec<_> = (0..5).filter_map(|_| backoff.next()).collect();
1608
1609        // All delays should be positive
1610        for d in &delays {
1611            assert!(*d > Duration::ZERO);
1612        }
1613
1614        // With jitter, at least one delay should differ from exact values
1615        let exact_values = [
1616            Duration::from_millis(100),
1617            Duration::from_millis(200),
1618            Duration::from_millis(400),
1619            Duration::from_millis(800),
1620            Duration::from_secs(1),
1621        ];
1622
1623        let all_exact = delays
1624            .iter()
1625            .zip(exact_values.iter())
1626            .all(|(d, e)| approx_eq(*d, *e));
1627
1628        assert!(
1629            !all_exact,
1630            "With jitter enabled, delays should not all match exact exponential values"
1631        );
1632    }
1633
1634    #[tokio::test(start_paused = true)]
1635    async fn idle_timeout_returns_item_when_stream_has_data() {
1636        let mut stream = futures::stream::iter(vec![1, 2, 3]);
1637        let result = next_with_idle_timeout(&mut stream, Some(290)).await;
1638        assert_eq!(result, Some(1));
1639    }
1640
1641    #[tokio::test(start_paused = true)]
1642    async fn zero_timeout_does_not_trigger_idle_timeout() {
1643        let mut stream = futures::stream::once(async {
1644            tokio::time::sleep(Duration::from_secs(6)).await;
1645            42
1646        })
1647        .boxed();
1648
1649        let result = next_with_idle_timeout(&mut stream, Some(0)).await;
1650        assert_eq!(result, Some(42));
1651    }
1652
1653    #[tokio::test(start_paused = true)]
1654    async fn idle_timeout_returns_none_on_dead_connection() {
1655        let mut stream = futures::stream::pending::<i32>();
1656        // NB tokio auto-advances virtual time when the runtime is idle so next_with_idle_timeout resolves immediately
1657        let result = next_with_idle_timeout(&mut stream, Some(290)).await;
1658        assert_eq!(result, None);
1659    }
1660
1661    #[tokio::test(start_paused = true)]
1662    async fn idle_timeout_returns_none_when_stream_ends() {
1663        let mut stream = futures::stream::empty::<i32>();
1664        let result = next_with_idle_timeout(&mut stream, Some(290)).await;
1665        assert_eq!(result, None);
1666    }
1667}
kube_runtime/watcher.rs

kube_runtime/
watcher.rs
