Broadcast shared reflector

examples/Cargo.toml

@@ -130,6 +130,10 @@ path = "log_stream.rs"
 name = "multi_watcher"
 path = "multi_watcher.rs"
 
+[[example]]
+name = "broadcast_reflector"
+path = "broadcast_reflector.rs"
+
 [[example]]
 name = "pod_api"
 path = "pod_api.rs"

examples/broadcast_reflector.rs

@@ -0,0 +1,113 @@
+use futures::{future, lock::Mutex, pin_mut, stream, StreamExt};
+use k8s_openapi::api::{
+    apps::v1::Deployment,
+    core::v1::{ConfigMap, Secret},
+};
+use kube::{
+    api::ApiResource,
+    runtime::{
+        broadcaster,
+        controller::Action,
+        reflector::multi_dispatcher::{BroadcastStream, MultiDispatcher},
+        watcher, Controller,
+    },
+    Api, Client, ResourceExt,
+};
+use std::{fmt::Debug, pin::pin, sync::Arc, time::Duration};
+use thiserror::Error;
+use tracing::*;
+
+#[derive(Debug, Error)]
+enum Infallible {}
+
+// A generic reconciler that can be used with any object whose type is known at
+// compile time. Will simply log its kind on reconciliation.
+async fn reconcile<K>(_obj: Arc<K>, _ctx: Arc<()>) -> Result<Action, Infallible>
+where
+    K: ResourceExt<DynamicType = ()>,
+{
+    let kind = K::kind(&());
+    info!("Reconciled {kind}");
+    Ok(Action::await_change())
+}
+
+fn error_policy<K: ResourceExt>(_: Arc<K>, _: &Infallible, _ctx: Arc<()>) -> Action {
+    info!("error");
+    Action::requeue(Duration::from_secs(10))
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    tracing_subscriber::fmt::init();
+    let client = Client::try_default().await?;
+
+    let writer = MultiDispatcher::new(128);
+
+    // multireflector stream
+    let combo_stream = Arc::new(Mutex::new(stream::select_all(vec![])));
+    let watcher = broadcaster(writer.clone(), BroadcastStream::new(combo_stream.clone()));
+
+    combo_stream.lock().await.push(
+        watcher::watcher(
+            Api::all_with(client.clone(), &ApiResource::erase::<Deployment>(&())),
+            Default::default(),
+        )
+        .boxed(),
+    );
+
+    // watching config maps, but ignoring in the final configuration
+    combo_stream.lock().await.push(
+        watcher::watcher(
+            Api::all_with(client.clone(), &ApiResource::erase::<ConfigMap>(&())),
+            Default::default(),
+        )
+        .boxed(),
+    );
+
+    // Combine duplicate type streams with narrowed down selection
+    combo_stream.lock().await.push(
+        watcher::watcher(
+            Api::default_namespaced_with(client.clone(), &ApiResource::erase::<Secret>(&())),
+            Default::default(),
+        )
+        .boxed(),
+    );
+    combo_stream.lock().await.push(
+        watcher::watcher(
+            Api::namespaced_with(client.clone(), "kube-system", &ApiResource::erase::<Secret>(&())),
+            Default::default(),
+        )
+        .boxed(),
+    );
+
+    let (sub, reader) = writer.subscribe::<Deployment>();
+    let deploy = Controller::for_shared_stream(sub, reader)
+        .shutdown_on_signal()
+        .run(reconcile, error_policy, Arc::new(()))
+        .for_each(|res| async move {
+            match res {
+                Ok(v) => info!("Reconciled deployment {v:?}"),
+                Err(error) => warn!(%error, "Failed to reconcile metadata"),
+            };
+        });
+
+    let (sub, reader) = writer.subscribe::<Secret>();
+    let secret = Controller::for_shared_stream(sub, reader)
+        .shutdown_on_signal()
+        .run(reconcile, error_policy, Arc::new(()))
+        .for_each(|res| async move {
+            match res {
+                Ok(v) => info!("Reconciled secret {v:?}"),
+                Err(error) => warn!(%error, "Failed to reconcile metadata"),
+            };
+        });
+
+    info!("long watches starting");
+    tokio::select! {
+        r = watcher.for_each(|_| future::ready(())) => println!("watcher exit: {r:?}"),
+        x = deploy => println!("deployments exit: {x:?}"),
+        x = secret => println!("secrets exit: {x:?}"),
+    }
+
+    Ok(())
+}

kube-core/src/dynamic.rs

@@ -5,6 +5,7 @@
 use crate::{
     metadata::TypeMeta,
     resource::{DynamicResourceScope, Resource},
+    GroupVersionKind,
 };
 
 use k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta;
@@ -73,6 +74,12 @@
     ) -> Result<K, ParseDynamicObjectError> {
         Ok(serde_json::from_value(serde_json::to_value(self)?)?)
     }
+
+    /// Returns the group, version, and kind (GVK) of this resource.
+    pub fn gvk(&self) -> Option<GroupVersionKind> {
+        let gvk = self.types.clone()?;
+        gvk.try_into().ok()
+    }
 }
 
 impl Resource for DynamicObject {

kube-core/src/gvk.rs

@@ -12,7 +12,7 @@ use thiserror::Error;
 pub struct ParseGroupVersionError(pub String);
 
 /// Core information about an API Resource.
-#[derive(Deserialize, Serialize, Debug, Clone, PartialEq, Eq, Hash)]
+#[derive(Deserialize, Serialize, Debug, Clone, PartialEq, Eq, Hash, Default)]
 pub struct GroupVersionKind {
     /// API group
     pub group: String,

kube-core/src/metadata.rs

@@ -51,6 +51,22 @@
             kind: K::kind(&()).into(),
         }
     }
+
+    /// Construct a new `TypeMeta` for the object from the list `TypeMeta`.
+    ///
+    /// ```
+    /// # use k8s_openapi::api::core::v1::Pod;
+    /// # use kube_core::TypeMeta;
+    ///
+    /// let mut type_meta = TypeMeta::resource::<Pod>();
+    /// type_meta.kind = "PodList".to_string();
+    /// assert_eq!(type_meta.clone().singular_list().unwrap().kind, "Pod");
+    /// assert_eq!(type_meta.clone().singular_list().unwrap().api_version, "v1");
+    /// ```
+    pub fn singular_list(self) -> Option<Self> {
+        let kind = self.kind.strip_suffix("List")?.to_string();
+        (!kind.is_empty()).then_some(Self { kind, ..self })
+    }
 }
 
 /// A generic representation of any object with `ObjectMeta`.

kube-core/src/resource.rs

@@ -8,6 +8,8 @@
 
 pub use k8s_openapi::{ClusterResourceScope, NamespaceResourceScope, ResourceScope, SubResourceScope};
 
+use crate::GroupVersionKind;
+
 /// Indicates that a [`Resource`] is of an indeterminate dynamic scope.
 pub struct DynamicResourceScope {}
 impl ResourceScope for DynamicResourceScope {}
@@ -54,6 +56,11 @@
     /// This is known as the resource in apimachinery, we rename it for disambiguation.
     fn plural(dt: &Self::DynamicType) -> Cow<'_, str>;
 
+    /// Generates an object reference for the resource
+    fn gvk(dt: &Self::DynamicType) -> GroupVersionKind {
+        GroupVersionKind::gvk(&Self::group(dt), &Self::version(dt), &Self::kind(dt))
+    }
+
     /// Creates a url path for http requests for this resource
     fn url_path(dt: &Self::DynamicType, namespace: Option<&str>) -> String {
         let n = if let Some(ns) = namespace {

kube-runtime/src/lib.rs

@@ -32,6 +32,7 @@ pub mod watcher;
 
 pub use controller::{applier, Config, Controller};
 pub use finalizer::finalizer;
+#[cfg(feature = "unstable-runtime-subscribe")] pub use reflector::broadcaster;
 pub use reflector::reflector;
 pub use scheduler::scheduler;
 pub use utils::WatchStreamExt;

kube-runtime/src/reflector/dispatcher.rs

@@ -6,12 +6,17 @@
 
 use educe::Educe;
 use futures::Stream;
+#[cfg(feature = "unstable-runtime-subscribe")]
+use kube_client::{api::DynamicObject, Resource};
 use pin_project::pin_project;
+#[cfg(feature = "unstable-runtime-subscribe")] use serde::de::DeserializeOwned;
 use std::task::ready;
 
 use crate::reflector::{ObjectRef, Store};
+#[cfg(feature = "unstable-runtime-subscribe")] use crate::watcher::Event;
 use async_broadcast::{InactiveReceiver, Receiver, Sender};
 
+#[cfg(feature = "unstable-runtime-subscribe")] use super::store::Writer;
 use super::Lookup;
 
 #[derive(Educe)]
@@ -125,7 +130,7 @@
 impl<K> Stream for ReflectHandle<K>
 where
     K: Lookup + Clone,
-    K::DynamicType: Eq + std::hash::Hash + Clone + Default,
+    K::DynamicType: Eq + std::hash::Hash + Clone,
 {
     type Item = Arc<K>;
 
@@ -141,6 +146,97 @@
     }
 }
 
+/// A handle to a shared dynamic object stream
+///
+/// [`TypedReflectHandle`]s are created by calling [`subscribe()`] on a [`TypedDispatcher`],
+/// Each shared stream reader should be polled independently and driven to readiness
+/// to avoid deadlocks. When the [`TypedDispatcher`]'s buffer is filled, backpressure
+/// will be applied on the root stream side.
+///
+/// When the root stream is dropped, or it ends, all [`TypedReflectHandle`]s
+/// subscribed to the shared stream will also terminate after all events yielded by
+/// the root stream have been observed. This means [`TypedReflectHandle`] streams
+/// can still be polled after the root stream has been dropped.
+#[cfg(feature = "unstable-runtime-subscribe")]
+#[pin_project]
+pub struct TypedReflectHandle<K>
+where
+    K: Lookup + Clone + 'static,
+    K::DynamicType: Eq + std::hash::Hash + Clone,
+    K: DeserializeOwned,
+{
+    #[pin]
+    rx: Receiver<Event<DynamicObject>>,
+    store: Writer<K>,
+}
+
+#[cfg(feature = "unstable-runtime-subscribe")]
+impl<K> TypedReflectHandle<K>
+where
+    K: Lookup + Clone + 'static,
+    K::DynamicType: Eq + std::hash::Hash + Clone + Default,
+    K: DeserializeOwned,
+{
+    pub(super) fn new(rx: Receiver<Event<DynamicObject>>) -> TypedReflectHandle<K> {
+        Self {
+            rx,
+            // Initialize a ready store by default
+            store: {
+                let mut store: Writer<K> = Default::default();
+                store.apply_shared_watcher_event(&Event::InitDone);
+                store
+            },
+        }
+    }
+
+    pub fn reader(&self) -> Store<K> {
+        self.store.as_reader()
+    }
+}
+
+#[cfg(feature = "unstable-runtime-subscribe")]
+impl<K> Stream for TypedReflectHandle<K>
+where
+    K: Resource + Clone + 'static,
+    K::DynamicType: Eq + std::hash::Hash + Clone + Default,
+    K: DeserializeOwned,
+{
+    type Item = Arc<K>;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let mut this = self.project();
+        loop {
+            return match ready!(this.rx.as_mut().poll_next(cx)) {
+                Some(event) => {
+                    let obj = match event {
+                        Event::InitApply(obj) | Event::Apply(obj)
+                            if obj.gvk() == Some(K::gvk(&Default::default())) =>
+                        {
+                            obj.try_parse::<K>().ok().map(Arc::new).inspect(|o| {
+                                this.store.apply_shared_watcher_event(&Event::Apply(o.clone()));
+                            })
+                        }
+                        Event::Delete(obj) if obj.gvk() == Some(K::gvk(&Default::default())) => {
+                            obj.try_parse::<K>().ok().map(Arc::new).inspect(|o| {
+                                this.store.apply_shared_watcher_event(&Event::Delete(o.clone()));
+                            })
+                        }
+                        _ => None,
+                    };
+
+                    // Skip propagating all objects which do not belong to the cache
+                    if obj.is_none() {
+                        continue;
+                    }
+
+                    Poll::Ready(obj)
+                }
+                None => Poll::Ready(None),
+            };
+        }
+    }
+}
+
 #[cfg(feature = "unstable-runtime-subscribe")]
 #[cfg(test)]
 pub(crate) mod test {

kube-runtime/src/reflector/mod.rs

@@ -1,6 +1,7 @@
 //! Caches objects in memory
 
 mod dispatcher;
+#[cfg(feature = "unstable-runtime-subscribe")] pub mod multi_dispatcher;
 mod object_ref;
 pub mod store;
 
@@ -11,6 +12,12 @@
 use crate::watcher;
 use async_stream::stream;
 use futures::{Stream, StreamExt};
+#[cfg(feature = "unstable-runtime-subscribe")]
+use kube_client::api::DynamicObject;
+#[cfg(feature = "unstable-runtime-subscribe")]
+use multi_dispatcher::BroadcastStream;
+#[cfg(feature = "unstable-runtime-subscribe")]
+use multi_dispatcher::MultiDispatcher;
 use std::hash::Hash;
 #[cfg(feature = "unstable-runtime-subscribe")] pub use store::store_shared;
 pub use store::{store, Store};
@@ -134,6 +141,28 @@
     }
 }
 
+// broadcaster uses a common stream of DynamicObject events to distribute to any subscribed typed watcher.
+#[cfg(feature = "unstable-runtime-subscribe")]
+pub fn broadcaster<W>(
+    mut writer: MultiDispatcher,
+    mut stream: BroadcastStream<W>,
+) -> impl Stream<Item = W::Item>
+where
+    W: Stream<Item = watcher::Result<watcher::Event<DynamicObject>>> + Unpin,
+{
+    stream! {
+        while let Some(event) = stream.next().await {
+            match event {
+                Ok(ev) => {
+                    writer.broadcast_event(&ev).await;
+                    yield Ok(ev);
+                },
+                Err(ev) => yield Err(ev)
+            }
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::{reflector, store, ObjectRef};