WIP: Allow updating cassandra nodepool resource requests

pkg/apis/navigator/types.go

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

pkg/apis/navigator/v1alpha1/types.go

@@ -114,6 +114,9 @@ type CassandraClusterStatus struct {
 type CassandraClusterNodePoolStatus struct {
 	// The number of replicas in the node pool that are currently 'Ready'.
 	ReadyReplicas int32 `json:"readyReplicas"`
+
+	// The applied resource requirements for this nodepool
+	Resources v1.ResourceRequirements `json:"resources,omitempty"`
 }
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

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

@@ -197,6 +197,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.Resources = in.Resources
 	return nil
 }
 
@@ -207,6 +208,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.Resources = in.Resources
 	return nil
 }
 

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

@@ -152,6 +152,7 @@ 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
+	in.Resources.DeepCopyInto(&out.Resources)
 	return
 }
 
@@ -206,7 +207,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

@@ -76,11 +76,15 @@ func ValidateCassandraClusterUpdate(old, new *navigator.CassandraCluster) field.
 				restorePersistence := newNp.Persistence
 				newNp.Persistence = oldNp.Persistence
 
+				restoreResources := newNp.Resources
+				newNp.Resources = oldNp.Resources
+
 				if !reflect.DeepEqual(newNp, oldNp) {
 					allErrs = append(allErrs, field.Forbidden(field.NewPath("spec"), "updates to nodepool for fields other than 'replicas' and 'persistence' are forbidden."))
 				}
 				newNp.Replicas = restoreReplicas
 				newNp.Persistence = restorePersistence
+				newNp.Resources = restoreResources
 
 				break
 			}

pkg/apis/navigator/zz_generated.deepcopy.go

@@ -152,6 +152,7 @@ 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
+	in.Resources.DeepCopyInto(&out.Resources)
 	return
 }
 
@@ -206,7 +207,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/controllers/cassandra/actions/setresources.go

@@ -0,0 +1,83 @@
+package actions
+
+import (
+	"fmt"
+
+	corev1 "k8s.io/api/core/v1"
+
+	"github.com/golang/glog"
+	"github.com/pkg/errors"
+
+	"github.com/jetstack/navigator/pkg/apis/navigator/v1alpha1"
+	"github.com/jetstack/navigator/pkg/controllers"
+	"github.com/jetstack/navigator/pkg/controllers/cassandra/nodepool"
+	"github.com/jetstack/navigator/pkg/util/resources"
+)
+
+type SetResources struct {
+	Cluster  *v1alpha1.CassandraCluster
+	NodePool *v1alpha1.CassandraClusterNodePool
+}
+
+var _ controllers.Action = &SetResources{}
+
+func (a *SetResources) Name() string {
+	return "SetResources"
+}
+
+func (a *SetResources) Execute(s *controllers.State) error {
+	baseSet := nodepool.StatefulSetForCluster(a.Cluster, a.NodePool)
+	existingSet, err := s.StatefulSetLister.
+		StatefulSets(baseSet.Namespace).Get(baseSet.Name)
+	if err != nil {
+		return errors.Wrap(err, "unable to find statefulset")
+	}
+
+	var cassContainerIndex int
+	var container *corev1.Container
+	for i, _ := range existingSet.Spec.Template.Spec.Containers {
+		if existingSet.Spec.Template.Spec.Containers[i].Name == "cassandra" {
+			cassContainerIndex = i
+			container = &existingSet.Spec.Template.Spec.Containers[i]
+		}
+	}
+
+	if container == nil {
+		return fmt.Errorf("unable to find cassandra container in StatefulSet %s/%s",
+			existingSet.Namespace, existingSet.Name,
+		)
+	}
+
+	if resources.RequirementsEqual(container.Resources, a.NodePool.Resources) {
+		glog.V(4).Infof(
+			"SetResources not necessary because StatefulSet '%s/%s' "+
+				"already has the desired resources value: %v",
+			existingSet.Namespace, existingSet.Name,
+			container.Resources,
+		)
+		return nil
+	}
+
+	newSet := existingSet.DeepCopy()
+	newSet.Spec.Template.Spec.Containers[cassContainerIndex].Resources = a.NodePool.Resources
+	glog.V(4).Infof(
+		"Setting cassandra resources %s/%s from %v to %v",
+		newSet.Namespace, newSet.Name,
+		existingSet.Spec.Template.Spec.Containers[cassContainerIndex].Resources,
+		a.NodePool.Resources,
+	)
+	_, err = s.Clientset.AppsV1beta1().
+		StatefulSets(newSet.Namespace).Update(newSet)
+	if err != nil {
+		return errors.Wrap(err, "unable to update statefulset resources")
+	}
+	s.Recorder.Eventf(
+		a.Cluster,
+		corev1.EventTypeNormal,
+		a.Name(),
+		"SetResources: NodePool=%s/%s/%s, Resources=%v",
+		a.Cluster.Namespace, a.Cluster.Name, a.NodePool.Name,
+		a.NodePool.Resources,
+	)
+	return nil
+}

pkg/controllers/cassandra/cluster_control.go

@@ -15,6 +15,7 @@ import (
 	"github.com/jetstack/navigator/pkg/controllers/cassandra/seedlabeller"
 	"github.com/jetstack/navigator/pkg/controllers/cassandra/service"
 	"github.com/jetstack/navigator/pkg/controllers/cassandra/serviceaccount"
+	"github.com/jetstack/navigator/pkg/util/resources"
 )
 
 const (
@@ -212,6 +213,16 @@ func NextAction(c *v1alpha1.CassandraCluster) controllers.Action {
 				NodePool: &np,
 			}
 		}
+
+		if !resources.RequirementsEqual(np.Resources, nps.Resources) {
+			return &actions.SetResources{
+				Cluster:  c,
+				NodePool: &np,
+			}
+		} else {
+			glog.Warningf("requirementsEqual")
+		}
+
 	}
 	return nil
 }

pkg/controllers/cassandra/nodepool/nodepool.go

@@ -1,6 +1,7 @@
 package nodepool
 
 import (
+	apiv1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/kubernetes"
 	appslisters "k8s.io/client-go/listers/apps/v1beta1"
 	"k8s.io/client-go/tools/record"
@@ -48,6 +49,18 @@ func (e *defaultCassandraClusterNodepoolControl) updateStatus(cluster *v1alpha1.
 		npName := ss.Labels[v1alpha1.CassandraNodePoolNameLabel]
 		nps := cluster.Status.NodePools[npName]
 		nps.ReadyReplicas = ss.Status.ReadyReplicas
+
+		var container *apiv1.Container
+		for i, _ := range ss.Spec.Template.Spec.Containers {
+			if ss.Spec.Template.Spec.Containers[i].Name == "cassandra" {
+				container = &ss.Spec.Template.Spec.Containers[i]
+			}
+		}
+
+		if container != nil {
+			nps.Resources = container.Resources
+		}
+
 		cluster.Status.NodePools[npName] = nps
 	}
 	return nil

pkg/util/resources/resources.go

@@ -0,0 +1,25 @@
+package resources
+
+import (
+	apiv1 "k8s.io/api/core/v1"
+)
+
+func RequirementsEqual(a, b apiv1.ResourceRequirements) bool {
+	if a.Limits.Cpu().Cmp(*b.Limits.Cpu()) != 0 {
+		return false
+	}
+
+	if a.Limits.Memory().Cmp(*b.Limits.Memory()) != 0 {
+		return false
+	}
+
+	if a.Requests.Cpu().Cmp(*b.Requests.Cpu()) != 0 {
+		return false
+	}
+
+	if a.Requests.Memory().Cmp(*b.Requests.Memory()) != 0 {
+		return false
+	}
+
+	return true
+}