feat: Supports new Https connection type in streams processing (#3150)

Author: jwongmongodb

.changelog/3150.txt

@@ -0,0 +1,11 @@
+```release-note:enhancement
+resource/mongodbatlas_stream_connection: Adds `Https` connection
+```
+
+```release-note:enhancement
+data-source/mongodbatlas_stream_connection: Adds `Https` connection
+```
+
+```release-note:enhancement
+data-source/mongodbatlas_stream_connections: Adds `Https` connection
+```

docs/data-sources/stream_connection.md

@@ -20,7 +20,7 @@ data "mongodbatlas_stream_connection" "example" {
 
 ## Attributes Reference
 
-* `type` - Type of connection. Can be `Cluster`, `Kafka`, `Sample`, or `AWSLambda`.
+* `type` - Type of connection. Can be `AWSLambda`, `Cluster`, `Https`, `Kafka` or `Sample`.
 
 If `type` is of value `Cluster` the following additional attributes are defined:
 * `cluster_name` - Name of the cluster configured for this connection.
@@ -36,6 +36,10 @@ If `type` is of value `Kafka` the following additional attributes are defined:
 If `type` is of value `AWSLambda` the following additional attributes are defined:
 * `aws` - The configuration for AWS Lambda connection. See [AWS](#AWS)
 
+If `type` is of value `Https` the following additional attributes are defined:
+* `url` - URL of the HTTPs endpoint that will be used for creating a connection.
+* `headers` - A map of key-value pairs for optional headers.
+
 ### Authentication
 
 * `mechanism` - Style of authentication. Can be one of `PLAIN`, `SCRAM-256`, or `SCRAM-512`.

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. `Cluster`, `Kafka`, `Sample`, or `AWSLambda`.
+* `type` - Type of connection. `AWSLambda`, `Cluster`, `Https`, `Kafka` or `Sample`.
 
 If `type` is of value `Cluster` the following additional attributes are defined:
 * `cluster_name` - Name of the cluster configured for this connection.
@@ -48,6 +48,10 @@ If `type` is of value `Kafka` the following additional attributes are defined:
 If `type` is of value `AWSLambda` the following additional attributes are defined::
 * `aws` - The configuration for AWS Lambda connection. See [AWS](#AWS)
 
+If `type` is of value `Https` the following additional attributes are defined:
+* `url` - URL of the HTTPs endpoint that will be used for creating a connection.
+* `headers` - A map of key-value pairs for optional headers.
+
 ### Authentication
 
 * `mechanism` - Style of authentication. Can be one of `PLAIN`, `SCRAM-256`, or `SCRAM-512`.

docs/resources/stream_connection.md

@@ -81,12 +81,28 @@ resource "mongodbatlas_stream_connection" "test" {
 
 ```
 
+### Example Https Connection
+
+```terraform
+resource "mongodbatlas_stream_connection" "example-https" {
+  project_id      = var.project_id
+  instance_name   = mongodbatlas_stream_instance.example.instance_name
+  connection_name = "https_connection_tf_new"
+  type            = "Https"
+  url             = "https://example.com"
+  headers = {
+    key1 = "value1"
+    key2 = "value2"
+  }
+}
+```
+
 ## 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 `Cluster`, `Kafka`, `Sample`, or `AWSLambda`.
+* `type` - (Required) Type of connection. Can be `AWSLambda`, `Cluster`, `Https`, `Kafka` or `Sample`.
 
 If `type` is of value `Cluster` the following additional arguments are defined:
 * `cluster_name` - Name of the cluster configured for this connection.
@@ -102,6 +118,10 @@ If `type` is of value `Kafka` the following additional arguments are defined:
 If `type` is of value `AWSLambda` the following additional arguments are defined:
 * `aws` - The configuration for AWS Lambda connection. See [AWS](#AWS)
 
+If `type` is of value `Https` the following additional attributes are defined:
+* `url` - URL of the HTTPs endpoint that will be used for creating a connection.
+* `headers` - A map of key-value pairs for optional headers.
+
 ### Authentication
 
 * `mechanism` - Style of authentication. Can be one of `PLAIN`, `SCRAM-256`, or `SCRAM-512`.

examples/mongodbatlas_stream_connection/main.tf

@@ -80,6 +80,18 @@ resource "mongodbatlas_stream_connection" "example-aws-lambda" {
   }
 }
 
+resource "mongodbatlas_stream_connection" "example-https" {
+  project_id      = var.project_id
+  instance_name   = mongodbatlas_stream_instance.example.instance_name
+  connection_name = "HttpsConnection"
+  type            = "Https"
+  url             = "https://example.com"
+  headers = {
+    key1 = "value1"
+    key2 = "value2"
+  }
+}
+
 data "mongodbatlas_stream_connection" "example-kafka-ssl" {
   project_id      = var.project_id
   instance_name   = mongodbatlas_stream_instance.example.instance_name

internal/service/streamconnection/data_source_stream_connection_test.go

@@ -20,8 +20,8 @@ func TestAccStreamDSStreamConnection_kafkaPlaintext(t *testing.T) {
 		CheckDestroy:             CheckDestroyStreamConnection,
 		Steps: []resource.TestStep{
 			{
-				Config: streamConnectionDataSourceConfig(kafkaStreamConnectionConfig(projectID, instanceName, "user", "rawpassword", "localhost:9092,localhost:9092", "earliest", kafkaNetworkingPublic, false)),
-				Check:  kafkaStreamConnectionAttributeChecks(dataSourceName, instanceName, "user", "rawpassword", "localhost:9092,localhost:9092", "earliest", networkingTypePublic, false, false),
+				Config: streamConnectionDataSourceConfig(configureKafka(projectID, instanceName, "user", "rawpassword", "localhost:9092,localhost:9092", "earliest", kafkaNetworkingPublic, false)),
+				Check:  checkKafkaAttributes(dataSourceName, instanceName, "user", "rawpassword", "localhost:9092,localhost:9092", "earliest", networkingTypePublic, false, false),
 			},
 		},
 	})
@@ -39,8 +39,8 @@ func TestAccStreamDSStreamConnection_kafkaSSL(t *testing.T) {
 		CheckDestroy:             CheckDestroyStreamConnection,
 		Steps: []resource.TestStep{
 			{
-				Config: streamConnectionDataSourceConfig(kafkaStreamConnectionConfig(projectID, instanceName, "user", "rawpassword", "localhost:9092", "earliest", kafkaNetworkingPublic, true)),
-				Check:  kafkaStreamConnectionAttributeChecks(dataSourceName, instanceName, "user", "rawpassword", "localhost:9092", "earliest", networkingTypePublic, true, false),
+				Config: streamConnectionDataSourceConfig(configureKafka(projectID, instanceName, "user", "rawpassword", "localhost:9092", "earliest", kafkaNetworkingPublic, true)),
+				Check:  checkKafkaAttributes(dataSourceName, instanceName, "user", "rawpassword", "localhost:9092", "earliest", networkingTypePublic, true, false),
 			},
 		},
 	})
@@ -58,8 +58,8 @@ func TestAccStreamDSStreamConnection_cluster(t *testing.T) {
 		CheckDestroy:             CheckDestroyStreamConnection,
 		Steps: []resource.TestStep{
 			{
-				Config: streamConnectionDataSourceConfig(clusterStreamConnectionConfig(projectID, instanceName, clusterName)),
-				Check:  clusterStreamConnectionAttributeChecks(dataSourceName, clusterName),
+				Config: streamConnectionDataSourceConfig(configureCluster(projectID, instanceName, clusterName)),
+				Check:  checkClusterAttributes(dataSourceName, clusterName),
 			},
 		},
 	})
@@ -78,8 +78,8 @@ func TestAccStreamDSStreamConnection_sample(t *testing.T) {
 		CheckDestroy:             CheckDestroyStreamConnection,
 		Steps: []resource.TestStep{
 			{
-				Config: streamConnectionDataSourceConfig(sampleStreamConnectionConfig(projectID, instanceName, sampleName)),
-				Check:  sampleStreamConnectionAttributeChecks(dataSourceName, instanceName, sampleName),
+				Config: streamConnectionDataSourceConfig(configureSampleStream(projectID, instanceName, sampleName)),
+				Check:  checkSampleStreamAttributes(dataSourceName, instanceName, sampleName),
 			},
 		},
 	})

internal/service/streamconnection/data_source_stream_connections_test.go

@@ -21,7 +21,7 @@ func TestAccStreamDSStreamConnections_basic(t *testing.T) {
 		CheckDestroy:             CheckDestroyStreamConnection,
 		Steps: []resource.TestStep{
 			{
-				Config: streamConnectionsDataSourceConfig(kafkaStreamConnectionConfig(projectID, instanceName, "user", "rawpassword", "localhost:9092,localhost:9092", "earliest", kafkaNetworkingPublic, false)),
+				Config: streamConnectionsDataSourceConfig(configureKafka(projectID, instanceName, "user", "rawpassword", "localhost:9092,localhost:9092", "earliest", kafkaNetworkingPublic, false)),
 				Check:  streamConnectionsAttributeChecks(dataSourceName, nil, nil, 1),
 			},
 		},
@@ -40,7 +40,7 @@ func TestAccStreamDSStreamConnections_withPageConfig(t *testing.T) {
 		CheckDestroy:             CheckDestroyStreamConnection,
 		Steps: []resource.TestStep{
 			{
-				Config: streamConnectionsWithPageAttrDataSourceConfig(kafkaStreamConnectionConfig(projectID, instanceName, "user", "rawpassword", "localhost:9092,localhost:9092", "earliest", kafkaNetworkingPublic, false)),
+				Config: streamConnectionsWithPageAttrDataSourceConfig(configureKafka(projectID, instanceName, "user", "rawpassword", "localhost:9092,localhost:9092", "earliest", kafkaNetworkingPublic, false)),
 				Check:  streamConnectionsAttributeChecks(dataSourceName, admin.PtrInt(2), admin.PtrInt(1), 0),
 			},
 		},

internal/service/streamconnection/model_stream_connection.go

@@ -18,6 +18,7 @@ func NewStreamConnectionReq(ctx context.Context, plan *TFStreamConnectionModel)
 		Type:             plan.Type.ValueStringPointer(),
 		ClusterName:      plan.ClusterName.ValueStringPointer(),
 		BootstrapServers: plan.BootstrapServers.ValueStringPointer(),
+		Url:              plan.URL.ValueStringPointer(),
 	}
 	if !plan.Authentication.IsNull() {
 		authenticationModel := &TFConnectionAuthenticationModel{}
@@ -87,9 +88,34 @@ func NewStreamConnectionReq(ctx context.Context, plan *TFStreamConnectionModel)
 		}
 	}
 
+	if !plan.Headers.IsNull() {
+		headersMap := make(map[string]string)
+		if diags := plan.Headers.ElementsAs(ctx, &headersMap, true); diags.HasError() {
+			return nil, diags
+		}
+		streamConnection.Headers = &headersMap
+	}
+
 	return &streamConnection, nil
 }
 
+func NewStreamConnectionUpdateReq(ctx context.Context, plan *TFStreamConnectionModel) (*admin.StreamsConnection, diag.Diagnostics) {
+	streamConnection, diags := NewStreamConnectionReq(ctx, plan)
+	if diags.HasError() {
+		return nil, diags
+	}
+
+	headersMap := make(map[string]string)
+	// only set headers if the plan is not empty, otherwise the headers will be removed by sending an empty headers map to the PATCH endpoint
+	if !plan.Headers.IsNull() && !plan.Headers.IsUnknown() {
+		if diags := plan.Headers.ElementsAs(ctx, &headersMap, true); diags.HasError() {
+			return nil, diags
+		}
+	}
+	streamConnection.Headers = &headersMap
+	return streamConnection, nil
+}
+
 func NewTFStreamConnection(ctx context.Context, projID, instanceName string, currAuthConfig *types.Object, apiResp *admin.StreamsConnection) (*TFStreamConnectionModel, diag.Diagnostics) {
 	rID := fmt.Sprintf("%s-%s-%s", instanceName, projID, conversion.SafeString(apiResp.Name))
 	connectionModel := TFStreamConnectionModel{
@@ -100,6 +126,7 @@ func NewTFStreamConnection(ctx context.Context, projID, instanceName string, cur
 		Type:             types.StringPointerValue(apiResp.Type),
 		ClusterName:      types.StringPointerValue(apiResp.ClusterName),
 		BootstrapServers: types.StringPointerValue(apiResp.BootstrapServers),
+		URL:              types.StringPointerValue(apiResp.Url),
 	}
 
 	authModel, diags := newTFConnectionAuthenticationModel(ctx, currAuthConfig, apiResp.Authentication)
@@ -170,6 +197,16 @@ func NewTFStreamConnection(ctx context.Context, projID, instanceName string, cur
 		connectionModel.AWS = aws
 	}
 
+	connectionModel.Headers = types.MapNull(types.StringType)
+	// this is to handle the case where empty headers are returned as an empty map from the API, which is equivalent to a null value
+	if apiResp.Headers != nil && len(*apiResp.Headers) > 0 {
+		mapValue, diags := types.MapValueFrom(ctx, types.StringType, apiResp.Headers)
+		if diags.HasError() {
+			return nil, diags
+		}
+		connectionModel.Headers = mapValue
+	}
+
 	return &connectionModel, nil
 }