Exploring shared cache based on multi reflectors

Signed-off-by: Danil-Grigorev <[email protected]>

examples/Cargo.toml

@@ -22,6 +22,7 @@ ws = ["kube/ws"]
 latest = ["k8s-openapi/latest"]
 
 [dev-dependencies]
+parking_lot.workspace = true
 tokio-util.workspace = true
 assert-json-diff.workspace = true
 garde = { version = "0.22.0", default-features = false, features = ["derive"] }

examples/multi_reflector.rs

@@ -1,74 +1,84 @@
-use futures::{future, StreamExt};
+use futures::{future, stream, StreamExt};
 use k8s_openapi::api::{
     apps::v1::Deployment,
     core::v1::{ConfigMap, Secret},
 };
 use kube::{
+    api::{DynamicObject, GroupVersionKind},
+    core::TypedResource,
     runtime::{
-        reflector,
-        reflector::{ObjectRef, Store},
-        watcher, WatchStreamExt,
+        reflector::store::CacheWriter,
+        watcher::{self, dynamic_watcher},
+        WatchStreamExt,
     },
-    Api, Client,
+    Api, Client, Resource,
 };
+use parking_lot::RwLock;
+use serde::de::DeserializeOwned;
 use std::sync::Arc;
 use tracing::*;
 
-// This does not work because Resource trait is not dyn safe.
-/*
-use std::any::TypeId;
 use std::collections::HashMap;
-use k8s_openapi::NamespaceResourceScope;
-use kube::api::{Resource, ResourceExt};
-struct MultiStore {
-     stores: HashMap<TypeId, Store<dyn Resource<DynamicType = (), Scope = NamespaceResourceScope>>>,
- }
-impl MultiStore {
-    fn get<K: Resource<DynamicType = ()>>(&self, name: &str, ns: &str) -> Option<Arc<K>> {
-        let oref = ObjectRef::<K>::new(name).within(ns);
-        if let Some(store) = self.stores.get(&TypeId::of::<K>()) {
-            store.get(oref)
-        } else {
-            None
-        }
-    }
-}*/
 
-// explicit store can work
-struct MultiStore {
-    deploys: Store<Deployment>,
-    cms: Store<ConfigMap>,
-    secs: Store<Secret>,
+type Cache = Arc<RwLock<HashMap<LookupKey, Arc<DynamicObject>>>>;
+
+#[derive(Default, Clone, Hash, PartialEq, Eq, Debug)]
+struct LookupKey {
+    gvk: GroupVersionKind,
+    name: Option<String>,
+    namespace: Option<String>,
 }
-// but using generics to help out won't because the K needs to be concretised
-/*
-impl MultiStore {
-    fn get<K: Resource<DynamicType = ()>>(&self, name: &str, ns: &str) -> Option<Arc<Option<K>>> {
-        let oref = ObjectRef::<K>::new(name).within(ns);
-        let kind = K::kind(&()).to_owned();
-        match kind.as_ref() {
-            "Deployment" => self.deploys.get(&ObjectRef::new(name).within(ns)),
-            "ConfigMap" => self.cms.get(&ObjectRef::new(name).within(ns)),
-            "Secret" => self.secs.get(&ObjectRef::new(name).within(ns)),
-            _ => None,
+
+impl LookupKey {
+    fn new<R: TypedResource>(resource: &R) -> LookupKey {
+        let meta = resource.meta();
+        LookupKey {
+            gvk: resource.gvk(),
+            name: meta.name.clone(),
+            namespace: meta.namespace.clone(),
         }
-        None
     }
 }
-*/
-// so left with this
 
-impl MultiStore {
-    fn get_deploy(&self, name: &str, ns: &str) -> Option<Arc<Deployment>> {
-        self.deploys.get(&ObjectRef::<Deployment>::new(name).within(ns))
-    }
+#[derive(Default, Clone)]
+struct MultiCache {
+    store: Cache,
+}
 
-    fn get_secret(&self, name: &str, ns: &str) -> Option<Arc<Secret>> {
-        self.secs.get(&ObjectRef::<Secret>::new(name).within(ns))
+impl MultiCache {
+    fn get<K: Resource<DynamicType = impl Default> + DeserializeOwned + Clone>(
+        &self,
+        name: &str,
+        ns: &str,
+    ) -> Option<Arc<K>> {
+        let obj = self
+            .store
+            .read()
+            .get(&LookupKey {
+                gvk: K::gvk(&Default::default()),
+                name: Some(name.into()),
+                namespace: if !ns.is_empty() { Some(ns.into()) } else { None },
+            })?
+            .as_ref()
+            .clone();
+        obj.try_parse().ok().map(Arc::new)
     }
+}
 
-    fn get_cm(&self, name: &str, ns: &str) -> Option<Arc<ConfigMap>> {
-        self.cms.get(&ObjectRef::<ConfigMap>::new(name).within(ns))
+impl CacheWriter<DynamicObject> for MultiCache {
+    /// Applies a single watcher event to the store
+    fn apply_watcher_event(&mut self, event: &watcher::Event<DynamicObject>) {
+        match event {
+            watcher::Event::Init | watcher::Event::InitDone(_) => {}
+            watcher::Event::Delete(obj) => {
+                self.store.write().remove(&LookupKey::new(obj));
+            }
+            watcher::Event::InitApply(obj) | watcher::Event::Apply(obj) => {
+                self.store
+                    .write()
+                    .insert(LookupKey::new(obj), Arc::new(obj.clone()));
+            }
+        }
     }
 }
 
@@ -77,60 +87,36 @@ async fn main() -> anyhow::Result<()> {
     tracing_subscriber::fmt::init();
     let client = Client::try_default().await?;
 
-    let deploys: Api<Deployment> = Api::default_namespaced(client.clone());
-    let cms: Api<ConfigMap> = Api::default_namespaced(client.clone());
-    let secret: Api<Secret> = Api::default_namespaced(client.clone());
-
-    let (dep_reader, dep_writer) = reflector::store::<Deployment>();
-    let (cm_reader, cm_writer) = reflector::store::<ConfigMap>();
-    let (sec_reader, sec_writer) = reflector::store::<Secret>();
+    // multistore
+    let combo_stream = stream::select_all(vec![
+        dynamic_watcher(Api::<Deployment>::all(client.clone()), Default::default()).boxed(),
+        dynamic_watcher(Api::<ConfigMap>::all(client.clone()), Default::default()).boxed(),
+        dynamic_watcher(Api::<Secret>::all(client.clone()), Default::default()).boxed(),
+    ]);
 
-    let cfg = watcher::Config::default();
-    let dep_watcher = watcher(deploys, cfg.clone())
-        .reflect(dep_writer)
-        .applied_objects()
-        .for_each(|_| future::ready(()));
-    let cm_watcher = watcher(cms, cfg.clone())
-        .reflect(cm_writer)
+    let multi_writer = MultiCache::default();
+    let watcher = combo_stream
+        .reflect(multi_writer.clone())
         .applied_objects()
         .for_each(|_| future::ready(()));
-    let sec_watcher = watcher(secret, cfg)
-        .reflect(sec_writer)
-        .applied_objects()
-        .for_each(|_| future::ready(()));
-    // poll these forever
-
-    // multistore
-    let stores = MultiStore {
-        deploys: dep_reader,
-        cms: cm_reader,
-        secs: sec_reader,
-    };
 
     // simulate doing stuff with the stores from some other thread
     tokio::spawn(async move {
-        // Show state every 5 seconds of watching
-        info!("waiting for them to be ready");
-        stores.deploys.wait_until_ready().await.unwrap();
-        stores.cms.wait_until_ready().await.unwrap();
-        stores.secs.wait_until_ready().await.unwrap();
-        info!("stores initialised");
         // can use helper accessors
-        info!(
-            "common cm: {:?}",
-            stores.get_cm("kube-root-ca.crt", "kube-system").unwrap()
-        );
         loop {
             tokio::time::sleep(std::time::Duration::from_secs(5)).await;
+            info!("cache content: {:?}", multi_writer.store.read().keys());
+            info!(
+                "common cm: {:?}",
+                multi_writer.get::<ConfigMap>("kube-root-ca.crt", "kube-system")
+            );
             // access individual sub stores
-            info!("Current deploys count: {}", stores.deploys.state().len());
+            info!("Current objects count: {}", multi_writer.store.read().len());
         }
     });
-    // info!("long watches starting");
+    info!("long watches starting");
     tokio::select! {
-        r = dep_watcher => println!("dep watcher exit: {r:?}"),
-        r = cm_watcher => println!("cm watcher exit: {r:?}"),
-        r = sec_watcher => println!("sec watcher exit: {r:?}"),
+        r = watcher => println!("watcher exit: {r:?}"),
     }
 
     Ok(())

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/lib.rs

@@ -35,7 +35,7 @@ pub mod gvk;
 pub use gvk::{GroupVersion, GroupVersionKind, GroupVersionResource};
 
 pub mod metadata;
-pub use metadata::{ListMeta, ObjectMeta, PartialObjectMeta, PartialObjectMetaExt, TypeMeta};
+pub use metadata::{ListMeta, ObjectMeta, PartialObjectMeta, PartialObjectMetaExt, TypeMeta, TypedResource};
 
 pub mod labels;
 

kube-core/src/metadata.rs

@@ -4,7 +4,7 @@ use std::{borrow::Cow, marker::PhantomData};
 pub use k8s_openapi::apimachinery::pkg::apis::meta::v1::{ListMeta, ObjectMeta};
 use serde::{Deserialize, Serialize};
 
-use crate::{DynamicObject, Resource};
+use crate::{ApiResource, DynamicObject, GroupVersionKind, Resource};
 
 /// Type information that is flattened into every kubernetes object
 #[derive(Deserialize, Serialize, Clone, Default, Debug, Eq, PartialEq, Hash)]
@@ -175,6 +175,123 @@ impl<K: Resource> Resource for PartialObjectMeta<K> {
     }
 }
 
+///
+pub trait TypedResource: Resource + Sized {
+    ///
+    fn type_meta(&self) -> TypeMeta;
+    ///
+    fn gvk(&self) -> GroupVersionKind;
+    ///
+    fn kind(&self) -> Cow<'_, str>;
+    ///
+    fn group(&self) -> Cow<'_, str>;
+    ///
+    fn version(&self) -> Cow<'_, str>;
+    ///
+    fn plural(&self) -> Cow<'_, str>;
+}
+
+impl<K> TypedResource for K
+where
+    K: Resource,
+    (K, K::DynamicType): TypedResourceImpl<Resource = K>,
+{
+    fn type_meta(&self) -> TypeMeta {
+        <(K, K::DynamicType) as TypedResourceImpl>::type_meta(self)
+    }
+
+    fn gvk(&self) -> GroupVersionKind {
+        <(K, K::DynamicType) as TypedResourceImpl>::gvk(self)
+    }
+
+    fn kind(&self) -> Cow<'_, str> {
+        <(K, K::DynamicType) as TypedResourceImpl>::kind(self)
+    }
+    ///
+    fn group(&self) -> Cow<'_, str> {
+        <(K, K::DynamicType) as TypedResourceImpl>::group(self)
+    }
+    ///
+    fn version(&self) -> Cow<'_, str> {
+        <(K, K::DynamicType) as TypedResourceImpl>::version(self)
+    }
+    ///
+    fn plural(&self) -> Cow<'_, str> {
+        <(K, K::DynamicType) as TypedResourceImpl>::plural(self)
+    }
+}
+
+#[doc(hidden)]
+// Workaround for https://github.com/rust-lang/rust/issues/20400
+pub trait TypedResourceImpl {
+    type Resource: Resource;
+    fn type_meta(res: &Self::Resource) -> TypeMeta;
+    fn gvk(res: &Self::Resource) -> GroupVersionKind;
+    fn kind(res: &Self::Resource) -> Cow<'_, str>;
+    fn group(res: &Self::Resource) -> Cow<'_, str>;
+    fn version(res: &Self::Resource) -> Cow<'_, str>;
+    fn plural(res: &Self::Resource) -> Cow<'_, str>;
+}
+
+impl<K> TypedResourceImpl for (K, ())
+where
+    K: Resource<DynamicType = ()>,
+{
+    type Resource = K;
+
+    fn type_meta(_: &Self::Resource) -> TypeMeta {
+        TypeMeta::resource::<K>()
+    }
+
+    fn gvk(res: &Self::Resource) -> GroupVersionKind {
+        GroupVersionKind::gvk(&res.group(), &res.version(), &res.kind())
+    }
+
+    fn kind(_: &Self::Resource) -> Cow<'_, str> {
+        K::kind(&())
+    }
+
+    fn group(_: &Self::Resource) -> Cow<'_, str> {
+        K::group(&())
+    }
+
+    fn version(_: &Self::Resource) -> Cow<'_, str> {
+        K::version(&())
+    }
+
+    fn plural(_: &Self::Resource) -> Cow<'_, str> {
+        K::plural(&())
+    }
+}
+
+impl TypedResourceImpl for (DynamicObject, ApiResource) {
+    type Resource = DynamicObject;
+
+    fn type_meta(obj: &Self::Resource) -> TypeMeta {
+        obj.types.clone().unwrap_or_default()
+    }
+
+    fn gvk(res: &Self::Resource) -> GroupVersionKind {
+        res.type_meta().try_into().unwrap_or_default()
+    }
+
+    fn kind(res: &Self::Resource) -> Cow<'_, str> {
+        Cow::from(res.type_meta().kind)
+    }
+
+    fn group(res: &Self::Resource) -> Cow<'_, str> {
+        Cow::from(res.gvk().group)
+    }
+
+    fn version(res: &Self::Resource) -> Cow<'_, str> {
+        Cow::from(res.gvk().version)
+    }
+
+    fn plural(res: &Self::Resource) -> Cow<'_, str> {
+        Cow::from(ApiResource::from_gvk(&res.gvk()).plural)
+    }
+}
+
 #[cfg(test)]
 mod test {
     use super::{ObjectMeta, PartialObjectMeta, PartialObjectMetaExt};

kube-core/src/resource.rs

@@ -8,6 +8,8 @@ use std::{borrow::Cow, collections::BTreeMap};
 
 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 @@ pub trait Resource {
     /// 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 {