Add Tags to DMS Replication Instance

client/mocks/builders_test.go

@@ -307,6 +307,15 @@ func buildDmsReplicationInstances(t *testing.T, ctrl *gomock.Controller) client.
 		&databasemigrationservice.DescribeReplicationInstancesOutput{
 			ReplicationInstances: []databasemigrationserviceTypes.ReplicationInstance{l},
 		}, nil)
+	lt := databasemigrationserviceTypes.Tag{}
+	if err := faker.FakeData(&lt); err != nil {
+		t.Fatal(err)
+	}
+	lt.ResourceArn = l.ReplicationInstanceArn
+	m.EXPECT().ListTagsForResource(gomock.Any(), gomock.Any(), gomock.Any()).Return(
+		&databasemigrationservice.ListTagsForResourceOutput{
+			TagList: []databasemigrationserviceTypes.Tag{lt},
+		}, nil)
 	return client.Services{
 		DMS: m,
 	}

client/services.go

@@ -147,6 +147,7 @@ type DirectconnectClient interface {
 //go:generate mockgen -package=mocks -destination=./mocks/mock_databasemigrationservice.go . DatabasemigrationserviceClient
 type DatabasemigrationserviceClient interface {
 	DescribeReplicationInstances(ctx context.Context, params *databasemigrationservice.DescribeReplicationInstancesInput, optFns ...func(*databasemigrationservice.Options)) (*databasemigrationservice.DescribeReplicationInstancesOutput, error)
+	ListTagsForResource(ctx context.Context, params *databasemigrationservice.ListTagsForResourceInput, optFns ...func(*databasemigrationservice.Options)) (*databasemigrationservice.ListTagsForResourceOutput, error)
 }
 
 //go:generate mockgen -package=mocks -destination=./mocks/mock_ec2.go . Ec2Client

resources/dms_replication_instances.go

@@ -19,6 +19,7 @@ func DmsReplicationInstances() *schema.Table {
 		Multiplex:    client.AccountRegionMultiplex,
 		IgnoreError:  client.IgnoreAccessDeniedServiceDisabled,
 		DeleteFilter: client.DeleteAccountRegionFilter,
+		Options:      schema.TableCreationOptions{PrimaryKeys: []string{"account_id", "arn"}},
 		Columns: []schema.Column{
 			{
 				Name:        "account_id",
@@ -32,6 +33,11 @@ func DmsReplicationInstances() *schema.Table {
 				Type:        schema.TypeString,
 				Resolver:    client.ResolveAWSRegion,
 			},
+			{
+				Name:        "tags",
+				Description: "Any tags assigned to the resource",
+				Type:        schema.TypeJSON,
+			},
 			{
 				Name:        "allocated_storage",
 				Description: "The amount of storage (in gigabytes) that is allocated for the replication instance.",
@@ -251,21 +257,45 @@ func DmsReplicationInstances() *schema.Table {
 //                                               Table Resolver Functions
 // ====================================================================================================================
 func fetchDmsReplicationInstances(ctx context.Context, meta schema.ClientMeta, _ *schema.Resource, res chan interface{}) error {
-	var config databasemigrationservice.DescribeReplicationInstancesInput
 	c := meta.(*client.Client)
 	svc := c.Services().DMS
-	output, err := svc.DescribeReplicationInstances(ctx, &config, func(options *databasemigrationservice.Options) {
+
+	var describeReplicationInstancesInput *databasemigrationservice.DescribeReplicationInstancesInput
+	describeReplicationInstancesOutput, err := svc.DescribeReplicationInstances(ctx, describeReplicationInstancesInput, func(options *databasemigrationservice.Options) {
 		options.Region = c.Region
 	})
 	if err != nil {
 		return err
 	}
-	res <- output.ReplicationInstances
+
+	listTagsForResourceInput := databasemigrationservice.ListTagsForResourceInput{}
+	for _, replicationInstance := range describeReplicationInstancesOutput.ReplicationInstances {
+		listTagsForResourceInput.ResourceArnList = append(listTagsForResourceInput.ResourceArnList, *replicationInstance.ReplicationInstanceArn)
+	}
+	var listTagsForResourceOutput *databasemigrationservice.ListTagsForResourceOutput
+	listTagsForResourceOutput, err = svc.ListTagsForResource(ctx, &listTagsForResourceInput, func(options *databasemigrationservice.Options) {
+		options.Region = c.Region
+	})
+	replicationInstanceTags := make(map[string]map[string]interface{})
+	for _, tag := range listTagsForResourceOutput.TagList {
+		if replicationInstanceTags[*tag.ResourceArn] == nil {
+			replicationInstanceTags[*tag.ResourceArn] = make(map[string]interface{})
+		}
+		replicationInstanceTags[*tag.ResourceArn][*tag.Key] = *tag.Value
+	}
+
+	for _, replicationInstance := range describeReplicationInstancesOutput.ReplicationInstances {
+		wrapper := DmsReplicationInstanceWrapper{
+			ReplicationInstance: replicationInstance,
+			Tags:                replicationInstanceTags[*replicationInstance.ReplicationInstanceArn],
+		}
+		res <- wrapper
+	}
 	return nil
 }
 
 func fetchDmsReplicationInstanceReplicationSubnetGroupSubnets(_ context.Context, _ schema.ClientMeta, parent *schema.Resource, res chan interface{}) error {
-	replicationInstance, ok := parent.Item.(types.ReplicationInstance)
+	replicationInstance, ok := parent.Item.(DmsReplicationInstanceWrapper)
 	if !ok {
 		return fmt.Errorf("not dms replication instance")
 	}
@@ -274,10 +304,15 @@ func fetchDmsReplicationInstanceReplicationSubnetGroupSubnets(_ context.Context,
 }
 
 func fetchDmsReplicationInstanceVpcSecurityGroups(_ context.Context, _ schema.ClientMeta, parent *schema.Resource, res chan interface{}) error {
-	replicationInstance, ok := parent.Item.(types.ReplicationInstance)
+	replicationInstance, ok := parent.Item.(DmsReplicationInstanceWrapper)
 	if !ok {
 		return fmt.Errorf("not dms replication instance")
 	}
 	res <- replicationInstance.VpcSecurityGroups
 	return nil
 }
+
+type DmsReplicationInstanceWrapper struct {
+	types.ReplicationInstance
+	Tags map[string]interface{}
+}