Rework Reconcile scheduled sync

cmd/jetstream-controller/main.go

@@ -36,7 +36,6 @@ import (
 	klog "k8s.io/klog/v2"
 
 	ctrl "sigs.k8s.io/controller-runtime"
-	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/healthz"
 )
 
@@ -111,10 +110,10 @@ func run() error {
 		}
 
 		controllerCfg := &controller.Config{
-			ReadOnly:     *readOnly,
-			Namespace:    *namespace,
-			CacheDir:     *cacheDir,
-			SyncInterval: *controlLoopSyncInterval,
+			ReadOnly:        *readOnly,
+			Namespace:       *namespace,
+			CacheDir:        *cacheDir,
+			RequeueInterval: *controlLoopSyncInterval,
 		}
 
 		return runControlLoop(config, natsCfg, controllerCfg)
@@ -171,9 +170,6 @@ func runControlLoop(config *rest.Config, natsCfg *controller.NatsConfig, control
 	mgr, err := ctrl.NewManager(config, ctrl.Options{
 		Scheme: scheme,
 		Logger: klog.NewKlogr().WithName("controller-runtime"),
-		Cache: cache.Options{
-			SyncPeriod: &controllerCfg.SyncInterval,
-		},
 	})
 	if err != nil {
 		return fmt.Errorf("unable to start manager: %w", err)

internal/controller/consumer_controller.go

@@ -132,7 +132,8 @@ func (r *ConsumerReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 		}
 		return ctrl.Result{}, fmt.Errorf("create or update: %s", err)
 	}
-	return ctrl.Result{}, nil
+
+	return ctrl.Result{RequeueAfter: r.RequeueInterval()}, nil
 }
 
 func (r *ConsumerReconciler) deleteConsumer(ctx context.Context, log logr.Logger, consumer *api.Consumer) error {

internal/controller/jetstream_controller.go

@@ -3,6 +3,7 @@ package controller
 import (
 	"context"
 	"fmt"
+	"math/rand/v2"
 	"os"
 	"path/filepath"
 	"strings"
@@ -34,6 +35,8 @@ type JetStreamController interface {
 	//
 	// Returns the error of the operation or errors during client setup.
 	WithJetStreamClient(opts api.ConnectionOpts, ns string, op func(js jetstream.JetStream) error) error
+
+	RequeueInterval() time.Duration
 }
 
 func NewJSController(k8sClient client.Client, natsConfig *NatsConfig, controllerConfig *Config) (JetStreamController, error) {
@@ -52,6 +55,18 @@ type jsController struct {
 	cacheDir         string
 }
 
+func (c *jsController) RequeueInterval() time.Duration {
+	// Stagger the requeue slightly
+	interval := c.controllerConfig.RequeueInterval
+
+	// Allow up to a 10% variance
+	intervalRange := float64(interval.Nanoseconds()) * 0.1
+
+	randomFactor := (rand.Float64() * 2) - 1.0
+
+	return time.Duration(float64(interval.Nanoseconds()) + (intervalRange * randomFactor))
+}
+
 func (c *jsController) ReadOnly() bool {
 	return c.controllerConfig.ReadOnly
 }

internal/controller/keyvalue_controller.go

@@ -120,7 +120,8 @@ func (r *KeyValueReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 	if err := r.createOrUpdate(ctx, log, keyValue); err != nil {
 		return ctrl.Result{}, fmt.Errorf("create or update: %s", err)
 	}
-	return ctrl.Result{}, nil
+
+	return ctrl.Result{RequeueAfter: r.RequeueInterval()}, nil
 }
 
 func (r *KeyValueReconciler) deleteKeyValue(ctx context.Context, log logr.Logger, keyValue *api.KeyValue) error {

internal/controller/objectstore_controller.go

@@ -121,7 +121,8 @@ func (r *ObjectStoreReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 	if err := r.createOrUpdate(ctx, log, objectStore); err != nil {
 		return ctrl.Result{}, fmt.Errorf("create or update: %s", err)
 	}
-	return ctrl.Result{}, nil
+
+	return ctrl.Result{RequeueAfter: r.RequeueInterval()}, nil
 }
 
 func (r *ObjectStoreReconciler) deleteObjectStore(ctx context.Context, log logr.Logger, objectStore *api.ObjectStore) error {

internal/controller/register.go

@@ -13,10 +13,10 @@ import (
 //
 // Namespace restricts the controller to resources of the given namespace.
 type Config struct {
-	ReadOnly     bool
-	Namespace    string
-	SyncInterval time.Duration
-	CacheDir     string
+	ReadOnly        bool
+	Namespace       string
+	RequeueInterval time.Duration
+	CacheDir        string
 }
 
 // RegisterAll registers all available jetStream controllers to the manager.

internal/controller/stream_controller.go

@@ -117,7 +117,8 @@ func (r *StreamReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr
 	if err := r.createOrUpdate(ctx, log, stream); err != nil {
 		return ctrl.Result{}, fmt.Errorf("create or update: %s", err)
 	}
-	return ctrl.Result{}, nil
+
+	return ctrl.Result{RequeueAfter: r.RequeueInterval()}, nil
 }
 
 func (r *StreamReconciler) deleteStream(ctx context.Context, log logr.Logger, stream *api.Stream) error {