Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 8 additions & 1 deletion pkg/controllers/disruption/consolidation.go
Original file line number Diff line number Diff line change
Expand Up @@ -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]) {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should we also update the block comment above with this change? something like "if Spot is unavailable at consolidation time, skip the Spot requirement and let the replacement launch as OD. The pricing filter picked the instance based on available offerings, so falling back to OD does not regress cost" ? WDYT?

results.NewNodeClaims[0].Requirements.Add(scheduling.NewRequirement(v1.CapacityTypeLabelKey, corev1.NodeSelectorOpIn, v1.CapacityTypeSpot))
}

Expand All @@ -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:
Expand Down
93 changes: 93 additions & 0 deletions pkg/controllers/disruption/consolidation_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
Loading