From 9d688acf8cc6db35e01693512e39c028c82adadc Mon Sep 17 00:00:00 2001 From: jld-adriano <98129582+jld-adriano@users.noreply.github.com> Date: Sun, 21 Jun 2026 21:34:50 +0000 Subject: [PATCH 1/2] Prefer reserved capacity during consolidation Co-authored-by: capy-ai[bot] <230910855+capy-ai[bot]@users.noreply.github.com> --- go.mod | 16 +- go.sum | 28 ++-- pkg/controllers/disruption/consolidation.go | 102 +++++++++++- .../disruption/consolidation_test.go | 155 ++++++++++++++++++ pkg/controllers/disruption/helpers.go | 10 +- pkg/controllers/disruption/validation.go | 54 +++--- 6 files changed, 306 insertions(+), 59 deletions(-) diff --git a/go.mod b/go.mod index 0c221faeb6..7b469905b5 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module sigs.k8s.io/karpenter -go 1.25.7 +go 1.25.11 require ( github.com/Pallinder/go-randomdata v1.2.0 @@ -22,7 +22,7 @@ require ( github.com/samber/lo v1.52.0 go.uber.org/multierr v1.11.0 go.uber.org/zap v1.27.1 - golang.org/x/text v0.33.0 + golang.org/x/text v0.37.0 golang.org/x/time v0.14.0 k8s.io/api v0.35.0 k8s.io/apiextensions-apiserver v0.35.0 @@ -81,13 +81,13 @@ require ( github.com/x448/float16 v0.8.4 // indirect go.yaml.in/yaml/v2 v2.4.3 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect - golang.org/x/mod v0.32.0 // indirect - golang.org/x/net v0.49.0 // indirect + golang.org/x/mod v0.35.0 // indirect + golang.org/x/net v0.55.0 // indirect golang.org/x/oauth2 v0.34.0 // indirect - golang.org/x/sync v0.19.0 // indirect - golang.org/x/sys v0.40.0 // indirect - golang.org/x/term v0.39.0 // indirect - golang.org/x/tools v0.41.0 // indirect + golang.org/x/sync v0.20.0 // indirect + golang.org/x/sys v0.45.0 // indirect + golang.org/x/term v0.43.0 // indirect + golang.org/x/tools v0.44.0 // indirect gomodules.xyz/jsonpatch/v2 v2.5.0 // indirect google.golang.org/protobuf v1.36.11 // indirect gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect diff --git a/go.sum b/go.sum index 7b53fcca2e..7448a95fca 100644 --- a/go.sum +++ b/go.sum @@ -182,24 +182,24 @@ go.yaml.in/yaml/v2 v2.4.3 h1:6gvOSjQoTB3vt1l+CU+tSyi/HOjfOjRLJ4YwYZGwRO0= go.yaml.in/yaml/v2 v2.4.3/go.mod h1:zSxWcmIDjOzPXpjlTTbAsKokqkDNAVtZO0WOMiT90s8= go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc= go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= -golang.org/x/mod v0.32.0 h1:9F4d3PHLljb6x//jOyokMv3eX+YDeepZSEo3mFJy93c= -golang.org/x/mod v0.32.0/go.mod h1:SgipZ/3h2Ci89DlEtEXWUk/HteuRin+HHhN+WbNhguU= -golang.org/x/net v0.49.0 h1:eeHFmOGUTtaaPSGNmjBKpbng9MulQsJURQUAfUwY++o= -golang.org/x/net v0.49.0/go.mod h1:/ysNB2EvaqvesRkuLAyjI1ycPZlQHM3q01F02UY/MV8= +golang.org/x/mod v0.35.0 h1:Ww1D637e6Pg+Zb2KrWfHQUnH2dQRLBQyAtpr/haaJeM= +golang.org/x/mod v0.35.0/go.mod h1:+GwiRhIInF8wPm+4AoT6L0FA1QWAad3OMdTRx4tFYlU= +golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8= +golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww= golang.org/x/oauth2 v0.34.0 h1:hqK/t4AKgbqWkdkcAeI8XLmbK+4m4G5YeQRrmiotGlw= golang.org/x/oauth2 v0.34.0/go.mod h1:lzm5WQJQwKZ3nwavOZ3IS5Aulzxi68dUSgRHujetwEA= -golang.org/x/sync v0.19.0 h1:vV+1eWNmZ5geRlYjzm2adRgW2/mcpevXNg50YZtPCE4= -golang.org/x/sync v0.19.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= -golang.org/x/sys v0.40.0 h1:DBZZqJ2Rkml6QMQsZywtnjnnGvHza6BTfYFWY9kjEWQ= -golang.org/x/sys v0.40.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= -golang.org/x/term v0.39.0 h1:RclSuaJf32jOqZz74CkPA9qFuVTX7vhLlpfj/IGWlqY= -golang.org/x/term v0.39.0/go.mod h1:yxzUCTP/U+FzoxfdKmLaA0RV1WgE0VY7hXBwKtY/4ww= -golang.org/x/text v0.33.0 h1:B3njUFyqtHDUI5jMn1YIr5B0IE2U0qck04r6d4KPAxE= -golang.org/x/text v0.33.0/go.mod h1:LuMebE6+rBincTi9+xWTY8TztLzKHc/9C1uBCG27+q8= +golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= +golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY= +golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/term v0.43.0 h1:S4RLU2sB31O/NCl+zFN9Aru9A/Cq2aqKpTZJ6B+DwT4= +golang.org/x/term v0.43.0/go.mod h1:lrhlHNdQJHO+1qVYiHfFKVuVioJIheAc3fBSMFYEIsk= +golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc= +golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38= golang.org/x/time v0.14.0 h1:MRx4UaLrDotUKUdCIqzPC48t1Y9hANFKIRpNx+Te8PI= golang.org/x/time v0.14.0/go.mod h1:eL/Oa2bBBK0TkX57Fyni+NgnyQQN4LitPmob2Hjnqw4= -golang.org/x/tools v0.41.0 h1:a9b8iMweWG+S0OBnlU36rzLp20z1Rp10w+IY2czHTQc= -golang.org/x/tools v0.41.0/go.mod h1:XSY6eDqxVNiYgezAVqqCeihT4j1U2CCsqvH3WhQpnlg= +golang.org/x/tools v0.44.0 h1:UP4ajHPIcuMjT1GqzDWRlalUEoY+uzoZKnhOjbIPD2c= +golang.org/x/tools v0.44.0/go.mod h1:KA0AfVErSdxRZIsOVipbv3rQhVXTnlU6UhKxHd1seDI= gomodules.xyz/jsonpatch/v2 v2.5.0 h1:JELs8RLM12qJGXU4u/TO3V25KW8GreMKl9pdkk14RM0= gomodules.xyz/jsonpatch/v2 v2.5.0/go.mod h1:AH3dM2RI6uoBZxn3LVrfvJ3E0/9dG4cSrbuBJT4moAY= google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= diff --git a/pkg/controllers/disruption/consolidation.go b/pkg/controllers/disruption/consolidation.go index ca7eb68e70..0a7cfcaadf 100644 --- a/pkg/controllers/disruption/consolidation.go +++ b/pkg/controllers/disruption/consolidation.go @@ -20,6 +20,7 @@ import ( "context" "errors" "fmt" + "math" "sort" "time" @@ -136,7 +137,7 @@ func (c *consolidation) sortCandidates(candidates []*Candidate) []*Candidate { func (c *consolidation) computeConsolidation(ctx context.Context, candidates ...*Candidate) (Command, error) { var err error // Run scheduling simulation to compute consolidation option - results, err := SimulateScheduling(ctx, c.kubeClient, c.cluster, c.provisioner, candidates...) + results, err := simulateScheduling(ctx, c.kubeClient, c.cluster, c.provisioner, []pscheduling.Options{pscheduling.DisableReservedCapacityFallback}, candidates...) if err != nil { // if a candidate node is now deleting, just retry if errors.Is(err, errCandidateDeleting) { @@ -154,26 +155,37 @@ func (c *consolidation) computeConsolidation(ctx context.Context, candidates ... return Command{}, nil } + // get the current node price based on the offering + // fallback if we can't find the specific zonal pricing data + candidatePrice := getCandidatePrices(candidates) + // were we able to schedule all the pods on the inflight candidates? if len(results.NewNodeClaims) == 0 { + if hasNonEmptyReservedCandidatesMovingToNonReservedCapacity(candidates, results) { + if len(candidates) == 1 { + c.recorder.Publish(disruptionevents.Unconsolidatable(candidates[0].Node, candidates[0].NodeClaim, "Can't consolidate non-empty reserved node onto non-reserved capacity")...) + } + return Command{}, nil + } return Command{ Candidates: candidates, Results: results, }, nil } - // we're not going to turn a single node into multiple candidates if len(results.NewNodeClaims) != 1 { + if len(candidates) == 1 { + cmd := c.computeReservedMultiNodeReplacement(candidates, results, candidatePrice) + if cmd.Decision() != NoOpDecision { + return cmd, nil + } + } if len(candidates) == 1 { c.recorder.Publish(disruptionevents.Unconsolidatable(candidates[0].Node, candidates[0].NodeClaim, fmt.Sprintf("Can't remove without creating %d candidates", len(results.NewNodeClaims)))...) } return Command{}, nil } - // get the current node price based on the offering - // fallback if we can't find the specific zonal pricing data - candidatePrice := getCandidatePrices(candidates) - allExistingAreSpot := true for _, cn := range candidates { if cn.capacityType != v1.CapacityTypeSpot { @@ -228,6 +240,84 @@ func (c *consolidation) computeConsolidation(ctx context.Context, candidates ... return cmd, nil } +func (c *consolidation) computeReservedMultiNodeReplacement(candidates []*Candidate, results pscheduling.Results, candidatePrice float64) Command { + if !allNodeClaimsLaunchReservedOnly(results.NewNodeClaims) { + return Command{} + } + var err error + for i := range results.NewNodeClaims { + results.NewNodeClaims[i].InstanceTypeOptions = results.NewNodeClaims[i].InstanceTypeOptions.OrderByPrice(results.NewNodeClaims[i].Requirements) + results.NewNodeClaims[i], err = results.NewNodeClaims[i].RemoveInstanceTypeOptionsByPriceAndMinValues(results.NewNodeClaims[i].Requirements, candidatePrice) + if err != nil { + c.recorder.Publish(disruptionevents.Unconsolidatable(candidates[0].Node, candidates[0].NodeClaim, fmt.Sprintf("Filtering by price: %v", err))...) + return Command{} + } + if len(results.NewNodeClaims[i].InstanceTypeOptions) == 0 { + c.recorder.Publish(disruptionevents.Unconsolidatable(candidates[0].Node, candidates[0].NodeClaim, "Can't replace with a cheaper node")...) + return Command{} + } + } + if price, ok := replacementNodeClaimsPrice(results.NewNodeClaims); !ok || price >= candidatePrice { + c.recorder.Publish(disruptionevents.Unconsolidatable(candidates[0].Node, candidates[0].NodeClaim, "Can't replace with cheaper reserved nodes")...) + return Command{} + } + cmd := Command{ + Candidates: candidates, + Replacements: replacementsFromNodeClaims(results.NewNodeClaims...), + Results: results, + } + cmd.EmitCandidateEvents(c.recorder) + return cmd +} + +func hasNonEmptyReservedCandidatesMovingToNonReservedCapacity(candidates []*Candidate, results pscheduling.Results) bool { + for _, candidate := range candidates { + if candidate.capacityType != v1.CapacityTypeReserved || len(candidate.reschedulablePods) == 0 { + continue + } + for _, p := range candidate.reschedulablePods { + existingNode, ok := lo.Find(results.ExistingNodes, func(n *pscheduling.ExistingNode) bool { + return lo.ContainsBy(n.Pods, func(scheduledPod *corev1.Pod) bool { + return podsEqual(p, scheduledPod) + }) + }) + if !ok || existingNode.Labels()[v1.CapacityTypeLabelKey] != v1.CapacityTypeReserved { + return true + } + } + } + return false +} + +func allNodeClaimsLaunchReservedOnly(nodeClaims []*pscheduling.NodeClaim) bool { + return len(nodeClaims) > 0 && lo.EveryBy(nodeClaims, func(nc *pscheduling.NodeClaim) bool { + capacityTypeRequirement := nc.Requirements.Get(v1.CapacityTypeLabelKey) + return capacityTypeRequirement.Len() == 1 && capacityTypeRequirement.Has(v1.CapacityTypeReserved) + }) +} + +func replacementNodeClaimsPrice(nodeClaims []*pscheduling.NodeClaim) (float64, bool) { + var price float64 + for _, nodeClaim := range nodeClaims { + if len(nodeClaim.InstanceTypeOptions) == 0 { + return 0.0, false + } + launchPrice := nodeClaim.InstanceTypeOptions[0].Offerings.Available().WorstLaunchPrice(nodeClaim.Requirements) + if launchPrice == math.MaxFloat64 { + return 0.0, false + } + price += launchPrice + } + return price, true +} + +func podsEqual(a, b *corev1.Pod) bool { + if a.UID != "" && b.UID != "" { + return a.UID == b.UID + } + return client.ObjectKeyFromObject(a) == client.ObjectKeyFromObject(b) +} + // 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 c391a79e63..e86c3416a7 100644 --- a/pkg/controllers/disruption/consolidation_test.go +++ b/pkg/controllers/disruption/consolidation_test.go @@ -4616,6 +4616,161 @@ var _ = Describe("Consolidation", func() { Entry("from on-demand", v1.CapacityTypeOnDemand), Entry("from spot", v1.CapacityTypeSpot), ) + It("does not delete a non-empty reserved node onto existing non-reserved capacity", func() { + rs := test.ReplicaSet() + ExpectApplied(ctx, env.Client, rs) + Expect(env.Client.Get(ctx, client.ObjectKeyFromObject(rs), rs)).To(Succeed()) + pod := test.Pod(test.PodOptions{ + ObjectMeta: metav1.ObjectMeta{Labels: labels, + OwnerReferences: []metav1.OwnerReference{ + { + APIVersion: "apps/v1", + Kind: "ReplicaSet", + Name: rs.Name, + UID: rs.UID, + Controller: lo.ToPtr(true), + BlockOwnerDeletion: lo.ToPtr(true), + }, + }}, + }) + ExpectApplied(ctx, env.Client, rs, pod, reservedNode, reservedNodeClaim, node, nodeClaim, nodePool) + ExpectManualBinding(ctx, env.Client, pod, reservedNode) + ExpectMakeNodesAndNodeClaimsInitializedAndStateUpdated(ctx, env.Client, nodeStateController, nodeClaimStateController, []*corev1.Node{reservedNode, node}, []*v1.NodeClaim{reservedNodeClaim, nodeClaim}) + + c := disruption.MakeConsolidation(fakeClock, cluster, env.Client, prov, cloudProvider, recorder, queue) + singleNodeConsolidation := disruption.NewSingleNodeConsolidation(c, disruption.WithValidator(NopValidator{})) + budgets, err := disruption.BuildDisruptionBudgetMapping(ctx, cluster, fakeClock, env.Client, cloudProvider, recorder, singleNodeConsolidation.Reason()) + Expect(err).To(Succeed()) + candidates, err := disruption.GetCandidates(ctx, cluster, env.Client, recorder, fakeClock, cloudProvider, singleNodeConsolidation.ShouldDisrupt, singleNodeConsolidation.Class(), queue) + Expect(err).To(Succeed()) + reservedCandidate, ok := lo.Find(candidates, func(candidate *disruption.Candidate) bool { + return candidate.NodeClaim.Name == reservedNodeClaim.Name + }) + Expect(ok).To(BeTrue()) + + cmds, err := singleNodeConsolidation.ComputeCommands(ctx, budgets, reservedCandidate) + Expect(err).To(Succeed()) + Expect(cmds).To(BeEmpty()) + }) + It("can replace one on-demand node with multiple reserved nodeclaims", func() { + reservationID := "r-small-reserved" + largeOnDemand := fake.NewInstanceType(fake.InstanceTypeOptions{ + Name: "large-on-demand", + Resources: corev1.ResourceList{ + corev1.ResourceCPU: resource.MustParse("8"), + corev1.ResourceMemory: resource.MustParse("8Gi"), + corev1.ResourcePods: resource.MustParse("10"), + }, + Offerings: cloudprovider.Offerings{{ + Available: true, + Price: 100.0, + Requirements: scheduling.NewLabelRequirements(map[string]string{ + v1.CapacityTypeLabelKey: v1.CapacityTypeOnDemand, + corev1.LabelTopologyZone: "test-zone-1", + }), + }}, + }) + smallReserved := fake.NewInstanceType(fake.InstanceTypeOptions{ + Name: "small-reserved", + Resources: corev1.ResourceList{ + corev1.ResourceCPU: resource.MustParse("4"), + corev1.ResourceMemory: resource.MustParse("4Gi"), + corev1.ResourcePods: resource.MustParse("10"), + }, + Offerings: cloudprovider.Offerings{{ + Available: true, + Price: 0.00001, + ReservationCapacity: 2, + Requirements: scheduling.NewLabelRequirements(map[string]string{ + v1.CapacityTypeLabelKey: v1.CapacityTypeReserved, + corev1.LabelTopologyZone: "test-zone-1", + v1alpha1.LabelReservationID: reservationID, + }), + }}, + }) + cloudProvider.InstanceTypes = []*cloudprovider.InstanceType{largeOnDemand, smallReserved} + + nodePool.Spec.Template.Spec.Requirements = []v1.NodeSelectorRequirementWithMinValues{ + { + Key: v1.CapacityTypeLabelKey, + Operator: corev1.NodeSelectorOpIn, + Values: []string{v1.CapacityTypeOnDemand, v1.CapacityTypeReserved}, + }, + } + onDemandNodeClaim, onDemandNode := test.NodeClaimAndNode(v1.NodeClaim{ + ObjectMeta: metav1.ObjectMeta{ + Labels: map[string]string{ + v1.NodePoolLabelKey: nodePool.Name, + corev1.LabelInstanceTypeStable: largeOnDemand.Name, + v1.CapacityTypeLabelKey: v1.CapacityTypeOnDemand, + corev1.LabelTopologyZone: "test-zone-1", + }, + }, + Status: v1.NodeClaimStatus{ + Allocatable: map[corev1.ResourceName]resource.Quantity{ + corev1.ResourceCPU: resource.MustParse("8"), + corev1.ResourceMemory: resource.MustParse("8Gi"), + corev1.ResourcePods: resource.MustParse("10"), + }, + }, + }) + onDemandNodeClaim.StatusConditions().SetTrue(v1.ConditionTypeConsolidatable) + rs := test.ReplicaSet() + ExpectApplied(ctx, env.Client, rs) + Expect(env.Client.Get(ctx, client.ObjectKeyFromObject(rs), rs)).To(Succeed()) + pods := test.Pods(2, test.PodOptions{ + ResourceRequirements: corev1.ResourceRequirements{ + Requests: corev1.ResourceList{ + corev1.ResourceCPU: resource.MustParse("3"), + corev1.ResourceMemory: resource.MustParse("100Mi"), + }, + }, + ObjectMeta: metav1.ObjectMeta{Labels: labels, + OwnerReferences: []metav1.OwnerReference{ + { + APIVersion: "apps/v1", + Kind: "ReplicaSet", + Name: rs.Name, + UID: rs.UID, + Controller: lo.ToPtr(true), + BlockOwnerDeletion: lo.ToPtr(true), + }, + }}, + }) + ExpectApplied(ctx, env.Client, rs, pods[0], pods[1], onDemandNode, onDemandNodeClaim, nodePool) + ExpectManualBinding(ctx, env.Client, pods[0], onDemandNode) + ExpectManualBinding(ctx, env.Client, pods[1], onDemandNode) + ExpectMakeNodesAndNodeClaimsInitializedAndStateUpdated(ctx, env.Client, nodeStateController, nodeClaimStateController, []*corev1.Node{onDemandNode}, []*v1.NodeClaim{onDemandNodeClaim}) + + c := disruption.MakeConsolidation(fakeClock, cluster, env.Client, prov, cloudProvider, recorder, queue) + singleNodeConsolidation := disruption.NewSingleNodeConsolidation(c, disruption.WithValidator(NopValidator{})) + budgets, err := disruption.BuildDisruptionBudgetMapping(ctx, cluster, fakeClock, env.Client, cloudProvider, recorder, singleNodeConsolidation.Reason()) + Expect(err).To(Succeed()) + candidates, err := disruption.GetCandidates(ctx, cluster, env.Client, recorder, fakeClock, cloudProvider, singleNodeConsolidation.ShouldDisrupt, singleNodeConsolidation.Class(), queue) + Expect(err).To(Succeed()) + onDemandCandidate, ok := lo.Find(candidates, func(candidate *disruption.Candidate) bool { + return candidate.NodeClaim.Name == onDemandNodeClaim.Name + }) + Expect(ok).To(BeTrue()) + + cmds, err := singleNodeConsolidation.ComputeCommands(ctx, budgets, onDemandCandidate) + Expect(err).To(Succeed()) + Expect(cmds).To(HaveLen(1)) + Expect(cmds[0].Replacements).To(HaveLen(2)) + for _, replacement := range cmds[0].Replacements { + Expect(replacement.Requirements.Get(v1.CapacityTypeLabelKey).Values()).To(ConsistOf(v1.CapacityTypeReserved)) + Expect(replacement.Requirements.Get(cloudprovider.ReservationIDLabel).Values()).To(ConsistOf(reservationID)) + } + + validator := disruption.NewSingleConsolidationValidator(c) + _, err = validator.Validate(ctx, cmds[0], 0) + Expect(err).ToNot(HaveOccurred()) + + cmds[0].Replacements[0].Requirements[v1.CapacityTypeLabelKey] = scheduling.NewRequirement(v1.CapacityTypeLabelKey, corev1.NodeSelectorOpIn, v1.CapacityTypeOnDemand) + _, err = validator.Validate(ctx, cmds[0], 0) + Expect(err).To(HaveOccurred()) + Expect(disruption.IsValidationError(err)).To(BeTrue()) + }) }) Context("Preferences", func() { It("should consolidate a node through deletion when ignoring preferences", func() { diff --git a/pkg/controllers/disruption/helpers.go b/pkg/controllers/disruption/helpers.go index 9fa2891b0b..0f316e9028 100644 --- a/pkg/controllers/disruption/helpers.go +++ b/pkg/controllers/disruption/helpers.go @@ -47,9 +47,15 @@ import ( var errCandidateDeleting = fmt.Errorf("candidate is deleting") -//nolint:gocyclo func SimulateScheduling(ctx context.Context, kubeClient client.Client, cluster *state.Cluster, provisioner *provisioning.Provisioner, candidates ...*Candidate, +) (scheduling.Results, error) { + return simulateScheduling(ctx, kubeClient, cluster, provisioner, nil, candidates...) +} + +//nolint:gocyclo +func simulateScheduling(ctx context.Context, kubeClient client.Client, cluster *state.Cluster, provisioner *provisioning.Provisioner, + schedulerOpts []scheduling.Options, candidates ...*Candidate, ) (scheduling.Results, error) { candidateNames := sets.NewString(lo.Map(candidates, func(t *Candidate, i int) string { return t.Name() })...) nodes := cluster.DeepCopyNodes() @@ -94,7 +100,7 @@ func SimulateScheduling(ctx context.Context, kubeClient client.Client, cluster * } pods = append(pods, deletingNodePods...) - var opts []scheduling.Options + opts := append([]scheduling.Options{}, schedulerOpts...) if options.FromContext(ctx).PreferencePolicy == options.PreferencePolicyIgnore { opts = append(opts, scheduling.IgnorePreferences) } diff --git a/pkg/controllers/disruption/validation.go b/pkg/controllers/disruption/validation.go index 85307e8e24..43dee0763c 100644 --- a/pkg/controllers/disruption/validation.go +++ b/pkg/controllers/disruption/validation.go @@ -29,6 +29,7 @@ import ( v1 "sigs.k8s.io/karpenter/pkg/apis/v1" "sigs.k8s.io/karpenter/pkg/cloudprovider" "sigs.k8s.io/karpenter/pkg/controllers/provisioning" + pscheduling "sigs.k8s.io/karpenter/pkg/controllers/provisioning/scheduling" "sigs.k8s.io/karpenter/pkg/controllers/state" "sigs.k8s.io/karpenter/pkg/events" ) @@ -298,7 +299,7 @@ func (v *validation) validateCommand(ctx context.Context, cmd Command, candidate if len(candidates) == 0 { return NewValidationError(fmt.Errorf("no candidates")) } - results, err := SimulateScheduling(ctx, v.kubeClient, v.cluster, v.provisioner, candidates...) + results, err := simulateScheduling(ctx, v.kubeClient, v.cluster, v.provisioner, []pscheduling.Options{pscheduling.DisableReservedCapacityFallback}, candidates...) if err != nil { return fmt.Errorf("simluating scheduling, %w", err) } @@ -307,11 +308,6 @@ func (v *validation) validateCommand(ctx context.Context, cmd Command, candidate } // We want to ensure that the re-simulated scheduling using the current cluster state produces the same result. - // There are three possible options for the number of new candidates that we need to handle: - // len(NewNodeClaims) == 0, as long as we weren't expecting a new node, this is valid - // len(NewNodeClaims) > 1, something in the cluster changed so that the candidates we were going to delete can no longer - // be deleted without producing more than one node - // len(NewNodeClaims) == 1, as long as the noe looks like what we were expecting, this is valid if len(results.NewNodeClaims) == 0 { if len(cmd.Replacements) == 0 { // scheduling produced zero new NodeClaims and we weren't expecting any, so this is valid. @@ -322,36 +318,36 @@ func (v *validation) validateCommand(ctx context.Context, cmd Command, candidate return NewSchedulingValidationError(fmt.Errorf("scheduling simulation produced new results")) } - // we need more than one replacement node which is never valid currently (all of our node replacement is m->1, never m->n) - if len(results.NewNodeClaims) > 1 { + if len(results.NewNodeClaims) != len(cmd.Replacements) { return NewSchedulingValidationError(fmt.Errorf("scheduling simulation produced new results")) } - // we now know that scheduling simulation wants to create one new node - if len(cmd.Replacements) == 0 { - // but we weren't expecting any new NodeClaims, so this is invalid + if !replacementsMatchNodeClaims(cmd.Replacements, results.NewNodeClaims) { return NewSchedulingValidationError(fmt.Errorf("scheduling simulation produced new results")) } + return nil +} - // We know that the scheduling simulation wants to create a new node and that the command we are verifying wants - // to create a new node. The scheduling simulation doesn't apply any filtering to instance types, so it may include - // instance types that we don't want to launch which were filtered out when the lifecycleCommand was created. To - // check if our lifecycleCommand is valid, we just want to ensure that the list of instance types we are considering - // creating are a subset of what scheduling says we should create. We check for a subset since the scheduling - // simulation here does no price filtering, so it will include more expensive types. - // - // This is necessary since consolidation only wants cheaper NodeClaims. Suppose consolidation determined we should delete - // a 4xlarge and replace it with a 2xlarge. If things have changed and the scheduling simulation we just performed - // now says that we need to launch a 4xlarge. It's still launching the correct number of NodeClaims, but it's just - // as expensive or possibly more so we shouldn't validate. - if !instanceTypesAreSubset(cmd.Replacements[0].InstanceTypeOptions, results.NewNodeClaims[0].InstanceTypeOptions) { - return NewSchedulingValidationError(fmt.Errorf("scheduling simulation produced new results")) +func replacementsMatchNodeClaims(replacements []*Replacement, nodeClaims []*pscheduling.NodeClaim) bool { + unmatchedNodeClaims := lo.SliceToMap(nodeClaims, func(nc *pscheduling.NodeClaim) (*pscheduling.NodeClaim, struct{}) { + return nc, struct{}{} + }) + for _, replacement := range replacements { + match, ok := lo.Find(nodeClaims, func(nodeClaim *pscheduling.NodeClaim) bool { + if _, ok := unmatchedNodeClaims[nodeClaim]; !ok { + return false + } + if !instanceTypesAreSubset(replacement.InstanceTypeOptions, nodeClaim.InstanceTypeOptions) { + return false + } + return nodeClaim.Requirements.Compatible(replacement.Requirements) == nil + }) + if !ok { + return false + } + delete(unmatchedNodeClaims, match) } - - // Now we know: - // - current scheduling simulation says to create a new node with types T = {T_0, T_1, ..., T_n} - // - our lifecycle command says to create a node with types {U_0, U_1, ..., U_n} where U is a subset of T - return nil + return true } // getValidationFailureReason categorizes validation errors into specific failure types From b01b405f8da7712bb1a8baf2c2423ec5fde0014e Mon Sep 17 00:00:00 2001 From: jld-adriano <98129582+jld-adriano@users.noreply.github.com> Date: Sun, 21 Jun 2026 22:30:01 +0000 Subject: [PATCH 2/2] fix: preserve spot validation while matching reserved replacements Co-authored-by: capy-ai[bot] <230910855+capy-ai[bot]@users.noreply.github.com> --- pkg/controllers/disruption/validation.go | 27 ++++++++++++++++++++++-- 1 file changed, 25 insertions(+), 2 deletions(-) diff --git a/pkg/controllers/disruption/validation.go b/pkg/controllers/disruption/validation.go index 43dee0763c..08922100d2 100644 --- a/pkg/controllers/disruption/validation.go +++ b/pkg/controllers/disruption/validation.go @@ -32,6 +32,7 @@ import ( pscheduling "sigs.k8s.io/karpenter/pkg/controllers/provisioning/scheduling" "sigs.k8s.io/karpenter/pkg/controllers/state" "sigs.k8s.io/karpenter/pkg/events" + "sigs.k8s.io/karpenter/pkg/scheduling" ) type ValidationError struct { @@ -337,10 +338,10 @@ func replacementsMatchNodeClaims(replacements []*Replacement, nodeClaims []*psch if _, ok := unmatchedNodeClaims[nodeClaim]; !ok { return false } - if !instanceTypesAreSubset(replacement.InstanceTypeOptions, nodeClaim.InstanceTypeOptions) { + if !replacementMatchesNodeClaim(replacement, nodeClaim) { return false } - return nodeClaim.Requirements.Compatible(replacement.Requirements) == nil + return true }) if !ok { return false @@ -350,6 +351,28 @@ func replacementsMatchNodeClaims(replacements []*Replacement, nodeClaims []*psch return true } +func replacementMatchesNodeClaim(replacement *Replacement, nodeClaim *pscheduling.NodeClaim) bool { + if !instanceTypesAreSubset(replacement.InstanceTypeOptions, nodeClaim.InstanceTypeOptions) { + return false + } + return reservedRequirementsCompatible(replacement.Requirements, nodeClaim.Requirements) +} + +func reservedRequirementsCompatible(replacementRequirements, nodeClaimRequirements scheduling.Requirements) bool { + reservedRequirements := scheduling.NewRequirements() + if replacementRequirements.Get(v1.CapacityTypeLabelKey).Has(v1.CapacityTypeReserved) || lo.ContainsBy(cloudprovider.ReservedCapacityLabels.UnsortedList(), replacementRequirements.Has) { + if replacementRequirements.Has(v1.CapacityTypeLabelKey) { + reservedRequirements.Add(replacementRequirements.Get(v1.CapacityTypeLabelKey)) + } + for label := range cloudprovider.ReservedCapacityLabels { + if replacementRequirements.Has(label) { + reservedRequirements.Add(replacementRequirements.Get(label)) + } + } + } + return len(reservedRequirements) == 0 || nodeClaimRequirements.Compatible(reservedRequirements) == nil +} + // getValidationFailureReason categorizes validation errors into specific failure types func getValidationFailureReason(err error) string { if err == nil {