feat: allow backoff with a rate limiter in the work applier (#113)

Author: michaelawyu

cmd/memberagent/main.go

@@ -87,14 +87,21 @@ var (
 	propertyProvider             = flag.String("property-provider", "none", "The property provider to use for the agent.")
 	region                       = flag.String("region", "", "The region where the member cluster resides.")
 	cloudConfigFile              = flag.String("cloud-config", "/etc/kubernetes/provider/config.json", "The path to the cloud cloudconfig file.")
-	availabilityCheckInterval    = flag.Int("availability-check-interval", 5, "The interval in seconds between attempts to check for resource availability when resources are not yet available.")
-	driftDetectionInterval       = flag.Int("drift-detection-interval", 15, "The interval in seconds between attempts to detect configuration drifts in the cluster.")
 	watchWorkWithPriorityQueue   = flag.Bool("enable-watch-work-with-priority-queue", false, "If set, the apply_work controller will watch/reconcile work objects that are created new or have recent updates")
 	watchWorkReconcileAgeMinutes = flag.Int("watch-work-reconcile-age", 60, "maximum age (in minutes) of work objects for apply_work controller to watch/reconcile")
 	deletionWaitTime             = flag.Int("deletion-wait-time", 5, "The time the work-applier will wait for work object to be deleted before updating the applied work owner reference")
 	enablePprof                  = flag.Bool("enable-pprof", false, "enable pprof profiling")
 	pprofPort                    = flag.Int("pprof-port", 6065, "port for pprof profiling")
 	hubPprofPort                 = flag.Int("hub-pprof-port", 6066, "port for hub pprof profiling")
+	// Work applier requeue rate limiter settings.
+	workApplierRequeueRateLimiterAttemptsWithFixedDelay                              = flag.Int("work-applier-requeue-rate-limiter-attempts-with-fixed-delay", 1, "If set, the work applier will requeue work objects with a fixed delay for the specified number of attempts before switching to exponential backoff.")
+	workApplierRequeueRateLimiterFixedDelaySeconds                                   = flag.Float64("work-applier-requeue-rate-limiter-fixed-delay-seconds", 5.0, "If set, the work applier will requeue work objects with this fixed delay in seconds for the specified number of attempts before switching to exponential backoff.")
+	workApplierRequeueRateLimiterExponentialBaseForSlowBackoff                       = flag.Float64("work-applier-requeue-rate-limiter-exponential-base-for-slow-backoff", 1.2, "If set, the work applier will start to back off slowly at this factor after it finished requeueing with fixed delays, until it reaches the slow backoff delay cap. Its value should be larger than 1.0 and no larger than 100.0")
+	workApplierRequeueRateLimiterInitialSlowBackoffDelaySeconds                      = flag.Float64("work-applier-requeue-rate-limiter-initial-slow-backoff-delay-seconds", 2, "If set, the work applier will start to back off slowly at this delay in seconds.")
+	workApplierRequeueRateLimiterMaxSlowBackoffDelaySeconds                          = flag.Float64("work-applier-requeue-rate-limiter-max-slow-backoff-delay-seconds", 15, "If set, the work applier will not back off longer than this value in seconds when it is in the slow backoff stage.")
+	workApplierRequeueRateLimiterExponentialBaseForFastBackoff                       = flag.Float64("work-applier-requeue-rate-limiter-exponential-base-for-fast-backoff", 1.2, "If set, the work applier will start to back off fast at this factor after it completes the slow backoff stage, until it reaches the fast backoff delay cap. Its value should be larger than the base value for the slow backoff stage.")
+	workApplierRequeueRateLimiterMaxFastBackoffDelaySeconds                          = flag.Float64("work-applier-requeue-rate-limiter-max-fast-backoff-delay-seconds", 900, "If set, the work applier will not back off longer than this value in seconds when it is in the fast backoff stage.")
+	workApplierRequeueRateLimiterSkipToFastBackoffForAvailableOrDiffReportedWorkObjs = flag.Bool("work-applier-requeue-rate-limiter-skip-to-fast-backoff-for-available-or-diff-reported-work-objs", true, "If set, the rate limiter will skip the slow backoff stage and start fast backoff immediately for work objects that are available or have diff reported.")
 )
 
 func init() {
@@ -382,6 +389,45 @@ func Start(ctx context.Context, hubCfg, memberConfig *rest.Config, hubOpts, memb
 			return err
 		}
 		// create the work controller, so we can pass it to the internal member cluster reconciler
+
+		// Set up the requeue rate limiter for the work applier.
+		//
+		// With default settings, the rate limiter will:
+		// * allow 1 attempt of fixed delay; this helps give objects a bit of headroom to get available (or have
+		//   diffs reported).
+		// * use a fixed delay of 5 seconds for the first attempt.
+		//
+		//   Important (chenyu1): before the introduction of the requeue rate limiter, the work
+		//   applier uses static requeue intervals, specifically 5 seconds (if the work object is unavailable),
+		//   and 15 seconds (if the work object is available). There are a number of test cases that
+		//   implicitly assume this behavior (e.g., a test case might expect that the availability check completes
+		//   w/in 10 seconds), which is why the rate limiter uses the 5 seconds fixed requeue delay by default.
+		//   If you need to change this value and see that some test cases begin to fail, update the test
+		//   cases accordingly.
+		// * after completing all attempts with fixed delay, switch to slow exponential backoff with a base of
+		//   1.2 with an initial delay of 2 seconds and a cap of 15 seconds (12 requeues in total, ~90 seconds in total);
+		//   this is to allow fast checkups in cases where objects are not yet available or have not yet reported diffs.
+		// * after completing the slow backoff stage, switch to a fast exponential backoff with a base of 1.5
+		//   with an initial delay of 15 seconds and a cap of 15 minutes (10 requeues in total, ~42 minutes in total).
+		// * for Work objects that are available or have diffs reported, skip the slow backoff stage and
+		//   start fast backoff immediately.
+		//
+		// The requeue pattern is essentially:
+		// * 1 attempts of requeue with fixed delay (5 seconds); then
+		// * 12 attempts of requeues with slow exponential backoff (factor of 1.2, ~90 seconds in total); then
+		// * 10 attempts of requeues with fast exponential backoff (factor of 1.5, ~42 minutes in total);
+		// * afterwards, requeue with a delay of 15 minutes indefinitely.
+		requeueRateLimiter := workapplier.NewRequeueMultiStageWithExponentialBackoffRateLimiter(
+			*workApplierRequeueRateLimiterAttemptsWithFixedDelay,
+			*workApplierRequeueRateLimiterFixedDelaySeconds,
+			*workApplierRequeueRateLimiterExponentialBaseForSlowBackoff,
+			*workApplierRequeueRateLimiterInitialSlowBackoffDelaySeconds,
+			*workApplierRequeueRateLimiterMaxSlowBackoffDelaySeconds,
+			*workApplierRequeueRateLimiterExponentialBaseForFastBackoff,
+			*workApplierRequeueRateLimiterMaxFastBackoffDelaySeconds,
+			*workApplierRequeueRateLimiterSkipToFastBackoffForAvailableOrDiffReportedWorkObjs,
+		)
+
 		workController := workapplier.NewReconciler(
 			hubMgr.GetClient(),
 			targetNS,
@@ -394,11 +440,10 @@ func Start(ctx context.Context, hubCfg, memberConfig *rest.Config, hubOpts, memb
 			5,
 			// Use the default worker count (4) for parallelized manifest processing.
 			parallelizer.DefaultNumOfWorkers,
-			time.Second*time.Duration(*availabilityCheckInterval),
-			time.Second*time.Duration(*driftDetectionInterval),
 			time.Minute*time.Duration(*deletionWaitTime),
 			*watchWorkWithPriorityQueue,
 			*watchWorkReconcileAgeMinutes,
+			requeueRateLimiter,
 		)
 
 		if err = workController.SetupWithManager(hubMgr); err != nil {

pkg/controllers/internalmembercluster/v1beta1/member_suite_test.go

@@ -379,7 +379,7 @@ var _ = BeforeSuite(func() {
 
 	// This controller is created for testing purposes only; no reconciliation loop is actually
 	// run.
-	workApplier1 = workapplier.NewReconciler(hubClient, member1ReservedNSName, nil, nil, nil, nil, 0, 1, time.Second*5, time.Second*5, time.Minute, true, 60)
+	workApplier1 = workapplier.NewReconciler(hubClient, member1ReservedNSName, nil, nil, nil, nil, 0, 1, time.Minute, false, 60, nil)
 
 	propertyProvider1 = &manuallyUpdatedProvider{}
 	member1Reconciler, err := NewReconciler(ctx, hubClient, member1Cfg, member1Client, workApplier1, propertyProvider1)
@@ -402,7 +402,7 @@ var _ = BeforeSuite(func() {
 
 	// This controller is created for testing purposes only; no reconciliation loop is actually
 	// run.
-	workApplier2 = workapplier.NewReconciler(hubClient, member2ReservedNSName, nil, nil, nil, nil, 0, 1, time.Second*5, time.Second*5, time.Minute, true, 60)
+	workApplier2 = workapplier.NewReconciler(hubClient, member2ReservedNSName, nil, nil, nil, nil, 0, 1, time.Minute, false, 60, nil)
 
 	member2Reconciler, err := NewReconciler(ctx, hubClient, member2Cfg, member2Client, workApplier2, nil)
 	Expect(err).NotTo(HaveOccurred())