Allow cleaning up old replicasets while the new one is not ready (#59)

Author: jackkleeman

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "restate-operator"
-version = "1.8.1"
+version = "1.8.2"
 authors = ["restate.dev"]
 edition = "2021"
 rust-version = "1.86"

Cargo.toml

@@ -2,4 +2,4 @@ apiVersion: v2
 name: restate-operator-helm
 description: An operator for Restate clusters
 type: application
-version: "1.8.1"
+version: "1.8.2"

charts/restate-operator-helm/Chart.yaml

@@ -330,13 +330,44 @@ impl RestateDeployment {
                 .cloned()
                 .unwrap_or_default()
         }) {
-            if let Some(cluster_name) = &self.spec.restate.register.cluster {
-                // wait for the cluster to be ready before registering to it
-                validate_cluster_status(rsc_api, cluster_name).await?;
-            };
+            let valid = async {
+                if let Some(cluster_name) = &self.spec.restate.register.cluster {
+                    // wait for the cluster to be ready before registering to it
+                    validate_cluster_status(rsc_api, cluster_name).await?;
+                }
 
-            // wait for the replicaset to be ready before registering it
-            validate_replica_set_status(replicaset.status.as_ref(), self.spec.replicas)?;
+                // wait for the replicaset to be ready before registering it
+                validate_replica_set_status(replicaset.status.as_ref(), self.spec.replicas)?;
+
+                Ok(())
+            }
+            .await;
+
+            match valid {
+                Ok(()) => {}
+                // there is a chicken and egg situation if the cluster is out of capacity; the new version can't become ready until
+                // old versions are removed. so we remove them aggressively here
+                Err(ready_err @ Error::DeploymentNotReady { .. }) => {
+                    match reconcilers::replicaset::cleanup_old_replicasets(
+                        namespace,
+                        &ctx,
+                        &rs_api,
+                        &my_uid,
+                        self,
+                        &deployments,
+                        Some(&versioned_name), // exclude the replicaset which may not be registered
+                    )
+                    .await
+                    {
+                        Ok((_, _)) => return Err(ready_err),
+                        Err(cleanup_err) => {
+                            error!("Failed to clean up old replicasets while waiting for current replicaset to become ready: {cleanup_err}");
+                            return Err(ready_err);
+                        }
+                    }
+                }
+                Err(err) => return Err(err),
+            }
 
             // Register the latest version with Restate cluster using the service URL
             let deployment_id = self
@@ -385,6 +416,7 @@ impl RestateDeployment {
             &my_uid,
             self,
             &deployments,
+            Some(&versioned_name),
         )
         .await?;
 
@@ -667,6 +699,7 @@ impl RestateDeployment {
             &my_uid,
             self,
             &deployments,
+            None,
         )
         .await?;
 

src/controllers/restatedeployment/controller.rs

@@ -162,15 +162,29 @@ pub async fn cleanup_old_replicasets(
     rsd_uid: &str,
     rsd: &RestateDeployment,
     deployments: &HashMap<String, bool>,
+    except_rs: Option<&str>,
 ) -> Result<(i32, Option<chrono::DateTime<chrono::Utc>>)> {
     let replicasets_cell = std::cell::Cell::new(Vec::new());
 
     let _ = ctx.replicasets_store.find(|rs| {
+        let rs_namespace = match &rs.metadata.namespace.as_deref() {
+            Some("") | None => "default",
+            Some(ns) => ns,
+        };
         // replicasets in the same ns
-        if rs.metadata.namespace.as_deref() != Some(namespace) {
+        if rs_namespace != namespace {
             return false;
         }
 
+        // not the current version if we are actively trying to register it
+        if let Some(except_rs) = except_rs {
+            let rs_name = rs.name_any();
+
+            if rs_name == except_rs {
+                return false;
+            }
+        }
+
         // replicasets owned by this restatedeployment (we make no attempt to handle orphaned ones if a rsd was deleted with --cascade=orphan and then recreated)
         if !rs.owner_references().iter().any(|reference| {
             reference.uid == rsd_uid && reference.kind == RestateDeployment::kind(&())