Repository: crossplane/provider-aws (go)

Issue:
ReplicationGroup: Add support for shards (node groups) scaling
### What problem are you facing?
It is impossible to change the number of node groups for a ReplicationGroup after deployment. Redis (cluster mode enabled) replication group does allow this, but Crossplane does not.

Probably related to https://github.com/crossplane/provider-aws/blob/master/pkg/clients/elasticache/elasticache.go#L193-L212 and https://github.com/crossplane/provider-aws/blob/f600ff88ad64df73295c96e4446be83b608ddfa4/apis/cache/v1beta1/replication_group_types.go#L424 (immutable)

### How could Crossplane help solve your problem?
This is probably a common use case which Crossplane should be able to handle.


Pull request description:
elasticache: Scale replicationgroup shards
### Description of your changes
Scale up and down replication groups as requested by the user.

On scale down, the AWS API requires us to pick the nodes to remove. I have not considered what the best option is, I picked the first node in the list.

Co-authored-by: sb1-oscar <sb1-oscar@users.noreply.github.com>

Fixes #531

I have:

- [x] Read and followed Crossplane's [contribution process].
- [x] Run `make reviewable test` to ensure this PR is ready for review.

### How has this code been tested

Tested changing the CR to scale up/down. I will have to retest since this PR has been rebased and it's a while since I wrote it.

[contribution process]: https://git.io/fj2m9


Patch:
diff --git a/pkg/clients/elasticache/elasticache.go b/pkg/clients/elasticache/elasticache.go
index a4eb769a5..2df149f19 100644
--- a/pkg/clients/elasticache/elasticache.go
+++ b/pkg/clients/elasticache/elasticache.go
@@ -59,6 +59,8 @@ type Client interface {
 	CreateCacheCluster(context.Context, *elasticache.CreateCacheClusterInput, ...func(*elasticache.Options)) (*elasticache.CreateCacheClusterOutput, error)
 	DeleteCacheCluster(context.Context, *elasticache.DeleteCacheClusterInput, ...func(*elasticache.Options)) (*elasticache.DeleteCacheClusterOutput, error)
 	ModifyCacheCluster(context.Context, *elasticache.ModifyCacheClusterInput, ...func(*elasticache.Options)) (*elasticache.ModifyCacheClusterOutput, error)
+
+	ModifyReplicationGroupShardConfiguration(context.Context, *elasticache.ModifyReplicationGroupShardConfigurationInput, ...func(*elasticache.Options)) (*elasticache.ModifyReplicationGroupShardConfigurationOutput, error)
 }
 
 // NewClient returns a new ElastiCache client. Credentials must be passed as
@@ -147,6 +149,26 @@ func NewModifyReplicationGroupInput(g v1beta1.ReplicationGroupParameters, id str
 	}
 }
 
+// NewModifyReplicationGroupShardConfigurationInput returns ElastiCache replication group
+// shard configuration modification input suitable for use with the AWS API.
+func NewModifyReplicationGroupShardConfigurationInput(g v1beta1.ReplicationGroupParameters, id string, rg elasticachetypes.ReplicationGroup) *elasticache.ModifyReplicationGroupShardConfigurationInput {
+	input := &elasticache.ModifyReplicationGroupShardConfigurationInput{
+		ApplyImmediately:   g.ApplyModificationsImmediately,
+		NodeGroupCount:     int32(*g.NumNodeGroups),
+		ReplicationGroupId: aws.String(id),
+	}
+
+	// For scale down we must name the nodes. This code picks the oldest rg
+	// now, but there might be a better algorithm, such as the one with least
+	// data
+	remove := len(rg.NodeGroups) - int(input.NodeGroupCount)
+	for i := 0; i < remove; i++ {
+		input.NodeGroupsToRemove = append(input.NodeGroupsToRemove, aws.ToString(rg.NodeGroups[i].NodeGroupId))
+	}
+
+	return input
+}
+
 // NewDeleteReplicationGroupInput returns ElastiCache replication group deletion
 // input suitable for use with the AWS API.
 func NewDeleteReplicationGroupInput(id string) *elasticache.DeleteReplicationGroupInput {
@@ -208,6 +230,12 @@ func LateInitialize(s *v1beta1.ReplicationGroupParameters, rg elasticachetypes.R
 	}
 }
 
+// ReplicationGroupShardConfigurationNeedsUpdate returns true if the supplied ReplicationGroup and
+// the configuration shards.
+func ReplicationGroupShardConfigurationNeedsUpdate(kube v1beta1.ReplicationGroupParameters, rg elasticachetypes.ReplicationGroup) bool {
+	return kube.NumNodeGroups != nil && *kube.NumNodeGroups != len(rg.NodeGroups)
+}
+
 // ReplicationGroupNeedsUpdate returns true if the supplied ReplicationGroup and
 // the configuration of its member clusters differ from given desired state.
 func ReplicationGroupNeedsUpdate(kube v1beta1.ReplicationGroupParameters, rg elasticachetypes.ReplicationGroup, ccList []elasticachetypes.CacheCluster) bool {
diff --git a/pkg/clients/elasticache/fake/fake.go b/pkg/clients/elasticache/fake/fake.go
index cd145b307..5dd619edb 100644
--- a/pkg/clients/elasticache/fake/fake.go
+++ b/pkg/clients/elasticache/fake/fake.go
@@ -40,6 +40,8 @@ type MockClient struct {
 	MockCreateCacheCluster    func(context.Context, *elasticache.CreateCacheClusterInput, []func(*elasticache.Options)) (*elasticache.CreateCacheClusterOutput, error)
 	MockDeleteCacheCluster    func(context.Context, *elasticache.DeleteCacheClusterInput, []func(*elasticache.Options)) (*elasticache.DeleteCacheClusterOutput, error)
 	MockModifyCacheCluster    func(context.Context, *elasticache.ModifyCacheClusterInput, []func(*elasticache.Options)) (*elasticache.ModifyCacheClusterOutput, error)
+
+	MockModifyReplicationGroupShardConfiguration func(context.Context, *elasticache.ModifyReplicationGroupShardConfigurationInput, []func(*elasticache.Options)) (*elasticache.ModifyReplicationGroupShardConfigurationOutput, error)
 }
 
 // DescribeReplicationGroups calls the underlying
@@ -72,6 +74,12 @@ func (c *MockClient) DescribeCacheClusters(ctx context.Context, i *elasticache.D
 	return c.MockDescribeCacheClusters(ctx, i, opts)
 }
 
+// ModifyReplicationGroupShardConfiguration calls the underlying
+// MockModifyReplicationGroupShardConfiguration method.
+func (c *MockClient) ModifyReplicationGroupShardConfiguration(ctx context.Context, i *elasticache.ModifyReplicationGroupShardConfigurationInput, opts ...func(*elasticache.Options)) (*elasticache.ModifyReplicationGroupShardConfigurationOutput, error) {
+	return c.MockModifyReplicationGroupShardConfiguration(ctx, i, opts)
+}
+
 // DescribeCacheSubnetGroups calls the underlying
 // MockDescribeCacheSubnetGroups method.
 func (c *MockClient) DescribeCacheSubnetGroups(ctx context.Context, i *elasticache.DescribeCacheSubnetGroupsInput, opts ...func(*elasticache.Options)) (*elasticache.DescribeCacheSubnetGroupsOutput, error) {
diff --git a/pkg/controller/cache/managed.go b/pkg/controller/cache/managed.go
index 96102a9d0..1cde583ba 100644
--- a/pkg/controller/cache/managed.go
+++ b/pkg/controller/cache/managed.go
@@ -55,6 +55,7 @@ const (
 	errCreateReplicationGroup   = "cannot create ElastiCache replication group"
 	errModifyReplicationGroup   = "cannot modify ElastiCache replication group"
 	errDeleteReplicationGroup   = "cannot delete ElastiCache replication group"
+	errModifyReplicationGroupSC = "cannot modify ElastiCache replication group shard configuration"
 )
 
 // SetupReplicationGroup adds a controller that reconciles ReplicationGroups.
@@ -146,7 +147,7 @@ func (e *external) Observe(ctx context.Context, mg resource.Managed) (managed.Ex
 
 	return managed.ExternalObservation{
 		ResourceExists:    true,
-		ResourceUpToDate:  !elasticache.ReplicationGroupNeedsUpdate(cr.Spec.ForProvider, rg, ccList),
+		ResourceUpToDate:  !elasticache.ReplicationGroupNeedsUpdate(cr.Spec.ForProvider, rg, ccList) && !elasticache.ReplicationGroupShardConfigurationNeedsUpdate(cr.Spec.ForProvider, rg),
 		ConnectionDetails: elasticache.ConnectionEndpoint(rg),
 	}, nil
 }
@@ -195,7 +196,23 @@ func (e *external) Update(ctx context.Context, mg resource.Managed) (managed.Ext
 	if cr.Status.AtProvider.Status != v1beta1.StatusAvailable {
 		return managed.ExternalUpdate{}, nil
 	}
-	_, err := e.client.ModifyReplicationGroup(ctx, elasticache.NewModifyReplicationGroupInput(cr.Spec.ForProvider, meta.GetExternalName(cr)))
+
+	rsp, err := e.client.DescribeReplicationGroups(ctx, elasticache.NewDescribeReplicationGroupsInput(meta.GetExternalName(cr)))
+	if err != nil {
+		return managed.ExternalUpdate{}, awsclient.Wrap(err, errDescribeReplicationGroup)
+	}
+	rg := rsp.ReplicationGroups[0]
+
+	if elasticache.ReplicationGroupShardConfigurationNeedsUpdate(cr.Spec.ForProvider, rg) {
+		_, err = e.client.ModifyReplicationGroupShardConfiguration(ctx, elasticache.NewModifyReplicationGroupShardConfigurationInput(cr.Spec.ForProvider, meta.GetExternalName(cr), rg))
+		if err != nil {
+			return managed.ExternalUpdate{}, awsclient.Wrap(err, errModifyReplicationGroupSC)
+		}
+		// we can only do one change at a time, so we'll have to return early here
+		return managed.ExternalUpdate{}, nil
+	}
+
+	_, err = e.client.ModifyReplicationGroup(ctx, elasticache.NewModifyReplicationGroupInput(cr.Spec.ForProvider, meta.GetExternalName(cr)))
 	return managed.ExternalUpdate{}, awsclient.Wrap(err, errModifyReplicationGroup)
 }
 


Test patch:
diff --git a/pkg/clients/elasticache/elasticache_test.go b/pkg/clients/elasticache/elasticache_test.go
index 059ce649b..29eef5b37 100644
--- a/pkg/clients/elasticache/elasticache_test.go
+++ b/pkg/clients/elasticache/elasticache_test.go
@@ -293,6 +293,79 @@ func TestNewModifyReplicationGroupInput(t *testing.T) {
 	}
 }
 
+func TestNewModifyReplicationGroupShardConfigurationInput(t *testing.T) {
+	cases := []struct {
+		name     string
+		params   v1beta1.ReplicationGroupParameters
+		observed elasticachetypes.ReplicationGroup
+		want     *elasticache.ModifyReplicationGroupShardConfigurationInput
+	}{
+		{
+			name:   "ScaleUp",
+			params: replicationGroup.Spec.ForProvider,
+			observed: elasticachetypes.ReplicationGroup{
+				NodeGroups: []elasticachetypes.NodeGroup{
+					{
+						NodeGroupId: aws.String("ng-01"),
+					},
+				},
+			},
+			want: &elasticache.ModifyReplicationGroupShardConfigurationInput{
+				ApplyImmediately:   true,
+				NodeGroupCount:     2,
+				ReplicationGroupId: aws.String(name, aws.FieldRequired),
+			},
+		},
+		{
+			name:   "ScaleDown",
+			params: replicationGroup.Spec.ForProvider,
+			observed: elasticachetypes.ReplicationGroup{
+				NodeGroups: []elasticachetypes.NodeGroup{
+					{NodeGroupId: aws.String("ng-01")},
+					{NodeGroupId: aws.String("ng-02")},
+					{NodeGroupId: aws.String("ng-03")},
+				},
+			},
+			want: &elasticache.ModifyReplicationGroupShardConfigurationInput{
+				ApplyImmediately:   true,
+				NodeGroupCount:     2,
+				NodeGroupsToRemove: []string{"ng-01"},
+				ReplicationGroupId: aws.String(name),
+			},
+		},
+		{
+			name: "ApplyImmediatelyFromRG",
+			params: v1beta1.ReplicationGroupParameters{
+				ApplyModificationsImmediately: false,
+				NumNodeGroups:                 &numNodeGroups,
+			},
+			observed: elasticachetypes.ReplicationGroup{
+				NodeGroups: []elasticachetypes.NodeGroup{
+					{NodeGroupId: aws.String("ng-01")},
+					{NodeGroupId: aws.String("ng-02")},
+					{NodeGroupId: aws.String("ng-03")},
+				},
+			},
+			want: &elasticache.ModifyReplicationGroupShardConfigurationInput{
+				ApplyImmediately:   false,
+				NodeGroupCount:     2,
+				NodeGroupsToRemove: []string{"ng-01"},
+				ReplicationGroupId: aws.String(name),
+			},
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			got := NewModifyReplicationGroupShardConfigurationInput(tc.params, name, tc.observed)
+
+			if diff := cmp.Diff(tc.want, got, cmpopts.IgnoreTypes(document.NoSerde{})); diff != "" {
+				t.Errorf("NewModifyReplicationGroupShardConfigurationInput(...): -want, +got:\n%s", diff)
+			}
+		})
+	}
+}
+
 func TestNewDeleteReplicationGroupInput(t *testing.T) {
 	cases := []struct {
 		name string
@@ -680,6 +753,52 @@ func TestReplicationGroupNeedsUpdate(t *testing.T) {
 	}
 }
 
+func TestReplicationGroupShardConfigurationNeedsUpdate(t *testing.T) {
+	cases := []struct {
+		name   string
+		kube   v1beta1.ReplicationGroupParameters
+		rg     elasticachetypes.ReplicationGroup
+		ccList []elasticachetypes.CacheCluster
+		want   bool
+	}{
+		{
+			name: "NodeMismatch",
+			kube: replicationGroup.Spec.ForProvider, // 2
+			rg: elasticachetypes.ReplicationGroup{
+				NodeGroups: make([]elasticachetypes.NodeGroup, 3),
+			},
+			want: true,
+		},
+		{
+			name: "UpToDate",
+			kube: replicationGroup.Spec.ForProvider,
+			rg: elasticachetypes.ReplicationGroup{
+				NodeGroups: make([]elasticachetypes.NodeGroup, numNodeGroups),
+			},
+			want: false,
+		},
+		{
+			name: "NilNumNodes",
+			kube: v1beta1.ReplicationGroupParameters{
+				NumNodeGroups: nil,
+			},
+			rg: elasticachetypes.ReplicationGroup{
+				NodeGroups: make([]elasticachetypes.NodeGroup, numNodeGroups),
+			},
+			want: false,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			got := ReplicationGroupShardConfigurationNeedsUpdate(tc.kube, tc.rg)
+			if got != tc.want {
+				t.Errorf("ReplicationGroupShardConfigurationNeedsUpdate(...): want %t, got %t", tc.want, got)
+			}
+		})
+	}
+}
+
 func TestCacheClusterNeedsUpdate(t *testing.T) {
 	cases := []struct {
 		name string
diff --git a/pkg/controller/cache/managed_test.go b/pkg/controller/cache/managed_test.go
index 0f5664ba1..66e9f0e0f 100644
--- a/pkg/controller/cache/managed_test.go
+++ b/pkg/controller/cache/managed_test.go
@@ -117,6 +117,10 @@ func withTags(tagMaps ...map[string]string) replicationGroupModifier {
 	return func(r *v1beta1.ReplicationGroup) { r.Spec.ForProvider.Tags = tagList }
 }
 
+func withNumNodeGroups(n int) replicationGroupModifier {
+	return func(r *v1beta1.ReplicationGroup) { r.Spec.ForProvider.NumNodeGroups = &n }
+}
+
 func replicationGroup(rm ...replicationGroupModifier) *v1beta1.ReplicationGroup {
 	r := &v1beta1.ReplicationGroup{
 		ObjectMeta: objectMeta,
@@ -452,6 +456,53 @@ func TestUpdate(t *testing.T) {
 			),
 			returnsErr: true,
 		},
+		{
+			name: "CallsModifyReplicationGroupShardConfiguration",
+			e: &external{client: &fake.MockClient{
+				MockDescribeReplicationGroups: func(ctx context.Context, _ *elasticache.DescribeReplicationGroupsInput, opts []func(*elasticache.Options)) (*elasticache.DescribeReplicationGroupsOutput, error) {
+					return &elasticache.DescribeReplicationGroupsOutput{
+						ReplicationGroups: []types.ReplicationGroup{{
+							Status:                 aws.String(v1beta1.StatusAvailable),
+							MemberClusters:         []string{cacheClusterID},
+							AutomaticFailover:      types.AutomaticFailoverStatusEnabled,
+							NodeGroups:             []types.NodeGroup{{NodeGroupId: aws.String("ng-01")}, {NodeGroupId: aws.String("ng-02")}},
+							CacheNodeType:          aws.String(cacheNodeType),
+							SnapshotRetentionLimit: aws.Int32(int32(snapshotRetentionLimit)),
+							SnapshotWindow:         aws.String(snapshotWindow),
+							ClusterEnabled:         aws.Bool(true),
+							ConfigurationEndpoint:  &types.Endpoint{Address: aws.String(host), Port: int32(port)},
+						}},
+					}, nil
+				},
+				MockDescribeCacheClusters: func(ctx context.Context, _ *elasticache.DescribeCacheClustersInput, opts []func(*elasticache.Options)) (*elasticache.DescribeCacheClustersOutput, error) {
+					return &elasticache.DescribeCacheClustersOutput{
+						CacheClusters: []types.CacheCluster{{
+							EngineVersion:              aws.String(engineVersion),
+							PreferredMaintenanceWindow: aws.String("never!"), // This field needs to be updated.
+						}},
+					}, nil
+
+				},
+				MockModifyReplicationGroupShardConfiguration: func(ctx context.Context, _ *elasticache.ModifyReplicationGroupShardConfigurationInput, opts []func(*elasticache.Options)) (*elasticache.ModifyReplicationGroupShardConfigurationOutput, error) {
+					return nil, errorBoom
+				},
+			}},
+			r: replicationGroup(
+				withReplicationGroupID(name),
+				withProviderStatus(v1beta1.StatusAvailable),
+				withConditions(xpv1.Available()),
+				withMemberClusters([]string{cacheClusterID}),
+				withNumNodeGroups(3),
+			),
+			want: replicationGroup(
+				withReplicationGroupID(name),
+				withProviderStatus(v1beta1.StatusAvailable),
+				withConditions(xpv1.Available()),
+				withMemberClusters([]string{cacheClusterID}),
+				withNumNodeGroups(3),
+			),
+			returnsErr: true,
+		},
 	}
 
 	for _, tc := range cases {