Add retry mechanism for patch operations to handle resource version conflicts (#207)

fixes: aws-controllers-k8s/community#2267

ACK controllers were experiencing race conditions when multiple controllers attempted to patch the same Kubernetes resource simultaneously. This occurred because Kubernetes uses optimistic concurrency control through resource versions - each resource has a version that must match when performing updates. When multiple controllers read a resource and then attempt to patch it, only the first patch succeeds while subsequent patches fail with HTTP 409 conflict errors due to stale resource versions.

Description of changes:

A retry mechanism using Kubernetes client-go's standard `retry.RetryOnConflict` function with `retry.DefaultBackoff` configuration. When a patch operation encounters a resource version conflict, the mechanism automatically refreshes the resource version by fetching the latest state from the API server, then retries the patch operation with the updated version. This process repeats up to 5 times with exponential backoff (starting at 100ms, doubling each time, capped at 1 second), ensuring that temporary conflicts are resolved automatically while avoiding excessive API server load.

Ref:
- https://github.com/kubernetes-sigs/kro/blob/c1bc05c5384245d3ef2a5104198459732552b148/pkg/controller/resourcegraphdefinition/controller_status.go#L34-L75
- https://github.com/prometheus-operator/prometheus-operator/blob/3ff38ebe6216c28da344abd6e4f698831309b959/pkg/k8sutil/k8sutil.go#L239-L376
- https://github.com/prometheus-operator/prometheus-operator/blob/3ff38ebe6216c28da344abd6e4f698831309b959/pkg/k8sutil/k8sutil.go#L513-L522

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.

Author: rushmash91

pkg/runtime/adoption_reconciler.go

@@ -375,7 +375,7 @@ func (r *adoptionReconciler) patchAdoptedCondition(
 		adoptedCondition.Status = corev1.ConditionTrue
 	}
 
-	return r.patchStatus(ctx, res, base)
+	return r.patchResourceStatus(ctx, res, base)
 }
 
 // isAdopted returns true if the AdoptedResource is in a terminal adoption state
@@ -411,7 +411,7 @@ func (r *adoptionReconciler) markManaged(
 ) error {
 	base := res.DeepCopy()
 	k8sctrlutil.AddFinalizer(res, adoptionFinalizerString)
-	return r.patchMetadataAndSpec(ctx, res, base)
+	return r.patchResourceMetadataAndSpec(ctx, res, base)
 }
 
 // markUnmanaged removes the supplied resource from management by ACK.
@@ -423,7 +423,7 @@ func (r *adoptionReconciler) markUnmanaged(
 ) error {
 	base := res.DeepCopy()
 	k8sctrlutil.RemoveFinalizer(res, adoptionFinalizerString)
-	return r.patchMetadataAndSpec(ctx, res, base)
+	return r.patchResourceMetadataAndSpec(ctx, res, base)
 }
 
 // handleReconcileError will handle errors from reconcile handlers, which
@@ -561,12 +561,12 @@ func (r *adoptionReconciler) getRegion(
 	return ackv1alpha1.AWSRegion(r.cfg.Region)
 }
 
-// patchMetadataAndSpec patches the Metadata and Spec for AdoptedResource into
+// patchResourceMetadataAndSpec patches the Metadata and Spec for AdoptedResource into
 // k8s. The adopted resource 'res' also gets updated with content returned from
 // apiserver.
-// TODO(vijtrip2@): Refactor this and use single 'patchMetadataAndSpec' method
+// TODO(vijtrip2@): Refactor this and use single 'patchResourceMetadataAndSpec' method
 // for reconciler and adoptionReconciler
-func (r *adoptionReconciler) patchMetadataAndSpec(
+func (r *adoptionReconciler) patchResourceMetadataAndSpec(
 	ctx context.Context,
 	res *ackv1alpha1.AdoptedResource,
 	base *ackv1alpha1.AdoptedResource,
@@ -576,21 +576,22 @@ func (r *adoptionReconciler) patchMetadataAndSpec(
 	// Keep a copy of status field to reset the status of 'res' after patch call
 	resStatusCopy := res.DeepCopy().Status
 
-	err := patchWithoutCancel(ctx, r.kc, res, client.MergeFrom(base))
+	err := patchMetadataAndSpec(ctx, r.kc, res, client.MergeFrom(base))
 	res.Status = resStatusCopy
 	return err
 }
 
-// patchStatus patches the Status for AdoptedResource into k8s. The adopted
+// patchResourceStatus patches the Status for AdoptedResource into k8s. The adopted
 // resource 'res' also gets updated with the content returned from apiserver.
-// TODO(vijtrip2): Refactor this and use single 'patchStatus' method
+// TODO(vijtrip2): Refactor this and use single 'patchResourceStatus' method
 // for reconciler and adoptionReconciler
-func (r *adoptionReconciler) patchStatus(
+func (r *adoptionReconciler) patchResourceStatus(
 	ctx context.Context,
 	res *ackv1alpha1.AdoptedResource,
 	base *ackv1alpha1.AdoptedResource,
 ) error {
-	return patchStatusWithoutCancel(ctx, r.kc, res, client.MergeFrom(base))
+	rlog := ackrtlog.FromContext(ctx)
+	return patchStatus(ctx, r.kc, r.apiReader, res, client.MergeFrom(base), rlog)
 }
 
 // NewAdoptionReconciler returns a new adoptionReconciler object

pkg/runtime/field_export_reconciler.go

@@ -150,7 +150,7 @@ func (r *fieldExportReconciler) Sync(
 
 	r.resetConditions(ctx, &desired)
 	defer func() {
-		r.patchStatus(ctx, &desired, latest)
+		r.patchResourceStatus(ctx, &desired, latest)
 	}()
 
 	// Get the field from the resource
@@ -324,7 +324,7 @@ func (r *fieldExportReconciler) writeToConfigMap(
 	cm.Data[key] = sourceValue
 
 	ackrtlog.DebugFieldExport(r.log, desired, "patching target config map")
-	err = patchWithoutCancel(ctx, r.kc, cm, patch)
+	err = patchMetadataAndSpec(ctx, r.kc, cm, patch)
 	if err != nil {
 		return err
 	}
@@ -371,7 +371,7 @@ func (r *fieldExportReconciler) writeToSecret(
 	secret.Data[key] = []byte(sourceValue)
 
 	ackrtlog.DebugFieldExport(r.log, desired, "patching target secret")
-	err = patchWithoutCancel(ctx, r.kc, secret, patch)
+	err = patchMetadataAndSpec(ctx, r.kc, secret, patch)
 	if err != nil {
 		return err
 	}
@@ -519,16 +519,17 @@ func (r *fieldExportReconciler) patchTerminalCondition(
 	return nil
 }
 
-// patchStatus patches the Status for FieldExport into k8s. The field export
+// patchResourceStatus patches the Status for FieldExport into k8s. The field export
 // 'res' also gets updated with the content returned from apiserver.
-// TODO(vijtrip2): Refactor this and use single 'patchStatus' method
+// TODO(vijtrip2): Refactor this and use single 'patchResourceStatus' method
 // for all reconcilers
-func (r *fieldExportReconciler) patchStatus(
+func (r *fieldExportReconciler) patchResourceStatus(
 	ctx context.Context,
 	res *ackv1alpha1.FieldExport,
 	base *ackv1alpha1.FieldExport,
 ) error {
-	return patchStatusWithoutCancel(ctx, r.kc, res, client.MergeFrom(base))
+	rlog := ackrtlog.FromContext(ctx)
+	return patchStatus(ctx, r.kc, r.apiReader, res, client.MergeFrom(base), rlog)
 }
 
 // markManaged places the supplied resource under the management of ACK.
@@ -541,7 +542,7 @@ func (r *fieldExportReconciler) markManaged(
 	if !k8sctrlutil.ContainsFinalizer(res, fieldExportFinalizerString) {
 		base := res.DeepCopy()
 		k8sctrlutil.AddFinalizer(res, fieldExportFinalizerString)
-		return r.patchMetadataAndSpec(ctx, res, base)
+		return r.patchResourceMetadataAndSpec(ctx, res, base)
 	}
 	return nil
 }
@@ -556,17 +557,17 @@ func (r *fieldExportReconciler) markUnmanaged(
 	if k8sctrlutil.ContainsFinalizer(res, fieldExportFinalizerString) {
 		base := res.DeepCopy()
 		k8sctrlutil.RemoveFinalizer(res, fieldExportFinalizerString)
-		return r.patchMetadataAndSpec(ctx, res, base)
+		return r.patchResourceMetadataAndSpec(ctx, res, base)
 	}
 	return nil
 }
 
-// patchMetadataAndSpec patches the Metadata and Spec for FieldExport into
+// patchResourceMetadataAndSpec patches the Metadata and Spec for FieldExport into
 // k8s. The field export 'res' also gets updated with content returned from
 // apiserver.
-// TODO(vijtrip2@): Refactor this and use single 'patchMetadataAndSpec' method
+// TODO(vijtrip2@): Refactor this and use single 'patchResourceMetadataAndSpec' method
 // for all reconcilers
-func (r *fieldExportReconciler) patchMetadataAndSpec(
+func (r *fieldExportReconciler) patchResourceMetadataAndSpec(
 	ctx context.Context,
 	res *ackv1alpha1.FieldExport,
 	base *ackv1alpha1.FieldExport,
@@ -576,7 +577,7 @@ func (r *fieldExportReconciler) patchMetadataAndSpec(
 	// Keep a copy of status field to reset the status of 'res' after patch call
 	resStatusCopy := res.DeepCopy().Status
 
-	err := patchWithoutCancel(ctx, r.kc, res, client.MergeFrom(base))
+	err := patchMetadataAndSpec(ctx, r.kc, res, client.MergeFrom(base))
 	res.Status = resStatusCopy
 	return err
 }

pkg/runtime/reconciler.go

@@ -892,7 +892,7 @@ func (r *resourceReconciler) patchResourceMetadataAndSpec(
 	lorig := latestCleaned.DeepCopy()
 	patch := client.MergeFrom(desiredCleaned.RuntimeObject())
 
-	err = patchWithoutCancel(ctx, r.kc, latestCleaned.RuntimeObject(), patch)
+	err = patchMetadataAndSpec(ctx, r.kc, latestCleaned.RuntimeObject(), patch)
 
 	if err == nil {
 		if rlog.IsDebugEnabled() {
@@ -930,7 +930,7 @@ func (r *resourceReconciler) patchResourceStatus(
 	lobj := latest.DeepCopy().RuntimeObject()
 	patch := client.MergeFrom(dobj)
 
-	err = patchStatusWithoutCancel(ctx, r.kc, lobj, patch)
+	err = patchStatus(ctx, r.kc, r.apiReader, lobj, patch, rlog)
 
 	if err == nil {
 		if rlog.IsDebugEnabled() {

pkg/runtime/util.go

@@ -20,6 +20,7 @@ import (
 	"strings"
 
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/util/retry"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	ackv1alpha1 "github.com/aws-controllers-k8s/runtime/apis/core/v1alpha1"
@@ -41,6 +42,10 @@ const (
 	AdoptionPolicy_Adopt AdoptionPolicy = "adopt"
 	// AdoptPolicy is ...
 	AdoptionPolicy_AdoptOrCreate AdoptionPolicy = "adopt-or-create"
+
+	// Operation types for patch operations
+	OperationType_MetadataSpec = "metadata+spec"
+	OperationType_Status       = "status"
 )
 
 // IsAdopted returns true if the supplied AWSResource was created with a
@@ -144,34 +149,102 @@ func getAdoptionFields(res acktypes.AWSResource) string {
 	return ""
 }
 
-// patchWithoutCancel performs a patch operation using context.WithoutCancel to prevent
+// patchObject performs a patch operation using context.WithoutCancel to prevent
 // patch operations from being cancelled while preserving context values.
+// It automatically determines whether to patch spec/metadata or status based on operationType.
 //
 // NOTE(rushmash91): The 30s SIGTERM grace period acts as the effective timeout -
 // no additional timeout needed to avoid interfering with normal Kubernetes client
 // timeout/retry strategy.
-func patchWithoutCancel(
+func patchObject(
 	ctx context.Context,
 	kc client.Client,
 	obj client.Object,
 	patch client.Patch,
+	operationType string,
 ) error {
 	patchCtx := context.WithoutCancel(ctx)
+	if operationType == OperationType_Status {
+		return kc.Status().Patch(patchCtx, obj, patch)
+	}
 	return kc.Patch(patchCtx, obj, patch)
 }
 
-// patchStatusWithoutCancel performs a status patch operation using context.WithoutCancel
-// to prevent patch operations from being cancelled while preserving context values.
+// patchWithRetry performs a patch operation with retry on conflicts using client-go's standard retry mechanism.
+// This helps handle race conditions where multiple controllers try to update the same resource.
 //
-// NOTE(rushmash91): The 30s SIGTERM grace period acts as the effective timeout -
-// no additional timeout needed to avoid interfering with normal Kubernetes client
-// timeout/retry strategy.
-func patchStatusWithoutCancel(
+// When a conflict occurs (HTTP 409), it refreshes the resource version and retries the patch operation.
+func patchWithRetry(
 	ctx context.Context,
 	kc client.Client,
+	apiReader client.Reader,
 	obj client.Object,
 	patch client.Patch,
+	logger acktypes.Logger,
+	operationType string,
 ) error {
-	patchCtx := context.WithoutCancel(ctx)
-	return kc.Status().Patch(patchCtx, obj, patch)
+	attempt := 0
+
+	return retry.RetryOnConflict(retry.DefaultBackoff, func() error {
+		attempt++
+
+		// For retry attempts (after first attempt), refresh the object from the API server
+		if attempt > 1 {
+			logger.Debug(fmt.Sprintf("%s patch conflict detected, refreshing resource version", operationType),
+				"attempt", attempt,
+				"object", client.ObjectKeyFromObject(obj))
+
+			key := client.ObjectKeyFromObject(obj)
+			freshObject := obj.DeepCopyObject().(client.Object)
+
+			err := apiReader.Get(ctx, key, freshObject) 
+			if err != nil {
+				logger.Info(fmt.Sprintf("failed to refresh resource version during %s patch retry", operationType),
+					"attempt", attempt,
+					"object", key,
+					"error", err.Error())
+				return err
+			}
+
+			// Update the resource version on our object
+			obj.SetResourceVersion(freshObject.GetResourceVersion())
+		}
+
+		err := patchObject(ctx, kc, obj, patch, operationType)
+		if err == nil && attempt > 1 {
+			logger.Debug(fmt.Sprintf("%s patch succeeded after retry", operationType),
+				"attempts", attempt,
+				"object", client.ObjectKeyFromObject(obj))
+		}
+
+		if err != nil && attempt == 1 {
+			logger.Debug(fmt.Sprintf("%s patch failed on first attempt", operationType),
+				"object", client.ObjectKeyFromObject(obj),
+				"error", err.Error())
+		}
+
+		return err
+	})
+}
+
+// patchMetadataAndSpec performs a patch operation using client-go's standard retry mechanism on conflicts.
+func patchMetadataAndSpec(
+	ctx context.Context,
+	kc client.Client,
+	obj client.Object,
+	patch client.Patch,
+) error {
+	return patchObject(ctx, kc, obj, patch, OperationType_MetadataSpec)
+}
+
+// patchStatus performs a status patch operation using client-go's standard retry mechanism on conflicts.
+func patchStatus(
+	ctx context.Context,
+	kc client.Client,
+	apiReader client.Reader,
+	obj client.Object,
+	patch client.Patch,
+	logger acktypes.Logger,
+) error {
+	return patchWithRetry(ctx, kc, apiReader, obj, patch, logger, OperationType_Status)
 }