feat(fault-quarantine): replace GET + PUT with PATCH

Signed-off-by: Ajay Mishra <ajmishra@nvidia.com>

Author: XRFXLP

commons/pkg/kubeclient/nodepatch.go

@@ -17,25 +17,25 @@ package kubeclient
 import (
 	"encoding/json"
 	"fmt"
+	"reflect"
 
 	v1 "k8s.io/api/core/v1"
 )
 
-// NodeMergePatch builds an RFC 7386 JSON merge patch carrying the label and
-// annotation differences between original and modified. It returns a nil patch when
-// the two already agree, so callers can skip the write instead of spending an API
-// call on a no-op.
+// NodeMergePatch builds an RFC 7386 JSON merge patch carrying differences in labels,
+// annotations, taints, and unschedulable state. It returns a nil patch when the two
+// nodes already agree, so callers can skip a no-op write.
 //
 // The patch is assembled key by key rather than by marshalling modified, because
 // marshalling a Node emits every populated field. A caller that read original from an
 // informer cache holding a projected Node would then patch the projection's gaps back
 // over the real object. Emitting only keys that differ means fields missing from both
 // sides are left untouched.
 //
-// Spec fields such as taints and unschedulable are deliberately out of scope: a merge
-// patch replaces a list wholesale, so patching taints from a projected Node whose Spec
-// had been cleared would silently drop every taint on the real object.
+// Taints are emitted only when the caller changed them. A projected Node whose Spec
+// is empty on both sides therefore cannot erase taints from the real object.
 func NodeMergePatch(original, modified *v1.Node) ([]byte, error) {
+	root := map[string]any{}
 	metadata := map[string]any{}
 
 	if labels := stringMapMergePatch(original.Labels, modified.Labels); labels != nil {
@@ -46,18 +46,49 @@ func NodeMergePatch(original, modified *v1.Node) ([]byte, error) {
 		metadata["annotations"] = annotations
 	}
 
-	if len(metadata) == 0 {
+	if len(metadata) > 0 {
+		root["metadata"] = metadata
+	}
+
+	spec := map[string]any{}
+	if !taintsEqual(original.Spec.Taints, modified.Spec.Taints) {
+		spec["taints"] = modified.Spec.Taints
+	}
+
+	if original.Spec.Unschedulable != modified.Spec.Unschedulable {
+		spec["unschedulable"] = modified.Spec.Unschedulable
+	}
+
+	if len(spec) > 0 {
+		root["spec"] = spec
+	}
+
+	if len(root) == 0 {
 		return nil, nil
 	}
 
-	patch, err := json.Marshal(map[string]any{"metadata": metadata})
+	patch, err := json.Marshal(root)
 	if err != nil {
 		return nil, fmt.Errorf("marshal merge patch for node %s: %w", original.Name, err)
 	}
 
 	return patch, nil
 }
 
+func taintsEqual(original, modified []v1.Taint) bool {
+	if len(original) != len(modified) {
+		return false
+	}
+
+	for idx := range original {
+		if !reflect.DeepEqual(original[idx], modified[idx]) {
+			return false
+		}
+	}
+
+	return true
+}
+
 // stringMapMergePatch returns the merge patch entries that turn original into
 // modified: added and changed keys map to their new value, removed keys map to nil so
 // the API server deletes them. It returns nil when the two maps already agree.

commons/pkg/kubeclient/nodepatch_test.go

@@ -132,7 +132,7 @@ func TestNodeMergePatch_ProjectedFields_LeavesThemAlone(t *testing.T) {
 		"a cleared Spec must never reach the patch, or real taints would be dropped")
 }
 
-func TestNodeMergePatch_SpecChanges_ReturnsNoPatch(t *testing.T) {
+func TestNodeMergePatch_SpecChanges_ReturnsExpectedPatch(t *testing.T) {
 	original := node(nil, nil)
 	modified := original.DeepCopy()
 	modified.Spec.Unschedulable = true
@@ -141,5 +141,12 @@ func TestNodeMergePatch_SpecChanges_ReturnsNoPatch(t *testing.T) {
 	patch, err := NodeMergePatch(original, modified)
 	require.NoError(t, err)
 
-	assert.Nil(t, patch, "spec is out of scope until a caller needs it")
+	assert.JSONEq(t,
+		`{"spec":{"taints":[{"key":"held","effect":"NoSchedule"}],"unschedulable":true}}`,
+		string(patch),
+	)
+
+	patch, err = NodeMergePatch(modified, original)
+	require.NoError(t, err)
+	assert.JSONEq(t, `{"spec":{"taints":null,"unschedulable":false}}`, string(patch))
 }

fault-quarantine/pkg/informer/k8s_client.go

@@ -27,6 +27,7 @@ import (
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
@@ -57,6 +58,7 @@ type FaultQuarantineClient struct {
 	cordonedReasonLabelKey   string
 	uncordonedReasonLabelKey string
 	operationMutex           sync.Map // map[string]*sync.Mutex for per-node locking
+	lastWrittenNodeVersion   sync.Map // map[string]string
 }
 
 // NewFaultQuarantineClient constructs a FaultQuarantineClient using the
@@ -171,26 +173,82 @@ func (c *FaultQuarantineClient) UpdateNode(ctx context.Context, nodeName string,
 	}
 
 	return retry.OnError(backoff, isRetryableError, func() error {
-		node, err := c.Clientset.CoreV1().Nodes().Get(ctx, nodeName, metav1.GetOptions{})
+		current, err := c.nodeForPatch(ctx, nodeName)
 		if err != nil {
 			return err
 		}
 
-		if err := updateFn(node); err != nil {
+		desired := current.DeepCopy()
+		if err := updateFn(desired); err != nil {
 			return err
 		}
 
-		_, err = c.Clientset.CoreV1().Nodes().Update(ctx, node, metav1.UpdateOptions{})
+		patch, err := kubeclient.NodeMergePatch(current, desired)
 		if err != nil {
 			return err
 		}
 
-		slog.Debug("Updated node", "node", nodeName)
+		if patch == nil {
+			return nil
+		}
+
+		updated, err := c.Clientset.CoreV1().Nodes().Patch(
+			ctx,
+			nodeName,
+			types.MergePatchType,
+			patch,
+			metav1.PatchOptions{},
+		)
+		if err != nil {
+			return err
+		}
+
+		c.lastWrittenNodeVersion.Store(nodeName, updated.ResourceVersion)
+		slog.Debug("Patched node", "node", nodeName)
 
 		return nil
 	})
 }
 
+func (c *FaultQuarantineClient) nodeForPatch(ctx context.Context, nodeName string) (*v1.Node, error) {
+	if node, ready := c.cachedNodeForPatch(nodeName); ready {
+		return node, nil
+	}
+
+	// The informer may not be running in tests or may not have observed our previous
+	// write yet. A live read preserves consecutive updates to the same node.
+	node, err := c.Clientset.CoreV1().Nodes().Get(ctx, nodeName, metav1.GetOptions{})
+	if err != nil {
+		return nil, err
+	}
+
+	c.lastWrittenNodeVersion.Delete(nodeName)
+
+	return node, nil
+}
+
+func (c *FaultQuarantineClient) cachedNodeForPatch(nodeName string) (*v1.Node, bool) {
+	if c.NodeInformer == nil || !c.NodeInformer.HasSynced() {
+		return nil, false
+	}
+
+	node, err := c.NodeInformer.GetNode(nodeName)
+	if err != nil {
+		return nil, false
+	}
+
+	lastWrittenVersion, pending := c.lastWrittenNodeVersion.Load(nodeName)
+	if pending && node.ResourceVersion != lastWrittenVersion.(string) {
+		return nil, false
+	}
+
+	if pending {
+		c.lastWrittenNodeVersion.Delete(nodeName)
+	}
+
+	return node.DeepCopy(), true
+}
+
 func isRetryableError(err error) bool {
 	if errors.IsConflict(err) {
 		return true
@@ -376,29 +434,25 @@ func (c *FaultQuarantineClient) applyTaints(
 		return nil
 	}
 
-	existingTaints := make(map[config.Taint]v1.Taint)
+	existingTaints := make(map[config.Taint]struct{})
 	for _, taint := range node.Spec.Taints {
-		existingTaints[config.Taint{Key: taint.Key, Value: taint.Value, Effect: string(taint.Effect)}] = taint
+		existingTaints[config.Taint{Key: taint.Key, Value: taint.Value, Effect: string(taint.Effect)}] = struct{}{}
 	}
 
 	for _, taintConfig := range taints {
 		key := config.Taint{Key: taintConfig.Key, Value: taintConfig.Value, Effect: string(taintConfig.Effect)}
 
 		if _, exists := existingTaints[key]; !exists {
 			slog.InfoContext(ctx, "Tainting node", "node", nodename, "taintConfig", taintConfig)
-			existingTaints[key] = v1.Taint{
+			node.Spec.Taints = append(node.Spec.Taints, v1.Taint{
 				Key:    taintConfig.Key,
 				Value:  taintConfig.Value,
 				Effect: v1.TaintEffect(taintConfig.Effect),
-			}
+			})
+			existingTaints[key] = struct{}{}
 		}
 	}
 
-	node.Spec.Taints = []v1.Taint{}
-	for _, taint := range existingTaints {
-		node.Spec.Taints = append(node.Spec.Taints, taint)
-	}
-
 	return nil
 }
 

fault-quarantine/pkg/informer/k8s_client_patch_test.go

@@ -0,0 +1,108 @@
+// Copyright (c) 2026, NVIDIA CORPORATION.  All rights reserved.
+//
+// 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 informer
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/kubernetes/fake"
+	corelisters "k8s.io/client-go/listers/core/v1"
+	k8stesting "k8s.io/client-go/testing"
+	"k8s.io/client-go/tools/cache"
+)
+
+func cachedNodeInformer(t *testing.T, node *v1.Node) *NodeInformer {
+	t.Helper()
+
+	indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{})
+	require.NoError(t, indexer.Add(node))
+
+	return &NodeInformer{
+		lister:         corelisters.NewNodeLister(indexer),
+		informerSynced: func() bool { return true },
+	}
+}
+
+func TestUpdateNode_CachedNode_UsesPatchAndSkipsNoOp(t *testing.T) {
+	node := &v1.Node{ObjectMeta: metav1.ObjectMeta{Name: "node-1", ResourceVersion: "1"}}
+	clientset := fake.NewSimpleClientset(node.DeepCopy())
+	client := &FaultQuarantineClient{
+		Clientset:    clientset,
+		NodeInformer: cachedNodeInformer(t, node.DeepCopy()),
+	}
+
+	clientset.ClearActions()
+	_, err := client.QuarantineNodeAndSetAnnotations(
+		context.Background(),
+		node.Name,
+		nil,
+		true,
+		nil,
+		nil,
+	)
+	require.NoError(t, err)
+
+	actions := clientset.Actions()
+	require.Len(t, actions, 1)
+	patchAction, ok := actions[0].(k8stesting.PatchAction)
+	require.True(t, ok)
+	assert.Equal(t, types.MergePatchType, patchAction.GetPatchType())
+	assert.JSONEq(t, `{"spec":{"unschedulable":true}}`, string(patchAction.GetPatch()))
+
+	updated, err := clientset.CoreV1().Nodes().Get(t.Context(), node.Name, metav1.GetOptions{})
+	require.NoError(t, err)
+	assert.True(t, updated.Spec.Unschedulable)
+
+	client.NodeInformer = cachedNodeInformer(t, updated.DeepCopy())
+	clientset.ClearActions()
+	_, err = client.QuarantineNodeAndSetAnnotations(
+		context.Background(),
+		node.Name,
+		nil,
+		true,
+		nil,
+		nil,
+	)
+	require.NoError(t, err)
+	assert.Empty(t, clientset.Actions())
+}
+
+func TestNodeForPatch_PreviousWriteNotInCache_FallsBackToLiveGet(t *testing.T) {
+	cached := &v1.Node{ObjectMeta: metav1.ObjectMeta{Name: "node-1", ResourceVersion: "1"}}
+	live := cached.DeepCopy()
+	live.ResourceVersion = "2"
+	live.Annotations = map[string]string{"first-write": "preserved"}
+
+	clientset := fake.NewSimpleClientset(live)
+	client := &FaultQuarantineClient{
+		Clientset:    clientset,
+		NodeInformer: cachedNodeInformer(t, cached),
+	}
+	client.lastWrittenNodeVersion.Store(cached.Name, live.ResourceVersion)
+	clientset.ClearActions()
+
+	node, err := client.nodeForPatch(t.Context(), cached.Name)
+	require.NoError(t, err)
+	assert.Equal(t, live.ResourceVersion, node.ResourceVersion)
+	assert.Equal(t, "preserved", node.Annotations["first-write"])
+	require.Len(t, clientset.Actions(), 1)
+	assert.Equal(t, "get", clientset.Actions()[0].GetVerb())
+}

fault-quarantine/pkg/informer/k8s_client_test.go

@@ -172,6 +172,10 @@ func measureCordonThroughput(t *testing.T, prefix string, nodeCount int, qps flo
 	)
 	require.NoError(t, err)
 
+	stopCh := make(chan struct{})
+	t.Cleanup(func() { close(stopCh) })
+	require.NoError(t, client.NodeInformer.Run(stopCh))
+
 	start := time.Now()
 	for _, nodeName := range nodeNames {
 		_, err := client.QuarantineNodeAndSetAnnotations(ctx, nodeName, nil, true, nil, nil)
@@ -182,7 +186,7 @@ func measureCordonThroughput(t *testing.T, prefix string, nodeCount int, qps flo
 }
 
 // TestQuarantineNodeAndSetAnnotations_QPSControlledCordonThroughput_HigherQPSIncreasesThroughput
-// exercises FQ's real GET+UPDATE cordon path against envtest.
+// exercises FQ's cache-read + PATCH cordon path against envtest.
 func TestQuarantineNodeAndSetAnnotations_QPSControlledCordonThroughput_HigherQPSIncreasesThroughput(t *testing.T) {
 	const (
 		nodeCount = 10