Implemented a node replace flow

instaclustr · Dec 21, 2023 · a225611 · a225611
1 parent 46b12b2
commit a225611
Show file tree

Hide file tree

Showing 5 changed files with 457 additions and 247 deletions.
diff --git a/config/samples/onpremises/clusters_v1beta1_cassandra.yaml b/config/samples/onpremises/clusters_v1beta1_cassandra.yaml
@@ -1,12 +1,15 @@
 apiVersion: clusters.instaclustr.com/v1beta1
 kind: Cassandra
 metadata:
-  name: cassandra-on-prem-cluster
+  name: cassandra-on-prem-cluster-test
 spec:
-  name: "danylo-on-prem-cassandra"
+  name: "rostyslp-on-prem-cassandra"
   version: "4.0.10"
   privateNetworkCluster: false
   onPremisesSpec:
+    cloudInitScriptRef:
+      name: instaclustr-cloud-init-secret
+      namespace: default
     storageClassName: managed-csi-premium
     osDiskSize: 20Gi
     dataDiskSize: 200Gi
@@ -15,9 +18,6 @@ spec:
     nodeCPU: 2
     nodeMemory: 8192Mi
     osImageURL: "https://s3.amazonaws.com/debian-bucket/debian-11-generic-amd64-20230601-1398.raw"
-    cloudInitScriptRef:
-      namespace: default
-      name: instaclustr-cloud-init-secret
   dataCentres:
     - name: "onPremCassandra"
       region: "CLIENT_DC"

diff --git a/controllers/clusters/cassandra_controller.go b/controllers/clusters/cassandra_controller.go
@@ -800,7 +800,58 @@ func (r *CassandraReconciler) newWatchStatusJob(c *v1beta1.Cassandra) scheduler.
 			return err
 		}
 
+		// TODO move everything not related to the watch status job to other places
 		if !areStatusesEqual(&iCassandra.Status.ClusterStatus, &c.Status.ClusterStatus) {
+			if len(c.Status.DataCentres) != 0 &&
+				len(iCassandra.Status.DataCentres) != 0 {
+				bootstrap := newOnPremisesBootstrap(
+					r.Client,
+					c,
+					r.EventRecorder,
+					iCassandra.Status.ClusterStatus,
+					c.Spec.OnPremisesSpec,
+					newExposePorts(c.GetExposePorts()),
+					c.GetHeadlessPorts(),
+					c.Spec.PrivateNetworkCluster,
+				)
+				newNode, oldNode := getReplacedNodes(c.Status.DataCentres[0], iCassandra.Status.DataCentres[0])
+				if oldNode != nil {
+					deleteUsed, err := deleteReplacedNodeResources(context.TODO(), r.Client, oldNode.ID)
+					if err != nil {
+						l.Error(err, "Cannot delete replaced node resources",
+							"old node id", oldNode.ID)
+
+						return err
+					}
+
+					if deleteUsed {
+						l.Info("Node replace is in progress. Deleting old vm resources",
+							"cluster name", c.Name,
+							"old id", oldNode.ID)
+
+						return nil
+					}
+
+					l.Info("Node replace is in progress. Creating new resources",
+						"bootstrap", bootstrap,
+						"new node", newNode)
+
+					err = handleNodeReplace(
+						context.TODO(),
+						bootstrap,
+						oldNode,
+						newNode)
+					if err != nil {
+						l.Error(err, "Error while handling node replace operation",
+							"bootstrap struct", bootstrap,
+							"old node", oldNode,
+							"new node", newNode)
+
+						return err
+					}
+				}
+			}
+
 			l.Info("Updating cluster status",
 				"status from Instaclustr", iCassandra.Status.ClusterStatus,
 				"status from k8s", c.Status.ClusterStatus)