feat: Move the finalization handling out of the Controller (#81)

BREAKING CHANGE: This means that adding and removing finalizers is now optional

Author: lfrancke

Cargo.toml

@@ -10,8 +10,9 @@ chrono = "0.4"
 const_format = "0.2"
 either = "1"
 futures = "0.3"
+json-patch = "0.2"
 k8s-openapi = { version = "0.11", default-features = false }
-kube = { version = "0.49", default-features = false }
+kube = { version = "0.49", default-features = false, features = ["jsonpatch"] }
 kube-runtime = "0.49"
 lazy_static = "1"
 regex = "1"

src/client.rs

@@ -124,6 +124,20 @@ impl Client {
             .await
     }
 
+    /// Patches a resource using the `JSON` patch strategy described in [JavaScript Object Notation (JSON) Patch](https://tools.ietf.org/html/rfc6902).
+    pub async fn json_patch<T>(&self, resource: &T, patch: json_patch::Patch) -> OperatorResult<T>
+    where
+        T: Clone + DeserializeOwned + Meta,
+    {
+        // The `()` type is not used. I need to provide _some_ type just to get it to compile.
+        // But the type is not used _at all_ for the `Json` variant so I'd argue it's okay to
+        // provide any type here.
+        // This is definitely a hack though but there is currently no better way.
+        // See also: https://github.com/clux/kube-rs/pull/456
+        let patch = Patch::Json::<()>(patch);
+        self.patch(resource, patch, &self.patch_params).await
+    }
+
     async fn patch<T, P>(
         &self,
         resource: &T,

src/controller.rs

@@ -111,14 +111,14 @@
 //!     let controller = Controller::new(pods_api);
 //!
 //!     let strategy = FooStrategy {};
-//!     controller.run(client, strategy).await;
+//!     controller.run(client, strategy, Duration::from_secs(10)).await;
 //! }
 //! ```
 //!
 use crate::client::Client;
-use crate::error::{Error, OperatorResult};
+use crate::error::Error;
+use crate::reconcile;
 use crate::reconcile::{ReconcileFunctionAction, ReconciliationContext};
-use crate::{finalizer, reconcile};
 
 use async_trait::async_trait;
 use futures::StreamExt;
@@ -131,7 +131,7 @@ use std::fmt::{Debug, Display};
 use std::future::Future;
 use std::pin::Pin;
 use std::time::Duration;
-use tracing::{debug, error, info, trace, Instrument};
+use tracing::{debug, error, trace, Instrument};
 use uuid::Uuid;
 
 /// Every operator needs to provide an implementation of this trait as it provides the operator specific business logic.
@@ -239,14 +239,23 @@ where
         self
     }
 
-    /// Call this method once your Controller object is fully configured.
-    /// It'll start talking to Kubernetes and will call the `Strategy` implementation.
-    pub async fn run<S>(self, client: Client, strategy: S)
+    /// Call this method once your Controller object is fully configured to start the reconciliation.
+    ///
+    /// # Arguments
+    ///
+    /// * `client` - The Client to access Kubernetes
+    /// * `strategy` - This implements the domain/business logic and the framework will call its methods for each reconcile operation
+    /// * `requeue_timeout` - Whenever a `Requeue` is returned this is the timeout/duration after which the same object will be requeued
+    pub async fn run<S>(self, client: Client, strategy: S, requeue_timeout: Duration)
     where
         S: ControllerStrategy<Item = T> + Send + Sync + 'static,
         S::State: Send,
     {
-        let context = Context::new(ControllerContext { client, strategy });
+        let context = Context::new(ControllerContext {
+            client,
+            strategy,
+            requeue_timeout,
+        });
 
         self.kube_controller
             .run(reconcile, error_policy, context)
@@ -285,6 +294,7 @@ where
 {
     client: Client,
     strategy: S,
+    requeue_timeout: Duration,
 }
 
 /// This method contains the logic of reconciling an object (the desired state) we received with the actual state.
@@ -307,18 +317,13 @@ where
     debug!(?resource, "Beginning reconciliation");
     let context = context.get_ref();
 
-    let client = &context.client;
     let strategy = &context.strategy;
 
-    if handle_deletion(&resource, client.clone(), &strategy.finalizer_name()).await?
-        == ReconcileFunctionAction::Done
-    {
-        return Ok(reconcile::create_non_requeuing_reconciler_action());
-    }
-
-    add_finalizer(&resource, client.clone(), &strategy.finalizer_name()).await?;
-
-    let rc = ReconciliationContext::new(context.client.clone(), resource.clone());
+    let rc = ReconciliationContext::new(
+        context.client.clone(),
+        resource.clone(),
+        context.requeue_timeout,
+    );
 
     let mut state = match strategy.init_reconcile_state(rc).in_current_span().await {
         Ok(state) => state,
@@ -373,38 +378,3 @@ where
     trace!(%err, "Reconciliation error, calling strategy error_policy");
     context.get_ref().strategy.error_policy()
 }
-
-async fn handle_deletion<T>(
-    resource: &T,
-    client: Client,
-    finalizer_name: &str,
-) -> OperatorResult<ReconcileFunctionAction>
-where
-    T: Clone + DeserializeOwned + Meta + Send + Sync + 'static,
-{
-    trace!("[handle_deletion] Begin");
-    if !finalizer::has_deletion_stamp(resource) {
-        debug!("Resource not deleted, continuing",);
-        return Ok(ReconcileFunctionAction::Continue);
-    }
-
-    info!("Removing finalizer [{}]", finalizer_name,);
-    finalizer::remove_finalizer(client, resource, finalizer_name).await?;
-
-    Ok(ReconcileFunctionAction::Requeue(Duration::from_secs(10)))
-}
-
-async fn add_finalizer<T>(resource: &T, client: Client, finalizer_name: &str) -> OperatorResult<()>
-where
-    T: Clone + Debug + DeserializeOwned + Meta + Send + Sync + 'static,
-{
-    trace!("[add_finalizer] Begin");
-
-    if finalizer::has_finalizer(resource, finalizer_name) {
-        debug!("Finalizer already exists, continuing...",);
-    } else {
-        debug!("Finalizer missing, adding now and continuing...",);
-        finalizer::add_finalizer(client, resource, finalizer_name).await?;
-    }
-    Ok(())
-}

src/finalizer.rs

@@ -1,10 +1,11 @@
 use crate::client::Client;
 use crate::error::{Error, OperatorResult};
 
-use k8s_openapi::Resource;
+use json_patch::{PatchOperation, RemoveOperation, TestOperation};
 use kube::api::Meta;
 use serde::de::DeserializeOwned;
 use serde_json::json;
+use tracing::debug;
 
 /// Checks whether our own finalizer is in the list of finalizers for the provided object.
 pub fn has_finalizer<T>(resource: &T, finalizer: &str) -> bool
@@ -17,25 +18,46 @@ where
     };
 }
 
-/// Adds our finalizer to the list of finalizers.
-pub async fn add_finalizer<T>(client: Client, resource: &T, finalizer: &str) -> OperatorResult<T>
+/// This will add the passed finalizer to the list of finalizers for the resource if it doesn't exist yet
+/// and will update the resource in Kubernetes.
+///
+/// It'll return `true` if we changed the object in Kubernetes and `false` if no modification was needed.
+/// If the object is currently being deleted this _will_ return an Error!
+pub async fn add_finalizer<T>(
+    client: &Client,
+    resource: &T,
+    finalizer: &str,
+) -> OperatorResult<bool>
 where
-    T: Resource + Clone + Meta + DeserializeOwned,
+    T: Clone + Meta + DeserializeOwned,
 {
+    if has_finalizer(resource, finalizer) {
+        debug!("Finalizer [{}] already exists, continuing...", finalizer);
+
+        return Ok(false);
+    }
+
     let new_metadata = json!({
         "metadata": {
             "finalizers": [finalizer.to_string()]
         }
     });
-    client.merge_patch(resource, new_metadata).await
+    client.merge_patch(resource, new_metadata).await?;
+    Ok(true)
 }
 
 /// Removes our finalizer from a resource object.
 ///
 /// # Arguments
-/// `name` - is the name of the resource we want to patch
-/// `namespace` is the namespace of where the resource to patch lives
-pub async fn remove_finalizer<T>(client: Client, resource: &T, finalizer: &str) -> OperatorResult<T>
+///
+/// * `client` - The Client to access Kubernetes
+/// * `resource` - is the resource we want to remove the finalizer from
+/// * `finalizer` - this is the actual finalizer string that we want to remove
+pub async fn remove_finalizer<T>(
+    client: &Client,
+    resource: &T,
+    finalizer: &str,
+) -> OperatorResult<T>
 where
     T: Clone + DeserializeOwned + Meta,
 {
@@ -51,7 +73,7 @@ where
         None => Err(Error::MissingObjectKey {
             key: ".metadata.finalizers",
         }),
-        Some(mut finalizers) => {
+        Some(finalizers) => {
             let index = finalizers
                 .iter()
                 .position(|cur_finalizer| cur_finalizer == finalizer);
@@ -60,14 +82,18 @@ where
                 // We found our finalizer which means that we now need to handle our deletion logic
                 // And then remove the finalizer from the list.
 
-                finalizers.swap_remove(index);
-                let new_metadata = json!({
-                    "metadata": {
-                        "finalizers": finalizers
-                    }
-                });
+                let finalizer_path = format!("/metadata/finalizers/{}", index);
+                let patch = json_patch::Patch(vec![
+                    PatchOperation::Test(TestOperation {
+                        path: finalizer_path.clone(),
+                        value: finalizer.into(),
+                    }),
+                    PatchOperation::Remove(RemoveOperation {
+                        path: finalizer_path,
+                    }),
+                ]);
 
-                client.merge_patch(resource, new_metadata).await
+                client.json_patch(resource, patch).await
             } else {
                 Err(Error::MissingObjectKey {
                     key: ".metadata.finalizers",

src/reconcile.rs

@@ -1,6 +1,6 @@
 use crate::client::Client;
 use crate::error::{Error, OperatorResult};
-use crate::{conditions, controller_ref, podutils};
+use crate::{conditions, controller_ref, finalizer, podutils};
 
 use crate::conditions::ConditionStatus;
 use k8s_openapi::api::core::v1::Pod;
@@ -9,7 +9,9 @@ use kube::api::{ListParams, Meta, ObjectMeta};
 use kube_runtime::controller::ReconcilerAction;
 use serde::de::DeserializeOwned;
 use std::future::Future;
+use std::pin::Pin;
 use std::time::Duration;
+use tracing::{debug, info};
 
 pub type ReconcileResult<E> = std::result::Result<ReconcileFunctionAction, E>;
 
@@ -58,11 +60,20 @@ pub fn create_requeuing_reconcile_function_action(secs: u64) -> ReconcileFunctio
 pub struct ReconciliationContext<T> {
     pub client: Client,
     pub resource: T,
+    pub requeue_timeout: Duration,
 }
 
 impl<T> ReconciliationContext<T> {
-    pub fn new(client: Client, resource: T) -> Self {
-        ReconciliationContext { client, resource }
+    pub fn new(client: Client, resource: T, requeue_timeout: Duration) -> Self {
+        ReconciliationContext {
+            client,
+            resource,
+            requeue_timeout,
+        }
+    }
+
+    fn requeue(&self) -> ReconcileFunctionAction {
+        ReconcileFunctionAction::Requeue(self.requeue_timeout)
     }
 }
 
@@ -122,6 +133,63 @@ where
             })
     }
 
+    /// This reconcile function can be added to the chain to automatically handle deleted objects
+    /// using finalizers.
+    ///
+    /// It'll add a finalizer to the object if it's not there yet, if the `deletion_timestamp` is set
+    /// it'll call the provided handler function and it'll remove the finalizer if the handler completes
+    /// with a `Done` result.
+    ///
+    /// If the object is not deleted this function will return a `Continue` event.
+    ///
+    /// # Arguments
+    ///
+    /// * `handler` - This future will be completed if the object has been marked for deletion
+    /// * `finalizer` - The finalizer to add and/or check for
+    /// * `requeue_if_changed` - If this is `true` we'll return a `Requeue` immediately if we had to
+    ///     change the resource due to the addition of the finalizer
+    pub async fn handle_deletion(
+        &self,
+        handler: Pin<Box<dyn Future<Output = Result<ReconcileFunctionAction, Error>> + Send + '_>>,
+        finalizer: &str,
+        requeue_if_changed: bool,
+    ) -> ReconcileResult<Error>
+    where
+        T: Clone + DeserializeOwned + Meta + Send + Sync + 'static,
+    {
+        let being_deleted = finalizer::has_deletion_stamp(&self.resource);
+
+        // Try to add a finalizer but only if the deletion_timestamp is not already set
+        // Kubernetes forbids setting new finalizers on objects under deletion and will return this error:
+        // Forbidden: no new finalizers can be added if the object is being deleted, found new finalizers []string{\"foo\"}
+        if !being_deleted
+            && finalizer::add_finalizer(&self.client, &self.resource, finalizer).await?
+            && requeue_if_changed
+        {
+            return Ok(self.requeue());
+        }
+
+        if !being_deleted {
+            debug!("Resource not deleted, continuing",);
+            return Ok(ReconcileFunctionAction::Continue);
+        }
+
+        if !finalizer::has_finalizer(&self.resource, finalizer) {
+            debug!("Resource being deleted but our finalizer is already gone, there might be others but we're done here!");
+            return Ok(ReconcileFunctionAction::Done);
+        }
+
+        match handler.await? {
+            ReconcileFunctionAction::Continue => Ok(ReconcileFunctionAction::Continue),
+            ReconcileFunctionAction::Done => {
+                info!("Removing finalizer [{}]", finalizer,);
+                finalizer::remove_finalizer(&self.client, &self.resource, finalizer).await?;
+                Ok(ReconcileFunctionAction::Done)
+            }
+            ReconcileFunctionAction::Requeue(_) => Ok(self.requeue()),
+        }
+    }
+
     /// Creates a new [`Condition`] for the `resource` this context contains.
     ///
     /// It's a convenience function that passes through all parameters and builds a `Condition`
@@ -173,6 +241,26 @@ where
         );
         self.set_condition(condition).await
     }
+
+    /// A reconciler function to add to our finalizer to the list of finalizers.
+    /// It is a wrapper around [`finalizer::add_finalizer`].
+    ///
+    /// It can return `Continue` or `Requeue` depending on the `requeue` argument and the state of the resource.
+    /// If the finalizer already exists it'll _always_ return `Continue`.
+    ///
+    /// There is a more full-featured alternative to this function ([`handle_deletion`]).
+    ///
+    /// # Arguments
+    ///
+    /// * `finalizer` - The finalizer to add
+    /// * `requeue` - If `true` this function will return `Requeue` if the object was changed (i.e. the finalizer was added) otherwise it'll return `Continue`
+    pub async fn add_finalizer(&self, finalizer: &str, requeue: bool) -> ReconcileResult<Error> {
+        if finalizer::add_finalizer(&self.client, &self.resource, finalizer).await? && requeue {
+            Ok(self.requeue())
+        } else {
+            Ok(ReconcileFunctionAction::Continue)
+        }
+    }
 }
 
 /// This returns `false` for Pods that have no OwnerReference (with a Controller flag)