config: make max stagger window configurable, default to 2hrs

This patch makes the stagger window for reclaimspace & keyrotation
CronJobs configurable.

The default value is 2hrs and it can be disabled by setting the value
to: 0

Signed-off-by: Niraj Yadav <niryadav@redhat.com>

Author: black-dragon74

cmd/manager/main.go

@@ -107,6 +107,7 @@ func main() {
 	flag.BoolVar(&enableAuth, "enable-auth", true, "Enables TLS and adds bearer token to the headers (enabled by default)")
 	flag.IntVar(&cfg.MaxGroupPVC, "max-group-pvc", cfg.MaxGroupPVC, "Maximum number of PVCs allowed in a volume group")
 	flag.IntVar(&cfg.CSIAddonsNodeRetryDelay, util.CsiaddonsNodeRetryDelayKey, cfg.CSIAddonsNodeRetryDelay, "Duration, in seconds, the CSIAddonsNode reconciler must wait before retrying the connection to csi-addons sidecar")
+	flag.IntVar(&cfg.KrRsCronJobStaggerWindow, util.KrRsCronJobStaggerWindowKey, cfg.KrRsCronJobStaggerWindow, "Duration, in hours, that the CronJobs for KeyRotation and ReclaimSpace are staggered within. Defaults to 2 hours, set as 0 to disable.")
 	opts := zap.Options{
 		Development: true,
 		TimeEncoder: zapcore.ISO8601TimeEncoder,
@@ -234,8 +235,9 @@ func main() {
 		os.Exit(1)
 	}
 	if err = (&controllers.ReclaimSpaceCronJobReconciler{
-		Client: mgr.GetClient(),
-		Scheme: mgr.GetScheme(),
+		Client:        mgr.GetClient(),
+		Scheme:        mgr.GetScheme(),
+		StaggerWindow: cfg.KrRsCronJobStaggerWindow,
 	}).SetupWithManager(mgr, ctrlOptions); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "ReclaimSpaceCronJob")
 		os.Exit(1)
@@ -299,8 +301,9 @@ func main() {
 		os.Exit(1)
 	}
 	if err = (&controllers.EncryptionKeyRotationCronJobReconciler{
-		Client: mgr.GetClient(),
-		Scheme: mgr.GetScheme(),
+		Client:        mgr.GetClient(),
+		Scheme:        mgr.GetScheme(),
+		StaggerWindow: cfg.KrRsCronJobStaggerWindow,
 	}).SetupWithManager(mgr, ctrlOptions); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "EncryptionKeyRotationCronJob")
 		os.Exit(1)

deploy/controller/csi-addons-config.yaml

@@ -10,3 +10,4 @@ data:
   "max-concurrent-reconciles": "100"
   "max-group-pvcs": "100"
   "csi-addons-node-retry-delay": "5"
+  "kr-rs-cronjob-stagger-window": "2"

docs/csi-addons-config.md

@@ -4,13 +4,14 @@ CSI-Addons Operator can consume configuration from a ConfigMap named `csi-addons
 in the same namespace as the operator. This enables configuration of the operator to persist across
 upgrades. The ConfigMap can support the following configuration options:
 
-| Option                        | Default value | Description                                                                                         |
-| ----------------------------- | ------------- | --------------------------------------------------------------------------------------------------- |
-| `reclaim-space-timeout`       | `"3m"`        | Timeout for reclaimspace operation                                                                  |
-| `max-concurrent-reconciles`   | `"100"`       | Maximum number of concurrent reconciles                                                             |
-| `max-group-pvcs`              | `"100"`       | Maximum number of PVCs allowed in a volume group                                                    |
-| `csi-addons-node-retry-delay` | `"5"`         | Duration, in seconds, that csi-addons reconcile must wait before retrying connection to the sidecar |
-| `schedule-precedence`         | `"pvc"`       | The order in which the schedule annotation should be read                                           |
+| Option                         | Default value | Description                                                                                                              |
+| ------------------------------ | ------------- | ------------------------------------------------------------------------------------------------------------------------ |
+| `reclaim-space-timeout`        | `"3m"`        | Timeout for reclaimspace operation                                                                                       |
+| `max-concurrent-reconciles`    | `"100"`       | Maximum number of concurrent reconciles                                                                                  |
+| `max-group-pvcs`               | `"100"`       | Maximum number of PVCs allowed in a volume group                                                                         |
+| `csi-addons-node-retry-delay`  | `"5"`         | Duration, in seconds, that csi-addons reconcile must wait before retrying connection to the sidecar                      |
+| `schedule-precedence`          | `"pvc"`       | The order in which the schedule annotation should be read                                                                |
+| `kr-rs-cronjob-stagger-window` | `"2"`         | Maximum stagger window, in hours, for key rotation and reclaim space CronJob schedules. Set to `0` to disable staggering |
 
 [`csi-addons-config` ConfigMap](../deploy/controller/csi-addons-config.yaml) is provided as an example.
 

internal/controller/csiaddons/encryptionkeyrotationcronjob_controller.go

@@ -40,7 +40,8 @@ import (
 // EncryptionKeyRotationCronJobReconciler reconciles a EncryptionKeyRotationCronJob object
 type EncryptionKeyRotationCronJobReconciler struct {
 	client.Client
-	Scheme *runtime.Scheme
+	Scheme        *runtime.Scheme
+	StaggerWindow int
 }
 
 //+kubebuilder:rbac:groups=csiaddons.openshift.io,resources=encryptionkeyrotationcronjobs,verbs=get;list;watch;create;update;patch;delete
@@ -123,7 +124,7 @@ func (r *EncryptionKeyRotationCronJobReconciler) Reconcile(ctx context.Context,
 		return ctrl.Result{}, nil
 	}
 
-	missedRun, nextRun, err := getNextScheduleForKeyRotation(krcJob, time.Now())
+	missedRun, nextRun, err := getNextScheduleForKeyRotation(krcJob, time.Now(), r.StaggerWindow)
 	if err != nil {
 		logger.Error(err, "failed to get next schedule for jobs", "schedule", krcJob.Spec.Schedule)
 
@@ -293,7 +294,8 @@ func (r *EncryptionKeyRotationCronJobReconciler) deleteOldEncryptionKeyRotationJ
 // An error is returned if start is missed more than 100 times
 func getNextScheduleForKeyRotation(
 	krcJob *csiaddonsv1alpha1.EncryptionKeyRotationCronJob,
-	now time.Time) (time.Time, time.Time, error) {
+	now time.Time,
+	staggerWindow int) (time.Time, time.Time, error) {
 	sched, err := cron.ParseStandard(krcJob.Spec.Schedule)
 	if err != nil {
 		return time.Time{}, time.Time{}, fmt.Errorf("unparsable schedule %q: %v", krcJob.Spec.Schedule, err)
@@ -315,7 +317,7 @@ func getNextScheduleForKeyRotation(
 	}
 
 	rawNext := sched.Next(now)
-	staggeredNext := utils.GetStaggeredNext(krcJob.UID, rawNext, sched)
+	staggeredNext := utils.GetStaggeredNext(krcJob.UID, rawNext, sched, staggerWindow)
 	if earliestTime.After(now) {
 		return time.Time{}, staggeredNext, nil
 	}

internal/controller/csiaddons/reclaimspacecronjob_controller.go

@@ -40,7 +40,8 @@ import (
 // ReclaimSpaceCronJobReconciler reconciles a ReclaimSpaceCronJob object
 type ReclaimSpaceCronJobReconciler struct {
 	client.Client
-	Scheme *runtime.Scheme
+	Scheme        *runtime.Scheme
+	StaggerWindow int
 }
 
 var (
@@ -129,7 +130,7 @@ func (r *ReclaimSpaceCronJobReconciler) Reconcile(ctx context.Context, req ctrl.
 	}
 
 	// figure out the next times that we need to create jobs at (or anything we missed).
-	missedRun, nextRun, err := getNextSchedule(rsCronJob, time.Now())
+	missedRun, nextRun, err := getNextSchedule(rsCronJob, time.Now(), r.StaggerWindow)
 	if err != nil {
 		logger.Error(err, "Failed to Parse out CronJob schedule", "schedule", rsCronJob.Spec.Schedule)
 		// invalid schedule, do not requeue.
@@ -356,7 +357,8 @@ func getScheduledTimeForRSJob(rsJob *csiaddonsv1alpha1.ReclaimSpaceJob) (*time.T
 // This function returns error if there are more than 100 missed start times.
 func getNextSchedule(
 	rsCronJob *csiaddonsv1alpha1.ReclaimSpaceCronJob,
-	now time.Time) (time.Time, time.Time, error) {
+	now time.Time,
+	staggerWindow int) (time.Time, time.Time, error) {
 	sched, err := cron.ParseStandard(rsCronJob.Spec.Schedule)
 	if err != nil {
 		return time.Time{}, time.Time{}, fmt.Errorf("unparsable schedule %q: %v", rsCronJob.Spec.Schedule, err)
@@ -378,7 +380,7 @@ func getNextSchedule(
 	}
 
 	rawNext := sched.Next(now)
-	staggeredNext := utils.GetStaggeredNext(rsCronJob.UID, rawNext, sched)
+	staggeredNext := utils.GetStaggeredNext(rsCronJob.UID, rawNext, sched, staggerWindow)
 	if earliestTime.After(now) {
 		return time.Time{}, staggeredNext, nil
 	}

internal/controller/csiaddons/reclaimspacecronjob_controller_test.go

@@ -220,7 +220,7 @@ func TestGetNextSchedule(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			gotLastMissed, gotNextSchedule, err := getNextSchedule(tt.args.rsCronJob, tt.args.now)
+			gotLastMissed, gotNextSchedule, err := getNextSchedule(tt.args.rsCronJob, tt.args.now, 2)
 			if (err != nil) != tt.wantErr {
 				t.Errorf("getNextSchedule() error = %v, wantErr %v", err, tt.wantErr)
 				return
@@ -235,7 +235,7 @@ func TestGetNextSchedule(t *testing.T) {
 			}
 
 			sched := mustParse(tt.args.rsCronJob.Spec.Schedule)
-			staggered := utils.GetStaggeredNext(tt.args.rsCronJob.UID, tt.nextSchedule, sched)
+			staggered := utils.GetStaggeredNext(tt.args.rsCronJob.UID, tt.nextSchedule, sched, 2)
 			if !gotNextSchedule.Equal(staggered) {
 				t.Errorf("getNextSchedule() got next schedule = %v, want %v", gotNextSchedule, staggered)
 			}

internal/controller/utils/reclaimspace_keyrotation.go

@@ -18,14 +18,10 @@ package utils
 
 import (
 	"errors"
-	"hash/fnv"
-	"time"
 
 	csiaddonsv1alpha1 "github.com/csi-addons/kubernetes-csi-addons/api/csiaddons/v1alpha1"
-	"github.com/robfig/cron/v3"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	apiTypes "k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
@@ -103,17 +99,6 @@ func SetSpec(obj client.Object, schedule, pvcName string) {
 	}
 }
 
-func GetSchedule(obj client.Object) string {
-	switch v := obj.(type) {
-	case *csiaddonsv1alpha1.EncryptionKeyRotationCronJob:
-		return v.Spec.Schedule
-	case *csiaddonsv1alpha1.ReclaimSpaceCronJob:
-		return v.Spec.Schedule
-	default:
-		return ""
-	}
-}
-
 // ExtractOwnerNameFromPVCObj extracts owner.Name from the object if it is
 // of type `T` and has a PVC as its owner.
 func ExtractOwnerNameFromPVCObj[T client.Object](rawObj client.Object) []string {
@@ -132,33 +117,3 @@ func ExtractOwnerNameFromPVCObj[T client.Object](rawObj client.Object) []string
 
 	return []string{owner.Name}
 }
-
-// GetStaggeredNext returns a deterministic, UID-based staggered time computed from a schedule.
-// The staggered window is capped at a maximum of 2hrs but is adjusted for smaller intervals.
-func GetStaggeredNext(uid apiTypes.UID, nextTime time.Time, sched cron.Schedule) time.Time {
-	// cron does not expose interval directly
-	// We can determine it by subtracting two consecutive intervals
-	afterNext := sched.Next(nextTime)
-	interval := afterNext.Sub(nextTime)
-
-	// We do not want the stagger window to be any larger than a max of 2hrs
-	const maxCap = 2 * time.Hour
-	staggerWindow := min(interval, maxCap)
-
-	// To prevent the schedule from jumping in bw the reconciles,
-	// we use the UID and hash it, this way it remains deterministic
-	h := fnv.New32a()
-	if _, err := h.Write([]byte(string(uid))); err != nil {
-		return nextTime
-	}
-	hash := h.Sum32()
-
-	// Just a safety net
-	stgrWindowSeconds := int64(staggerWindow.Seconds())
-	if stgrWindowSeconds <= 0 {
-		return nextTime
-	}
-	offsetSeconds := int64(hash) % stgrWindowSeconds // Modulo ensures it to be < stgrWindowSeconds
-
-	return nextTime.Add(time.Duration(offsetSeconds) * time.Second)
-}

internal/controller/utils/reclaimspace_keyrotation_test.go

@@ -18,11 +18,8 @@ package utils
 
 import (
 	"testing"
-	"time"
 
 	csiaddonsv1alpha1 "github.com/csi-addons/kubernetes-csi-addons/api/csiaddons/v1alpha1"
-	"github.com/robfig/cron/v3"
-	apitypes "k8s.io/apimachinery/pkg/types"
 
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
@@ -90,115 +87,3 @@ func TestSetSpec(t *testing.T) {
 		})
 	}
 }
-
-func TestGetStaggeredNext(t *testing.T) {
-	baseTime, _ := time.Parse(time.RFC3339, "2026-01-01T12:00:00Z")
-
-	mustParse := func(sched string) cron.Schedule {
-		s, err := cron.ParseStandard(sched)
-		if err != nil {
-			t.Fatalf("Failed to parse cron spec %q: %v", sched, err)
-		}
-		return s
-	}
-
-	tests := []struct {
-		name     string
-		uid      apitypes.UID
-		schedule string
-		baseNext time.Time
-		maxDelta time.Duration // The staggered offset msut be less than this value
-	}{
-		{
-			name:     "Hourly schedule - capped at 1h interval",
-			uid:      apitypes.UID("job-hourly"),
-			schedule: "@hourly",
-			baseNext: baseTime,
-			maxDelta: 1 * time.Hour,
-		},
-		{
-			name:     "Daily schedule - capped at 2h max cap",
-			uid:      apitypes.UID("job-daily"),
-			schedule: "@daily",
-			baseNext: baseTime,
-			maxDelta: 2 * time.Hour,
-		},
-		{
-			name:     "1 Minute schedule - capped at 1m interval",
-			uid:      apitypes.UID("job-minute"),
-			schedule: "* * * * *",
-			baseNext: baseTime,
-			maxDelta: 1 * time.Minute,
-		},
-		{
-			name:     "13 Minute schedule - capped at 13m interval",
-			uid:      apitypes.UID("job-13m"),
-			schedule: "*/13 * * * *",
-			baseNext: baseTime,
-			maxDelta: 13 * time.Minute,
-		},
-		{
-			// This last test case is special, avoid modifications
-			name:     "Determinism Check A (Run 1)", // Do not modify, used to check when to store the result
-			uid:      apitypes.UID("static-uid"),    // Do not modify, used to assert determinisim below
-			schedule: "@daily",
-			baseNext: baseTime,
-			maxDelta: 2 * time.Hour,
-		},
-	}
-
-	// Will store the result of the above special test case
-	var deterministicResult time.Time
-	var collisionResult time.Time
-
-	for i, tc := range tests {
-		t.Run(tc.name, func(t *testing.T) {
-			sched := mustParse(tc.schedule)
-			got := GetStaggeredNext(tc.uid, tc.baseNext, sched)
-
-			// Staggered time must be >= original time
-			if got.Before(tc.baseNext) {
-				t.Errorf("Result %v is before base time %v", got, tc.baseNext)
-			}
-
-			// Offset must be < min(cronInterval, maxCapof2hrs)
-			offset := got.Sub(tc.baseNext)
-			if offset >= tc.maxDelta {
-				t.Errorf("Offset %v exceeds allowed max delta %v", offset, tc.maxDelta)
-			}
-
-			// Store results of the last test case for determinism
-			if tc.name == "Determinism Check A (Run 1)" {
-				deterministicResult = got
-				collisionResult = got
-			}
-		})
-
-		// to ensure the run after the above TCs
-		if i == len(tests)-1 {
-			t.Run("Determinism Check B (Run 2 - Same UID)", func(t *testing.T) {
-				sched := mustParse("@daily")
-				got := GetStaggeredNext("static-uid", baseTime, sched)
-
-				if !got.Equal(deterministicResult) {
-					t.Errorf("Expected %v, got %v. Result not deterministic.", deterministicResult, got)
-				}
-			})
-
-			t.Run("Collision Check (Different UID)", func(t *testing.T) {
-				sched := mustParse("@daily")
-				got := GetStaggeredNext("different-uid", baseTime, sched)
-
-				// It is theoretically possible for hashes to collide, but highly unlikely
-				// especially within a 2hr window
-				if got.Equal(collisionResult) {
-					t.Logf("Warning: Different UIDs resulted in exact same time %v. "+
-						"This is rare but possible (hash collision).", got)
-				} else {
-					t.Logf("Verified different UIDs produced different offsets: %v vs %v",
-						collisionResult.Sub(baseTime), got.Sub(baseTime))
-				}
-			})
-		}
-	}
-}