fix: skip adding spot requirement during consolidation when no spot offerings are available

Author: bwagner5

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:

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)