From b346b0de2ea02a3924c1665afd87dec4789eac6b Mon Sep 17 00:00:00 2001 From: Brandon Wagner Date: Thu, 20 Aug 2026 14:43:42 -0400 Subject: [PATCH] fix: skip adding spot requirement during consolidation when no spot offerings are available --- pkg/controllers/disruption/consolidation.go | 9 +- .../disruption/consolidation_test.go | 93 +++++++++++++++++++ 2 files changed, 101 insertions(+), 1 deletion(-) diff --git a/pkg/controllers/disruption/consolidation.go b/pkg/controllers/disruption/consolidation.go index 7f4ea07dde..bcde2884ec 100644 --- a/pkg/controllers/disruption/consolidation.go +++ b/pkg/controllers/disruption/consolidation.go @@ -238,7 +238,7 @@ func (c *consolidation) computeConsolidation(ctx context.Context, candidates ... // should fail and we'll just leave the node alone. We don't need to do the same for reserved since the requirements // are injected on by the scheduler. ctReq := results.NewNodeClaims[0].Requirements.Get(v1.CapacityTypeLabelKey) - if ctReq.Has(v1.CapacityTypeSpot) && ctReq.Has(v1.CapacityTypeOnDemand) { + if ctReq.Has(v1.CapacityTypeSpot) && ctReq.Has(v1.CapacityTypeOnDemand) && hasSpotOffering(results.NewNodeClaims[0]) { results.NewNodeClaims[0].Requirements.Add(scheduling.NewRequirement(v1.CapacityTypeLabelKey, corev1.NodeSelectorOpIn, v1.CapacityTypeSpot)) } @@ -253,6 +253,13 @@ func (c *consolidation) computeConsolidation(ctx context.Context, candidates ... return cmd, nil } +// hasSpotOffering returns true if the replacement has an available Spot offering. +func hasSpotOffering(nodeClaim *pscheduling.NodeClaim) bool { + return lo.ContainsBy(nodeClaim.InstanceTypeOptions, func(it *cloudprovider.InstanceType) bool { + return it.Offerings.Available().Compatible(nodeClaim.Requirements).HasCompatible(cloudprovider.SpotRequirement) + }) +} + // Compute command to execute spot-to-spot consolidation if: // 1. The SpotToSpotConsolidation feature flag is set to true. // 2. For single-node consolidation: diff --git a/pkg/controllers/disruption/consolidation_test.go b/pkg/controllers/disruption/consolidation_test.go index 8808a7e9af..3eb99dfca6 100644 --- a/pkg/controllers/disruption/consolidation_test.go +++ b/pkg/controllers/disruption/consolidation_test.go @@ -4027,6 +4027,99 @@ var _ = Describe("Consolidation", func() { Entry("if the candidate is on-demand node", false), Entry("if the candidate is spot node", true), ) + DescribeTable("can merge 3 nodes into 1 without available Spot offerings", func(requirements []v1.NodeSelectorRequirementWithMinValues, includeUnavailableSpot bool) { + if requirements != nil { + nodePool.Spec.Template.Spec.Requirements = requirements + } + offerings := func(price float64) []cloudprovider.Offering { + result := []cloudprovider.Offering{{ + Available: true, + Requirements: scheduling.NewLabelRequirements(map[string]string{v1.CapacityTypeLabelKey: v1.CapacityTypeOnDemand, corev1.LabelTopologyZone: "test-zone-1"}), + Price: price, + }} + if includeUnavailableSpot { + result = append(result, cloudprovider.Offering{ + Available: false, + Requirements: scheduling.NewLabelRequirements(map[string]string{v1.CapacityTypeLabelKey: v1.CapacityTypeSpot, corev1.LabelTopologyZone: "test-zone-1"}), + Price: price / 2, + }) + } + return result + } + currentInstance := fake.NewInstanceType("current-on-demand-only", + fake.WithOfferings(offerings(1.5)...), + ) + replacementInstance := fake.NewInstanceType("replacement-on-demand-only", + fake.WithOfferings(offerings(1.0)...), + ) + cloudProvider.InstanceTypes = []*cloudprovider.InstanceType{currentInstance, replacementInstance} + ExpectSingletonReconciled(ctx, pricingController) + + nodeClaims, nodes = test.NodeClaimsAndNodes(3, v1.NodeClaim{ + ObjectMeta: metav1.ObjectMeta{ + Labels: map[string]string{ + v1.NodePoolLabelKey: nodePool.Name, + corev1.LabelInstanceTypeStable: currentInstance.Name, + v1.CapacityTypeLabelKey: v1.CapacityTypeOnDemand, + corev1.LabelTopologyZone: "test-zone-1", + }, + }, + Status: v1.NodeClaimStatus{ + Allocatable: map[corev1.ResourceName]resource.Quantity{ + corev1.ResourceCPU: resource.MustParse("32"), + corev1.ResourcePods: resource.MustParse("100"), + }, + }, + }) + for i := range nodeClaims { + nodeClaims[i].StatusConditions().SetTrue(v1.ConditionTypeConsolidatable) + } + + rs := test.ReplicaSet() + ExpectApplied(ctx, env.Client, rs) + pods := test.Pods(3, test.PodOptions{ + ObjectMeta: metav1.ObjectMeta{Labels: labels, + OwnerReferences: []metav1.OwnerReference{ + { + APIVersion: "apps/v1", + Kind: "ReplicaSet", + Name: rs.Name, + UID: rs.UID, + Controller: new(true), + BlockOwnerDeletion: new(true), + }, + }}}) + + ExpectApplied(ctx, env.Client, rs, pods[0], pods[1], pods[2], nodeClaims[0], nodes[0], nodeClaims[1], nodes[1], nodeClaims[2], nodes[2], nodePool) + ExpectMakeNodesInitialized(ctx, env.Client, env.Clock, nodes[0], nodes[1], nodes[2]) + + ExpectManualBinding(ctx, env.Client, pods[0], nodes[0]) + ExpectManualBinding(ctx, env.Client, pods[1], nodes[1]) + ExpectManualBinding(ctx, env.Client, pods[2], nodes[2]) + + ExpectMakeNodesAndNodeClaimsInitializedAndStateUpdated(ctx, env.Client, env.Clock, nodeStateController, nodeClaimStateController, []*corev1.Node{nodes[0], nodes[1], nodes[2]}, []*v1.NodeClaim{nodeClaims[0], nodeClaims[1], nodeClaims[2]}) + ExpectSingletonReconciled(ctx, disruptionController) + + cmds := queue.GetCommands() + Expect(cmds).To(HaveLen(1)) + Expect(cmds[0].Replacements[0].Requirements.Get(v1.CapacityTypeLabelKey).Has(v1.CapacityTypeOnDemand)).To(BeTrue()) + Expect(cmds[0].Replacements[0].Requirements.Get(v1.CapacityTypeLabelKey).Has(v1.CapacityTypeSpot)).To(BeFalse()) + + ExpectMakeNewNodeClaimsReady(ctx, env.Client, env.Clock, cluster, cloudProvider, cmds[0]) + ExpectObjectReconciled(ctx, env.Client, queue, cmds[0].Candidates[0].NodeClaim) + ExpectNodeClaimsCascadeDeletion(ctx, env.Client, nodeClaims[0], nodeClaims[1], nodeClaims[2]) + + Expect(ExpectNodeClaims(ctx, env.Client)).To(HaveLen(1)) + Expect(ExpectNodes(ctx, env.Client)).To(HaveLen(1)) + ExpectNotFound(ctx, env.Client, nodeClaims[0], nodes[0], nodeClaims[1], nodes[1], nodeClaims[2], nodes[2]) + }, + Entry("when the NodePool does not constrain capacity type", nil, false), + Entry("when the NodePool allows Spot and on-demand", []v1.NodeSelectorRequirementWithMinValues{{ + Key: v1.CapacityTypeLabelKey, + Operator: corev1.NodeSelectorOpIn, + Values: []string{v1.CapacityTypeSpot, v1.CapacityTypeOnDemand}, + }}, true), + ) It("can merge 3 nodes into 1 if the candidates have both spot and on-demand", func() { // By default all the 3 nodeClaims are OD. nodeClaims = lo.Ternary(false, spotNodeClaims, nodeClaims)