diff --git a/pkg/controllers/disruption/helpers.go b/pkg/controllers/disruption/helpers.go index 12d17e2ed4..0ed3b2e14c 100644 --- a/pkg/controllers/disruption/helpers.go +++ b/pkg/controllers/disruption/helpers.go @@ -79,6 +79,7 @@ func SimulateScheduling(ctx context.Context, kubeClient client.Client, cluster * if err != nil { return scheduling.Results{}, fmt.Errorf("determining pending pods, %w", err) } + pods = excludeUnprovisionablePods(ctx, cluster, clk, pods) // Don't provision capacity for pods which will not get evicted due to fully blocking PDBs. // Since Karpenter doesn't know when these pods will be successfully evicted, spinning up capacity until diff --git a/pkg/controllers/disruption/metrics.go b/pkg/controllers/disruption/metrics.go index 9d800ad46b..5cc5aa9e22 100644 --- a/pkg/controllers/disruption/metrics.go +++ b/pkg/controllers/disruption/metrics.go @@ -59,6 +59,7 @@ const ( policyLabel = "policy" outcomeLabel = "outcome" reasonLabel = "reason" + dispositionLabel = "disposition" replacementCountLabel = "replacement_count" capacityTypeTransitionLabel = "capacity_type_transition" instanceTypeLabel = "instance_type" @@ -422,6 +423,20 @@ var ( }, []string{ConsolidationTypeLabel}, ) + // SimulationPendingPods splits the pending backlog a disruption simulation received into the pods + // it scheduled and the pods it left out because the provisioner's latest pass found every NodePool + // incompatible with them (see unprovisionablepods.go). The excluded count is the population that + // was inflating every candidate's simulation without being able to affect its verdict. + SimulationPendingPods = opmetrics.NewPrometheusGauge( + crmetrics.Registry, + prometheus.GaugeOpts{ + Namespace: metrics.Namespace, + Subsystem: voluntaryDisruptionSubsystem, + Name: "simulation_pending_pods", + Help: "Number of pending pods the latest disruption scheduling simulation received from the provisioning backlog, by disposition: simulated pods entered the solve; excluded_unprovisionable pods were left out because the provisioner's most recent simulation found every NodePool incompatible with them, independent of cluster state, and that verdict is younger than DISRUPTION_UNPROVISIONABLE_POD_TTL.", + }, + []string{dispositionLabel}, + ) ConsolidationPassOutcomesTotal = opmetrics.NewPrometheusCounter( crmetrics.Registry, prometheus.CounterOpts{ diff --git a/pkg/controllers/disruption/unprovisionablepods.go b/pkg/controllers/disruption/unprovisionablepods.go new file mode 100644 index 0000000000..9620c7d593 --- /dev/null +++ b/pkg/controllers/disruption/unprovisionablepods.go @@ -0,0 +1,134 @@ +/* +Copyright The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package disruption + +import ( + "context" + "sync" + "time" + + "github.com/samber/lo" + corev1 "k8s.io/api/core/v1" + "k8s.io/klog/v2" + "k8s.io/utils/clock" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/log" + + "sigs.k8s.io/karpenter/pkg/controllers/state" + "sigs.k8s.io/karpenter/pkg/operator/options" +) + +// A disruption simulation schedules the whole pending backlog alongside the candidate's pods, because +// the backlog competes for the same capacity. A pending pod that every NodePool rejects on the pod and +// the NodePool alone - it tolerates none of the NodePool's taints, its requirements contradict the +// NodePool's, or no instance type with an available offering meets them (a capacity type whose every +// offering is exhausted, say) - opens no NodeClaim in the provisioner's simulation or in any +// disruption simulation, since those differ only in which nodes exist and what runs on them, and +// never shares or sizes a replacement. Simulating it is pure cost, paid once per candidate, and a +// backlog of thousands of such pods turns each candidate's budget into a timeout. +// +// What such a pod can still do is bind to an existing node another pod vacates, which kube-scheduler +// decides after the fact whether or not the simulation modeled it. Leaving the pod out lets a +// simulation hand that room to a candidate's pods instead; if the pending pod takes it first, the +// candidate's pods return to the backlog and provisioning launches for them - the outcome any pod that +// turns pending between simulation and execution already produces. +// +// The provisioner already computes the verdict: every provisioning pass places each pending pod or +// records an error for it, and only an error of that pass-invariant kind (see +// scheduling.IsIncompatibleWithAllNodePools) becomes an unprovisionable verdict in the cluster state. +// Errors that hinge on the pass - a NodePool at its limit, which a candidate's removal can lift; a +// reserved offering the strict provisioning mode deferred but the fallback mode of a disruption +// simulation may grant; topology, DRA or minValues outcomes - record no verdict, so those pods stay +// in every simulation. A disruption simulation excludes the pods whose verdict is younger than +// DISRUPTION_UNPROVISIONABLE_POD_TTL. Nothing here re-derives schedulability, so the exclusion +// follows the provisioner exactly and is refreshed as often as it runs. The TTL bounds what happens +// when it does not, and when instance type availability moves under a verdict: a verdict older than +// the TTL is ignored and the pod is simulated again. +// +// Normal provisioning is untouched: it reads the backlog directly, never through this filter, so an +// excluded pod is retried by every provisioning pass and re-enters simulations the moment one places it. + +const ( + simulationPodsDispositionSimulated = "simulated" + simulationPodsDispositionExcluded = "excluded_unprovisionable" + + // unprovisionablePodsLogInterval spaces the Info-level log of excluded pods. The gauge carries + // the continuous signal; the log is a periodic, sampled confirmation of which pods it counts. + unprovisionablePodsLogInterval = time.Minute + unprovisionablePodsLogSample = 5 +) + +// partitionUnprovisionablePods splits the pending backlog into the pods a disruption simulation +// should schedule and the pods whose latest provisioning verdict, younger than ttl, found every +// NodePool incompatible with them. A ttl of zero or less excludes nothing. Both returned slices are +// freshly allocated; pods is not modified. +func partitionUnprovisionablePods(cluster *state.Cluster, clk clock.Clock, ttl time.Duration, pods []*corev1.Pod) (simulated, excluded []*corev1.Pod) { + if ttl <= 0 { + return append([]*corev1.Pod(nil), pods...), nil + } + now := clk.Now() + return lo.FilterReject(pods, func(p *corev1.Pod, _ int) bool { + verdict := cluster.PodUnprovisionableTime(client.ObjectKeyFromObject(p)) + return verdict.IsZero() || now.Sub(verdict) >= ttl + }) +} + +// unprovisionablePodsLog rate-limits the Info-level report of excluded pods across every +// simulation of every disruption method, which otherwise run several times a minute. +type unprovisionablePodsLog struct { + mu sync.Mutex + last time.Time +} + +var excludedPodsLog unprovisionablePodsLog + +// shouldLog reports whether at least unprovisionablePodsLogInterval has passed since the last +// accepted log and, if so, records now as the last accepted time. +func (l *unprovisionablePodsLog) shouldLog(now time.Time) bool { + l.mu.Lock() + defer l.mu.Unlock() + if !l.last.IsZero() && now.Sub(l.last) < unprovisionablePodsLogInterval { + return false + } + l.last = now + return true +} + +// excludeUnprovisionablePods applies partitionUnprovisionablePods to a simulation's pending backlog +// using the configured TTL, records both populations on the simulation pending pods gauge, and logs +// a rate-limited sample of the excluded pods. +func excludeUnprovisionablePods(ctx context.Context, cluster *state.Cluster, clk clock.Clock, pods []*corev1.Pod) []*corev1.Pod { + ttl := options.FromContext(ctx).DisruptionUnprovisionablePodTTL + simulated, excluded := partitionUnprovisionablePods(cluster, clk, ttl, pods) + SimulationPendingPods.Set(float64(len(simulated)), map[string]string{dispositionLabel: simulationPodsDispositionSimulated}) + SimulationPendingPods.Set(float64(len(excluded)), map[string]string{dispositionLabel: simulationPodsDispositionExcluded}) + if len(excluded) == 0 { + return simulated + } + logger := log.FromContext(ctx).WithValues( + "excluded", len(excluded), + "simulated", len(simulated), + "ttl", ttl, + "sample", lo.Map(lo.Subset(excluded, 0, unprovisionablePodsLogSample), func(p *corev1.Pod, _ int) string { return klog.KObj(p).String() }), + ) + if excludedPodsLog.shouldLog(clk.Now()) { + logger.Info("excluding pending pods the provisioner could not place from disruption simulation") + } else { + logger.V(1).Info("excluding pending pods the provisioner could not place from disruption simulation") + } + return simulated +} diff --git a/pkg/controllers/disruption/unprovisionablepods_internal_test.go b/pkg/controllers/disruption/unprovisionablepods_internal_test.go new file mode 100644 index 0000000000..63fa52dbe9 --- /dev/null +++ b/pkg/controllers/disruption/unprovisionablepods_internal_test.go @@ -0,0 +1,120 @@ +/* +Copyright The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package disruption + +import ( + "context" + "errors" + "slices" + "testing" + "time" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/utils/clock" + clocktesting "k8s.io/utils/clock/testing" + + "sigs.k8s.io/karpenter/pkg/cloudprovider/fake" + "sigs.k8s.io/karpenter/pkg/controllers/state" +) + +func namespacedTestPod(name string) *corev1.Pod { + return &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "default"}} +} + +func podNames(pods []*corev1.Pod) []string { + names := make([]string, 0, len(pods)) + for _, p := range pods { + names = append(names, p.Name) + } + return names +} + +func newVerdictCluster(clk clock.Clock) *state.Cluster { + return state.NewCluster(clk, nil, fake.NewCloudProvider()) +} + +func TestPartitionUnprovisionablePodsExcludesFreshVerdictsOnly(t *testing.T) { + clk := clocktesting.NewFakeClock(time.Now()) + cluster := newVerdictCluster(clk) + ttl := 2 * time.Minute + + stale := namespacedTestPod("stale") + cluster.MarkPodsUnprovisionable([]*corev1.Pod{stale}) + clk.Step(ttl) + + fresh := namespacedTestPod("fresh") + cluster.MarkPodsUnprovisionable([]*corev1.Pod{fresh}) + clk.Step(ttl / 2) + + never := namespacedTestPod("never-decided") + placed := namespacedTestPod("placed") + cluster.MarkPodsUnprovisionable([]*corev1.Pod{placed}) + cluster.MarkPodSchedulingDecisions(context.Background(), nil, nil, map[string][]*corev1.Pod{"existing-node": {placed}}) + // An error the provisioner does not classify as pass-invariant is no verdict at all. + errored := namespacedTestPod("errored") + cluster.MarkPodSchedulingDecisions(context.Background(), map[*corev1.Pod]error{errored: errors.New("node limits have been exhausted for nodepool")}, nil, nil) + + backlog := []*corev1.Pod{never, fresh, stale, placed, errored} + simulated, excluded := partitionUnprovisionablePods(cluster, clk, ttl, backlog) + + // A verdict exactly ttl old has expired; only the one younger than ttl excludes its pod. + if got := podNames(simulated); !slices.Equal(got, []string{"never-decided", "stale", "placed", "errored"}) { + t.Fatalf("unexpected simulated pods %v", got) + } + if got := podNames(excluded); !slices.Equal(got, []string{"fresh"}) { + t.Fatalf("unexpected excluded pods %v", got) + } + if got := podNames(backlog); !slices.Equal(got, []string{"never-decided", "fresh", "stale", "placed", "errored"}) { + t.Fatalf("input backlog was modified: %v", got) + } +} + +func TestPartitionUnprovisionablePodsDisabledByZeroTTL(t *testing.T) { + clk := clocktesting.NewFakeClock(time.Now()) + cluster := newVerdictCluster(clk) + + pod := namespacedTestPod("fresh") + cluster.MarkPodsUnprovisionable([]*corev1.Pod{pod}) + + backlog := []*corev1.Pod{pod} + simulated, excluded := partitionUnprovisionablePods(cluster, clk, 0, backlog) + if len(excluded) != 0 || len(simulated) != 1 { + t.Fatalf("expected a zero ttl to exclude nothing, got simulated=%v excluded=%v", podNames(simulated), podNames(excluded)) + } + // The caller appends candidate pods to the returned slice, so it must not share the memoized backlog's backing array. + if &simulated[0] == &backlog[0] { + t.Fatalf("returned slice aliases the input backlog") + } +} + +func TestUnprovisionablePodsLogRateLimits(t *testing.T) { + var l unprovisionablePodsLog + now := time.Now() + if !l.shouldLog(now) { + t.Fatalf("the first report must be logged") + } + if l.shouldLog(now.Add(unprovisionablePodsLogInterval - time.Second)) { + t.Fatalf("a report inside the interval must be suppressed") + } + if !l.shouldLog(now.Add(unprovisionablePodsLogInterval)) { + t.Fatalf("a report at the interval must be logged") + } + if l.shouldLog(now.Add(unprovisionablePodsLogInterval + unprovisionablePodsLogInterval/2)) { + t.Fatalf("the interval restarts from the last accepted report") + } +} diff --git a/pkg/controllers/disruption/unprovisionablepods_test.go b/pkg/controllers/disruption/unprovisionablepods_test.go new file mode 100644 index 0000000000..ac0e9f34ce --- /dev/null +++ b/pkg/controllers/disruption/unprovisionablepods_test.go @@ -0,0 +1,387 @@ +/* +Copyright The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package disruption_test + +import ( + "slices" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "github.com/samber/lo" + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/resource" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" + + v1 "sigs.k8s.io/karpenter/pkg/apis/v1" + "sigs.k8s.io/karpenter/pkg/controllers/disruption" + "sigs.k8s.io/karpenter/pkg/controllers/provisioning/scheduling" + "sigs.k8s.io/karpenter/pkg/operator/options" + "sigs.k8s.io/karpenter/pkg/test" + . "sigs.k8s.io/karpenter/pkg/test/expectations" + "sigs.k8s.io/karpenter/pkg/utils/pdb" + "sigs.k8s.io/karpenter/pkg/utils/resources" +) + +// These tests cover the exclusion of pending pods every NodePool is incompatible with from +// disruption scheduling simulations (unprovisionablepods.go). The first fixture is the two-node, +// three-pod layout of "can delete nodes with a permanently pending pod": nodes[1] hosts one pod that +// fits on nodes[0], so single-node consolidation deletes nodes[1] regardless of the pending backlog. +// What varies is the backlog: a pod pinned to a capacity type no NodePool offers, which every +// provisioning pass records an incompatibility verdict for, or a pod any NodePool can launch for. +// The second fixture is a NodePool at its node limit, where the provisioner's rejection of a pending +// pod is lifted by the very node removal a disruption simulation performs. +var _ = Describe("Unprovisionable Pending Pods", func() { + const ttl = 2 * time.Minute + var nodePool *v1.NodePool + var nodeClaims []*v1.NodeClaim + var nodes []*corev1.Node + var rs *appsv1.ReplicaSet + var labels = map[string]string{"app": "test"} + var excluded = map[string]string{"disposition": "excluded_unprovisionable"} + var simulated = map[string]string{"disposition": "simulated"} + + BeforeEach(func() { + disruption.SimulationPendingPods.Reset() + ctx = options.ToContext(ctx, test.Options(test.OptionsFields{DisruptionUnprovisionablePodTTL: lo.ToPtr(ttl)})) + nodePool = test.NodePool(v1.NodePool{ + Spec: v1.NodePoolSpec{ + Disruption: v1.Disruption{ + ConsolidationPolicy: v1.ConsolidationPolicyWhenEmptyOrUnderutilized, + Budgets: []v1.Budget{{Nodes: "100%"}}, + ConsolidateAfter: v1.MustParseNillableDuration("0s"), + }, + }, + }) + nodeClaims, nodes = test.NodeClaimsAndNodes(2, v1.NodeClaim{ + ObjectMeta: metav1.ObjectMeta{ + Labels: map[string]string{ + v1.NodePoolLabelKey: nodePool.Name, + corev1.LabelInstanceTypeStable: mostExpensiveInstance.Name, + v1.CapacityTypeLabelKey: mostExpensiveOffering.Requirements.Get(v1.CapacityTypeLabelKey).Any(), + corev1.LabelTopologyZone: mostExpensiveOffering.Requirements.Get(corev1.LabelTopologyZone).Any(), + }, + }, + Status: v1.NodeClaimStatus{ + Allocatable: map[corev1.ResourceName]resource.Quantity{ + corev1.ResourceCPU: resource.MustParse("32"), + corev1.ResourcePods: resource.MustParse("100"), + }, + }, + }) + for _, nc := range nodeClaims { + nc.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(3, 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, pods[0], pods[1], pods[2], nodeClaims[0], nodes[0], nodeClaims[1], nodes[1], nodePool) + ExpectManualBinding(ctx, env.Client, pods[0], nodes[0]) + ExpectManualBinding(ctx, env.Client, pods[1], nodes[0]) + ExpectManualBinding(ctx, env.Client, pods[2], nodes[1]) + }) + + // reservedOnlyPod is pinned to a capacity type the fake cloud provider never offers, so the + // provisioner can place it nowhere: the shape of a reserved-only pod once every reservation is + // exhausted. + reservedOnlyPod := func() *corev1.Pod { + return test.UnschedulablePod(test.PodOptions{ + NodeSelector: map[string]string{v1.CapacityTypeLabelKey: v1.CapacityTypeReserved}, + }) + } + + // consolidate syncs cluster state and runs one disruption pass, returning the commands it queued. + consolidate := func() []*disruption.Command { + GinkgoHelper() + ExpectMakeNodesAndNodeClaimsInitializedAndStateUpdated(ctx, env.Client, env.Clock, nodeStateController, nodeClaimStateController, nodes, nodeClaims) + ExpectSingletonReconciled(ctx, disruptionController) + return queue.GetCommands() + } + + It("should exclude a pending pod the provisioner could not place and still consolidate", func() { + pending := reservedOnlyPod() + ExpectProvisionedNoBinding(ctx, env.Client, cluster, cloudProvider, prov, pending) + Expect(cluster.PodUnprovisionableTime(client.ObjectKeyFromObject(pending)).IsZero()).To(BeFalse()) + Expect(ExpectNodeClaims(ctx, env.Client)).To(HaveLen(2)) + + cmds := consolidate() + ExpectMetricGaugeValue(disruption.SimulationPendingPods, 1, excluded) + ExpectMetricGaugeValue(disruption.SimulationPendingPods, 0, simulated) + + Expect(cmds).To(HaveLen(1)) + Expect(cmds[0].Candidates).ToNot(BeEmpty()) + Expect(cmds[0].Reason()).To(Equal(v1.DisruptionReasonUnderutilized)) + }) + It("should exclude the pod from drift simulations and still replace the drifted node", func() { + nodeClaims[1].StatusConditions().SetTrue(v1.ConditionTypeDrifted) + ExpectApplied(ctx, env.Client, nodeClaims[1]) + pending := reservedOnlyPod() + ExpectProvisionedNoBinding(ctx, env.Client, cluster, cloudProvider, prov, pending) + Expect(cluster.PodUnprovisionableTime(client.ObjectKeyFromObject(pending)).IsZero()).To(BeFalse()) + + cmds := consolidate() + ExpectMetricGaugeValue(disruption.SimulationPendingPods, 1, excluded) + ExpectMetricGaugeValue(disruption.SimulationPendingPods, 0, simulated) + + Expect(cmds).To(HaveLen(1)) + Expect(cmds[0].Reason()).To(Equal(v1.DisruptionReasonDrifted)) + Expect(cmds[0].Candidates).To(HaveLen(1)) + Expect(cmds[0].Candidates[0].Name()).To(Equal(nodes[1].Name)) + }) + It("should simulate the pod again once its verdict is older than the TTL", func() { + pending := reservedOnlyPod() + ExpectProvisionedNoBinding(ctx, env.Client, cluster, cloudProvider, prov, pending) + env.Clock.Step(ttl) + + cmds := consolidate() + ExpectMetricGaugeValue(disruption.SimulationPendingPods, 0, excluded) + ExpectMetricGaugeValue(disruption.SimulationPendingPods, 1, simulated) + Expect(cmds).To(HaveLen(1)) + }) + It("should simulate every pending pod when the TTL is zero", func() { + ctx = options.ToContext(ctx, test.Options(test.OptionsFields{DisruptionUnprovisionablePodTTL: lo.ToPtr(time.Duration(0))})) + pending := reservedOnlyPod() + ExpectProvisionedNoBinding(ctx, env.Client, cluster, cloudProvider, prov, pending) + Expect(cluster.PodUnprovisionableTime(client.ObjectKeyFromObject(pending)).IsZero()).To(BeFalse()) + + cmds := consolidate() + ExpectMetricGaugeValue(disruption.SimulationPendingPods, 0, excluded) + ExpectMetricGaugeValue(disruption.SimulationPendingPods, 1, simulated) + Expect(cmds).To(HaveLen(1)) + }) + It("should simulate a pending pod the provisioner has not decided on", func() { + pending := reservedOnlyPod() + ExpectApplied(ctx, env.Client, pending) + + consolidate() + ExpectMetricGaugeValue(disruption.SimulationPendingPods, 0, excluded) + ExpectMetricGaugeValue(disruption.SimulationPendingPods, 1, simulated) + }) + It("should simulate a pending pod the provisioner is launching capacity for", func() { + pending := test.UnschedulablePod(test.PodOptions{ + ResourceRequirements: corev1.ResourceRequirements{Requests: corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("1")}}, + }) + ExpectProvisionedNoBinding(ctx, env.Client, cluster, cloudProvider, prov, pending) + Expect(cluster.PodUnprovisionableTime(client.ObjectKeyFromObject(pending)).IsZero()).To(BeTrue()) + + consolidate() + ExpectMetricGaugeValue(disruption.SimulationPendingPods, 0, excluded) + ExpectMetricGaugeValue(disruption.SimulationPendingPods, 1, simulated) + }) + It("should keep retrying an excluded pod in provisioning and simulate it again once it is placeable", func() { + pending := test.UnschedulablePod(test.PodOptions{ + NodeSelector: map[string]string{"example.com/pool": "reserved"}, + }) + ExpectProvisionedNoBinding(ctx, env.Client, cluster, cloudProvider, prov, pending) + Expect(cluster.PodUnprovisionableTime(client.ObjectKeyFromObject(pending)).IsZero()).To(BeFalse()) + Expect(ExpectNodeClaims(ctx, env.Client)).To(HaveLen(2)) + + // A NodePool that can launch for the pod appears. The next provisioning pass, which reads the + // backlog directly, opens a NodeClaim for it and clears the verdict. + reservedPool := test.NodePool(v1.NodePool{ + Spec: v1.NodePoolSpec{ + Template: v1.NodeClaimTemplate{ + Spec: v1.NodeClaimTemplateSpec{ + Requirements: []v1.NodeSelectorRequirementWithMinValues{{ + Key: "example.com/pool", Operator: corev1.NodeSelectorOpIn, Values: []string{"reserved"}, + }}, + }, + }, + }, + }) + ExpectApplied(ctx, env.Client, reservedPool) + ExpectProvisionedNoBinding(ctx, env.Client, cluster, cloudProvider, prov, pending) + Expect(cluster.PodUnprovisionableTime(client.ObjectKeyFromObject(pending)).IsZero()).To(BeTrue()) + Expect(ExpectNodeClaims(ctx, env.Client)).To(HaveLen(3)) + + consolidate() + ExpectMetricGaugeValue(disruption.SimulationPendingPods, 0, excluded) + ExpectMetricGaugeValue(disruption.SimulationPendingPods, 1, simulated) + }) +}) + +// A pending pod the provisioner rejected only because its NodePool was at its node limit is placeable +// inside a disruption simulation: SimulateScheduling removes the candidate from the cluster state, the +// NodePool regains one node of headroom and the scheduler opens a replacement that must be sized for +// the pending pod as well as the candidate's pods. Such a rejection records no verdict, so the pod is +// simulated and the replacement is the same one a simulation with the exclusion disabled produces. +// A pod the capped NodePool is incompatible with regardless of limits keeps its verdict and stays out. +// SimulateScheduling is the path every disruption method (single-node and multi-node consolidation, +// drift, and command validation) schedules through, so the check covers all of them. +var _ = Describe("Unprovisionable Pending Pods with NodePool limits", func() { + var nodePool *v1.NodePool + var nodeClaims []*v1.NodeClaim + var nodes []*corev1.Node + + BeforeEach(func() { + disruption.SimulationPendingPods.Reset() + nodePool = test.NodePool(v1.NodePool{ + Spec: v1.NodePoolSpec{ + Limits: v1.Limits(corev1.ResourceList{resources.Node: resource.MustParse("2")}), + Disruption: v1.Disruption{ + ConsolidationPolicy: v1.ConsolidationPolicyWhenEmptyOrUnderutilized, + Budgets: []v1.Budget{{Nodes: "100%"}}, + ConsolidateAfter: v1.MustParseNillableDuration("0s"), + }, + }, + }) + nodeClaims, nodes = test.NodeClaimsAndNodes(2, v1.NodeClaim{ + ObjectMeta: metav1.ObjectMeta{ + Labels: map[string]string{ + v1.NodePoolLabelKey: nodePool.Name, + corev1.LabelInstanceTypeStable: mostExpensiveInstance.Name, + v1.CapacityTypeLabelKey: mostExpensiveOffering.Requirements.Get(v1.CapacityTypeLabelKey).Any(), + corev1.LabelTopologyZone: mostExpensiveOffering.Requirements.Get(corev1.LabelTopologyZone).Any(), + }, + }, + Status: v1.NodeClaimStatus{ + Allocatable: map[corev1.ResourceName]resource.Quantity{ + corev1.ResourceCPU: resource.MustParse("32"), + corev1.ResourcePods: resource.MustParse("100"), + }, + }, + }) + for _, nc := range nodeClaims { + nc.StatusConditions().SetTrue(v1.ConditionTypeConsolidatable) + } + // nodes[0] is full; nodes[1], the candidate, hosts one small pod that cannot move to nodes[0]. + full := test.Pods(2, test.PodOptions{ResourceRequirements: corev1.ResourceRequirements{Requests: corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("16")}}}) + candidatePod := test.Pod(test.PodOptions{ResourceRequirements: corev1.ResourceRequirements{Requests: corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("1")}}}) + ExpectApplied(ctx, env.Client, full[0], full[1], candidatePod, nodeClaims[0], nodes[0], nodeClaims[1], nodes[1], nodePool) + ExpectManualBinding(ctx, env.Client, full[0], nodes[0]) + ExpectManualBinding(ctx, env.Client, full[1], nodes[0]) + ExpectManualBinding(ctx, env.Client, candidatePod, nodes[1]) + ExpectMakeNodesAndNodeClaimsInitializedAndStateUpdated(ctx, env.Client, env.Clock, nodeStateController, nodeClaimStateController, nodes, nodeClaims) + }) + + // simulateRemoving runs the disruption scheduling simulation for the removal of the given nodes under the given TTL. + simulateRemoving := func(ttl time.Duration, removed ...*corev1.Node) scheduling.Results { + GinkgoHelper() + ctx = options.ToContext(ctx, test.Options(test.OptionsFields{DisruptionUnprovisionablePodTTL: lo.ToPtr(ttl)})) + nodePoolMap, nodePoolToInstanceTypesMap, err := disruption.BuildNodePoolMap(ctx, env.Client, cloudProvider) + Expect(err).To(Succeed()) + pdbs, err := pdb.NewLimits(ctx, env.Client) + Expect(err).To(Succeed()) + candidates := lo.Map(removed, func(n *corev1.Node, _ int) *disruption.Candidate { + candidate, err := disruption.NewCandidate(ctx, env.Client, recorder, env.Clock, ExpectStateNodeExists(cluster, n), pdbs, nodePoolMap, nodePoolToInstanceTypesMap, queue, disruption.GracefulDisruptionClass) + Expect(err).To(Succeed()) + return candidate + }) + results, err := disruption.SimulateScheduling(ctx, env.Client, cluster, prov, env.Clock, recorder, nil, candidates...) + Expect(err).To(Succeed()) + return results + } + // simulate runs the disruption scheduling simulation for the removal of nodes[1], the candidate, under the given TTL. + simulate := func(ttl time.Duration) scheduling.Results { + GinkgoHelper() + return simulateRemoving(ttl, nodes[1]) + } + podNames := func(pods []*corev1.Pod) []string { + return lo.Map(pods, func(p *corev1.Pod, _ int) string { return p.Name }) + } + // replacementPodSets returns the sorted pod names of each new NodeClaim, ordered by NodeClaim size. + replacementPodSets := func(results scheduling.Results) [][]string { + sets := lo.Map(results.NewNodeClaims, func(nc *scheduling.NodeClaim, _ int) []string { + names := podNames(nc.Pods) + slices.Sort(names) + return names + }) + slices.SortFunc(sets, func(a, b []string) int { return len(a) - len(b) }) + return sets + } + + It("should simulate a pod the provisioner rejected on NodePool limits and size the replacement for it", func() { + // Too big for the 31 free CPUs on the candidate, fits a fresh node, but the NodePool is at its two-node + // limit while both nodes exist, so the provisioner rejects it. + pending := test.UnschedulablePod(test.PodOptions{ + ResourceRequirements: corev1.ResourceRequirements{Requests: corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("40")}}, + }) + ExpectProvisionedNoBinding(ctx, env.Client, cluster, cloudProvider, prov, pending) + Expect(ExpectNodeClaims(ctx, env.Client)).To(HaveLen(2)) + Expect(cluster.PodUnprovisionableTime(client.ObjectKeyFromObject(pending)).IsZero()).To(BeTrue()) + + baseline := simulate(0) + Expect(baseline.PodErrors).NotTo(HaveKey(pending)) + Expect(baseline.NewNodeClaims).To(HaveLen(1)) + Expect(podNames(baseline.NewNodeClaims[0].Pods)).To(ContainElement(pending.Name)) + + filtered := simulate(2 * time.Minute) + ExpectMetricGaugeValue(disruption.SimulationPendingPods, 0, map[string]string{"disposition": "excluded_unprovisionable"}) + ExpectMetricGaugeValue(disruption.SimulationPendingPods, 1, map[string]string{"disposition": "simulated"}) + Expect(filtered.PodErrors).NotTo(HaveKey(pending)) + Expect(filtered.NewNodeClaims).To(HaveLen(1)) + Expect(podNames(filtered.NewNodeClaims[0].Pods)).To(ConsistOf(podNames(baseline.NewNodeClaims[0].Pods))) + Expect(len(filtered.NewNodeClaims[0].InstanceTypeOptions)).To(Equal(len(baseline.NewNodeClaims[0].InstanceTypeOptions))) + }) + It("should simulate a pod the provisioner rejected on NodePool limits when several nodes are removed at once", func() { + // Removing both nodes frees the whole two-node limit: the full node's pods, the candidate's pod and the pending + // pod all need replacements, as in a multi-node consolidation simulation. + pending := test.UnschedulablePod(test.PodOptions{ + ResourceRequirements: corev1.ResourceRequirements{Requests: corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("40")}}, + }) + ExpectProvisionedNoBinding(ctx, env.Client, cluster, cloudProvider, prov, pending) + Expect(ExpectNodeClaims(ctx, env.Client)).To(HaveLen(2)) + Expect(cluster.PodUnprovisionableTime(client.ObjectKeyFromObject(pending)).IsZero()).To(BeTrue()) + + baseline := simulateRemoving(0, nodes[0], nodes[1]) + Expect(baseline.PodErrors).NotTo(HaveKey(pending)) + Expect(lo.Flatten(replacementPodSets(baseline))).To(ContainElement(pending.Name)) + + filtered := simulateRemoving(2*time.Minute, nodes[0], nodes[1]) + ExpectMetricGaugeValue(disruption.SimulationPendingPods, 0, map[string]string{"disposition": "excluded_unprovisionable"}) + ExpectMetricGaugeValue(disruption.SimulationPendingPods, 1, map[string]string{"disposition": "simulated"}) + Expect(filtered.PodErrors).NotTo(HaveKey(pending)) + Expect(replacementPodSets(filtered)).To(Equal(replacementPodSets(baseline))) + }) + It("should still exclude a pod the NodePool at its limit is incompatible with", func() { + // Pinned to a capacity type no instance type offers: the limit is not what keeps it pending, and the + // headroom the candidate's removal frees changes nothing for it. + pending := test.UnschedulablePod(test.PodOptions{ + NodeSelector: map[string]string{v1.CapacityTypeLabelKey: v1.CapacityTypeReserved}, + }) + ExpectProvisionedNoBinding(ctx, env.Client, cluster, cloudProvider, prov, pending) + Expect(ExpectNodeClaims(ctx, env.Client)).To(HaveLen(2)) + Expect(cluster.PodUnprovisionableTime(client.ObjectKeyFromObject(pending)).IsZero()).To(BeFalse()) + + baseline := simulate(0) + Expect(baseline.PodErrors).To(HaveKey(pending)) + Expect(baseline.NewNodeClaims).To(HaveLen(1)) + + filtered := simulate(2 * time.Minute) + ExpectMetricGaugeValue(disruption.SimulationPendingPods, 1, map[string]string{"disposition": "excluded_unprovisionable"}) + ExpectMetricGaugeValue(disruption.SimulationPendingPods, 0, map[string]string{"disposition": "simulated"}) + Expect(filtered.PodErrors).To(BeEmpty()) + Expect(filtered.NewNodeClaims).To(HaveLen(1)) + Expect(podNames(filtered.NewNodeClaims[0].Pods)).To(ConsistOf(podNames(baseline.NewNodeClaims[0].Pods))) + }) +}) diff --git a/pkg/controllers/provisioning/provisioner.go b/pkg/controllers/provisioning/provisioner.go index f0e51fa85d..50c952fe5f 100644 --- a/pkg/controllers/provisioning/provisioner.go +++ b/pkg/controllers/provisioning/provisioner.go @@ -467,6 +467,15 @@ func (p *Provisioner) Schedule(ctx context.Context) (scheduler.Results, error) { // Only passing existing nodes here and not new nodeClaims because // these nodeClaims don't have a name until they are created filterVirtualPodMapping(results.ExistingNodeToPodMapping())) + // Of the errored pods, only those every NodePool rejected for a reason that holds regardless of cluster state get + // an unprovisionable verdict for disruption simulations to rely on. Limits, reserved offerings, topology, volumes + // and DRA are all pass-dependent and record nothing; a NodePool at its limits still reports an incompatibility it + // would have with the pod anyway. Because such a verdict does not depend on what else the pass + // scheduled, one from a deadline-cut pass is as sound as one from a complete pass, even for a pod the deadline + // kept from being retried. + p.cluster.MarkPodsUnprovisionable(lo.Reject(results.PodsIncompatibleWithAllNodePools(), func(pod *corev1.Pod, _ int) bool { + return IsVirtualPod(pod) + })) results.Record(ctx, p.recorder, p.cluster) return results, nil } diff --git a/pkg/controllers/provisioning/scheduling/nodeclaim.go b/pkg/controllers/provisioning/scheduling/nodeclaim.go index 4d09fa10ec..8b6af346ee 100644 --- a/pkg/controllers/provisioning/scheduling/nodeclaim.go +++ b/pkg/controllers/provisioning/scheduling/nodeclaim.go @@ -26,6 +26,7 @@ import ( "unique" "github.com/samber/lo" + "go.uber.org/multierr" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/resource" "k8s.io/apimachinery/pkg/util/sets" @@ -59,6 +60,10 @@ type NodeClaim struct { // this expansion. reservedOfferings cloudprovider.Offerings reservedOfferingMode ReservedOfferingMode + // instanceTypesBeforeLimits holds the template's full instance type set when NodePool limits trimmed it before this + // NodeClaim was built, and is nil otherwise. An instance type filter failure over a trimmed set may not recur once + // limits free up, so it only counts as an incompatibility if the full set rejects the pod too. + instanceTypesBeforeLimits []*cloudprovider.InstanceType } // ReservedOfferingError indicates a NodeClaim couldn't be created or a pod couldn't be added to an exxisting NodeClaim @@ -80,6 +85,59 @@ func (e ReservedOfferingError) Unwrap() error { return e.error } +// NodePoolIncompatibleError marks a rejection that follows from the pod and the NodePool alone — the NodePool's taints, +// requirements or instance types — and not from anything that differs between scheduling passes over the same +// NodePools: which nodes exist and what runs on them, topology counts, NodePool limits, DRA device state or the +// reserved-offering mode. A pod every NodePool rejects this way cannot open a NodeClaim in any such pass, which is what +// lets disruption simulations leave it out (see Results.PodsIncompatibleWithAllNodePools). Error() is lazy so wrapping an +// InstanceTypeFilterError stays cheap. +type NodePoolIncompatibleError struct { + error +} + +func NewNodePoolIncompatibleError(err error) NodePoolIncompatibleError { + return NodePoolIncompatibleError{error: err} +} + +func IsNodePoolIncompatibleError(err error) bool { + nie := &NodePoolIncompatibleError{} + return errors.As(err, nie) +} + +func (e NodePoolIncompatibleError) Unwrap() error { + return e.error +} + +// IsIncompatibleWithAllNodePools reports whether err, the error Solve attached to a pod, consists solely of +// NodePoolIncompatibleErrors: every NodePool rejected the pod for a reason that holds in any scheduling pass over the +// same NodePools. Errors from other stages (no NodePools, DRA, minValues truncation) or any single pass-dependent +// NodePool rejection (limits, topology, reserved offerings) make it false. +func IsIncompatibleWithAllNodePools(err error) bool { + errs := multierr.Errors(err) + return len(errs) > 0 && lo.EveryBy(errs, IsNodePoolIncompatibleError) +} + +// NodePoolLimitError marks a NodePool that rejected a pod only because launching for it would breach the NodePool's +// limits given the nodes that exist in this pass. Removing a node, as every disruption simulation does, can lift it, so +// it is never a NodePoolIncompatibleError. A NodePool that would reject the pod even with unlimited capacity reports +// that incompatibility instead (see incompatibleIgnoringLimits). +type NodePoolLimitError struct { + error +} + +func NewNodePoolLimitError(err error) NodePoolLimitError { + return NodePoolLimitError{error: err} +} + +func IsNodePoolLimitError(err error) bool { + nle := &NodePoolLimitError{} + return errors.As(err, nle) +} + +func (e NodePoolLimitError) Unwrap() error { + return e.error +} + var nodeID int64 func NewNodeClaim( @@ -124,14 +182,14 @@ func NewNodeClaim( func (n *NodeClaim) CanAdd(ctx context.Context, pod *corev1.Pod, podData *PodData, relaxMinValues bool, allocator *dynamicresources.Allocator) (updatedRequirements scheduling.Requirements, updatedInstanceTypes []*cloudprovider.InstanceType, offeringsToReserve []*cloudprovider.Offering, allocationResult *dynamicresources.AllocationResult, err error) { // Check Taints if err := scheduling.Taints(n.Spec.Taints).ToleratesPod(pod); err != nil { - return nil, nil, nil, nil, err + return nil, nil, nil, nil, NewNodePoolIncompatibleError(err) } baseRequirements := scheduling.NewRequirements(n.Requirements.Values()...) // Check NodeClaim Affinity Requirements if err := baseRequirements.Compatible(podData.Requirements, scheduling.AllowUndefinedWellKnownLabels); err != nil { - return nil, nil, nil, nil, fmt.Errorf("incompatible requirements, %w", err) + return nil, nil, nil, nil, NewNodePoolIncompatibleError(fmt.Errorf("incompatible requirements, %w", err)) } baseRequirements.Add(podData.Requirements.Values()...) @@ -146,14 +204,22 @@ func (n *NodeClaim) CanAdd(ctx context.Context, pod *corev1.Pod, podData *PodDat // Try each volume topology alternative. We need to iterate here because the selected // volume topology constraints affect downstream topology checks (e.g., pod anti-affinity). var lastErr error + allIncompatible := true for _, volReqs := range volumeAlternatives { reqs, its, ofs, result, err := n.tryVolumeAlternative(ctx, pod, podData, baseRequirements, volReqs, relaxMinValues, allocator) if err != nil { lastErr = err + allIncompatible = allIncompatible && IsNodePoolIncompatibleError(err) continue } return reqs, its, ofs, result, nil } + // The pod is incompatible with this NodePool only if every volume alternative is; if one failed for a reason that + // may not recur, the reported error must not claim otherwise. + nie := &NodePoolIncompatibleError{} + if !allIncompatible && errors.As(lastErr, nie) { + lastErr = nie.error + } return nil, nil, nil, nil, lastErr } @@ -218,8 +284,11 @@ func (n *NodeClaim) tryVolumeAlternative(ctx context.Context, pod *corev1.Pod, p } } if err != nil { - // We avoid wrapping this err because calling String() on InstanceTypeFilterError is an expensive operation - // due to calls to resources.Merge and stringifying the nodeClaimRequirements + // We avoid wrapping this err with fmt.Errorf because calling String() on InstanceTypeFilterError is an expensive + // operation due to calls to resources.Merge and stringifying the nodeClaimRequirements. + if n.instanceTypeFilterFailureIsInvariant(pod, podData, relaxMinValues) { + return nil, nil, nil, nil, NewNodePoolIncompatibleError(err) + } return nil, nil, nil, nil, err } // Apply the DRA-specific instance type filter: only instance types whose device allocation succeeded survive. @@ -241,6 +310,49 @@ func (n *NodeClaim) tryVolumeAlternative(ctx context.Context, pod *corev1.Pod, p return nodeClaimRequirements, remaining, ofs, allocationResult, nil } +// instanceTypeFilterFailureIsInvariant reports whether a failed instance type filter for pod rules the pod out of this +// NodePool in every scheduling pass over the same NodePools. That holds when the filter saw the pod's own requirements +// only — neither volumes, DRA nor topology could have tightened them — and, if NodePool limits trimmed the instance +// types it saw, when the untrimmed set rejects the pod as well. +func (n *NodeClaim) instanceTypeFilterFailureIsInvariant(pod *corev1.Pod, podData *PodData, relaxMinValues bool) bool { + if len(podData.VolumeRequirements) > 0 || podData.HasResourceClaimRequests || n.topology.Constrains(pod) { + return false + } + if n.instanceTypesBeforeLimits == nil { + return true + } + return incompatibleIgnoringLimits(&n.NodeClaimTemplate, n.topology, n.daemonOverheadGroups, n.instanceTypesBeforeLimits, pod, podData, relaxMinValues) != nil +} + +// incompatibleIgnoringLimits returns why pod could not join a NodeClaim built from template even if NodePool limits left +// every one of instanceTypes available: the template's taints, its requirements, or the instance types themselves +// (with the pod's own requirements and requests) reject the pod. It returns nil when the pod is compatible, and also +// when volumes, DRA or topology could tighten the pod's requirements, since those depend on the scheduling pass and +// the instance type check would not be conclusive. It has no side effects on the scheduler's state. +func incompatibleIgnoringLimits( + template *NodeClaimTemplate, + topology *Topology, + daemonOverheadGroups []DaemonOverheadGroup, + instanceTypes []*cloudprovider.InstanceType, + pod *corev1.Pod, + podData *PodData, + relaxMinValues bool, +) error { + if err := scheduling.Taints(template.Spec.Taints).ToleratesPod(pod); err != nil { + return err + } + requirements := scheduling.NewRequirements(template.Requirements.Values()...) + if err := requirements.Compatible(podData.Requirements, scheduling.AllowUndefinedWellKnownLabels); err != nil { + return fmt.Errorf("incompatible requirements, %w", err) + } + if len(podData.VolumeRequirements) > 0 || podData.HasResourceClaimRequests || topology.Constrains(pod) { + return nil + } + requirements.Add(podData.Requirements.Values()...) + _, _, err := filterInstanceTypesByRequirements(instanceTypes, requirements, pod, podData.Requests, daemonOverheadGroups, podData.Requests, relaxMinValues) + return err +} + // Add updates the NodeClaim to schedule the pod to this NodeClaim, updating // the NodeClaim with new requirements, instance types, and offerings to reserve // based on the pod scheduling diff --git a/pkg/controllers/provisioning/scheduling/scheduler.go b/pkg/controllers/provisioning/scheduling/scheduler.go index 48743b8822..c0df49acb1 100644 --- a/pkg/controllers/provisioning/scheduling/scheduler.go +++ b/pkg/controllers/provisioning/scheduling/scheduler.go @@ -403,6 +403,16 @@ func (r Results) DRAErrors() map[*corev1.Pod]error { }) } +// PodsIncompatibleWithAllNodePools returns the pods every NodePool rejected for a reason that holds in any scheduling +// pass over the same NodePools (see IsIncompatibleWithAllNodePools). Solve records a pod's most recent attempt, and +// such a rejection does not depend on the other pods in the batch or on how far the pass got, so the verdict stands +// even for a pass cut short by its deadline. +func (r Results) PodsIncompatibleWithAllNodePools() []*corev1.Pod { + return lo.Keys(lo.PickBy(r.PodErrors, func(_ *corev1.Pod, err error) bool { + return IsIncompatibleWithAllNodePools(err) + })) +} + func (r Results) NodePoolToPodMapping() map[string][]*corev1.Pod { result := make(map[string][]*corev1.Pod) @@ -756,12 +766,12 @@ func (s *Scheduler) addToNewNodeClaim(ctx context.Context, pod *corev1.Pod) erro // Node limits can be enforced early, since we know exactly how much capacity in nodes will be consumed by any instance type (1 node). nodesRemaining, ok := remaining[resources.Node] if ok && nodesRemaining.IsZero() { - errs[i] = serrors.Wrap(fmt.Errorf("node limits have been exhausted for nodepool"), "NodePool", klog.KRef("", s.nodeClaimTemplates[i].NodePoolName)) + errs[i] = s.limitRejection(s.nodeClaimTemplates[i], pod, "node limits have been exhausted for nodepool") return true } its = filterByRemainingResources(its, remaining) if len(its) == 0 { - errs[i] = serrors.Wrap(fmt.Errorf("all available instance types exceed limits for nodepool"), "NodePool", klog.KRef("", s.nodeClaimTemplates[i].NodePoolName)) + errs[i] = s.limitRejection(s.nodeClaimTemplates[i], pod, "all available instance types exceed limits for nodepool") return true } else if len(s.nodeClaimTemplates[i].InstanceTypeOptions) != len(its) { log.FromContext(ctx).V(1).WithValues( @@ -772,6 +782,9 @@ func (s *Scheduler) addToNewNodeClaim(ctx context.Context, pod *corev1.Pod) erro } } nodeClaim := NewNodeClaim(s.nodeClaimTemplates[i], s.topology, s.daemonOverheadGroups[s.nodeClaimTemplates[i]], its, s.reservationManager, s.reservedOfferingMode) + if len(its) != len(s.nodeClaimTemplates[i].InstanceTypeOptions) { + nodeClaim.instanceTypesBeforeLimits = s.nodeClaimTemplates[i].InstanceTypeOptions + } r, its, ofs, result, err := nodeClaim.CanAdd(ctx, pod, s.cachedPodData[pod.UID], s.minValuesPolicy == karpopts.MinValuesPolicyBestEffort, s.allocator) if err != nil { errs[i] = err @@ -1042,6 +1055,16 @@ func volumeZoneReq(volumeReqs []scheduling.Requirements) *scheduling.Requirement return merged } +// limitRejection builds the error for a NodePool whose limits leave no room to launch for pod. A pod the NodePool would +// reject even with unlimited capacity gets that rejection, a NodePoolIncompatibleError, so that a NodePool sitting at +// its limits never hides an incompatibility the pod has with it; otherwise the rejection is a NodePoolLimitError. +func (s *Scheduler) limitRejection(template *NodeClaimTemplate, pod *corev1.Pod, reason string) error { + if err := incompatibleIgnoringLimits(template, s.topology, s.daemonOverheadGroups[template], template.InstanceTypeOptions, pod, s.cachedPodData[pod.UID], s.minValuesPolicy == karpopts.MinValuesPolicyBestEffort); err != nil { + return NewNodePoolIncompatibleError(err) + } + return NewNodePoolLimitError(serrors.Wrap(errors.New(reason), "NodePool", klog.KRef("", template.NodePoolName))) +} + // parallelizeUntil is an implementation of workqueue.ParallelizeUntil that modifies the // doWorkPiece so that a worker always finishes its work when it pulls a piece off of pieces // The function returns a bool that represents whether the worker should continue doing work diff --git a/pkg/controllers/provisioning/scheduling/topology.go b/pkg/controllers/provisioning/scheduling/topology.go index 0f6b505d98..af2370d704 100644 --- a/pkg/controllers/provisioning/scheduling/topology.go +++ b/pkg/controllers/provisioning/scheduling/topology.go @@ -621,6 +621,24 @@ func (t *Topology) buildNamespaceList(ctx context.Context, namespace string, nam return selected, nil } +// Constrains reports whether any topology could tighten pod p's placement: p owns a topology spread, affinity or +// anti-affinity, or another pod's anti-affinity selects it. It is a superset of getMatchingTopologies for every node, +// since it ignores the node filter, so a false answer means no pass over any cluster state adds topology requirements +// for p. +func (t *Topology) Constrains(p *corev1.Pod) bool { + for _, tg := range t.topologyGroups { + if tg.IsOwnedBy(p.UID) { + return true + } + } + for _, tg := range t.inverseTopologyGroups { + if tg.selects(p) { + return true + } + } + return false +} + // getMatchingTopologies returns a sorted list of topologies that either control the scheduling of pod p, or for which // the topology selects pod p and the scheduling of p affects the count per topology domain func (t *Topology) getMatchingTopologies(p *corev1.Pod, taints []corev1.Taint, requirements scheduling.Requirements, compatibilityOptions ...option.Function[scheduling.CompatibilityOptions]) []*TopologyGroup { diff --git a/pkg/controllers/provisioning/scheduling/unprovisionableverdict_test.go b/pkg/controllers/provisioning/scheduling/unprovisionableverdict_test.go new file mode 100644 index 0000000000..29ca5c0ade --- /dev/null +++ b/pkg/controllers/provisioning/scheduling/unprovisionableverdict_test.go @@ -0,0 +1,367 @@ +/* +Copyright The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package scheduling_test + +import ( + "context" + "errors" + "fmt" + "sync/atomic" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "github.com/samber/lo" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/resource" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" + + v1 "sigs.k8s.io/karpenter/pkg/apis/v1" + "sigs.k8s.io/karpenter/pkg/cloudprovider" + "sigs.k8s.io/karpenter/pkg/cloudprovider/fake" + "sigs.k8s.io/karpenter/pkg/controllers/provisioning/scheduling" + "sigs.k8s.io/karpenter/pkg/operator/options" + pscheduling "sigs.k8s.io/karpenter/pkg/scheduling" + "sigs.k8s.io/karpenter/pkg/test" + . "sigs.k8s.io/karpenter/pkg/test/expectations" + "sigs.k8s.io/karpenter/pkg/test/v1alpha1" + "sigs.k8s.io/karpenter/pkg/utils/resources" +) + +// These tests pin down which provisioning failures become an unprovisionable verdict in the cluster +// state (state.Cluster.PodUnprovisionableTime), the verdict disruption simulations use to leave a +// pending pod out. Only a rejection every NodePool issues on the pod and the NodePool alone - taints, +// requirements, or no instance type for the pod's own requirements - qualifies +// (scheduling.IsIncompatibleWithAllNodePools). Anything that can come out differently in another +// scheduling pass over the same NodePools records nothing: NodePool limits, which a node removal lifts; +// a reserved offering the strict provisioning mode deferred; topology, which depends on what else is +// running; and a pass the Solve deadline cut short before the pod was retried. Limits only shield a pod +// the NodePool could otherwise take: a NodePool at its limits still reports an incompatibility it has +// with the pod, so a capped NodePool does not keep every pod in every simulation. +var _ = Describe("Unprovisionable Verdicts", func() { + var nodePool *v1.NodePool + verdict := func(pod *corev1.Pod) bool { + return !cluster.PodUnprovisionableTime(client.ObjectKeyFromObject(pod)).IsZero() + } + + BeforeEach(func() { + nodePool = test.NodePool() + }) + + Context("incompatible with every NodePool", func() { + It("should record a verdict for a pod that tolerates no NodePool's taints", func() { + nodePool.Spec.Template.Spec.Taints = []corev1.Taint{{Key: "dedicated", Value: "gpu", Effect: corev1.TaintEffectNoSchedule}} + ExpectApplied(ctx, env.Client, nodePool) + pod := test.UnschedulablePod() + ExpectProvisionedNoBinding(ctx, env.Client, cluster, cloudProvider, prov, pod) + Expect(ExpectNodeClaims(ctx, env.Client)).To(BeEmpty()) + Expect(verdict(pod)).To(BeTrue()) + }) + It("should record a verdict for a pod whose requirements contradict every NodePool's", func() { + nodePool.Spec.Template.Spec.Requirements = []v1.NodeSelectorRequirementWithMinValues{{ + Key: v1.CapacityTypeLabelKey, Operator: corev1.NodeSelectorOpIn, Values: []string{v1.CapacityTypeOnDemand}, + }} + ExpectApplied(ctx, env.Client, nodePool) + pod := test.UnschedulablePod(test.PodOptions{NodeSelector: map[string]string{v1.CapacityTypeLabelKey: v1.CapacityTypeSpot}}) + ExpectProvisionedNoBinding(ctx, env.Client, cluster, cloudProvider, prov, pod) + Expect(ExpectNodeClaims(ctx, env.Client)).To(BeEmpty()) + Expect(verdict(pod)).To(BeTrue()) + }) + It("should record a verdict for a pod pinned to a capacity type no instance type offers", func() { + ExpectApplied(ctx, env.Client, nodePool) + pod := test.UnschedulablePod(test.PodOptions{NodeSelector: map[string]string{v1.CapacityTypeLabelKey: v1.CapacityTypeReserved}}) + ExpectProvisionedNoBinding(ctx, env.Client, cluster, cloudProvider, prov, pod) + Expect(ExpectNodeClaims(ctx, env.Client)).To(BeEmpty()) + Expect(verdict(pod)).To(BeTrue()) + }) + It("should record a verdict for a pod no instance type is large enough for", func() { + ExpectApplied(ctx, env.Client, nodePool) + pod := test.UnschedulablePod(test.PodOptions{ + ResourceRequirements: corev1.ResourceRequirements{Requests: corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("10000")}}, + }) + ExpectProvisionedNoBinding(ctx, env.Client, cluster, cloudProvider, prov, pod) + Expect(ExpectNodeClaims(ctx, env.Client)).To(BeEmpty()) + Expect(verdict(pod)).To(BeTrue()) + }) + It("should not record a verdict when a second NodePool rejects the pod for a pass-dependent reason", func() { + nodePool.Spec.Template.Spec.Taints = []corev1.Taint{{Key: "dedicated", Value: "gpu", Effect: corev1.TaintEffectNoSchedule}} + limited := test.NodePool(v1.NodePool{Spec: v1.NodePoolSpec{Limits: v1.Limits(corev1.ResourceList{resources.Node: resource.MustParse("0")})}}) + ExpectApplied(ctx, env.Client, nodePool, limited) + pod := test.UnschedulablePod() + ExpectProvisionedNoBinding(ctx, env.Client, cluster, cloudProvider, prov, pod) + Expect(ExpectNodeClaims(ctx, env.Client)).To(BeEmpty()) + Expect(verdict(pod)).To(BeFalse()) + }) + }) + + Context("NodePool limits", func() { + It("should not record a verdict for a pod rejected because the NodePool's node limit is exhausted", func() { + nodePool.Spec.Limits = v1.Limits(corev1.ResourceList{resources.Node: resource.MustParse("0")}) + ExpectApplied(ctx, env.Client, nodePool) + pod := test.UnschedulablePod() + ExpectProvisionedNoBinding(ctx, env.Client, cluster, cloudProvider, prov, pod) + Expect(ExpectNodeClaims(ctx, env.Client)).To(BeEmpty()) + Expect(verdict(pod)).To(BeFalse()) + }) + It("should not record a verdict for a pod every remaining instance type exceeds the NodePool's limits for", func() { + nodePool.Spec.Limits = v1.Limits(corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("1")}) + ExpectApplied(ctx, env.Client, nodePool) + pod := test.UnschedulablePod() + ExpectProvisionedNoBinding(ctx, env.Client, cluster, cloudProvider, prov, pod) + Expect(ExpectNodeClaims(ctx, env.Client)).To(BeEmpty()) + Expect(verdict(pod)).To(BeFalse()) + }) + It("should not record a verdict when limits trimmed the instance types the pod was filtered against", func() { + // The 4-CPU limit leaves only the smallest instance types, none of which fits the pod; with the full set the + // pod would fit, so the filter failure is a consequence of the limit and not of the pod's requirements. + nodePool.Spec.Limits = v1.Limits(corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("4")}) + ExpectApplied(ctx, env.Client, nodePool) + pod := test.UnschedulablePod(test.PodOptions{ + ResourceRequirements: corev1.ResourceRequirements{Requests: corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("8")}}, + }) + ExpectProvisionedNoBinding(ctx, env.Client, cluster, cloudProvider, prov, pod) + Expect(ExpectNodeClaims(ctx, env.Client)).To(BeEmpty()) + Expect(verdict(pod)).To(BeFalse()) + }) + It("should record a verdict for a pod that tolerates no taint of a NodePool at its node limit", func() { + nodePool.Spec.Limits = v1.Limits(corev1.ResourceList{resources.Node: resource.MustParse("0")}) + nodePool.Spec.Template.Spec.Taints = []corev1.Taint{{Key: "dedicated", Value: "gpu", Effect: corev1.TaintEffectNoSchedule}} + ExpectApplied(ctx, env.Client, nodePool) + pod := test.UnschedulablePod() + ExpectProvisionedNoBinding(ctx, env.Client, cluster, cloudProvider, prov, pod) + Expect(ExpectNodeClaims(ctx, env.Client)).To(BeEmpty()) + Expect(verdict(pod)).To(BeTrue()) + }) + It("should record a verdict for a pod whose requirements contradict those of a NodePool at its node limit", func() { + nodePool.Spec.Limits = v1.Limits(corev1.ResourceList{resources.Node: resource.MustParse("0")}) + nodePool.Spec.Template.Spec.Requirements = []v1.NodeSelectorRequirementWithMinValues{{ + Key: v1.CapacityTypeLabelKey, Operator: corev1.NodeSelectorOpIn, Values: []string{v1.CapacityTypeOnDemand}, + }} + ExpectApplied(ctx, env.Client, nodePool) + pod := test.UnschedulablePod(test.PodOptions{NodeSelector: map[string]string{v1.CapacityTypeLabelKey: v1.CapacityTypeSpot}}) + ExpectProvisionedNoBinding(ctx, env.Client, cluster, cloudProvider, prov, pod) + Expect(ExpectNodeClaims(ctx, env.Client)).To(BeEmpty()) + Expect(verdict(pod)).To(BeTrue()) + }) + It("should record a verdict for a pod no instance type of a NodePool at its limits could take", func() { + nodePool.Spec.Limits = v1.Limits(corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("1")}) + ExpectApplied(ctx, env.Client, nodePool) + pod := test.UnschedulablePod(test.PodOptions{NodeSelector: map[string]string{v1.CapacityTypeLabelKey: v1.CapacityTypeReserved}}) + ExpectProvisionedNoBinding(ctx, env.Client, cluster, cloudProvider, prov, pod) + Expect(ExpectNodeClaims(ctx, env.Client)).To(BeEmpty()) + Expect(verdict(pod)).To(BeTrue()) + }) + It("should record a verdict when the instance types limits trimmed away would reject the pod too", func() { + nodePool.Spec.Limits = v1.Limits(corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("4")}) + ExpectApplied(ctx, env.Client, nodePool) + pod := test.UnschedulablePod(test.PodOptions{NodeSelector: map[string]string{v1.CapacityTypeLabelKey: v1.CapacityTypeReserved}}) + ExpectProvisionedNoBinding(ctx, env.Client, cluster, cloudProvider, prov, pod) + Expect(ExpectNodeClaims(ctx, env.Client)).To(BeEmpty()) + Expect(verdict(pod)).To(BeTrue()) + }) + It("should not record a verdict for a topology-constrained pod a NodePool at its limits rejects", func() { + nodePool.Spec.Limits = v1.Limits(corev1.ResourceList{resources.Node: resource.MustParse("0")}) + ExpectApplied(ctx, env.Client, nodePool) + pod := test.UnschedulablePod(test.PodOptions{ + ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{"app": "spread"}}, + ResourceRequirements: corev1.ResourceRequirements{Requests: corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("10000")}}, + TopologySpreadConstraints: []corev1.TopologySpreadConstraint{{ + MaxSkew: 1, + TopologyKey: corev1.LabelTopologyZone, + WhenUnsatisfiable: corev1.DoNotSchedule, + LabelSelector: &metav1.LabelSelector{MatchLabels: map[string]string{"app": "spread"}}, + }}, + }) + ExpectProvisionedNoBinding(ctx, env.Client, cluster, cloudProvider, prov, pod) + Expect(ExpectNodeClaims(ctx, env.Client)).To(BeEmpty()) + Expect(verdict(pod)).To(BeFalse()) + }) + }) + + Context("topology", func() { + It("should not record a verdict for a pod whose affinity no other pod satisfies", func() { + ExpectApplied(ctx, env.Client, nodePool) + pod := test.UnschedulablePod(test.PodOptions{ + PodRequirements: []corev1.PodAffinityTerm{{ + LabelSelector: &metav1.LabelSelector{MatchLabels: map[string]string{"app": "absent"}}, + TopologyKey: corev1.LabelHostname, + }}, + }) + ExpectProvisionedNoBinding(ctx, env.Client, cluster, cloudProvider, prov, pod) + Expect(ExpectNodeClaims(ctx, env.Client)).To(BeEmpty()) + Expect(verdict(pod)).To(BeFalse()) + }) + It("should not record a verdict for a topology-constrained pod no instance type is large enough for", func() { + ExpectApplied(ctx, env.Client, nodePool) + pod := test.UnschedulablePod(test.PodOptions{ + ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{"app": "spread"}}, + ResourceRequirements: corev1.ResourceRequirements{Requests: corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("10000")}}, + TopologySpreadConstraints: []corev1.TopologySpreadConstraint{{ + MaxSkew: 1, + TopologyKey: corev1.LabelTopologyZone, + WhenUnsatisfiable: corev1.DoNotSchedule, + LabelSelector: &metav1.LabelSelector{MatchLabels: map[string]string{"app": "spread"}}, + }}, + }) + ExpectProvisionedNoBinding(ctx, env.Client, cluster, cloudProvider, prov, pod) + Expect(ExpectNodeClaims(ctx, env.Client)).To(BeEmpty()) + Expect(verdict(pod)).To(BeFalse()) + }) + }) + + Context("reserved offerings", func() { + BeforeEach(func() { + nodePool.Spec.Template.Spec.Requirements = []v1.NodeSelectorRequirementWithMinValues{{ + Key: v1.CapacityTypeLabelKey, + Operator: corev1.NodeSelectorOpIn, + Values: []string{v1.CapacityTypeSpot, v1.CapacityTypeOnDemand, v1.CapacityTypeReserved}, + }} + cloudProvider.Reset() + cloudProvider.InstanceTypes = []*cloudprovider.InstanceType{ + fake.NewInstanceType("large-instance-type", fake.WithResources(map[corev1.ResourceName]resource.Quantity{corev1.ResourceCPU: resource.MustParse("6"), corev1.ResourceMemory: resource.MustParse("6Gi")})), + fake.NewInstanceType("medium-instance-type", fake.WithResources(map[corev1.ResourceName]resource.Quantity{corev1.ResourceCPU: resource.MustParse("3"), corev1.ResourceMemory: resource.MustParse("3Gi")})), + fake.NewInstanceType("small-instance-type", fake.WithResources(map[corev1.ResourceName]resource.Quantity{corev1.ResourceCPU: resource.MustParse("2"), corev1.ResourceMemory: resource.MustParse("2Gi")})), + } + for _, it := range cloudProvider.InstanceTypes[1:] { + it.Requirements.Get(v1.CapacityTypeLabelKey).Insert(v1.CapacityTypeReserved) + it.Offerings = append(it.Offerings, &cloudprovider.Offering{ + ReservationCapacity: 1, + Available: true, + Requirements: pscheduling.NewLabelRequirements(map[string]string{ + v1.CapacityTypeLabelKey: v1.CapacityTypeReserved, + corev1.LabelTopologyZone: "test-zone-1", + v1alpha1.LabelReservationID: fmt.Sprintf("r-%s", it.Name), + }), + Price: fake.PriceFromResources(it.Capacity) / 100_000.0, + }) + } + ctx = options.ToContext(ctx, test.Options(test.OptionsFields{FeatureGates: test.FeatureGates{ReservedCapacity: lo.ToPtr(true)}})) + }) + It("should not record a verdict for a pod deferred by reservation contention within the pass", func() { + // Three pods compete for two reservations of capacity one. The strict provisioning mode schedules one pod per + // pass and defers the others with a ReservedOfferingError; the next pass places another of them. + ExpectApplied(ctx, env.Client, nodePool) + pods := lo.Times(3, func(_ int) *corev1.Pod { + return test.UnschedulablePod(test.PodOptions{ + ResourceRequirements: corev1.ResourceRequirements{Requests: corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("1800m")}}, + }) + }) + result := ExpectProvisioned(ctx, env.Client, cluster, cloudProvider, prov, pods...) + Expect(result.Bindings).To(HaveLen(1)) + deferred := lo.Filter(pods, func(p *corev1.Pod, _ int) bool { return result.Get(p) == nil }) + Expect(deferred).To(HaveLen(2)) + for _, p := range deferred { + Expect(verdict(p)).To(BeFalse()) + } + + result = ExpectProvisioned(ctx, env.Client, cluster, cloudProvider, prov, deferred...) + Expect(result.Bindings).To(HaveLen(1)) + }) + }) + + Context("Solve deadline", func() { + placed := func(results scheduling.Results, p *corev1.Pod) bool { + return lo.ContainsBy(results.NewNodeClaims, func(nc *scheduling.NodeClaim) bool { + return lo.ContainsBy(nc.Pods, func(np *corev1.Pod) bool { return np.UID == p.UID }) + }) + } + + It("should not classify a pod as incompatible when the deadline cut the pass before its retry", func() { + // a (popped first, 2 CPU) has a required hostname affinity to b (1 CPU). a fails its first attempt because no b + // exists yet, is requeued, b opens a NodeClaim, and a's retry lands next to b. Cutting the pass between a's + // first failure and its retry leaves a in PodErrors with a topology error, which is not an incompatibility. + ExpectApplied(ctx, env.Client, nodePool) + bLabels := map[string]string{"app": "b"} + b := test.UnschedulablePod(test.PodOptions{ + ObjectMeta: metav1.ObjectMeta{Labels: bLabels}, + ResourceRequirements: corev1.ResourceRequirements{Requests: corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("1")}}, + }) + a := test.UnschedulablePod(test.PodOptions{ + ResourceRequirements: corev1.ResourceRequirements{Requests: corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("2")}}, + PodRequirements: []corev1.PodAffinityTerm{{ + LabelSelector: &metav1.LabelSelector{MatchLabels: bLabels}, + TopologyKey: corev1.LabelHostname, + }}, + }) + ExpectApplied(ctx, env.Client, a, b) + + s, err := prov.NewScheduler(ctx, []*corev1.Pod{a, b}, nil, nil) + Expect(err).ToNot(HaveOccurred()) + results, err := s.Solve(ctx, []*corev1.Pod{a, b}) + Expect(err).ToNot(HaveOccurred()) + Expect(results.PodErrors).To(BeEmpty()) + Expect(placed(results, a)).To(BeTrue()) + Expect(placed(results, b)).To(BeTrue()) + + for n := int64(1); n <= 500; n++ { + s, err := prov.NewScheduler(ctx, []*corev1.Pod{a, b}, nil, nil) + Expect(err).ToNot(HaveOccurred()) + results, err := s.Solve(deadlineAfterErrCalls(ctx, n), []*corev1.Pod{a, b}) + if !errors.Is(err, context.DeadlineExceeded) { + continue + } + if _, aErrored := results.PodErrors[a]; !aErrored || placed(results, a) || !placed(results, b) { + continue + } + Expect(results.PodsIncompatibleWithAllNodePools()).To(BeEmpty()) + return + } + Fail("found no deadline point between a's first failure and its retry") + }) + It("should classify a pod every NodePool is incompatible with even when the deadline cut the pass", func() { + nodePool.Spec.Template.Spec.Taints = []corev1.Taint{{Key: "dedicated", Value: "gpu", Effect: corev1.TaintEffectNoSchedule}} + ExpectApplied(ctx, env.Client, nodePool) + pod := test.UnschedulablePod() + ExpectApplied(ctx, env.Client, pod) + + for n := int64(1); n <= 500; n++ { + s, err := prov.NewScheduler(ctx, []*corev1.Pod{pod}, nil, nil) + Expect(err).ToNot(HaveOccurred()) + results, err := s.Solve(deadlineAfterErrCalls(ctx, n), []*corev1.Pod{pod}) + if !errors.Is(err, context.DeadlineExceeded) { + continue + } + if _, errored := results.PodErrors[pod]; !errored { + continue + } + Expect(results.PodsIncompatibleWithAllNodePools()).To(ConsistOf(pod)) + return + } + Fail("found no deadline point after the pod's first failure") + }) + }) +}) + +// deadlineContext reports context.DeadlineExceeded from Err() once it has been consulted remaining times. Done() is +// inherited and never closes; Solve and trySchedule only poll Err(). +type deadlineContext struct { + context.Context + remaining atomic.Int64 +} + +func deadlineAfterErrCalls(parent context.Context, n int64) *deadlineContext { + c := &deadlineContext{Context: parent} + c.remaining.Store(n) + return c +} + +func (c *deadlineContext) Err() error { + if c.remaining.Add(-1) <= 0 { + return context.DeadlineExceeded + } + return c.Context.Err() +} diff --git a/pkg/controllers/state/cluster.go b/pkg/controllers/state/cluster.go index 413740bf82..f9fd0bc26d 100644 --- a/pkg/controllers/state/cluster.go +++ b/pkg/controllers/state/cluster.go @@ -72,6 +72,7 @@ type Cluster struct { podsSchedulableTimes sync.Map // pod namespaced name -> time when it was first marked as able to fit to a node podHealthyNodePoolScheduledTime sync.Map // pod namespaced name -> time when pod scheduled to a nodePool that has NodeRegistrationHealthy=true, is marked as able to fit to a node podToNodeClaim sync.Map // pod namespaced name -> nodeClaim name + podsUnprovisionableTimes sync.Map // pod namespaced name -> time of the latest provisioning simulation that found every NodePool incompatible with it clusterStateMu sync.RWMutex // Separate mutex as this is called in some places that mu is held // A monotonically increasing timestamp representing the time state of the @@ -122,6 +123,7 @@ func NewCluster(clk clock.Clock, client client.Client, cloudProvider cloudprovid podsSchedulingAttempted: sync.Map{}, podHealthyNodePoolScheduledTime: sync.Map{}, podToNodeClaim: sync.Map{}, + podsUnprovisionableTimes: sync.Map{}, } } @@ -506,6 +508,10 @@ func (c *Cluster) MarkPodSchedulingDecisions(ctx context.Context, podErrors map[ } c.podHealthyNodePoolScheduledTime.Delete(nn) c.podToNodeClaim.Delete(nn) + // A scheduling error on its own is not an unprovisionable verdict: it may hinge on this pass's cluster state + // (NodePool limits, topology, reservation contention). The caller re-marks the pods whose error is + // pass-invariant through MarkPodsUnprovisionable; everything else loses any stale verdict here. + c.podsUnprovisionableTimes.Delete(nn) } for nodePoolName, pods := range npPods { nodePool := &v1.NodePool{} @@ -522,6 +528,7 @@ func (c *Cluster) MarkPodSchedulingDecisions(ctx context.Context, podErrors map[ if podutils.IsScheduled(p) { continue } + c.podsUnprovisionableTimes.Delete(nn) c.podsSchedulableTimes.LoadOrStore(nn, now) _, alreadyExists := c.podsSchedulingAttempted.LoadOrStore(nn, now) // If we already attempted this, we don't need to emit another metric. @@ -548,11 +555,38 @@ func (c *Cluster) MarkPodSchedulingDecisions(ctx context.Context, podErrors map[ func (c *Cluster) UpdatePodToNodeClaimMapping(ncPods map[string][]*corev1.Pod) { for ncName, pods := range ncPods { for _, p := range pods { - c.podToNodeClaim.Store(client.ObjectKeyFromObject(p), ncName) + nn := client.ObjectKeyFromObject(p) + c.podToNodeClaim.Store(nn, ncName) + c.podsUnprovisionableTimes.Delete(nn) } } } +// MarkPodsUnprovisionable records that the provisioner's most recent scheduling simulation rejected +// each pod for a reason that does not depend on cluster state: every NodePool is incompatible with +// the pod's taints, requirements or instance-type needs (scheduling.IsIncompatibleWithAllNodePools). +// Call it after MarkPodSchedulingDecisions for the same pass, which clears the previous verdict of +// every pod that errored. +func (c *Cluster) MarkPodsUnprovisionable(pods []*corev1.Pod) { + now := c.clock.Now() + for _, pod := range pods { + c.podsUnprovisionableTimes.Store(client.ObjectKeyFromObject(pod), now) + } +} + +// PodUnprovisionableTime returns when the provisioner's most recent scheduling simulation last +// rejected the pod for a cluster-state-independent reason (see MarkPodsUnprovisionable), which no +// scheduling pass over the same NodePools can undo. It is zero when the latest simulation placed the +// pod, rejected it for a reason that may not hold in another pass, or never considered it. Unlike +// PodSchedulingDecisionTime, which remembers the first decision, this follows the latest one, so a +// caller can tell a fresh verdict from a stale one. +func (c *Cluster) PodUnprovisionableTime(podKey types.NamespacedName) time.Time { + if val, found := c.podsUnprovisionableTimes.Load(podKey); found { + return val.(time.Time) + } + return time.Time{} +} + // PodSchedulingDecisionTime returns when Karpenter first decided if a pod could schedule a pod in scheduling simulations. // This returns 0, false if Karpenter never made a decision on the pod. func (c *Cluster) PodSchedulingDecisionTime(podKey types.NamespacedName) time.Time { @@ -604,6 +638,7 @@ func (c *Cluster) ClearPodSchedulingMappings(podKey types.NamespacedName) { c.podsSchedulingAttempted.Delete(podKey) c.podHealthyNodePoolScheduledTime.Delete(podKey) c.podToNodeClaim.Delete(podKey) + c.podsUnprovisionableTimes.Delete(podKey) } // MarkUnconsolidated marks the cluster state as being unconsolidated. This should be called in any situation where @@ -665,6 +700,7 @@ func (c *Cluster) Reset() { c.podAcks = sync.Map{} c.podsSchedulingAttempted = sync.Map{} c.podsSchedulableTimes = sync.Map{} + c.podsUnprovisionableTimes = sync.Map{} c.bufferPodCounts = map[string]int{} } diff --git a/pkg/controllers/state/suite_test.go b/pkg/controllers/state/suite_test.go index 2812e470ec..73aeaf4d6b 100644 --- a/pkg/controllers/state/suite_test.go +++ b/pkg/controllers/state/suite_test.go @@ -167,6 +167,84 @@ var _ = Describe("Pod Healthy NodePool", func() { }) }) +var _ = Describe("Pod Unprovisionable Time", func() { + It("should be zero for a pod no simulation has decided on", func() { + pod := test.Pod() + Expect(cluster.PodUnprovisionableTime(client.ObjectKeyFromObject(pod)).IsZero()).To(BeTrue()) + }) + It("should record the time a pod was found incompatible with every NodePool", func() { + pod := test.Pod() + cluster.MarkPodSchedulingDecisions(ctx, map[*corev1.Pod]error{pod: fmt.Errorf("incompatible requirements")}, nil, nil) + cluster.MarkPodsUnprovisionable([]*corev1.Pod{pod}) + Expect(cluster.PodUnprovisionableTime(client.ObjectKeyFromObject(pod))).To(Equal(env.Clock.Now())) + }) + It("should not record a verdict for a scheduling error alone", func() { + pod := test.Pod() + cluster.MarkPodSchedulingDecisions(ctx, map[*corev1.Pod]error{pod: fmt.Errorf("node limits have been exhausted for nodepool")}, nil, nil) + Expect(cluster.PodUnprovisionableTime(client.ObjectKeyFromObject(pod)).IsZero()).To(BeTrue()) + }) + It("should follow the latest verdict rather than the first", func() { + pod := test.Pod() + nn := client.ObjectKeyFromObject(pod) + cluster.MarkPodSchedulingDecisions(ctx, map[*corev1.Pod]error{pod: fmt.Errorf("first")}, nil, nil) + cluster.MarkPodsUnprovisionable([]*corev1.Pod{pod}) + first := cluster.PodUnprovisionableTime(nn) + + env.Clock.Step(time.Minute) + cluster.MarkPodSchedulingDecisions(ctx, map[*corev1.Pod]error{pod: fmt.Errorf("second")}, nil, nil) + cluster.MarkPodsUnprovisionable([]*corev1.Pod{pod}) + Expect(cluster.PodUnprovisionableTime(nn)).To(Equal(first.Add(time.Minute))) + // The first-decision timestamp is unaffected. + Expect(cluster.PodSchedulingDecisionTime(nn)).To(Equal(first)) + }) + It("should clear the verdict once a later pass fails the pod for a reason that may not recur", func() { + pod := test.Pod() + nn := client.ObjectKeyFromObject(pod) + cluster.MarkPodSchedulingDecisions(ctx, map[*corev1.Pod]error{pod: fmt.Errorf("incompatible requirements")}, nil, nil) + cluster.MarkPodsUnprovisionable([]*corev1.Pod{pod}) + Expect(cluster.PodUnprovisionableTime(nn).IsZero()).To(BeFalse()) + + cluster.MarkPodSchedulingDecisions(ctx, map[*corev1.Pod]error{pod: fmt.Errorf("node limits have been exhausted for nodepool")}, nil, nil) + Expect(cluster.PodUnprovisionableTime(nn).IsZero()).To(BeTrue()) + }) + It("should clear the verdict once a simulation places the pod on a NodePool", func() { + pod := test.Pod() + nn := client.ObjectKeyFromObject(pod) + cluster.MarkPodsUnprovisionable([]*corev1.Pod{pod}) + Expect(cluster.PodUnprovisionableTime(nn).IsZero()).To(BeFalse()) + + cluster.MarkPodSchedulingDecisions(ctx, nil, map[string][]*corev1.Pod{nodePool.Name: {pod}}, nil) + Expect(cluster.PodUnprovisionableTime(nn).IsZero()).To(BeTrue()) + }) + It("should clear the verdict once a simulation places the pod on an existing node", func() { + pod := test.Pod() + nn := client.ObjectKeyFromObject(pod) + cluster.MarkPodsUnprovisionable([]*corev1.Pod{pod}) + Expect(cluster.PodUnprovisionableTime(nn).IsZero()).To(BeFalse()) + + cluster.MarkPodSchedulingDecisions(ctx, nil, nil, map[string][]*corev1.Pod{"existing-node": {pod}}) + Expect(cluster.PodUnprovisionableTime(nn).IsZero()).To(BeTrue()) + }) + It("should clear the verdict when the pod is deleted", func() { + pod := test.Pod() + nn := client.ObjectKeyFromObject(pod) + cluster.MarkPodsUnprovisionable([]*corev1.Pod{pod}) + Expect(cluster.PodUnprovisionableTime(nn).IsZero()).To(BeFalse()) + + cluster.DeletePod(nn) + Expect(cluster.PodUnprovisionableTime(nn).IsZero()).To(BeTrue()) + }) + It("should clear the verdict when cluster state is reset", func() { + pod := test.Pod() + nn := client.ObjectKeyFromObject(pod) + cluster.MarkPodsUnprovisionable([]*corev1.Pod{pod}) + Expect(cluster.PodUnprovisionableTime(nn).IsZero()).To(BeFalse()) + + cluster.Reset() + Expect(cluster.PodUnprovisionableTime(nn).IsZero()).To(BeTrue()) + }) +}) + var _ = Describe("Pod Ack", func() { It("should only mark pods as schedulable once", func() { pod := test.Pod() diff --git a/pkg/operator/options/options.go b/pkg/operator/options/options.go index 27c88bc72d..a6b9f0f269 100644 --- a/pkg/operator/options/options.go +++ b/pkg/operator/options/options.go @@ -113,6 +113,7 @@ type Options struct { ConsolidationAttributeReplacements bool ConsolidationSkipUnchangedNegatives bool ConsolidationNegativeCacheTTL time.Duration + DisruptionUnprovisionablePodTTL time.Duration NodeClaimInitializationTimeout time.Duration ODToSpotConsolidation bool topologyCountCacheModeRaw string @@ -167,6 +168,7 @@ func (o *Options) AddFlags(fs *FlagSet) { fs.DurationVar(&o.ConsolidationCandidateTimeout, "consolidation-candidate-timeout", env.WithDefaultDuration("CONSOLIDATION_CANDIDATE_TIMEOUT", 10*time.Second), "The maximum time a single consolidation candidate's scheduling simulation may run before it is abandoned and the walk moves on. The pass timeout bounds discovery in aggregate; this bounds one candidate, so a pass degrades into finding fewer commands rather than none. 0 disables the per-candidate bound.") fs.BoolVar(&o.ConsolidationAttributeReplacements, "consolidation-attribute-replacements", env.WithDefaultBool("CONSOLIDATION_ATTRIBUTE_REPLACEMENTS", true), "Count only the new NodeClaims that host a disrupted pod as a command's replacements, for every disruption method. A disruption simulation also schedules the cluster's pending pods, and the capacity it opens for them would otherwise be launched and waited on by the command, priced against a consolidation candidate, and counted against the replacement bound. Disable to restore the unattributed behavior.") fs.BoolVarWithEnv(&o.ConsolidationSkipUnchangedNegatives, "consolidation-skip-unchanged-negatives", "CONSOLIDATION_SKIP_UNCHANGED_NEGATIVES", false, "When set, a single-node consolidation candidate whose previous simulation ended in a no-op is skipped while its fingerprint - Node and NodeClaim resourceVersions, NodePool generation, reschedulable pod set, and the NodePool's instance type revision - is unchanged and the verdict is younger than consolidation-negative-cache-ttl. Only no-op verdicts are cached, so a stale entry can only delay a node's consolidation, never disrupt one wrongly; the cache is dropped whenever a pass admits a command. Lookup outcomes are counted regardless of this flag, so the hit rate is measurable before skipping is enabled.") + fs.DurationVar(&o.DisruptionUnprovisionablePodTTL, "disruption-unprovisionable-pod-ttl", env.WithDefaultDuration("DISRUPTION_UNPROVISIONABLE_POD_TTL", 2*time.Minute), "How long a pending pod that the provisioner's most recent scheduling simulation found incompatible with every NodePool - by taints, requirements or instance types, independent of cluster state - stays excluded from disruption scheduling simulations. Such a pod cannot open a NodeClaim in any simulation, so simulating it only spends the candidate's budget; pods rejected for reasons a simulation could change (NodePool limits, reserved offerings, topology) are never excluded. The verdict is refreshed by every provisioning pass and cleared as soon as one places the pod or fails it for another reason. The TTL bounds staleness when provisioning stalls: a verdict older than it is ignored and the pod is simulated again. 0 disables the exclusion.") fs.DurationVar(&o.ConsolidationNegativeCacheTTL, "consolidation-negative-cache-ttl", env.WithDefaultDuration("CONSOLIDATION_NEGATIVE_CACHE_TTL", 5*time.Minute), "How long a cached no-op consolidation verdict remains valid. The fingerprint covers the candidate's own inputs; the TTL bounds what it cannot see, chiefly capacity elsewhere in the fleet freeing up: a verdict older than the TTL is never served. A fully no-op pass marks the fleet consolidated for up to five minutes whether or not caching is enabled, so a TTL below that does not make rechecks more frequent - it only tightens which verdicts the recheck may reuse.") fs.BoolVarWithEnv(&o.ODToSpotConsolidation, "od-to-spot-consolidation", "OD_TO_SPOT_CONSOLIDATION", true, "When set, a consolidation candidate running on-demand whose replacement found nothing cheaper is re-evaluated against spot offerings only, restricted to the zones whose spot price beats the candidate. The replacement launch is pinned to spot and those zones, so insufficient spot capacity fails the launch instead of falling back to on-demand. Enabled by default; set to false to opt out.") fs.Float64Var(&o.ConsolidationReplaceMinSavings, "consolidation-replace-min-savings", env.WithDefaultFloat64("CONSOLIDATION_REPLACE_MIN_SAVINGS", 0), "The fraction of the disrupted nodes' price that any consolidation replacement must save before it is accepted, on top of the usual cheaper-than-candidate check. Applies to every replace decision, including spot-to-spot and the split fallback (which uses the larger of this and consolidation-split-min-savings); delete decisions are unaffected. Replacement launches are also restricted to instance types that meet the margin. 0 accepts any cheaper replacement.") @@ -242,6 +244,9 @@ func (o *Options) validateConsolidation() error { } func (o *Options) validateNegativeCache() error { + if o.DisruptionUnprovisionablePodTTL < 0 { + return fmt.Errorf("validating cli flags / env vars, DISRUPTION_UNPROVISIONABLE_POD_TTL must be >= 0, got %s", o.DisruptionUnprovisionablePodTTL) + } if o.ConsolidationNegativeCacheTTL <= 0 && o.ConsolidationSkipUnchangedNegatives { return fmt.Errorf("validating cli flags / env vars, CONSOLIDATION_NEGATIVE_CACHE_TTL must be > 0 when CONSOLIDATION_SKIP_UNCHANGED_NEGATIVES is set, got %s", o.ConsolidationNegativeCacheTTL) } diff --git a/pkg/test/options.go b/pkg/test/options.go index 1c9361d30c..5added553a 100644 --- a/pkg/test/options.go +++ b/pkg/test/options.go @@ -61,6 +61,7 @@ type OptionsFields struct { ConsolidationAttributeReplacements *bool ConsolidationSkipUnchangedNegatives *bool ConsolidationNegativeCacheTTL *time.Duration + DisruptionUnprovisionablePodTTL *time.Duration NodeClaimInitializationTimeout *time.Duration ODToSpotConsolidation *bool TopologyCountCacheMode *options.TopologyCountCacheMode @@ -118,6 +119,7 @@ func Options(overrides ...OptionsFields) *options.Options { ConsolidationAttributeReplacements: lo.FromPtrOr(opts.ConsolidationAttributeReplacements, true), ConsolidationSkipUnchangedNegatives: lo.FromPtrOr(opts.ConsolidationSkipUnchangedNegatives, false), ConsolidationNegativeCacheTTL: lo.FromPtrOr(opts.ConsolidationNegativeCacheTTL, 5*time.Minute), + DisruptionUnprovisionablePodTTL: lo.FromPtrOr(opts.DisruptionUnprovisionablePodTTL, 2*time.Minute), NodeClaimInitializationTimeout: lo.FromPtrOr(opts.NodeClaimInitializationTimeout, 0), ODToSpotConsolidation: lo.FromPtrOr(opts.ODToSpotConsolidation, false), TopologyCountCacheMode: lo.FromPtrOr(opts.TopologyCountCacheMode, options.TopologyCountCacheModeOff),