feat: Use staggered schedule for CronJobs

This patch includes the following changes around the determined
schedule for created (ReclaimSpace/KeyRotation)CronJobs:

- Prevents thundering-herd behavior by staggering the schedule by
a max offset of 2hrs.
  - If a job's schedule requires it to run more frequently, the stagger
    window is adjusted accordingly. IOW, if schedule < 2hrs; staggerWindow = schedule
- The offset is calculated deterministically using UID to ensure the same offset
b/w multiple reconciles.
- If a job has missed its last execution, the stagger is not applied
to let the job run immediately (might lead to thundering-herd but a fair
tradeoff in happy path?)

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

Author: black-dragon74

api/csiaddons/v1alpha1/encryptionkeyrotationcronjob_types.go

@@ -36,6 +36,9 @@ type EncryptionKeyRotationJobTemplateSpec struct {
 // EncryptionKeyRotationCronJobSpec defines the desired state of EncryptionKeyRotationCronJob
 type EncryptionKeyRotationCronJobSpec struct {
 	// The schedule in Cron format, see https://en.wikipedia.org/wiki/Cron.
+	// A deterministic, UID-based stagger offset is applied to spread
+	// execution across the "cronjob-stagger-window" (default: 2 hours,
+	// set to 0 to disable) configured in the csi-addons-config ConfigMap.
 	// +kubebuilder:validation:Required
 	// +kubebuilder:validation:Pattern:=.+
 	Schedule string `json:"schedule"`

api/csiaddons/v1alpha1/reclaimspacecronjob_types.go

@@ -36,6 +36,9 @@ type ReclaimSpaceJobTemplateSpec struct {
 // ReclaimSpaceCronJobSpec defines the desired state of ReclaimSpaceJob
 type ReclaimSpaceCronJobSpec struct {
 	// The schedule in Cron format, see https://en.wikipedia.org/wiki/Cron.
+	// A deterministic, UID-based stagger offset is applied to spread
+	// execution across the "cronjob-stagger-window" (default: 2 hours,
+	// set to 0 to disable) configured in the csi-addons-config ConfigMap.
 	// +kubebuilder:validation:Required
 	// +kubebuilder:validation:Pattern:=.+
 	Schedule string `json:"schedule"`

config/crd/bases/csiaddons.openshift.io_encryptionkeyrotationcronjobs.yaml

@@ -146,7 +146,11 @@ spec:
                 - spec
                 type: object
               schedule:
-                description: The schedule in Cron format, see https://en.wikipedia.org/wiki/Cron.
+                description: |-
+                  The schedule in Cron format, see https://en.wikipedia.org/wiki/Cron.
+                  A deterministic, UID-based stagger offset is applied to spread
+                  execution across the "cronjob-stagger-window" (default: 2 hours,
+                  set to 0 to disable) configured in the csi-addons-config ConfigMap.
                 pattern: .+
                 type: string
               startingDeadlineSeconds:

config/crd/bases/csiaddons.openshift.io_reclaimspacecronjobs.yaml

@@ -145,7 +145,11 @@ spec:
                 - spec
                 type: object
               schedule:
-                description: The schedule in Cron format, see https://en.wikipedia.org/wiki/Cron.
+                description: |-
+                  The schedule in Cron format, see https://en.wikipedia.org/wiki/Cron.
+                  A deterministic, UID-based stagger offset is applied to spread
+                  execution across the "cronjob-stagger-window" (default: 2 hours,
+                  set to 0 to disable) configured in the csi-addons-config ConfigMap.
                 pattern: .+
                 type: string
               startingDeadlineSeconds:

deploy/controller/crds.yaml

@@ -304,7 +304,11 @@ spec:
                 - spec
                 type: object
               schedule:
-                description: The schedule in Cron format, see https://en.wikipedia.org/wiki/Cron.
+                description: |-
+                  The schedule in Cron format, see https://en.wikipedia.org/wiki/Cron.
+                  A deterministic, UID-based stagger offset is applied to spread
+                  execution across the "cronjob-stagger-window" (default: 2 hours,
+                  set to 0 to disable) configured in the csi-addons-config ConfigMap.
                 pattern: .+
                 type: string
               startingDeadlineSeconds:
@@ -1004,7 +1008,11 @@ spec:
                 - spec
                 type: object
               schedule:
-                description: The schedule in Cron format, see https://en.wikipedia.org/wiki/Cron.
+                description: |-
+                  The schedule in Cron format, see https://en.wikipedia.org/wiki/Cron.
+                  A deterministic, UID-based stagger offset is applied to spread
+                  execution across the "cronjob-stagger-window" (default: 2 hours,
+                  set to 0 to disable) configured in the csi-addons-config ConfigMap.
                 pattern: .+
                 type: string
               startingDeadlineSeconds:

internal/controller/csiaddons/encryptionkeyrotationcronjob_controller.go

@@ -313,8 +313,11 @@ func getNextScheduleForKeyRotation(
 			earliestTime = schedulingDeadline
 		}
 	}
+
+	rawNext := sched.Next(now)
+	staggeredNext := utils.GetStaggeredNext(krcJob.UID, rawNext, sched)
 	if earliestTime.After(now) {
-		return time.Time{}, sched.Next(now), nil
+		return time.Time{}, staggeredNext, nil
 	}
 
 	starts := 0
@@ -333,7 +336,7 @@ func getNextScheduleForKeyRotation(
 					" delete and recreate encryptionkeyrotationjob")
 		}
 	}
-	return lastMissed, sched.Next(now), nil
+	return lastMissed, staggeredNext, nil
 }
 
 // constructEncryptionKeyRotationJob forms an EncryptionKeyRotationJob for a given

internal/controller/csiaddons/reclaimspacecronjob_controller.go

@@ -376,8 +376,11 @@ func getNextSchedule(
 			earliestTime = schedulingDeadline
 		}
 	}
+
+	rawNext := sched.Next(now)
+	staggeredNext := utils.GetStaggeredNext(rsCronJob.UID, rawNext, sched)
 	if earliestTime.After(now) {
-		return time.Time{}, sched.Next(now), nil
+		return time.Time{}, staggeredNext, nil
 	}
 
 	starts := 0
@@ -396,5 +399,5 @@ func getNextSchedule(
 					" delete and recreate reclaimspacecronjob")
 		}
 	}
-	return lastMissed, sched.Next(now), nil
+	return lastMissed, staggeredNext, nil
 }

internal/controller/csiaddons/reclaimspacecronjob_controller_test.go

@@ -22,9 +22,12 @@ import (
 	"time"
 
 	csiaddonsv1alpha1 "github.com/csi-addons/kubernetes-csi-addons/api/csiaddons/v1alpha1"
+	"github.com/csi-addons/kubernetes-csi-addons/internal/controller/utils"
+	"github.com/robfig/cron/v3"
 
 	"github.com/stretchr/testify/assert"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/utils/ptr"
 )
 
@@ -113,6 +116,13 @@ func TestGetScheduledTimeForRSJob(t *testing.T) {
 func TestGetNextSchedule(t *testing.T) {
 	now := time.Now()
 	expectedLastMissed := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.Local)
+	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
+	}
 	type args struct {
 		rsCronJob *csiaddonsv1alpha1.ReclaimSpaceCronJob
 		now       time.Time
@@ -128,6 +138,9 @@ func TestGetNextSchedule(t *testing.T) {
 			name: "Valid schedule, no deadline, last schedule time exists",
 			args: args{
 				rsCronJob: &csiaddonsv1alpha1.ReclaimSpaceCronJob{
+					ObjectMeta: metav1.ObjectMeta{
+						UID: types.UID("valid-sched-1"),
+					},
 					Spec: csiaddonsv1alpha1.ReclaimSpaceCronJobSpec{
 						Schedule: "0 0 * * *",
 					},
@@ -145,6 +158,9 @@ func TestGetNextSchedule(t *testing.T) {
 			name: "Valid schedule, deadline not exceeded, last schedule time exists",
 			args: args{
 				rsCronJob: &csiaddonsv1alpha1.ReclaimSpaceCronJob{
+					ObjectMeta: metav1.ObjectMeta{
+						UID: types.UID("valid-sched-2"),
+					},
 					Spec: csiaddonsv1alpha1.ReclaimSpaceCronJobSpec{
 						Schedule:                "0 0 * * *",
 						StartingDeadlineSeconds: ptr.To(int64(3600)),
@@ -163,6 +179,9 @@ func TestGetNextSchedule(t *testing.T) {
 			name: "Valid schedule, deadline exceeded, last schedule time exists, missed schedules < 100",
 			args: args{
 				rsCronJob: &csiaddonsv1alpha1.ReclaimSpaceCronJob{
+					ObjectMeta: metav1.ObjectMeta{
+						UID: types.UID("valid-sched-3"),
+					},
 					Spec: csiaddonsv1alpha1.ReclaimSpaceCronJobSpec{
 						Schedule:                "*/1 * * * *",
 						StartingDeadlineSeconds: ptr.To(int64(6000)),
@@ -181,6 +200,9 @@ func TestGetNextSchedule(t *testing.T) {
 			name: "Valid schedule, deadline exceeded, last schedule time exists, missed schedules > 100",
 			args: args{
 				rsCronJob: &csiaddonsv1alpha1.ReclaimSpaceCronJob{
+					ObjectMeta: metav1.ObjectMeta{
+						UID: types.UID("valid-sched-4"),
+					},
 					Spec: csiaddonsv1alpha1.ReclaimSpaceCronJobSpec{
 						Schedule:                "*/1 * * * *",
 						StartingDeadlineSeconds: ptr.To(int64(6060)),
@@ -203,11 +225,19 @@ func TestGetNextSchedule(t *testing.T) {
 				t.Errorf("getNextSchedule() error = %v, wantErr %v", err, tt.wantErr)
 				return
 			}
+
+			if tt.wantErr {
+				return
+			}
+
 			if !gotLastMissed.Equal(tt.lastMissed) && !gotLastMissed.Equal(time.Time{}) {
 				t.Errorf("getNextSchedule() got last missed = %v, want %v", gotLastMissed, tt.lastMissed)
 			}
-			if !gotNextSchedule.Equal(tt.nextSchedule) {
-				t.Errorf("getNextSchedule() got next schedule = %v, want %v", gotNextSchedule, tt.nextSchedule)
+
+			sched := mustParse(tt.args.rsCronJob.Spec.Schedule)
+			staggered := utils.GetStaggeredNext(tt.args.rsCronJob.UID, tt.nextSchedule, sched)
+			if !gotNextSchedule.Equal(staggered) {
+				t.Errorf("getNextSchedule() got next schedule = %v, want %v", gotNextSchedule, staggered)
 			}
 		})
 	}

internal/controller/utils/reclaimspace_keyrotation.go

@@ -18,10 +18,14 @@ 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"
 )
 
@@ -128,3 +132,33 @@ 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,8 +18,11 @@ 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"
 )
@@ -87,3 +90,115 @@ 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))
+				}
+			})
+		}
+	}
+}