fix(stackablectl): Re-run GVK discovery after resolution failure (#294)

* fix(stackablectl): Re-run GVK discovery after resolution failure

* Use tokio::sync::RwLock

* Remove comment

* Update rust/stackable-cockpit/src/utils/k8s/client.rs

Co-authored-by: Nick <[email protected]>

* tracing

* tracing2

* changelog

---------

Co-authored-by: Nick <[email protected]>

Author: sbernauer

rust/stackable-cockpit/src/utils/k8s/client.rs

@@ -7,12 +7,14 @@ use k8s_openapi::api::{
 use kube::{
     api::{ListParams, Patch, PatchParams, PostParams},
     core::{DynamicObject, GroupVersionKind, ObjectList, ObjectMeta, TypeMeta},
-    discovery::Scope,
+    discovery::{ApiCapabilities, ApiResource, Scope},
     Api, Discovery, ResourceExt,
 };
 use serde::Deserialize;
 use snafu::{OptionExt, ResultExt, Snafu};
 use stackable_operator::{commons::listener::Listener, kvp::Labels};
+use tokio::sync::RwLock;
+use tracing::info;
 
 use crate::{
     platform::{cluster, credentials::Credentials},
@@ -39,12 +41,15 @@ pub enum Error {
     #[snafu(display("failed to deserialize YAML data"))]
     DeserializeYaml { source: serde_yaml::Error },
 
+    #[snafu(display("failed to run GVK discovery"))]
+    GVKDiscoveryRun { source: kube::error::Error },
+
+    #[snafu(display("GVK {gvk:?} is not known"))]
+    GVKUnkown { gvk: GroupVersionKind },
+
     #[snafu(display("failed to deploy manifest because type of object {object:?} is not set"))]
     ObjectType { object: DynamicObject },
 
-    #[snafu(display("failed to deploy manifest because GVK {gvk:?} cannot be resolved"))]
-    DiscoveryResolve { gvk: GroupVersionKind },
-
     #[snafu(display("failed to convert byte string into UTF-8 string"))]
     ByteStringConvert { source: FromUtf8Error },
 
@@ -66,7 +71,9 @@ pub enum Error {
 
 pub struct Client {
     client: kube::Client,
-    discovery: Discovery,
+
+    // Choosing an [`RwLock`] here, as their can be many reads in parallel, but running a discovery is very rare
+    discovery: RwLock<Discovery>,
 }
 
 impl Client {
@@ -77,10 +84,7 @@ impl Client {
             .await
             .context(KubeClientCreateSnafu)?;
 
-        let discovery = Discovery::new(client.clone())
-            .run()
-            .await
-            .context(KubeClientFetchSnafu)?;
+        let discovery = RwLock::new(Self::run_discovery(client.clone()).await?);
 
         Ok(Self { client, discovery })
     }
@@ -112,9 +116,9 @@ impl Client {
 
             let gvk = Self::gvk_of_typemeta(object_type);
             let (resource, capabilities) = self
-                .discovery
                 .resolve_gvk(&gvk)
-                .context(DiscoveryResolveSnafu { gvk })?;
+                .await?
+                .context(GVKUnkownSnafu { gvk })?;
 
             let api: Api<DynamicObject> = match capabilities.scope {
                 Scope::Cluster => {
@@ -147,9 +151,9 @@ impl Client {
         gvk: &GroupVersionKind,
         namespace: Option<&str>,
     ) -> Result<Option<ObjectList<DynamicObject>>, Error> {
-        let object_api_resource = match self.discovery.resolve_gvk(gvk) {
-            Some((object_api_resource, _)) => object_api_resource,
-            None => {
+        let object_api_resource = match self.resolve_gvk(gvk).await {
+            Ok(Some((object_api_resource, _))) => object_api_resource,
+            _ => {
                 return Ok(None);
             }
         };
@@ -175,9 +179,9 @@ impl Client {
         object_name: &str,
         gvk: &GroupVersionKind,
     ) -> Result<Option<DynamicObject>, Error> {
-        let object_api_resource = match self.discovery.resolve_gvk(gvk) {
-            Some((object_api_resource, _)) => object_api_resource,
-            None => {
+        let object_api_resource = match self.resolve_gvk(gvk).await {
+            Ok(Some((object_api_resource, _))) => object_api_resource,
+            _ => {
                 return Ok(None);
             }
         };
@@ -383,6 +387,43 @@ impl Client {
         endpoints_api.get(name).await.context(KubeClientFetchSnafu)
     }
 
+    /// Try to resolve the given [`GroupVersionKind`]. In case the resolution fails a discovery is run to pull in new
+    /// GVKs that are not present in the [`Discovery`] cache. Afterwards a normal resolution is issued.
+    async fn resolve_gvk(
+        &self,
+        gvk: &GroupVersionKind,
+    ) -> Result<Option<(ApiResource, ApiCapabilities)>> {
+        let resolved = self.discovery.read().await.resolve_gvk(gvk);
+
+        Ok(match resolved {
+            Some(resolved) => Some(resolved),
+            None => {
+                info!(?gvk, "discovery did not include gvk");
+
+                // We take the lock early here to avoid running multiple discoveries in parallel (as they are expensive)
+                let mut old_discovery = self.discovery.write().await;
+
+                // We create a new Discovery object here, as [`Discovery::run`] consumes self
+                let new_discovery = Self::run_discovery(self.client.clone()).await?;
+                *old_discovery = new_discovery;
+
+                // Release the lock as quickly as possible
+                drop(old_discovery);
+                self.discovery.read().await.resolve_gvk(gvk)
+            }
+        })
+    }
+
+    /// Creates a new [`Discovery`] object and immediatly runs a discovery.
+    #[tracing::instrument(skip_all)]
+    async fn run_discovery(client: kube::client::Client) -> Result<Discovery> {
+        info!("running discovery");
+        Discovery::new(client)
+            .run()
+            .await
+            .context(GVKDiscoveryRunSnafu)
+    }
+
     /// Extracts the [`GroupVersionKind`] from [`TypeMeta`].
     fn gvk_of_typemeta(type_meta: &TypeMeta) -> GroupVersionKind {
         match type_meta.api_version.split_once('/') {

rust/stackablectl/CHANGELOG.md

@@ -4,11 +4,16 @@ All notable changes to this project will be documented in this file.
 
 ## [Unreleased]
 
+### Fixed
+
+- Re-run GVK discovery after resolution failure ([#294]).
+
 ## [24.3.3] - 2024-05-13
 
 - Bump Rust, Go and Node dependencies ([#238]).
 
 [#238]: https://github.com/stackabletech/stackable-cockpit/pull/238
+[#294]: https://github.com/stackabletech/stackable-cockpit/pull/294
 
 ## [24.3.2] - 2024-04-25