perf(disruption): exclude pending pods every NodePool is incompatible with from simulations

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

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{

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
+}

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")
+	}
+}