kube_runtime/utils/
predicate.rs

1use crate::watcher::Error;
2use core::{
3    pin::Pin,
4    task::{Context, Poll, ready},
5};
6use futures::Stream;
7use kube_client::{Resource, api::ObjectMeta};
8use pin_project::pin_project;
9use std::{
10    collections::{
11        HashMap,
12        hash_map::{DefaultHasher, Entry},
13    },
14    hash::{Hash, Hasher},
15    marker::PhantomData,
16    time::{Duration, Instant},
17};
18
19fn hash<T: Hash + ?Sized>(t: &T) -> u64 {
20    let mut hasher = DefaultHasher::new();
21    t.hash(&mut hasher);
22    hasher.finish()
23}
24
25/// Private cache key that includes UID in equality for predicate filtering
26#[derive(Debug, Clone, PartialEq, Eq, Hash)]
27struct PredicateCacheKey {
28    name: String,
29    namespace: Option<String>,
30    uid: Option<String>,
31}
32
33impl From<&ObjectMeta> for PredicateCacheKey {
34    fn from(meta: &ObjectMeta) -> Self {
35        Self {
36            name: meta.name.clone().unwrap_or_default(),
37            namespace: meta.namespace.clone(),
38            uid: meta.uid.clone(),
39        }
40    }
41}
42
43/// A predicate is a hasher of Kubernetes objects stream filtering
44pub trait Predicate<K> {
45    /// A predicate only needs to implement optional hashing when keys exist
46    fn hash_property(&self, obj: &K) -> Option<u64>;
47
48    /// Returns a `Predicate` that falls back to an alternate property if the first does not exist
49    ///
50    /// The alternate predicate is only used when the first returns `None`, not when its
51    /// hash is unchanged. In particular, `generation.fallback(resource_version)` can still
52    /// suppress deletions for objects with an unchanged generation.
53    ///
54    /// Generation support generally depends on the resource type: Deployments and custom
55    /// resources normally populate it, while `ConfigMaps` and Secrets normally do not.
56    /// Thus, `generation.fallback(resource_version)` is useful as a shared policy across
57    /// resource types, but the fallback is redundant for a type known to always provide
58    /// generation. Generation is not normally a field that appears and disappears over
59    /// an object's lifetime, though support can vary by API implementation and version.
60    ///
61    /// Persisted objects returned by ordinary API-server list/watch requests have a
62    /// resource version, so reversing the order to `resource_version.fallback(generation)`
63    /// generally makes the generation fallback redundant. Locally constructed objects
64    /// and non-persisted API responses need not have a resource version.
65    ///
66    /// # Usage
67    ///
68    /// ```
69    /// # use k8s_openapi::api::core::v1::Pod;
70    /// use kube::runtime::{predicates, Predicate};
71    /// # fn blah<K>(a: impl Predicate<K>) {}
72    /// let pred = predicates::generation.fallback(predicates::resource_version);
73    /// blah::<Pod>(pred);
74    /// ```
75    fn fallback<F: Predicate<K>>(self, f: F) -> Fallback<Self, F>
76    where
77        Self: Sized,
78    {
79        Fallback(self, f)
80    }
81
82    /// Returns a `Predicate` that combines all available hashes
83    ///
84    /// # Usage
85    ///
86    /// ```
87    /// # use k8s_openapi::api::core::v1::Pod;
88    /// use kube::runtime::{predicates, Predicate};
89    /// # fn blah<K>(a: impl Predicate<K>) {}
90    /// let pred = predicates::labels.combine(predicates::annotations);
91    /// blah::<Pod>(pred);
92    /// ```
93    fn combine<F: Predicate<K>>(self, f: F) -> Combine<Self, F>
94    where
95        Self: Sized,
96    {
97        Combine(self, f)
98    }
99}
100
101impl<K, F: Fn(&K) -> Option<u64>> Predicate<K> for F {
102    fn hash_property(&self, obj: &K) -> Option<u64> {
103        (self)(obj)
104    }
105}
106
107/// See [`Predicate::fallback`]
108#[derive(Copy, Clone, Debug, PartialEq, Eq)]
109pub struct Fallback<A, B>(pub(super) A, pub(super) B);
110impl<A, B, K> Predicate<K> for Fallback<A, B>
111where
112    A: Predicate<K>,
113    B: Predicate<K>,
114{
115    fn hash_property(&self, obj: &K) -> Option<u64> {
116        self.0.hash_property(obj).or_else(|| self.1.hash_property(obj))
117    }
118}
119/// See [`Predicate::combine`]
120#[derive(Copy, Clone, Debug, PartialEq, Eq)]
121pub struct Combine<A, B>(pub(super) A, pub(super) B);
122impl<A, B, K> Predicate<K> for Combine<A, B>
123where
124    A: Predicate<K>,
125    B: Predicate<K>,
126{
127    fn hash_property(&self, obj: &K) -> Option<u64> {
128        match (self.0.hash_property(obj), self.1.hash_property(obj)) {
129            // pass on both missing properties so people can chain .fallback
130            (None, None) => None,
131            // but any other combination of properties are hashed together
132            (a, b) => Some(hash(&(a, b))),
133        }
134    }
135}
136
137/// Configuration for predicate filtering with cache TTL
138#[derive(Debug, Clone)]
139pub struct Config {
140    /// Time-to-live for cache entries
141    ///
142    /// Entries not seen for at least this long will be evicted from the cache.
143    /// Default is 1 hour.
144    ttl: Duration,
145}
146
147impl Config {
148    /// Set the time-to-live for cache entries
149    ///
150    /// Entries not seen for at least this long will be evicted from the cache.
151    #[must_use]
152    pub fn ttl(mut self, ttl: Duration) -> Self {
153        self.ttl = ttl;
154        self
155    }
156}
157
158impl Default for Config {
159    fn default() -> Self {
160        Self {
161            // Default to 1 hour TTL - long enough to avoid unnecessary reconciles
162            // but short enough to prevent unbounded memory growth
163            ttl: Duration::from_secs(3600),
164        }
165    }
166}
167
168/// Cache entry storing predicate hash and last access time
169#[derive(Debug, Clone)]
170struct CacheEntry {
171    hash: u64,
172    last_seen: Instant,
173}
174
175#[pin_project]
176/// Stream returned by the [`predicate_filter`](super::WatchStreamExt::predicate_filter) method.
177#[must_use = "streams do nothing unless polled"]
178pub struct PredicateFilter<St, K: Resource, P: Predicate<K>> {
179    #[pin]
180    stream: St,
181    predicate: P,
182    cache: HashMap<PredicateCacheKey, CacheEntry>,
183    config: Config,
184    last_evicted: Instant,
185    // K: Resource necessary to get .meta() of an object during polling
186    _phantom: PhantomData<K>,
187}
188impl<St, K, P> PredicateFilter<St, K, P>
189where
190    St: Stream<Item = Result<K, Error>>,
191    K: Resource,
192    P: Predicate<K>,
193{
194    pub(super) fn new(stream: St, predicate: P, config: Config) -> Self {
195        Self {
196            stream,
197            predicate,
198            cache: HashMap::new(),
199            config,
200            last_evicted: Instant::now(),
201            _phantom: PhantomData,
202        }
203    }
204}
205impl<St, K, P> Stream for PredicateFilter<St, K, P>
206where
207    St: Stream<Item = Result<K, Error>>,
208    K: Resource,
209    P: Predicate<K>,
210{
211    type Item = Result<K, Error>;
212
213    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
214        let mut me = self.project();
215
216        let now = Instant::now();
217        let ttl = me.config.ttl;
218        // Reclaim expired entries periodically instead of on every poll. Per-entry
219        // freshness is enforced below, so this scan is only memory hygiene and does
220        // not affect which events are emitted.
221        if now.duration_since(*me.last_evicted) >= ttl {
222            me.cache
223                .retain(|_, entry| now.duration_since(entry.last_seen) < ttl);
224            *me.last_evicted = now;
225        }
226
227        Poll::Ready(loop {
228            break match ready!(me.stream.as_mut().poll_next(cx)) {
229                Some(Ok(obj)) => {
230                    if let Some(val) = me.predicate.hash_property(&obj) {
231                        let key = PredicateCacheKey::from(obj.meta());
232
233                        // Upsert in a single lookup. An entry older than the TTL is
234                        // treated as absent, matching the evict-before-every-poll
235                        // semantics now that eviction is amortized.
236                        let changed = match me.cache.entry(key) {
237                            Entry::Occupied(mut e) => {
238                                let stale = now.duration_since(e.get().last_seen) >= ttl;
239                                let changed = stale || e.get().hash != val;
240                                e.insert(CacheEntry {
241                                    hash: val,
242                                    last_seen: now,
243                                });
244                                changed
245                            }
246                            Entry::Vacant(e) => {
247                                e.insert(CacheEntry {
248                                    hash: val,
249                                    last_seen: now,
250                                });
251                                true
252                            }
253                        };
254
255                        if changed {
256                            Some(Ok(obj))
257                        } else {
258                            continue;
259                        }
260                    } else {
261                        // if we can't evaluate predicate, always emit K
262                        Some(Ok(obj))
263                    }
264                }
265                Some(Err(err)) => Some(Err(err)),
266                None => return Poll::Ready(None),
267            };
268        })
269    }
270}
271
272/// Predicate functions for [`WatchStreamExt::predicate_filter`](crate::WatchStreamExt::predicate_filter)
273///
274/// These functions just return a hash of commonly compared values,
275/// to help decide whether to pass a watch event along or not.
276///
277/// # Choosing a predicate
278///
279/// | Predicate | Use case | Deletion behavior |
280/// | --- | --- | --- |
281/// | [`generation`] | Observe desired-state changes while ignoring status-only updates, for resources supporting generation | Can suppress deletion-related updates and deleted objects when generation is unchanged |
282/// | [`resource_version`] | Observe object revisions, including status-only changes | Retains observed deletes whose resource version differs from the cached version |
283/// | [`labels`], [`annotations`], [`finalizers`] | Observe changes to those fields | Can suppress deletions when the selected fields are unchanged |
284/// | [`fallback`](crate::Predicate::fallback) | Use another predicate when the first has no value | See [`Predicate::fallback`](crate::Predicate::fallback) for the deletion caveat |
285///
286/// # Deletion handling
287///
288/// Predicates operate on decoded objects (versus Events).
289/// After [`touched_objects`](crate::WatchStreamExt::touched_objects), an object from a
290/// delete event is filtered using the same rules as any other object. Deletion does not
291/// necessarily change `metadata.generation` or populate `metadata.deletionTimestamp`,
292/// so predicates based on unchanged fields can suppress that object.
293///
294/// Use [`resource_version`] when observed deletions should trigger reconciliation, for
295/// example to recreate a missing owned resource. The Kubernetes apiserver stamps the
296/// delete revision onto the deleted object. This predicate also passes status-only
297/// changes, so it does not provide the same filtering as [`generation`].
298///
299/// No predicate guarantees delivery of every deletion: events can be missed while a
300/// watcher is unavailable. Reconcilers should check current desired and actual state,
301/// and use a recovery mechanism such as periodic requeues when missed triggers matter.
302/// Use [finalizers](crate::finalizer::finalizer) for cleanup that must complete before
303/// deletion. The initial deletion marking bumps a positive generation, so a generation-only
304/// predicate lets that update through to start cleanup. It can still suppress the final
305/// delete event and the update that adds the finalizer, since adding a finalizer leaves
306/// generation unchanged. See the [finalizer documentation](crate::finalizer::finalizer).
307///
308/// Functional rewrite of the [controller-runtime/predicate module](https://github.com/kubernetes-sigs/controller-runtime/blob/main/pkg/predicate/predicate.go).
309pub mod predicates {
310    use super::hash;
311    use kube_client::{Resource, ResourceExt};
312
313    /// Hash the generation of a Resource K
314    ///
315    /// Useful for ignoring status-only updates on resources that support generation.
316    ///
317    /// Note that the final delete event need not change generation, so this can suppress
318    /// objects from delete events. See the [module documentation](crate::predicates).
319    pub fn generation<K: Resource>(obj: &K) -> Option<u64> {
320        obj.meta().generation.map(|g| hash(&g))
321    }
322
323    /// Hash the resource version of a Resource K
324    ///
325    /// Use this when observed deletions should trigger reconciliation: the Kubernetes
326    /// apiserver stamps the delete revision onto the deleted object. Status-only changes
327    /// also change resource version and pass this filter. This does not guarantee that
328    /// every deletion is observed; see the [module documentation](crate::predicates).
329    pub fn resource_version<K: Resource>(obj: &K) -> Option<u64> {
330        obj.meta().resource_version.as_ref().map(hash)
331    }
332
333    /// Hash the labels of a Resource K
334    pub fn labels<K: Resource>(obj: &K) -> Option<u64> {
335        Some(hash(obj.labels()))
336    }
337
338    /// Hash the annotations of a Resource K
339    pub fn annotations<K: Resource>(obj: &K) -> Option<u64> {
340        Some(hash(obj.annotations()))
341    }
342
343    /// Hash the finalizers of a Resource K
344    pub fn finalizers<K: Resource>(obj: &K) -> Option<u64> {
345        Some(hash(obj.finalizers()))
346    }
347}
348
349#[cfg(test)]
350pub(crate) mod tests {
351    use std::{pin::pin, task::Poll};
352
353    use super::{Config, Error, PredicateFilter, predicates};
354    use futures::{FutureExt, StreamExt, poll, stream};
355    use kube_client::Resource;
356    use serde_json::json;
357
358    #[tokio::test]
359    async fn predicate_filtering_hides_equal_predicate_values() {
360        use k8s_openapi::api::core::v1::Pod;
361        let mkobj = |g: i32| {
362            let p: Pod = serde_json::from_value(json!({
363                "apiVersion": "v1",
364                "kind": "Pod",
365                "metadata": {
366                    "name": "blog",
367                    "generation": Some(g),
368                },
369                "spec": {
370                    "containers": [{
371                      "name": "blog",
372                      "image": "clux/blog:0.1.0"
373                    }],
374                }
375            }))
376            .unwrap();
377            p
378        };
379        let data = stream::iter([
380            Ok(mkobj(1)),
381            Err(Error::NoResourceVersion),
382            Ok(mkobj(1)),
383            Ok(mkobj(2)),
384        ]);
385        let mut rx = pin!(PredicateFilter::new(
386            data,
387            predicates::generation,
388            Config::default()
389        ));
390
391        // mkobj(1) passed through
392        let first = rx.next().now_or_never().unwrap().unwrap().unwrap();
393        assert_eq!(first.meta().generation, Some(1));
394
395        // Error passed through
396        assert!(matches!(
397            poll!(rx.next()),
398            Poll::Ready(Some(Err(Error::NoResourceVersion)))
399        ));
400        // (no repeat mkobj(1) - same generation)
401        // mkobj(2) next
402        let second = rx.next().now_or_never().unwrap().unwrap().unwrap();
403        assert_eq!(second.meta().generation, Some(2));
404        assert!(matches!(poll!(rx.next()), Poll::Ready(None)));
405    }
406
407    #[tokio::test]
408    async fn predicate_filtering_should_handle_recreated_resources_with_same_generation() {
409        use k8s_openapi::api::core::v1::Pod;
410
411        let mkobj = |g: i32, uid: &str| {
412            let p: Pod = serde_json::from_value(json!({
413                "apiVersion": "v1",
414                "kind": "Pod",
415                "metadata": {
416                    "name": "blog",
417                    "namespace": "default",
418                    "generation": Some(g),
419                    "uid": uid,
420                },
421                "spec": {
422                    "containers": [{
423                      "name": "blog",
424                      "image": "clux/blog:0.1.0"
425                    }],
426                }
427            }))
428            .unwrap();
429            p
430        };
431
432        // Simulate: create (gen=1, uid=1) -> update (gen=1, uid=1) -> delete ->
433        // create (gen=1, uid=2) -> delete -> create (gen=2, uid=3)
434        let data = stream::iter([
435            Ok(mkobj(1, "uid-1")),
436            Ok(mkobj(1, "uid-1")),
437            Ok(mkobj(1, "uid-2")),
438            Ok(mkobj(2, "uid-3")),
439        ]);
440        let mut rx = pin!(PredicateFilter::new(
441            data,
442            predicates::generation,
443            Config::default()
444        ));
445
446        // mkobj(1, uid-1) passed through
447        let first = rx.next().now_or_never().unwrap().unwrap().unwrap();
448        assert_eq!(first.meta().generation, Some(1));
449        assert_eq!(first.meta().uid.as_deref(), Some("uid-1"));
450
451        // (no repeat mkobj(1, uid-1) - same UID and generation)
452        // mkobj(1, uid-2) next - different UID detected
453        let second = rx.next().now_or_never().unwrap().unwrap().unwrap();
454        assert_eq!(second.meta().generation, Some(1));
455        assert_eq!(second.meta().uid.as_deref(), Some("uid-2"));
456
457        // mkobj(2, uid-3) next
458        let third = rx.next().now_or_never().unwrap().unwrap().unwrap();
459        assert_eq!(third.meta().generation, Some(2));
460        assert_eq!(third.meta().uid.as_deref(), Some("uid-3"));
461
462        assert!(matches!(poll!(rx.next()), Poll::Ready(None)));
463    }
464
465    #[tokio::test]
466    async fn predicate_cache_ttl_evicts_expired_entries() {
467        use futures::{SinkExt, channel::mpsc};
468        use k8s_openapi::api::core::v1::Pod;
469        use std::time::Duration;
470
471        let mkobj = |g: i32, uid: &str| {
472            let p: Pod = serde_json::from_value(json!({
473                "apiVersion": "v1",
474                "kind": "Pod",
475                "metadata": {
476                    "name": "blog",
477                    "namespace": "default",
478                    "generation": Some(g),
479                    "uid": uid,
480                },
481                "spec": {
482                    "containers": [{
483                      "name": "blog",
484                      "image": "clux/blog:0.1.0"
485                    }],
486                }
487            }))
488            .unwrap();
489            p
490        };
491
492        // Use a very short TTL for testing
493        let config = Config::default().ttl(Duration::from_millis(50));
494
495        // Use a channel so we can send items with delays
496        let (mut tx, rx) = mpsc::unbounded();
497        let mut filtered = pin!(PredicateFilter::new(
498            rx.map(Ok::<_, Error>),
499            predicates::generation,
500            config
501        ));
502
503        // Send first object
504        tx.send(mkobj(1, "uid-1")).await.unwrap();
505        let first = filtered.next().now_or_never().unwrap().unwrap().unwrap();
506        assert_eq!(first.meta().generation, Some(1));
507
508        // Send same object immediately - should be filtered
509        tx.send(mkobj(1, "uid-1")).await.unwrap();
510        assert!(matches!(poll!(filtered.next()), Poll::Pending));
511
512        // Wait for TTL to expire
513        tokio::time::sleep(Duration::from_millis(100)).await;
514
515        // Send same object after TTL - should pass through due to eviction
516        tx.send(mkobj(1, "uid-1")).await.unwrap();
517        let second = filtered.next().now_or_never().unwrap().unwrap().unwrap();
518        assert_eq!(second.meta().generation, Some(1));
519    }
520
521    #[tokio::test]
522    async fn predicate_filter_compiles_with_dynamic_resource() {
523        use kube::api::DynamicObject;
524
525        let obj: DynamicObject = serde_json::from_value(json!({
526            "apiVersion": "v1",
527            "kind": "Pod",
528            "metadata": {
529                "name": "blog",
530                "generation": Some(1),
531            },
532        }))
533        .unwrap();
534        let data = stream::iter([Ok(obj)]);
535        let mut rx = pin!(PredicateFilter::new(
536            data,
537            predicates::generation,
538            Config::default()
539        ));
540
541        let first = rx.next().now_or_never().unwrap().unwrap().unwrap();
542        assert_eq!(first.meta().generation, Some(1));
543
544        // Mostly, this check is ensuring that predicate filters compile with
545        // `DynamicObject`, not testing any behaviour.
546    }
547}
kube_runtime/utils/predicate.rs

kube_runtime/utils/
predicate.rs
