Declarative Tests for ScalingUpAndDown and Async rollback fixes

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-node-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
 }
 
@@ -76,6 +79,9 @@ func (rm *resourceManager) CustomModifyReplicationGroupSetOutput(
 	}
 	ko.Spec.LogDeliveryConfigurations = logDeliveryConfig
 
+	// Keep the value of desired for CacheNodeType.
+	ko.Spec.CacheNodeType = r.ko.Spec.CacheNodeType
+
 	rm.setAnnotationsFields(r, ko)
 	return ko, nil
 }
@@ -267,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
@@ -297,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,45 @@ 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" {
+		var lastRequestedNodeGroupConfig []*svcapitypes.NodeGroupConfiguration
+		_ = json.Unmarshal([]byte(val), &lastRequestedNodeGroupConfig)
+		return !reflect.DeepEqual(desiredNodeGroupConfig, lastRequestedNodeGroupConfig)
+	}
+
+	// This means there is delta and no value in annotation or in Spec
+	return true
+}