fix: allow non-churn empty nodes to be disrupted

pkg/controllers/disruption/consolidation_test.go

@@ -2302,7 +2302,7 @@ var _ = Describe("Consolidation", func() {
 			// and delete the old one
 			ExpectNotFound(ctx, env.Client, nodeClaims[1], nodes[1])
 		})
-		It("does not delete nodes when there is pod churn", func() {
+		It("does not delete nodes with pod churn, deletes nodes without pod churn", func() {
 			// create our RS so we can link a pod to it
 			ExpectApplied(ctx, env.Client, nodePool)
 			for i := range 2 {
@@ -2329,7 +2329,10 @@ var _ = Describe("Consolidation", func() {
 			wg.Wait()
 			Expect(err).To(Succeed())
 			Expect(results).To(Equal(pscheduling.Results{}))
-			Expect(cmd).To(Equal(disruption.Command{}))
+			Expect(cmd.Candidates()).To(HaveLen(1))
+			// the test validator manually binds a pod to nodes[0], causing it to no longer be eligible
+			Expect(cmd.Candidates()[0].StateNode.Node.Name).To(Equal(nodes[1].Name))
+			Expect(cmd.Decision()).To(Equal(disruption.DeleteDecision))
 
 			Expect(emptyConsolidation.IsConsolidated()).To(BeFalse())
 

pkg/controllers/disruption/emptiness.go

@@ -21,6 +21,7 @@ import (
 	"errors"
 	"fmt"
 
+	"github.com/samber/lo"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 
 	v1 "sigs.k8s.io/karpenter/pkg/apis/v1"
@@ -60,22 +61,18 @@ func (e *Emptiness) ComputeCommand(ctx context.Context, disruptionBudgetMapping
 	}
 	candidates = e.sortCandidates(candidates)
 
-	empty := make([]*Candidate, 0, len(candidates))
 	constrainedByBudgets := false
-	for _, candidate := range candidates {
+	empty := lo.Filter(candidates, func(candidate *Candidate, _ int) bool {
 		if len(candidate.reschedulablePods) > 0 {
-			continue
+			return false
 		}
 		if disruptionBudgetMapping[candidate.NodePool.Name] == 0 {
-			// set constrainedByBudgets to true if any node was a candidate but was constrained by a budget
 			constrainedByBudgets = true
-			continue
+			return false
 		}
-		// If there's disruptions allowed for the candidate's nodepool,
-		// add it to the list of candidates, and decrement the budget.
-		empty = append(empty, candidate)
 		disruptionBudgetMapping[candidate.NodePool.Name]--
-	}
+		return true
+	})
 	// none empty, so do nothing
 	if len(empty) == 0 {
 		// if there are no candidates, but a nodepool had a fully blocking budget,

pkg/controllers/disruption/types.go

@@ -141,6 +141,10 @@ func (c Command) Decision() Decision {
 	}
 }
 
+func (c Command) Candidates() []*Candidate {
+	return c.candidates
+}
+
 func (c Command) LogValues() []any {
 	podCount := lo.Reduce(c.candidates, func(_ int, cd *Candidate, _ int) int { return len(cd.reschedulablePods) }, 0)
 

pkg/controllers/disruption/validation.go

@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/samber/lo"
 	"k8s.io/utils/clock"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
@@ -149,6 +150,32 @@ func (c *ConsolidationValidator) isValid(ctx context.Context, cmd Command, valid
 	return nil
 }
 
+func (e *EmptinessValidator) validateCandidates(ctx context.Context, candidates ...*Candidate) ([]*Candidate, error) {
+	validatedCandidates, err := GetCandidates(ctx, e.cluster, e.kubeClient, e.recorder, e.clock, e.cloudProvider, e.filter, GracefulDisruptionClass, e.queue)
+	if err != nil {
+		return nil, fmt.Errorf("constructing validation candidates, %w", err)
+	}
+	if len(validatedCandidates) == 0 {
+		return nil, NewValidationError(fmt.Errorf("%d candidates are no longer valid", len(candidates)))
+	}
+	disruptionBudgetMapping, err := BuildDisruptionBudgetMapping(ctx, e.cluster, e.clock, e.kubeClient, e.cloudProvider, e.recorder, e.reason)
+	if err != nil {
+		return nil, fmt.Errorf("building disruption budgets, %w", err)
+	}
+
+	if valid := lo.Filter(validatedCandidates, func(cn *Candidate, _ int) bool {
+		if e.cluster.IsNodeNominated(cn.ProviderID()) || disruptionBudgetMapping[cn.NodePool.Name] == 0 {
+			return false
+		}
+		disruptionBudgetMapping[cn.NodePool.Name]--
+		return true
+	}); len(valid) > 0 {
+		return valid, nil
+	}
+	return nil, NewValidationError(fmt.Errorf("a candidate failed validation because it was nominated for a pod or would violate disruption budgets"))
+
+}
+
 // ValidateCandidates gets the current representation of the provided candidates and ensures that they are all still valid.
 // For a candidate to still be valid, the following conditions must be met:
 //

pkg/controllers/nodeclaim/disruption/drift.go

@@ -135,7 +135,7 @@ func instanceTypeNotFound(its []*cloudprovider.InstanceType, nodeClaim *v1.NodeC
 }
 
 // Eligible fields for drift are described in the docs
-// https://karpenter.sh/docs/concepts/deprovisioning/#drift
+// https://karpenter.sh/docs/concepts/disruption/#drift
 func areStaticFieldsDrifted(nodePool *v1.NodePool, nodeClaim *v1.NodeClaim) cloudprovider.DriftReason {
 	nodePoolHash, foundNodePoolHash := nodePool.Annotations[v1.NodePoolHashAnnotationKey]
 	nodePoolHashVersion, foundNodePoolHashVersion := nodePool.Annotations[v1.NodePoolHashVersionAnnotationKey]