feat: adds tier attribute to stream processor resource and datasource

kpatel71716 · kpatel71716 · commit e29e4f411a14 · 2025-12-09T05:26:00.000-06:00
diff --git a/.changelog/todo.txt b/.changelog/todo.txt
@@ -0,0 +1,7 @@
+```release-note:enhancement
+resource/mongodbatlas_stream_processor: Adds the `tier` attribute
+```
+
+```release-note:enhancement
+data-source/mongodbatlas_stream_processor: Adds the `tier` attribute
+```
diff --git a/docs/data-sources/stream_processor.md b/docs/data-sources/stream_processor.md
@@ -140,6 +140,7 @@ output "stream_processors_results" {
 
 **NOTE** When a Stream Processor is updated without specifying the state, it is stopped and then restored to previous state upon update completion.
 - `stats` (String) The stats associated with the stream processor. Refer to the [MongoDB Atlas Docs](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-stream-processor/#view-statistics-of-a-stream-processor) for more information.
+- `tier` (String) Selected tier for the Stream Processor. Configures Memory / VCPU allowances.
 
 <a id="nestedatt--options"></a>
 ### Nested Schema for `options`
diff --git a/docs/resources/stream_processor.md b/docs/resources/stream_processor.md
@@ -70,6 +70,7 @@ resource "mongodbatlas_stream_processor" "stream-processor-sample-example" {
     { "$emit" = { "connectionName" : resource.mongodbatlas_stream_connection.example-cluster.connection_name, "db" : "sample", "coll" : "solar", "timeseries" : { "timeField" : "_ts" } } }
   ])
   state = "STARTED"
+  tier  = "SP30"
 }
 
 resource "mongodbatlas_stream_processor" "stream-processor-cluster-to-kafka-example" {
@@ -141,9 +142,10 @@ output "stream_processors_results" {
 - `delete_on_create_timeout` (Boolean) Indicates whether to delete the resource being created if a timeout is reached when waiting for completion. When set to `true` and timeout occurs, it triggers the deletion and returns immediately without waiting for deletion to complete. When set to `false`, the timeout will not trigger resource deletion. If you suspect a transient error when the value is `true`, wait before retrying to allow resource deletion to finish. Default is `true`.
 - `instance_name` (String, Deprecated) Label that identifies the stream processing workspace.
 - `options` (Attributes) Optional configuration for the stream processor. (see [below for nested schema](#nestedatt--options))
-- `state` (String) The state of the stream processor. Commonly occurring states are 'CREATED', 'STARTED', 'STOPPED' and 'FAILED'. Used to start or stop the Stream Processor. Valid values are `CREATED`, `STARTED` or `STOPPED`. When a Stream Processor is created without specifying the state, it will default to `CREATED` state. When a Stream Processor is updated without specifying the state, it will default to the Previous state. 
+- `state` (String) The state of the stream processor. Commonly occurring states are 'CREATED', 'STARTED', 'STOPPED' and 'FAILED'. Used to start or stop the Stream Processor. Valid values are `CREATED`, `STARTED` or `STOPPED`. When a Stream Processor is created without specifying the state, it will default to `CREATED` state. When a Stream Processor is updated without specifying the state, it will default to the Previous state.
 
 **NOTE** When a Stream Processor is updated without specifying the state, it is stopped and then restored to previous state upon update completion.
+- `tier` (String) Selected tier for the Stream Processor. Configures Memory / VCPU allowances.
 - `timeouts` (Attributes) (see [below for nested schema](#nestedatt--timeouts))
 - `workspace_name` (String) Label that identifies the stream processing workspace.
 
diff --git a/internal/service/streamprocessor/model.go b/internal/service/streamprocessor/model.go
@@ -114,6 +114,7 @@ func NewStreamProcessorWithStats(ctx context.Context, projectID, instanceName, w
 		ProjectID:     types.StringPointerValue(&projectID),
 		State:         types.StringPointerValue(&apiResp.State),
 		Stats:         statsTF,
+		Tier:          types.StringPointerValue(apiResp.Tier),
 	}
 
 	if workspaceName != "" {
@@ -158,6 +159,7 @@ func NewTFStreamprocessorDSModel(ctx context.Context, projectID, instanceName, w
 		ProjectID:     types.StringPointerValue(&projectID),
 		State:         types.StringPointerValue(&apiResp.State),
 		Stats:         statsTF,
+		Tier:          types.StringPointerValue(apiResp.Tier),
 	}
 
 	if workspaceName != "" {
diff --git a/internal/service/streamprocessor/resource.go b/internal/service/streamprocessor/resource.go
@@ -107,13 +107,12 @@ func (r *streamProcessorRS) Create(ctx context.Context, req resource.CreateReque
 	}
 
 	if needsStarting {
-		_, err := connV2.StreamsApi.StartStreamProcessorWithParams(ctx,
-			&admin.StartStreamProcessorApiParams{
-				GroupId:       projectID,
-				TenantName:    workspaceOrInstanceName,
-				ProcessorName: processorName,
-			},
-		).Execute()
+		startWithOptions := &admin.StreamsStartStreamProcessorWith{}
+		if plan.Tier.ValueString() != "" {
+			startWithOptions.SetTier(plan.Tier.ValueString())
+		}
+
+		_, err := connV2.StreamsApi.StartStreamProcessorWith(ctx, projectID, workspaceOrInstanceName, processorName, startWithOptions).Execute()
 		if err != nil {
 			resp.Diagnostics.AddError(errorCreateStart, err.Error())
 			return
@@ -237,13 +236,12 @@ func (r *streamProcessorRS) Update(ctx context.Context, req resource.UpdateReque
 
 	// start the stream processor if the desired state is started
 	if plannedState == StartedState {
-		_, err := r.Client.AtlasV2.StreamsApi.StartStreamProcessorWithParams(ctx,
-			&admin.StartStreamProcessorApiParams{
-				GroupId:       projectID,
-				TenantName:    workspaceOrInstanceName,
-				ProcessorName: processorName,
-			},
-		).Execute()
+		startWithOptions := &admin.StreamsStartStreamProcessorWith{}
+		if plan.Tier.ValueString() != "" {
+			startWithOptions.SetTier(plan.Tier.ValueString())
+		}
+
+		_, err := r.Client.AtlasV2.StreamsApi.StartStreamProcessorWith(ctx, projectID, workspaceOrInstanceName, processorName, startWithOptions).Execute()
 		if err != nil {
 			resp.Diagnostics.AddError("Error starting stream processor", err.Error())
 			return
diff --git a/internal/service/streamprocessor/resource_schema.go b/internal/service/streamprocessor/resource_schema.go
@@ -95,6 +95,14 @@ func ResourceSchema(ctx context.Context) schema.Schema {
 				Computed:            true,
 				MarkdownDescription: "The stats associated with the stream processor. Refer to the [MongoDB Atlas Docs](https://www.mongodb.com/docs/atlas/atlas-stream-processing/manage-stream-processor/#view-statistics-of-a-stream-processor) for more information.",
 			},
+			"tier": schema.StringAttribute{
+				Optional:            true,
+				Computed:            true,
+				MarkdownDescription: "Selected tier for the Stream Processor. Configures Memory / VCPU allowances.",
+				PlanModifiers: []planmodifier.String{
+					stringplanmodifier.UseStateForUnknown(),
+				},
+			},
 			"timeouts": timeouts.Attributes(ctx, timeouts.Opts{
 				Create: true,
 			}),
@@ -120,6 +128,7 @@ type TFStreamProcessorRSModel struct {
 	ProjectID             types.String         `tfsdk:"project_id"`
 	State                 types.String         `tfsdk:"state"`
 	Stats                 types.String         `tfsdk:"stats"`
+	Tier                  types.String         `tfsdk:"tier"`
 	Timeouts              timeouts.Value       `tfsdk:"timeouts"`
 	DeleteOnCreateTimeout types.Bool           `tfsdk:"delete_on_create_timeout"`
 }
@@ -155,6 +164,7 @@ type TFStreamProcessorDSModel struct {
 	ProjectID     types.String `tfsdk:"project_id"`
 	State         types.String `tfsdk:"state"`
 	Stats         types.String `tfsdk:"stats"`
+	Tier          types.String `tfsdk:"tier"`
 }
 
 type TFStreamProcessorsDSModel struct {
diff --git a/internal/service/streamprocessor/resource_test.go b/internal/service/streamprocessor/resource_test.go
@@ -53,6 +53,33 @@ func TestAccStreamProcessor_basic(t *testing.T) {
 	resource.ParallelTest(t, *basicTestCase(t))
 }
 
+func TestAccStreamProcessor_withTier(t *testing.T) {
+	var (
+		projectID, workspaceName = acc.ProjectIDExecutionWithStreamInstance(t)
+		randomSuffix             = acctest.RandString(5)
+		processorName            = "new-processor-tier" + randomSuffix
+		tier                     = "SP30"
+	)
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:                 func() { acc.PreCheckBasic(t) },
+		ProtoV6ProviderFactories: acc.TestAccProviderV6Factories,
+		CheckDestroy:             checkDestroyStreamProcessor,
+		Steps: []resource.TestStep{
+			{
+				Config: configWithTier(t, projectID, workspaceName, processorName, tier),
+				Check: resource.ComposeAggregateTestCheckFunc(
+					resource.TestCheckResourceAttr(resourceName, "project_id", projectID),
+					resource.TestCheckResourceAttr(resourceName, "workspace_name", workspaceName),
+					resource.TestCheckResourceAttr(resourceName, "processor_name", processorName),
+					resource.TestCheckResourceAttr(resourceName, "state", streamprocessor.StartedState),
+					resource.TestCheckResourceAttr(resourceName, "tier", tier),
+				),
+			},
+			importStep(),
+		}})
+}
+
 func basicTestCase(t *testing.T) *resource.TestCase {
 	t.Helper()
 	var (
@@ -673,6 +700,44 @@ func config(t *testing.T, projectID, workspaceName, processorName, state, nameSu
 	`, projectID, workspaceName, processorName, pipeline, stateConfig, optionsStr, dependsOnStr, timeoutConfig, deleteOnCreateTimeoutConfig) + otherConfig
 }
 
+func configWithTier(t *testing.T, projectID, workspaceName, processorName, tier string) string {
+	t.Helper()
+
+	return fmt.Sprintf(`
+	resource "mongodbatlas_stream_connection" "sample_stream_solar" {
+		project_id      = %[1]q
+		workspace_name  = %[2]q
+		connection_name = "sample_stream_solar"
+		type            = "Sample"
+	}
+
+	resource "mongodbatlas_stream_processor" "processor" {
+		project_id     = %[1]q
+		workspace_name = %[2]q
+		processor_name = %[3]q
+		pipeline = jsonencode([
+			{ "$source" = { "connectionName" = mongodbatlas_stream_connection.sample_stream_solar.connection_name } },
+			{ "$emit" = { "connectionName" = "__testLog" } }
+		])
+		state = "STARTED"
+		tier  = %[4]q
+	}
+
+	data "mongodbatlas_stream_processor" "test" {
+		project_id     = %[1]q
+		workspace_name = %[2]q
+		processor_name = %[3]q
+		depends_on     = [mongodbatlas_stream_processor.processor]
+	}
+
+	data "mongodbatlas_stream_processors" "test" {
+		project_id     = %[1]q
+		workspace_name = %[2]q
+		depends_on     = [mongodbatlas_stream_processor.processor]
+	}
+	`, projectID, workspaceName, processorName, tier)
+}
+
 func configMigration(t *testing.T, projectID, instanceName, processorName, state, nameSuffix string, src, dest connectionConfig, timeoutConfig string, deleteOnCreateTimeout *bool) string {
 	t.Helper()
 	stateConfig := ""

Original file line number	Diff line number	Diff line change
`@@ -114,6 +114,7 @@ func NewStreamProcessorWithStats(ctx context.Context, projectID, instanceName, w`
`114`	`114`	`ProjectID: types.StringPointerValue(&projectID),`
`115`	`115`	`State: types.StringPointerValue(&apiResp.State),`
`116`	`116`	`Stats: statsTF,`
	`117`	`+ Tier: types.StringPointerValue(apiResp.Tier),`
`117`	`118`	`}`
`118`	`119`
`119`	`120`	`if workspaceName != "" {`
`@@ -158,6 +159,7 @@ func NewTFStreamprocessorDSModel(ctx context.Context, projectID, instanceName, w`
`158`	`159`	`ProjectID: types.StringPointerValue(&projectID),`
`159`	`160`	`State: types.StringPointerValue(&apiResp.State),`
`160`	`161`	`Stats: statsTF,`
	`162`	`+ Tier: types.StringPointerValue(apiResp.Tier),`
`161`	`163`	`}`
`162`	`164`
`163`	`165`	`if workspaceName != "" {`