Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions client/finalizer.go
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ func AddFinalizerAndUpdate(ctx context.Context, c client.Client, obj client.Obje
if controllerutil.ContainsFinalizer(obj, finalizer) {
return nil
}
_, err := UpdateOnConflict(ctx, c, c, obj, func(obj client.Object) error {
_, err := PatchOnConflict(ctx, c, c, obj, func(obj client.Object) error {
controllerutil.AddFinalizer(obj, finalizer)
return nil
})
Expand All @@ -38,7 +38,7 @@ func RemoveFinalizerAndUpdate(ctx context.Context, c client.Client, obj client.O
if !controllerutil.ContainsFinalizer(obj, finalizer) {
return nil
}
_, err := UpdateOnConflict(ctx, c, c, obj, func(obj client.Object) error {
_, err := PatchOnConflict(ctx, c, c, obj, func(obj client.Object) error {
controllerutil.RemoveFinalizer(obj, finalizer)
return nil
})
Expand Down
6 changes: 3 additions & 3 deletions client/finalizer_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ import (
func TestAddFinalizerAndUpdate(t *testing.T) {
c := fake.NewClientBuilder().WithScheme(scheme.Scheme).Build()

pod := newTestPod()
pod := newTestPod("test-pod")
err := c.Create(context.Background(), pod)
require.NoError(t, err)

Expand Down Expand Up @@ -62,7 +62,7 @@ func TestAddFinalizerAndUpdate(t *testing.T) {
func TestRemoveFinalizerAndUpdate(t *testing.T) {
c := fake.NewClientBuilder().WithScheme(scheme.Scheme).Build()

pod := newTestPod()
pod := newTestPod("test-pod")
err := c.Create(context.Background(), pod)
require.NoError(t, err)

Expand Down Expand Up @@ -95,7 +95,7 @@ func TestRemoveFinalizerAndUpdate(t *testing.T) {
func TestRemoveFinalizerAndDelete(t *testing.T) {
c := fake.NewClientBuilder().WithScheme(scheme.Scheme).Build()

pod := newTestPod()
pod := newTestPod("test-pod")
err := c.Create(context.Background(), pod)
require.NoError(t, err)

Expand Down
29 changes: 29 additions & 0 deletions client/suit_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,29 @@
/*
Copyright 2025 The KusionStack Authors.

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 client

import (
"testing"

"github.com/stretchr/testify/suite"
)

// In order for 'go test' to run this suite, we need to create
// a normal test function and pass our suite to suite.Run
func TestClientTestSuite(t *testing.T) {
suite.Run(t, new(clientTestSuite))
}
153 changes: 134 additions & 19 deletions client/util.go
Original file line number Diff line number Diff line change
Expand Up @@ -37,43 +37,48 @@ type UpdateWriter interface {
Update(ctx context.Context, obj client.Object, opts ...client.UpdateOption) error
}

// UpdateOnConflict attempts to update a resource while avoiding conflicts that may arise from concurrent modifications.
// It utilizes the mutateFn function to apply changes to the original obj and then attempts an update using the writer,
// which can be either client.Writer or client.StatusWriter.
// In case of an update failure due to a conflict, UpdateOnConflict will retrieve the latest version of the object using
// the reader and attempt the update again.
// The retry mechanism adheres to the retry.DefaultBackoff policy.
// UpdateOnConflict attempts to update a resource while avoiding conflicts that may arise
// from concurrent modifications. It utilizes the mutateFn function to apply changes to the
// input obj and then attempts an update using the writer, which can be either client.Writer
// or client.StatusWriter.
//
// In case of an update failure due to a conflict, UpdateOnConflict will retrieve the latest version
// of the object using the reader and attempt the update again. The retry mechanism adheres to the
// retry.DefaultBackoff policy.
//
// NOTE: It must be ensured that the input object has not been modified in any way after being obtained
// from cache or apiserver.
func UpdateOnConflict[T client.Object](
ctx context.Context,
reader client.Reader,
writer UpdateWriter,
original T,
inputObj T,
mutateFn MutateFn[T],
) (changed bool, err error) {
key := client.ObjectKeyFromObject(original)
key := client.ObjectKeyFromObject(inputObj)
first := true

err = retry.RetryOnConflict(retry.DefaultBackoff, func() error {
if !first {
// refresh object
if innerErr := reader.Get(ctx, key, original); innerErr != nil {
if innerErr := reader.Get(ctx, key, inputObj); innerErr != nil {
return innerErr
}
} else {
first = false
}

existing := original.DeepCopyObject()
if innerErr := mutate(mutateFn, key, original); innerErr != nil {
original := inputObj.DeepCopyObject()
if innerErr := mutate(mutateFn, key, inputObj); innerErr != nil {
return innerErr
}

if equality.Semantic.DeepEqual(existing, original) {
if equality.Semantic.DeepEqual(original, inputObj) {
// nothing changed, skip update
return nil
}

if innerErr := writer.Update(ctx, original); innerErr != nil {
if innerErr := writer.Update(ctx, inputObj); innerErr != nil {
return innerErr
}

Expand Down Expand Up @@ -106,24 +111,24 @@ func CreateOrUpdateOnConflict[T client.Object](
ctx context.Context,
reader client.Reader,
writer client.Writer,
original T,
inputObj T,
f MutateFn[T],
) (controllerutil.OperationResult, error) {
key := client.ObjectKeyFromObject(original)
if err := reader.Get(ctx, key, original); err != nil {
key := client.ObjectKeyFromObject(inputObj)
if err := reader.Get(ctx, key, inputObj); err != nil {
if !apierrors.IsNotFound(err) {
return controllerutil.OperationResultNone, err
}
// not found, create it
if err := mutate(f, key, original); err != nil {
if err := mutate(f, key, inputObj); err != nil {
return controllerutil.OperationResultNone, err
}
if err := writer.Create(ctx, original); err != nil {
if err := writer.Create(ctx, inputObj); err != nil {
return controllerutil.OperationResultNone, err
}
return controllerutil.OperationResultCreated, nil
}
changed, err := UpdateOnConflict(ctx, reader, writer, original, f)
changed, err := UpdateOnConflict(ctx, reader, writer, inputObj, f)
if err != nil {
return controllerutil.OperationResultNone, err
}
Expand All @@ -132,3 +137,113 @@ func CreateOrUpdateOnConflict[T client.Object](
}
return controllerutil.OperationResultUpdated, nil
}

type PatchWriter interface {
// Patch patches the given obj in the Kubernetes cluster. obj must be a
// struct pointer so that obj can be updated with the content returned by the Server.
Patch(ctx context.Context, obj client.Object, patch client.Patch, opts ...client.PatchOption) error
}

// Patch applies changes to a Kubernetes object using the merge patch strategy.
// It creates a merge patch from the original object, applies the mutateFn to modify the object,
// and then applies the patch if there are actual changes.
// Returns true if changes were applied, false otherwise.
func Patch[T client.Object](
ctx context.Context,
writer PatchWriter,
inputObj T,
mutateFn MutateFn[T],
) (changed bool, err error) {
original := inputObj.DeepCopyObject().(client.Object)

mergePatch := client.MergeFrom(original)

// modify inputObj object
err = mutateFn(inputObj)
if err != nil {
return false, err
}

// calculate patch data
patchData, err := mergePatch.Data(inputObj)
if err != nil {
return false, err
}

if len(patchData) == 0 || string(patchData) == "{}" {
// nothing changed, skip update
return false, nil
}

err = writer.Patch(ctx, inputObj, mergePatch)
if err != nil {
return false, err
}

// patched
return true, nil
}

// PatchOnConflict attempts to patch a resource while avoiding conflicts that may arise
// from concurrent modifications. It utilizes the mutateFn function to apply changes to the
// input obj and then attempts an mergePatch using the writer, which can be either client.Writer
// or client.StatusWriter.
//
// In case of an update failure due to a conflict, PatchOnConflict will retrieve the latest version
// of the object using the reader and attempt the update again. The retry mechanism adheres to the
// retry.DefaultBackoff policy.
//
// NOTE: It must be ensured that the input object has not been modified in any way after being obtained
// from cache or apiserver.
func PatchOnConflict[T client.Object](
ctx context.Context,
reader client.Reader,
writer PatchWriter,
inputObj T,
mutateFn MutateFn[T],
) (changed bool, err error) {
key := client.ObjectKeyFromObject(inputObj)
first := true

err = retry.RetryOnConflict(retry.DefaultBackoff, func() error {
if !first {
// refresh object
if innerErr := reader.Get(ctx, key, inputObj); innerErr != nil {
return innerErr
}
} else {
first = false
}

original := inputObj.DeepCopyObject().(client.Object)

// modify inputObj object
if innerErr := mutateFn(inputObj); innerErr != nil {
return innerErr
}

mergePatch := client.MergeFrom(original)

// calculate patch data
patchData, err := mergePatch.Data(inputObj)
if err != nil {
return err
}

if len(patchData) == 0 || string(patchData) == "{}" {
// nothing changed, skip update
return nil
}

// use resourceVersion as optimisticLock
mergePatchWithOptimisticLock := client.MergeFromWithOptions(original, client.MergeFromWithOptimisticLock{})
if innerErr := writer.Patch(ctx, inputObj, mergePatchWithOptimisticLock); innerErr != nil {
return innerErr
}

changed = true
return nil
})

return changed, err
}
Loading
Loading