Support Table tags updates (#22)

Issue: aws-controllers-k8s/community#1029

This patch adds hooks and delta helpers to properly update Table tags.

Description of changes:
- Drop the generate updateTable function and use a custom one
- Add hooks to properly compute the tags delta and make necessary API
calls to update them
- Add e2e tests for table tag updates

By submitting this pull request, I confirm that my contribution is made under
the terms of the Apache 2.0 license.

Author: a-hilaly

Diff for: apis/v1alpha1/ack-generate-metadata.yaml

@@ -1,13 +1,13 @@
 ack_generate_info:
-  build_date: "2021-10-27T21:04:06Z"
-  build_hash: d2b063806d25cfcae4f2d4eb44f8e3f713b23e8e
-  go_version: go1.15
+  build_date: "2021-12-22T13:40:06Z"
+  build_hash: 6f17f51682dc0d16c36aa456fd22855ce9282fbc
+  go_version: go1.16.4
   version: v0.15.2
 api_directory_checksum: c1d144a18336326f141e97e6800b47f64ed992cc
 api_version: v1alpha1
 aws_sdk_go_version: v1.38.47
 generator_config_info:
-  file_checksum: 3d4ab94742ecf92212b94a5e47bdda8258589718
+  file_checksum: 2c5f9585b1b0356e601e17efc7f5c47bdabdf4ab
   original_file_name: generator.yaml
 last_modification:
   reason: API generation

Diff for: apis/v1alpha1/generator.yaml

@@ -7,6 +7,10 @@ operations:
     primary_identifier_field_name: BackupArn
 resources:
   Table:
+    fields:
+      Tags:
+        compare:
+          is_ignored: true
     exceptions:
       errors:
         404:
@@ -15,11 +19,13 @@ resources:
         - InternalServerError
         - LimitExceededException
         - ResourceInUseException
+    update_operation:
+      custom_method_name: customUpdateTable
     hooks:
+      delta_pre_compare:
+        code: customPreCompare(delta, a, b)
       sdk_read_one_post_set_output:
         template_path: hooks/table/sdk_read_one_post_set_output.go.tpl
-      sdk_update_pre_build_request:
-        template_path: hooks/table/sdk_update_pre_build_request.go.tpl
       sdk_delete_pre_build_request:
         template_path: hooks/table/sdk_delete_pre_build_request.go.tpl
   GlobalTable:

Diff for: generator.yaml

@@ -7,6 +7,10 @@ operations:
     primary_identifier_field_name: BackupArn
 resources:
   Table:
+    fields:
+      Tags:
+        compare:
+          is_ignored: true
     exceptions:
       errors:
         404:
@@ -15,11 +19,13 @@ resources:
         - InternalServerError
         - LimitExceededException
         - ResourceInUseException
+    update_operation:
+      custom_method_name: customUpdateTable
     hooks:
+      delta_pre_compare:
+        code: customPreCompare(delta, a, b)
       sdk_read_one_post_set_output:
         template_path: hooks/table/sdk_read_one_post_set_output.go.tpl
-      sdk_update_pre_build_request:
-        template_path: hooks/table/sdk_update_pre_build_request.go.tpl
       sdk_delete_pre_build_request:
         template_path: hooks/table/sdk_delete_pre_build_request.go.tpl
   GlobalTable:

Diff for: helm/templates/deployment.yaml

@@ -75,3 +75,12 @@ spec:
           value: {{ join "," .Values.resourceTags | quote }}
       terminationGracePeriodSeconds: 10
       nodeSelector: {{ toYaml .Values.deployment.nodeSelector | nindent 8 }}
+      {{ if .Values.deployment.tolerations -}}
+      tolerations: {{ toYaml .Values.deployment.tolerations | nindent 8 }}
+      {{ end -}}
+      {{ if .Values.deployment.affinity -}}
+      affinity: {{ toYaml .Values.deployment.affinity | nindent 8 }}
+      {{ end -}}
+      {{ if .Values.deployment.priorityClassName -}}
+      priorityClassName: {{ .Values.deployment.priorityClassName -}}
+      {{ end -}}

Diff for: helm/values.yaml

@@ -15,9 +15,20 @@ deployment:
   annotations: {}
   labels: {}
   containerPort: 8080
+  # Which nodeSelector to set?
+  # See: https://kubernetes.io/docs/concepts/scheduling-eviction/assign-pod-node/#nodeselector
   nodeSelector:
     kubernetes.io/os: linux
-
+  # Which tolerations to set?
+  # See: https://kubernetes.io/docs/concepts/scheduling-eviction/taint-and-toleration/
+  tolerations: {}
+  # What affinity to set?
+  # See: https://kubernetes.io/docs/concepts/scheduling-eviction/assign-pod-node/#affinity-and-anti-affinity
+  affinity: {}
+  # Which priorityClassName to set?
+  # See: https://kubernetes.io/docs/concepts/scheduling-eviction/pod-priority-preemption/#pod-priority
+  priorityClassName:
+  
 metrics:
   service:
     # Set to true to automatically create a Kubernetes Service resource for the

Diff for: pkg/resource/backup/sdk.go

@@ -14,11 +14,18 @@
 package table
 
 import (
+	"context"
 	"errors"
 	"time"
 
-	"github.com/aws-controllers-k8s/dynamodb-controller/apis/v1alpha1"
+	ackcompare "github.com/aws-controllers-k8s/runtime/pkg/compare"
 	ackrequeue "github.com/aws-controllers-k8s/runtime/pkg/requeue"
+	ackrtlog "github.com/aws-controllers-k8s/runtime/pkg/runtime/log"
+	ackutil "github.com/aws-controllers-k8s/runtime/pkg/util"
+	svcsdk "github.com/aws/aws-sdk-go/service/dynamodb"
+	corev1 "k8s.io/api/core/v1"
+
+	"github.com/aws-controllers-k8s/dynamodb-controller/apis/v1alpha1"
 )
 
 var (
@@ -95,3 +102,242 @@ func isTableUpdating(r *resource) bool {
 	dbis := *r.ko.Status.TableStatus
 	return dbis == string(v1alpha1.TableStatus_SDK_UPDATING)
 }
+
+func (rm *resourceManager) customUpdateTable(
+	ctx context.Context,
+	desired *resource,
+	latest *resource,
+	delta *ackcompare.Delta,
+) (updated *resource, err error) {
+	rlog := ackrtlog.FromContext(ctx)
+	exit := rlog.Trace("rm.customUpdateTable")
+	defer exit(err)
+
+	if isTableDeleting(latest) {
+		msg := "table is currently being deleted"
+		setSyncedCondition(desired, corev1.ConditionFalse, &msg, nil)
+		return desired, requeueWaitWhileDeleting
+	}
+	if isTableCreating(latest) {
+		msg := "table is currently being created"
+		setSyncedCondition(desired, corev1.ConditionFalse, &msg, nil)
+		return desired, requeueWaitWhileCreating
+	}
+	if isTableUpdating(latest) {
+		msg := "table is currently being updated"
+		setSyncedCondition(desired, corev1.ConditionFalse, &msg, nil)
+		return desired, requeueWaitWhileUpdating
+	}
+	if tableHasTerminalStatus(latest) {
+		msg := "table is in '" + *latest.ko.Status.TableStatus + "' status"
+		setTerminalCondition(desired, corev1.ConditionTrue, &msg, nil)
+		setSyncedCondition(desired, corev1.ConditionTrue, nil, nil)
+		return desired, nil
+	}
+
+	// Merge in the information we read from the API call above to the copy of
+	// the original Kubernetes object we passed to the function
+	ko := desired.ko.DeepCopy()
+	rm.setStatusDefaults(ko)
+
+	if delta.DifferentAt("Spec.Tags") {
+		if err := rm.syncTableTags(ctx, latest, desired); err != nil {
+			return nil, err
+		}
+	}
+
+	// TODO(hilalymh): support updating all table field
+	return &resource{ko}, nil
+}
+
+// syncTableTags updates a dynamodb table tags.
+//
+// TODO(hilalymh): move this function to a common utility file. This function can be reused
+// to tag GlobalTable resources.
+func (rm *resourceManager) syncTableTags(
+	ctx context.Context,
+	latest *resource,
+	desired *resource,
+) (err error) {
+	rlog := ackrtlog.FromContext(ctx)
+	exit := rlog.Trace("rm.syncTableTags")
+	defer exit(err)
+
+	added, updated, removed := computeTagsDelta(latest.ko.Spec.Tags, desired.ko.Spec.Tags)
+
+	// There are no API calls to update an existing tag. To update a tag we will have to first
+	// delete it and then recreate it with the new value.
+
+	// Tags to remove
+	for _, updatedTag := range updated {
+		removed = append(removed, updatedTag.Key)
+	}
+	// Tags to create
+	added = append(added, updated...)
+
+	if len(removed) > 0 {
+		_, err = rm.sdkapi.UntagResourceWithContext(
+			ctx,
+			&svcsdk.UntagResourceInput{
+				ResourceArn: (*string)(latest.ko.Status.ACKResourceMetadata.ARN),
+				TagKeys:     removed,
+			},
+		)
+		rm.metrics.RecordAPICall("GET", "UntagResource", err)
+		if err != nil {
+			return err
+		}
+	}
+
+	if len(added) > 0 {
+		_, err = rm.sdkapi.TagResourceWithContext(
+			ctx,
+			&svcsdk.TagResourceInput{
+				ResourceArn: (*string)(latest.ko.Status.ACKResourceMetadata.ARN),
+				Tags:        sdkTagsFromResourceTags(added),
+			},
+		)
+		rm.metrics.RecordAPICall("GET", "UntagResource", err)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// equalTags returns true if two Tag arrays are equal regardless of the order
+// of their elements.
+func equalTags(
+	a []*v1alpha1.Tag,
+	b []*v1alpha1.Tag,
+) bool {
+	added, updated, removed := computeTagsDelta(a, b)
+	return len(added) == 0 && len(updated) == 0 && len(removed) == 0
+}
+
+// resourceTagsFromSDKTags transforms a *svcsdk.Tag array to a *v1alpha1.Tag array.
+func resourceTagsFromSDKTags(svcTags []*svcsdk.Tag) []*v1alpha1.Tag {
+	tags := make([]*v1alpha1.Tag, len(svcTags))
+	for i := range svcTags {
+		tags[i] = &v1alpha1.Tag{
+			Key:   svcTags[i].Key,
+			Value: svcTags[i].Value,
+		}
+	}
+	return tags
+}
+
+// svcTagsFromResourceTags transforms a *v1alpha1.Tag array to a *svcsdk.Tag array.
+func sdkTagsFromResourceTags(rTags []*v1alpha1.Tag) []*svcsdk.Tag {
+	tags := make([]*svcsdk.Tag, len(rTags))
+	for i := range rTags {
+		tags[i] = &svcsdk.Tag{
+			Key:   rTags[i].Key,
+			Value: rTags[i].Value,
+		}
+	}
+	return tags
+}
+
+// computeTagsDelta compares two Tag arrays and return three different list
+// containing the added, updated and removed tags.
+// The removed tags only contains the Key of tags
+func computeTagsDelta(
+	a []*v1alpha1.Tag,
+	b []*v1alpha1.Tag,
+) (added, updated []*v1alpha1.Tag, removed []*string) {
+	var visitedIndexes []string
+mainLoop:
+	for _, aElement := range a {
+		visitedIndexes = append(visitedIndexes, *aElement.Key)
+		for _, bElement := range b {
+			if equalStrings(aElement.Key, bElement.Key) {
+				if !equalStrings(aElement.Value, bElement.Value) {
+					updated = append(updated, bElement)
+				}
+				continue mainLoop
+			}
+		}
+		removed = append(removed, aElement.Key)
+	}
+	for _, bElement := range b {
+		if !ackutil.InStrings(*bElement.Key, visitedIndexes) {
+			added = append(added, bElement)
+		}
+	}
+	return added, updated, removed
+}
+
+func equalStrings(a, b *string) bool {
+	if a == nil {
+		return b == nil || *b == ""
+	}
+	return (*a == "" && b == nil) || *a == *b
+}
+
+// setResourceAdditionalFields will describe the fields that are not return by
+// DescribeTable calls
+func (rm *resourceManager) setResourceAdditionalFields(
+	ctx context.Context,
+	ko *v1alpha1.Table,
+) (err error) {
+	rlog := ackrtlog.FromContext(ctx)
+	exit := rlog.Trace("rm.setResourceAdditionalFields")
+	defer exit(err)
+
+	ko.Spec.Tags, err = rm.getResourceTagsPagesWithContext(ctx, string(*ko.Status.ACKResourceMetadata.ARN))
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// getResourceTagsPagesWithContext queries the list of tags of a given resource.
+func (rm *resourceManager) getResourceTagsPagesWithContext(ctx context.Context, resourceARN string) ([]*v1alpha1.Tag, error) {
+	var err error
+	rlog := ackrtlog.FromContext(ctx)
+	exit := rlog.Trace("rm.getResourceTagsPagesWithContext")
+	defer exit(err)
+
+	tags := []*v1alpha1.Tag{}
+
+	var token *string = nil
+	for {
+		var listTagsOfResourceOutput *svcsdk.ListTagsOfResourceOutput
+		listTagsOfResourceOutput, err = rm.sdkapi.ListTagsOfResourceWithContext(
+			ctx,
+			&svcsdk.ListTagsOfResourceInput{
+				NextToken:   token,
+				ResourceArn: &resourceARN,
+			},
+		)
+		rm.metrics.RecordAPICall("GET", "ListTagsOfResource", err)
+		if err != nil {
+			return nil, err
+		}
+		tags = append(tags, resourceTagsFromSDKTags(listTagsOfResourceOutput.Tags)...)
+		if listTagsOfResourceOutput.NextToken == nil {
+			break
+		}
+		token = listTagsOfResourceOutput.NextToken
+	}
+	return tags, nil
+}
+
+func customPreCompare(
+	delta *ackcompare.Delta,
+	a *resource,
+	b *resource,
+) {
+	// TODO(hilalymh): customDeltaFunctions for AttributeDefintions
+	// TODO(hilalymh): customDeltaFunctions for GlobalSecondaryIndexes
+
+	if len(a.ko.Spec.Tags) != len(b.ko.Spec.Tags) {
+		delta.Add("Spec.Tags", a.ko.Spec.Tags, b.ko.Spec.Tags)
+	} else if a.ko.Spec.Tags != nil && b.ko.Spec.Tags != nil {
+		if !equalTags(a.ko.Spec.Tags, b.ko.Spec.Tags) {
+			delta.Add("Spec.Tags", a.ko.Spec.Tags, b.ko.Spec.Tags)
+		}
+	}
+}

Diff for: pkg/resource/table/delta.go

@@ -0,0 +1,132 @@
+// Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License"). You may
+// not use this file except in compliance with the License. A copy of the
+// License is located at
+//
+//     http://aws.amazon.com/apache2.0/
+//
+// or in the "license" file accompanying this file. This file is distributed
+// on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+// express or implied. See the License for the specific language governing
+// permissions and limitations under the License.
+
+package table
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/aws/aws-sdk-go/aws"
+
+	"github.com/aws-controllers-k8s/dynamodb-controller/apis/v1alpha1"
+)
+
+var (
+	Tag1 = &v1alpha1.Tag{
+		Key:   aws.String("k1"),
+		Value: aws.String("v1"),
+	}
+	Tag2 = &v1alpha1.Tag{
+		Key:   aws.String("k2"),
+		Value: aws.String("v2"),
+	}
+	Tag2Updated = &v1alpha1.Tag{
+		Key:   aws.String("k2"),
+		Value: aws.String("v2-updated"),
+	}
+	Tag3 = &v1alpha1.Tag{
+		Key:   aws.String("k3"),
+		Value: aws.String("v3"),
+	}
+)
+
+func Test_computeTagsDelta(t *testing.T) {
+	type args struct {
+		a []*v1alpha1.Tag
+		b []*v1alpha1.Tag
+	}
+	tests := []struct {
+		name        string
+		args        args
+		wantAdded   []*v1alpha1.Tag
+		wantUpdated []*v1alpha1.Tag
+		wantRemoved []*string
+	}{
+		{
+			name: "nil arrays",
+			args: args{
+				a: nil,
+				b: nil,
+			},
+			wantAdded:   nil,
+			wantRemoved: nil,
+			wantUpdated: nil,
+		},
+		{
+			name: "empty arrays",
+			args: args{
+				a: []*v1alpha1.Tag{},
+				b: []*v1alpha1.Tag{},
+			},
+			wantAdded:   nil,
+			wantRemoved: nil,
+			wantUpdated: nil,
+		},
+		{
+			name: "added tags",
+			args: args{
+				a: []*v1alpha1.Tag{},
+				b: []*v1alpha1.Tag{Tag1, Tag2},
+			},
+			wantAdded:   []*v1alpha1.Tag{Tag1, Tag2},
+			wantRemoved: nil,
+			wantUpdated: nil,
+		},
+		{
+			name: "removed tags",
+			args: args{
+				a: []*v1alpha1.Tag{Tag1, Tag2},
+				b: nil,
+			},
+			wantAdded:   nil,
+			wantRemoved: []*string{aws.String("k1"), aws.String("k2")},
+			wantUpdated: nil,
+		},
+		{
+			name: "updated tags",
+			args: args{
+				a: []*v1alpha1.Tag{Tag1, Tag2},
+				b: []*v1alpha1.Tag{Tag1, Tag2Updated},
+			},
+			wantAdded:   nil,
+			wantRemoved: nil,
+			wantUpdated: []*v1alpha1.Tag{Tag2Updated},
+		},
+		{
+			name: "added, updated and removed tags",
+			args: args{
+				a: []*v1alpha1.Tag{Tag1, Tag2},
+				// remove Tag1, update Tag2 and add Tag3
+				b: []*v1alpha1.Tag{Tag2Updated, Tag3},
+			},
+			wantAdded:   []*v1alpha1.Tag{Tag3},
+			wantRemoved: []*string{aws.String("k1")},
+			wantUpdated: []*v1alpha1.Tag{Tag2Updated},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			gotAdded, gotUpdated, gotRemoved := computeTagsDelta(tt.args.a, tt.args.b)
+			if !reflect.DeepEqual(gotAdded, tt.wantAdded) {
+				t.Errorf("computeTagsDelta() gotAdded = %v, want %v", gotAdded, tt.wantAdded)
+			}
+			if !reflect.DeepEqual(gotUpdated, tt.wantUpdated) {
+				t.Errorf("computeTagsDelta() gotUpdated = %v, want %v", gotUpdated, tt.wantUpdated)
+			}
+			if !reflect.DeepEqual(gotRemoved, tt.wantRemoved) {
+				t.Errorf("computeTagsDelta() gotRemoved = %v, want %v", gotRemoved, tt.wantRemoved)
+			}
+		})
+	}
+}

Diff for: pkg/resource/table/hooks.go

@@ -3,4 +3,7 @@
 	}
 	if isTableUpdating(&resource{ko}) {
 		return &resource{ko}, requeueWaitWhileUpdating
+	}
+	if err := rm.setResourceAdditionalFields(ctx, ko); err != nil {
+		return nil, err
 	}

Diff for: pkg/resource/table/hooks_test.go

@@ -20,6 +20,7 @@
 
 from acktest.k8s import resource as k8s
 from acktest.resources import load_resource_file
+from acktest.aws.identity import get_region
 
 SERVICE_NAME = "dynamodb"
 CRD_GROUP = "dynamodb.services.k8s.aws"
@@ -60,4 +61,21 @@ def wait_for_cr_status(
             f"Wait for status: {desired_status} timed out. Actual status: {actual_status}"
         )
 
-    assert actual_status == desired_status
+    assert actual_status == desired_status
+
+def get_resource_tags(resource_arn: str):
+    region = get_region()
+    ddb_client = boto3.client('dynamodb', region_name=region)
+    tags = []
+    next_token = ""
+    while True:
+        resp = ddb_client.list_tags_of_resource(
+            ResourceArn=resource_arn,
+            NextToken=next_token,
+        )
+        tags += resp['Tags']
+        if not 'NextToken' in resp.keys():
+            break
+        next_token = resp['NextToken']
+    
+    return tags

Diff for: pkg/resource/table/sdk.go

@@ -7,10 +7,10 @@ spec:
   attributeDefinitions:
     - attributeName: ForumName
       attributeType: S
-    - attributeName: Subject
-      attributeType: S
     - attributeName: LastPostDateTime
       attributeType: S
+    - attributeName: Subject
+      attributeType: S
   keySchema:
     - attributeName: ForumName
       keyType: HASH

Diff for: templates/hooks/table/sdk_read_one_post_set_output.go.tpl

@@ -22,12 +22,17 @@
 
 from acktest.resources import random_suffix_name
 from acktest.k8s import resource as k8s
-from e2e import service_marker, CRD_GROUP, CRD_VERSION, load_dynamodb_resource, wait_for_cr_status
+from e2e import (
+    service_marker, CRD_GROUP, CRD_VERSION,
+    load_dynamodb_resource, wait_for_cr_status, 
+    get_resource_tags,
+)
 from e2e.replacement_values import REPLACEMENT_VALUES
 
 RESOURCE_PLURAL = "tables"
 
 DELETE_WAIT_AFTER_SECONDS = 15
+UPDATE_TAGS_WAIT_AFTER_SECONDS = 5
 
 @pytest.fixture(scope="module")
 def dynamodb_client():
@@ -50,7 +55,7 @@ def get_table(self, dynamodb_client, table_name: str) -> dict:
     def table_exists(self, dynamodb_client, table_name: str) -> bool:
         return self.get_table(dynamodb_client, table_name) is not None
 
-    def test_smoke(self, dynamodb_client):
+    def test_create_delete(self, dynamodb_client):
         resource_name = random_suffix_name("table", 32)
 
         replacements = REPLACEMENT_VALUES.copy()
@@ -92,6 +97,73 @@ def test_smoke(self, dynamodb_client):
 
         time.sleep(DELETE_WAIT_AFTER_SECONDS)
 
+        # Check DynamoDB Table doesn't exists
+        exists = self.table_exists(dynamodb_client, resource_name)
+        assert not exists
+
+    def test_table_update_tags(self, dynamodb_client):
+        resource_name = random_suffix_name("table", 32)
+
+        replacements = REPLACEMENT_VALUES.copy()
+        replacements["TABLE_NAME"] = resource_name
+
+        # Load Table CR
+        resource_data = load_dynamodb_resource(
+            "table_forums",
+            additional_replacements=replacements,
+        )
+        logging.debug(resource_data)
+
+        # Create k8s resource
+        ref = k8s.CustomResourceReference(
+            CRD_GROUP, CRD_VERSION, RESOURCE_PLURAL,
+            resource_name, namespace="default",
+        )
+        k8s.create_custom_resource(ref, resource_data)
+        cr = k8s.wait_resource_consumed_by_controller(ref)
+
+        assert cr is not None
+        assert k8s.get_resource_exists(ref)
+
+        wait_for_cr_status(
+            ref,
+            "tableStatus",
+            "ACTIVE",
+            10,
+            5,
+        )
+
+        # Check DynamoDB Table exists
+        exists = self.table_exists(dynamodb_client, resource_name)
+        assert exists
+
+        # Get CR latest revision
+        cr = k8s.wait_resource_consumed_by_controller(ref)
+
+        # Update table list of tags
+        tags = [
+            {
+                "key": "key1",
+                "value": "value1",
+            },
+        ]
+        cr["spec"]["tags"] = tags
+
+        # Patch k8s resource
+        k8s.patch_custom_resource(ref, cr)
+        time.sleep(UPDATE_TAGS_WAIT_AFTER_SECONDS)
+
+        table_tags = get_resource_tags(cr["status"]["ackResourceMetadata"]["arn"])
+        assert len(table_tags) == len(tags)
+        assert table_tags[0]['Key'] == tags[0]['key']
+        assert table_tags[0]['Value'] == tags[0]['value']
+
+        # Delete k8s resource
+        _, deleted = k8s.delete_custom_resource(ref)
+        assert deleted is True
+
+        time.sleep(DELETE_WAIT_AFTER_SECONDS)
+
         # Check DynamoDB Table doesn't exists
         exists = self.table_exists(dynamodb_client, resource_name)
         assert not exists