Merge branch 'main' into xrfxlp/1597

Author: XRFXLP

commons/pkg/kubeclient/nodepatch.go

@@ -19,6 +19,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"reflect"
 	"sync"
 	"time"
 
@@ -50,7 +51,15 @@ func (p *NodePatcher) Patch(
 	cached *v1.Node,
 	mutate func(*v1.Node) error,
 ) (bool, error) {
-	current, err := p.currentNode(ctx, nodes, nodeName, cached)
+	var current *v1.Node
+
+	err := retry.OnError(nodePatchBackoff(), isRetryableNodePatchError, func() error {
+		var err error
+
+		current, err = p.currentNode(ctx, nodes, nodeName, cached)
+
+		return err
+	})
 	if err != nil {
 		return false, err
 	}
@@ -85,11 +94,13 @@ func (p *NodePatcher) Patch(
 		if errors.IsConflict(err) {
 			patchErr := err
 
-			current, err = nodes.Get(ctx, nodeName, metav1.GetOptions{})
+			refreshed, err := nodes.Get(ctx, nodeName, metav1.GetOptions{})
 			if err != nil {
 				return fmt.Errorf("refresh node %q after patch conflict: %w", nodeName, err)
 			}
 
+			current = refreshed
+
 			return fmt.Errorf("patch node %q: %w", nodeName, patchErr)
 		}
 
@@ -134,8 +145,6 @@ func (p *NodePatcher) currentNode(
 		return nil, fmt.Errorf("refresh node %q while pending write is not in cache: %w", nodeName, err)
 	}
 
-	p.pendingVersions.CompareAndDelete(nodeName, writtenVersionValue)
-
 	return current, nil
 }
 
@@ -156,40 +165,36 @@ func isRetryableNodePatchError(err error) bool {
 		errors.IsServiceUnavailable(err)
 }
 
-// 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.
 //
-// CreateTwoWayMergePatch compares metadata-only projections of the two Nodes.
-// Excluding every other field from both inputs ensures an informer projection cannot
-// patch its gaps back over the live object.
+// CreateTwoWayMergePatch compares projections containing only the fields this helper
+// supports. Excluding every other field from both inputs ensures an informer
+// projection cannot patch its gaps back over the live object.
 //
-// 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) {
-	originalMetadata := &v1.Node{
-		ObjectMeta: metav1.ObjectMeta{
-			Labels:      original.Labels,
-			Annotations: original.Annotations,
-		},
-	}
-	modifiedMetadata := &v1.Node{
-		ObjectMeta: metav1.ObjectMeta{
-			Labels:      modified.Labels,
-			Annotations: modified.Annotations,
-		},
+	originalProjection := projectNodePatchableFields(original)
+	modifiedProjection := projectNodePatchableFields(modified)
+
+	specChanged := !reflect.DeepEqual(original.Spec.Taints, modified.Spec.Taints) ||
+		original.Spec.Unschedulable != modified.Spec.Unschedulable
+	if specChanged {
+		// Lists in spec, such as taints, are replaced wholesale. ResourceVersion
+		// prevents a stale list from overwriting a concurrent update.
+		modifiedProjection.ResourceVersion = original.ResourceVersion
 	}
 
-	originalJSON, err := json.Marshal(originalMetadata)
+	originalJSON, err := json.Marshal(originalProjection)
 	if err != nil {
-		return nil, fmt.Errorf("marshal original metadata for node %q: %w", original.Name, err)
+		return nil, fmt.Errorf("marshal original patch projection for node %q: %w", original.Name, err)
 	}
 
-	modifiedJSON, err := json.Marshal(modifiedMetadata)
+	modifiedJSON, err := json.Marshal(modifiedProjection)
 	if err != nil {
-		return nil, fmt.Errorf("marshal modified metadata for node %q: %w", original.Name, err)
+		return nil, fmt.Errorf("marshal modified patch projection for node %q: %w", original.Name, err)
 	}
 
 	patch, err := strategicpatch.CreateTwoWayMergePatch(originalJSON, modifiedJSON, v1.Node{})
@@ -203,3 +208,18 @@ func NodeMergePatch(original, modified *v1.Node) ([]byte, error) {
 
 	return patch, nil
 }
+
+// projectNodePatchableFields restricts patch generation to the Node fields this
+// helper intentionally supports, preventing callbacks from patching unrelated fields.
+func projectNodePatchableFields(node *v1.Node) *v1.Node {
+	return &v1.Node{
+		ObjectMeta: metav1.ObjectMeta{
+			Labels:      node.Labels,
+			Annotations: node.Annotations,
+		},
+		Spec: v1.NodeSpec{
+			Taints:        node.Spec.Taints,
+			Unschedulable: node.Spec.Unschedulable,
+		},
+	}
+}

commons/pkg/kubeclient/nodepatch_test.go

@@ -79,8 +79,8 @@ func TestNodePatcher_CachedNode_UsesPatchAndSkipsNoOp(t *testing.T) {
 	assert.Empty(t, clientset.Actions())
 }
 
-func TestNodePatcher_PreviousWriteNotInCache_ReadsLiveNode(t *testing.T) {
-	current := node(map[string]string{"a": "1"}, nil)
+func TestNodePatcher_NoOpBetweenWrites_KeepsReadingLiveNode(t *testing.T) {
+	current := node(nil, map[string]string{"events": "base"})
 	clientset := fake.NewSimpleClientset(current.DeepCopy())
 	var patcher NodePatcher
 
@@ -90,7 +90,7 @@ func TestNodePatcher_PreviousWriteNotInCache_ReadsLiveNode(t *testing.T) {
 		current.Name,
 		current,
 		func(node *v1.Node) error {
-			node.Labels["b"] = "2"
+			node.Annotations["events"] += "|first"
 			return nil
 		},
 	)
@@ -111,12 +111,29 @@ func TestNodePatcher_PreviousWriteNotInCache_ReadsLiveNode(t *testing.T) {
 	require.Len(t, clientset.Actions(), 1)
 	assert.Equal(t, "get", clientset.Actions()[0].GetVerb())
 
+	clientset.ClearActions()
+	changed, err = patcher.Patch(
+		context.Background(),
+		clientset.CoreV1().Nodes(),
+		current.Name,
+		stale,
+		func(node *v1.Node) error {
+			node.Annotations["events"] += "|second"
+			return nil
+		},
+	)
+	require.NoError(t, err)
+	assert.True(t, changed)
+	require.Len(t, clientset.Actions(), 2)
+	assert.Equal(t, "get", clientset.Actions()[0].GetVerb())
+	assert.Equal(t, "patch", clientset.Actions()[1].GetVerb())
+
 	updated, err := clientset.CoreV1().Nodes().Get(t.Context(), current.Name, metav1.GetOptions{})
 	require.NoError(t, err)
-	assert.Equal(t, "2", updated.Labels["b"])
+	assert.Equal(t, "base|first|second", updated.Annotations["events"])
 }
 
-func TestNodePatcher_LiveReadFailure_PreservesPendingVersion(t *testing.T) {
+func TestNodePatcher_LiveReadRetriesTransientFailure(t *testing.T) {
 	current := node(map[string]string{"a": "1"}, nil)
 	clientset := fake.NewSimpleClientset(current.DeepCopy())
 	var patcher NodePatcher
@@ -128,22 +145,12 @@ func TestNodePatcher_LiveReadFailure_PreservesPendingVersion(t *testing.T) {
 	clientset.PrependReactor("get", "nodes", func(k8stesting.Action) (bool, runtime.Object, error) {
 		getAttempts++
 		if getAttempts == 1 {
-			return true, nil, assert.AnError
+			return true, nil, apierrors.NewTooManyRequests("try again", 0)
 		}
 
 		return false, nil, nil
 	})
 
-	_, err := patcher.Patch(
-		context.Background(),
-		clientset.CoreV1().Nodes(),
-		current.Name,
-		stale,
-		func(*v1.Node) error { return nil },
-	)
-	require.ErrorIs(t, err, assert.AnError)
-	assert.ErrorContains(t, err, `refresh node "node-1" while pending write is not in cache`)
-
 	changed, err := patcher.Patch(
 		context.Background(),
 		clientset.CoreV1().Nodes(),
@@ -198,12 +205,34 @@ func TestNodePatcher_Conflict_RefreshesLiveNodeBeforeRetry(t *testing.T) {
 	assert.Equal(t, "true", updated.Labels["desired"])
 }
 
-func TestNodeMergePatch_MetadataChanges_ReturnsExpectedPatch(t *testing.T) {
+func TestNodeMergePatch_ReturnsExpectedPatch(t *testing.T) {
+	resourceVersionOriginal := node(nil, nil)
+	resourceVersionOriginal.ResourceVersion = "42"
+	resourceVersionModified := resourceVersionOriginal.DeepCopy()
+	resourceVersionModified.Spec.Unschedulable = true
+
+	projected := &v1.Node{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:            "node-1",
+			ResourceVersion: "1",
+			Labels:          map[string]string{"gpu": "true"},
+			Annotations:     map[string]string{"kept": "yes"},
+		},
+	}
+	projectedModified := projected.DeepCopy()
+	projectedModified.Labels["driver.installed"] = "true"
+
+	specOriginal := node(nil, nil)
+	specModified := specOriginal.DeepCopy()
+	specModified.Spec.Unschedulable = true
+	specModified.Spec.Taints = []v1.Taint{{Key: "held", Effect: v1.TaintEffectNoSchedule}}
+
 	tests := []struct {
 		name     string
 		original *v1.Node
 		modified *v1.Node
 		expected string
+		excluded []string
 	}{
 		{
 			name:     "no change produces no patch",
@@ -253,6 +282,31 @@ func TestNodeMergePatch_MetadataChanges_ReturnsExpectedPatch(t *testing.T) {
 			modified: node(map[string]string{"a": "1"}, nil),
 			expected: `{"metadata":{"labels":{"a":"1"}}}`,
 		},
+		{
+			name:     "spec change includes original resource version",
+			original: resourceVersionOriginal,
+			modified: resourceVersionModified,
+			expected: `{"metadata":{"resourceVersion":"42"},"spec":{"unschedulable":true}}`,
+		},
+		{
+			name:     "projected fields remain absent",
+			original: projected,
+			modified: projectedModified,
+			expected: `{"metadata":{"labels":{"driver.installed":"true"}}}`,
+			excluded: []string{"annotations", "spec"},
+		},
+		{
+			name:     "sets spec fields",
+			original: specOriginal,
+			modified: specModified,
+			expected: `{"metadata":{"resourceVersion":"1"},"spec":{"taints":[{"key":"held","effect":"NoSchedule"}],"unschedulable":true}}`,
+		},
+		{
+			name:     "clears spec fields",
+			original: specModified,
+			modified: specOriginal,
+			expected: `{"metadata":{"resourceVersion":"1"},"spec":{"taints":null,"unschedulable":null}}`,
+		},
 	}
 
 	for _, tt := range tests {
@@ -266,49 +320,9 @@ func TestNodeMergePatch_MetadataChanges_ReturnsExpectedPatch(t *testing.T) {
 			}
 
 			assert.JSONEq(t, tt.expected, string(patch))
+			for _, field := range tt.excluded {
+				assert.NotContains(t, string(patch), field)
+			}
 		})
 	}
 }
-
-// TestNodeMergePatchLeavesProjectedFieldsAlone pins the reason the patch is built key
-// by key. Informer caches often hold a projected Node — the labeler's transform keeps
-// only one annotation and clears Spec entirely — and a patch derived from that
-// projection must not describe the fields the projection dropped, or it would erase
-// them on the real object.
-func TestNodeMergePatch_ProjectedFields_LeavesThemAlone(t *testing.T) {
-	projected := &v1.Node{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:            "node-1",
-			ResourceVersion: "1",
-			Labels:          map[string]string{"gpu": "true"},
-			Annotations:     map[string]string{"kept": "yes"},
-		},
-	}
-
-	modified := projected.DeepCopy()
-	modified.Labels["driver.installed"] = "true"
-
-	patch, err := NodeMergePatch(projected, modified)
-	require.NoError(t, err)
-
-	assert.JSONEq(t,
-		`{"metadata":{"labels":{"driver.installed":"true"}}}`,
-		string(patch),
-	)
-	assert.NotContains(t, string(patch), "annotations",
-		"an untouched annotation must not appear in the patch")
-	assert.NotContains(t, string(patch), "spec",
-		"a cleared Spec must never reach the patch, or real taints would be dropped")
-}
-
-func TestNodeMergePatch_SpecChanges_ReturnsNoPatch(t *testing.T) {
-	original := node(nil, nil)
-	modified := original.DeepCopy()
-	modified.Spec.Unschedulable = true
-	modified.Spec.Taints = []v1.Taint{{Key: "held", Effect: v1.TaintEffectNoSchedule}}
-
-	patch, err := NodeMergePatch(original, modified)
-	require.NoError(t, err)
-
-	assert.Nil(t, patch, "spec is out of scope until a caller needs it")
-}