Consider DC-level config when validating numToken updates in webhook (f…

…ixes #1222) (#1358)

CHANGELOG/CHANGELOG-1.18.md

@@ -15,4 +15,5 @@ When cutting a new release, update the `unreleased` heading to the tag being gen
 
 ## unreleased
 
-* [FEATURE] [#1310](https://github.com/k8ssandra/k8ssandra-operator/issues/1310) Enhance the MedusaBackupSchedule API to allow scheduling purge tasks
+* [FEATURE] [#1310](https://github.com/k8ssandra/k8ssandra-operator/issues/1310) Enhance the MedusaBackupSchedule API to allow scheduling purge tasks
+* [BUGFIX] [#1222](https://github.com/k8ssandra/k8ssandra-operator/issues/1222) Consider DC-level config when validating numToken updates in webhook

apis/k8ssandra/v1alpha1/k8ssandracluster_types.go

@@ -17,7 +17,6 @@ limitations under the License.
 package v1alpha1
 
 import (
-	"github.com/Masterminds/semver/v3"
 	cassdcapi "github.com/k8ssandra/cass-operator/apis/cassandra/v1beta1"
 	medusaapi "github.com/k8ssandra/k8ssandra-operator/apis/medusa/v1alpha1"
 	reaperapi "github.com/k8ssandra/k8ssandra-operator/apis/reaper/v1alpha1"
@@ -542,13 +541,6 @@ func (sd *ServerDistribution) IsDse() bool {
 	return *sd == ServerDistributionDse
 }
 
-func (kc *K8ssandraCluster) DefaultNumTokens(serverVersion *semver.Version) float64 {
-	if kc.Spec.Cassandra.ServerType.IsCassandra() && serverVersion.Major() == 3 {
-		return float64(256)
-	}
-	return float64(16)
-}
-
 // GetClusterIdHash should be used to derive short form unique identifiers for the cluster,
 // this is to be used to name resources that are cluster specific in preference of concatenations
 // of the namespaced name, as the latter are becoming too long and causing issues due to DNS name length

apis/k8ssandra/v1alpha1/k8ssandracluster_webhook.go

@@ -18,11 +18,11 @@ package v1alpha1
 
 import (
 	"fmt"
+	"github.com/Masterminds/semver/v3"
 	"strings"
 
 	"k8s.io/apimachinery/pkg/util/validation"
 
-	"github.com/Masterminds/semver/v3"
 	"github.com/k8ssandra/k8ssandra-operator/pkg/clientcache"
 	"github.com/pkg/errors"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -150,26 +150,8 @@ func (r *K8ssandraCluster) ValidateUpdate(old runtime.Object) (admission.Warning
 		}
 	}
 
-	oldCassConfig := oldCluster.Spec.Cassandra.DatacenterOptions.CassandraConfig
-	newCassConfig := r.Spec.Cassandra.DatacenterOptions.CassandraConfig
-	if oldCassConfig != nil && newCassConfig != nil {
-		oldNumTokens, oldNumTokensExists := oldCassConfig.CassandraYaml["num_tokens"]
-		newNumTokens, newNumTokensExists := newCassConfig.CassandraYaml["num_tokens"]
-
-		if !oldNumTokensExists {
-			cassVersion, err := semver.NewVersion(oldCluster.Spec.Cassandra.ServerVersion)
-			if err != nil {
-				return nil, err
-			}
-			defaultNumTokens := oldCluster.DefaultNumTokens(cassVersion)
-			if newNumTokensExists && newNumTokens.(float64) != defaultNumTokens {
-				return nil, ErrNumTokens
-			}
-		} else {
-			if oldNumTokens != newNumTokens {
-				return nil, ErrNumTokens
-			}
-		}
+	if err := validateUpdateNumTokens(oldCluster.Spec.Cassandra, r.Spec.Cassandra); err != nil {
+		return nil, err
 	}
 
 	// Verify that the cluster name override was not changed
@@ -192,6 +174,71 @@ func (r *K8ssandraCluster) ValidateUpdate(old runtime.Object) (admission.Warning
 	return nil, nil
 }
 
+func validateUpdateNumTokens(
+	oldCassandra *CassandraClusterTemplate,
+	newCassandra *CassandraClusterTemplate,
+) error {
+	oldNumTokensPerDc, err := numTokensPerDc(oldCassandra)
+	if err != nil {
+		return err
+	}
+	newNumTokensPerDc, err := numTokensPerDc(newCassandra)
+	if err != nil {
+		return err
+	}
+
+	for dcName, newNumTokens := range newNumTokensPerDc {
+		oldNumTokens, oldExists := oldNumTokensPerDc[dcName]
+		if oldExists && oldNumTokens != newNumTokens {
+			return ErrNumTokens
+		}
+	}
+	return nil
+}
+
+func numTokensPerDc(cassandra *CassandraClusterTemplate) (map[string]interface{}, error) {
+	var globalNumTokens interface{}
+	globalConfig := cassandra.DatacenterOptions.CassandraConfig
+	if globalConfig != nil {
+		globalNumTokens = globalConfig.CassandraYaml["num_tokens"]
+	}
+
+	numTokensPerDc := make(map[string]interface{})
+	for _, dc := range cassandra.Datacenters {
+		var numTokens interface{}
+		// Try to set from DC config
+		config := dc.DatacenterOptions.CassandraConfig
+		if config != nil {
+			numTokens = config.CassandraYaml["num_tokens"]
+		}
+		// Otherwise, try from global config
+		if numTokens == nil {
+			numTokens = globalNumTokens
+		}
+		// Otherwise, use version-specific default
+		if numTokens == nil {
+			versionString := dc.ServerVersion
+			if versionString == "" {
+				versionString = cassandra.ServerVersion
+			}
+			if versionString == "" {
+				return nil, errors.New("serverVersion should be set globally or at DC level")
+			}
+			version, err := semver.NewVersion(versionString)
+			if err != nil {
+				return nil, err
+			}
+			if cassandra.ServerType.IsCassandra() && version.Major() == 3 {
+				numTokens = float64(256)
+			} else {
+				numTokens = float64(16)
+			}
+		}
+		numTokensPerDc[dc.Meta.Name] = numTokens
+	}
+	return numTokensPerDc, nil
+}
+
 // ValidateDelete implements webhook.Validator so a webhook will be registered for the type
 func (r *K8ssandraCluster) ValidateDelete() (admission.Warnings, error) {
 	webhookLog.Info("validate K8ssandraCluster delete", "name", r.Name)

apis/k8ssandra/v1alpha1/k8ssandracluster_webhook_test.go

@@ -216,6 +216,7 @@ func testStorageConfigValidation(t *testing.T) {
 	required := require.New(t)
 	createNamespace(required, "storage-namespace")
 	cluster := createMinimalClusterObj("storage-test", "storage-namespace")
+	cluster.Spec.Cassandra.DatacenterOptions.ServerVersion = "3.11.10"
 
 	cluster.Spec.Cassandra.DatacenterOptions.StorageConfig = nil
 	err := k8sClient.Create(ctx, cluster)
@@ -493,3 +494,86 @@ func testMedusaNonLocalNamespace(t *testing.T) {
 	required.Error(err)
 	required.Contains(err.Error(), "Medusa config must be namespace local")
 }
+
+// TestValidateUpdateNumTokens is a unit test for numTokens updates.
+func TestValidateUpdateNumTokens(t *testing.T) {
+	type config struct {
+		globalVersion   string
+		dcVersion       string
+		globalNumTokens int
+		dcNumTokens     int
+	}
+	type testCase struct {
+		oldConfig config
+		newConfig config
+		expected  error
+	}
+	testCases := []testCase{
+		{
+			oldConfig: config{globalVersion: "3.11.10"},
+			newConfig: config{globalVersion: "4.1.3"},
+			expected:  ErrNumTokens,
+		},
+		{
+			oldConfig: config{globalVersion: "3.11.10"},
+			newConfig: config{globalNumTokens: 256},
+			expected:  nil,
+		},
+		{
+			oldConfig: config{globalVersion: "3.11.10"},
+			newConfig: config{dcNumTokens: 256},
+			expected:  nil,
+		},
+		{
+			oldConfig: config{dcVersion: "3.11.10"},
+			newConfig: config{globalNumTokens: 256},
+			expected:  nil,
+		},
+		{
+			oldConfig: config{dcVersion: "3.11.10"},
+			newConfig: config{dcNumTokens: 256},
+			expected:  nil,
+		},
+		{
+			oldConfig: config{dcNumTokens: 10},
+			newConfig: config{dcNumTokens: 10},
+			expected:  nil,
+		},
+		{
+			oldConfig: config{globalNumTokens: 10},
+			newConfig: config{dcNumTokens: 10},
+			expected:  nil,
+		},
+	}
+	toCassandra := func(c config) *CassandraClusterTemplate {
+		cassandra := &CassandraClusterTemplate{ServerType: ServerDistributionCassandra}
+		options := DatacenterOptions{ServerVersion: c.globalVersion}
+		if c.globalNumTokens != 0 {
+			options.CassandraConfig = &CassandraConfig{
+				CassandraYaml: unstructured.Unstructured{"num_tokens": float64(c.globalNumTokens)}}
+		}
+		cassandra.DatacenterOptions = options
+		dc := CassandraDatacenterTemplate{
+			Meta: EmbeddedObjectMeta{Name: "dc1"},
+		}
+		dcOptions := DatacenterOptions{ServerVersion: c.dcVersion}
+		if c.dcNumTokens != 0 {
+			dcOptions.CassandraConfig = &CassandraConfig{
+				CassandraYaml: unstructured.Unstructured{"num_tokens": float64(c.dcNumTokens)}}
+		}
+		dc.DatacenterOptions = dcOptions
+		cassandra.Datacenters = []CassandraDatacenterTemplate{dc}
+		return cassandra
+	}
+	for _, testCase := range testCases {
+		oldCassandra := toCassandra(testCase.oldConfig)
+		newCassandra := toCassandra(testCase.newConfig)
+		err := validateUpdateNumTokens(oldCassandra, newCassandra)
+		if testCase.expected == nil {
+			require.NoError(t, err)
+		} else {
+			require.Error(t, err)
+			require.Equal(t, testCase.expected.Error(), err.Error())
+		}
+	}
+}