Add vertical scaling options to scaling api for clickpipes terraform (#322)

tpanetti · web-flow · commit 16e0e2969ad8 · 2025-07-29T17:49:56.000-07:00
diff --git a/examples/clickpipe/multiple_pipes_example/main.tf b/examples/clickpipe/multiple_pipes_example/main.tf
@@ -22,12 +22,14 @@ variable "number_of_pipes" {
 resource "clickhouse_clickpipe" "multiple" {
   for_each = toset([for i in range(1, var.number_of_pipes + 1) : tostring(i)])
 
-  name        = "📈 multiple pipe ${each.key}"
+  name = "📈 multiple pipe ${each.key}"
 
   service_id = var.service_id
 
   scaling = {
-    replicas = tonumber(each.key)
+    replicas               = tonumber(each.key)
+    replica_cpu_millicores = 250
+    replica_memory_gb      = 1.0
   }
 
   state = "Running"
@@ -47,7 +49,7 @@ resource "clickhouse_clickpipe" "multiple" {
   }
 
   destination = {
-    table    = "multiple_pipes_example_${each.key}"
+    table         = "multiple_pipes_example_${each.key}"
     managed_table = true
 
     table_definition = {
@@ -58,116 +60,116 @@ resource "clickhouse_clickpipe" "multiple" {
 
     columns = [
       {
-        name: "area",
-        type: "Int64"
+        name : "area",
+        type : "Int64"
       },
       {
-        name: "averageSignal",
-        type: "Int64"
+        name : "averageSignal",
+        type : "Int64"
       },
       {
-        name: "cell",
-        type: "Int64"
+        name : "cell",
+        type : "Int64"
       },
       {
-        name: "changeable",
-        type: "Int64"
+        name : "changeable",
+        type : "Int64"
       },
       {
-        name: "created",
-        type: "DateTime64(9)"
+        name : "created",
+        type : "DateTime64(9)"
       },
       {
-        name: "lat",
-        type: "Float64"
+        name : "lat",
+        type : "Float64"
       },
       {
-        name: "lon",
-        type: "Float64"
+        name : "lon",
+        type : "Float64"
       },
       {
-        name: "mcc",
-        type: "Int64"
+        name : "mcc",
+        type : "Int64"
       },
       {
-        name: "net",
-        type: "Int64"
+        name : "net",
+        type : "Int64"
       },
       {
-        name: "radio",
-        type: "String"
+        name : "radio",
+        type : "String"
       },
       {
-        name: "range",
-        type: "Int64"
+        name : "range",
+        type : "Int64"
       },
       {
-        name: "samples-renamed",
-        type: "Int64"
+        name : "samples-renamed",
+        type : "Int64"
       },
       {
-        name: "unit",
-        type: "Int64"
+        name : "unit",
+        type : "Int64"
       },
       {
-        name: "updated",
-        type: "DateTime64(9)"
+        name : "updated",
+        type : "DateTime64(9)"
       }
     ]
   }
 
   field_mappings = [
     {
-      source_field: "averageSignal",
-      destination_field: "averageSignal"
+      source_field : "averageSignal",
+      destination_field : "averageSignal"
     },
     {
-      source_field: "cell",
-      destination_field: "cell"
+      source_field : "cell",
+      destination_field : "cell"
     },
     {
-      source_field: "changeable",
-      destination_field: "changeable"
+      source_field : "changeable",
+      destination_field : "changeable"
     },
     {
-      source_field: "created",
-      destination_field: "created"
+      source_field : "created",
+      destination_field : "created"
     },
     {
-      source_field: "lat",
-      destination_field: "lat"
+      source_field : "lat",
+      destination_field : "lat"
     },
     {
-      source_field: "lon",
-      destination_field: "lon"
+      source_field : "lon",
+      destination_field : "lon"
     },
     {
-      source_field: "mcc",
-      destination_field: "mcc"
+      source_field : "mcc",
+      destination_field : "mcc"
     },
     {
-      source_field: "net",
-      destination_field: "net"
+      source_field : "net",
+      destination_field : "net"
     },
     {
-      source_field: "radio",
-      destination_field: "radio"
+      source_field : "radio",
+      destination_field : "radio"
     },
     {
-      source_field: "range",
-      destination_field: "range"
+      source_field : "range",
+      destination_field : "range"
     },
     {
-      source_field: "samples",
-      destination_field: "samples-renamed"
+      source_field : "samples",
+      destination_field : "samples-renamed"
     },
     {
-      source_field: "unit",
-      destination_field: "unit"
+      source_field : "unit",
+      destination_field : "unit"
     },
     {
-      source_field: "updated",
-      destination_field: "updated"
+      source_field : "updated",
+      destination_field : "updated"
     }
   ]
 }
diff --git a/examples/resources/clickhouse_clickpipe/resource.tf b/examples/resources/clickhouse_clickpipe/resource.tf
@@ -1,21 +1,23 @@
 resource "clickhouse_clickpipe" "kafka_clickpipe" {
-  name           = "My Kafka ClickPipe"
-  description    = "Data pipeline from Kafka to ClickHouse"
+  name        = "My Kafka ClickPipe"
+  description = "Data pipeline from Kafka to ClickHouse"
 
-  service_id     = "e9465b4b-f7e5-4937-8e21-8d508b02843d"
+  service_id = "e9465b4b-f7e5-4937-8e21-8d508b02843d"
 
   scaling {
-    replicas = 1
+    replicas               = 2
+    replica_cpu_millicores = 250
+    replica_memory_gb      = 1.0
   }
 
   state = "Running"
 
   source {
     kafka {
-      type = "confluent"
-      format = "JSONEachRow"
+      type    = "confluent"
+      format  = "JSONEachRow"
       brokers = "my-kafka-broker:9092"
-      topics = "my_topic"
+      topics  = "my_topic"
 
       consumer_group = "clickpipe-test"
 
@@ -27,9 +29,9 @@ resource "clickhouse_clickpipe" "kafka_clickpipe" {
   }
 
   destination {
-    table    = "my_table"
+    table         = "my_table"
     managed_table = true
-    
+
     tableDefinition {
       engine {
         type = "MergeTree"
@@ -49,7 +51,7 @@ resource "clickhouse_clickpipe" "kafka_clickpipe" {
 
   field_mappings = [
     {
-      source_field = "my_field"
+      source_field      = "my_field"
       destination_field = "my_field1"
     }
   ]
diff --git a/pkg/internal/api/clickpipe.go b/pkg/internal/api/clickpipe.go
@@ -152,8 +152,10 @@ var ClickPipeKafkaOffsetStrategies = []string{
 }
 
 type ClickPipeScalingRequest struct {
-	Replicas    *int64 `json:"replicas,omitempty"`
-	Concurrency *int64 `json:"concurrency,omitempty"`
+	Replicas             *int64   `json:"replicas,omitempty"`
+	ReplicaCpuMillicores *int64   `json:"replicaCpuMillicores,omitempty"`
+	ReplicaMemoryGb      *float64 `json:"replicaMemoryGb,omitempty"`
+	Concurrency          *int64   `json:"concurrency,omitempty"`
 }
 
 type ClickPipeStateRequest struct {
diff --git a/pkg/internal/api/clickpipe_models.go b/pkg/internal/api/clickpipe_models.go
@@ -3,7 +3,9 @@ package api
 import "time"
 
 type ClickPipeScaling struct {
-	Replicas *int64 `json:"replicas,omitempty"`
+	Replicas             *int64   `json:"replicas,omitempty"`
+	ReplicaCpuMillicores *int64   `json:"replicaCpuMillicores,omitempty"`
+	ReplicaMemoryGb      *float64 `json:"replicaMemoryGb,omitempty"`
 }
 
 type ClickPipeSourceCredentials struct {
diff --git a/pkg/resource/clickpipe.go b/pkg/resource/clickpipe.go
@@ -9,6 +9,7 @@ import (
 
 	"github.com/ClickHouse/terraform-provider-clickhouse/pkg/internal/api"
 	"github.com/ClickHouse/terraform-provider-clickhouse/pkg/resource/models"
+	"github.com/hashicorp/terraform-plugin-framework-validators/float64validator"
 	"github.com/hashicorp/terraform-plugin-framework-validators/int64validator"
 	"github.com/hashicorp/terraform-plugin-framework-validators/stringvalidator"
 	"github.com/hashicorp/terraform-plugin-framework/attr"
@@ -110,6 +111,20 @@ func (c *ClickPipeResource) Schema(_ context.Context, _ resource.SchemaRequest,
 							int64validator.Between(1, 10),
 						},
 					},
+					"replica_cpu_millicores": schema.Int64Attribute{
+						Description: "The CPU allocation per replica in millicores. Must be between 125 and 2000.",
+						Optional:    true,
+						Validators: []validator.Int64{
+							int64validator.Between(125, 2000),
+						},
+					},
+					"replica_memory_gb": schema.Float64Attribute{
+						Description: "The memory allocation per replica in GB. Must be between 0.5 and 8.0.",
+						Optional:    true,
+						Validators: []validator.Float64{
+							float64validator.Between(0.5, 8.0),
+						},
+					},
 				},
 				Optional: true,
 				Computed: true,
@@ -741,17 +756,40 @@ func (c *ClickPipeResource) Create(ctx context.Context, request resource.CreateR
 	}
 
 	if !plan.Scaling.IsUnknown() && !plan.Scaling.IsNull() {
-		replicasModel := models.ClickPipeScalingModel{}
-		response.Diagnostics.Append(plan.Scaling.As(ctx, &replicasModel, basetypes.ObjectAsOptions{})...)
+		scalingModel := models.ClickPipeScalingModel{}
+		response.Diagnostics.Append(plan.Scaling.As(ctx, &scalingModel, basetypes.ObjectAsOptions{})...)
 
 		var desiredReplicas *int64
-		if !replicasModel.Replicas.IsUnknown() && !replicasModel.Replicas.IsNull() && createdClickPipe.Scaling.Replicas != nil && *createdClickPipe.Scaling.Replicas != replicasModel.Replicas.ValueInt64() {
-			desiredReplicas = replicasModel.Replicas.ValueInt64Pointer()
+		var desiredCpuMillicores *int64
+		var desiredMemoryGb *float64
+		needsScaling := false
+
+		if !scalingModel.Replicas.IsUnknown() && !scalingModel.Replicas.IsNull() &&
+			(createdClickPipe.Scaling == nil || createdClickPipe.Scaling.Replicas == nil ||
+				*createdClickPipe.Scaling.Replicas != scalingModel.Replicas.ValueInt64()) {
+			desiredReplicas = scalingModel.Replicas.ValueInt64Pointer()
+			needsScaling = true
+		}
+
+		if !scalingModel.ReplicaCpuMillicores.IsUnknown() && !scalingModel.ReplicaCpuMillicores.IsNull() &&
+			(createdClickPipe.Scaling == nil || createdClickPipe.Scaling.ReplicaCpuMillicores == nil ||
+				*createdClickPipe.Scaling.ReplicaCpuMillicores != scalingModel.ReplicaCpuMillicores.ValueInt64()) {
+			desiredCpuMillicores = scalingModel.ReplicaCpuMillicores.ValueInt64Pointer()
+			needsScaling = true
+		}
+
+		if !scalingModel.ReplicaMemoryGb.IsUnknown() && !scalingModel.ReplicaMemoryGb.IsNull() &&
+			(createdClickPipe.Scaling == nil || createdClickPipe.Scaling.ReplicaMemoryGb == nil ||
+				*createdClickPipe.Scaling.ReplicaMemoryGb != scalingModel.ReplicaMemoryGb.ValueFloat64()) {
+			desiredMemoryGb = scalingModel.ReplicaMemoryGb.ValueFloat64Pointer()
+			needsScaling = true
 		}
 
-		if desiredReplicas != nil {
+		if needsScaling {
 			scalingRequest := api.ClickPipeScaling{
-				Replicas: desiredReplicas,
+				Replicas:             desiredReplicas,
+				ReplicaCpuMillicores: desiredCpuMillicores,
+				ReplicaMemoryGb:      desiredMemoryGb,
 			}
 
 			if createdClickPipe, err = c.client.ScalingClickPipe(ctx, serviceID, createdClickPipe.ID, scalingRequest); err != nil {
@@ -1014,9 +1052,11 @@ func (c *ClickPipeResource) syncClickPipeState(ctx context.Context, state *model
 
 	state.State = types.StringValue(clickPipe.State)
 
-	if clickPipe.Scaling != nil && clickPipe.Scaling.Replicas != nil {
+	if clickPipe.Scaling != nil {
 		scalingModel := models.ClickPipeScalingModel{
-			Replicas: types.Int64PointerValue(clickPipe.Scaling.Replicas),
+			Replicas:             types.Int64PointerValue(clickPipe.Scaling.Replicas),
+			ReplicaCpuMillicores: types.Int64PointerValue(clickPipe.Scaling.ReplicaCpuMillicores),
+			ReplicaMemoryGb:      types.Float64PointerValue(clickPipe.Scaling.ReplicaMemoryGb),
 		}
 
 		state.Scaling = scalingModel.ObjectValue()
@@ -1394,11 +1434,13 @@ func (c *ClickPipeResource) Update(ctx context.Context, req resource.UpdateReque
 	}
 
 	if !plan.Scaling.Equal(state.Scaling) {
-		replicasModel := models.ClickPipeScalingModel{}
-		response.Diagnostics.Append(plan.Scaling.As(ctx, &replicasModel, basetypes.ObjectAsOptions{})...)
+		scalingModel := models.ClickPipeScalingModel{}
+		response.Diagnostics.Append(plan.Scaling.As(ctx, &scalingModel, basetypes.ObjectAsOptions{})...)
 
 		scalingRequest := api.ClickPipeScaling{
-			Replicas: replicasModel.Replicas.ValueInt64Pointer(),
+			Replicas:             scalingModel.Replicas.ValueInt64Pointer(),
+			ReplicaCpuMillicores: scalingModel.ReplicaCpuMillicores.ValueInt64Pointer(),
+			ReplicaMemoryGb:      scalingModel.ReplicaMemoryGb.ValueFloat64Pointer(),
 		}
 
 		if _, err := c.client.ScalingClickPipe(ctx, state.ServiceID.ValueString(), state.ID.ValueString(), scalingRequest); err != nil {
diff --git a/pkg/resource/models/clickpipe_resource.go b/pkg/resource/models/clickpipe_resource.go

Original file line number	Diff line number	Diff line change
`@@ -152,8 +152,10 @@ var ClickPipeKafkaOffsetStrategies = []string{`
`152`	`152`	`}`
`153`	`153`
`154`	`154`	`type ClickPipeScalingRequest struct {`
`155`		- Replicas *int64 `json:"replicas,omitempty"`
`156`		- Concurrency *int64 `json:"concurrency,omitempty"`
	`155`	+ Replicas *int64 `json:"replicas,omitempty"`
	`156`	+ ReplicaCpuMillicores *int64 `json:"replicaCpuMillicores,omitempty"`
	`157`	+ ReplicaMemoryGb *float64 `json:"replicaMemoryGb,omitempty"`
	`158`	+ Concurrency *int64 `json:"concurrency,omitempty"`
`157`	`159`	`}`
`158`	`160`
`159`	`161`	`type ClickPipeStateRequest struct {`
Original file line number	Diff line number	Diff line change
`@@ -3,7 +3,9 @@ package api`
`3`	`3`	`import "time"`
`4`	`4`
`5`	`5`	`type ClickPipeScaling struct {`
`6`		- Replicas *int64 `json:"replicas,omitempty"`
	`6`	+ Replicas *int64 `json:"replicas,omitempty"`
	`7`	+ ReplicaCpuMillicores *int64 `json:"replicaCpuMillicores,omitempty"`
	`8`	+ ReplicaMemoryGb *float64 `json:"replicaMemoryGb,omitempty"`
`7`	`9`	`}`
`8`	`10`
`9`	`11`	`type ClickPipeSourceCredentials struct {`