feat(cluster): expose broker autoscaling on streamnative_pulsar_cluster

cloud/provider.go

@@ -81,18 +81,19 @@ func init() {
 		"instance_engine":              "The streamnative cloud instance engine, supporting 'ursa' and 'classic', default 'classic'",
 		"location": "The location of the pulsar cluster, " +
 			"supported location https://docs.streamnative.io/docs/cluster#cluster-location",
-		"release_channel":         "The release channel of the pulsar cluster subscribe to, it must to be lts or rapid, default rapid",
-		"bookie_replicas":         "The number of bookie replicas",
-		"broker_replicas":         "The number of broker replicas",
-		"compute_unit_per_broker": "compute unit per broker, 1 compute unit is 2 cpu and 8gb memory",
-		"storage_unit_per_bookie": "storage unit per bookie, 1 storage unit is 2 cpu and 8gb memory",
-		"cluster_ready":           "Pulsar cluster is ready, it will be set to 'True' after the cluster is ready",
-		"instance_ready":          "Pulsar instance is ready, it will be set to 'True' after the instance is ready",
-		"websocket_enabled":       "Whether the websocket is enabled",
-		"function_enabled":        "Whether the function is enabled",
-		"transaction_enabled":     "Whether the transaction is enabled",
-		"kafka":                   "Controls the kafka protocol config of pulsar cluster",
-		"mqtt":                    "Controls the mqtt protocol config of pulsar cluster",
+		"release_channel":            "The release channel of the pulsar cluster subscribe to, it must to be lts or rapid, default rapid",
+		"bookie_replicas":            "The number of bookie replicas",
+		"broker_replicas":            "The number of broker replicas",
+		"broker_auto_scaling_policy": "Autoscaling bounds for brokers. When set, the number of brokers is managed automatically between min_replicas and max_replicas and broker_replicas no longer takes effect. Requires the autoscaling feature to be enabled for the organization, and is not supported on serverless instances.",
+		"compute_unit_per_broker":    "compute unit per broker, 1 compute unit is 2 cpu and 8gb memory",
+		"storage_unit_per_bookie":    "storage unit per bookie, 1 storage unit is 2 cpu and 8gb memory",
+		"cluster_ready":              "Pulsar cluster is ready, it will be set to 'True' after the cluster is ready",
+		"instance_ready":             "Pulsar instance is ready, it will be set to 'True' after the instance is ready",
+		"websocket_enabled":          "Whether the websocket is enabled",
+		"function_enabled":           "Whether the function is enabled",
+		"transaction_enabled":        "Whether the transaction is enabled",
+		"kafka":                      "Controls the kafka protocol config of pulsar cluster",
+		"mqtt":                       "Controls the mqtt protocol config of pulsar cluster",
 		"categories": "Controls the audit log categories config of pulsar cluster, supported categories: " +
 			"\"Management\", \"Describe\", \"Produce\", \"Consume\"",
 		"lakehouse_type":          "The type of the lakehouse",

cloud/pulsar_cluster_state_test.go

@@ -20,6 +20,7 @@ import (
 	"time"
 
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/terraform"
 	cloudv1alpha1 "github.com/streamnative/cloud-api-server/pkg/apis/cloud/v1alpha1"
 	"github.com/stretchr/testify/assert"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -295,3 +296,206 @@ func TestMaintenanceWindowEqualWhenDifferent(t *testing.T) {
 
 	assert.False(t, maintenanceWindowEqual(expected, actual))
 }
+
+func TestExpandBrokerAutoScalingPolicy(t *testing.T) {
+	policy := expandBrokerAutoScalingPolicy([]interface{}{
+		map[string]interface{}{
+			"min_replicas": 2,
+			"max_replicas": 6,
+		},
+	})
+
+	assert.NotNil(t, policy)
+	assert.Equal(t, int32(6), policy.MaxReplicas)
+	assert.NotNil(t, policy.MinReplicas)
+	assert.Equal(t, int32(2), *policy.MinReplicas)
+}
+
+// min_replicas is Required, so the schema rejects an omitted value before expand is reached. What
+// expand must never do is drop a minimum it was given: sn-operator's GetAutoScalingPolicyWithDefault
+// returns a nil policy whenever MinReplicas is nil, which builds no HorizontalPodAutoscaler and
+// leaves autoscaling silently off.
+func TestExpandBrokerAutoScalingPolicyAlwaysSendsMinReplicas(t *testing.T) {
+	policy := expandBrokerAutoScalingPolicy([]interface{}{
+		map[string]interface{}{
+			"min_replicas": 1,
+			"max_replicas": 4,
+		},
+	})
+
+	assert.NotNil(t, policy)
+	assert.Equal(t, int32(4), policy.MaxReplicas)
+	assert.NotNil(t, policy.MinReplicas)
+	assert.Equal(t, int32(1), *policy.MinReplicas)
+}
+
+func TestBrokerAutoScalingPolicySchemaRequiresMinReplicas(t *testing.T) {
+	elem := resourcePulsarCluster().Schema["broker_auto_scaling_policy"].Elem.(*schema.Resource)
+
+	assert.True(t, elem.Schema["min_replicas"].Required,
+		"min_replicas must be Required; an absent minimum disables the downstream HPA")
+	assert.False(t, elem.Schema["min_replicas"].Computed,
+		"min_replicas must not be Computed; a server-filled value would mask the absent minimum")
+}
+
+func TestExpandBrokerAutoScalingPolicyEmpty(t *testing.T) {
+	assert.Nil(t, expandBrokerAutoScalingPolicy(nil))
+	assert.Nil(t, expandBrokerAutoScalingPolicy([]interface{}{}))
+	assert.Nil(t, expandBrokerAutoScalingPolicy([]interface{}{nil}))
+}
+
+func TestFlattenBrokerAutoScalingPolicy(t *testing.T) {
+	minReplicas := int32(3)
+	flattened := flattenBrokerAutoScalingPolicy(&cloudv1alpha1.AutoScalingPolicy{
+		MinReplicas: &minReplicas,
+		MaxReplicas: 9,
+	})
+
+	assert.Len(t, flattened, 1)
+	assert.Equal(t, map[string]interface{}{"min_replicas": 3, "max_replicas": 9}, flattened[0])
+	assert.Equal(t, []interface{}{}, flattenBrokerAutoScalingPolicy(nil))
+}
+
+func TestBrokerAutoScalingPolicyEqual(t *testing.T) {
+	minReplicas := int32(2)
+	expected := &cloudv1alpha1.AutoScalingPolicy{MinReplicas: &minReplicas, MaxReplicas: 6}
+	actualMin := int32(2)
+	actual := &cloudv1alpha1.AutoScalingPolicy{MinReplicas: &actualMin, MaxReplicas: 6}
+
+	assert.True(t, brokerAutoScalingPolicyEqual(expected, actual))
+	assert.True(t, brokerAutoScalingPolicyEqual(nil, nil))
+	assert.False(t, brokerAutoScalingPolicyEqual(expected, nil))
+	assert.False(t, brokerAutoScalingPolicyEqual(nil, actual))
+}
+
+func TestBrokerAutoScalingPolicyEqualWhenDifferent(t *testing.T) {
+	minReplicas := int32(2)
+	expected := &cloudv1alpha1.AutoScalingPolicy{MinReplicas: &minReplicas, MaxReplicas: 6}
+
+	differentMax := &cloudv1alpha1.AutoScalingPolicy{MinReplicas: &minReplicas, MaxReplicas: 7}
+	assert.False(t, brokerAutoScalingPolicyEqual(expected, differentMax))
+
+	otherMin := int32(3)
+	differentMin := &cloudv1alpha1.AutoScalingPolicy{MinReplicas: &otherMin, MaxReplicas: 6}
+	assert.False(t, brokerAutoScalingPolicyEqual(expected, differentMin))
+
+	droppedMin := &cloudv1alpha1.AutoScalingPolicy{MaxReplicas: 6}
+	assert.False(t, brokerAutoScalingPolicyEqual(expected, droppedMin))
+}
+
+// A min_replicas the user did not ask for is the control plane's to choose, so a server-filled value
+// must not read as drift.
+// The dry-run comparison used to treat a nil expected minimum as "whatever the control plane fills
+// in", which hid the case this PR now prevents: a request carrying no minimum builds no HPA at all.
+// min_replicas is Required, so a nil on either side is a real mismatch rather than a defaulted value.
+func TestBrokerAutoScalingPolicyEqualRejectsMissingMinReplicas(t *testing.T) {
+	serverFilled := int32(2)
+	withMin := &cloudv1alpha1.AutoScalingPolicy{MinReplicas: &serverFilled, MaxReplicas: 6}
+	withoutMin := &cloudv1alpha1.AutoScalingPolicy{MaxReplicas: 6}
+
+	assert.False(t, brokerAutoScalingPolicyEqual(withoutMin, withMin),
+		"a minimum the request never asked for is not an accepted policy")
+	assert.False(t, brokerAutoScalingPolicyEqual(withMin, withoutMin),
+		"a dropped minimum means the policy was not accepted as sent")
+	assert.True(t, brokerAutoScalingPolicyEqual(withoutMin, withoutMin),
+		"two policies without a minimum still compare equal")
+}
+
+func TestValidateBrokerAutoScalingPolicyAccepted(t *testing.T) {
+	minReplicas := int32(2)
+	expected := &cloudv1alpha1.AutoScalingPolicy{MinReplicas: &minReplicas, MaxReplicas: 6}
+
+	assert.NoError(t, validateBrokerAutoScalingPolicyAccepted(expected, expected, "CREATE"))
+}
+
+func TestValidateBrokerAutoScalingPolicyAcceptedWhenDropped(t *testing.T) {
+	expected := &cloudv1alpha1.AutoScalingPolicy{MaxReplicas: 6}
+
+	err := validateBrokerAutoScalingPolicyAccepted(expected, nil, "CREATE")
+	assert.EqualError(t, err,
+		"ERROR_CREATE_PULSAR_CLUSTER: broker_auto_scaling_policy is not enabled for this organization")
+}
+
+// The control plane pins serverless clusters to its own policy; catching that as a mismatch is what
+// stops a silently overridden configuration from being reported as applied.
+func TestValidateBrokerAutoScalingPolicyAcceptedWhenOverridden(t *testing.T) {
+	requestedMin := int32(1)
+	expected := &cloudv1alpha1.AutoScalingPolicy{MinReplicas: &requestedMin, MaxReplicas: 10}
+	pinnedMin := int32(2)
+	actual := &cloudv1alpha1.AutoScalingPolicy{MinReplicas: &pinnedMin, MaxReplicas: 3}
+
+	err := validateBrokerAutoScalingPolicyAccepted(expected, actual, "UPDATE")
+	assert.EqualError(t, err,
+		"ERROR_UPDATE_PULSAR_CLUSTER: broker_auto_scaling_policy is not enabled for this organization")
+}
+
+// An unmanaged policy must never reach state. The block is Optional with no Computed, so a policy
+// Terraform did not write would make an absent configuration plan broker_auto_scaling_policy.# from
+// 1 to 0 -- rejected by the update guard on serverless, and a silent disable on dedicated.
+func TestBrokerAutoScalingPolicyForState(t *testing.T) {
+	minReplicas := int32(2)
+	policy := &cloudv1alpha1.AutoScalingPolicy{MinReplicas: &minReplicas, MaxReplicas: 6}
+	managed := []interface{}{map[string]interface{}{"min_replicas": 2, "max_replicas": 6}}
+
+	assert.Equal(t, managed, brokerAutoScalingPolicyForState(policy, true, false),
+		"a policy Terraform manages on a dedicated cluster is surfaced")
+	assert.Equal(t, []interface{}{}, brokerAutoScalingPolicyForState(policy, false, false),
+		"a policy set out of band is not adopted into state")
+	assert.Equal(t, []interface{}{}, brokerAutoScalingPolicyForState(policy, true, true),
+		"the serverless admission-injected policy is never surfaced")
+	assert.Equal(t, []interface{}{}, brokerAutoScalingPolicyForState(nil, true, false),
+		"an absent policy clears state so removal drift is still detected")
+}
+
+// Plan-level regression for the same issue: once refresh leaves an unmanaged policy out of state, a
+// configuration without the block must produce no broker_auto_scaling_policy change at all.
+func TestBrokerAutoScalingPolicyUnmanagedProducesNoPlan(t *testing.T) {
+	state := &terraform.InstanceState{
+		ID:         "test-cluster",
+		Attributes: map[string]string{"id": "test-cluster"},
+	}
+
+	diff, err := resourcePulsarCluster().SimpleDiff(
+		context.Background(), state, terraform.NewResourceConfigRaw(map[string]interface{}{}), nil)
+	assert.NoError(t, err)
+	if diff != nil {
+		_, planned := diff.Attributes["broker_auto_scaling_policy.#"]
+		assert.False(t, planned,
+			"an unmanaged policy absent from state must not plan a broker_auto_scaling_policy change")
+	}
+}
+
+func TestValidateBrokerAutoScalingPolicyBounds(t *testing.T) {
+	boundsFor := func(minReplicas, maxReplicas int) []interface{} {
+		return []interface{}{
+			map[string]interface{}{"min_replicas": minReplicas, "max_replicas": maxReplicas},
+		}
+	}
+
+	err := validateBrokerAutoScalingPolicyBounds(boundsFor(10, 5))
+	assert.Error(t, err, "an inverted range must fail at plan time, not at Kubernetes reconcile")
+	assert.Contains(t, err.Error(), "min_replicas (10) must be less than or equal to max_replicas (5)")
+
+	assert.NoError(t, validateBrokerAutoScalingPolicyBounds(boundsFor(3, 3)), "an equal range is valid")
+	assert.NoError(t, validateBrokerAutoScalingPolicyBounds(boundsFor(2, 6)), "an ascending range is valid")
+	assert.NoError(t, validateBrokerAutoScalingPolicyBounds(nil), "an absent block carries no bounds")
+	assert.NoError(t, validateBrokerAutoScalingPolicyBounds([]interface{}{nil}), "a nil entry is ignored")
+}
+
+// The inverted range must surface through the resource's CustomizeDiff, not just the helper.
+func TestBrokerAutoScalingPolicyBoundsRejectedAtPlan(t *testing.T) {
+	state := &terraform.InstanceState{
+		ID:         "test-cluster",
+		Attributes: map[string]string{"id": "test-cluster"},
+	}
+
+	_, err := resourcePulsarCluster().SimpleDiff(context.Background(), state,
+		terraform.NewResourceConfigRaw(map[string]interface{}{
+			"broker_auto_scaling_policy": []interface{}{
+				map[string]interface{}{"min_replicas": 10, "max_replicas": 5},
+			},
+		}), nil)
+
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "min_replicas (10) must be less than or equal to max_replicas (5)")
+}