🐛 Fakeclient: Allow concurrent patching if RV is unset or null

pkg/client/fake/client.go

@@ -468,6 +468,11 @@ func (t versionedTracker) updateObject(gvr schema.GroupVersionResource, obj runt
 		switch {
 		case allowsUnconditionalUpdate(gvk):
 			accessor.SetResourceVersion(oldAccessor.GetResourceVersion())
+			// This is needed because if the patch explicitly sets the RV to null, the client-go reaction we use
+			// to apply it and whose output we process here will have it unset. It is not clear why the Kubernetes
+			// apiserver accepts such a patch, but it does so we just copy that behavior.
+			// Kubernetes apiserver behavior can be checked like this:
+			// `kubectl patch configmap foo --patch '{"metadata":{"annotations":{"foo":"bar"},"resourceVersion":null}}' -v=9`
 		case bytes.
 			Contains(debug.Stack(), []byte("sigs.k8s.io/controller-runtime/pkg/client/fake.(*fakeClient).Patch")):
 			// We apply patches using a client-go reaction that ends up calling the trackers Update. As we can't change
@@ -904,6 +909,16 @@ func (c *fakeClient) patch(obj client.Object, patch client.Patch, opts ...client
 		return err
 	}
 
+	// retryOnConflict unless the patch includes a resourceVersion. We have to
+	// compare the resource version of newObj with oldObject because newObj is
+	// oldObj + the patch. It is important to note that the RV in it might be
+	// different from the RV in obj, as oldObj is fetched from the tracker and
+	// a concurrent actor could have modified it.
+	retryOnConflict := true
+	if newObj.GetResourceVersion() != oldAccessor.GetResourceVersion() {
+		retryOnConflict = false
+	}
+
 	// Validate that deletionTimestamp has not been changed
 	if !deletionTimestampEqual(newObj, oldAccessor) {
 		return fmt.Errorf("rejected patch, metadata.deletionTimestamp immutable")
@@ -912,6 +927,9 @@ func (c *fakeClient) patch(obj client.Object, patch client.Patch, opts ...client
 	reaction := testing.ObjectReaction(c.tracker)
 	handled, o, err := reaction(action)
 	if err != nil {
+		if retryOnConflict && apierrors.IsConflict(err) {
+			return c.patch(obj, patch, opts...)
+		}
 		return err
 	}
 	if !handled {

pkg/client/fake/client_test.go

@@ -21,6 +21,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"strconv"
+	"sync"
 	"time"
 
 	"github.com/google/go-cmp/cmp"
@@ -580,7 +581,7 @@ var _ = Describe("Fake client", func() {
 			Expect(obj.ObjectMeta.ResourceVersion).To(Equal("1000"))
 		})
 
-		It("should allow patch with non-set ResourceVersion for a resource that doesn't allow unconditional updates", func() {
+		It("should allow patch when the patch sets RV to 'null'", func() {
 			schemeBuilder := &scheme.Builder{GroupVersion: schema.GroupVersion{Group: "test", Version: "v1"}}
 			schemeBuilder.Register(&WithPointerMeta{}, &WithPointerMetaList{})
 
@@ -605,6 +606,7 @@ var _ = Describe("Fake client", func() {
 						"foo": "bar",
 					},
 				}}
+
 			Expect(cl.Patch(context.Background(), newObj, client.MergeFrom(original))).To(Succeed())
 
 			patched := &WithPointerMeta{}
@@ -2134,6 +2136,61 @@ var _ = Describe("Fake client", func() {
 		Expect(apierrors.IsNotFound(err)).To(BeTrue())
 	})
 
+	It("should allow concurrent patches to a configMap", func() {
+		scheme := runtime.NewScheme()
+		Expect(corev1.AddToScheme(scheme)).To(Succeed())
+
+		obj := &corev1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:            "foo",
+				ResourceVersion: "0",
+			},
+		}
+		cl := NewClientBuilder().WithScheme(scheme).WithObjects(obj).Build()
+		wg := sync.WaitGroup{}
+		wg.Add(5)
+
+		for i := range 5 {
+			go func() {
+				defer wg.Done()
+				defer GinkgoRecover()
+
+				newObj := obj.DeepCopy()
+				newObj.Data = map[string]string{"foo": strconv.Itoa(i)}
+				Expect(cl.Patch(context.Background(), newObj, client.MergeFrom(obj))).To(Succeed())
+			}()
+		}
+		wg.Wait()
+	})
+
+	It("should not allow concurrent patches to a configMap if the patch contains a ResourceVersion", func() {
+		scheme := runtime.NewScheme()
+		Expect(corev1.AddToScheme(scheme)).To(Succeed())
+
+		obj := &corev1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:            "foo",
+				ResourceVersion: "0",
+			},
+		}
+		cl := NewClientBuilder().WithScheme(scheme).WithObjects(obj).Build()
+		wg := sync.WaitGroup{}
+		wg.Add(5)
+
+		for i := range 5 {
+			go func() {
+				defer wg.Done()
+				defer GinkgoRecover()
+
+				newObj := obj.DeepCopy()
+				newObj.ResourceVersion = "1" // include an invalid RV to cause a conflict
+				newObj.Data = map[string]string{"foo": strconv.Itoa(i)}
+				Expect(apierrors.IsConflict(cl.Patch(context.Background(), newObj, client.MergeFrom(obj)))).To(BeTrue())
+			}()
+		}
+		wg.Wait()
+	})
+
 	It("disallows scale subresources on unsupported built-in types", func() {
 		scheme := runtime.NewScheme()
 		Expect(corev1.AddToScheme(scheme)).To(Succeed())
@@ -2288,8 +2345,8 @@ func (t *WithPointerMetaList) DeepCopyObject() runtime.Object {
 }
 
 type WithPointerMeta struct {
-	*metav1.TypeMeta
-	*metav1.ObjectMeta
+	*metav1.TypeMeta   `json:",inline"`
+	*metav1.ObjectMeta `json:"metadata,omitempty"`
 }
 
 func (t *WithPointerMeta) DeepCopy() *WithPointerMeta {