The 'ensure_crd_created' method retries on failure. (#79)

* The 'ensure_crd_created' method retries on failure.

* Docs + properly recognized restarted state in wait_ready

* Add test for wait_ready method. Make default namespace from kubeconfig accessible.

* Do not panic on Applied event, as pod status might change.

* Test wait_created timeout on nonexisting resource

* Cargo fmt after merge with main

* Reword wait_created documentation and add an example

* Used stream approach for resource waiting

* Kubernetes-dependent tests ignored by default. CONTRIBUTING guide.

* Run all tests, not only ignored

* Renamed wait_ready to wait_created in crd.rs

* Removed explicit timeout on 'ensure_crd_created' function in crd.rs

Author: Pscheidl

.github/workflows/rust.yml

@@ -20,10 +20,22 @@ jobs:
           profile: minimal
           toolchain: nightly
           override: true
+      - name: Start K3S
+        uses: debianmaster/[email protected]
+        id: k3s
+        with:
+          version: 'latest'
+      - uses: actions-rs/cargo@v1
+        name: Run tests
+        with:
+          command: test
       - uses: actions-rs/cargo@v1
+        name: Run ignored-by-default tests
         with:
           command: test
+          args: -- --ignored
       - uses: actions-rs/cargo@v1
+        name: Release build
         with:
           command: build
           args: --release

CONTRIBUTING.adoc

@@ -0,0 +1,23 @@
+= OPERATOR-RS CONTRIBUTING GUIDE
+
+Contributions in any form are welcome and appreciated. The recommended approach for code contributions is to fork this repository and create a pull request.
+
+== CODE
+
+* Use the *stable* branch of Rust, not nightly.
+* Code must be formatted with https://github.com/rust-lang/rustfmt[rustfmt]. Easiest way to install `rustfmt` is to to invoke `cargo install rustfmt` and then format the project with `cargo fmt`.
+* Rustdoc: Structs, functions and all other elements are documented. In general, documentation style follows the https://doc.rust-lang.org/rustdoc/what-is-rustdoc.html[Rustdoc] guide.
+* Tests dependent on Kubernetes runtime are prefixed with `k8s_test_` and ignored.
+
+
+== TESTS
+
+Tests requiring Kubernetes are *disabled* by default, yet enabled in link:.github/workflows/rust.yml[rust.yml] workflow. All others tests are *enabled* by default. This implies `cargo test` won't run Kubernetes-dependent tests by default. To run ignored test(s), invoke `cargo test -- --ignored`.
+
+Kubernetes tests require a `KUBECONFIG` environment variable pointing to a valid kubeconfig file. The kubeconfig must:
+
+* Point to a reachable Kubernetes cluster,
+* Use account with enough permissions to run the tests,
+* Be readable by the user running the tests.
+
+On Linux, a convenient way to install local Kubernetes cluster is https://k3s.io/:[K3S.io]. Invoking `curl -sfL https://get.k3s.io | sh -` installs latest version of K3S and exports a `KUBECONFIG` at `/etc/rancher/k3s/k3s.yaml`. To run Kubernetes-dependent tests with K3S, set the `KUBECONFIG` environment variable using `export KUBECONFIG=/etc/rancher/k3s/k3s.yaml`. Make sure it is readable by user invoking the tests by using `chown`/`chmod`. Running the `cargo test -- --ignored` command afterwards runs all tests, including Kubernetes tests. K3S is also leveraged in link:.github/workflows/rust.yml[rust.yml] workflow.

README.adoc

@@ -1,5 +1,10 @@
-= operator-rs
+= OPERATOR-RS
 
 This is a simple library that includes all kinds of helper methods, structs and enums that can be used to write a Kubernetes Controller/Operator with the https://github.com/clux/kube-rs[kube-rs] crate.
 
-WARNING: This is very new, undocumented and untested!
+WARNING: This is very new, undocumented and partially tested (see the link:CONTRIBUTING.adoc[contributing guide])!
+
+
+== CONTRIBUTING
+
+Contributions are much welcome. A detailed guide on contributing, development and testing is to be found at link:CONTRIBUTING.adoc[CONTRIBUTING].

src/client.rs

@@ -4,13 +4,15 @@ use crate::label_selector;
 use crate::podutils;
 
 use either::Either;
+use futures::StreamExt;
 use k8s_openapi::apimachinery::pkg::apis::meta::v1::{Condition, LabelSelector};
 use k8s_openapi::Resource;
 use kube::api::{DeleteParams, ListParams, Meta, Patch, PatchParams, PostParams};
 use kube::client::{Client as KubeClient, Status};
-use kube::Api;
+use kube::{Api, Config};
 use serde::de::DeserializeOwned;
 use serde::Serialize;
+use std::convert::TryFrom;
 use tracing::trace;
 
 /// This `Client` can be used to access Kubernetes.
@@ -21,10 +23,16 @@ pub struct Client {
     patch_params: PatchParams,
     post_params: PostParams,
     delete_params: DeleteParams,
+    /// Default namespace as defined in the kubeconfig this client has been created from.
+    pub default_namespace: String,
 }
 
 impl Client {
-    pub fn new(client: KubeClient, field_manager: Option<String>) -> Self {
+    pub fn new(
+        client: KubeClient,
+        field_manager: Option<String>,
+        default_namespace: String,
+    ) -> Self {
         Client {
             client,
             post_params: PostParams {
@@ -39,6 +47,7 @@ impl Client {
                 ..PatchParams::default()
             },
             delete_params: DeleteParams::default(),
+            default_namespace,
         }
     }
 
@@ -236,7 +245,7 @@ impl Client {
     /// It checks whether the resource is already deleted by looking at the `deletion_timestamp`
     /// of the resource using the [`finalizer::has_deletion_stamp`] method.
     /// If that is the case it'll return a `Ok(None)`.
-    ///    
+    ///
     /// In case the object is actually deleted or marked for deletion there are two possible
     /// return types.
     /// Which of the two are returned depends on the API being called.
@@ -309,11 +318,167 @@ impl Client {
     {
         Api::namespaced(self.client.clone(), namespace)
     }
+
+    /// Waits indefinitely until resources matching given `ListParams` are created in Kubernetes.
+    /// If the resource is already present, this method just returns. Makes no assumptions about resource's state,
+    /// e.g. a pod created could be created, but not in a ready state.
+    ///
+    /// # Arguments
+    ///
+    /// - `namespace` - Optional namespace to look for the resources in.
+    /// - `lp` - Parameters to filter resources to wait for in given namespace.
+    ///
+    /// # Example
+    ///
+    /// ```no_run
+    /// use kube::api::ListParams;
+    /// use std::time::Duration;
+    /// use tokio::time::error::Elapsed;
+    /// use k8s_openapi::api::core::v1::Pod;
+    /// use stackable_operator::client::{Client, create_client};
+    ///
+    /// #[tokio::main]
+    /// async fn main(){
+    /// let client: Client = create_client(None).await.expect("Unable to construct client.");
+    /// let lp: ListParams =
+    ///         ListParams::default().fields(&format!("metadata.name=nonexistent-pod"));
+    ///
+    /// // Will time out in 1 second unless the nonexistent-pod actually exists
+    ///  let wait_created_result: Result<(), Elapsed> = tokio::time::timeout(
+    ///          Duration::from_secs(1),
+    ///          client.wait_created::<Pod>(Some(client.default_namespace.clone()), lp.clone()),
+    ///      )
+    ///      .await;
+    /// }
+    /// ```
+    ///
+    pub async fn wait_created<T>(&self, namespace: Option<String>, lp: ListParams)
+    where
+        T: Meta + Clone + DeserializeOwned + Send + 'static,
+    {
+        let api: Api<T> = self.get_api(namespace);
+        let watcher = kube_runtime::watcher(api, lp).boxed();
+        kube_runtime::utils::try_flatten_applied(watcher)
+            .skip_while(|res| std::future::ready(res.is_err()))
+            .next()
+            .await;
+    }
 }
 
 pub async fn create_client(field_manager: Option<String>) -> OperatorResult<Client> {
+    let kubeconfig: Config = kube::Config::infer().await?;
+    let default_namespace = kubeconfig.default_ns.clone();
     Ok(Client::new(
-        kube::Client::try_default().await?,
+        kube::Client::try_from(kubeconfig)?,
         field_manager,
+        default_namespace,
     ))
 }
+
+#[cfg(test)]
+mod tests {
+    use futures::StreamExt;
+    use k8s_openapi::api::core::v1::{Container, Pod, PodSpec};
+    use kube::api::{ListParams, Meta, ObjectMeta, PostParams};
+    use kube_runtime::watcher::Event;
+    use std::time::Duration;
+    use tokio::time::error::Elapsed;
+
+    #[tokio::test]
+    #[ignore = "Tests depending on Kubernetes are not ran by default"]
+    async fn k8s_test_wait_created() {
+        let client = super::create_client(None)
+            .await
+            .expect("KUBECONFIG variable must be configured.");
+
+        // Definition of the pod the `wait_created` function will be waiting for.
+        let pod_to_wait_for: Pod = Pod {
+            metadata: ObjectMeta {
+                name: Some("test-wait-created-busybox".to_owned()),
+                ..ObjectMeta::default()
+            },
+            spec: Some(PodSpec {
+                containers: vec![Container {
+                    name: "test-wait-created-busybox".to_owned(),
+                    image: Some("busybox:latest".to_owned()),
+                    image_pull_policy: Some("IfNotPresent".to_owned()),
+                    command: Some(vec!["sleep".into(), "infinity".into()]),
+                    ..Container::default()
+                }],
+                termination_grace_period_seconds: Some(1),
+                ..PodSpec::default()
+            }),
+            ..Pod::default()
+        };
+        let api = client.get_api(Some(client.default_namespace.clone()));
+        let created_pod = api
+            .create(&PostParams::default(), &pod_to_wait_for)
+            .await
+            .expect("Test pod not created.");
+        let lp: ListParams = ListParams::default().fields(&format!(
+            "metadata.name={}",
+            created_pod
+                .metadata
+                .name
+                .as_ref()
+                .expect("Expected busybox pod to have metadata")
+        ));
+        // First, let the tested `wait_creation` function wait until the resource is present.
+        // Timeout is not acceptable
+        tokio::time::timeout(
+            Duration::from_secs(30), // Busybox is ~5MB and sub 1 sec to start.
+            client.wait_created::<Pod>(Some(client.default_namespace.clone()), lp.clone()),
+        )
+        .await
+        .expect("The tested wait_created function timed out.");
+
+        // A second, manually constructed watcher is used to verify the ListParams filter out the correct resource
+        // and the `wait_created` function returned when the correct resources had been detected.
+        let mut ready_watcher = kube_runtime::watcher::<Pod>(api, lp).boxed();
+        while let Some(result) = ready_watcher.next().await {
+            match result {
+                Ok(event) => match event {
+                    Event::Applied(pod) => {
+                        assert_eq!("test-wait-created-busybox", pod.name());
+                    }
+                    Event::Restarted(pods) => {
+                        assert_eq!(1, pods.len());
+                        assert_eq!("test-wait-created-busybox", &pods[0].name());
+                        break;
+                    }
+                    Event::Deleted(_) => {
+                        panic!("Not expected the test_wait_created busybox pod to be deleted");
+                    }
+                },
+                Err(_) => {
+                    panic!("Error while waiting for readiness.");
+                }
+            }
+        }
+
+        client
+            .delete(&created_pod)
+            .await
+            .expect("Expected test_wait_created pod to be deleted.");
+    }
+
+    #[tokio::test]
+    #[ignore = "Tests depending on Kubernetes are not ran by default"]
+    async fn k8s_test_wait_created_timeout() {
+        let client = super::create_client(None)
+            .await
+            .expect("KUBECONFIG variable must be configured.");
+
+        let lp: ListParams =
+            ListParams::default().fields(&format!("metadata.name=nonexistent-pod"));
+
+        // There is no such pod, therefore the `wait_created` function call times out.
+        let wait_created_result: Result<(), Elapsed> = tokio::time::timeout(
+            Duration::from_secs(1),
+            client.wait_created::<Pod>(Some(client.default_namespace.clone()), lp.clone()),
+        )
+        .await;
+
+        assert!(wait_created_result.is_err());
+    }
+}

src/crd.rs

@@ -1,10 +1,13 @@
-use crate::client::Client;
-use crate::error::{Error, OperatorResult};
+use std::time::Duration;
 
 use k8s_openapi::apiextensions_apiserver::pkg::apis::apiextensions::v1::CustomResourceDefinition;
 use kube::error::ErrorResponse;
 use tracing::info;
 
+use crate::client::Client;
+use crate::error::{Error, OperatorResult};
+use kube::api::ListParams;
+
 /// This trait can be implemented to allow automatic handling
 /// (e.g. creation) of `CustomResourceDefinition`s in Kubernetes.
 pub trait Crd {
@@ -61,10 +64,16 @@ where
     }
 }
 
-/// This makes sure the CRD is registered in the apiserver.
-/// Currently this does not retry internally.
-/// This means that running it again _might_ work in case of transient errors.
-// TODO: Make sure to wait until it's enabled in the apiserver
+/// Makes sure CRD of given type `T` is running and accepted by the Kubernetes apiserver.
+/// If the CRD already exists at the time this method is invoked, this method exits.
+/// If there is no CRD of type `T` yet, it will attempt to create it and verify k8s apiserver
+/// applied the CRD. This method retries indefinitely. Use timeout on the `future` returned
+/// to apply time limit constraint.
+///
+/// # Parameters
+/// - `client`: Client to connect to Kubernetes API and create the CRD with
+/// - `timeout`: If specified, retries creating the CRD for given `Duration`. If not specified,
+///     retries indefinitely.
 pub async fn ensure_crd_created<T>(client: Client) -> OperatorResult<()>
 where
     T: Crd,
@@ -74,8 +83,15 @@ where
         Ok(())
     } else {
         info!("CRD not detected in Kubernetes. Attempting to create it.");
-        create::<T>(client).await
-        // TODO: Maybe retry?
+
+        loop {
+            if let Ok(res) = create::<T>(client.clone()).await {
+                break res;
+            }
+            tokio::time::sleep(Duration::from_millis(100)).await;
+        }
+        wait_created::<T>(client.clone()).await?;
+        Ok(())
     }
 }
 
@@ -90,3 +106,18 @@ where
     let crd: CustomResourceDefinition = serde_yaml::from_str(T::CRD_DEFINITION)?;
     client.create(&crd).await.and(Ok(()))
 }
+
+/// Waits until CRD of given type `T` is applied to Kubernetes.
+pub async fn wait_created<T>(client: Client) -> OperatorResult<()>
+where
+    T: Crd,
+{
+    let lp: ListParams = ListParams {
+        field_selector: Some(format!("metadata.name={}", T::RESOURCE_NAME)),
+        ..ListParams::default()
+    };
+    client
+        .wait_created::<CustomResourceDefinition>(None, lp)
+        .await;
+    Ok(())
+}

src/error.rs

@@ -23,6 +23,11 @@ pub enum Error {
 
     #[error("LabelSelector is invalid: {message}")]
     InvalidLabelSelector { message: String },
+    #[error("Operation timed out: {source}")]
+    TimeoutError {
+        #[from]
+        source: tokio::time::error::Elapsed,
+    },
 }
 
 pub type OperatorResult<T> = std::result::Result<T, Error>;