Add WatchStreamExt::default_backoff shorthand

examples/node_watcher.rs

@@ -1,4 +1,3 @@
-use backoff::ExponentialBackoff;
 use futures::{pin_mut, TryStreamExt};
 use k8s_openapi::api::core::v1::{Event, Node};
 use kube::{
@@ -16,9 +15,7 @@ async fn main() -> anyhow::Result<()> {
     let nodes: Api<Node> = Api::all(client.clone());
 
     let wc = watcher::Config::default().labels("beta.kubernetes.io/arch=amd64");
-    let obs = watcher(nodes, wc)
-        .backoff(ExponentialBackoff::default())
-        .applied_objects();
+    let obs = watcher(nodes, wc).default_backoff().applied_objects();
 
     pin_mut!(obs);
     while let Some(n) = obs.try_next().await? {

examples/pod_watcher.rs

@@ -15,6 +15,7 @@ async fn main() -> anyhow::Result<()> {
 
     watcher(api, watcher::Config::default())
         .applied_objects()
+        .default_backoff()
         .try_for_each(|p| async move {
             info!("saw {}", p.name_any());
             if let Some(unready_reason) = pod_unready(&p) {

kube-runtime/src/controller/mod.rs

@@ -9,7 +9,7 @@ use crate::{
     },
     scheduler::{scheduler, ScheduleRequest},
     utils::{trystream_try_via, CancelableJoinHandle, KubeRuntimeStreamExt, StreamBackoff, WatchStreamExt},
-    watcher::{self, watcher, Config},
+    watcher::{self, watcher, Config, DefaultBackoff},
 };
 use backoff::backoff::Backoff;
 use derivative::Derivative;
@@ -541,7 +541,7 @@ where
         trigger_selector.push(self_watcher);
         Self {
             trigger_selector,
-            trigger_backoff: Box::new(watcher::default_backoff()),
+            trigger_backoff: Box::<DefaultBackoff>::default(),
             graceful_shutdown_selector: vec![
                 // Fallback future, ensuring that we never terminate if no additional futures are added to the selector
                 future::pending().boxed(),
@@ -624,7 +624,7 @@ where
         trigger_selector.push(self_watcher);
         Self {
             trigger_selector,
-            trigger_backoff: Box::new(watcher::default_backoff()),
+            trigger_backoff: Box::<DefaultBackoff>::default(),
             graceful_shutdown_selector: vec![
                 // Fallback future, ensuring that we never terminate if no additional futures are added to the selector
                 future::pending().boxed(),
@@ -710,14 +710,14 @@ where
     /// # use k8s_openapi::api::core::v1::ConfigMap;
     /// # use k8s_openapi::api::apps::v1::StatefulSet;
     /// # use kube::runtime::controller::Action;
-    /// # use kube::runtime::{predicates, watcher, Controller, WatchStreamExt};
+    /// # use kube::runtime::{predicates, metadata_watcher, watcher, Controller, WatchStreamExt};
     /// # use kube::{Api, Client, Error, ResourceExt};
     /// # use std::sync::Arc;
     /// # type CustomResource = ConfigMap;
     /// # async fn reconcile(_: Arc<CustomResource>, _: Arc<()>) -> Result<Action, Error> { Ok(Action::await_change()) }
     /// # fn error_policy(_: Arc<CustomResource>, _: &kube::Error, _: Arc<()>) -> Action { Action::await_change() }
     /// # async fn doc(client: kube::Client) {
-    /// let sts_stream = watcher(Api::<StatefulSet>::all(client.clone()), watcher::Config::default())
+    /// let sts_stream = metadata_watcher(Api::<StatefulSet>::all(client.clone()), watcher::Config::default())
     ///     .touched_objects()
     ///     .predicate_filter(predicates::generation);
     ///

kube-runtime/src/utils/watch_ext.rs

@@ -8,12 +8,23 @@ use crate::{
 };
 #[cfg(feature = "unstable-runtime-predicates")] use kube_client::Resource;
 
+use crate::watcher::DefaultBackoff;
 use backoff::backoff::Backoff;
 use futures::{Stream, TryStream};
 
 /// Extension trait for streams returned by [`watcher`](watcher()) or [`reflector`](crate::reflector::reflector)
 pub trait WatchStreamExt: Stream {
-    /// Apply a [`Backoff`] policy to a [`Stream`] using [`StreamBackoff`]
+    /// Apply the [`DefaultBackoff`] watcher [`Backoff`] policy
+    ///
+    /// This is recommended for controllers that want to play nicely with the apiserver.
+    fn default_backoff(self) -> StreamBackoff<Self, DefaultBackoff>
+    where
+        Self: TryStream + Sized,
+    {
+        StreamBackoff::new(self, DefaultBackoff::default())
+    }
+
+    /// Apply a specific [`Backoff`] policy to a [`Stream`] using [`StreamBackoff`]
     fn backoff<B>(self, b: B) -> StreamBackoff<Self, B>
     where
         B: Backoff,

kube-runtime/src/watcher.rs

@@ -640,19 +640,51 @@ pub fn watch_object<K: Resource + Clone + DeserializeOwned + Debug + Send + 'sta
 
 /// Default watch [`Backoff`] inspired by Kubernetes' client-go.
 ///
-/// Note that the exact parameters used herein should not be considered stable.
-/// The parameters currently optimize for being kind to struggling apiservers.
-/// See [client-go's reflector source](/~https://github.com/kubernetes/client-go/blob/980663e185ab6fc79163b1c2565034f6d58368db/tools/cache/reflector.go#L177-L181)
-/// for more details.
+/// This fn has been moved into [`DefaultBackoff`].
 #[must_use]
-pub fn default_backoff() -> impl Backoff + Send + Sync {
-    let expo = backoff::ExponentialBackoff {
-        initial_interval: Duration::from_millis(800),
-        max_interval: Duration::from_secs(30),
-        randomization_factor: 1.0,
-        multiplier: 2.0,
-        max_elapsed_time: None,
-        ..ExponentialBackoff::default()
-    };
-    ResetTimerBackoff::new(expo, Duration::from_secs(120))
+#[deprecated(
+    since = "0.84.0",
+    note = "replaced by `watcher::DefaultBackoff`. This fn will be removed in 0.88.0."
+)]
+pub fn default_backoff() -> DefaultBackoff {
+    DefaultBackoff::default()
+}
+
+/// Default watcher backoff inspired by Kubernetes' client-go.
+///
+/// The parameters currently optimize for being kind to struggling apiservers.
+/// The exact parameters are taken from
+/// [client-go's reflector source](/~https://github.com/kubernetes/client-go/blob/980663e185ab6fc79163b1c2565034f6d58368db/tools/cache/reflector.go#L177-L181)
+/// and should not be considered stable.
+///
+/// This struct implements [`Backoff`] and is the default strategy used
+/// when calling `WatchStreamExt::default_backoff`. If you need to create
+/// this manually then [`DefaultBackoff::default`] can be used.
+pub struct DefaultBackoff(Strategy);
+type Strategy = ResetTimerBackoff<ExponentialBackoff>;
+
+impl Default for DefaultBackoff {
+    fn default() -> Self {
+        Self(ResetTimerBackoff::new(
+            backoff::ExponentialBackoff {
+                initial_interval: Duration::from_millis(800),
+                max_interval: Duration::from_secs(30),
+                randomization_factor: 1.0,
+                multiplier: 2.0,
+                max_elapsed_time: None,
+                ..ExponentialBackoff::default()
+            },
+            Duration::from_secs(120),
+        ))
+    }
+}
+
+impl Backoff for DefaultBackoff {
+    fn next_backoff(&mut self) -> Option<Duration> {
+        self.0.next_backoff()
+    }
+
+    fn reset(&mut self) {
+        self.0.reset()
+    }
 }