WIP: cassandra upgrade

docs/cassandra.rst

@@ -191,6 +191,7 @@ Supported Configuration Changes
 Navigator supports the following changes to a Cassandra cluster:
 
  * :ref:`create-cluster-cassandra`: Add all initially configured node pools and nodes.
+ * :ref:`minor-upgrade-cassandra`: Trigger a rolling upgrade of Cassandra nodes by increasing the minor and / or patch components of ``CassandraCluster.Spec.Version``.
  * :ref:`scale-out-cassandra`: Increase ``CassandraCluster.Spec.NodePools[0].Replicas`` to add more C* nodes to a ``nodepool``.
 
 Navigator does not currently support any other changes to the Cassandra cluster configuration.
@@ -200,7 +201,6 @@ Unsupported Configuration Changes
 
 The following configuration changes are not currently supported but will be supported in the near future:
 
- * Minor Upgrade: Trigger a rolling Cassandra upgrade by increasing the minor and / or patch components of ``CassandraCluster.Spec.Version``.
  * Scale In: Decrease ``CassandraCluster.Spec.NodePools[0].Replicas`` to remove C* nodes from a ``nodepool``.
 
 The following configuration changes are not currently supported:
@@ -220,6 +220,19 @@ in order of ``NodePool`` and according to the process described in :ref:`scale-o
 The order of node creation is determined by the order of the entries in the ``CassandraCluster.Spec.NodePools`` list.
 You can look at ``CassandraCluster.Status.NodePools`` to see the current state.
 
+.. _minor-upgrade-cassandra:
+
+Minor Upgrade
+~~~~~~~~~~~~~
+
+If you increment the minor or patch number in ``CassandraCluster.Spec.Version``, Navigator will trigger a rolling update of the existing C* nodes.
+
+C* nodes are upgraded serially, in order of NodePool and Pod ordinal, starting with the pod with the highest ordinal in the first NodePool.
+
+`StatefulSet Rolling Updates <https://kubernetes.io/docs/concepts/workloads/controllers/statefulset/#rolling-updates>`_ describes the update process in more detail.
+
+.. note:: Major version upgrades are not yet supported.
+
 .. _scale-out-cassandra:
 
 Scale Out

hack/e2e.sh

@@ -273,12 +273,23 @@ function test_cassandracluster() {
          stdout_equals "${CASS_VERSION}" \
          kubectl --namespace "${namespace}" \
          get pilots \
-         --output 'jsonpath={.items[0].status.cassandra.version}'
+         --selector "navigator.jetstack.io/cassandra-cluster-name=${CASS_NAME}" \
+         --output 'jsonpath={.items[*].status.cassandra.version}'
     then
         kubectl --namespace "${namespace}" get pilots -o yaml
         fail_test "Pilots failed to report the expected version"
     fi
 
+    if ! retry TIMEOUT=300 \
+         stdout_equals "${CASS_VERSION}" \
+         kubectl --namespace "${namespace}" \
+         get cassandracluster "${CASS_NAME}" \
+         --output 'jsonpath={.status.nodePools.*.version}'
+    then
+        kubectl --namespace "${namespace}" get cassandracluster -o yaml
+        fail_test "NodePools failed to report the expected version"
+    fi
+
     # Wait 5 minutes for cassandra to start and listen for CQL queries.
     if ! retry TIMEOUT=300 cql_connect \
          "${namespace}" \
@@ -310,6 +321,43 @@ function test_cassandracluster() {
         --debug \
         --execute="INSERT INTO space1.testtable1(key, value) VALUES('testkey1', 'testvalue1')"
 
+    # Upgrade to newer patch version
+    export CASS_VERSION="3.11.2"
+    kubectl apply \
+            --namespace "${namespace}" \
+            --filename \
+            <(envsubst \
+                  '$NAVIGATOR_IMAGE_REPOSITORY:$NAVIGATOR_IMAGE_TAG:$NAVIGATOR_IMAGE_PULLPOLICY:$CASS_NAME:$CASS_REPLICAS:$CASS_CQL_PORT:$CASS_VERSION' \
+                  < "${SCRIPT_DIR}/testdata/cass-cluster-test.template.yaml")
+
+    # The cluster is upgraded
+    if ! retry TIMEOUT=300 kube_event_exists "${namespace}" \
+         "navigator-controller:CassandraCluster:Normal:UpdateVersion"
+    then
+        fail_test "An UpdateVersion event was not recorded"
+    fi
+
+    if ! retry TIMEOUT=300 \
+         stdout_equals "${CASS_VERSION}" \
+         kubectl --namespace "${namespace}" \
+         get pilots \
+         --selector "navigator.jetstack.io/cassandra-cluster-name=${CASS_NAME}" \
+         --output 'jsonpath={.items[*].status.cassandra.version}'
+    then
+        kubectl --namespace "${namespace}" get pilots -o yaml
+        fail_test "Pilots failed to report the expected version"
+    fi
+
+    if ! retry TIMEOUT=300 \
+         stdout_equals "${CASS_VERSION}" \
+         kubectl --namespace "${namespace}" \
+         get cassandracluster "${CASS_NAME}" \
+         --output 'jsonpath={.status.nodePools.*.version}'
+    then
+        kubectl --namespace "${namespace}" get cassandracluster -o yaml
+        fail_test "NodePools failed to report the expected version"
+    fi
+
     # Delete the Cassandra pod and wait for the CQL service to become
     # unavailable (readiness probe fails)
 

hack/testdata/values.yaml

@@ -31,3 +31,4 @@ controller:
     repository: quay.io/jetstack/navigator-controller
     tag: build
     pullPolicy: Never
+  logLevel: 4

internal/test/util/generate/generate.go

@@ -7,6 +7,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/jetstack/navigator/pkg/apis/navigator/v1alpha1"
+	"github.com/jetstack/navigator/pkg/cassandra/version"
 )
 
 type PilotConfig struct {
@@ -121,6 +122,7 @@ func StatefulSet(c StatefulSetConfig) *apps.StatefulSet {
 
 type CassandraClusterConfig struct {
 	Name, Namespace string
+	Version         version.Version
 }
 
 func CassandraCluster(c CassandraClusterConfig) *v1alpha1.CassandraCluster {
@@ -129,6 +131,9 @@ func CassandraCluster(c CassandraClusterConfig) *v1alpha1.CassandraCluster {
 			Name:      c.Name,
 			Namespace: c.Namespace,
 		},
+		Spec: v1alpha1.CassandraClusterSpec{
+			Version: c.Version,
+		},
 	}
 }
 

pkg/apis/navigator/types.go

@@ -51,6 +51,7 @@ type CassandraClusterStatus struct {
 
 type CassandraClusterNodePoolStatus struct {
 	ReadyReplicas int32
+	Version       *version.Version
 }
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

pkg/apis/navigator/v1alpha1/types.go

@@ -82,6 +82,12 @@ type CassandraClusterStatus struct {
 
 type CassandraClusterNodePoolStatus struct {
 	ReadyReplicas int32 `json:"readyReplicas"`
+	// The lowest version of Cassandra found to be running in this nodepool,
+	// as reported by the Cassandra process.
+	// nil or empty if the lowest version can not be determined,
+	// or if the lowest version has not yet been determined
+	// +optional
+	Version *version.Version `json:"version,omitempty"`
 }
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

pkg/apis/navigator/v1alpha1/zz_generated.conversion.go

@@ -202,6 +202,7 @@ func Convert_navigator_CassandraClusterNodePool_To_v1alpha1_CassandraClusterNode
 
 func autoConvert_v1alpha1_CassandraClusterNodePoolStatus_To_navigator_CassandraClusterNodePoolStatus(in *CassandraClusterNodePoolStatus, out *navigator.CassandraClusterNodePoolStatus, s conversion.Scope) error {
 	out.ReadyReplicas = in.ReadyReplicas
+	out.Version = (*version.Version)(unsafe.Pointer(in.Version))
 	return nil
 }
 
@@ -212,6 +213,7 @@ func Convert_v1alpha1_CassandraClusterNodePoolStatus_To_navigator_CassandraClust
 
 func autoConvert_navigator_CassandraClusterNodePoolStatus_To_v1alpha1_CassandraClusterNodePoolStatus(in *navigator.CassandraClusterNodePoolStatus, out *CassandraClusterNodePoolStatus, s conversion.Scope) error {
 	out.ReadyReplicas = in.ReadyReplicas
+	out.Version = (*version.Version)(unsafe.Pointer(in.Version))
 	return nil
 }
 

pkg/apis/navigator/v1alpha1/zz_generated.deepcopy.go

@@ -116,6 +116,15 @@ func (in *CassandraClusterNodePool) DeepCopy() *CassandraClusterNodePool {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *CassandraClusterNodePoolStatus) DeepCopyInto(out *CassandraClusterNodePoolStatus) {
 	*out = *in
+	if in.Version != nil {
+		in, out := &in.Version, &out.Version
+		if *in == nil {
+			*out = nil
+		} else {
+			*out = new(version.Version)
+			**out = **in
+		}
+	}
 	return
 }
 
@@ -170,7 +179,9 @@ func (in *CassandraClusterStatus) DeepCopyInto(out *CassandraClusterStatus) {
 		in, out := &in.NodePools, &out.NodePools
 		*out = make(map[string]CassandraClusterNodePoolStatus, len(*in))
 		for key, val := range *in {
-			(*out)[key] = val
+			newVal := new(CassandraClusterNodePoolStatus)
+			val.DeepCopyInto(newVal)
+			(*out)[key] = *newVal
 		}
 	}
 	return

pkg/apis/navigator/validation/cassandra.go

@@ -1,6 +1,7 @@
 package validation
 
 import (
+	"fmt"
 	"reflect"
 
 	apimachineryvalidation "k8s.io/apimachinery/pkg/api/validation"
@@ -10,6 +11,8 @@ import (
 	"github.com/jetstack/navigator/pkg/apis/navigator"
 )
 
+var supportedMajorVersions = sets.NewInt64(3)
+
 func ValidateCassandraClusterNodePool(np *navigator.CassandraClusterNodePool, fldPath *field.Path) field.ErrorList {
 	// TODO: call k8s.io/kubernetes/pkg/apis/core/validation.ValidateResourceRequirements on np.Resources
 	// this will require vendoring kubernetes/kubernetes.
@@ -27,6 +30,26 @@ func ValidateCassandraClusterUpdate(old, new *navigator.CassandraCluster) field.
 
 	fldPath := field.NewPath("spec")
 
+	if new.Spec.Version.LessThan(&old.Spec.Version) {
+		allErrs = append(
+			allErrs,
+			field.Forbidden(
+				fldPath.Child("version"),
+				"cannot perform version downgrades",
+			),
+		)
+	}
+
+	if new.Spec.Version.Major != old.Spec.Version.Major {
+		allErrs = append(
+			allErrs,
+			field.Forbidden(
+				fldPath.Child("version"),
+				"cannot perform major version upgrades",
+			),
+		)
+	}
+
 	npPath := fldPath.Child("nodePools")
 	for i, newNp := range new.Spec.NodePools {
 		idxPath := npPath.Index(i)
@@ -60,6 +83,21 @@ func ValidateCassandraClusterUpdate(old, new *navigator.CassandraCluster) field.
 
 func ValidateCassandraClusterSpec(spec *navigator.CassandraClusterSpec, fldPath *field.Path) field.ErrorList {
 	allErrs := ValidateNavigatorClusterConfig(&spec.NavigatorClusterConfig, fldPath)
+
+	if !supportedMajorVersions.Has(spec.Version.Major) {
+		allErrs = append(
+			allErrs,
+			field.Forbidden(
+				fldPath.Child("version"),
+				fmt.Sprintf(
+					"%s is not supported. Supported major versions are: %v",
+					spec.Version,
+					supportedMajorVersions.List(),
+				),
+			),
+		)
+	}
+
 	npPath := fldPath.Child("nodePools")
 	allNames := sets.String{}
 	for i, np := range spec.NodePools {

pkg/apis/navigator/validation/cassandra_test.go

@@ -17,7 +17,7 @@ var (
 			Namespace: "bar",
 		},
 		Spec: navigator.CassandraClusterSpec{
-			Version: *version.New("5.6.2"),
+			Version: *version.New("3.11.2"),
 			Image:   &validImageSpec,
 			NavigatorClusterConfig: validNavigatorClusterConfig,
 			NodePools: []navigator.CassandraClusterNodePool{
@@ -128,6 +128,15 @@ func TestValidateCassandraClusterUpdate(t *testing.T) {
 		return c
 	}
 
+	setVersion := func(
+		c *navigator.CassandraCluster,
+		v *version.Version,
+	) *navigator.CassandraCluster {
+		c = c.DeepCopy()
+		c.Spec.Version = *v
+		return c
+	}
+
 	tests := map[string]testT{
 		"unchanged cluster": {
 			old: validCassCluster,
@@ -147,6 +156,24 @@ func TestValidateCassandraClusterUpdate(t *testing.T) {
 			old: setPersistence(validCassCluster, navigator.PersistenceConfig{Enabled: false}),
 			new: validCassCluster,
 		},
+		"decrease version": {
+			old:           setVersion(validCassCluster, validCassCluster.Spec.Version.BumpMinor()),
+			new:           validCassCluster,
+			errorExpected: true,
+		},
+		"major version upgrade": {
+			old:           validCassCluster,
+			new:           setVersion(validCassCluster, validCassCluster.Spec.Version.BumpMajor()),
+			errorExpected: true,
+		},
+		"minor version upgrade": {
+			old: validCassCluster,
+			new: setVersion(validCassCluster, validCassCluster.Spec.Version.BumpMinor()),
+		},
+		"patch version upgrade": {
+			old: validCassCluster,
+			new: setVersion(validCassCluster, validCassCluster.Spec.Version.BumpPatch()),
+		},
 	}
 
 	for title, persistence := range persistenceErrorCases {

pkg/apis/navigator/zz_generated.deepcopy.go

@@ -116,6 +116,15 @@ func (in *CassandraClusterNodePool) DeepCopy() *CassandraClusterNodePool {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *CassandraClusterNodePoolStatus) DeepCopyInto(out *CassandraClusterNodePoolStatus) {
 	*out = *in
+	if in.Version != nil {
+		in, out := &in.Version, &out.Version
+		if *in == nil {
+			*out = nil
+		} else {
+			*out = new(version.Version)
+			**out = **in
+		}
+	}
 	return
 }
 
@@ -170,7 +179,9 @@ func (in *CassandraClusterStatus) DeepCopyInto(out *CassandraClusterStatus) {
 		in, out := &in.NodePools, &out.NodePools
 		*out = make(map[string]CassandraClusterNodePoolStatus, len(*in))
 		for key, val := range *in {
-			(*out)[key] = val
+			newVal := new(CassandraClusterNodePoolStatus)
+			val.DeepCopyInto(newVal)
+			(*out)[key] = *newVal
 		}
 	}
 	return