Asynchronous rollback fixes for Scaling

Handle scaling in/out, Scaling up/down rollback
fixes.

Author: nmvk

pkg/resource/replication_group/annotations.go

@@ -22,4 +22,14 @@ const (
 	// LogDeliveryConfigurationRequest structs passed in as input to either the create or modify API called most
 	// recently
 	AnnotationLastRequestedLDCs = svcapitypes.AnnotationPrefix + "last-requested-log-delivery-configurations"
+	// AnnotationLastRequestedCNT is an annotation whose value is passed in as input to either the create or modify API
+	// called most recently
+	AnnotationLastRequestedCNT = svcapitypes.AnnotationPrefix + "last-requested-cache-node-type"
+	// AnnotationLastRequestedNNG is an annotation whose value is passed in as input to either the create or modify API
+	// called most recently
+	AnnotationLastRequestedNNG = svcapitypes.AnnotationPrefix + "last-requested-num-node-groups"
+	// AnnotationLastRequestedNGC is an annotation whose value is the marshaled list of pointers to
+	// NodeGroupConfiguration structs passed in as input to either the create or modify API called most
+	// recently
+	AnnotationLastRequestedNGC = svcapitypes.AnnotationPrefix + "last-requested-num-group-configuration"
 )

pkg/resource/replication_group/custom_set_output.go

@@ -16,6 +16,7 @@ package replication_group
 import (
 	"context"
 	"encoding/json"
+	"strconv"
 
 	svcapitypes "github.com/aws-controllers-k8s/elasticache-controller/apis/v1alpha1"
 	ackv1alpha1 "github.com/aws-controllers-k8s/runtime/apis/core/v1alpha1"
@@ -57,6 +58,8 @@ func (rm *resourceManager) CustomCreateReplicationGroupSetOutput(
 ) (*svcapitypes.ReplicationGroup, error) {
 	rm.customSetOutput(resp.ReplicationGroup, ko)
 	rm.setAnnotationsFields(r, ko)
+	rm.setLastRequestedNodeGroupConfiguration(r, ko)
+	rm.setLastRequestedNumNodeGroups(r, ko)
 	return ko, nil
 }
 
@@ -270,20 +273,36 @@ func (rm *resourceManager) provideEvents(
 
 // setAnnotationsFields copies the desired object's annotations, populates any
 // relevant fields, and sets the latest object's annotations to this newly populated map.
+// Fields that are handled by custom modify implementation are not set here.
 // This should only be called upon a successful create or modify call.
 func (rm *resourceManager) setAnnotationsFields(
 	r *resource,
 	ko *svcapitypes.ReplicationGroup,
 ) {
+	annotations := getAnnotationsFields(r, ko)
+
+	rm.setLastRequestedLogDeliveryConfigurations(r, annotations)
+	rm.setLastRequestedCacheNodeType(r, annotations)
+	ko.ObjectMeta.Annotations = annotations
+}
+
+// getAnnotationsFields return the annotations map that would be used to set the fields
+func getAnnotationsFields(
+	r *resource,
+	ko *svcapitypes.ReplicationGroup) map[string]string {
+
+	if ko.ObjectMeta.Annotations != nil {
+		return ko.ObjectMeta.Annotations
+	}
+
 	desiredAnnotations := r.ko.ObjectMeta.GetAnnotations()
 	annotations := make(map[string]string)
 	for k, v := range desiredAnnotations {
 		annotations[k] = v
 	}
 
-	rm.setLastRequestedLogDeliveryConfigurations(r, annotations)
-
 	ko.ObjectMeta.Annotations = annotations
+	return annotations
 }
 
 // setLastRequestedLogDeliveryConfigurations copies desired.Spec.LogDeliveryConfigurations
@@ -300,3 +319,43 @@ func (rm *resourceManager) setLastRequestedLogDeliveryConfigurations(
 		annotations[AnnotationLastRequestedLDCs] = string(lastRequestedConfigs)
 	}
 }
+
+// setLastRequestedCacheNodeType copies desired.Spec.CacheNodeType into the annotation
+// of the object.
+func (rm *resourceManager) setLastRequestedCacheNodeType(
+	r *resource,
+	annotations map[string]string,
+) {
+	if r.ko.Spec.CacheNodeType != nil {
+		annotations[AnnotationLastRequestedCNT] = *r.ko.Spec.CacheNodeType
+	}
+}
+
+// setLastRequestedNodeGroupConfiguration copies desired.spec.NodeGroupConfiguration into the
+// annotation of the object
+func (rm *resourceManager) setLastRequestedNodeGroupConfiguration(
+	r *resource,
+	ko *svcapitypes.ReplicationGroup,
+) {
+	annotations := getAnnotationsFields(r, ko)
+	lastRequestedConfigs, err := json.Marshal(r.ko.Spec.NodeGroupConfiguration)
+	if err != nil {
+		annotations[AnnotationLastRequestedNGC] = "null"
+	} else {
+		annotations[AnnotationLastRequestedNGC] = string(lastRequestedConfigs)
+	}
+}
+
+// setLastRequestedNumNodeGroups copies desired.spec.NumNodeGroups into the
+// annotation of the object
+func (rm *resourceManager) setLastRequestedNumNodeGroups(
+	r *resource,
+	ko *svcapitypes.ReplicationGroup,
+) {
+	annotations := getAnnotationsFields(r, ko)
+	if r.ko.Spec.NumNodeGroups != nil {
+		annotations[AnnotationLastRequestedNNG] = strconv.Itoa(int(*r.ko.Spec.NumNodeGroups))
+	} else {
+		annotations[AnnotationLastRequestedNNG] = "null"
+	}
+}

pkg/resource/replication_group/custom_update_api.go

@@ -15,11 +15,14 @@ package replication_group
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"github.com/aws-controllers-k8s/runtime/pkg/requeue"
 	"github.com/aws/aws-sdk-go/aws/awserr"
 	"github.com/pkg/errors"
+	"reflect"
 	"sort"
+	"strconv"
 
 	svcapitypes "github.com/aws-controllers-k8s/elasticache-controller/apis/v1alpha1"
 	ackcompare "github.com/aws-controllers-k8s/runtime/pkg/compare"
@@ -70,6 +73,28 @@ func (rm *resourceManager) CustomModifyReplicationGroup(
 			requeue.DefaultRequeueAfterDuration)
 	}
 
+	// Handle the asynchronous rollback case for while Scaling down.
+	// This means that we have already attempted to apply the CacheNodeType once and
+	// were not successful hence we will set a terminal condition.
+	if !cacheNodeTypeRequiresUpdate(desired) && delta.DifferentAt("Spec.CacheNodeType") {
+		return nil, awserr.New("InvalidParameterCombination", "Cannot update CacheNodeType, "+
+			"Please refer to Events for more details", nil)
+
+	}
+
+	// Handle the asynchronous rollback for Resharding.
+	if !nodeGroupRequiresUpdate(desired) && rm.shardConfigurationsDiffer(desired, latest) {
+
+		return nil, awserr.New("InvalidParameterCombination", "Cannot update NodeGroups, "+
+			"Please refer to Events for more details", nil)
+	}
+
+	// Handle NodeGroupConfiguration asynchronous rollback situations other than Resharding.
+	if !nodeGroupRequiresUpdate(desired) && (rm.replicaCountDifference(desired, latest) != 0 && !delta.DifferentAt("Spec.ReplicasPerNodeGroup")) {
+		return nil, awserr.New("InvalidParameterCombination", "Cannot update NodeGroupConfiguration, "+
+			"Please refer to Events for more details", nil)
+	}
+
 	// Order of operations when diffs map to multiple updates APIs:
 	// 1. When automaticFailoverEnabled differs:
 	//		if automaticFailoverEnabled == false; do nothing in this custom logic, let the modify execute first.
@@ -316,7 +341,18 @@ func (rm *resourceManager) updateShardConfiguration(
 		rm.log.V(1).Info("Error during ModifyReplicationGroupShardConfiguration", "error", respErr)
 		return nil, respErr
 	}
-	return rm.setReplicationGroupOutput(desired, resp.ReplicationGroup)
+
+	r, err := rm.setReplicationGroupOutput(desired, resp.ReplicationGroup)
+
+	if err != nil {
+		return r, err
+	}
+
+	ko := r.ko.DeepCopy()
+	// Update the annotations since API call was successful
+	rm.setLastRequestedNodeGroupConfiguration(desired, ko)
+	rm.setLastRequestedNumNodeGroups(desired, ko)
+	return &resource{ko}, nil
 }
 
 // newIncreaseReplicaCountRequestPayload returns an SDK-specific struct for the HTTP request
@@ -691,3 +727,44 @@ func (rm *resourceManager) newModifyReplicationGroupRequestPayload(
 
 	return input
 }
+
+// cacheNodeTypeRequiresUpdate retrieves the last requested cacheNodeType saved in annotations and compares them
+// to the current desired cacheNodeType
+func cacheNodeTypeRequiresUpdate(desired *resource) bool {
+	annotations := desired.ko.ObjectMeta.GetAnnotations()
+	if val, ok := annotations[AnnotationLastRequestedCNT]; ok && desired.ko.Spec.CacheNodeType != nil {
+		return val != *desired.ko.Spec.CacheNodeType
+	}
+
+	// This means there is delta and no value in annotation or in Spec
+	return true
+}
+
+// nodeGroupRequiresUpdate retrieves the last applied NumNodeGroups and NodeGroupConfiguration and compares them
+// to the current desired NumNodeGroups and NodeGroupConfiguration
+func nodeGroupRequiresUpdate(desired *resource) bool {
+	annotations := desired.ko.ObjectMeta.GetAnnotations()
+
+	if val, ok := annotations[AnnotationLastRequestedNNG]; ok && val != "null" {
+		numNodes, err := strconv.ParseInt(val, 10, 64)
+
+		if err != nil {
+			return false
+		}
+
+		if numNodes != *desired.ko.Spec.NumNodeGroups {
+			return true
+		}
+
+		return false
+	}
+
+	desiredNodeGroupConfig := desired.ko.Spec.NodeGroupConfiguration
+	if val, ok := annotations[AnnotationLastRequestedNGC]; ok && val != "null" {
+		_ = json.Unmarshal([]byte(val), &desiredNodeGroupConfig)
+		return !reflect.DeepEqual(desiredNodeGroupConfig, val)
+	}
+
+	// This means there is delta and no value in annotation or in Spec
+	return true
+}

pkg/resource/replication_group/custom_update_api_test.go

@@ -25,6 +25,7 @@ import (
 	"go.uber.org/zap/zapcore"
 	"path/filepath"
 	ctrlrtzap "sigs.k8s.io/controller-runtime/pkg/log/zap"
+	"strconv"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -290,8 +291,7 @@ func TestCustomModifyReplicationGroup(t *testing.T) {
 		desired := provideResource()
 		latest := provideResource()
 		var delta ackcompare.Delta
-		var ctx context.Context
-		res, err := rm.CustomModifyReplicationGroup(ctx, desired, latest, &delta)
+		res, err := rm.CustomModifyReplicationGroup(context.TODO(), desired, latest, &delta)
 		assert.Nil(res)
 		assert.Nil(err)
 	})
@@ -306,8 +306,7 @@ func TestCustomModifyReplicationGroup_Unavailable(t *testing.T) {
 		desired := provideResource()
 		latest := provideResourceWithStatus("modifying")
 		var delta ackcompare.Delta
-		var ctx context.Context
-		res, err := rm.CustomModifyReplicationGroup(ctx, desired, latest, &delta)
+		res, err := rm.CustomModifyReplicationGroup(context.TODO(), desired, latest, &delta)
 		assert.Nil(res)
 		assert.NotNil(err)
 		var requeueNeededAfter *requeue.RequeueNeededAfter
@@ -329,8 +328,7 @@ func TestCustomModifyReplicationGroup_NodeGroup_Unvailable(t *testing.T) {
 			nodeGroup.Status = &unavailableStatus
 		}
 		var delta ackcompare.Delta
-		var ctx context.Context
-		res, err := rm.CustomModifyReplicationGroup(ctx, desired, latest, &delta)
+		res, err := rm.CustomModifyReplicationGroup(context.TODO(), desired, latest, &delta)
 		assert.Nil(res)
 		assert.NotNil(err)
 		var requeueNeededAfter *requeue.RequeueNeededAfter
@@ -357,9 +355,8 @@ func TestCustomModifyReplicationGroup_NodeGroup_MemberClusters_mismatch(t *testi
 			nodeGroup.Status = &availableStatus
 		}
 		var delta ackcompare.Delta
-		var ctx context.Context
 		require.NotNil(latest.ko.Status.MemberClusters)
-		res, err := rm.CustomModifyReplicationGroup(ctx, desired, latest, &delta)
+		res, err := rm.CustomModifyReplicationGroup(context.TODO(), desired, latest, &delta)
 		assert.Nil(res)
 		assert.NotNil(err) // due to surplus member cluster
 		var requeueNeededAfter *requeue.RequeueNeededAfter
@@ -384,14 +381,61 @@ func TestCustomModifyReplicationGroup_NodeGroup_available(t *testing.T) {
 			nodeGroup.Status = &availableStatus
 		}
 		var delta ackcompare.Delta
-		var ctx context.Context
 		require.NotNil(latest.ko.Status.MemberClusters)
-		res, err := rm.CustomModifyReplicationGroup(ctx, desired, latest, &delta)
+		res, err := rm.CustomModifyReplicationGroup(context.TODO(), desired, latest, &delta)
 		assert.Nil(res)
 		assert.Nil(err)
 	})
 }
 
+func TestCustomModifyReplicationGroup_Scaling_Async_Rollback(t *testing.T) {
+	assert := assert.New(t)
+	t.Run("ScaleDownRollback=Diff", func(t *testing.T) {
+		desired := provideResource()
+		latest := provideResource()
+		rgId := "RGID"
+		desired.ko.Spec.ReplicationGroupID = &rgId
+		latest.ko.Spec.ReplicationGroupID = &rgId
+		desired.ko.ObjectMeta.Annotations = make(map[string]string)
+		desiredCacheNodeType := "cache.t3.micro"
+		currentCacheNodeType := "cache.m5.large"
+		desired.ko.Annotations[AnnotationLastRequestedCNT] = desiredCacheNodeType
+		desired.ko.Spec.CacheNodeType = &desiredCacheNodeType
+
+		rm := provideResourceManager()
+
+		var delta ackcompare.Delta
+		delta.Add("Spec.CacheNodeType", currentCacheNodeType, desiredCacheNodeType)
+
+		res, err := rm.CustomModifyReplicationGroup(context.TODO(), desired, latest, &delta)
+		assert.Nil(res)
+		assert.NotNil(err)
+		assert.Equal("InvalidParameterCombination: Cannot update CacheNodeType, Please refer to Events for more details", err.Error())
+	})
+
+	t.Run("ScaleInRollback=Diff", func(t *testing.T) {
+		desired := provideResource()
+		latest := provideResource()
+		rgId := "RGID"
+		desired.ko.Spec.ReplicationGroupID = &rgId
+		latest.ko.Spec.ReplicationGroupID = &rgId
+		desired.ko.ObjectMeta.Annotations = make(map[string]string)
+
+		desiredNodeGroup := int64(4)
+		currentNodeGroup := int64(3)
+		desired.ko.Annotations[AnnotationLastRequestedNNG] = strconv.Itoa(int(desiredNodeGroup))
+		desired.ko.Spec.NumNodeGroups = &desiredNodeGroup
+		rm := provideResourceManager()
+
+		var delta ackcompare.Delta
+		delta.Add("Spec.NumNodeGroups", currentNodeGroup, desiredNodeGroup)
+
+		res, err := rm.CustomModifyReplicationGroup(context.TODO(), desired, latest, &delta)
+		assert.Nil(res)
+		assert.NotNil(err)
+		assert.Equal("InvalidParameterCombination: Cannot update NodeGroups, Please refer to Events for more details", err.Error())
+	})
+}
 func TestCustomModifyReplicationGroup_ScaleUpAndDown_And_Resharding(t *testing.T) {
 	assert := assert.New(t)
 
@@ -402,16 +446,26 @@ func TestCustomModifyReplicationGroup_ScaleUpAndDown_And_Resharding(t *testing.T
 		rgId := "RGID"
 		desired.ko.Spec.ReplicationGroupID = &rgId
 		latest.ko.Spec.ReplicationGroupID = &rgId
+		desired.ko.ObjectMeta.Annotations = make(map[string]string)
+		desiredCacheNodeType := "cache.m5.large"
+		currentCacheNodeType := "cache.t3.small"
+		desired.ko.Annotations[AnnotationLastRequestedCNT] = currentCacheNodeType
+		desired.ko.Spec.CacheNodeType = &desiredCacheNodeType
 
+		desiredNodeGroup := int64(4)
+		currentNodeGroup := int64(3)
+		desired.ko.Annotations[AnnotationLastRequestedNNG] = strconv.Itoa(int(currentNodeGroup))
+		desired.ko.Spec.NumNodeGroups = &desiredNodeGroup
+		allowedNodeModifications := []*string{&desiredCacheNodeType}
+		desired.ko.Status.AllowedScaleUpModifications = allowedNodeModifications
 		mocksdkapi := &mocksvcsdkapi.ElastiCacheAPI{}
 		rm := provideResourceManagerWithMockSDKAPI(mocksdkapi)
 
 		var delta ackcompare.Delta
-		delta.Add("Spec.CacheNodeType", "cache.t3.small", "cache.m5.large")
-		delta.Add("Spec.NumNodeGroups", 3, 4)
+		delta.Add("Spec.CacheNodeType", currentCacheNodeType, desiredCacheNodeType)
+		delta.Add("Spec.NumNodeGroups", currentNodeGroup, desiredNodeGroup)
 
-		var ctx context.Context
-		res, err := rm.CustomModifyReplicationGroup(ctx, desired, latest, &delta)
+		res, err := rm.CustomModifyReplicationGroup(context.TODO(), desired, latest, &delta)
 		assert.Nil(res)
 		assert.Nil(err)
 		assert.Empty(mocksdkapi.Calls)
@@ -429,15 +483,19 @@ func TestCustomModifyReplicationGroup_ScaleUpAndDown_And_Resharding(t *testing.T
 
 		var delta ackcompare.Delta
 		delta.Add("Spec.CacheNodeType", "cache.t3.small", "cache.t3.micro")
+		desired.ko.ObjectMeta.Annotations = make(map[string]string)
+		cacheNodeType := "cache.t3.small"
+		desired.ko.Annotations[AnnotationLastRequestedCNT] = "cache.t3.micro"
+		desired.ko.Spec.CacheNodeType = &cacheNodeType
 		oldshardCount := int64(4)
 		newShardCount := int64(10)
 		delta.Add("Spec.NumNodeGroups", oldshardCount, newShardCount)
 		desired.ko.Spec.NumNodeGroups = &newShardCount
 		latest.ko.Spec.NumNodeGroups = &oldshardCount
+		desired.ko.Annotations[AnnotationLastRequestedNNG] = strconv.Itoa(int(oldshardCount))
 		mocksdkapi.On("ModifyReplicationGroupShardConfigurationWithContext", mock.Anything, mock.Anything).Return(nil,
 			awserr.New("Invalid", "Invalid error", nil))
-		var ctx context.Context
-		res, err := rm.CustomModifyReplicationGroup(ctx, desired, latest, &delta)
+		res, err := rm.CustomModifyReplicationGroup(context.TODO(), desired, latest, &delta)
 		assert.Nil(res)
 		assert.NotNil(err)
 		assert.NotEmpty(mocksdkapi.Calls)