Align watcher::Event init/page variants (#1504)

* Align `watcher::Event` variants for initialisation **Merges - `Event::InitPage` and `Event::InitApply`** This has several consequences; - buffering of pages done in watcher (need to benchmark this since it is likely worse memory wise) - no buffering needed in any flatteners (further simplification can be done later) Signed-off-by: clux <[email protected]> * get tests to compile Signed-off-by: clux <[email protected]> * fix tests Signed-off-by: clux <[email protected]> * Event::into_iter helpers are now unnecessary (and unused) Signed-off-by: clux <[email protected]> * avoid reversing event list from pages Signed-off-by: clux <[email protected]> * can kill smallvec now Signed-off-by: clux <[email protected]> * docs for the enum were bad, update after proofread Signed-off-by: clux <[email protected]> * typing tweaks spotted from codereview and two other minor ones Signed-off-by: clux <[email protected]> --------- Signed-off-by: clux <[email protected]>
kube-rs · Jun 4, 2024 · 6ce3978 · 6ce3978
1 parent de3fe1e
commit 6ce3978
Show file tree

Hide file tree

Showing 11 changed files with 113 additions and 108 deletions.
diff --git a/Cargo.toml b/Cargo.toml
@@ -74,7 +74,6 @@ secrecy = "0.8.0"
 serde = "1.0.130"
 serde_json = "1.0.68"
 serde_yaml = "0.9.19"
-smallvec = "1.7.0"
 syn = "2.0.38"
 tame-oauth = "0.10.0"
 tempfile = "3.1.0"

diff --git a/kube-runtime/Cargo.toml b/kube-runtime/Cargo.toml
@@ -34,7 +34,6 @@ futures = { workspace = true, features = ["async-await"] }
 kube-client = { path = "../kube-client", version = "=0.91.0", default-features = false, features = ["jsonpatch", "client"] }
 derivative.workspace = true
 serde.workspace = true
-smallvec.workspace = true
 ahash.workspace = true
 parking_lot.workspace = true
 pin-project.workspace = true

diff --git a/kube-runtime/src/reflector/dispatcher.rs b/kube-runtime/src/reflector/dispatcher.rs
@@ -148,7 +148,7 @@ pub(crate) mod test {
         watcher::{Error, Event},
         WatchStreamExt,
     };
-    use std::{sync::Arc, task::Poll, vec};
+    use std::{sync::Arc, task::Poll};
 
     use crate::reflector;
     use futures::{pin_mut, poll, stream, StreamExt};
@@ -166,9 +166,10 @@ pub(crate) mod test {
         let bar = testpod("bar");
         let st = stream::iter([
             Ok(Event::Apply(foo.clone())),
-            Err(Error::TooManyObjects),
+            Err(Error::NoResourceVersion),
             Ok(Event::Init),
-            Ok(Event::InitPage(vec![foo, bar])),
+            Ok(Event::InitApply(foo)),
+            Ok(Event::InitApply(bar)),
             Ok(Event::InitDone),
         ]);
 
@@ -187,7 +188,7 @@ pub(crate) mod test {
         assert_eq!(reader.len(), 1);
         assert!(matches!(
             poll!(reflect.next()),
-            Poll::Ready(Some(Err(Error::TooManyObjects)))
+            Poll::Ready(Some(Err(Error::NoResourceVersion)))
         ));
         assert_eq!(reader.len(), 1);
 
@@ -196,7 +197,11 @@ pub(crate) mod test {
         assert_eq!(reader.len(), 1);
 
         let restarted = poll!(reflect.next());
-        assert!(matches!(restarted, Poll::Ready(Some(Ok(Event::InitPage(_))))));
+        assert!(matches!(restarted, Poll::Ready(Some(Ok(Event::InitApply(_))))));
+        assert_eq!(reader.len(), 1);
+
+        let restarted = poll!(reflect.next());
+        assert!(matches!(restarted, Poll::Ready(Some(Ok(Event::InitApply(_))))));
         assert_eq!(reader.len(), 1);
 
         let restarted = poll!(reflect.next());
@@ -218,9 +223,10 @@ pub(crate) mod test {
         let st = stream::iter([
             Ok(Event::Delete(foo.clone())),
             Ok(Event::Apply(foo.clone())),
-            Err(Error::TooManyObjects),
+            Err(Error::NoResourceVersion),
             Ok(Event::Init),
-            Ok(Event::InitPage(vec![foo.clone(), bar.clone()])),
+            Ok(Event::InitApply(foo.clone())),
+            Ok(Event::InitApply(bar.clone())),
             Ok(Event::InitDone),
         ]);
 
@@ -249,7 +255,7 @@ pub(crate) mod test {
         // Errors are not propagated to subscribers.
         assert!(matches!(
             poll!(reflect.next()),
-            Poll::Ready(Some(Err(Error::TooManyObjects)))
+            Poll::Ready(Some(Err(Error::NoResourceVersion)))
         ));
         assert!(matches!(poll!(subscriber.next()), Poll::Pending));
 
@@ -262,7 +268,11 @@ pub(crate) mod test {
 
         assert!(matches!(
             poll!(reflect.next()),
-            Poll::Ready(Some(Ok(Event::InitPage(_))))
+            Poll::Ready(Some(Ok(Event::InitApply(_))))
+        ));
+        assert!(matches!(
+            poll!(reflect.next()),
+            Poll::Ready(Some(Ok(Event::InitApply(_))))
         ));
 
         assert!(matches!(
@@ -288,7 +298,8 @@ pub(crate) mod test {
         let st = stream::iter([
             Ok(Event::Apply(foo.clone())),
             Ok(Event::Init),
-            Ok(Event::InitPage(vec![foo.clone(), bar.clone()])),
+            Ok(Event::InitApply(foo.clone())),
+            Ok(Event::InitApply(bar.clone())),
             Ok(Event::InitDone),
         ]);
 
@@ -320,7 +331,13 @@ pub(crate) mod test {
 
         assert!(matches!(
             poll!(reflect.next()),
-            Poll::Ready(Some(Ok(Event::InitPage(_))))
+            Poll::Ready(Some(Ok(Event::InitApply(_))))
+        ));
+        assert_eq!(poll!(subscriber.next()), Poll::Pending);
+
+        assert!(matches!(
+            poll!(reflect.next()),
+            Poll::Ready(Some(Ok(Event::InitApply(_))))
         ));
         assert_eq!(poll!(subscriber.next()), Poll::Pending);
 

diff --git a/kube-runtime/src/reflector/mod.rs b/kube-runtime/src/reflector/mod.rs
@@ -243,7 +243,7 @@ mod tests {
             stream::iter(vec![
                 Ok(watcher::Event::Apply(cm_a.clone())),
                 Ok(watcher::Event::Init),
-                Ok(watcher::Event::InitPage(vec![cm_b.clone()])),
+                Ok(watcher::Event::InitApply(cm_b.clone())),
                 Ok(watcher::Event::InitDone),
             ]),
         )

diff --git a/kube-runtime/src/reflector/store.rs b/kube-runtime/src/reflector/store.rs
@@ -111,12 +111,6 @@ where
             watcher::Event::Init => {
                 self.buffer = AHashMap::new();
             }
-            watcher::Event::InitPage(new_objs) => {
-                let new_objs = new_objs
-                    .iter()
-                    .map(|obj| (obj.to_object_ref(self.dyntype.clone()), Arc::new(obj.clone())));
-                self.buffer.extend(new_objs);
-            }
             watcher::Event::InitApply(obj) => {
                 let key = obj.to_object_ref(self.dyntype.clone());
                 let obj = Arc::new(obj.clone());

diff --git a/kube-runtime/src/utils/event_flatten.rs b/kube-runtime/src/utils/event_flatten.rs
@@ -9,22 +9,17 @@ use pin_project::pin_project;
 #[pin_project]
 /// Stream returned by the [`applied_objects`](super::WatchStreamExt::applied_objects) and [`touched_objects`](super::WatchStreamExt::touched_objects) method.
 #[must_use = "streams do nothing unless polled"]
-pub struct EventFlatten<St, K> {
+pub struct EventFlatten<St> {
     #[pin]
     stream: St,
     emit_deleted: bool,
-    queue: std::vec::IntoIter<K>,
 }
-impl<St: TryStream<Ok = Event<K>>, K> EventFlatten<St, K> {
+impl<St: TryStream<Ok = Event<K>>, K> EventFlatten<St> {
     pub(super) fn new(stream: St, emit_deleted: bool) -> Self {
-        Self {
-            stream,
-            queue: vec![].into_iter(),
-            emit_deleted,
-        }
+        Self { stream, emit_deleted }
     }
 }
-impl<St, K> Stream for EventFlatten<St, K>
+impl<St, K> Stream for EventFlatten<St>
 where
     St: Stream<Item = Result<Event<K>, Error>>,
 {
@@ -33,9 +28,6 @@ where
     fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let mut me = self.project();
         Poll::Ready(loop {
-            if let Some(item) = me.queue.next() {
-                break Some(Ok(item));
-            }
             let var_name = match ready!(me.stream.as_mut().poll_next(cx)) {
                 Some(Ok(Event::Apply(obj) | Event::InitApply(obj))) => Some(Ok(obj)),
                 Some(Ok(Event::Delete(obj))) => {
@@ -45,10 +37,6 @@ where
                         continue;
                     }
                 }
-                Some(Ok(Event::InitPage(objs))) => {
-                    *me.queue = objs.into_iter();
-                    continue;
-                }
                 Some(Ok(Event::Init | Event::InitDone)) => continue,
                 Some(Err(err)) => Some(Err(err)),
                 None => return Poll::Ready(None),
@@ -72,8 +60,9 @@ pub(crate) mod tests {
             Ok(Event::Apply(1)),
             Ok(Event::Delete(0)),
             Ok(Event::Apply(2)),
-            Ok(Event::InitPage(vec![1, 2])),
-            Err(Error::TooManyObjects),
+            Ok(Event::InitApply(1)),
+            Ok(Event::InitApply(2)),
+            Err(Error::NoResourceVersion),
             Ok(Event::Apply(2)),
         ]);
         let mut rx = pin!(EventFlatten::new(data, false));
@@ -89,7 +78,7 @@ pub(crate) mod tests {
         // Error passed through
         assert!(matches!(
             poll!(rx.next()),
-            Poll::Ready(Some(Err(Error::TooManyObjects)))
+            Poll::Ready(Some(Err(Error::NoResourceVersion)))
         ));
         assert!(matches!(poll!(rx.next()), Poll::Ready(Some(Ok(2)))));
         assert!(matches!(poll!(rx.next()), Poll::Ready(None)));

diff --git a/kube-runtime/src/utils/event_modify.rs b/kube-runtime/src/utils/event_modify.rs
@@ -46,7 +46,7 @@ where
 
 #[cfg(test)]
 pub(crate) mod test {
-    use std::{pin::pin, task::Poll, vec};
+    use std::{pin::pin, task::Poll};
 
     use super::{Error, Event, EventModify};
     use futures::{poll, stream, StreamExt};
@@ -55,8 +55,8 @@ pub(crate) mod test {
     async fn eventmodify_modifies_innner_value_of_event() {
         let st = stream::iter([
             Ok(Event::Apply(0)),
-            Err(Error::TooManyObjects),
-            Ok(Event::InitPage(vec![10])),
+            Err(Error::NoResourceVersion),
+            Ok(Event::InitApply(10)),
         ]);
         let mut ev_modify = pin!(EventModify::new(st, |x| {
             *x += 1;
@@ -69,13 +69,13 @@ pub(crate) mod test {
 
         assert!(matches!(
             poll!(ev_modify.next()),
-            Poll::Ready(Some(Err(Error::TooManyObjects)))
+            Poll::Ready(Some(Err(Error::NoResourceVersion)))
         ));
 
         let restarted = poll!(ev_modify.next());
         assert!(matches!(
             restarted,
-            Poll::Ready(Some(Ok(Event::InitPage(vec)))) if vec == [11]
+            Poll::Ready(Some(Ok(Event::InitApply(x)))) if x == 11
         ));
 
         assert!(matches!(poll!(ev_modify.next()), Poll::Ready(None)));

diff --git a/kube-runtime/src/utils/predicate.rs b/kube-runtime/src/utils/predicate.rs
@@ -225,7 +225,7 @@ pub(crate) mod tests {
         };
         let data = stream::iter([
             Ok(mkobj(1)),
-            Err(Error::TooManyObjects),
+            Err(Error::NoResourceVersion),
             Ok(mkobj(1)),
             Ok(mkobj(2)),
         ]);
@@ -238,7 +238,7 @@ pub(crate) mod tests {
         // Error passed through
         assert!(matches!(
             poll!(rx.next()),
-            Poll::Ready(Some(Err(Error::TooManyObjects)))
+            Poll::Ready(Some(Err(Error::NoResourceVersion)))
         ));
         // (no repeat mkobj(1) - same generation)
         // mkobj(2) next

diff --git a/kube-runtime/src/utils/reflect.rs b/kube-runtime/src/utils/reflect.rs
@@ -54,7 +54,7 @@ where
 
 #[cfg(test)]
 pub(crate) mod test {
-    use std::{pin::pin, task::Poll, vec};
+    use std::{pin::pin, task::Poll};
 
     use super::{Error, Event, Reflect};
     use crate::reflector;
@@ -73,9 +73,10 @@ pub(crate) mod test {
         let bar = testpod("bar");
         let st = stream::iter([
             Ok(Event::Apply(foo.clone())),
-            Err(Error::TooManyObjects),
+            Err(Error::NoResourceVersion),
             Ok(Event::Init),
-            Ok(Event::InitPage(vec![foo, bar])),
+            Ok(Event::InitApply(foo)),
+            Ok(Event::InitApply(bar)),
             Ok(Event::InitDone),
         ]);
         let (reader, writer) = reflector::store();
@@ -91,7 +92,7 @@ pub(crate) mod test {
 
         assert!(matches!(
             poll!(reflect.next()),
-            Poll::Ready(Some(Err(Error::TooManyObjects)))
+            Poll::Ready(Some(Err(Error::NoResourceVersion)))
         ));
         assert_eq!(reader.len(), 1);
 
@@ -102,7 +103,10 @@ pub(crate) mod test {
         assert_eq!(reader.len(), 1);
 
         let restarted = poll!(reflect.next());
-        assert!(matches!(restarted, Poll::Ready(Some(Ok(Event::InitPage(_))))));
+        assert!(matches!(restarted, Poll::Ready(Some(Ok(Event::InitApply(_))))));
+        assert_eq!(reader.len(), 1);
+        let restarted = poll!(reflect.next());
+        assert!(matches!(restarted, Poll::Ready(Some(Ok(Event::InitApply(_))))));
         assert_eq!(reader.len(), 1);
 
         assert!(matches!(

diff --git a/kube-runtime/src/utils/watch_ext.rs b/kube-runtime/src/utils/watch_ext.rs
@@ -36,7 +36,7 @@ pub trait WatchStreamExt: Stream {
     /// Flatten a [`watcher()`] stream into a stream of applied objects
     ///
     /// All Added/Modified events are passed through, and critical errors bubble up.
-    fn applied_objects<K>(self) -> EventFlatten<Self, K>
+    fn applied_objects<K>(self) -> EventFlatten<Self>
     where
         Self: Stream<Item = Result<watcher::Event<K>, watcher::Error>> + Sized,
     {
@@ -46,7 +46,7 @@ pub trait WatchStreamExt: Stream {
     /// Flatten a [`watcher()`] stream into a stream of touched objects
     ///
     /// All Added/Modified/Deleted events are passed through, and critical errors bubble up.
-    fn touched_objects<K>(self) -> EventFlatten<Self, K>
+    fn touched_objects<K>(self) -> EventFlatten<Self>
     where
         Self: Stream<Item = Result<watcher::Event<K>, watcher::Error>> + Sized,
     {