feat: implement Balanced consolidation policy

Adds a new consolidationPolicy value, Balanced, that scores each
consolidation move and rejects moves where the disruption outweighs the
savings. Gated behind --feature-gates BalancedConsolidation=true.

The scoring formula compares savings and disruption as fractions of
NodePool totals: score = savings_fraction / disruption_fraction. A move
is approved when score >= 1/consolidationThreshold (default 2). The
scoring step is a filter inserted after scheduling feasibility and price
comparison. It can only reject moves, never create them. If scoring has
a bug that incorrectly approves, the move was already feasible and
cost-saving. If it incorrectly rejects, the cluster is less optimized
but not disrupted.

API:
- consolidationPolicy: Balanced (new enum value)
- consolidationThreshold: 1-3 (default 2, requires Balanced)

Implementation:
- balanced.go: scoring formula, NodePool totals, candidate pre-filter,
  cross-NodePool move evaluation
- Feature gate, API validation (CEL + runtime), defaulting
- ShouldDisrupt accepts Balanced, sets ConsolidationPolicyUnsupported
  status condition when gate is disabled
- Score-based candidate ranking for single-node consolidation
- Events (ConsolidationApproved/Rejected on Node+NodeClaim for
  single-node, NodePool for multi-node)
- Metrics (consolidation_score histogram, consolidation_moves_total
  counter)

Tests (31 new):
- 15 unit tests covering all RFC worked examples
- 9 integration tests (NodePool totals, cross-pool, candidate price)
- 3 feature gate tests
- 5 validation + 4 defaulting tests
- 4 score-based ranking tests
- 1 status condition test

See designs/balanced-consolidation.md (PR #2942) for the full RFC.

Author: jamesmt-aws

karpenter-triage

@@ -0,0 +1 @@
+Subproject commit 28d417564c9e463343ff641f14d1ec1b0c51389a

pkg/apis/crds/karpenter.sh_nodeclaims.yaml

@@ -80,6 +80,7 @@ type NodePoolSpec struct {
 	Replicas *int64 `json:"replicas,omitempty"`
 }
 
+// +kubebuilder:validation:XValidation:rule="!has(self.consolidationThreshold) || self.consolidationPolicy == 'Balanced'",message="consolidationThreshold is only valid when consolidationPolicy is Balanced"
 type Disruption struct {
 	//nolint:kubeapilinter
 	// ConsolidateAfter is the duration the controller will wait
@@ -96,10 +97,19 @@ type Disruption struct {
 	// algorithm. This policy defaults to "WhenEmptyOrUnderutilized" if not specified
 	// When replicas is set, ConsolidationPolicy is simply ignored
 	// +kubebuilder:default:="WhenEmptyOrUnderutilized"
-	// +kubebuilder:validation:Enum:={WhenEmpty,WhenEmptyOrUnderutilized}
+	// +kubebuilder:validation:Enum:={WhenEmpty,WhenEmptyOrUnderutilized,Balanced}
 	// +optional
 	ConsolidationPolicy ConsolidationPolicy `json:"consolidationPolicy,omitempty"`
 	//nolint:kubeapilinter
+	// ConsolidationThreshold controls how much disruption one unit of savings can buy
+	// when using the Balanced consolidation policy. A move is approved when
+	// score >= 1/consolidationThreshold. Higher values approve more aggressively.
+	// Only valid when consolidationPolicy is "Balanced". Defaults to 2.
+	// +kubebuilder:validation:Minimum:=1
+	// +kubebuilder:validation:Maximum:=3
+	// +optional
+	ConsolidationThreshold *int32 `json:"consolidationThreshold,omitempty"`
+	//nolint:kubeapilinter
 	// Budgets is a list of Budgets.
 	// If there are multiple active budgets, Karpenter uses
 	// the most restrictive value. If left undefined,
@@ -158,8 +168,12 @@ type ConsolidationPolicy string
 const (
 	ConsolidationPolicyWhenEmpty                ConsolidationPolicy = "WhenEmpty"
 	ConsolidationPolicyWhenEmptyOrUnderutilized ConsolidationPolicy = "WhenEmptyOrUnderutilized"
+	ConsolidationPolicyBalanced                 ConsolidationPolicy = "Balanced"
 )
 
+// DefaultConsolidationThreshold is the default value for ConsolidationThreshold when using the Balanced policy.
+const DefaultConsolidationThreshold int32 = 2
+
 // DisruptionReason defines valid reasons for disruption budgets.
 // +kubebuilder:validation:Enum={Underutilized,Empty,Drifted}
 type DisruptionReason string

pkg/apis/crds/karpenter.sh_nodeoverlays.yaml

@@ -0,0 +1,191 @@
+/*
+Copyright The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package v1
+
+import (
+	"context"
+	"strings"
+	"testing"
+)
+
+func int32Ptr(v int32) *int32 { return &v }
+
+// --- Test 2: Validation ---
+
+// TestValidateConsolidationThreshold_RejectsWithoutBalanced verifies that
+// RuntimeValidate returns an error when consolidationThreshold is set but
+// consolidationPolicy is not Balanced.
+func TestValidateConsolidationThreshold_RejectsWithoutBalanced(t *testing.T) {
+	np := &NodePool{}
+	np.Spec.Disruption.ConsolidationPolicy = ConsolidationPolicyWhenEmptyOrUnderutilized
+	np.Spec.Disruption.ConsolidationThreshold = int32Ptr(2)
+
+	err := np.RuntimeValidate(context.Background())
+	if err == nil {
+		t.Fatal("expected error when consolidationThreshold is set with non-Balanced policy")
+	}
+	if !strings.Contains(err.Error(), "consolidationThreshold is only valid when consolidationPolicy is Balanced") {
+		t.Errorf("unexpected error message: %s", err.Error())
+	}
+}
+
+// TestValidateConsolidationThreshold_RejectsWithWhenEmpty verifies rejection
+// when policy is WhenEmpty.
+func TestValidateConsolidationThreshold_RejectsWithWhenEmpty(t *testing.T) {
+	np := &NodePool{}
+	np.Spec.Disruption.ConsolidationPolicy = ConsolidationPolicyWhenEmpty
+	np.Spec.Disruption.ConsolidationThreshold = int32Ptr(2)
+
+	err := np.RuntimeValidate(context.Background())
+	if err == nil {
+		t.Fatal("expected error when consolidationThreshold is set with WhenEmpty policy")
+	}
+}
+
+// TestValidateConsolidationThreshold_PassesWithBalanced verifies that validation
+// passes when consolidationThreshold is set with consolidationPolicy: Balanced.
+func TestValidateConsolidationThreshold_PassesWithBalanced(t *testing.T) {
+	np := &NodePool{}
+	np.Spec.Disruption.ConsolidationPolicy = ConsolidationPolicyBalanced
+	np.Spec.Disruption.ConsolidationThreshold = int32Ptr(2)
+
+	err := np.Spec.Disruption.validateConsolidationThreshold()
+	if err != nil {
+		t.Errorf("expected no error for Balanced policy with threshold, got: %s", err)
+	}
+}
+
+// TestValidateConsolidationThreshold_PassesWithNilThreshold verifies that
+// validation passes when consolidationThreshold is nil regardless of policy.
+func TestValidateConsolidationThreshold_PassesWithNilThreshold(t *testing.T) {
+	np := &NodePool{}
+	np.Spec.Disruption.ConsolidationPolicy = ConsolidationPolicyWhenEmptyOrUnderutilized
+	np.Spec.Disruption.ConsolidationThreshold = nil
+
+	err := np.Spec.Disruption.validateConsolidationThreshold()
+	if err != nil {
+		t.Errorf("expected no error when threshold is nil, got: %s", err)
+	}
+}
+
+// TestValidateConsolidationThreshold_BoundaryValues tests the valid range [1, 3].
+// Note: The range [1, 3] is enforced by kubebuilder CEL validation at the API level.
+// The RuntimeValidate only checks the policy association. These tests verify the
+// policy check works for each valid value.
+func TestValidateConsolidationThreshold_BoundaryValues(t *testing.T) {
+	tests := []struct {
+		name      string
+		threshold int32
+		policy    ConsolidationPolicy
+		wantErr   bool
+	}{
+		{"threshold 1 with Balanced", 1, ConsolidationPolicyBalanced, false},
+		{"threshold 2 with Balanced", 2, ConsolidationPolicyBalanced, false},
+		{"threshold 3 with Balanced", 3, ConsolidationPolicyBalanced, false},
+		{"threshold 1 with WhenEmptyOrUnderutilized", 1, ConsolidationPolicyWhenEmptyOrUnderutilized, true},
+		{"threshold 3 with WhenEmptyOrUnderutilized", 3, ConsolidationPolicyWhenEmptyOrUnderutilized, true},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			d := &Disruption{
+				ConsolidationPolicy:    tc.policy,
+				ConsolidationThreshold: int32Ptr(tc.threshold),
+			}
+			err := d.validateConsolidationThreshold()
+			if tc.wantErr && err == nil {
+				t.Errorf("expected error, got nil")
+			}
+			if !tc.wantErr && err != nil {
+				t.Errorf("expected no error, got: %s", err)
+			}
+		})
+	}
+}
+
+// --- Test 3: Defaulting ---
+
+// TestSetDefaults_BalancedPolicyDefaultsThresholdTo2 verifies that SetDefaults
+// sets consolidationThreshold to 2 when consolidationPolicy is Balanced and
+// threshold is nil.
+func TestSetDefaults_BalancedPolicyDefaultsThresholdTo2(t *testing.T) {
+	np := &NodePool{}
+	np.Spec.Disruption.ConsolidationPolicy = ConsolidationPolicyBalanced
+	np.Spec.Disruption.ConsolidationThreshold = nil
+
+	np.SetDefaults(context.Background())
+
+	if np.Spec.Disruption.ConsolidationThreshold == nil {
+		t.Fatal("expected consolidationThreshold to be set, got nil")
+	}
+	if *np.Spec.Disruption.ConsolidationThreshold != DefaultConsolidationThreshold {
+		t.Errorf("expected threshold %d, got %d", DefaultConsolidationThreshold, *np.Spec.Disruption.ConsolidationThreshold)
+	}
+	if *np.Spec.Disruption.ConsolidationThreshold != 2 {
+		t.Errorf("expected threshold 2, got %d", *np.Spec.Disruption.ConsolidationThreshold)
+	}
+}
+
+// TestSetDefaults_NonBalancedPolicyDoesNotSetThreshold verifies that SetDefaults
+// does NOT set consolidationThreshold when policy is not Balanced.
+func TestSetDefaults_NonBalancedPolicyDoesNotSetThreshold(t *testing.T) {
+	policies := []ConsolidationPolicy{
+		ConsolidationPolicyWhenEmpty,
+		ConsolidationPolicyWhenEmptyOrUnderutilized,
+	}
+
+	for _, policy := range policies {
+		t.Run(string(policy), func(t *testing.T) {
+			np := &NodePool{}
+			np.Spec.Disruption.ConsolidationPolicy = policy
+			np.Spec.Disruption.ConsolidationThreshold = nil
+
+			np.SetDefaults(context.Background())
+
+			if np.Spec.Disruption.ConsolidationThreshold != nil {
+				t.Errorf("expected threshold to remain nil for %s policy, got %d", policy, *np.Spec.Disruption.ConsolidationThreshold)
+			}
+		})
+	}
+}
+
+// TestSetDefaults_DoesNotOverrideExplicitThreshold verifies that SetDefaults
+// does NOT override an explicitly set threshold.
+func TestSetDefaults_DoesNotOverrideExplicitThreshold(t *testing.T) {
+	np := &NodePool{}
+	np.Spec.Disruption.ConsolidationPolicy = ConsolidationPolicyBalanced
+	np.Spec.Disruption.ConsolidationThreshold = int32Ptr(3)
+
+	np.SetDefaults(context.Background())
+
+	if *np.Spec.Disruption.ConsolidationThreshold != 3 {
+		t.Errorf("expected threshold to remain 3, got %d", *np.Spec.Disruption.ConsolidationThreshold)
+	}
+}
+
+// TestSetDefaults_ExplicitThreshold1Preserved verifies threshold=1 is preserved.
+func TestSetDefaults_ExplicitThreshold1Preserved(t *testing.T) {
+	np := &NodePool{}
+	np.Spec.Disruption.ConsolidationPolicy = ConsolidationPolicyBalanced
+	np.Spec.Disruption.ConsolidationThreshold = int32Ptr(1)
+
+	np.SetDefaults(context.Background())
+
+	if *np.Spec.Disruption.ConsolidationThreshold != 1 {
+		t.Errorf("expected threshold to remain 1, got %d", *np.Spec.Disruption.ConsolidationThreshold)
+	}
+}

pkg/apis/crds/karpenter.sh_nodepools.yaml

@@ -21,4 +21,9 @@ import (
 )
 
 // SetDefaults for the NodePool
-func (in *NodePool) SetDefaults(_ context.Context) {}
+func (in *NodePool) SetDefaults(_ context.Context) {
+	if in.Spec.Disruption.ConsolidationPolicy == ConsolidationPolicyBalanced && in.Spec.Disruption.ConsolidationThreshold == nil {
+		threshold := DefaultConsolidationThreshold
+		in.Spec.Disruption.ConsolidationThreshold = &threshold
+	}
+}

pkg/apis/v1/nodepool.go

@@ -29,6 +29,9 @@ const (
 	ConditionTypeNodeClassReady = "NodeClassReady"
 	// ConditionTypeNodeRegistrationHealthy = "NodeRegistrationHealthy" condition indicates if a misconfiguration exists that is preventing successful node launch/registrations that requires manual investigation
 	ConditionTypeNodeRegistrationHealthy = "NodeRegistrationHealthy"
+	// ConditionTypeConsolidationPolicyUnsupported indicates that the NodePool's consolidationPolicy
+	// is set to Balanced but the BalancedConsolidation feature gate is disabled.
+	ConditionTypeConsolidationPolicyUnsupported = "ConsolidationPolicyUnsupported"
 )
 
 // NodePoolStatus defines the observed state of NodePool

pkg/apis/v1/nodepool_balanced_test.go

@@ -26,10 +26,17 @@ import (
 
 // RuntimeValidate will be used to validate any part of the CRD that can not be validated at CRD creation
 func (in *NodePool) RuntimeValidate(ctx context.Context) (errs error) {
-	errs = multierr.Combine(in.Spec.Template.validateLabels(), in.Spec.Template.Spec.validateTaints(), in.Spec.Template.Spec.validateRequirements(ctx), in.Spec.Template.validateRequirementsNodePoolKeyDoesNotExist())
+	errs = multierr.Combine(in.Spec.Template.validateLabels(), in.Spec.Template.Spec.validateTaints(), in.Spec.Template.Spec.validateRequirements(ctx), in.Spec.Template.validateRequirementsNodePoolKeyDoesNotExist(), in.Spec.Disruption.validateConsolidationThreshold())
 	return errs
 }
 
+func (in *Disruption) validateConsolidationThreshold() error {
+	if in.ConsolidationThreshold != nil && in.ConsolidationPolicy != ConsolidationPolicyBalanced {
+		return fmt.Errorf("consolidationThreshold is only valid when consolidationPolicy is Balanced")
+	}
+	return nil
+}
+
 func (in *NodeClaimTemplate) validateLabels() (errs error) {
 	for key, value := range in.Labels {
 		if key == NodePoolLabelKey {