Prevent changes of DC name or both adding and removing a DC at the sa…

…me time.

apis/k8ssandra/v1alpha1/k8ssandracluster_types.go

@@ -573,3 +573,34 @@ func (kc *K8ssandraCluster) GetClusterIdHash() string {
 func (k *K8ssandraCluster) GenerationChanged() bool {
 	return k.Status.ObservedGeneration < k.Generation
 }
+
+func DcAdded(oldSpec K8ssandraClusterSpec, newSpec K8ssandraClusterSpec) bool {
+	oldDcs := make([]string, 0)
+	for _, dc := range oldSpec.Cassandra.Datacenters {
+		oldDcs = append(oldDcs, dc.Meta.Name)
+	}
+	wasAdded := false
+	for _, dc := range newSpec.Cassandra.Datacenters {
+		if !utils.SliceContains(oldDcs, dc.Meta.Name) {
+			wasAdded = true
+			break
+		}
+	}
+	return wasAdded
+
+}
+
+func DcRemoved(oldSpec K8ssandraClusterSpec, newSpec K8ssandraClusterSpec) bool {
+	newDcs := make([]string, 0)
+	for _, dc := range newSpec.Cassandra.Datacenters {
+		newDcs = append(newDcs, dc.Meta.Name)
+	}
+	wasRemoved := false
+	for _, dc := range oldSpec.Cassandra.Datacenters {
+		if !utils.SliceContains(newDcs, dc.Meta.Name) {
+			wasRemoved = true
+			break
+		}
+	}
+	return wasRemoved
+}

apis/k8ssandra/v1alpha1/k8ssandracluster_webhook.go

@@ -18,9 +18,10 @@ package v1alpha1
 
 import (
 	"fmt"
+	"strings"
+
 	"github.com/Masterminds/semver/v3"
 	reaperapi "github.com/k8ssandra/k8ssandra-operator/apis/reaper/v1alpha1"
-	"strings"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/validation"
@@ -172,7 +173,9 @@ func (r *K8ssandraCluster) ValidateUpdate(old runtime.Object) (admission.Warning
 	if err := validateUpdateNumTokens(oldCluster.Spec.Cassandra, r.Spec.Cassandra); err != nil {
 		return nil, err
 	}
-
+	if DcRemoved(oldCluster.Spec, r.Spec) && DcAdded(oldCluster.Spec, r.Spec) {
+		return nil, fmt.Errorf("renaming, as well as adding and removing DCs at the same time is prohibited as it can cause data loss")
+	}
 	// Verify that the cluster name override was not changed
 	if r.Spec.Cassandra.ClusterName != oldCluster.Spec.Cassandra.ClusterName {
 		return nil, ErrClusterName

controllers/k8ssandra/k8ssandracluster_controller.go

@@ -119,6 +119,10 @@ func (r *K8ssandraClusterReconciler) reconcile(ctx context.Context, kc *api.K8ss
 		return recResult.Output()
 	}
 
+	if err := validateK8ssandraCluster(*kc); err != nil {
+		return reconcile.Result{}, err
+	}
+
 	if kc.Spec.Cassandra == nil {
 		// TODO handle the scenario of CassandraClusterTemplate being set to nil after having a non-nil value
 		return ctrl.Result{}, nil

controllers/k8ssandra/k8ssandracluster_controller_test.go

@@ -116,7 +116,7 @@ func TestK8ssandraCluster(t *testing.T) {
 	t.Run("ApplyClusterWithEncryptionOptionsExternalSecrets", testEnv.ControllerTest(ctx, applyClusterWithEncryptionOptionsExternalSecrets))
 	t.Run("StopDatacenter", testEnv.ControllerTest(ctx, stopDc))
 	t.Run("ConvertSystemReplicationAnnotation", testEnv.ControllerTest(ctx, convertSystemReplicationAnnotation))
-	t.Run("ChangeClusterNameFails", testEnv.ControllerTest(ctx, changeClusterNameFails))
+	t.Run("ChangeClusterDcNameFails", testEnv.ControllerTest(ctx, changeClusterDcNameFails))
 	t.Run("InjectContainersAndVolumes", testEnv.ControllerTest(ctx, injectContainersAndVolumes))
 	t.Run("CreateMultiDcDseCluster", testEnv.ControllerTest(ctx, createMultiDcDseCluster))
 	t.Run("PerNodeConfiguration", testEnv.ControllerTest(ctx, perNodeConfiguration))
@@ -2334,7 +2334,7 @@ func convertSystemReplicationAnnotation(t *testing.T, ctx context.Context, f *fr
 
 // Create a cluster with server and client encryption but client encryption stores missing.
 // Verify that dc1 never gets created.
-func changeClusterNameFails(t *testing.T, ctx context.Context, f *framework.Framework, namespace string) {
+func changeClusterDcNameFails(t *testing.T, ctx context.Context, f *framework.Framework, namespace string) {
 	require := require.New(t)
 
 	clusterName := "cluster-with-encryption"
@@ -2366,6 +2366,13 @@ func changeClusterNameFails(t *testing.T, ctx context.Context, f *framework.Fram
 						K8sContext: f.DataPlaneContexts[0],
 						Size:       dc1Size,
 					},
+					{
+						Meta: api.EmbeddedObjectMeta{
+							Name: "dc2",
+						},
+						K8sContext: f.DataPlaneContexts[0],
+						Size:       dc1Size,
+					},
 				},
 				ServerEncryptionStores: &encryption.Stores{
 					KeystoreSecretRef: &encryption.SecretKeySelector{LocalObjectReference: corev1.LocalObjectReference{
@@ -2398,6 +2405,10 @@ func changeClusterNameFails(t *testing.T, ctx context.Context, f *framework.Fram
 	err = f.Client.Get(ctx, client.ObjectKey{Namespace: namespace, Name: clusterName}, k8c)
 	require.NoError(err, "failed to get K8ssandraCluster")
 
+	k8c.Spec.Cassandra.Datacenters[0].Meta.Name = "newDC1"
+	err = f.Client.Update(ctx, k8c)
+	require.Error(err, "failed to update K8ssandraCluster")
+
 	// Change the cluster name
 	k8c.Spec.Cassandra.ClusterName = newClusterName
 	err = f.Client.Update(ctx, k8c)

controllers/k8ssandra/validate_cluster_update.go

@@ -0,0 +1,37 @@
+package k8ssandra
+
+import (
+	"fmt"
+
+	api "github.com/k8ssandra/k8ssandra-operator/apis/k8ssandra/v1alpha1"
+	"github.com/k8ssandra/k8ssandra-operator/pkg/utils"
+)
+
+func validateK8ssandraCluster(kc api.K8ssandraCluster) error {
+	oldDCs := make([]string, 0)
+	for dcName, _ := range kc.Status.Datacenters {
+		oldDCs = append(oldDCs, dcName)
+	}
+	newDcs := make([]string, 0)
+	for _, dc := range kc.Spec.Cassandra.Datacenters {
+		newDcs = append(newDcs, dc.Meta.Name)
+	}
+	wasAdded := false
+	wasRemoved := true
+	for _, dc := range kc.Spec.Cassandra.Datacenters {
+		if !utils.SliceContains(oldDCs, dc.Meta.Name) {
+			wasAdded = true
+			break
+		}
+	}
+	for dcName, _ := range kc.Status.Datacenters {
+		if !utils.SliceContains(newDcs, dcName) {
+			wasRemoved = true
+			break
+		}
+	}
+	if wasAdded && wasRemoved {
+		return fmt.Errorf("cannot add and remove datacenters at the same time")
+	}
+	return nil
+}