kube_runtime/utils/watch_ext.rs
1use crate::{
2 utils::{
3 event_decode::EventDecode,
4 event_modify::EventModify,
5 predicate::{Config as PredicateConfig, Predicate, PredicateFilter},
6 stream_backoff::StreamBackoff,
7 },
8 watcher,
9};
10use kube_client::Resource;
11
12use crate::{
13 reflector::store::Writer,
14 utils::{Backoff, Reflect},
15};
16
17use crate::watcher::DefaultBackoff;
18use futures::{Stream, TryStream};
19
20/// Extension trait for streams returned by [`watcher`](watcher()) or [`reflector`](crate::reflector::reflector)
21pub trait WatchStreamExt: Stream {
22 /// Apply the [`DefaultBackoff`] watcher [`Backoff`] policy
23 ///
24 /// This is recommended for controllers that want to play nicely with the apiserver.
25 fn default_backoff(self) -> StreamBackoff<Self, DefaultBackoff>
26 where
27 Self: TryStream + Sized,
28 {
29 StreamBackoff::new(self, DefaultBackoff::default())
30 }
31
32 /// Apply a specific [`Backoff`] policy to a [`Stream`] using [`StreamBackoff`]
33 fn backoff<B>(self, b: B) -> StreamBackoff<Self, B>
34 where
35 B: Backoff,
36 Self: TryStream + Sized,
37 {
38 StreamBackoff::new(self, b)
39 }
40
41 /// Decode a [`watcher()`] stream into a stream of applied objects
42 ///
43 /// All Added/Modified events are passed through, and critical errors bubble up.
44 fn applied_objects<K>(self) -> EventDecode<Self>
45 where
46 Self: Stream<Item = Result<watcher::Event<K>, watcher::Error>> + Sized,
47 {
48 EventDecode::new(self, false)
49 }
50
51 /// Decode a [`watcher()`] stream into a stream of touched objects
52 ///
53 /// All Added/Modified/Deleted events are passed through, and critical errors bubble up.
54 /// Only the objects are retained; the original event type is discarded. An object
55 /// from a delete event need not have a deletion timestamp or a changed generation.
56 /// A subsequent [`predicate_filter`](Self::predicate_filter) can therefore suppress
57 /// it if the selected properties are unchanged. See [deletion handling](crate::predicates).
58 fn touched_objects<K>(self) -> EventDecode<Self>
59 where
60 Self: Stream<Item = Result<watcher::Event<K>, watcher::Error>> + Sized,
61 {
62 EventDecode::new(self, true)
63 }
64
65 /// Modify elements of a [`watcher()`] stream.
66 ///
67 /// Calls [`watcher::Event::modify()`] on every element.
68 /// Stream shorthand for `stream.map_ok(|event| { event.modify(f) })`.
69 ///
70 /// ```no_run
71 /// # use std::pin::pin;
72 /// # use futures::{Stream, StreamExt, TryStreamExt};
73 /// # use kube::{Api, Client, ResourceExt};
74 /// # use kube_runtime::{watcher, WatchStreamExt};
75 /// # use k8s_openapi::api::apps::v1::Deployment;
76 /// # async fn wrapper() -> Result<(), Box<dyn std::error::Error>> {
77 /// # let client: kube::Client = todo!();
78 /// let deploys: Api<Deployment> = Api::all(client);
79 /// let mut truncated_deploy_stream = pin!(watcher(deploys, watcher::Config::default())
80 /// .modify(|deploy| {
81 /// deploy.managed_fields_mut().clear();
82 /// deploy.status = None;
83 /// })
84 /// .applied_objects());
85 ///
86 /// while let Some(d) = truncated_deploy_stream.try_next().await? {
87 /// println!("Truncated Deployment: '{:?}'", serde_json::to_string(&d)?);
88 /// }
89 /// # Ok(())
90 /// # }
91 /// ```
92 fn modify<F, K>(self, f: F) -> EventModify<Self, F>
93 where
94 Self: Stream<Item = Result<watcher::Event<K>, watcher::Error>> + Sized,
95 F: FnMut(&mut K),
96 {
97 EventModify::new(self, f)
98 }
99
100 /// Filter a stream based on [`predicates`](crate::predicates).
101 ///
102 /// This will filter out repeat calls where the predicate returns the same result.
103 /// Common use case for this is to avoid repeat events for status updates
104 /// by filtering on [`predicates::generation`](crate::predicates::generation).
105 ///
106 /// The cache entries have a configurable time-to-live (TTL) to prevent unbounded
107 /// memory growth. By default, entries expire after 1 hour.
108 ///
109 /// ## Deletion handling
110 ///
111 /// This method filters decoded objects without access to their original event type.
112 /// After [`touched_objects`](Self::touched_objects), a generation-only predicate can
113 /// suppress objects from delete events because deletion need not change generation.
114 /// Use [`predicates::resource_version`](crate::predicates::resource_version) when
115 /// observed deletions should remain reconciliation triggers:
116 ///
117 /// ```no_run
118 /// # use kube::Api;
119 /// # use kube_runtime::{watcher, WatchStreamExt, predicates};
120 /// # use k8s_openapi::api::apps::v1::Deployment;
121 /// # fn example(api: Api<Deployment>) {
122 /// let stream = watcher(api, watcher::Config::default())
123 /// .default_backoff()
124 /// .touched_objects()
125 /// .predicate_filter(predicates::resource_version, Default::default());
126 /// # }
127 /// ```
128 ///
129 /// This also passes status-only changes and does not guarantee delivery of every
130 /// deletion. See the [predicate documentation](crate::predicates) for choosing a
131 /// predicate and handling cleanup with finalizers.
132 ///
133 /// ## Usage
134 /// ```no_run
135 /// # use std::pin::pin;
136 /// # use futures::{Stream, StreamExt, TryStreamExt};
137 /// use kube::{Api, Client, ResourceExt};
138 /// use kube_runtime::{watcher, WatchStreamExt, predicates};
139 /// use k8s_openapi::api::apps::v1::Deployment;
140 /// # async fn wrapper() -> Result<(), Box<dyn std::error::Error>> {
141 /// # let client: kube::Client = todo!();
142 /// let deploys: Api<Deployment> = Api::default_namespaced(client);
143 /// let mut changed_deploys = pin!(watcher(deploys, watcher::Config::default())
144 /// .applied_objects()
145 /// .predicate_filter(predicates::generation, Default::default()));
146 ///
147 /// while let Some(d) = changed_deploys.try_next().await? {
148 /// println!("saw Deployment '{} with hitherto unseen generation", d.name_any());
149 /// }
150 /// # Ok(())
151 /// # }
152 /// ```
153 fn predicate_filter<K, P>(self, predicate: P, config: PredicateConfig) -> PredicateFilter<Self, K, P>
154 where
155 Self: Stream<Item = Result<K, watcher::Error>> + Sized,
156 K: Resource + 'static,
157 P: Predicate<K> + 'static,
158 {
159 PredicateFilter::new(self, predicate, config)
160 }
161
162 /// Reflect a [`watcher()`] stream into a [`Store`] through a [`Writer`]
163 ///
164 /// Returns the stream unmodified, but passes every [`watcher::Event`] through a [`Writer`].
165 /// This populates a [`Store`] as the stream is polled.
166 ///
167 /// ## Usage
168 /// ```no_run
169 /// # use futures::{Stream, StreamExt, TryStreamExt};
170 /// # use std::time::Duration;
171 /// # use tracing::{info, warn};
172 /// use kube::{Api, Client, ResourceExt};
173 /// use kube_runtime::{watcher, WatchStreamExt, reflector};
174 /// use k8s_openapi::api::apps::v1::Deployment;
175 /// # async fn wrapper() -> Result<(), Box<dyn std::error::Error>> {
176 /// # let client: kube::Client = todo!();
177 ///
178 /// let deploys: Api<Deployment> = Api::default_namespaced(client);
179 /// let (reader, writer) = reflector::store::<Deployment>();
180 ///
181 /// tokio::spawn(async move {
182 /// // start polling the store once the reader is ready
183 /// reader.wait_until_ready().await.unwrap();
184 /// loop {
185 /// let names = reader.state().iter().map(|d| d.name_any()).collect::<Vec<_>>();
186 /// info!("Current {} deploys: {:?}", names.len(), names);
187 /// tokio::time::sleep(Duration::from_secs(10)).await;
188 /// }
189 /// });
190 ///
191 /// // configure the watcher stream and populate the store while polling
192 /// watcher(deploys, watcher::Config::default())
193 /// .reflect(writer)
194 /// .applied_objects()
195 /// .for_each(|res| async move {
196 /// match res {
197 /// Ok(o) => info!("saw {}", o.name_any()),
198 /// Err(e) => warn!("watcher error: {}", e),
199 /// }
200 /// })
201 /// .await;
202 ///
203 /// # Ok(())
204 /// # }
205 /// ```
206 ///
207 /// [`Store`]: crate::reflector::Store
208 fn reflect<K>(self, writer: Writer<K>) -> Reflect<Self, K>
209 where
210 Self: Stream<Item = watcher::Result<watcher::Event<K>>> + Sized,
211 K: Resource + Clone + 'static,
212 K::DynamicType: Eq + std::hash::Hash + Clone,
213 {
214 Reflect::new(self, writer)
215 }
216
217 /// Reflect a shared [`watcher()`] stream into a [`Store`] through a [`Writer`]
218 ///
219 /// Returns the stream unmodified, but passes every [`watcher::Event`]
220 /// through a [`Writer`]. This populates a [`Store`] as the stream is
221 /// polled. When the [`watcher::Event`] is not an error or a
222 /// [`watcher::Event::Deleted`] then its inner object will also be
223 /// propagated to subscribers.
224 ///
225 /// Subscribers can be created by calling [`subscribe()`] on a [`Writer`].
226 /// This will return a [`ReflectHandle`] stream that should be polled
227 /// independently. When the root stream is dropped, or it ends, all [`ReflectHandle`]s
228 /// subscribed to the stream will also terminate after all events yielded by
229 /// the root stream have been observed. This means [`ReflectHandle`] streams
230 /// can still be polled after the root stream has been dropped.
231 ///
232 /// ## Warning
233 ///
234 /// If the root [`Stream`] is not polled, [`ReflectHandle`] streams will
235 /// never receive any events. This will cause the streams to deadlock since
236 /// the root stream will apply backpressure when downstream readers are not
237 /// consuming events.
238 ///
239 ///
240 /// [`Store`]: crate::reflector::Store
241 /// [`subscribe()`]: crate::reflector::store::Writer::subscribe()
242 /// [`Stream`]: futures::stream::Stream
243 /// [`ReflectHandle`]: crate::reflector::dispatcher::ReflectHandle
244 /// ## Usage
245 /// ```no_run
246 /// # use futures::StreamExt;
247 /// # use std::time::Duration;
248 /// # use tracing::{info, warn};
249 /// use kube::{Api, ResourceExt};
250 /// use kube_runtime::{watcher, WatchStreamExt, reflector};
251 /// use k8s_openapi::api::apps::v1::Deployment;
252 /// # async fn wrapper() -> Result<(), Box<dyn std::error::Error>> {
253 /// # let client: kube::Client = todo!();
254 ///
255 /// let deploys: Api<Deployment> = Api::default_namespaced(client);
256 /// let subscriber_buf_sz = 100;
257 /// let (reader, writer) = reflector::store_shared::<Deployment>(subscriber_buf_sz);
258 /// let subscriber = writer.subscribe().unwrap();
259 ///
260 /// tokio::spawn(async move {
261 /// // start polling the store once the reader is ready
262 /// reader.wait_until_ready().await.unwrap();
263 /// loop {
264 /// let names = reader.state().iter().map(|d| d.name_any()).collect::<Vec<_>>();
265 /// info!("Current {} deploys: {:?}", names.len(), names);
266 /// tokio::time::sleep(Duration::from_secs(10)).await;
267 /// }
268 /// });
269 ///
270 /// tokio::spawn(async move {
271 /// // subscriber can be used to receive applied_objects
272 /// subscriber.for_each(|obj| async move {
273 /// info!("saw in subscriber {}", &obj.name_any())
274 /// }).await;
275 /// });
276 ///
277 /// // configure the watcher stream and populate the store while polling
278 /// watcher(deploys, watcher::Config::default())
279 /// .reflect_shared(writer)
280 /// .applied_objects()
281 /// .for_each(|res| async move {
282 /// match res {
283 /// Ok(o) => info!("saw in root stream {}", o.name_any()),
284 /// Err(e) => warn!("watcher error in root stream: {}", e),
285 /// }
286 /// })
287 /// .await;
288 ///
289 /// # Ok(())
290 /// # }
291 /// ```
292 #[cfg(feature = "unstable-runtime-subscribe")]
293 fn reflect_shared<K>(self, writer: Writer<K>) -> impl Stream<Item = Self::Item>
294 where
295 Self: Stream<Item = watcher::Result<watcher::Event<K>>> + Sized,
296 K: Resource + Clone + 'static,
297 K::DynamicType: Eq + std::hash::Hash + Clone,
298 {
299 crate::reflector(writer, self)
300 }
301}
302
303impl<St: ?Sized> WatchStreamExt for St where St: Stream {}
304
305// Compile tests
306#[cfg(test)]
307pub(crate) mod tests {
308 use super::watcher;
309 use crate::{WatchStreamExt as _, predicates};
310 use futures::prelude::*;
311 use k8s_openapi::api::core::v1::Pod;
312 use kube_client::{Api, Resource};
313
314 fn compile_type<T>() -> T {
315 unimplemented!("not called - compile test only")
316 }
317
318 pub fn assert_stream<T, K>(x: T) -> T
319 where
320 T: Stream<Item = watcher::Result<K>> + Send,
321 K: Resource + Clone + Send + 'static,
322 {
323 x
324 }
325
326 // not #[test] because this is only a compile check verification
327 #[allow(dead_code, unused_must_use)]
328 fn test_watcher_stream_type_drift() {
329 let pred_watch = watcher(compile_type::<Api<Pod>>(), Default::default())
330 .touched_objects()
331 .predicate_filter(predicates::generation, Default::default())
332 .boxed();
333 assert_stream(pred_watch);
334 }
335}