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/config"
 "github.com/aws/aws-application-networking-k8s/pkg/controllers/eventhandlers"
 "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"
@@ -261,7 +260,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 {
@@ -366,7 +366,7 @@ func (r *routeReconciler) reconcileUpsert(ctx context.Context, req ctrl.Request,
 
 if svc == nil || svc.DnsEntry == nil || svc.DnsEntry.DomainName == nil {
 r.log.Infof(ctx, "Either service, dns entry, or domain name is not available. Will Retry")
-return errors.New(lattice.LATTICE_RETRY)
+return lattice_runtime.NewRetryError()
 }
 
 if err := r.updateRouteAnnotation(ctx, *svc.DnsEntry.DomainName, route); err != nil {

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"
@@ -244,12 +245,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
 }
@@ -281,7 +282,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"
@@ -398,7 +399,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))
 })
 
 }
@@ -459,7 +461,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
@@ -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

pkg/deploy/lattice/targets_synthesizer.go

@@ -11,6 +11,7 @@ import (
 
 "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"
 "github.com/aws/aws-application-networking-k8s/pkg/utils/gwlog"
 "github.com/aws/aws-application-networking-k8s/pkg/webhook"
@@ -107,7 +108,7 @@ func (t *targetsSynthesizer) PostSynthesize(ctx context.Context) error {
 }
 
 if requeueNeeded {
-return fmt.Errorf("%w: target status still in pending", RetryErr)
+return fmt.Errorf("%w: target status still in pending", lattice_runtime.NewRetryError())
 }
 return nil
 }