feat: Supports new AWSLambda connection type in streams processing (#3085)

* feat: Supports new AWSLambda connection type in streams processing

* feat: renaming and adding acc tests

* feat: add change log

* feat: minor change

* feat: add env vars to GH workflow

* CLOUDP-300287: merge master

* CLOUDP-300287: merge master

* CLOUDP-300287: merge master

* CLOUDP-300287: fix doc per comment

* changelog

* fix example

* trigger test fix

* Trigger Build

* fix workflow

* CLOUDP-300287: fix lint error

* CLOUDP-300287: test accpt test

* CLOUDP-300287: test accpt test

* CLOUDP-300287: test accpt test

* CLOUDP-300287: test accpt test

* CLOUDP-300287: test accpt test

* fix acc test

* don't hardcode project

* CLOUDP-300287: refactor by comments

* CLOUDP-300287: refactor by comments

* CLOUDP-300287: refactor by comments

* CLOUDP-300287: refactor by comments

---------

Co-authored-by: Oriol Arbusi <[email protected]>

Author: LijieZhang1998

.changelog/3085.txt

@@ -0,0 +1,7 @@
+```release-note:enhancement
+resource/mongodbatlas_stream_connection: Adds `AWSLambda` connection
+```
+
+```release-note:enhancement
+data-source/mongodbatlas_stream_connection: Adds `AWSLambda` connection
+```

.github/workflows/acceptance-tests-runner.yml

@@ -115,6 +115,12 @@ on:
       mongodb_atlas_project_ear_pe_aws_id:
         type: string
         required: true
+      mongodb_atlas_asp_project_ear_pe_id:
+        type: string
+        required: true
+      mongodb_atlas_asp_project_aws_role_arn:
+        type: string
+        required: true
     secrets: # all secrets are passed explicitly in this workflow
       mongodb_atlas_public_key:
         required: true
@@ -646,6 +652,8 @@ jobs:
           AWS_CUSTOMER_MASTER_KEY_ID: ${{ secrets.aws_customer_master_key_id }}
           MONGODB_ATLAS_PROJECT_EAR_PE_AWS_ID: ${{ inputs.mongodb_atlas_project_ear_pe_aws_id }}
           AWS_EAR_ROLE_ID: ${{ inputs.aws_ear_role_id }}
+          MONGODB_ATLAS_ASP_PROJECT_EAR_PE_ID: ${{ inputs.mongodb_atlas_asp_project_ear_pe_id }}
+          MONGODB_ATLAS_ASP_PROJECT_AWS_ROLE_ARN: ${{ inputs.mongodb_atlas_asp_project_aws_role_arn }}
         run: make testacc
 
   event_trigger:
@@ -1063,6 +1071,8 @@ jobs:
             AWS_REGION: ${{ vars.AWS_REGION }}
             AWS_VPC_CIDR_BLOCK: ${{ vars.AWS_VPC_CIDR_BLOCK }}
             AWS_VPC_ID: ${{ vars.AWS_VPC_ID }}
+            MONGODB_ATLAS_ASP_PROJECT_AWS_ROLE_ARN: ${{ inputs.mongodb_atlas_asp_project_aws_role_arn }}
+            MONGODB_ATLAS_ASP_PROJECT_EAR_PE_ID: ${{ inputs.mongodb_atlas_asp_project_ear_pe_id }}
             ACCTEST_PACKAGES: |
               ./internal/service/streamconnection
               ./internal/service/streaminstance

.github/workflows/acceptance-tests.yml

@@ -130,4 +130,6 @@ jobs:
       mongodb_atlas_rp_org_id: ${{ inputs.atlas_cloud_env == 'qa' && vars.MONGODB_ATLAS_RP_ORG_ID_QA || vars.MONGODB_ATLAS_RP_ORG_ID_DEV }}
       mongodb_atlas_flex_project_id: ${{ inputs.atlas_cloud_env == 'qa' && vars.MONGODB_ATLAS_FLEX_PROJECT_ID_QA || vars.MONGODB_ATLAS_FLEX_PROJECT_ID }} 
       confluent_cloud_network_id: ${{ vars.CONFLUENT_CLOUD_NETWORK_ID }} 
-      confluent_cloud_privatelink_access_id: ${{ vars.CONFLUENT_CLOUD_PRIVATELINK_ACCESS_ID }} 
+      confluent_cloud_privatelink_access_id: ${{ vars.CONFLUENT_CLOUD_PRIVATELINK_ACCESS_ID }}
+      mongodb_atlas_asp_project_ear_pe_id: ${{ inputs.atlas_cloud_env == 'qa' && vars.MONGODB_ATLAS_ASP_PROJECT_EAR_PE_ID_QA || vars.MONGODB_ATLAS_ASP_PROJECT_EAR_PE_ID_DEV }}
+      mongodb_atlas_asp_project_aws_role_arn: ${{ inputs.atlas_cloud_env == 'qa' && vars.MONGODB_ATLAS_ASP_PROJECT_AWS_ROLE_ARN_QA || vars.MONGODB_ATLAS_ASP_PROJECT_AWS_ROLE_ARN_DEV }}

.github/workflows/code-health.yml

@@ -9,8 +9,8 @@ on:
       - '*.md'
       - 'examples/**'
       - 'LICENSE'
-  pull_request: {}
-  workflow_dispatch: {}
+  pull_request:
+  workflow_dispatch:
 
 jobs:
   build:

docs/data-sources/stream_connection.md

@@ -20,7 +20,7 @@ data "mongodbatlas_stream_connection" "example" {
 
 ## Attributes Reference
 
-* `type` - Type of connection. Can be either `Cluster`, `Kafka` or `Sample`.
+* `type` - Type of connection. Can be `Cluster`, `Kafka`, `Sample`, or `AWSLambda`.
 
 If `type` is of value `Cluster` the following additional attributes are defined:
 * `cluster_name` - Name of the cluster configured for this connection.
@@ -33,6 +33,9 @@ If `type` is of value `Kafka` the following additional attributes are defined:
 * `security` - Properties for the secure transport connection to Kafka. For SSL, this can include the trusted certificate to use. See [security](#security).
 * `networking` - Networking Access Type can either be `PUBLIC` (default) or `VPC`. See [networking](#networking).
 
+If `type` is of value `AWSLambda` the following additional attributes are defined:
+* `aws` - The configuration for AWS Lambda connection. See [AWS](#AWS)
+
 ### Authentication
 
 * `mechanism` - Style of authentication. Can be one of `PLAIN`, `SCRAM-256`, or `SCRAM-512`.
@@ -56,5 +59,8 @@ If `type` is of value `Kafka` the following additional attributes are defined:
 * `type` - Selected networking type. Either `PUBLIC`, `VPC` or `PRIVATE_LINK`. Defaults to `PUBLIC`.
 * `connection_id` - Id of the Private Link connection when type is `PRIVATE_LINK`.
 
+### AWS
+* `role_arn` - Amazon Resource Name (ARN) that identifies the Amazon Web Services (AWS) Identity and Access Management (IAM) role that MongoDB Cloud assumes when it accesses resources in your AWS account. 
+
 To learn more, see: [MongoDB Atlas API - Stream Connection](https://www.mongodb.com/docs/atlas/reference/api-resources-spec/#tag/Streams/operation/getStreamConnection) Documentation.
 The [Terraform Provider Examples Section](https://github.com/mongodb/terraform-provider-mongodbatlas/blob/master/examples/mongodbatlas_stream_instance/atlas-streams-user-journey.md) also contains details on the overall support for Atlas Streams Processing in Terraform.

docs/data-sources/stream_connections.md

@@ -32,7 +32,7 @@ In addition to all arguments above, it also exports the following attributes:
 * `project_id` - Unique 24-hexadecimal digit string that identifies your project.
 * `instance_name` - Human-readable label that identifies the stream instance.
 * `connection_name` - Human-readable label that identifies the stream connection. In the case of the Sample type, this is the name of the sample source.
-* `type` - Type of connection. Can be either `Cluster`, `Kafka` or `Sample`.
+* `type` - Type of connection. `Cluster`, `Kafka`, `Sample`, or `AWSLambda`.
 
 If `type` is of value `Cluster` the following additional attributes are defined:
 * `cluster_name` - Name of the cluster configured for this connection.
@@ -45,6 +45,9 @@ If `type` is of value `Kafka` the following additional attributes are defined:
 * `security` - Properties for the secure transport connection to Kafka. For SSL, this can include the trusted certificate to use. See [security](#security).
 * `networking` - Networking Access Type can either be `PUBLIC` (default) or `VPC`. See [networking](#networking).
 
+If `type` is of value `AWSLambda` the following additional attributes are defined::
+* `aws` - The configuration for AWS Lambda connection. See [AWS](#AWS)
+
 ### Authentication
 
 * `mechanism` - Style of authentication. Can be one of `PLAIN`, `SCRAM-256`, or `SCRAM-512`.
@@ -68,5 +71,8 @@ If `type` is of value `Kafka` the following additional attributes are defined:
 * `type` - Selected networking type. Either `PUBLIC`, `VPC` or `PRIVATE_LINK`. Defaults to `PUBLIC`.
 * `connection_id` - Id of the Private Link connection when type is `PRIVATE_LINK`.
 
+### AWS
+* `role_arn` - Amazon Resource Name (ARN) that identifies the Amazon Web Services (AWS) Identity and Access Management (IAM) role that MongoDB Cloud assumes when it accesses resources in your AWS account.
+
 To learn more, see: [MongoDB Atlas API - Stream Connection](https://www.mongodb.com/docs/atlas/reference/api-resources-spec/#tag/Streams/operation/listStreamConnections) Documentation.
 The [Terraform Provider Examples Section](https://github.com/mongodb/terraform-provider-mongodbatlas/blob/master/examples/mongodbatlas_stream_instance/atlas-streams-user-journey.md) also contains details on the overall support for Atlas Streams Processing in Terraform.

docs/resources/stream_connection.md

@@ -66,12 +66,27 @@ resource "mongodbatlas_stream_connection" "test" {
 }
 ```
 
+### Example AWSLambda Connection
+
+```terraform
+resource "mongodbatlas_stream_connection" "test" {
+    project_id      = var.project_id
+    instance_name   = "NewInstance"
+    connection_name = "AWSLambdaConnection"
+    type            = "AWSLambda"
+    aws             = {
+      role_arn = "arn:aws:iam::123456789123:role/lambdaRole"
+    }
+}
+
+```
+
 ## Argument Reference
 
 * `project_id` - (Required) Unique 24-hexadecimal digit string that identifies your project.
 * `instance_name` - (Required) Human-readable label that identifies the stream instance.
 * `connection_name` - (Required) Human-readable label that identifies the stream connection. In the case of the Sample type, this is the name of the sample source.
-* `type` - (Required) Type of connection. Can be either `Cluster`, `Kafka` or `Sample`.
+* `type` - (Required) Type of connection. Can be `Cluster`, `Kafka`, `Sample`, or `AWSLambda`.
 
 If `type` is of value `Cluster` the following additional arguments are defined:
 * `cluster_name` - Name of the cluster configured for this connection.
@@ -84,6 +99,9 @@ If `type` is of value `Kafka` the following additional arguments are defined:
 * `security` - Properties for the secure transport connection to Kafka. For SSL, this can include the trusted certificate to use. See [security](#security).
 * `networking` - Networking Access Type can either be `PUBLIC` (default) or `VPC`. See [networking](#networking).
 
+If `type` is of value `AWSLambda` the following additional arguments are defined:
+* `aws` - The configuration for AWS Lambda connection. See [AWS](#AWS)
+
 ### Authentication
 
 * `mechanism` - Style of authentication. Can be one of `PLAIN`, `SCRAM-256`, or `SCRAM-512`.
@@ -107,6 +125,9 @@ If `type` is of value `Kafka` the following additional arguments are defined:
 * `type` - Selected networking type. Either `PUBLIC`, `VPC` or `PRIVATE_LINK`. Defaults to `PUBLIC`.
 * `connection_id` - Id of the Private Link connection when type is `PRIVATE_LINK`.
 
+### AWS
+* `role_arn` - Amazon Resource Name (ARN) that identifies the Amazon Web Services (AWS) Identity and Access Management (IAM) role that MongoDB Cloud assumes when it accesses resources in your AWS account.
+
 ## Import
 
 You can import a stream connection resource using the instance name, project ID, and connection name. The format must be `INSTANCE_NAME-PROJECT_ID-CONNECTION_NAME`. For example:

examples/mongodbatlas_stream_connection/main.tf

@@ -70,6 +70,16 @@ resource "mongodbatlas_stream_connection" "example-sample" {
   type            = "Sample"
 }
 
+resource "mongodbatlas_stream_connection" "example-aws-lambda" {
+  project_id      = var.project_id
+  instance_name   = mongodbatlas_stream_instance.example.instance_name
+  connection_name = "AWSLambdaConnection"
+  type            = "AWSLambda"
+  aws = {
+    role_arn = "arn:aws:iam::123456789123:role/lambdaRole"
+  }
+}
+
 data "mongodbatlas_stream_connection" "example-kafka-ssl" {
   project_id      = var.project_id
   instance_name   = mongodbatlas_stream_instance.example.instance_name

internal/service/streamconnection/model_stream_connection.go

@@ -77,6 +77,16 @@ func NewStreamConnectionReq(ctx context.Context, plan *TFStreamConnectionModel)
 		}
 	}
 
+	if !plan.AWS.IsNull() {
+		awsModel := &TFAWSModel{}
+		if diags := plan.AWS.As(ctx, awsModel, basetypes.ObjectAsOptions{}); diags.HasError() {
+			return nil, diags
+		}
+		streamConnection.Aws = &admin.StreamsAWSConnectionConfig{
+			RoleArn: awsModel.RoleArn.ValueStringPointer(),
+		}
+	}
+
 	return &streamConnection, nil
 }
 
@@ -149,6 +159,17 @@ func NewTFStreamConnection(ctx context.Context, projID, instanceName string, cur
 		connectionModel.Networking = networkingModel
 	}
 
+	connectionModel.AWS = types.ObjectNull(AWSObjectType.AttrTypes)
+	if apiResp.Aws != nil {
+		aws, diags := types.ObjectValueFrom(ctx, AWSObjectType.AttrTypes, TFAWSModel{
+			RoleArn: types.StringPointerValue(apiResp.Aws.RoleArn),
+		})
+		if diags.HasError() {
+			return nil, diags
+		}
+		connectionModel.AWS = aws
+	}
+
 	return &connectionModel, nil
 }
 

internal/service/streamconnection/model_stream_connection_test.go

@@ -26,6 +26,8 @@ const (
 	sampleConnectionName      = "sample_stream_solar"
 	networkingType            = "PUBLIC"
 	privatelinkNetworkingType = "PRIVATE_LINK"
+	awslambdaConnectionName   = "aws_lambda_connection"
+	sampleRoleArn             = "rn:aws:iam::123456789123:role/sample"
 )
 
 var configMap = map[string]string{
@@ -70,6 +72,7 @@ func TestStreamConnectionSDKToTFModel(t *testing.T) {
 				Security:        types.ObjectNull(streamconnection.ConnectionSecurityObjectType.AttrTypes),
 				DBRoleToExecute: tfDBRoleToExecuteObject(t, dbRole, dbRoleType),
 				Networking:      types.ObjectNull(streamconnection.NetworkingObjectType.AttrTypes),
+				AWS:             types.ObjectNull(streamconnection.AWSObjectType.AttrTypes),
 			},
 		},
 		{
@@ -102,6 +105,7 @@ func TestStreamConnectionSDKToTFModel(t *testing.T) {
 				Security:         tfSecurityObject(t, DummyCACert, securityProtocol),
 				DBRoleToExecute:  types.ObjectNull(streamconnection.DBRoleToExecuteObjectType.AttrTypes),
 				Networking:       types.ObjectNull(streamconnection.NetworkingObjectType.AttrTypes),
+				AWS:              types.ObjectNull(streamconnection.AWSObjectType.AttrTypes),
 			},
 		},
 		{
@@ -123,6 +127,7 @@ func TestStreamConnectionSDKToTFModel(t *testing.T) {
 				Security:        types.ObjectNull(streamconnection.ConnectionSecurityObjectType.AttrTypes),
 				DBRoleToExecute: types.ObjectNull(streamconnection.DBRoleToExecuteObjectType.AttrTypes),
 				Networking:      types.ObjectNull(streamconnection.NetworkingObjectType.AttrTypes),
+				AWS:             types.ObjectNull(streamconnection.AWSObjectType.AttrTypes),
 			},
 		},
 		{
@@ -155,6 +160,7 @@ func TestStreamConnectionSDKToTFModel(t *testing.T) {
 				Security:         tfSecurityObject(t, DummyCACert, securityProtocol),
 				DBRoleToExecute:  types.ObjectNull(streamconnection.DBRoleToExecuteObjectType.AttrTypes),
 				Networking:       types.ObjectNull(streamconnection.NetworkingObjectType.AttrTypes),
+				AWS:              types.ObjectNull(streamconnection.AWSObjectType.AttrTypes),
 			},
 		},
 		{
@@ -175,6 +181,29 @@ func TestStreamConnectionSDKToTFModel(t *testing.T) {
 				Security:        types.ObjectNull(streamconnection.ConnectionSecurityObjectType.AttrTypes),
 				DBRoleToExecute: types.ObjectNull(streamconnection.DBRoleToExecuteObjectType.AttrTypes),
 				Networking:      types.ObjectNull(streamconnection.NetworkingObjectType.AttrTypes),
+				AWS:             types.ObjectNull(streamconnection.AWSObjectType.AttrTypes),
+			},
+		},
+		{
+			name: "AWSLambda connection type with roleArn",
+			SDKResp: &admin.StreamsConnection{
+				Name: admin.PtrString(awslambdaConnectionName),
+				Type: admin.PtrString("AWSLambda"),
+				Aws:  &admin.StreamsAWSConnectionConfig{RoleArn: admin.PtrString(sampleRoleArn)},
+			},
+			providedProjID:       dummyProjectID,
+			providedInstanceName: instanceName,
+			expectedTFModel: &streamconnection.TFStreamConnectionModel{
+				ProjectID:       types.StringValue(dummyProjectID),
+				InstanceName:    types.StringValue(instanceName),
+				ConnectionName:  types.StringValue(awslambdaConnectionName),
+				Type:            types.StringValue("AWSLambda"),
+				Authentication:  types.ObjectNull(streamconnection.ConnectionAuthenticationObjectType.AttrTypes),
+				Config:          types.MapNull(types.StringType),
+				Security:        types.ObjectNull(streamconnection.ConnectionSecurityObjectType.AttrTypes),
+				DBRoleToExecute: types.ObjectNull(streamconnection.DBRoleToExecuteObjectType.AttrTypes),
+				Networking:      types.ObjectNull(streamconnection.NetworkingObjectType.AttrTypes),
+				AWS:             tfAWSLambdaConfigObject(t, sampleRoleArn),
 			},
 		},
 	}
@@ -238,8 +267,15 @@ func TestStreamConnectionsSDKToTFModel(t *testing.T) {
 						Name: admin.PtrString(sampleConnectionName),
 						Type: admin.PtrString("Sample"),
 					},
+					{
+						Name: admin.PtrString(awslambdaConnectionName),
+						Type: admin.PtrString("AWSLambda"),
+						Aws: &admin.StreamsAWSConnectionConfig{
+							RoleArn: admin.PtrString(sampleRoleArn),
+						},
+					},
 				},
-				TotalCount: admin.PtrInt(3),
+				TotalCount: admin.PtrInt(4),
 			},
 			providedConfig: &streamconnection.TFStreamConnectionsDSModel{
 				ProjectID:    types.StringValue(dummyProjectID),
@@ -252,7 +288,7 @@ func TestStreamConnectionsSDKToTFModel(t *testing.T) {
 				InstanceName: types.StringValue(instanceName),
 				PageNum:      types.Int64Value(1),
 				ItemsPerPage: types.Int64Value(3),
-				TotalCount:   types.Int64Value(3),
+				TotalCount:   types.Int64Value(4),
 				Results: []streamconnection.TFStreamConnectionModel{
 					{
 						ID:               types.StringValue(fmt.Sprintf("%s-%s-%s", instanceName, dummyProjectID, connectionName)),
@@ -266,6 +302,7 @@ func TestStreamConnectionsSDKToTFModel(t *testing.T) {
 						Security:         tfSecurityObject(t, DummyCACert, securityProtocol),
 						DBRoleToExecute:  types.ObjectNull(streamconnection.DBRoleToExecuteObjectType.AttrTypes),
 						Networking:       tfNetworkingObject(t, networkingType, nil),
+						AWS:              types.ObjectNull(streamconnection.AWSObjectType.AttrTypes),
 					},
 					{
 						ID:              types.StringValue(fmt.Sprintf("%s-%s-%s", instanceName, dummyProjectID, connectionName)),
@@ -279,6 +316,7 @@ func TestStreamConnectionsSDKToTFModel(t *testing.T) {
 						Security:        types.ObjectNull(streamconnection.ConnectionSecurityObjectType.AttrTypes),
 						DBRoleToExecute: tfDBRoleToExecuteObject(t, dbRole, dbRoleType),
 						Networking:      types.ObjectNull(streamconnection.NetworkingObjectType.AttrTypes),
+						AWS:             types.ObjectNull(streamconnection.AWSObjectType.AttrTypes),
 					},
 					{
 						ID:              types.StringValue(fmt.Sprintf("%s-%s-%s", instanceName, dummyProjectID, sampleConnectionName)),
@@ -292,6 +330,21 @@ func TestStreamConnectionsSDKToTFModel(t *testing.T) {
 						Security:        types.ObjectNull(streamconnection.ConnectionSecurityObjectType.AttrTypes),
 						DBRoleToExecute: types.ObjectNull(streamconnection.DBRoleToExecuteObjectType.AttrTypes),
 						Networking:      types.ObjectNull(streamconnection.NetworkingObjectType.AttrTypes),
+						AWS:             types.ObjectNull(streamconnection.AWSObjectType.AttrTypes),
+					},
+					{
+						ID:              types.StringValue(fmt.Sprintf("%s-%s-%s", instanceName, dummyProjectID, awslambdaConnectionName)),
+						ProjectID:       types.StringValue(dummyProjectID),
+						InstanceName:    types.StringValue(instanceName),
+						ConnectionName:  types.StringValue(awslambdaConnectionName),
+						Type:            types.StringValue("AWSLambda"),
+						ClusterName:     types.StringNull(),
+						Authentication:  types.ObjectNull(streamconnection.ConnectionAuthenticationObjectType.AttrTypes),
+						Config:          types.MapNull(types.StringType),
+						Security:        types.ObjectNull(streamconnection.ConnectionSecurityObjectType.AttrTypes),
+						DBRoleToExecute: types.ObjectNull(streamconnection.DBRoleToExecuteObjectType.AttrTypes),
+						Networking:      types.ObjectNull(streamconnection.NetworkingObjectType.AttrTypes),
+						AWS:             tfAWSLambdaConfigObject(t, sampleRoleArn),
 					},
 				},
 			},
@@ -413,6 +466,23 @@ func TestStreamInstanceTFToSDKCreateModel(t *testing.T) {
 				Type: admin.PtrString("Sample"),
 			},
 		},
+		{
+			name: "AWSLambda type TF state",
+			tfModel: &streamconnection.TFStreamConnectionModel{
+				ProjectID:      types.StringValue(dummyProjectID),
+				InstanceName:   types.StringValue(instanceName),
+				ConnectionName: types.StringValue(awslambdaConnectionName),
+				Type:           types.StringValue("AWSLambda"),
+				AWS:            tfAWSLambdaConfigObject(t, sampleRoleArn),
+			},
+			expectedSDKReq: &admin.StreamsConnection{
+				Name: admin.PtrString(awslambdaConnectionName),
+				Type: admin.PtrString("AWSLambda"),
+				Aws: &admin.StreamsAWSConnectionConfig{
+					RoleArn: admin.PtrString(sampleRoleArn),
+				},
+			},
+		},
 	}
 
 	for _, tc := range testCases {
@@ -503,3 +573,14 @@ func tfNetworkingObject(t *testing.T, networkingType string, connectionID *strin
 	}
 	return networking
 }
+
+func tfAWSLambdaConfigObject(t *testing.T, roleArn string) types.Object {
+	t.Helper()
+	aws, diags := types.ObjectValueFrom(context.Background(), streamconnection.AWSObjectType.AttrTypes, streamconnection.TFAWSModel{
+		RoleArn: types.StringValue(roleArn),
+	})
+	if diags.HasError() {
+		t.Errorf("failed to create terraform data model: %s", diags.Errors()[0].Summary())
+	}
+	return aws
+}