perf: Attempt to improve SimulateScheduling efficiency

Makefile

@@ -139,6 +139,12 @@ e2etest-dra: ## Run DRA e2e integration tests
 benchmark: ## Run benchmark tests for node overlay store
 	go test -bench=. -benchmem ./pkg/controllers/nodeoverlay/... -run=^$$
 
+benchmark-cow: ## Run copy-on-write state/disruption benchmarks (25/100/400 nodes -- fast, excludes 5000-node variants)
+	go test -tags=test_performance -bench=. -benchmem -run=^$$ ./pkg/controllers/state/... ./pkg/controllers/disruption/...
+
+benchmark-cow-5k: ## Run the long-running 5000-node copy-on-write benchmarks (several minutes; not part of benchmark-cow)
+	go test -tags=test_performance_5000 -bench=. -benchmem -run=^$$ ./pkg/controllers/state/... ./pkg/controllers/disruption/...
+
 deflake: ## Run randomized, racing tests until the test fails to catch flakes
 	go tool -modfile=go.tools.mod ginkgo \
 		--race \
@@ -187,4 +193,4 @@ download: ## Recursively "go mod download" on all directories where go.mod exist
 gen_instance_types:
 	go run kwok/tools/gen_instance_types.go > kwok/cloudprovider/instance_types.json
 
-.PHONY: help presubmit install-kwok uninstall-kwok build apply delete test test-memory test-dra e2etest-dra benchmark deflake vulncheck licenses verify download gen_instance_types setup-kind-dra delete-kind-dra apply-with-kind-dra
+.PHONY: help presubmit install-kwok uninstall-kwok build apply delete test test-memory test-dra e2etest-dra benchmark benchmark-cow benchmark-cow-5k deflake vulncheck licenses verify download gen_instance_types setup-kind-dra delete-kind-dra apply-with-kind-dra

pkg/controllers/disruption/concurrent_simulation_test.go

@@ -0,0 +1,107 @@
+/*
+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 (
+	"sync"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	v1 "sigs.k8s.io/karpenter/pkg/apis/v1"
+	"sigs.k8s.io/karpenter/pkg/controllers/disruption"
+	"sigs.k8s.io/karpenter/pkg/test"
+	. "sigs.k8s.io/karpenter/pkg/test/expectations"
+	"sigs.k8s.io/karpenter/pkg/utils/pdb"
+)
+
+// This spec exists specifically because of the SimulateScheduling deep-copy reduction: ExistingNode used to mutate
+// its embedded *state.StateNode directly, which meant cluster state could never safely be read concurrently by
+// multiple simulations. Now that ExistingNode clones its own usage trackers, and Cluster.Snapshot() memoizes a
+// pointer-slice behind a generation counter, many goroutines can call SimulateScheduling concurrently -- each
+// internally calling cluster.Snapshot(), which is itself exercised concurrently by this test (some goroutines hit
+// the cache, some race to rebuild it). Run with `-race` to verify both layers.
+var _ = Describe("Concurrent SimulateScheduling", func() {
+	It("should not race when multiple goroutines simulate scheduling against one shared snapshot", func() {
+		nodePool := test.NodePool(v1.NodePool{
+			Spec: v1.NodePoolSpec{
+				Disruption: v1.Disruption{
+					ConsolidateAfter:    v1.MustParseNillableDuration("0s"),
+					ConsolidationPolicy: v1.ConsolidationPolicyWhenEmptyOrUnderutilized,
+				},
+			},
+		})
+		ExpectApplied(ctx, env.Client, nodePool)
+
+		const numCandidates = 8
+		nodeClaims, nodes := test.NodeClaimsAndNodes(numCandidates, v1.NodeClaim{
+			ObjectMeta: metav1.ObjectMeta{
+				Labels: map[string]string{
+					v1.NodePoolLabelKey:            nodePool.Name,
+					corev1.LabelInstanceTypeStable: leastExpensiveInstance.Name,
+					v1.CapacityTypeLabelKey:        leastExpensiveOffering.Requirements.Get(v1.CapacityTypeLabelKey).Any(),
+					corev1.LabelTopologyZone:       leastExpensiveOffering.Requirements.Get(corev1.LabelTopologyZone).Any(),
+				},
+			},
+			Status: v1.NodeClaimStatus{
+				Allocatable: corev1.ResourceList{
+					corev1.ResourceCPU:  resource.MustParse("32"),
+					corev1.ResourcePods: resource.MustParse("100"),
+				},
+			},
+		})
+		for i := range numCandidates {
+			ExpectApplied(ctx, env.Client, nodeClaims[i], nodes[i])
+		}
+		ExpectMakeNodesAndNodeClaimsInitializedAndStateUpdated(ctx, env.Client, env.Clock, nodeStateController, nodeClaimStateController, nodes, nodeClaims)
+
+		pdbs, err := pdb.NewLimits(ctx, env.Client)
+		Expect(err).To(Succeed())
+		nodePoolMap, nodePoolToInstanceTypesMap, err := disruption.BuildNodePoolMap(ctx, env.Client, cloudProvider)
+		Expect(err).To(Succeed())
+
+		candidates := make([]*disruption.Candidate, numCandidates)
+		for i := range numCandidates {
+			stateNode := ExpectStateNodeExists(cluster, nodes[i])
+			c, err := disruption.NewCandidate(ctx, env.Client, recorder, env.Clock, stateNode, pdbs, nodePoolMap, nodePoolToInstanceTypesMap, queue, disruption.GracefulDisruptionClass)
+			Expect(err).To(Succeed())
+			candidates[i] = c
+		}
+
+		// Every goroutine calls SimulateScheduling concurrently; each internally calls cluster.Snapshot(), so this
+		// exercises concurrent readers of the generation-counter cache (some will hit it, some will race to
+		// rebuild it) without any of them mutating the nodes the others are relying on.
+		var wg sync.WaitGroup
+		errs := make([]error, numCandidates)
+		for i := range numCandidates {
+			wg.Add(1)
+			go func(idx int) {
+				defer wg.Done()
+				_, err := disruption.SimulateScheduling(ctx, env.Client, cluster, prov, env.Clock, recorder, nil, candidates[idx])
+				errs[idx] = err
+			}(i)
+		}
+		wg.Wait()
+
+		for i, err := range errs {
+			Expect(err).To(Succeed(), "candidate %d", i)
+		}
+	})
+})

pkg/controllers/disruption/consolidation.go

@@ -21,6 +21,7 @@ import (
 	"errors"
 	"fmt"
 	"sort"
+	"strings"
 	"time"
 
 	"github.com/samber/lo"
@@ -37,6 +38,7 @@ import (
 	pscheduling "sigs.k8s.io/karpenter/pkg/controllers/provisioning/scheduling"
 	"sigs.k8s.io/karpenter/pkg/controllers/state"
 	"sigs.k8s.io/karpenter/pkg/events"
+	"sigs.k8s.io/karpenter/pkg/metrics"
 	"sigs.k8s.io/karpenter/pkg/operator/options"
 	"sigs.k8s.io/karpenter/pkg/scheduling"
 )
@@ -156,10 +158,16 @@ func (c *consolidation) sortCandidates(_ context.Context, candidates []*Candidat
 // computeConsolidation computes a consolidation action to take
 //
 // nolint:gocyclo
-func (c *consolidation) computeConsolidation(ctx context.Context, candidates ...*Candidate) (Command, error) {
+func (c *consolidation) computeConsolidation(ctx context.Context, consolidationType string, candidates ...*Candidate) (Command, error) {
 	var err error
 	// Run scheduling simulation to compute consolidation option
+	stop := metrics.Measure(CandidateEvaluationDurationSeconds, map[string]string{
+		metrics.ReasonLabel:    strings.ToLower(string(v1.DisruptionReasonUnderutilized)),
+		ConsolidationTypeLabel: consolidationType,
+		StageLabel:             StageEvaluate,
+	})
 	results, err := SimulateScheduling(ctx, c.kubeClient, c.cluster, c.provisioner, c.clock, c.recorder, []pscheduling.Options{pscheduling.IsConsolidationSimulation}, candidates...)
+	stop()
 	if err != nil {
 		// if a candidate node is now deleting, just retry
 		if errors.Is(err, errCandidateDeleting) {

pkg/controllers/disruption/controller.go

@@ -147,7 +147,7 @@ func (c *Controller) Reconcile(ctx context.Context) (reconciler.Result, error) {
 	// Karpenter taints nodes with a karpenter.sh/disruption taint as part of the disruption process while it progresses in memory.
 	// If Karpenter restarts or fails with an error during a disruption action, some nodes can be left tainted.
 	// Idempotently remove this taint from candidates that are not in the orchestration queue before continuing.
-	outdatedNodes := lo.Reject(c.cluster.DeepCopyNodes(), func(s *state.StateNode, _ int) bool {
+	outdatedNodes := lo.Reject(c.cluster.Snapshot(), func(s *state.StateNode, _ int) bool {
 		return c.queue.HasAny(s.ProviderID()) || s.MarkedForDeletion()
 	})
 	if err := state.RequireNoScheduleTaint(ctx, c.kubeClient, false, outdatedNodes...); err != nil {

pkg/controllers/disruption/drift.go

@@ -21,6 +21,7 @@ import (
 	"errors"
 	"slices"
 	"sort"
+	"strings"
 
 	"github.com/samber/lo"
 	"k8s.io/utils/clock"
@@ -33,6 +34,7 @@ import (
 	"sigs.k8s.io/karpenter/pkg/controllers/provisioning"
 	"sigs.k8s.io/karpenter/pkg/controllers/state"
 	"sigs.k8s.io/karpenter/pkg/events"
+	"sigs.k8s.io/karpenter/pkg/metrics"
 )
 
 // Drift is a subreconciler that deletes drifted candidates.
@@ -81,7 +83,13 @@ func (d *Drift) ComputeCommands(ctx context.Context, disruptionBudgetMapping map
 			continue
 		}
 		// Check if we need to create any NodeClaims.
+		stop := metrics.Measure(CandidateEvaluationDurationSeconds, map[string]string{
+			metrics.ReasonLabel:    strings.ToLower(string(d.Reason())),
+			ConsolidationTypeLabel: d.ConsolidationType(),
+			StageLabel:             StageEvaluate,
+		})
 		results, err := SimulateScheduling(ctx, d.kubeClient, d.cluster, d.provisioner, d.clock, d.recorder, nil, candidate)
+		stop()
 		if err != nil {
 			// if a candidate is now deleting, just retry
 			if errors.Is(err, errCandidateDeleting) {

pkg/controllers/disruption/helpers.go

@@ -54,7 +54,10 @@ func SimulateScheduling(ctx context.Context, kubeClient client.Client, cluster *
 	schedulerOpts []scheduling.Options, candidates ...*Candidate,
 ) (scheduling.Results, error) {
 	candidateNames := sets.NewString(lo.Map(candidates, func(t *Candidate, i int) string { return t.Name() })...)
-	nodes := cluster.DeepCopyNodes()
+	// Snapshot is cheap now (see Cluster.Snapshot's doc comment) -- every call gets a point-in-time view that's
+	// always current as of this instant, so there's no benefit to threading a shared snapshot through callers
+	// anymore.
+	nodes := cluster.Snapshot()
 	deletingNodes := nodes.Deleting()
 	stateNodes := lo.Filter(nodes.Active(), func(n *state.StateNode, _ int) bool {
 		return !candidateNames.Has(n.Name())
@@ -204,7 +207,7 @@ func GetCandidatesWithTotals(ctx context.Context, cluster *state.Cluster, kubeCl
 	if err != nil {
 		return nil, nil, fmt.Errorf("tracking PodDisruptionBudgets, %w", err)
 	}
-	allNodes := cluster.DeepCopyNodes()
+	allNodes := cluster.Snapshot()
 	allCandidates := lo.FilterMap(allNodes, func(n *state.StateNode, _ int) (*Candidate, bool) {
 		cn, e := NewCandidate(ctx, kubeClient, recorder, clk, n, pdbs, nodePoolMap, nodePoolToInstanceTypesMap, queue, disruptionClass)
 		return cn, e == nil
@@ -263,7 +266,9 @@ func BuildDisruptionBudgetMapping(ctx context.Context, cluster *state.Cluster, c
 	disruptionBudgetMapping := map[string]int{}
 	numNodes := map[string]int{}   // map[nodepool] -> node count in nodepool
 	disrupting := map[string]int{} // map[nodepool] -> nodes undergoing disruption
-	for _, node := range cluster.DeepCopyNodes() {
+	// This loop is read-only (only counts nodes), so it's safe to iterate the live cluster state directly
+	// instead of taking a deep-copy.
+	for node := range cluster.Nodes() {
 		// We only consider nodes that we own and are initialized towards the total.
 		// If a node is launched/registered, but not initialized, pods aren't scheduled
 		// to the node, and these are treated as unhealthy until they're cleaned up.

pkg/controllers/disruption/metrics.go

@@ -30,6 +30,9 @@ const (
 	ConsolidationTypeLabel       = "consolidation_type"
 	CandidatesIneligible         = "candidates_ineligible"
 	policyLabel                  = "policy"
+	StageLabel                   = "stage"
+	StageEvaluate                = "evaluate"
+	StageValidate                = "validate"
 )
 
 func init() {
@@ -49,6 +52,17 @@ var (
 		},
 		[]string{metrics.ReasonLabel, ConsolidationTypeLabel},
 	)
+	CandidateEvaluationDurationSeconds = opmetrics.NewPrometheusHistogram(
+		crmetrics.Registry,
+		prometheus.HistogramOpts{
+			Namespace: metrics.Namespace,
+			Subsystem: voluntaryDisruptionSubsystem,
+			Name:      "candidate_evaluation_duration_seconds",
+			Help:      "Duration of a single SimulateScheduling call for one candidate or candidate batch. Labeled by disruption reason, consolidation type, and evaluation stage.",
+			Buckets:   metrics.DurationBuckets(),
+		},
+		[]string{metrics.ReasonLabel, ConsolidationTypeLabel, StageLabel},
+	)
 	DecisionsPerformedTotal = opmetrics.NewPrometheusCounter(
 		crmetrics.Registry,
 		prometheus.CounterOpts{

pkg/controllers/disruption/multinodeconsolidation.go

@@ -138,7 +138,7 @@ func (m *MultiNodeConsolidation) firstNConsolidationOption(ctx context.Context,
 		candidatesToConsolidate := candidates[0 : mid+1]
 
 		// Pass the timeout context to ensure sub-operations can be canceled
-		cmd, err := m.computeConsolidation(timeoutCtx, candidatesToConsolidate...)
+		cmd, err := m.computeConsolidation(timeoutCtx, m.ConsolidationType(), candidatesToConsolidate...)
 		// context deadline exceeded will return to the top of the loop and either return nothing or the last saved command
 		if err != nil {
 			if errors.Is(err, context.DeadlineExceeded) {

pkg/controllers/disruption/simulatescheduling_5k_benchmark_test.go

@@ -0,0 +1,39 @@
+//go:build test_performance_5000
+
+/*
+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 "testing"
+
+// 5000-node variants of simulatescheduling_benchmark_test.go's benchmarks, split into their own build-tagged
+// file. These are the slowest benchmarks in the whole copy-on-write suite (envtest-backed, ~830s for the full
+// disruption package run) -- excluded from the routine `make benchmark-cow` target and only runnable via the
+// separate `make benchmark-cow-5k` target. See simulatescheduling_benchmark_test.go (build tag
+// test_performance || test_performance_5000) for setupSimulateSchedulingBenchFixture and the shared
+// benchmarkSimulateScheduling/benchmarkClusterDeepCopyNodes/benchmarkClusterNodesIterate helpers, which are
+// available here too since that file's tag includes test_performance_5000.
+//
+// Run with:
+//
+//	KUBEBUILDER_ASSETS=<path> go test -tags=test_performance_5000 -run=XXX -bench=. ./pkg/controllers/disruption/...
+
+func BenchmarkSimulateScheduling_5000(b *testing.B) { benchmarkSimulateScheduling(b, 5000) }
+
+func BenchmarkClusterDeepCopyNodes_5000(b *testing.B) { benchmarkClusterDeepCopyNodes(b, 5000) }
+
+func BenchmarkClusterNodesIterate_5000(b *testing.B) { benchmarkClusterNodesIterate(b, 5000) }