perf(operator): cut reconcile-path allocations on large clusters

operator/cmd/manager/main.go

@@ -175,6 +175,10 @@ func main() {
 		//               apply-to-config stalls in e2e/core that are not yet explained. Do not
 		//               re-scope without reproducing that first.
 		Cache: cache.Options{
+			// Nothing here reads metadata.managedFields, and on Pods and Nodes it is roughly a
+			// third of the serialized object; dropping it before objects enter the cache
+			// shrinks every informer and every DeepCopy made from one.
+			DefaultTransform: cache.TransformStripManagedFields(),
 			ByObject: map[client.Object]cache.ByObject{
 				&batchv1.Job{}: {
 					Namespaces: map[string]cache.Config{

operator/internal/controller/cluster_state_v2.go

@@ -69,6 +69,14 @@ func (t *ObjectTracker) Track(obj client.Object) {
 	}
 }
 
+// Tracked reports whether an original for obj is already held, so callers can skip the
+// DeepCopy that Track would discard anyway.
+func (t *ObjectTracker) Tracked(obj client.Object) bool {
+	key := fmt.Sprintf("%s|%s|%s-%s", obj.GetObjectKind().GroupVersionKind().Kind, obj.GetNamespace(), obj.GetName(), obj.GetUID())
+	_, ok := t.objects[key]
+	return ok
+}
+
 type clusterState struct {
 	tracker  ObjectTracker
 	skyhooks []SkyhookNodes
@@ -94,20 +102,24 @@ func BuildState(skyhooks *v1alpha1.NodeWrightList, nodes *corev1.NodeList, deplo
 			nodes:        make([]wrapper.SkyhookNode, 0),
 			compartments: make(map[string]*wrapper.Compartment),
 		}
+		selector, err := metav1.LabelSelectorAsSelector(&skyhook.Spec.NodeSelector)
+		if err != nil {
+			return nil, err
+		}
 		for _, node := range nodes.Items {
-			skyNode, err := wrapper.NewSkyhookNode(&node, skyhook)
-			if err != nil {
-				return nil, err
+			if !selector.Matches(labels.Set(node.Labels)) { // note: if selector is empty, it selects all
+				continue
 			}
-
-			selector, err := metav1.LabelSelectorAsSelector(&skyhook.Spec.NodeSelector)
+			skyNode, err := wrapper.NewSkyhookNode(&node, skyhook)
 			if err != nil {
 				return nil, err
 			}
-			if selector.Matches(labels.Set(node.Labels)) { // note: if selector is empty, it selects all
+			// One original per Node, however many NodeWrights select it: Track keeps the first
+			// copy and drops later ones, so copying again for each NodeWright is pure garbage.
+			if !ret.tracker.Tracked(&node) {
 				ret.tracker.Track(node.DeepCopy())
-				ret.skyhooks[idx].AddNode(skyNode)
 			}
+			ret.skyhooks[idx].AddNode(skyNode)
 		}
 
 		// find deployment policy and all compartments + the default one

operator/internal/controller/cluster_state_v2_test.go

@@ -35,7 +35,9 @@ import (
 	wrapperMock "github.com/NVIDIA/nodewright/operator/internal/wrapper/mock"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	kptr "k8s.io/utils/ptr"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
 var testLogger = logr.Discard()
@@ -3959,3 +3961,85 @@ var _ = Describe("UpdateDrainBlockedCondition", func() {
 		Expect(drainBlocked(sn)).To(HaveField("Message", "1/1 nodes blocked draining (node-a); default/web-0 on node-a: "+pdbDetail))
 	})
 })
+
+var _ = Describe("BuildState selection and tracking", func() {
+	nodes := func(n int) *corev1.NodeList {
+		list := &corev1.NodeList{Items: make([]corev1.Node, n)}
+		for i := range list.Items {
+			list.Items[i] = corev1.Node{ObjectMeta: metav1.ObjectMeta{
+				Name:   fmt.Sprintf("node-%d", i),
+				UID:    types.UID(fmt.Sprintf("uid-%d", i)),
+				Labels: map[string]string{"pool": "gpu"},
+			}}
+		}
+		return list
+	}
+	noPolicies := &v1alpha1.DeploymentPolicyList{Items: []v1alpha1.DeploymentPolicy{}}
+	cpuOnly := metav1.LabelSelector{MatchLabels: map[string]string{"pool": "cpu"}}
+	// BuildGraph rejects a dependency with an empty version, so this NodeWright can never be wrapped.
+	brokenGraph := v1alpha1.NodeWrightSpec{Packages: v1alpha1.Packages{
+		"a": {PackageRef: v1alpha1.PackageRef{Name: "a", Version: "1"}, DependsOn: map[string]string{"b": ""}},
+	}}
+
+	It("does not build a wrapper, and so does not fail, for a nodewright that selects none of the nodes", func() {
+		skyhooks := &v1alpha1.NodeWrightList{Items: []v1alpha1.NodeWright{
+			{ObjectMeta: metav1.ObjectMeta{Name: "healthy", UID: "h"}},
+			{ObjectMeta: metav1.ObjectMeta{Name: "broken-but-idle", UID: "b"}, Spec: brokenGraph},
+		}}
+		skyhooks.Items[1].Spec.NodeSelector = cpuOnly
+		state, err := BuildState(skyhooks, nodes(3), noPolicies)
+		Expect(err).ToNot(HaveOccurred(), "an unselectable nodewright must not poison the pass for the others")
+		byName := map[string]SkyhookNodes{}
+		for _, sh := range state.skyhooks {
+			byName[sh.GetSkyhook().Name] = sh
+		}
+		Expect(byName["healthy"].GetNodes()).To(HaveLen(3))
+		Expect(byName["broken-but-idle"].GetNodes()).To(BeEmpty())
+	})
+
+	It("still fails for a nodewright whose graph cannot be built and which selects a node", func() {
+		skyhooks := &v1alpha1.NodeWrightList{Items: []v1alpha1.NodeWright{{ObjectMeta: metav1.ObjectMeta{Name: "broken", UID: "b"}, Spec: brokenGraph}}}
+		_, err := BuildState(skyhooks, nodes(1), noPolicies)
+		Expect(err).To(MatchError(ContainSubstring("DependsOn version is empty")))
+	})
+
+	It("fails on an invalid node selector even when there are no nodes", func() {
+		skyhooks := &v1alpha1.NodeWrightList{Items: []v1alpha1.NodeWright{{
+			ObjectMeta: metav1.ObjectMeta{Name: "bad-selector", UID: "s"},
+			Spec:       v1alpha1.NodeWrightSpec{NodeSelector: metav1.LabelSelector{MatchExpressions: []metav1.LabelSelectorRequirement{{Key: "pool", Operator: "Bogus"}}}},
+		}}}
+		_, err := BuildState(skyhooks, &corev1.NodeList{}, noPolicies)
+		Expect(err).To(HaveOccurred())
+	})
+
+	It("keeps one original per node, and that original is a copy the wrappers cannot reach", func() {
+		skyhooks := &v1alpha1.NodeWrightList{Items: []v1alpha1.NodeWright{
+			{ObjectMeta: metav1.ObjectMeta{Name: "a", UID: "a"}},
+			{ObjectMeta: metav1.ObjectMeta{Name: "b", UID: "b"}},
+		}}
+		state, err := BuildState(skyhooks, nodes(2), noPolicies)
+		Expect(err).ToNot(HaveOccurred())
+		// Track deduplicates by key, so this holds before and after the copy-once change; it pins
+		// the invariant that change relies on: the tracker only ever wants one copy per node.
+		Expect(state.tracker.objects).To(HaveLen(4))
+		live := state.skyhooks[0].GetNodes()[0].GetNode()
+		orig := state.tracker.GetOriginal(live)
+		Expect(orig).ToNot(BeIdenticalTo(live))
+		live.Labels["pool"] = "changed"
+		Expect(orig.(*corev1.Node).Labels["pool"]).To(Equal("gpu"))
+	})
+})
+
+var _ = Describe("ObjectTracker", func() {
+	It("reports tracked objects and keeps the first original", func() {
+		t := ObjectTracker{objects: make(map[string]client.Object)}
+		first := &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "n", UID: "u"}}
+		Expect(t.Tracked(first)).To(BeFalse())
+		t.Track(first)
+		Expect(t.Tracked(first)).To(BeTrue())
+		second := &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "n", UID: "u", Labels: map[string]string{"x": "y"}}}
+		Expect(t.Tracked(second)).To(BeTrue(), "same kind/namespace/name/uid is the same object")
+		t.Track(second)
+		Expect(t.GetOriginal(second)).To(BeIdenticalTo(first))
+	})
+})

operator/internal/controller/event_handler.go

@@ -111,44 +111,41 @@ func (h *globalDelayHandler) relevant(ctx context.Context, object client.Object)
 	case *v1alpha1.NodeWright:
 		return true, nil
 	case *corev1.Node:
-		list, err := h.dal.GetSkyhooks(ctx)
+		selectors, err := h.skyhookSelectors(ctx)
 		if err != nil {
-			return false, fmt.Errorf("listing nodewrights for node relevance check: %w", err)
+			return false, err
 		}
-		if list == nil {
-			return false, nil
+		nodeLabels := labels.Set(obj.Labels)
+		for _, selector := range selectors {
+			if selector.Matches(nodeLabels) {
+				return true, nil
+			}
 		}
-		return len(matchSelectors(list, obj.Labels)) > 0, nil
+		return false, nil
 	default:
 		return false, nil
 	}
 }
 
-// matchSelectors returns the Skyhooks whose node selector matches the given
-// labels.
-func matchSelectors(crs *v1alpha1.NodeWrightList, lbs map[string]string) []types.NamespacedName {
-
-	ret := make([]types.NamespacedName, 0)
-
-	for _, cr := range crs.Items {
-		match := false
-
-		selector, err := metav1.LabelSelectorAsSelector(&cr.Spec.NodeSelector)
+// skyhookSelectors reads the NodeWrights without the cache's deep copy (their status carries
+// per-node state) and lets only the parsed selectors out; the objects have no TypeMeta on
+// this path and must not escape. An unparseable selector matches every node, as admission
+// should have rejected it and waking the reconcile is the safe side.
+func (h *globalDelayHandler) skyhookSelectors(ctx context.Context) ([]labels.Selector, error) {
+	list, err := h.dal.GetSkyhooks(ctx, client.UnsafeDisableDeepCopy)
+	if err != nil {
+		return nil, fmt.Errorf("listing nodewrights for node relevance check: %w", err)
+	}
+	if list == nil {
+		return nil, nil
+	}
+	selectors := make([]labels.Selector, 0, len(list.Items))
+	for i := range list.Items {
+		selector, err := metav1.LabelSelectorAsSelector(&list.Items[i].Spec.NodeSelector)
 		if err != nil {
-			match = true
-		}
-
-		if selector.Matches(labels.Set(lbs)) {
-			match = true
-		}
-
-		if match {
-			ret = append(ret, types.NamespacedName{
-				Name:      cr.Name,
-				Namespace: cr.Namespace,
-			})
+			selector = labels.Everything()
 		}
+		selectors = append(selectors, selector)
 	}
-
-	return ret
+	return selectors, nil
 }

operator/internal/controller/event_handler_test.go

@@ -29,6 +29,7 @@ import (
 	"github.com/NVIDIA/nodewright/operator/internal/mocks/workqueue"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 )
@@ -54,7 +55,7 @@ var _ = Describe("Global delay handler", func() {
 
 		node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "foonode", Labels: matchingLabels}}
 
-		dalMock.EXPECT().GetSkyhooks(ctx).Return(skyhookList, nil).Times(4)
+		dalMock.EXPECT().GetSkyhooks(ctx, client.UnsafeDisableDeepCopy).Return(skyhookList, nil).Times(4)
 		queue.EXPECT().AddAfter(globalReconcileKey, delay).Times(4)
 
 		handler.Create(ctx, event.CreateEvent{Object: node}, queue)
@@ -72,7 +73,7 @@ var _ = Describe("Global delay handler", func() {
 		node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "foonode", Labels: map[string]string{"no": "match"}}}
 
 		// no AddAfter expectation: the mock fails the test if AddAfter is called.
-		dalMock.EXPECT().GetSkyhooks(ctx).Return(skyhookList, nil).Once()
+		dalMock.EXPECT().GetSkyhooks(ctx, client.UnsafeDisableDeepCopy).Return(skyhookList, nil).Once()
 
 		handler.Create(ctx, event.CreateEvent{Object: node}, queue)
 	})
@@ -103,6 +104,26 @@ var _ = Describe("Global delay handler", func() {
 		handler.Create(ctx, event.CreateEvent{Object: pod}, queue)
 	})
 
+	It("treats a skyhook whose selector cannot be parsed as matching every node", func() {
+
+		dalMock := &dalmock.MockDAL{}
+		queue := workqueue.NewTypedRateLimitingInterface[reconcile.Request](GinkgoT())
+		handler := &globalDelayHandler{logger: GinkgoLogr, dal: dalMock, delay: delay}
+
+		badSelector := &v1alpha1.NodeWrightList{Items: []v1alpha1.NodeWright{{
+			ObjectMeta: metav1.ObjectMeta{Name: "bad"},
+			Spec: v1alpha1.NodeWrightSpec{NodeSelector: metav1.LabelSelector{
+				MatchExpressions: []metav1.LabelSelectorRequirement{{Key: "pool", Operator: "Bogus"}},
+			}},
+		}}}
+		node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "foonode", Labels: map[string]string{"no": "match"}}}
+
+		dalMock.EXPECT().GetSkyhooks(ctx, client.UnsafeDisableDeepCopy).Return(badSelector, nil).Once()
+		queue.EXPECT().AddAfter(globalReconcileKey, delay).Once()
+
+		handler.Create(ctx, event.CreateEvent{Object: node}, queue)
+	})
+
 	It("does not enqueue when listing skyhooks errors", func() {
 
 		dalMock := &dalmock.MockDAL{}
@@ -112,7 +133,7 @@ var _ = Describe("Global delay handler", func() {
 		node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "foonode", Labels: matchingLabels}}
 
 		// no AddAfter expectation: a list error must not enqueue.
-		dalMock.EXPECT().GetSkyhooks(ctx).Return(nil, errors.New("boom")).Once()
+		dalMock.EXPECT().GetSkyhooks(ctx, client.UnsafeDisableDeepCopy).Return(nil, errors.New("boom")).Once()
 
 		handler.Create(ctx, event.CreateEvent{Object: node}, queue)
 	})