feat: Add support for LogGroup Tags Update (#42)

Issue [#2163](aws-controllers-k8s/community#2163)

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

Author: michaelhtm

pkg/resource/log_group/hook.go

@@ -22,6 +22,7 @@ import (
 	svcsdktypes "github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs/types"
 	"github.com/aws/aws-sdk-go/aws"
 
+	"github.com/aws-controllers-k8s/cloudwatchlogs-controller/pkg/sync"
 	svcapitypes "github.com/aws-controllers-k8s/cloudwatchlogs-controller/apis/v1alpha1"
 )
 
@@ -137,6 +138,8 @@ func (rm *resourceManager) removeSubscriptionFilter(
 	return output, nil
 }
 
+var getTags = sync.GetResourceTags
+
 // customUpdateLogGroup patches each of the resource properties in the backend AWS
 // service API and returns a new resource with updated fields.
 func (rm *resourceManager) customUpdateLogGroup(
@@ -152,9 +155,15 @@ func (rm *resourceManager) customUpdateLogGroup(
 	// 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()
-
+	ko.Status = latest.ko.Status
 	rm.setStatusDefaults(ko)
 
+	if delta.DifferentAt("Spec.Tags") {
+		err = sync.SyncResourceTags(ctx, rm.sdkapi, rm.metrics, string(*latest.ko.Status.ACKResourceMetadata.ARN), desired.ko.Spec.Tags, latest.ko.Spec.Tags, convertToOrderedACKTags)
+		if err != nil {
+			return &resource{ko}, err
+		}
+	}
 	if delta.DifferentAt("Spec.RetentionDays") {
 		if err := rm.updateRetentionPeriod(ctx, desired); err != nil {
 			return &resource{ko}, err

pkg/resource/log_group/sdk.go

@@ -0,0 +1,118 @@
+// 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 sync
+
+import (
+	"context"
+
+	ackcompare "github.com/aws-controllers-k8s/runtime/pkg/compare"
+	ackrtlog "github.com/aws-controllers-k8s/runtime/pkg/runtime/log"
+	acktags "github.com/aws-controllers-k8s/runtime/pkg/tags"
+	svcsdk "github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs"
+)
+
+type metricsRecorder interface {
+	RecordAPICall(opType string, opID string, err error)
+}
+
+type tagsClient interface {
+	TagResource(context.Context, *svcsdk.TagResourceInput, ...func(*svcsdk.Options)) (*svcsdk.TagResourceOutput, error)
+	ListTagsForResource(context.Context, *svcsdk.ListTagsForResourceInput, ...func(*svcsdk.Options)) (*svcsdk.ListTagsForResourceOutput, error)
+	UntagResource(context.Context, *svcsdk.UntagResourceInput, ...func(*svcsdk.Options)) (*svcsdk.UntagResourceOutput, error)
+}
+
+// GetResourceTags retrieves a resource list of tags.
+func GetResourceTags(
+	ctx context.Context,
+	client tagsClient,
+	mr metricsRecorder,
+	resourceARN string,
+) (map[string]*string, error) {
+	listTagsForResourceResponse, err := client.ListTagsForResource(
+		ctx,
+		&svcsdk.ListTagsForResourceInput{
+			ResourceArn: &resourceARN,
+		},
+	)
+	mr.RecordAPICall("GET", "ListTagsForResource", err)
+	if err != nil {
+		return nil, err
+	}
+	tags := map[string]*string{}
+	for key, val := range listTagsForResourceResponse.Tags {
+		tags[key] = &val
+	}
+	return tags, nil
+}
+
+// SyncResourceTags uses TagResource and UntagResource API Calls to add, remove
+// and update resource tags.
+func SyncResourceTags(
+	ctx context.Context,
+	client tagsClient,
+	mr metricsRecorder,
+	resourceARN string,
+	desiredTags map[string]*string,
+	currentTags map[string]*string,
+	convertToOrderedACKTags func(tags map[string]*string) (acktags.Tags, []string),
+) (err error) {
+	rlog := ackrtlog.FromContext(ctx)
+	exit := rlog.Trace("SyncResourceTags")
+	defer func() { exit(err) }()
+
+	desiredACKTags, _ := convertToOrderedACKTags(desiredTags)
+	currentACKTags, _ := convertToOrderedACKTags(currentTags)
+
+	added, _, toRemove := ackcompare.GetTagsDifference(currentACKTags, desiredACKTags)
+
+	for key := range toRemove {
+		if _, ok := added[key]; ok {
+			delete(toRemove, key)
+		}
+	}
+
+	var removed []string
+	for key := range toRemove {
+		removed = append(removed, key)
+	}
+
+	if len(removed) > 0 {
+		_, err = client.UntagResource(
+			ctx,
+			&svcsdk.UntagResourceInput{
+				ResourceArn: &resourceARN,
+				TagKeys:     removed,
+			},
+		)
+		mr.RecordAPICall("UPDATE", "UntagResource", err)
+		if err != nil {
+			return err
+		}
+	}
+
+	if len(added) > 0 {
+		_, err = client.TagResource(
+			ctx,
+			&svcsdk.TagResourceInput{
+				ResourceArn: &resourceARN,
+				Tags:        added,
+			},
+		)
+		mr.RecordAPICall("UPDATE", "TagResource", err)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}

pkg/sync/tags.go

@@ -4,4 +4,8 @@
 	ko.Spec.SubscriptionFilters, err = rm.getSubscriptionFilters(ctx, r.ko.Spec.Name)
 	if err != nil {
 		return nil, err
-	}
+	}
+	ko.Spec.Tags, err = getTags(ctx, rm.sdkapi, rm.metrics, string(*ko.Status.ACKResourceMetadata.ARN))
+	if err != nil {
+		return nil, err
+	}

templates/hooks/log_group/sdk_read_many_post_set_output.go.tpl

@@ -103,8 +103,8 @@ def get_tags(log_group_arn):
     c = boto3.client('logs')
     try:
         resp = c.list_tags_for_resource(
-            ResourceName=log_group_arn,
+            resourceArn=log_group_arn,
         )
-        return resp['TagList']
-    except c.exceptions.LogGroupNotFoundFault:
+        return resp['tags']
+    except c.exceptions.ResourceNotFoundException:
         return None

test/e2e/log_group.py

@@ -17,7 +17,7 @@
 import time
 
 import pytest
-
+from acktest import tags
 from acktest.k8s import resource as k8s
 from acktest.resources import random_suffix_name
 from e2e import service_marker, CRD_GROUP, CRD_VERSION, load_resource
@@ -87,19 +87,38 @@ def test_crud(self, _log_group):
         cr = k8s.get_resource(ref)
         assert 'creationTime' in cr['status']
         assert cr['status']['creationTime'] > 0
+        assert 'ackResourceMetadata' in cr['status']
+        assert 'arn' in cr['status']['ackResourceMetadata']
+        arn = cr['status']['ackResourceMetadata']['arn']
 
         # update retention period
         updated_retention = 3
         updates = {
             "spec": {
-                "retentionDays": updated_retention
+                "retentionDays": updated_retention,
+                "tags": {
+                    "newKey": "newVal"
+                }
             }
         }
 
         k8s.patch_custom_resource(ref, updates)
         time.sleep(UPDATE_WAIT_AFTER_SECONDS)
 
         assert log_group.exists_with_retention_period(log_group_name, updated_retention)
+        cr = k8s.get_resource(ref)
+
+        latest_tags = log_group.get_tags(arn)
+        desired_tags = cr['spec']['tags']
+        tags.assert_ack_system_tags(
+            tags=latest_tags,
+        )
+        tags.assert_equal_without_ack_tags(
+            expected=desired_tags,
+            actual=latest_tags,
+        )
+
+        
 
     @pytest.mark.resource_data({'resource_file': 'invalid/log_group_invalid_parameter'})
     def test_terminal_condition_invalid_parameter(self, _log_group):