Tech debt: Migrate kinesis resources to AWS SDK for Go v2

.changelog/35167.txt

@@ -0,0 +1,3 @@
+```release-note:new-resource
+aws_kinesis_resource_policy
+```

.github/labeler-pr-triage.yml

@@ -604,6 +604,7 @@ service/keyspaces:
 service/kinesis:
   - 'internal/service/kinesis/**/*'
   - 'website/**/kinesis_stream*'
+  - 'website/**/kinesis_resource_policy*'
 service/kinesisanalytics:
   - 'internal/service/kinesisanalytics/**/*'
   - 'website/**/kinesis_analytics_*'

go.mod

@@ -58,6 +58,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/kafka v1.28.6
 	github.com/aws/aws-sdk-go-v2/service/kendra v1.47.6
 	github.com/aws/aws-sdk-go-v2/service/keyspaces v1.7.6
+	github.com/aws/aws-sdk-go-v2/service/kinesis v1.24.7
 	github.com/aws/aws-sdk-go-v2/service/lambda v1.49.7
 	github.com/aws/aws-sdk-go-v2/service/lexmodelsv2 v1.38.6
 	github.com/aws/aws-sdk-go-v2/service/lightsail v1.33.0

go.sum

@@ -153,6 +153,8 @@ github.com/aws/aws-sdk-go-v2/service/kendra v1.47.6 h1:F/U90/tSb08JFrtztE2zvqG7g
 github.com/aws/aws-sdk-go-v2/service/kendra v1.47.6/go.mod h1:HRTCOLfl+Y63RoSezT8bhl1olWzEWNDwBOXCoWT02rw=
 github.com/aws/aws-sdk-go-v2/service/keyspaces v1.7.6 h1:UWYKr1x1B4Dk91BwJoCmJA9gAw01XjWQ+wFFM+60Ifw=
 github.com/aws/aws-sdk-go-v2/service/keyspaces v1.7.6/go.mod h1:xQtrNWsoGMqoQ/Xw/x+4zprlCMfkpFKgSku0n9v8Or4=
+github.com/aws/aws-sdk-go-v2/service/kinesis v1.24.7 h1:7Xy/miw2n9G6yi0qHey8Ro2pHR93cMB/r/PMXLMeZrI=
+github.com/aws/aws-sdk-go-v2/service/kinesis v1.24.7/go.mod h1:xOJOknNQF6owzT/d+ivXnNK7M+swiglnobX+zekpS6s=
 github.com/aws/aws-sdk-go-v2/service/lambda v1.49.7 h1:YCvhGwdiZ9tKTjoIOE8jLt+3JBK4quAQyhoMCWtxhQc=
 github.com/aws/aws-sdk-go-v2/service/lambda v1.49.7/go.mod h1:xqjYGK1M7YTmyfZBW8LVAx7QnefUb/mE5BglUnxtx6E=
 github.com/aws/aws-sdk-go-v2/service/lexmodelsv2 v1.38.6 h1:ga1Fi7oHPIGIRZCvklA/+Wnn7h8YQgGdDGu7lmtAWsE=

internal/service/kinesis/exports_test.go

@@ -0,0 +1,15 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package kinesis
+
+// Exports for use in tests only.
+var (
+	ResourceResourcePolicy = newResourcePolicyResource
+	ResourceStream         = resourceStream
+	ResourceStreamConsumer = resourceStreamConsumer
+
+	FindResourcePolicyByResourceARN = findResourcePolicyByResourceARN
+	FindStreamByName                = findStreamByName
+	FindStreamConsumerByARN         = findStreamConsumerByARN
+)

internal/service/kinesis/generate.go

@@ -1,7 +1,7 @@
 // Copyright (c) HashiCorp, Inc.
 // SPDX-License-Identifier: MPL-2.0
 
-//go:generate go run ../../generate/tags/main.go -ListTags -ListTagsOp=ListTagsForStream -ListTagsInIDElem=StreamName -ServiceTagsSlice -TagOp=AddTagsToStream -TagOpBatchSize=10 -TagInCustomVal=aws.StringMap(updatedTags.IgnoreAWS().Map()) -TagInIDElem=StreamName -UntagOp=RemoveTagsFromStream -UpdateTags -CreateTags
+//go:generate go run ../../generate/tags/main.go -AWSSDKVersion=2 -ListTags -ListTagsOp=ListTagsForStream -ListTagsInIDElem=StreamName -ServiceTagsSlice -TagOp=AddTagsToStream -TagOpBatchSize=10 -TagInCustomVal=updatedTags.IgnoreAWS().Map() -TagInIDElem=StreamName -UntagOp=RemoveTagsFromStream -UpdateTags -CreateTags
 //go:generate go run ../../generate/servicepackage/main.go
 // ONLY generate directives and package declaration! Do not add anything else to this file.
 

internal/service/kinesis/list_pages.go

@@ -0,0 +1,30 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package kinesis
+
+import (
+	"context"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/kinesis"
+)
+
+// Custom Kinesis service lister functions using the same format as generated code.
+
+func listShardsPages(ctx context.Context, conn *kinesis.Client, input *kinesis.ListShardsInput, fn func(*kinesis.ListShardsOutput, bool) bool) error {
+	for {
+		output, err := conn.ListShards(ctx, input)
+		if err != nil {
+			return err
+		}
+
+		lastPage := aws.ToString(output.NextToken) == ""
+		if !fn(output, lastPage) || lastPage {
+			break
+		}
+
+		input.NextToken = output.NextToken
+	}
+	return nil
+}

internal/service/kinesis/resource_policy.go

@@ -0,0 +1,228 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package kinesis
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/kinesis"
+	awstypes "github.com/aws/aws-sdk-go-v2/service/kinesis/types"
+	"github.com/hashicorp/terraform-plugin-framework/resource"
+	"github.com/hashicorp/terraform-plugin-framework/resource/schema"
+	"github.com/hashicorp/terraform-plugin-framework/resource/schema/planmodifier"
+	"github.com/hashicorp/terraform-plugin-framework/resource/schema/stringplanmodifier"
+	"github.com/hashicorp/terraform-plugin-framework/types"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/retry"
+	"github.com/hashicorp/terraform-provider-aws/internal/errs"
+	"github.com/hashicorp/terraform-provider-aws/internal/errs/fwdiag"
+	"github.com/hashicorp/terraform-provider-aws/internal/framework"
+	"github.com/hashicorp/terraform-provider-aws/internal/framework/flex"
+	fwtypes "github.com/hashicorp/terraform-provider-aws/internal/framework/types"
+	"github.com/hashicorp/terraform-provider-aws/internal/tfresource"
+	"github.com/hashicorp/terraform-provider-aws/names"
+)
+
+// @FrameworkResource(name="Resource Policy")
+func newResourcePolicyResource(context.Context) (resource.ResourceWithConfigure, error) {
+	r := &resourcePolicyResource{}
+
+	return r, nil
+}
+
+type resourcePolicyResource struct {
+	framework.ResourceWithConfigure
+	framework.WithImportByID
+}
+
+func (r *resourcePolicyResource) Metadata(_ context.Context, request resource.MetadataRequest, response *resource.MetadataResponse) {
+	response.TypeName = "aws_kinesis_resource_policy"
+}
+
+func (r *resourcePolicyResource) Schema(ctx context.Context, request resource.SchemaRequest, response *resource.SchemaResponse) {
+	response.Schema = schema.Schema{
+		Attributes: map[string]schema.Attribute{
+			names.AttrID: framework.IDAttribute(),
+			"policy": schema.StringAttribute{
+				CustomType: fwtypes.IAMPolicyType,
+				Required:   true,
+			},
+			"resource_arn": schema.StringAttribute{
+				CustomType: fwtypes.ARNType,
+				Required:   true,
+				PlanModifiers: []planmodifier.String{
+					stringplanmodifier.RequiresReplace(),
+				},
+			},
+		},
+	}
+}
+
+func (r *resourcePolicyResource) Create(ctx context.Context, request resource.CreateRequest, response *resource.CreateResponse) {
+	var data resourcePolicyResourceModel
+
+	response.Diagnostics.Append(request.Plan.Get(ctx, &data)...)
+	if response.Diagnostics.HasError() {
+		return
+	}
+
+	conn := r.Meta().KinesisClient(ctx)
+
+	input := &kinesis.PutResourcePolicyInput{
+		Policy:      flex.StringFromFramework(ctx, data.Policy),
+		ResourceARN: flex.StringFromFramework(ctx, data.ResourceARN),
+	}
+
+	_, err := conn.PutResourcePolicy(ctx, input)
+
+	if err != nil {
+		response.Diagnostics.AddError(fmt.Sprintf("creating Kinesis Resource Policy (%s)", data.ResourceARN.ValueString()), err.Error())
+
+		return
+	}
+
+	// Set values for unknowns.
+	data.setID()
+
+	response.Diagnostics.Append(response.State.Set(ctx, &data)...)
+}
+
+func (r *resourcePolicyResource) Read(ctx context.Context, request resource.ReadRequest, response *resource.ReadResponse) {
+	var data resourcePolicyResourceModel
+
+	response.Diagnostics.Append(request.State.Get(ctx, &data)...)
+
+	if response.Diagnostics.HasError() {
+		return
+	}
+
+	if err := data.InitFromID(); err != nil {
+		response.Diagnostics.AddError("parsing resource ID", err.Error())
+
+		return
+	}
+
+	conn := r.Meta().KinesisClient(ctx)
+
+	output, err := findResourcePolicyByResourceARN(ctx, conn, data.ResourceARN.ValueString())
+
+	if tfresource.NotFound(err) {
+		response.Diagnostics.Append(fwdiag.NewResourceNotFoundWarningDiagnostic(err))
+		response.State.RemoveResource(ctx)
+
+		return
+	}
+
+	if err != nil {
+		response.Diagnostics.AddError(fmt.Sprintf("reading Kinesis Resource Policy (%s)", data.ID.ValueString()), err.Error())
+
+		return
+	}
+
+	// Set attributes for import.
+	response.Diagnostics.Append(flex.Flatten(ctx, output, &data)...)
+	if response.Diagnostics.HasError() {
+		return
+	}
+
+	response.Diagnostics.Append(response.State.Set(ctx, &data)...)
+}
+
+func (r *resourcePolicyResource) Update(ctx context.Context, request resource.UpdateRequest, response *resource.UpdateResponse) {
+	var old, new resourcePolicyResourceModel
+
+	response.Diagnostics.Append(request.State.Get(ctx, &old)...)
+	if response.Diagnostics.HasError() {
+		return
+	}
+
+	response.Diagnostics.Append(request.Plan.Get(ctx, &new)...)
+	if response.Diagnostics.HasError() {
+		return
+	}
+
+	conn := r.Meta().KinesisClient(ctx)
+
+	input := &kinesis.PutResourcePolicyInput{
+		Policy:      flex.StringFromFramework(ctx, new.Policy),
+		ResourceARN: flex.StringFromFramework(ctx, new.ResourceARN),
+	}
+
+	_, err := conn.PutResourcePolicy(ctx, input)
+
+	if err != nil {
+		response.Diagnostics.AddError(fmt.Sprintf("updating Kinesis Resource Policy (%s)", new.ID.ValueString()), err.Error())
+
+		return
+	}
+
+	response.Diagnostics.Append(response.State.Set(ctx, &new)...)
+}
+
+func (r *resourcePolicyResource) Delete(ctx context.Context, request resource.DeleteRequest, response *resource.DeleteResponse) {
+	var data resourcePolicyResourceModel
+
+	response.Diagnostics.Append(request.State.Get(ctx, &data)...)
+	if response.Diagnostics.HasError() {
+		return
+	}
+
+	conn := r.Meta().KinesisClient(ctx)
+
+	_, err := conn.DeleteResourcePolicy(ctx, &kinesis.DeleteResourcePolicyInput{
+		ResourceARN: flex.StringFromFramework(ctx, data.ResourceARN),
+	})
+
+	if errs.IsA[*awstypes.ResourceNotFoundException](err) {
+		return
+	}
+
+	if err != nil {
+		response.Diagnostics.AddError(fmt.Sprintf("deleting Kinesis Resource Policy (%s)", data.ID.ValueString()), err.Error())
+
+		return
+	}
+}
+
+func findResourcePolicyByResourceARN(ctx context.Context, conn *kinesis.Client, resourceARN string) (*kinesis.GetResourcePolicyOutput, error) {
+	input := &kinesis.GetResourcePolicyInput{
+		ResourceARN: aws.String(resourceARN),
+	}
+
+	output, err := conn.GetResourcePolicy(ctx, input)
+
+	if errs.IsA[*awstypes.ResourceNotFoundException](err) {
+		return nil, &retry.NotFoundError{
+			LastError:   err,
+			LastRequest: input,
+		}
+	}
+
+	if err != nil {
+		return nil, err
+	}
+
+	if output == nil || output.Policy == nil {
+		return nil, tfresource.NewEmptyResultError(input)
+	}
+
+	return output, nil
+}
+
+type resourcePolicyResourceModel struct {
+	ID          types.String      `tfsdk:"id"`
+	Policy      fwtypes.IAMPolicy `tfsdk:"policy"`
+	ResourceARN fwtypes.ARN       `tfsdk:"resource_arn"`
+}
+
+func (data *resourcePolicyResourceModel) InitFromID() error {
+	data.ResourceARN = fwtypes.ARNValue(data.ID.ValueString())
+
+	return nil
+}
+
+func (data *resourcePolicyResourceModel) setID() {
+	data.ID = data.ResourceARN.StringValue
+}