refactor: k8s clients (#1332)

The refactor consists in

1. Remove all methods in `DynamicObjectManagers` that called methods in the inner elements. Converting it into a real "collection".
2. Make sure that `SyncK8sClient` methods always call `AsyncK8sClient` methods.

Author: danielorihuela

agent-control/src/k8s/client.rs

@@ -65,7 +65,12 @@ impl SyncK8sClient {
 
     pub fn apply_dynamic_object(&self, obj: &DynamicObject) -> Result<(), K8sError> {
         self.runtime
-            .block_on(self.async_client.dynamic_object_managers.apply(obj))
+            .block_on(self.async_client.apply_dynamic_object(obj))
+    }
+
+    pub fn apply_dynamic_object_if_changed(&self, obj: &DynamicObject) -> Result<(), K8sError> {
+        self.runtime
+            .block_on(self.async_client.apply_dynamic_object_if_changed(obj))
     }
 
     pub fn patch_dynamic_object(
@@ -74,24 +79,8 @@ impl SyncK8sClient {
         name: &str,
         patch: serde_json::Value,
     ) -> Result<DynamicObject, K8sError> {
-        self.runtime.block_on(
-            self.async_client
-                .dynamic_object_managers
-                .patch(tm, name, patch),
-        )
-    }
-
-    pub fn has_dynamic_object_changed(&self, obj: &DynamicObject) -> Result<bool, K8sError> {
         self.runtime
-            .block_on(self.async_client.dynamic_object_managers.has_changed(obj))
-    }
-
-    pub fn apply_dynamic_object_if_changed(&self, obj: &DynamicObject) -> Result<(), K8sError> {
-        self.runtime.block_on(
-            self.async_client
-                .dynamic_object_managers
-                .apply_if_changed(obj),
-        )
+            .block_on(self.async_client.patch_dynamic_object(tm, name, patch))
     }
 
     pub fn get_dynamic_object(
@@ -100,15 +89,16 @@ impl SyncK8sClient {
         name: &str,
     ) -> Result<Option<Arc<DynamicObject>>, K8sError> {
         self.runtime
-            .block_on(self.async_client.dynamic_object_managers.get(tm, name))
+            .block_on(self.async_client.get_dynamic_object(tm, name))
     }
+
     pub fn delete_dynamic_object(
         &self,
         tm: &TypeMeta,
         name: &str,
     ) -> Result<Either<DynamicObject, Status>, K8sError> {
         self.runtime
-            .block_on(self.async_client.dynamic_object_managers.delete(tm, name))
+            .block_on(self.async_client.delete_dynamic_object(tm, name))
     }
 
     pub fn delete_dynamic_object_collection(
@@ -118,14 +108,18 @@ impl SyncK8sClient {
     ) -> Result<Either<ObjectList<DynamicObject>, Status>, K8sError> {
         self.runtime.block_on(
             self.async_client
-                .dynamic_object_managers
-                .delete_collection(tm, label_selector),
+                .delete_dynamic_object_collection(tm, label_selector),
         )
     }
 
     pub fn list_dynamic_objects(&self, tm: &TypeMeta) -> Result<Vec<Arc<DynamicObject>>, K8sError> {
         self.runtime
-            .block_on(self.async_client.dynamic_object_managers.list(tm))
+            .block_on(self.async_client.list_dynamic_objects(tm))
+    }
+
+    pub fn has_dynamic_object_changed(&self, obj: &DynamicObject) -> Result<bool, K8sError> {
+        self.runtime
+            .block_on(self.async_client.has_dynamic_object_changed(obj))
     }
 
     pub fn delete_configmap_collection(&self, label_selector: &str) -> Result<(), K8sError> {
@@ -233,10 +227,6 @@ impl AsyncK8sClient {
         })
     }
 
-    pub fn dynamic_object_managers(&self) -> &DynamicObjectManagers {
-        &self.dynamic_object_managers
-    }
-
     // Due to the Kube-rs library we need to retrieve with two different calls the versions of each object and then fetch the available kinds
     pub async fn list_api_resources(&self) -> Result<Vec<APIResourceList>, K8sError> {
         let mut list = vec![];
@@ -355,6 +345,94 @@ impl AsyncK8sClient {
     pub fn default_namespace(&self) -> &str {
         self.client.default_namespace()
     }
+
+    pub async fn apply_dynamic_object(&self, obj: &DynamicObject) -> Result<(), K8sError> {
+        let type_meta = get_type_meta(obj)?;
+
+        self.dynamic_object_managers
+            .get_or_create(&type_meta)
+            .await?
+            .apply(obj)
+            .await
+    }
+
+    pub async fn apply_dynamic_object_if_changed(
+        &self,
+        obj: &DynamicObject,
+    ) -> Result<(), K8sError> {
+        let type_meta = get_type_meta(obj)?;
+
+        self.dynamic_object_managers
+            .get_or_create(&type_meta)
+            .await?
+            .apply_if_changed(obj)
+            .await
+    }
+
+    pub async fn patch_dynamic_object(
+        &self,
+        type_meta: &TypeMeta,
+        name: &str,
+        patch: serde_json::Value,
+    ) -> Result<DynamicObject, K8sError> {
+        self.dynamic_object_managers
+            .get_or_create(type_meta)
+            .await?
+            .patch(name, patch)
+            .await
+    }
+
+    pub async fn get_dynamic_object(
+        &self,
+        tm: &TypeMeta,
+        name: &str,
+    ) -> Result<Option<Arc<DynamicObject>>, K8sError> {
+        Ok(self
+            .dynamic_object_managers
+            .get_or_create(tm)
+            .await?
+            .get(name))
+    }
+
+    pub async fn delete_dynamic_object(
+        &self,
+        tm: &TypeMeta,
+        name: &str,
+    ) -> Result<Either<DynamicObject, Status>, K8sError> {
+        self.dynamic_object_managers
+            .get_or_create(tm)
+            .await?
+            .delete(name)
+            .await
+    }
+
+    pub async fn delete_dynamic_object_collection(
+        &self,
+        tm: &TypeMeta,
+        label_selector: &str,
+    ) -> Result<Either<ObjectList<DynamicObject>, Status>, K8sError> {
+        self.dynamic_object_managers
+            .get_or_create(tm)
+            .await?
+            .delete_collection(label_selector)
+            .await
+    }
+
+    pub async fn list_dynamic_objects(
+        &self,
+        tm: &TypeMeta,
+    ) -> Result<Vec<Arc<DynamicObject>>, K8sError> {
+        Ok(self.dynamic_object_managers.get_or_create(tm).await?.list())
+    }
+
+    pub async fn has_dynamic_object_changed(&self, obj: &DynamicObject) -> Result<bool, K8sError> {
+        let tm = get_type_meta(obj)?;
+
+        self.dynamic_object_managers
+            .get_or_create(&tm)
+            .await?
+            .has_changed(obj)
+    }
 }
 
 //  delete_collection has been moved outside the client to be able to use mockall in the client

agent-control/src/k8s/dynamic_object.rs

@@ -1,12 +1,11 @@
 //! This module holds helpers and to perform k8s operations with resources whose type is known at runtime
 //! (DynamicObjects).
 use super::{
-    client::{get_name, get_type_meta},
+    client::{delete_collection, get_name},
     error::K8sError,
     reflector::definition::{Reflector, ReflectorBuilder},
     utils::display_type,
 };
-use crate::k8s::client::delete_collection;
 use base64::engine::general_purpose::STANDARD;
 use either::Either;
 use kube::api::{ObjectList, Patch, PatchParams};
@@ -155,6 +154,13 @@ impl DynamicObjectManager {
         }
         Ok(either)
     }
+
+    pub async fn delete_collection(
+        &self,
+        label_selector: &str,
+    ) -> Result<Either<ObjectList<DynamicObject>, Status>, K8sError> {
+        delete_collection(&self.api, label_selector).await
+    }
 }
 
 /// Holds a collection of [DynamicObjectManager] by [TypeMeta] to perform operations with objects known at runtime.
@@ -175,79 +181,10 @@ impl DynamicObjectManagers {
         }
     }
 
-    pub async fn get(
-        &self,
-        type_meta: &TypeMeta,
-        name: &str,
-    ) -> Result<Option<Arc<DynamicObject>>, K8sError> {
-        Ok(self.get_or_create_manager(type_meta).await?.get(name))
-    }
-
-    pub async fn delete_collection(
-        &self,
-        type_meta: &TypeMeta,
-        label_selector: &str,
-    ) -> Result<Either<ObjectList<DynamicObject>, Status>, K8sError> {
-        let api = self.get_or_create_manager(type_meta).await?.api.clone();
-
-        delete_collection(&api, label_selector).await
-    }
-
-    pub async fn delete(
-        &self,
-        type_meta: &TypeMeta,
-        name: &str,
-    ) -> Result<Either<DynamicObject, Status>, K8sError> {
-        self.get_or_create_manager(type_meta)
-            .await?
-            .delete(name)
-            .await
-    }
-
-    pub async fn list(&self, type_meta: &TypeMeta) -> Result<Vec<Arc<DynamicObject>>, K8sError> {
-        Ok(self.get_or_create_manager(type_meta).await?.list())
-    }
-
-    pub async fn apply(&self, obj: &DynamicObject) -> Result<(), K8sError> {
-        let type_meta = &get_type_meta(obj)?;
-
-        self.get_or_create_manager(type_meta)
-            .await?
-            .apply(obj)
-            .await
-    }
-
-    pub async fn apply_if_changed(&self, obj: &DynamicObject) -> Result<(), K8sError> {
-        let type_meta = &get_type_meta(obj)?;
-
-        self.get_or_create_manager(type_meta)
-            .await?
-            .apply_if_changed(obj)
-            .await
-    }
-
-    pub async fn patch(
-        &self,
-        type_meta: &TypeMeta,
-        name: &str,
-        patch: serde_json::Value,
-    ) -> Result<DynamicObject, K8sError> {
-        self.get_or_create_manager(type_meta)
-            .await?
-            .patch(name, patch)
-            .await
-    }
-
-    pub async fn has_changed(&self, obj: &DynamicObject) -> Result<bool, K8sError> {
-        let type_meta = &get_type_meta(obj)?;
-
-        self.get_or_create_manager(type_meta)
-            .await?
-            .has_changed(obj)
-    }
-
-    /// Obtains the manager for the provided [TypeMeta]. If it does not exist it creates it and stores it.
-    async fn get_or_create_manager(
+    /// Returns the manager for the provided [TypeMeta].
+    ///
+    /// If it does not exist it creates it and stores it.
+    pub(super) async fn get_or_create(
         &self,
         type_meta: &TypeMeta,
     ) -> Result<Arc<DynamicObjectManager>, K8sError> {