🐛 fix(controller): snapshot CronJob drops scheduling constraints and wedges permanently

VERSION

@@ -1 +1 @@
-0.2.1
+0.2.2

api/v1alpha1/memgraphcluster_types.go

@@ -189,6 +189,16 @@ type HighAvailabilitySpec struct {
 	PreferredMain string `json:"preferredMain,omitempty"`
 }
 
+// SnapshotConcurrencyPolicy defines how concurrent snapshot job executions are handled
+// +kubebuilder:validation:Enum=Allow;Forbid;Replace
+type SnapshotConcurrencyPolicy string
+
+const (
+	SnapshotConcurrencyAllow   SnapshotConcurrencyPolicy = "Allow"
+	SnapshotConcurrencyForbid  SnapshotConcurrencyPolicy = "Forbid"
+	SnapshotConcurrencyReplace SnapshotConcurrencyPolicy = "Replace"
+)
+
 // SnapshotSpec defines snapshot and backup configuration
 type SnapshotSpec struct {
 	// Enabled enables periodic snapshots
@@ -201,6 +211,29 @@ type SnapshotSpec struct {
 	// +optional
 	Schedule string `json:"schedule,omitempty"`
 
+	// ConcurrencyPolicy controls what happens when a snapshot job is still running
+	// at the next scheduled tick. Forbid skips the tick, which is safe because it
+	// prevents overlapping CREATE SNAPSHOT runs.
+	// +kubebuilder:default=Forbid
+	// +optional
+	ConcurrencyPolicy SnapshotConcurrencyPolicy `json:"concurrencyPolicy,omitempty"`
+
+	// ActiveDeadlineSeconds is the maximum time a snapshot job may run before it is
+	// marked failed and its pods terminated. Without it, a job whose pod can never be
+	// scheduled stays Active forever and blocks every later run under Forbid.
+	// +kubebuilder:default=600
+	// +kubebuilder:validation:Minimum=60
+	// +optional
+	ActiveDeadlineSeconds *int64 `json:"activeDeadlineSeconds,omitempty"`
+
+	// StartingDeadlineSeconds bounds how far back the CronJob controller looks for
+	// missed schedules. Without it, more than 100 missed starts permanently disables
+	// scheduling with "too many missed start times".
+	// +kubebuilder:default=300
+	// +kubebuilder:validation:Minimum=10
+	// +optional
+	StartingDeadlineSeconds *int64 `json:"startingDeadlineSeconds,omitempty"`
+
 	// RetentionCount is the number of snapshots to retain on disk
 	// +kubebuilder:default=5
 	// +optional

config/crd/bases/memgraph.base14.io_memgraphclusters.yaml

@@ -1157,6 +1157,26 @@ spec:
               snapshot:
                 description: Snapshot defines snapshot and backup configuration
                 properties:
+                  activeDeadlineSeconds:
+                    default: 600
+                    description: |-
+                      ActiveDeadlineSeconds is the maximum time a snapshot job may run before it is
+                      marked failed and its pods terminated. Without it, a job whose pod can never be
+                      scheduled stays Active forever and blocks every later run under Forbid.
+                    format: int64
+                    minimum: 60
+                    type: integer
+                  concurrencyPolicy:
+                    default: Forbid
+                    description: |-
+                      ConcurrencyPolicy controls what happens when a snapshot job is still running
+                      at the next scheduled tick. Forbid skips the tick, which is safe because it
+                      prevents overlapping CREATE SNAPSHOT runs.
+                    enum:
+                    - Allow
+                    - Forbid
+                    - Replace
+                    type: string
                   enabled:
                     default: true
                     description: Enabled enables periodic snapshots
@@ -1215,6 +1235,15 @@ spec:
                     default: '*/15 * * * *'
                     description: Schedule is a cron expression for snapshot frequency
                     type: string
+                  startingDeadlineSeconds:
+                    default: 300
+                    description: |-
+                      StartingDeadlineSeconds bounds how far back the CronJob controller looks for
+                      missed schedules. Without it, more than 100 missed starts permanently disables
+                      scheduling with "too many missed start times".
+                    format: int64
+                    minimum: 10
+                    type: integer
                 type: object
               storage:
                 description: Storage defines the persistent storage configuration

config/samples/memgraph_v1alpha1_memgraphcluster.yaml

@@ -45,3 +45,9 @@ spec:
     enabled: true
     schedule: "*/15 * * * *"
     retentionCount: 5
+    # Forbid prevents overlapping CREATE SNAPSHOT runs. activeDeadlineSeconds
+    # ensures a job that cannot be scheduled fails rather than blocking the
+    # schedule forever.
+    concurrencyPolicy: Forbid
+    activeDeadlineSeconds: 600
+    startingDeadlineSeconds: 300

internal/controller/snapshot.go

@@ -10,6 +10,7 @@ import (
 	"go.uber.org/zap"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/equality"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -26,6 +27,14 @@ const (
 
 	// Shared volume name for snapshot data between containers
 	snapshotDataVolume = "snapshot-data"
+
+	// defaultActiveDeadlineSeconds bounds how long a snapshot job may run before it
+	// is failed, so a wedged job cannot block the schedule under Forbid.
+	defaultActiveDeadlineSeconds = int64(600)
+
+	// defaultStartingDeadlineSeconds bounds the CronJob controller's missed-start
+	// lookback window.
+	defaultStartingDeadlineSeconds = int64(300)
 )
 
 // SnapshotManager handles Memgraph snapshot operations
@@ -60,6 +69,23 @@ func buildSnapshotCronJob(cluster *memgraphv1alpha1.MemgraphCluster) *batchv1.Cr
 		schedule = "*/15 * * * *" // Default: every 15 minutes
 	}
 
+	// In-code defaults mirror the kubebuilder defaults so unit tests that build a
+	// bare struct — with no API server defaulting — get the same values.
+	concurrency := batchv1.ConcurrencyPolicy(cluster.Spec.Snapshot.ConcurrencyPolicy)
+	if concurrency == "" {
+		concurrency = batchv1.ForbidConcurrent
+	}
+
+	startingDeadline := cluster.Spec.Snapshot.StartingDeadlineSeconds
+	if startingDeadline == nil {
+		startingDeadline = ptr(defaultStartingDeadlineSeconds)
+	}
+
+	activeDeadline := cluster.Spec.Snapshot.ActiveDeadlineSeconds
+	if activeDeadline == nil {
+		activeDeadline = ptr(defaultActiveDeadlineSeconds)
+	}
+
 	// Use the same image as memgraph for the snapshot job
 	memgraphImage := cluster.Spec.Image
 	if memgraphImage == "" {
@@ -91,14 +117,16 @@ func buildSnapshotCronJob(cluster *memgraphv1alpha1.MemgraphCluster) *batchv1.Cr
 		},
 		Spec: batchv1.CronJobSpec{
 			Schedule:                   schedule,
-			ConcurrencyPolicy:          batchv1.ForbidConcurrent,
+			ConcurrencyPolicy:          concurrency,
+			StartingDeadlineSeconds:    startingDeadline,
 			SuccessfulJobsHistoryLimit: &successfulJobsHistoryLimit,
 			FailedJobsHistoryLimit:     &failedJobsHistoryLimit,
 			JobTemplate: batchv1.JobTemplateSpec{
 				ObjectMeta: metav1.ObjectMeta{
 					Labels: labelsForCluster(cluster),
 				},
 				Spec: batchv1.JobSpec{
+					ActiveDeadlineSeconds: activeDeadline,
 					Template: corev1.PodTemplateSpec{
 						ObjectMeta: metav1.ObjectMeta{
 							Labels: labelsForCluster(cluster),
@@ -113,6 +141,9 @@ func buildSnapshotCronJob(cluster *memgraphv1alpha1.MemgraphCluster) *batchv1.Cr
 									Type: corev1.SeccompProfileTypeRuntimeDefault,
 								},
 							},
+							NodeSelector:   cluster.Spec.NodeSelector,
+							Tolerations:    cluster.Spec.Tolerations,
+							Affinity:       cluster.Spec.Affinity,
 							InitContainers: initContainers,
 							Containers:     containers,
 							Volumes:        volumes,
@@ -346,6 +377,50 @@ func buildS3Env(cluster *memgraphv1alpha1.MemgraphCluster) []corev1.EnvVar {
 	return envVars
 }
 
+// snapshotCronJobNeedsUpdate reports whether an existing snapshot CronJob has
+// drifted from the desired spec.
+//
+// Only fields the operator sets are compared. Comparing whole specs would loop
+// forever: the API server defaults many PodSpec fields that the operator never
+// sets, so the two would never compare equal.
+func snapshotCronJobNeedsUpdate(existing, desired *batchv1.CronJob) bool {
+	if existing.Spec.Schedule != desired.Spec.Schedule ||
+		existing.Spec.ConcurrencyPolicy != desired.Spec.ConcurrencyPolicy ||
+		!equality.Semantic.DeepEqual(existing.Spec.StartingDeadlineSeconds, desired.Spec.StartingDeadlineSeconds) {
+		return true
+	}
+
+	existingJob := &existing.Spec.JobTemplate.Spec
+	desiredJob := &desired.Spec.JobTemplate.Spec
+	if !equality.Semantic.DeepEqual(existingJob.ActiveDeadlineSeconds, desiredJob.ActiveDeadlineSeconds) {
+		return true
+	}
+
+	e, d := &existingJob.Template.Spec, &desiredJob.Template.Spec
+	return !equality.Semantic.DeepEqual(e.NodeSelector, d.NodeSelector) ||
+		!equality.Semantic.DeepEqual(e.Tolerations, d.Tolerations) ||
+		!equality.Semantic.DeepEqual(e.Affinity, d.Affinity) ||
+		!snapshotContainersEqual(e.InitContainers, d.InitContainers) ||
+		!snapshotContainersEqual(e.Containers, d.Containers)
+}
+
+// snapshotContainersEqual compares containers on the fields the operator controls.
+// Full container comparison is avoided because the API server defaults
+// imagePullPolicy, terminationMessagePath and others.
+func snapshotContainersEqual(existing, desired []corev1.Container) bool {
+	if len(existing) != len(desired) {
+		return false
+	}
+	for i := range desired {
+		if existing[i].Name != desired[i].Name ||
+			existing[i].Image != desired[i].Image ||
+			!equality.Semantic.DeepEqual(existing[i].Args, desired[i].Args) {
+			return false
+		}
+	}
+	return true
+}
+
 // reconcileSnapshotCronJob ensures the snapshot CronJob exists and is configured correctly
 func (r *MemgraphClusterReconciler) reconcileSnapshotCronJob(ctx context.Context, cluster *memgraphv1alpha1.MemgraphCluster, log *zap.Logger) error {
 	// If snapshots are not enabled, ensure CronJob doesn't exist
@@ -375,11 +450,7 @@ func (r *MemgraphClusterReconciler) reconcileSnapshotCronJob(ctx context.Context
 		return err
 	}
 
-	// Update if schedule or S3 config changed
-	needsUpdate := existing.Spec.Schedule != desired.Spec.Schedule ||
-		len(existing.Spec.JobTemplate.Spec.Template.Spec.Containers) != len(desired.Spec.JobTemplate.Spec.Template.Spec.Containers)
-
-	if needsUpdate {
+	if snapshotCronJobNeedsUpdate(existing, desired) {
 		log.Info("updating snapshot CronJob",
 			zap.String("cronjob", existing.Name),
 			zap.String("oldSchedule", existing.Spec.Schedule),