Skip to content

K
k3s
Unverified Commit 6aad80cc authored by Kubernetes Submit Queue's avatar Kubernetes Submit Queue Committed by GitHub
Browse files
Options

Merge pull request #63146 from liggitt/remove-patch-retry

Automatic merge from submit-queue. If you want to cherry-pick this change to another branch, please follow the instructions <a href="https://github.com/kubernetes/community/blob/master/contributors/devel/cherry-picks.md">here</a>. collapse patch conflict retry onto GuaranteedUpdate xref https://github.com/kubernetes/kubernetes/issues/63104 This PR builds on https://github.com/kubernetes/kubernetes/pull/62868 1. When the incoming patch specified a resourceVersion that failed as a precondition, the patch handler would retry uselessly 5 times. This PR collapses onto GuaranteedUpdate, which immediately stops retrying in that case. 2. When the incoming patch did not specify a resourceVersion, and persisting to etcd contended with other etcd updates, the retry would try to detect patch conflicts with deltas from the first 'current object' retrieved from etcd and fail with a conflict error in that case. Given that the user did not provide any information about the starting version they expected their patch to apply to, this does not make sense, and results in arbitrary conflict errors, depending on when the patch was submitted relative to other changes made to the resource. This PR changes the patch application to be performed on the object retrieved from etcd identically on every attempt. fixes #58017 SMP is no longer computed for CRD objects fixes #42644 No special state is retained on the first attempt, so the patch handler correctly handles the cached storage optimistically trying with a cached object first /assign @lavalamp ```release-note fixed spurious "unable to find api field" errors patching custom resources ```
parents d4b67803 b526532c
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 172 additions and 90 deletions
staging/src/k8s.io/apiextensions-apiserver/test/integration/basic_test.go
View file @ 6aad80cc
...@@ -582,17 +582,6 @@ func TestPatch(t *testing.T) { ...@@ -582,17 +582,6 @@ func TestPatch(t *testing.T) {
t.Fatalf("unexpected error: %v", err) 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 // a patch with no change
createdNoxuInstance, err = noxuNamespacedResourceClient.Patch("foo", types.MergePatchType, patch) createdNoxuInstance, err = noxuNamespacedResourceClient.Patch("foo", types.MergePatchType, patch)
if err != nil { if err != nil {
......
staging/src/k8s.io/apiserver/pkg/endpoints/handlers/rest.go
View file @ 6aad80cc
...@@ -90,9 +90,6 @@ func (scope *RequestScope) AllowsStreamSchema(s string) bool { ...@@ -90,9 +90,6 @@ func (scope *RequestScope) AllowsStreamSchema(s string) bool {
return s == "watch" 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. // 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 { func ConnectResource(connecter rest.Connecter, scope RequestScope, admit admission.Interface, restPath string, isSubresource bool) http.HandlerFunc {
return func(w http.ResponseWriter, req *http.Request) { return func(w http.ResponseWriter, req *http.Request) {
......
staging/src/k8s.io/apiserver/pkg/endpoints/handlers/rest_test.go
View file @ 6aad80cc
...@@ -173,32 +173,46 @@ func (p *testPatcher) New() runtime.Object { ...@@ -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) { 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 // Simulate GuaranteedUpdate behavior (retries internally on etcd changes if the incoming resource doesn't pin resourceVersion)
if p.numUpdates > 0 { for {
currentPod = p.updatePod currentPod := p.startingPod
} if p.numUpdates > 0 {
p.numUpdates++ currentPod = p.updatePod
}
p.numUpdates++
obj, err := objInfo.UpdatedObject(ctx, currentPod) // Remember the current resource version
if err != nil { currentResourceVersion := currentPod.ResourceVersion
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))
}
if currentPod == nil { obj, err := objInfo.UpdatedObject(ctx, currentPod)
if err := createValidation(currentPod); err != nil { if err != nil {
return nil, false, err return nil, false, err
} }
} else { inPod := obj.(*example.Pod)
if err := updateValidation(currentPod, inPod); err != nil { if inPod.ResourceVersion == "" || inPod.ResourceVersion == "0" {
return nil, false, err 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) { func (p *testPatcher) Get(ctx context.Context, name string, options *metav1.GetOptions) (runtime.Object, error) {
...@@ -252,15 +266,21 @@ type patchTestCase struct { ...@@ -252,15 +266,21 @@ type patchTestCase struct {
// startingPod is used as the starting point for the first Update // startingPod is used as the starting point for the first Update
startingPod *example.Pod startingPod *example.Pod
// changedPod is the "destination" pod for the patch. The test will create a patch from the startingPod to the changedPod // changedPod can be set as the "destination" pod for the patch, and the test will compute a patch from the startingPod to the changedPod,
// to use when calling the patch operation // or patches can be set directly using strategicMergePatch, mergePatch, and jsonPatch
changedPod *example.Pod changedPod *example.Pod
strategicMergePatch string
mergePatch string
jsonPatch string
// updatePod is the pod that is used for conflict comparison and as the starting point for the second Update // updatePod is the pod that is used for conflict comparison and as the starting point for the second Update
updatePod *example.Pod updatePod *example.Pod
// expectedPod is the pod that you expect to get back after the patch is complete // expectedPod is the pod that you expect to get back after the patch is complete
expectedPod *example.Pod expectedPod *example.Pod
expectedError string expectedError string
// if set, indicates the number of times patching was expected to be attempted
expectedTries int
} }
func (tc *patchTestCase) Run(t *testing.T) { func (tc *patchTestCase) Run(t *testing.T) {
...@@ -303,39 +323,59 @@ func (tc *patchTestCase) Run(t *testing.T) { ...@@ -303,39 +323,59 @@ func (tc *patchTestCase) Run(t *testing.T) {
updatePod: tc.updatePod, updatePod: tc.updatePod,
} }
// TODO SUPPORT THIS!
if patchType == types.JSONPatchType {
continue
}
t.Logf("Working with patchType %v", patchType) t.Logf("Working with patchType %v", patchType)
originalObjJS, err := runtime.Encode(codec, tc.startingPod)
if err != nil {
t.Errorf("%s: unexpected error: %v", tc.name, err)
continue
}
changedJS, err := runtime.Encode(codec, tc.changedPod)
if err != nil {
t.Errorf("%s: unexpected error: %v", tc.name, err)
continue
}
patch := []byte{} patch := []byte{}
switch patchType { switch patchType {
case types.StrategicMergePatchType: case types.StrategicMergePatchType:
patch, err = strategicpatch.CreateTwoWayMergePatch(originalObjJS, changedJS, schemaReferenceObj) patch = []byte(tc.strategicMergePatch)
if err != nil { if len(patch) == 0 {
t.Errorf("%s: unexpected error: %v", tc.name, err) originalObjJS, err := runtime.Encode(codec, tc.startingPod)
continue if err != nil {
t.Errorf("%s: unexpected error: %v", tc.name, err)
continue
}
changedJS, err := runtime.Encode(codec, tc.changedPod)
if err != nil {
t.Errorf("%s: unexpected error: %v", tc.name, err)
continue
}
patch, err = strategicpatch.CreateTwoWayMergePatch(originalObjJS, changedJS, schemaReferenceObj)
if err != nil {
t.Errorf("%s: unexpected error: %v", tc.name, err)
continue
}
} }
case types.MergePatchType: case types.MergePatchType:
patch, err = jsonpatch.CreateMergePatch(originalObjJS, changedJS) patch = []byte(tc.mergePatch)
if err != nil { if len(patch) == 0 {
t.Errorf("%s: unexpected error: %v", tc.name, err) originalObjJS, err := runtime.Encode(codec, tc.startingPod)
if err != nil {
t.Errorf("%s: unexpected error: %v", tc.name, err)
continue
}
changedJS, err := runtime.Encode(codec, tc.changedPod)
if err != nil {
t.Errorf("%s: unexpected error: %v", tc.name, err)
continue
}
patch, err = jsonpatch.CreateMergePatch(originalObjJS, changedJS)
if err != nil {
t.Errorf("%s: unexpected error: %v", tc.name, err)
continue
}
}
case types.JSONPatchType:
patch = []byte(tc.jsonPatch)
if len(patch) == 0 {
// TODO SUPPORT THIS!
continue continue
} }
default:
t.Error("unsupported patch type")
} }
p := patcher{ p := patcher{
...@@ -377,6 +417,12 @@ func (tc *patchTestCase) Run(t *testing.T) { ...@@ -377,6 +417,12 @@ func (tc *patchTestCase) Run(t *testing.T) {
} }
} }
if tc.expectedTries > 0 {
if tc.expectedTries != testPatcher.numUpdates {
t.Errorf("%s: expected %d tries, got %d", tc.expectedTries, testPatcher.numUpdates)
}
}
if tc.expectedPod == nil { if tc.expectedPod == nil {
if resultObj != nil { if resultObj != nil {
t.Errorf("%s: unexpected result: %v", tc.name, resultObj) t.Errorf("%s: unexpected result: %v", tc.name, resultObj)
...@@ -538,6 +584,73 @@ func TestPatchResourceWithVersionConflict(t *testing.T) { ...@@ -538,6 +584,73 @@ func TestPatchResourceWithVersionConflict(t *testing.T) {
tc.Run(t) tc.Run(t)
} }
func TestPatchResourceWithStaleVersionConflict(t *testing.T) {
namespace := "bar"
name := "foo"
uid := types.UID("uid")
tc := &patchTestCase{
name: "TestPatchResourceWithStaleVersionConflict",
startingPod: &example.Pod{},
updatePod: &example.Pod{},
expectedError: `Operation cannot be fulfilled on pods.example.apiserver.k8s.io "foo": existing 2, new 1`,
expectedTries: 1,
}
// starting pod is at rv=2
tc.startingPod.Name = name
tc.startingPod.Namespace = namespace
tc.startingPod.UID = uid
tc.startingPod.ResourceVersion = "2"
tc.startingPod.APIVersion = examplev1.SchemeGroupVersion.String()
// same pod is still in place when attempting to persist the update
tc.updatePod = tc.startingPod
// patches are submitted with a stale rv=1
tc.mergePatch = `{"metadata":{"resourceVersion":"1"},"spec":{"nodeName":"foo"}}`
tc.strategicMergePatch = `{"metadata":{"resourceVersion":"1"},"spec":{"nodeName":"foo"}}`
tc.Run(t)
}
func TestPatchResourceWithRacingVersionConflict(t *testing.T) {
namespace := "bar"
name := "foo"
uid := types.UID("uid")
tc := &patchTestCase{
name: "TestPatchResourceWithRacingVersionConflict",
startingPod: &example.Pod{},
updatePod: &example.Pod{},
expectedError: `Operation cannot be fulfilled on pods.example.apiserver.k8s.io "foo": existing 3, new 2`,
expectedTries: 2,
}
// starting pod is at rv=2
tc.startingPod.Name = name
tc.startingPod.Namespace = namespace
tc.startingPod.UID = uid
tc.startingPod.ResourceVersion = "2"
tc.startingPod.APIVersion = examplev1.SchemeGroupVersion.String()
// pod with rv=3 is found when attempting to persist the update
tc.updatePod.Name = name
tc.updatePod.Namespace = namespace
tc.updatePod.UID = uid
tc.updatePod.ResourceVersion = "3"
tc.updatePod.APIVersion = examplev1.SchemeGroupVersion.String()
// patches are submitted with a rv=2
tc.mergePatch = `{"metadata":{"resourceVersion":"2"},"spec":{"nodeName":"foo"}}`
tc.strategicMergePatch = `{"metadata":{"resourceVersion":"2"},"spec":{"nodeName":"foo"}}`
tc.Run(t)
}
func TestPatchResourceWithConflict(t *testing.T) { func TestPatchResourceWithConflict(t *testing.T) {
namespace := "bar" namespace := "bar"
name := "foo" name := "foo"
...@@ -549,8 +662,7 @@ func TestPatchResourceWithConflict(t *testing.T) { ...@@ -549,8 +662,7 @@ func TestPatchResourceWithConflict(t *testing.T) {
startingPod: &example.Pod{}, startingPod: &example.Pod{},
changedPod: &example.Pod{}, changedPod: &example.Pod{},
updatePod: &example.Pod{}, updatePod: &example.Pod{},
expectedPod: &example.Pod{},
expectedError: `Operation cannot be fulfilled on pods.example.apiserver.k8s.io "foo": existing 2, new 1`,
} }
// See issue #63104 for discussion of how much sense this makes. // See issue #63104 for discussion of how much sense this makes.
...@@ -576,6 +688,13 @@ func TestPatchResourceWithConflict(t *testing.T) { ...@@ -576,6 +688,13 @@ func TestPatchResourceWithConflict(t *testing.T) {
tc.updatePod.APIVersion = examplev1.SchemeGroupVersion.String() tc.updatePod.APIVersion = examplev1.SchemeGroupVersion.String()
tc.updatePod.Spec.NodeName = "anywhere" 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) tc.Run(t)
} }
......
test/integration/apiserver/BUILD
View file @ 6aad80cc
...@@ -43,7 +43,6 @@ go_test( ...@@ -43,7 +43,6 @@ go_test(
"//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library", "//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/runtime/schema: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/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/features:go_default_library",
"//vendor/k8s.io/apiserver/pkg/util/feature:go_default_library", "//vendor/k8s.io/apiserver/pkg/util/feature:go_default_library",
"//vendor/k8s.io/client-go/kubernetes:go_default_library", "//vendor/k8s.io/client-go/kubernetes:go_default_library",
......
test/integration/apiserver/patch_test.go
View file @ 6aad80cc
...@@ -29,11 +29,10 @@ import ( ...@@ -29,11 +29,10 @@ import (
"k8s.io/apimachinery/pkg/api/meta" "k8s.io/apimachinery/pkg/api/meta"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/types"
"k8s.io/apiserver/pkg/endpoints/handlers"
"k8s.io/kubernetes/test/integration/framework" "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) { func TestPatchConflicts(t *testing.T) {
s, clientSet, closeFn := setup(t) s, clientSet, closeFn := setup(t)
defer closeFn() defer closeFn()
...@@ -41,7 +40,7 @@ func TestPatchConflicts(t *testing.T) { ...@@ -41,7 +40,7 @@ func TestPatchConflicts(t *testing.T) {
ns := framework.CreateTestingNamespace("status-code", s, t) ns := framework.CreateTestingNamespace("status-code", s, t)
defer framework.DeleteTestingNamespace(ns, s, t) defer framework.DeleteTestingNamespace(ns, s, t)
numOfConcurrentPatches := 2 * handlers.MaxRetryWhenPatchConflicts numOfConcurrentPatches := 100
UIDs := make([]types.UID, numOfConcurrentPatches) UIDs := make([]types.UID, numOfConcurrentPatches)
ownerRefs := []metav1.OwnerReference{} ownerRefs := []metav1.OwnerReference{}
...@@ -69,10 +68,8 @@ func TestPatchConflicts(t *testing.T) { ...@@ -69,10 +68,8 @@ func TestPatchConflicts(t *testing.T) {
successes := int32(0) successes := int32(0)
// Run a lot of simultaneous patch operations to exercise internal API server retry of patch application. // Run a lot of simultaneous patch operations to exercise internal API server retry of application of patches that do not specify resourceVersion.
// Internally, a patch API call retries up to MaxRetryWhenPatchConflicts times if the resource version of the object has changed. // They should all succeed.
// 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.
wg := sync.WaitGroup{} wg := sync.WaitGroup{}
for i := 0; i < numOfConcurrentPatches; i++ { for i := 0; i < numOfConcurrentPatches; i++ {
wg.Add(1) wg.Add(1)
...@@ -123,8 +120,8 @@ func TestPatchConflicts(t *testing.T) { ...@@ -123,8 +120,8 @@ func TestPatchConflicts(t *testing.T) {
} }
wg.Wait() wg.Wait()
if successes < handlers.MaxRetryWhenPatchConflicts { if successes < int32(numOfConcurrentPatches) {
t.Errorf("Expected at least %d successful patches for %s, got %d", handlers.MaxRetryWhenPatchConflicts, "secrets", successes) t.Errorf("Expected at least %d successful patches for %s, got %d", numOfConcurrentPatches, "secrets", successes)
} else { } else {
t.Logf("Got %d successful patches for %s", successes, "secrets") t.Logf("Got %d successful patches for %s", successes, "secrets")
} }
......
test/integration/client/BUILD
View file @ 6aad80cc
...@@ -22,7 +22,6 @@ go_test( ...@@ -22,7 +22,6 @@ go_test(
"//test/utils/image:go_default_library", "//test/utils/image:go_default_library",
"//vendor/k8s.io/api/core/v1: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/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:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/apis/meta/v1/unstructured:go_default_library", "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1/unstructured:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/fields:go_default_library", "//vendor/k8s.io/apimachinery/pkg/fields:go_default_library",
......
test/integration/client/client_test.go
View file @ 6aad80cc
...@@ -27,7 +27,6 @@ import ( ...@@ -27,7 +27,6 @@ import (
"k8s.io/api/core/v1" "k8s.io/api/core/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors" apierrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/api/meta"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/fields" "k8s.io/apimachinery/pkg/fields"
"k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/labels"
...@@ -286,23 +285,6 @@ func TestPatch(t *testing.T) { ...@@ -286,23 +285,6 @@ func TestPatch(t *testing.T) {
t.Logf("%v", string(jsonObj)) 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 return nil
} }
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment