Exponential backoff for reconciler requests

pkg/controllers/accesslogpolicy_controller.go

@@ -18,12 +18,13 @@ package controllers
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"reflect"
 
 	"golang.org/x/exp/slices"
 	corev1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/api/errors"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
@@ -116,8 +117,8 @@ func (r *accessLogPolicyReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 	res, retryErr := lattice_runtime.HandleReconcileError(recErr)
 	if res.RequeueAfter != 0 {
 		r.log.Infow(ctx, "requeue request", "name", req.Name, "requeueAfter", res.RequeueAfter)
-	} else if res.RequeueAfter == 0 && retryErr != nil {
-		r.log.Infow(ctx, "requeue request", "name", req.Name)
+	} else if retryErr != nil && !errors.Is(retryErr, reconcile.TerminalError(nil)) {
+		r.log.Infow(ctx, "requeue request using exponential backoff", "name", req.Name)
 	} else if retryErr == nil {
 		r.log.Infow(ctx, "reconciled", "name", req.Name)
 	}
@@ -250,7 +251,7 @@ func (r *accessLogPolicyReconciler) targetRefExists(ctx context.Context, alp *an
 		return false, fmt.Errorf("access Log Policy targetRef is for unsupported Kind: %s", alp.Spec.TargetRef.Kind)
 	}
 
-	if err != nil && !errors.IsNotFound(err) {
+	if err != nil && !apierrors.IsNotFound(err) {
 		return false, err
 	}
 

pkg/controllers/gateway_controller.go

@@ -18,6 +18,7 @@ package controllers
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	anv1alpha1 "github.com/aws/aws-application-networking-k8s/pkg/apis/applicationnetworking/v1alpha1"
@@ -32,13 +33,14 @@ import (
 	"github.com/aws/aws-application-networking-k8s/pkg/utils"
 	"github.com/aws/aws-application-networking-k8s/pkg/utils/gwlog"
 
-	"github.com/pkg/errors"
+	pkgerrors "github.com/pkg/errors"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/tools/record"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	gwv1 "sigs.k8s.io/gateway-api/apis/v1"
 
 	deploy "github.com/aws/aws-application-networking-k8s/pkg/deploy/lattice"
@@ -130,8 +132,8 @@ func (r *gatewayReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 	res, retryErr := lattice_runtime.HandleReconcileError(recErr)
 	if res.RequeueAfter != 0 {
 		r.log.Infow(ctx, "requeue request", "name", req.Name, "requeueAfter", res.RequeueAfter)
-	} else if res.RequeueAfter == 0 && retryErr != nil {
-		r.log.Infow(ctx, "requeue request", "name", req.Name)
+	} else if retryErr != nil && !errors.Is(retryErr, reconcile.TerminalError(nil)) {
+		r.log.Infow(ctx, "requeue request using exponential backoff", "name", req.Name)
 	} else if retryErr == nil {
 		r.log.Infow(ctx, "reconciled", "name", req.Name)
 	}
@@ -194,7 +196,7 @@ func (r *gatewayReconciler) reconcileUpsert(ctx context.Context, gw *gwv1.Gatewa
 	if err != nil {
 		err2 := r.updateGatewayAcceptStatus(ctx, gw, false)
 		if err2 != nil {
-			return errors.Wrap(err2, err.Error())
+			return pkgerrors.Wrap(err2, err.Error())
 		}
 	}
 
@@ -420,7 +422,7 @@ func UpdateGWListenerStatus(ctx context.Context, k8sClient client.Client, gw *gw
 	}
 
 	if err := k8sClient.Status().Patch(ctx, gw, client.MergeFrom(gwOld)); err != nil {
-		return errors.Wrapf(err, "listener update failed")
+		return pkgerrors.Wrapf(err, "listener update failed")
 	}
 
 	if hasValidListener {

pkg/controllers/route_controller.go

@@ -46,7 +46,6 @@ import (
 	"github.com/aws/aws-application-networking-k8s/pkg/controllers/eventhandlers"
 	"github.com/aws/aws-application-networking-k8s/pkg/controllers/predicates"
 	"github.com/aws/aws-application-networking-k8s/pkg/deploy"
-	"github.com/aws/aws-application-networking-k8s/pkg/deploy/lattice"
 	"github.com/aws/aws-application-networking-k8s/pkg/gateway"
 	"github.com/aws/aws-application-networking-k8s/pkg/k8s"
 	"github.com/aws/aws-application-networking-k8s/pkg/model/core"
@@ -262,7 +261,8 @@ func (r *routeReconciler) buildAndDeployModel(
 	r.log.Debugf(ctx, "stack: %s", json)
 
 	if err := r.stackDeployer.Deploy(ctx, stack); err != nil {
-		if errors.As(err, &lattice.RetryErr) {
+		var requeueNeededAfter *lattice_runtime.RequeueNeededAfter
+		if errors.As(err, &requeueNeededAfter) {
 			r.eventRecorder.Event(route.K8sObject(), corev1.EventTypeNormal,
 				k8s.RouteEventReasonRetryReconcile, "retry reconcile...")
 		} else {

pkg/controllers/serviceexport_controller.go

@@ -27,6 +27,7 @@ import (
 	"k8s.io/client-go/tools/record"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	anv1alpha1 "github.com/aws/aws-application-networking-k8s/pkg/apis/applicationnetworking/v1alpha1"
 	"github.com/aws/aws-application-networking-k8s/pkg/aws"
@@ -111,7 +112,16 @@ func (r *serviceExportReconciler) Reconcile(ctx context.Context, req ctrl.Reques
 	if recErr != nil {
 		r.log.Infow(ctx, "reconcile error", "name", req.Name, "message", recErr.Error())
 	}
-	return lattice_runtime.HandleReconcileError(recErr)
+
+	res, retryErr := lattice_runtime.HandleReconcileError(recErr)
+	if res.RequeueAfter != 0 {
+		r.log.Infow(ctx, "requeue request", "name", req.Name, "requeueAfter", res.RequeueAfter)
+	} else if retryErr != nil && !errors.Is(retryErr, reconcile.TerminalError(nil)) {
+		r.log.Infow(ctx, "requeue request using exponential backoff", "name", req.Name)
+	} else if retryErr == nil {
+		r.log.Infow(ctx, "reconciled", "name", req.Name)
+	}
+	return res, retryErr
 }
 
 func (r *serviceExportReconciler) reconcile(ctx context.Context, req ctrl.Request) error {

pkg/deploy/lattice/service_manager.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 
 	"github.com/aws/aws-application-networking-k8s/pkg/aws/services"
+	lattice_runtime "github.com/aws/aws-application-networking-k8s/pkg/runtime"
 	"github.com/aws/aws-application-networking-k8s/pkg/utils/gwlog"
 
 	"github.com/aws/aws-sdk-go/aws"
@@ -251,12 +252,12 @@ func (m *defaultServiceManager) updateAssociations(ctx context.Context, svc *Ser
 	return nil
 }
 
-// returns RetryErr on all non-active Sn-Svc association responses
+// returns RetryError on all non-active Sn-Svc association responses
 func handleCreateAssociationResp(resp *CreateSnSvcAssocResp) error {
 	status := aws.StringValue(resp.Status)
 	if status != vpclattice.ServiceNetworkServiceAssociationStatusActive {
 		return fmt.Errorf("%w: sn-service-association-id: %s, non-active status: %s",
-			RetryErr, aws.StringValue(resp.Id), status)
+			lattice_runtime.NewRetryError(), aws.StringValue(resp.Id), status)
 	}
 	return nil
 }
@@ -288,7 +289,7 @@ func associationsDiff(svc *Service, curAssocs []*SnSvcAssocSummary) ([]string, [
 		// TODO: we should have something more lightweight, retrying full reconciliation looks to heavy
 		if aws.StringValue(oldSn.Status) == vpclattice.ServiceNetworkServiceAssociationStatusDeleteInProgress {
 			return nil, nil, fmt.Errorf("%w: want to associate sn: %s to svc: %s, but status is: %s",
-				RetryErr, newSn, svc.LatticeServiceName(), *oldSn.Status)
+				lattice_runtime.NewRetryError(), newSn, svc.LatticeServiceName(), *oldSn.Status)
 		}
 		// TODO: if assoc in failed state, may be we should try to re-create?
 	}

pkg/deploy/lattice/service_manager_test.go

@@ -9,6 +9,7 @@ import (
 	mocks "github.com/aws/aws-application-networking-k8s/pkg/aws/services"
 	"github.com/aws/aws-application-networking-k8s/pkg/model/core"
 	model "github.com/aws/aws-application-networking-k8s/pkg/model/lattice"
+	lattice_runtime "github.com/aws/aws-application-networking-k8s/pkg/runtime"
 	"github.com/aws/aws-application-networking-k8s/pkg/utils/gwlog"
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/vpclattice"
@@ -508,7 +509,8 @@ func TestHandleSnSvcAssocResp(t *testing.T) {
 			Status: aws.String(vpclattice.ServiceNetworkServiceAssociationStatusCreateInProgress),
 		}
 		err := handleCreateAssociationResp(resp)
-		assert.True(t, errors.Is(err, RetryErr))
+		var requeueNeededAfter *lattice_runtime.RequeueNeededAfter
+		assert.True(t, errors.As(err, &requeueNeededAfter))
 	})
 
 }
@@ -569,7 +571,8 @@ func TestSnSvcAssocsDiff(t *testing.T) {
 			Status:             aws.String(vpclattice.ServiceNetworkServiceAssociationStatusDeleteInProgress),
 		}}
 		_, _, err := associationsDiff(svc, assocs)
-		assert.True(t, errors.Is(err, RetryErr))
+		var requeueNeededAfter *lattice_runtime.RequeueNeededAfter
+		assert.True(t, errors.As(err, &requeueNeededAfter))
 	})
 
 }

pkg/deploy/lattice/service_network_manager.go

@@ -2,12 +2,12 @@ package lattice
 
 import (
 	"context"
-	"errors"
 	"fmt"
 
 	"golang.org/x/exp/slices"
 
 	"github.com/aws/aws-application-networking-k8s/pkg/aws/services"
+	lattice_runtime "github.com/aws/aws-application-networking-k8s/pkg/runtime"
 	"github.com/aws/aws-application-networking-k8s/pkg/utils/gwlog"
 
 	"github.com/aws/aws-sdk-go/aws"
@@ -80,7 +80,7 @@ func (m *defaultServiceNetworkManager) UpsertVpcAssociation(ctx context.Context,
 		case vpclattice.ServiceNetworkVpcAssociationStatusActive:
 			return *resp.Arn, nil
 		default:
-			return *resp.Arn, fmt.Errorf("%w, vpc association status in %s", RetryErr, status)
+			return *resp.Arn, fmt.Errorf("%w, vpc association status in %s", lattice_runtime.NewRetryError(), status)
 		}
 	}
 }
@@ -125,7 +125,7 @@ func (m *defaultServiceNetworkManager) DeleteVpcAssociation(ctx context.Context,
 		if err != nil {
 			m.log.Infof(ctx, "Failed to delete association %s for %s, with response %s and err %s", *snva.Arn, snName, resp, err.Error())
 		}
-		return errors.New(LATTICE_RETRY)
+		return lattice_runtime.NewRetryError()
 	}
 	return nil
 }
@@ -163,7 +163,7 @@ func (m *defaultServiceNetworkManager) getActiveVpcAssociation(ctx context.Conte
 		return nil, nil
 	default:
 		// a mutation is in progress, try later
-		return nil, errors.New(LATTICE_RETRY)
+		return nil, lattice_runtime.NewRetryError()
 	}
 }
 
@@ -253,7 +253,7 @@ func (m *defaultServiceNetworkManager) updateServiceNetworkVpcAssociation(ctx co
 			SnvaSecurityGroupIds: updateSnvaResp.SecurityGroupIds,
 		}, nil
 	} else {
-		return model.ServiceNetworkStatus{}, fmt.Errorf("%w, update snva status: %s", RetryErr, *updateSnvaResp.Status)
+		return model.ServiceNetworkStatus{}, fmt.Errorf("%w, update snva status: %s", lattice_runtime.NewRetryError(), *updateSnvaResp.Status)
 	}
 }
 

pkg/deploy/lattice/service_network_manager_test.go

@@ -11,6 +11,7 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"github.com/aws/aws-application-networking-k8s/pkg/config"
+	lattice_runtime "github.com/aws/aws-application-networking-k8s/pkg/runtime"
 	"github.com/aws/aws-application-networking-k8s/pkg/utils/gwlog"
 
 	pkg_aws "github.com/aws/aws-application-networking-k8s/pkg/aws"
@@ -147,7 +148,8 @@ func Test_CreateOrUpdateServiceNetwork_SnAlreadyExist_ServiceNetworkVpcAssociati
 	resp, err := snMgr.CreateOrUpdate(ctx, &snCreateInput)
 
 	assert.NotNil(t, err)
-	assert.Equal(t, err, errors.New(LATTICE_RETRY))
+	var requeueNeededAfter *lattice_runtime.RequeueNeededAfter
+	assert.True(t, errors.As(err, &requeueNeededAfter))
 	assert.Equal(t, resp.ServiceNetworkARN, "")
 	assert.Equal(t, resp.ServiceNetworkID, "")
 }
@@ -199,7 +201,8 @@ func Test_CreateOrUpdateServiceNetwork_SnAlreadyExist_ServiceNetworkVpcAssociati
 	resp, err := snMgr.CreateOrUpdate(ctx, &snCreateInput)
 
 	assert.NotNil(t, err)
-	assert.Equal(t, err, errors.New(LATTICE_RETRY))
+	var requeueNeededAfter *lattice_runtime.RequeueNeededAfter
+	assert.True(t, errors.As(err, &requeueNeededAfter))
 	assert.Equal(t, resp.ServiceNetworkARN, "")
 	assert.Equal(t, resp.ServiceNetworkID, "")
 }
@@ -629,7 +632,8 @@ func Test_defaultServiceNetworkManager_CreateOrUpdate_SnExists_SnvaCreateInProgr
 	snMgr := NewDefaultServiceNetworkManager(gwlog.FallbackLogger, cloud)
 	_, err := snMgr.UpsertVpcAssociation(ctx, name, securityGroupIds)
 
-	assert.Equal(t, err, errors.New(LATTICE_RETRY))
+	var requeueNeededAfter *lattice_runtime.RequeueNeededAfter
+	assert.True(t, errors.As(err, &requeueNeededAfter))
 }
 
 func Test_defaultServiceNetworkManager_CreateOrUpdate_SnExists_SnvaExists_CannotUpdateSecurityGroupsFromNonemptyToEmpty(t *testing.T) {

pkg/deploy/lattice/target_group_manager.go

@@ -17,6 +17,7 @@ import (
 
 	pkg_aws "github.com/aws/aws-application-networking-k8s/pkg/aws"
 	model "github.com/aws/aws-application-networking-k8s/pkg/model/lattice"
+	lattice_runtime "github.com/aws/aws-application-networking-k8s/pkg/runtime"
 	"github.com/aws/aws-application-networking-k8s/pkg/utils/gwlog"
 )
 
@@ -115,7 +116,7 @@ func (s *defaultTargetGroupManager) create(ctx context.Context, modelTg *model.T
 		latticeTgStatus != vpclattice.TargetGroupStatusCreateInProgress {
 
 		s.log.Infof(ctx, "Target group is not in the desired state. State is %s, will retry", latticeTgStatus)
-		return model.TargetGroupStatus{}, errors.New(LATTICE_RETRY)
+		return model.TargetGroupStatus{}, lattice_runtime.NewRetryError()
 	}
 
 	// create-in-progress is considered success
@@ -219,7 +220,7 @@ func (s *defaultTargetGroupManager) Delete(ctx context.Context, modelTg *model.T
 
 	if drainCount > 0 {
 		// no point in trying to deregister may as well wait
-		return fmt.Errorf("cannot deregister targets for %s as %d targets are DRAINING", modelTg.Status.Id, drainCount)
+		return fmt.Errorf("%w: cannot deregister targets for %s as %d targets are DRAINING", lattice_runtime.NewRetryError(), modelTg.Status.Id, drainCount)
 	}
 
 	if len(targetsToDeregister) > 0 {
@@ -341,7 +342,7 @@ func (s *defaultTargetGroupManager) findTargetGroup(
 		if match {
 			switch status {
 			case vpclattice.TargetGroupStatusCreateInProgress, vpclattice.TargetGroupStatusDeleteInProgress:
-				return nil, errors.New(LATTICE_RETRY)
+				return nil, lattice_runtime.NewRetryError()
 			case vpclattice.TargetGroupStatusDeleteFailed, vpclattice.TargetGroupStatusActive:
 				return latticeTg, nil
 			}

pkg/deploy/lattice/target_group_manager_test.go

@@ -24,6 +24,7 @@ import (
 	"github.com/aws/aws-application-networking-k8s/pkg/config"
 	"github.com/aws/aws-application-networking-k8s/pkg/model/core"
 	model "github.com/aws/aws-application-networking-k8s/pkg/model/lattice"
+	lattice_runtime "github.com/aws/aws-application-networking-k8s/pkg/runtime"
 	"github.com/aws/aws-application-networking-k8s/pkg/utils/gwlog"
 )
 
@@ -363,7 +364,8 @@ func Test_CreateTargetGroup_ExistingTG_Status_Retry(t *testing.T) {
 			tgManager := NewTargetGroupManager(gwlog.FallbackLogger, cloud, nil)
 			_, err := tgManager.Upsert(ctx, &tgCreateInput)
 
-			assert.Equal(t, errors.New(LATTICE_RETRY), err)
+			var requeueNeededAfter *lattice_runtime.RequeueNeededAfter
+			assert.True(t, errors.As(err, &requeueNeededAfter))
 		})
 	}
 }
@@ -412,7 +414,8 @@ func Test_CreateTargetGroup_NewTG_RetryStatus(t *testing.T) {
 			tgManager := NewTargetGroupManager(gwlog.FallbackLogger, cloud, nil)
 			_, err := tgManager.Upsert(ctx, &tgCreateInput)
 
-			assert.Equal(t, errors.New(LATTICE_RETRY), err)
+			var requeueNeededAfter *lattice_runtime.RequeueNeededAfter
+			assert.True(t, errors.As(err, &requeueNeededAfter))
 		})
 	}
 }

pkg/deploy/lattice/target_group_synthesizer.go

@@ -60,7 +60,7 @@ func (t *TargetGroupSynthesizer) Synthesize(ctx context.Context) error {
 }
 func (t *TargetGroupSynthesizer) SynthesizeCreate(ctx context.Context) error {
 	var resTargetGroups []*model.TargetGroup
-	var returnErr = false
+	var firstError error
 
 	err := t.stack.ListResources(&resTargetGroups)
 	if err != nil {
@@ -91,12 +91,14 @@ func (t *TargetGroupSynthesizer) SynthesizeCreate(ctx context.Context) error {
 			resTargetGroup.Status = &tgStatus
 		} else {
 			t.log.Debugf(ctx, "Failed TargetGroupManager.Upsert %s due to %s", prefix, err)
-			returnErr = true
+			if firstError == nil {
+				firstError = err
+			}
 		}
 	}
 
-	if returnErr {
-		return fmt.Errorf("error during target group synthesis, will retry")
+	if firstError != nil {
+		return fmt.Errorf("error during target group synthesis, will retry: %w", firstError)
 	}
 
 	return nil
@@ -119,7 +121,7 @@ func (t *TargetGroupSynthesizer) SynthesizeDelete(ctx context.Context) error {
 		err := t.targetGroupManager.Delete(ctx, resTargetGroup)
 		if err != nil {
 			prefix := model.TgNamePrefix(resTargetGroup.Spec)
-			retErr = errors.Join(retErr, fmt.Errorf("failed TargetGroupManager.Delete %s due to %s", prefix, err))
+			retErr = errors.Join(retErr, fmt.Errorf("failed TargetGroupManager.Delete %s due to %w", prefix, err))
 		}
 	}