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
+}