Work on dynamic poller fixes (#519)

This apparently is still broken, working on a fix now

Author: michaeljguarino

internal/helpers/poll.go

@@ -14,13 +14,13 @@ import (
 // If syncFirstRun is set to true, it will execute the condition function synchronously first and then start
 // polling. Since error is returned synchronously, the only way to check for it is to use syncFirstRun.
 // Background poller does not sync errors. It can be stopped externally by cancelling the provided context.
-func DynamicBackgroundPollUntilContextCancel(ctx context.Context, getInterval func() time.Duration, syncFirstRun bool, condition wait.ConditionWithContextFunc) (err error) {
+func DynamicBackgroundPollUntilContextCancel(ctx context.Context, getInterval func() time.Duration, syncFirstRun bool, callback wait.ConditionWithContextFunc) (err error) {
 	if syncFirstRun {
-		_, err = condition(ctx)
+		_, err = callback(ctx)
 	}
 
 	go func() {
-		_ = DynamicPollUntilContextCancel(ctx, getInterval, condition)
+		_ = DynamicPollUntilContextCancel(ctx, getInterval, callback)
 	}()
 
 	return err
@@ -29,37 +29,34 @@ func DynamicBackgroundPollUntilContextCancel(ctx context.Context, getInterval fu
 func DynamicPollUntilContextCancel(
 	ctx context.Context,
 	intervalFunc func() time.Duration,
-	condition wait.ConditionWithContextFunc,
+	callback wait.ConditionWithContextFunc,
 ) error {
 	for {
 		interval := intervalFunc()
 
 		// Handle inactive state (interval == 0) and wait 1sec
-		for interval <= 0 {
-			ticker := time.NewTicker(time.Second)
+		if interval <= 0 {
 			select {
 			case <-ctx.Done():
-				ticker.Stop()
 				return ctx.Err()
-			case <-ticker.C:
-				interval = intervalFunc()
+			case <-time.After(time.Second):
+				continue
 			}
-			ticker.Stop()
 		}
 
-		jitter := time.Duration(rand.Int63n(int64(interval)) - int64(interval/2))
+		asInt := int64(interval)
+		jitter := time.Duration(rand.Int63n(asInt) - asInt/2)
 
-		// Active polling mode
 		select {
-		case <-ctx.Done():
-			return ctx.Err()
 		case <-time.After(interval + jitter):
-			if ok, err := func() (bool, error) {
+			if done, err := func() (bool, error) {
 				defer runtime.HandleCrashWithContext(ctx)
-				return condition(ctx)
-			}(); err != nil || ok {
+				return callback(ctx)
+			}(); err != nil || done {
 				return err
 			}
+		case <-ctx.Done():
+			return ctx.Err()
 		}
 	}
 }

pkg/controller/controller.go

@@ -62,6 +62,7 @@ func (c *Controller) SetupWithManager(manager *Manager) {
 	c.PollJitter = manager.PollJitter
 	c.DeQueueJitter = time.Second
 	c.lastPollTime = time.Now()
+	c.lastReconcileTime = time.Now()
 }
 
 // Start implements controller.Controller.
@@ -119,7 +120,7 @@ func (c *Controller) startPoller(ctx context.Context) {
 	defer c.Do.Shutdown()
 
 	klog.V(internallog.LogLevelTrace).InfoS("Starting controller poller", "ctrl", c.Name)
-	_ = helpers.DynamicPollUntilContextCancel(ctx, c.Do.GetPollInterval(), func(_ context.Context) (bool, error) {
+	err := helpers.DynamicPollUntilContextCancel(ctx, c.Do.GetPollInterval(), func(_ context.Context) (bool, error) {
 		defer func() {
 			c.lastPollTime = time.Now()
 		}()
@@ -129,11 +130,13 @@ func (c *Controller) startPoller(ctx context.Context) {
 		}
 
 		c.lastPollTime = time.Now()
-		time.Sleep(time.Duration(rand.Int63n(int64(c.PollJitter))))
-
 		// never stop
 		return false, nil
 	})
+	if err != nil {
+		klog.V(internallog.LogLevelDefault).ErrorS(err, "Controller poller failed", "ctrl", c.Name)
+	}
+
 	klog.V(internallog.LogLevelDefault).InfoS("Controller poller finished", "ctrl", c.Name)
 }
 
@@ -200,7 +203,7 @@ func (c *Controller) reconcile(ctx context.Context, req string) (_ reconcile.Res
 				for _, fn := range utilruntime.PanicHandlers {
 					fn(ctx, r)
 				}
-				err = fmt.Errorf("panic: %v [recovered]", r)
+				logf.FromContext(ctx).V(1).Error(err, fmt.Sprintf("Observed a panic in reconciler: %v", r))
 				return
 			}
 

pkg/controller/controller_manager.go

@@ -141,7 +141,7 @@ func (cm *Manager) startControllerSupervised(ctx context.Context, ctrl *Controll
 	// Make last controller action deadline 5 times the time of regular poll.
 	// It means that the controller hasn't polled/reconciled any resources.
 	// It could indicate that the controller might have died and should be restarted.
-	lastControllerActionDeadline := 5 * (cm.PollInterval + cm.PollJitter)
+
 	ticker := time.NewTicker(livenessCheckInterval)
 	defer ticker.Stop()
 
@@ -176,6 +176,15 @@ func (cm *Manager) startControllerSupervised(ctx context.Context, ctrl *Controll
 				cm.restartController(internalCtx, ctrl)
 			}()
 		case <-ticker.C:
+			lastControllerActionDeadline := 5 * ctrl.Do.GetPollInterval()()
+			if lastControllerActionDeadline <= 0 {
+				klog.V(log.LogLevelDebug).InfoS(
+					"Controller poll interval is 0, skipping last poll time check",
+					"name", ctrl.Name,
+				)
+				continue
+			}
+
 			lastPollTime := ctrl.LastPollTime()
 			klog.V(log.LogLevelDebug).InfoS(
 				"Controller last poll time check",
@@ -187,6 +196,7 @@ func (cm *Manager) startControllerSupervised(ctx context.Context, ctrl *Controll
 					"Controller unresponsive, restarting",
 					"ctrl", ctrl.Name,
 					"lastPollTime", lastPollTime.Format(time.RFC3339),
+					"lastControllerActionDeadline", lastControllerActionDeadline.String(),
 				)
 				cancel()
 				break
@@ -211,6 +221,7 @@ func (cm *Manager) startControllerSupervised(ctx context.Context, ctrl *Controll
 					"Controller unresponsive, restarting",
 					"ctrl", ctrl.Name,
 					"lastReconcileTime", lastPollTime.Format(time.RFC3339),
+					"lastControllerActionDeadline", lastControllerActionDeadline.String(),
 				)
 				cancel()
 			}

pkg/controller/restore/reconciler.go

@@ -89,7 +89,7 @@ func (s *RestoreReconciler) Shutdown() {
 }
 
 func (s *RestoreReconciler) GetPollInterval() func() time.Duration {
-	return func() time.Duration { return s.pollInterval } // use default poll interval
+	return func() time.Duration { return time.Duration(0) } // use default poll interval
 }
 
 func (s *RestoreReconciler) GetPublisher() (string, websocket.Publisher) {

pkg/controller/service/reconciler.go

@@ -144,7 +144,6 @@ func (s *ServiceReconciler) GetPollInterval() func() time.Duration {
 		if servicePollInterval := agentcommon.GetConfigurationManager().GetServicePollInterval(); servicePollInterval != nil {
 			return *servicePollInterval
 		}
-
 		return s.pollInterval
 	}
 }