collapse patch conflict retry onto GuaranteedUpdate

staging/src/k8s.io/apiextensions-apiserver/test/integration/basic_test.go

@@ -582,17 +582,6 @@ func TestPatch(t *testing.T) {
 		t.Fatalf("unexpected error: %v", err)
 	}
 
-	// this call waits for the resourceVersion to be reached in the cache before returning.
-	// We need to do this because the patch gets its initial object from the storage, and the cache serves that.
-	// If it is out of date, then our initial patch is applied to an old resource version, which conflicts
-	// and then the updated object shows a conflicting diff, which permanently fails the patch.
-	// This gives expected stability in the patch without retrying on an known number of conflicts below in the test.
-	// See https://issue.k8s.io/42644
-	_, err = noxuNamespacedResourceClient.Get("foo", metav1.GetOptions{ResourceVersion: createdNoxuInstance.GetResourceVersion()})
-	if err != nil {
-		t.Fatalf("unexpected error: %v", err)
-	}
-
 	// a patch with no change
 	createdNoxuInstance, err = noxuNamespacedResourceClient.Patch("foo", types.MergePatchType, patch)
 	if err != nil {

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/rest.go

@@ -90,9 +90,6 @@ func (scope *RequestScope) AllowsStreamSchema(s string) bool {
 	return s == "watch"
 }
 
-// MaxRetryWhenPatchConflicts is the maximum number of conflicts retry during a patch operation before returning failure
-const MaxRetryWhenPatchConflicts = 5
-
 // ConnectResource returns a function that handles a connect request on a rest.Storage object.
 func ConnectResource(connecter rest.Connecter, scope RequestScope, admit admission.Interface, restPath string, isSubresource bool) http.HandlerFunc {
 	return func(w http.ResponseWriter, req *http.Request) {

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/rest_test.go

@@ -173,32 +173,46 @@ func (p *testPatcher) New() runtime.Object {
 }
 
 func (p *testPatcher) Update(ctx context.Context, name string, objInfo rest.UpdatedObjectInfo, createValidation rest.ValidateObjectFunc, updateValidation rest.ValidateObjectUpdateFunc) (runtime.Object, bool, error) {
-	currentPod := p.startingPod
-	if p.numUpdates > 0 {
-		currentPod = p.updatePod
-	}
-	p.numUpdates++
+	// Simulate GuaranteedUpdate behavior (retries internally on etcd changes if the incoming resource doesn't pin resourceVersion)
+	for {
+		currentPod := p.startingPod
+		if p.numUpdates > 0 {
+			currentPod = p.updatePod
+		}
+		p.numUpdates++
 
-	obj, err := objInfo.UpdatedObject(ctx, currentPod)
-	if err != nil {
-		return nil, false, err
-	}
-	inPod := obj.(*example.Pod)
-	if inPod.ResourceVersion != p.updatePod.ResourceVersion {
-		return nil, false, apierrors.NewConflict(example.Resource("pods"), inPod.Name, fmt.Errorf("existing %v, new %v", p.updatePod.ResourceVersion, inPod.ResourceVersion))
-	}
+		// Remember the current resource version
+		currentResourceVersion := currentPod.ResourceVersion
 
-	if currentPod == nil {
-		if err := createValidation(currentPod); err != nil {
+		obj, err := objInfo.UpdatedObject(ctx, currentPod)
+		if err != nil {
 			return nil, false, err
 		}
-	} else {
-		if err := updateValidation(currentPod, inPod); err != nil {
-			return nil, false, err
+		inPod := obj.(*example.Pod)
+		if inPod.ResourceVersion == "" || inPod.ResourceVersion == "0" {
+			inPod.ResourceVersion = p.updatePod.ResourceVersion
+		}
+		if inPod.ResourceVersion != p.updatePod.ResourceVersion {
+			// If the patch didn't have an opinion on the resource version, retry like GuaranteedUpdate does
+			if inPod.ResourceVersion == currentResourceVersion {
+				continue
+			}
+			// If the patch changed the resource version and it mismatches, conflict
+			return nil, false, apierrors.NewConflict(example.Resource("pods"), inPod.Name, fmt.Errorf("existing %v, new %v", p.updatePod.ResourceVersion, inPod.ResourceVersion))
+		}
+
+		if currentPod == nil {
+			if err := createValidation(currentPod); err != nil {
+				return nil, false, err
+			}
+		} else {
+			if err := updateValidation(currentPod, inPod); err != nil {
+				return nil, false, err
+			}
 		}
-	}
 
-	return inPod, false, nil
+		return inPod, false, nil
+	}
 }
 
 func (p *testPatcher) Get(ctx context.Context, name string, options *metav1.GetOptions) (runtime.Object, error) {
@@ -549,8 +563,7 @@ func TestPatchResourceWithConflict(t *testing.T) {
 		startingPod: &example.Pod{},
 		changedPod:  &example.Pod{},
 		updatePod:   &example.Pod{},
-
-		expectedError: `Operation cannot be fulfilled on pods.example.apiserver.k8s.io "foo": existing 2, new 1`,
+		expectedPod: &example.Pod{},
 	}
 
 	// See issue #63104 for discussion of how much sense this makes.
@@ -576,6 +589,13 @@ func TestPatchResourceWithConflict(t *testing.T) {
 	tc.updatePod.APIVersion = examplev1.SchemeGroupVersion.String()
 	tc.updatePod.Spec.NodeName = "anywhere"
 
+	tc.expectedPod.Name = name
+	tc.expectedPod.Namespace = namespace
+	tc.expectedPod.UID = uid
+	tc.expectedPod.ResourceVersion = "2"
+	tc.expectedPod.APIVersion = examplev1.SchemeGroupVersion.String()
+	tc.expectedPod.Spec.NodeName = "there"
+
 	tc.Run(t)
 }
 

test/integration/apiserver/BUILD

@@ -43,7 +43,6 @@ go_test(
         "//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
-        "//vendor/k8s.io/apiserver/pkg/endpoints/handlers:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/features:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/util/feature:go_default_library",
         "//vendor/k8s.io/client-go/kubernetes:go_default_library",

test/integration/apiserver/patch_test.go

@@ -29,11 +29,10 @@ import (
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/apiserver/pkg/endpoints/handlers"
 	"k8s.io/kubernetes/test/integration/framework"
 )
 
-// Tests that the apiserver retries non-overlapping conflicts on patches
+// Tests that the apiserver retries patches
 func TestPatchConflicts(t *testing.T) {
 	s, clientSet, closeFn := setup(t)
 	defer closeFn()
@@ -41,7 +40,7 @@ func TestPatchConflicts(t *testing.T) {
 	ns := framework.CreateTestingNamespace("status-code", s, t)
 	defer framework.DeleteTestingNamespace(ns, s, t)
 
-	numOfConcurrentPatches := 2 * handlers.MaxRetryWhenPatchConflicts
+	numOfConcurrentPatches := 100
 
 	UIDs := make([]types.UID, numOfConcurrentPatches)
 	ownerRefs := []metav1.OwnerReference{}
@@ -69,10 +68,8 @@ func TestPatchConflicts(t *testing.T) {
 
 	successes := int32(0)
 
-	// Run a lot of simultaneous patch operations to exercise internal API server retry of patch application.
-	// Internally, a patch API call retries up to MaxRetryWhenPatchConflicts times if the resource version of the object has changed.
-	// If the resource version of the object changed between attempts, that means another one of our patch requests succeeded.
-	// That means if we run 2*MaxRetryWhenPatchConflicts patch attempts, we should see at least MaxRetryWhenPatchConflicts succeed.
+	// Run a lot of simultaneous patch operations to exercise internal API server retry of application of patches that do not specify resourceVersion.
+	// They should all succeed.
 	wg := sync.WaitGroup{}
 	for i := 0; i < numOfConcurrentPatches; i++ {
 		wg.Add(1)
@@ -123,8 +120,8 @@ func TestPatchConflicts(t *testing.T) {
 	}
 	wg.Wait()
 
-	if successes < handlers.MaxRetryWhenPatchConflicts {
-		t.Errorf("Expected at least %d successful patches for %s, got %d", handlers.MaxRetryWhenPatchConflicts, "secrets", successes)
+	if successes < int32(numOfConcurrentPatches) {
+		t.Errorf("Expected at least %d successful patches for %s, got %d", numOfConcurrentPatches, "secrets", successes)
 	} else {
 		t.Logf("Got %d successful patches for %s", successes, "secrets")
 	}

test/integration/client/BUILD

@@ -22,7 +22,6 @@ go_test(
         "//test/utils/image:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/api/errors:go_default_library",
-        "//vendor/k8s.io/apimachinery/pkg/api/meta:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1/unstructured:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/fields:go_default_library",

test/integration/client/client_test.go

@@ -27,7 +27,6 @@ import (
 
 	"k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
-	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/labels"
@@ -286,23 +285,6 @@ func TestPatch(t *testing.T) {
 			t.Logf("%v", string(jsonObj))
 		}
 
-		obj, err := result.Get()
-		if err != nil {
-			t.Fatal(err)
-		}
-		metadata, err := meta.Accessor(obj)
-		if err != nil {
-			t.Fatal(err)
-		}
-		// this call waits for the resourceVersion to be reached in the cache before returning.  We need to do this because
-		// the patch gets its initial object from the storage, and the cache serves that.  If it is out of date,
-		// then our initial patch is applied to an old resource version, which conflicts and then the updated object shows
-		// a conflicting diff, which permanently fails the patch.  This gives expected stability in the patch without
-		// retrying on an known number of conflicts below in the test.
-		if _, err := c.Core().Pods(ns.Name).Get(name, metav1.GetOptions{ResourceVersion: metadata.GetResourceVersion()}); err != nil {
-			t.Fatal(err)
-		}
-
 		return nil
 	}