Merge pull request #197 from takirala/tga/add-retry

fix(release-0.2): add retry during status updates

Author: k8s-ci-robot

controller/pkg/bucketclaim/bucketclaim.go

@@ -10,6 +10,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	kubeclientset "k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/record"
+	"k8s.io/client-go/util/retry"
 	"k8s.io/klog/v2"
 	"sigs.k8s.io/container-object-storage-interface/client/apis/objectstorage/v1alpha1"
 	bucketclientset "sigs.k8s.io/container-object-storage-interface/client/clientset/versioned"
@@ -216,18 +217,66 @@ func (b *BucketClaimListener) provisionBucketClaimOperation(ctx context.Context,
 		bucketClaim.Status.BucketReady = false
 	}
 
-	// Fetching the updated bucketClaim again, so that the update
-	// operation doesn't happen on an outdated version of the bucketClaim.
-	bucketClaim, err = b.bucketClaims(bucketClaim.ObjectMeta.Namespace).UpdateStatus(ctx, bucketClaim, metav1.UpdateOptions{})
+	// Update status with retry logic for conflict errors
+	// Store the status values we want to set before entering retry loop
+	statusBucketName := bucketClaim.Status.BucketName
+	statusBucketReady := bucketClaim.Status.BucketReady
+
+	err = retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		// Fetch the latest version of the BucketClaim
+		latest, getErr := b.bucketClaims(bucketClaim.Namespace).Get(
+			ctx,
+			bucketClaim.Name,
+			metav1.GetOptions{},
+		)
+		if getErr != nil {
+			return getErr
+		}
+
+		// Apply the status changes to the latest version
+		latest.Status.BucketName = statusBucketName
+		latest.Status.BucketReady = statusBucketReady
+
+		// Try to update the status
+		var updateErr error
+		bucketClaim, updateErr = b.bucketClaims(bucketClaim.Namespace).UpdateStatus(
+			ctx,
+			latest,
+			metav1.UpdateOptions{},
+		)
+		return updateErr
+	})
 	if err != nil {
 		klog.V(3).ErrorS(err, "Failed to update status of BucketClaim", "name", bucketClaim.ObjectMeta.Name)
 		return b.recordError(inputBucketClaim, v1.EventTypeWarning, v1alpha1.FailedCreateBucket, err)
 	}
 
 	// Add the finalizers so that bucketClaim is deleted
 	// only after the associated bucket is deleted.
-	controllerutil.AddFinalizer(bucketClaim, util.BucketClaimFinalizer)
-	_, err = b.bucketClaims(bucketClaim.ObjectMeta.Namespace).Update(ctx, bucketClaim, metav1.UpdateOptions{})
+	// Update with retry logic for conflict errors
+	err = retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		// Fetch the latest version of the BucketClaim
+		latest, getErr := b.bucketClaims(bucketClaim.Namespace).Get(
+			ctx,
+			bucketClaim.Name,
+			metav1.GetOptions{},
+		)
+		if getErr != nil {
+			return getErr
+		}
+
+		// Add the finalizer to the latest version
+		controllerutil.AddFinalizer(latest, util.BucketClaimFinalizer)
+
+		// Try to update
+		var updateErr error
+		bucketClaim, updateErr = b.bucketClaims(bucketClaim.Namespace).Update(
+			ctx,
+			latest,
+			metav1.UpdateOptions{},
+		)
+		return updateErr
+	})
 	if err != nil {
 		klog.V(3).ErrorS(err, "Failed to add finalizer BucketClaim", "name", bucketClaim.ObjectMeta.Name)
 		return b.recordError(inputBucketClaim, v1.EventTypeWarning, v1alpha1.FailedCreateBucket, err)

controller/pkg/bucketclaim/bucketclaim_test.go

@@ -3,6 +3,7 @@ package bucketclaim
 import (
 	"context"
 	"fmt"
+	"sync"
 	"testing"
 
 	v1 "k8s.io/api/core/v1"
@@ -13,6 +14,7 @@ import (
 	"sigs.k8s.io/container-object-storage-interface/client/apis/objectstorage/v1alpha1"
 	fakebucketclientset "sigs.k8s.io/container-object-storage-interface/client/clientset/versioned/fake"
 	"sigs.k8s.io/container-object-storage-interface/controller/pkg/util"
+	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 )
 
 var classGoldParameters = map[string]string{
@@ -352,3 +354,94 @@ func TestAddDeletedBucketClaim(t *testing.T) {
 		t.Fatalf("expected 0 buckets, got %d", len(bl.Items))
 	}
 }
+
+// Test retry logic for conflict errors during status update
+func TestRetryOnConflictStatusUpdate(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	client := fakebucketclientset.NewSimpleClientset()
+	kubeClient := fakekubeclientset.NewSimpleClientset()
+	eventRecorder := record.NewFakeRecorder(3)
+
+	listener := NewBucketClaimListener()
+	listener.InitializeKubeClient(kubeClient)
+	listener.InitializeBucketClient(client)
+	listener.InitializeEventRecorder(eventRecorder)
+
+	bucketclass, err := util.CreateBucketClass(ctx, client, &goldClass)
+	if err != nil {
+		t.Fatalf("Error occurred when creating BucketClass: %v", err)
+	}
+
+	bucketClaim, err := util.CreateBucketClaim(ctx, client, &bucketClaim1)
+	if err != nil {
+		t.Fatalf("Error occurred when creating BucketClaim: %v", err)
+	}
+
+	// Cleanup
+	defer util.DeleteObjects(ctx, client, *bucketClaim, *bucketclass)
+
+	// Simulate concurrent modification by updating the BucketClaim in a goroutine
+	// This will cause resourceVersion to change, simulating a conflict scenario
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for i := range 10 {
+			// Fetch and update the BucketClaim to change its resourceVersion
+			bc, getErr := client.ObjectstorageV1alpha1().BucketClaims(bucketClaim.Namespace).Get(
+				ctx,
+				bucketClaim.Name,
+				metav1.GetOptions{},
+			)
+			if getErr != nil {
+				return
+			}
+			// Add an annotation to change the resourceVersion
+			if bc.Annotations == nil {
+				bc.Annotations = make(map[string]string)
+			}
+			bc.Annotations[fmt.Sprintf("test-%d", i)] = "value"
+			_, _ = client.ObjectstorageV1alpha1().BucketClaims(bc.Namespace).Update(
+				ctx,
+				bc,
+				metav1.UpdateOptions{},
+			)
+		}
+	}()
+
+	// Call Add which should handle conflicts with retry logic
+	err = listener.Add(ctx, bucketClaim)
+	if err != nil {
+		t.Fatalf("Add should succeed even with concurrent modifications: %v", err)
+	}
+
+	// Wait for the goroutine to complete to ensure all concurrent updates are done
+	wg.Wait()
+
+	// Verify the final state - status should be updated correctly
+	updatedClaim, err := client.ObjectstorageV1alpha1().BucketClaims(bucketClaim.Namespace).Get(
+		ctx,
+		bucketClaim.Name,
+		metav1.GetOptions{},
+	)
+	if err != nil {
+		t.Fatalf("Error occurred when reading BucketClaim: %v", err)
+	}
+
+	// Verify status was updated
+	expectedBucketName := fmt.Sprintf("bucket-%s", bucketClaim.UID)
+	if updatedClaim.Status.BucketName != expectedBucketName {
+		t.Errorf("Expected BucketName %s, got %s", expectedBucketName, updatedClaim.Status.BucketName)
+	}
+
+	if updatedClaim.Status.BucketReady != false {
+		t.Errorf("Expected BucketReady to be false, got %v", updatedClaim.Status.BucketReady)
+	}
+
+	// Verify finalizer was added
+	if !controllerutil.ContainsFinalizer(updatedClaim, util.BucketClaimFinalizer) {
+		t.Errorf("Expected finalizer to be added, but it was not found")
+	}
+}

vendor/k8s.io/client-go/util/retry/OWNERS

@@ -756,6 +756,7 @@ k8s.io/client-go/util/consistencydetector
 k8s.io/client-go/util/flowcontrol
 k8s.io/client-go/util/homedir
 k8s.io/client-go/util/keyutil
+k8s.io/client-go/util/retry
 k8s.io/client-go/util/watchlist
 k8s.io/client-go/util/workqueue
 # k8s.io/klog/v2 v2.130.1