Add support for applying jitter when requeuing resources after reconcile

pkg/reconciler/managed/reconciler.go

@@ -18,6 +18,7 @@ package managed
 
 import (
 	"context"
+	"math/rand"
 	"strings"
 	"time"
 
@@ -45,7 +46,7 @@ const (
 	reconcileGracePeriod = 30 * time.Second
 	reconcileTimeout     = 1 * time.Minute
 
-	defaultpollInterval = 1 * time.Minute
+	defaultPollInterval = 1 * time.Minute
 	defaultGracePeriod  = 30 * time.Second
 )
 
@@ -467,7 +468,9 @@ type Reconciler struct {
 	client     client.Client
 	newManaged func() resource.Managed
 
-	pollInterval        time.Duration
+	pollInterval     time.Duration
+	pollIntervalHook PollIntervalHook
+
 	timeout             time.Duration
 	creationGracePeriod time.Duration
 
@@ -541,6 +544,36 @@ func WithPollInterval(after time.Duration) ReconcilerOption {
 	}
 }
 
+// PollIntervalHook represents the function type passed to the
+// WithPollIntervalHook option to support dynamic computation of the poll
+// interval.
+type PollIntervalHook func(managed resource.Managed, pollInterval time.Duration) time.Duration
+
+func defaultPollIntervalHook(_ resource.Managed, pollInterval time.Duration) time.Duration {
+	return pollInterval
+}
+
+// WithPollIntervalHook adds a hook that can be used to configure the
+// delay before an up-to-date resource is reconciled again after a successful
+// reconcile. If this option is passed multiple times, only the latest hook
+// will be used.
+func WithPollIntervalHook(hook PollIntervalHook) ReconcilerOption {
+	return func(r *Reconciler) {
+		r.pollIntervalHook = hook
+	}
+}
+
+// WithPollJitterHook adds a simple PollIntervalHook to add jitter to the poll
+// interval used when queuing a new reconciliation after a successful
+// reconcile. The added jitter will be a random duration between -jitter and
+// +jitter. This option wraps WithPollIntervalHook, and is subject to the same
+// constraint that only the latest hook will be used.
+func WithPollJitterHook(jitter time.Duration) ReconcilerOption {
+	return WithPollIntervalHook(func(managed resource.Managed, pollInterval time.Duration) time.Duration {
+		return pollInterval + time.Duration((rand.Float64()-0.5)*2*float64(jitter)) //#nosec G404 -- no need for secure randomness
+	})
+}
+
 // WithCreationGracePeriod configures an optional period during which we will
 // wait for the external API to report that a newly created external resource
 // exists. This allows us to tolerate eventually consistent APIs that do not
@@ -658,7 +691,8 @@ func NewReconciler(m manager.Manager, of resource.ManagedKind, o ...ReconcilerOp
 	r := &Reconciler{
 		client:                      m.GetClient(),
 		newManaged:                  nm,
-		pollInterval:                defaultpollInterval,
+		pollInterval:                defaultPollInterval,
+		pollIntervalHook:            defaultPollIntervalHook,
 		creationGracePeriod:         defaultGracePeriod,
 		timeout:                     reconcileTimeout,
 		managed:                     defaultMRManaged(m),
@@ -1080,9 +1114,10 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (resu
 		// after the specified poll interval in order to observe it and react
 		// accordingly.
 		// https://github.com/crossplane/crossplane/issues/289
-		log.Debug("External resource is up to date", "requeue-after", time.Now().Add(r.pollInterval))
+		reconcileAfter := r.pollIntervalHook(managed, r.pollInterval)
+		log.Debug("External resource is up to date", "requeue-after", time.Now().Add(reconcileAfter))
 		managed.SetConditions(xpv1.ReconcileSuccess())
-		return reconcile.Result{RequeueAfter: r.pollInterval}, errors.Wrap(r.client.Status().Update(ctx, managed), errUpdateManagedStatus)
+		return reconcile.Result{RequeueAfter: reconcileAfter}, errors.Wrap(r.client.Status().Update(ctx, managed), errUpdateManagedStatus)
 	}
 
 	if observation.Diff != "" {
@@ -1091,9 +1126,10 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (resu
 
 	// skip the update if the management policy is set to ignore updates
 	if !policy.ShouldUpdate() {
-		log.Debug("Skipping update due to managementPolicies. Reconciliation succeeded", "requeue-after", time.Now().Add(r.pollInterval))
+		reconcileAfter := r.pollIntervalHook(managed, r.pollInterval)
+		log.Debug("Skipping update due to managementPolicies. Reconciliation succeeded", "requeue-after", time.Now().Add(reconcileAfter))
 		managed.SetConditions(xpv1.ReconcileSuccess())
-		return reconcile.Result{RequeueAfter: r.pollInterval}, errors.Wrap(r.client.Status().Update(ctx, managed), errUpdateManagedStatus)
+		return reconcile.Result{RequeueAfter: reconcileAfter}, errors.Wrap(r.client.Status().Update(ctx, managed), errUpdateManagedStatus)
 	}
 
 	update, err := external.Update(externalCtx, managed)
@@ -1124,8 +1160,9 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (resu
 	// changes, so we requeue a speculative reconcile after the specified poll
 	// interval in order to observe it and react accordingly.
 	// https://github.com/crossplane/crossplane/issues/289
-	log.Debug("Successfully requested update of external resource", "requeue-after", time.Now().Add(r.pollInterval))
+	reconcileAfter := r.pollIntervalHook(managed, r.pollInterval)
+	log.Debug("Successfully requested update of external resource", "requeue-after", time.Now().Add(reconcileAfter))
 	record.Event(managed, event.Normal(reasonUpdated, "Successfully requested update of external resource"))
 	managed.SetConditions(xpv1.ReconcileSuccess())
-	return reconcile.Result{RequeueAfter: r.pollInterval}, errors.Wrap(r.client.Status().Update(ctx, managed), errUpdateManagedStatus)
+	return reconcile.Result{RequeueAfter: reconcileAfter}, errors.Wrap(r.client.Status().Update(ctx, managed), errUpdateManagedStatus)
 }

pkg/reconciler/managed/reconciler_test.go

@@ -50,8 +50,9 @@ func TestReconciler(t *testing.T) {
 	}
 
 	type want struct {
-		result reconcile.Result
-		err    error
+		result        reconcile.Result
+		resultCmpOpts []cmp.Option
+		err           error
 	}
 
 	errBoom := errors.New("boom")
@@ -324,7 +325,7 @@ func TestReconciler(t *testing.T) {
 					WithFinalizer(resource.FinalizerFns{AddFinalizerFn: func(_ context.Context, _ resource.Object) error { return nil }}),
 				},
 			},
-			want: want{result: reconcile.Result{RequeueAfter: defaultpollInterval}},
+			want: want{result: reconcile.Result{RequeueAfter: defaultPollInterval}},
 		},
 		"ExternalObserveError": {
 			reason: "Errors observing the external resource should trigger a requeue after a short wait.",
@@ -949,7 +950,121 @@ func TestReconciler(t *testing.T) {
 					WithFinalizer(resource.FinalizerFns{AddFinalizerFn: func(_ context.Context, _ resource.Object) error { return nil }}),
 				},
 			},
-			want: want{result: reconcile.Result{RequeueAfter: defaultpollInterval}},
+			want: want{result: reconcile.Result{RequeueAfter: defaultPollInterval}},
+		},
+		"ExternalResourceUpToDateWithJitter": {
+			reason: "When the external resource exists and is up to date a requeue should be triggered after a long wait with jitter added.",
+			args: args{
+				m: &fake.Manager{
+					Client: &test.MockClient{
+						MockGet: test.NewMockGetFn(nil),
+						MockStatusUpdate: test.MockSubResourceUpdateFn(func(_ context.Context, _ client.Object, _ ...client.SubResourceUpdateOption) error {
+							return nil
+						}),
+					},
+					Scheme: fake.SchemeWith(&fake.Managed{}),
+				},
+				mg: resource.ManagedKind(fake.GVK(&fake.Managed{})),
+				o: []ReconcilerOption{
+					WithInitializers(),
+					WithReferenceResolver(ReferenceResolverFn(func(_ context.Context, _ resource.Managed) error { return nil })),
+					WithExternalConnecter(ExternalConnectorFn(func(_ context.Context, mg resource.Managed) (ExternalClient, error) {
+						c := &ExternalClientFns{
+							ObserveFn: func(_ context.Context, _ resource.Managed) (ExternalObservation, error) {
+								return ExternalObservation{ResourceExists: true, ResourceUpToDate: true}, nil
+							},
+						}
+						return c, nil
+					})),
+					WithConnectionPublishers(),
+					WithFinalizer(resource.FinalizerFns{AddFinalizerFn: func(_ context.Context, _ resource.Object) error { return nil }}),
+					WithPollJitterHook(time.Second),
+				},
+			},
+			want: want{
+				result: reconcile.Result{RequeueAfter: defaultPollInterval},
+				resultCmpOpts: []cmp.Option{cmp.Comparer(func(l, r time.Duration) bool {
+					diff := l - r
+					if diff < 0 {
+						diff = -diff
+					}
+					return diff < time.Second
+				})},
+			},
+		},
+		"ExternalResourceUpToDateWithPollIntervalHook": {
+			reason: "When the external resource exists and is up to date a requeue should be triggered after a long wait processed by the interval hook.",
+			args: args{
+				m: &fake.Manager{
+					Client: &test.MockClient{
+						MockGet: test.NewMockGetFn(nil),
+						MockStatusUpdate: test.MockSubResourceUpdateFn(func(_ context.Context, _ client.Object, _ ...client.SubResourceUpdateOption) error {
+							return nil
+						}),
+					},
+					Scheme: fake.SchemeWith(&fake.Managed{}),
+				},
+				mg: resource.ManagedKind(fake.GVK(&fake.Managed{})),
+				o: []ReconcilerOption{
+					WithInitializers(),
+					WithReferenceResolver(ReferenceResolverFn(func(_ context.Context, _ resource.Managed) error { return nil })),
+					WithExternalConnecter(ExternalConnectorFn(func(_ context.Context, mg resource.Managed) (ExternalClient, error) {
+						c := &ExternalClientFns{
+							ObserveFn: func(_ context.Context, _ resource.Managed) (ExternalObservation, error) {
+								return ExternalObservation{ResourceExists: true, ResourceUpToDate: true}, nil
+							},
+						}
+						return c, nil
+					})),
+					WithConnectionPublishers(),
+					WithFinalizer(resource.FinalizerFns{AddFinalizerFn: func(_ context.Context, _ resource.Object) error { return nil }}),
+					WithPollIntervalHook(func(managed resource.Managed, pollInterval time.Duration) time.Duration {
+						return 2 * pollInterval
+					}),
+				},
+			},
+			want: want{
+				result: reconcile.Result{RequeueAfter: 2 * defaultPollInterval},
+			},
+		},
+		"ExternalResourceUpToDateWithMultiplePollIntervalHooks": {
+			reason: "When the external resource exists and is up to date a requeue should be triggered after a long wait processed by the latest interval hook.",
+			args: args{
+				m: &fake.Manager{
+					Client: &test.MockClient{
+						MockGet: test.NewMockGetFn(nil),
+						MockStatusUpdate: test.MockSubResourceUpdateFn(func(_ context.Context, _ client.Object, _ ...client.SubResourceUpdateOption) error {
+							return nil
+						}),
+					},
+					Scheme: fake.SchemeWith(&fake.Managed{}),
+				},
+				mg: resource.ManagedKind(fake.GVK(&fake.Managed{})),
+				o: []ReconcilerOption{
+					WithInitializers(),
+					WithReferenceResolver(ReferenceResolverFn(func(_ context.Context, _ resource.Managed) error { return nil })),
+					WithExternalConnecter(ExternalConnectorFn(func(_ context.Context, mg resource.Managed) (ExternalClient, error) {
+						c := &ExternalClientFns{
+							ObserveFn: func(_ context.Context, _ resource.Managed) (ExternalObservation, error) {
+								return ExternalObservation{ResourceExists: true, ResourceUpToDate: true}, nil
+							},
+						}
+						return c, nil
+					})),
+					WithConnectionPublishers(),
+					WithFinalizer(resource.FinalizerFns{AddFinalizerFn: func(_ context.Context, _ resource.Object) error { return nil }}),
+					WithPollJitterHook(time.Second),
+					WithPollIntervalHook(func(managed resource.Managed, pollInterval time.Duration) time.Duration {
+						return 2 * pollInterval
+					}),
+					WithPollIntervalHook(func(managed resource.Managed, pollInterval time.Duration) time.Duration {
+						return 3 * pollInterval
+					}),
+				},
+			},
+			want: want{
+				result: reconcile.Result{RequeueAfter: 3 * defaultPollInterval},
+			},
 		},
 		"UpdateExternalError": {
 			reason: "Errors while updating an external resource should trigger a requeue after a short wait.",
@@ -1077,7 +1192,7 @@ func TestReconciler(t *testing.T) {
 					WithFinalizer(resource.FinalizerFns{AddFinalizerFn: func(_ context.Context, _ resource.Object) error { return nil }}),
 				},
 			},
-			want: want{result: reconcile.Result{RequeueAfter: defaultpollInterval}},
+			want: want{result: reconcile.Result{RequeueAfter: defaultPollInterval}},
 		},
 		"ReconciliationPausedSuccessful": {
 			reason: `If a managed resource has the pause annotation with value "true", there should be no further requeue requests.`,
@@ -1182,7 +1297,7 @@ func TestReconciler(t *testing.T) {
 					WithFinalizer(resource.FinalizerFns{AddFinalizerFn: func(_ context.Context, _ resource.Object) error { return nil }}),
 				},
 			},
-			want: want{result: reconcile.Result{RequeueAfter: defaultpollInterval}},
+			want: want{result: reconcile.Result{RequeueAfter: defaultPollInterval}},
 		},
 		"ReconciliationPausedError": {
 			reason: `If a managed resource has the pause annotation with value "true" and the status update due to reconciliation being paused fails, error should be reported causing an exponentially backed-off requeue.`,
@@ -1418,7 +1533,7 @@ func TestReconciler(t *testing.T) {
 					WithFinalizer(resource.FinalizerFns{AddFinalizerFn: func(_ context.Context, _ resource.Object) error { return nil }}),
 				},
 			},
-			want: want{result: reconcile.Result{RequeueAfter: defaultpollInterval}},
+			want: want{result: reconcile.Result{RequeueAfter: defaultPollInterval}},
 		},
 		"ManagementPolicyAllCreateSuccessful": {
 			reason: "Successful managed resource creation using management policy all should trigger a requeue after a short wait.",
@@ -1544,7 +1659,7 @@ func TestReconciler(t *testing.T) {
 					WithFinalizer(resource.FinalizerFns{AddFinalizerFn: func(_ context.Context, _ resource.Object) error { return nil }}),
 				},
 			},
-			want: want{result: reconcile.Result{RequeueAfter: defaultpollInterval}},
+			want: want{result: reconcile.Result{RequeueAfter: defaultPollInterval}},
 		},
 		"ManagementPolicyAllUpdateSuccessful": {
 			reason: "A successful managed resource update using management policies should trigger a requeue after a long wait.",
@@ -1589,7 +1704,7 @@ func TestReconciler(t *testing.T) {
 					WithFinalizer(resource.FinalizerFns{AddFinalizerFn: func(_ context.Context, _ resource.Object) error { return nil }}),
 				},
 			},
-			want: want{result: reconcile.Result{RequeueAfter: defaultpollInterval}},
+			want: want{result: reconcile.Result{RequeueAfter: defaultPollInterval}},
 		},
 		"ManagementPolicyUpdateUpdateSuccessful": {
 			reason: "A successful managed resource update using management policies should trigger a requeue after a long wait.",
@@ -1634,7 +1749,7 @@ func TestReconciler(t *testing.T) {
 					WithFinalizer(resource.FinalizerFns{AddFinalizerFn: func(_ context.Context, _ resource.Object) error { return nil }}),
 				},
 			},
-			want: want{result: reconcile.Result{RequeueAfter: defaultpollInterval}},
+			want: want{result: reconcile.Result{RequeueAfter: defaultPollInterval}},
 		},
 		"ManagementPolicySkipLateInitialize": {
 			reason: "Should skip updating a managed resource to persist late initialized fields and should trigger a requeue after a long wait.",
@@ -1677,7 +1792,7 @@ func TestReconciler(t *testing.T) {
 					WithFinalizer(resource.FinalizerFns{AddFinalizerFn: func(_ context.Context, _ resource.Object) error { return nil }}),
 				},
 			},
-			want: want{result: reconcile.Result{RequeueAfter: defaultpollInterval}},
+			want: want{result: reconcile.Result{RequeueAfter: defaultPollInterval}},
 		},
 	}
 
@@ -1690,7 +1805,7 @@ func TestReconciler(t *testing.T) {
 				t.Errorf("\nReason: %s\nr.Reconcile(...): -want error, +got error:\n%s", tc.reason, diff)
 			}
 
-			if diff := cmp.Diff(tc.want.result, got); diff != "" {
+			if diff := cmp.Diff(tc.want.result, got, tc.want.resultCmpOpts...); diff != "" {
 				t.Errorf("\nReason: %s\nr.Reconcile(...): -want, +got:\n%s", tc.reason, diff)
 			}
 		})