Scale up/down and Scale in/out memory considerations

pkg/resource/replication_group/custom_update_api.go

@@ -101,6 +101,19 @@ func (rm *resourceManager) CustomModifyReplicationGroup(
 		return rm.decreaseReplicaCount(ctx, desired, latest)
 	}
 
+	// If there is a scale up modification, then we would prioritize it
+	// over increase/decrease shards. This is important since performing
+	// scale in without scale up might fail due to insufficient memory.
+	if delta.DifferentAt("Spec.CacheNodeType") && desired.ko.Status.AllowedScaleUpModifications != nil {
+		if desired.ko.Spec.CacheNodeType != nil {
+			for _, scaleUpInstance := range desired.ko.Status.AllowedScaleUpModifications {
+				if *scaleUpInstance == *desired.ko.Spec.CacheNodeType {
+					return nil, nil
+				}
+			}
+		}
+	}
+
 	// increase/decrease shards
 	if rm.shardConfigurationsDiffer(desired, latest) {
 		return rm.updateShardConfiguration(ctx, desired, latest)

pkg/resource/replication_group/custom_update_api_test.go

@@ -19,6 +19,7 @@ import (
 	"github.com/aws-controllers-k8s/elasticache-controller/pkg/testutil"
 	ackmetrics "github.com/aws-controllers-k8s/runtime/pkg/metrics"
 	"github.com/aws-controllers-k8s/runtime/pkg/requeue"
+	"github.com/aws/aws-sdk-go/aws/awserr"
 	"github.com/pkg/errors"
 	"github.com/stretchr/testify/mock"
 	"go.uber.org/zap/zapcore"
@@ -391,6 +392,60 @@ func TestCustomModifyReplicationGroup_NodeGroup_available(t *testing.T) {
 	})
 }
 
+func TestCustomModifyReplicationGroup_ScaleUpAndDown_And_Resharding(t *testing.T)  {
+	assert := assert.New(t)
+
+	// Tests
+	t.Run("ScaleInAndScaleUp=Diff", func(t *testing.T) {
+		desired := provideResource()
+		latest := provideResource()
+		rgId := "RGID"
+		desired.ko.Spec.ReplicationGroupID = &rgId
+		latest.ko.Spec.ReplicationGroupID = &rgId
+
+		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)
+
+		var ctx context.Context
+		res, err := rm.CustomModifyReplicationGroup(ctx, desired, latest, &delta)
+		assert.Nil(res)
+		assert.Nil(err)
+		assert.Empty(mocksdkapi.Calls)
+	})
+
+	t.Run("ScaleOutAndScaleDown=Diff", func(t *testing.T) {
+		desired := provideResource()
+		latest := provideResource()
+		rgId := "RGID"
+		desired.ko.Spec.ReplicationGroupID = &rgId
+		latest.ko.Spec.ReplicationGroupID = &rgId
+
+		mocksdkapi := &mocksvcsdkapi.ElastiCacheAPI{}
+		rm := provideResourceManagerWithMockSDKAPI(mocksdkapi)
+
+		var delta ackcompare.Delta
+		delta.Add("Spec.CacheNodeType", "cache.t3.small", "cache.t3.micro")
+		oldshardCount := int64(4)
+		newShardCount := int64(10)
+		delta.Add("Spec.NumNodeGroups", oldshardCount, newShardCount)
+		desired.ko.Spec.NumNodeGroups = &newShardCount
+		latest.ko.Spec.NumNodeGroups = &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)
+		assert.Nil(res)
+		assert.NotNil(err)
+		assert.NotEmpty(mocksdkapi.Calls)
+		assert.Equal("ModifyReplicationGroupShardConfigurationWithContext", mocksdkapi.Calls[0].Method)
+	})
+
+}
+
 // TestReplicaCountDifference tests scenarios to check if desired, latest replica count
 // configurations differ
 func TestReplicaCountDifference(t *testing.T) {