diff --git a/VERSION b/VERSION index 0c62199..ee1372d 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.2.1 +0.2.2 diff --git a/api/v1alpha1/memgraphcluster_types.go b/api/v1alpha1/memgraphcluster_types.go index a7c4f42..5de8011 100644 --- a/api/v1alpha1/memgraphcluster_types.go +++ b/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 diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go index 4d5a0e5..11741bc 100644 --- a/api/v1alpha1/zz_generated.deepcopy.go +++ b/api/v1alpha1/zz_generated.deepcopy.go @@ -297,6 +297,16 @@ func (in *ServiceNamesSpec) DeepCopy() *ServiceNamesSpec { // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *SnapshotSpec) DeepCopyInto(out *SnapshotSpec) { *out = *in + if in.ActiveDeadlineSeconds != nil { + in, out := &in.ActiveDeadlineSeconds, &out.ActiveDeadlineSeconds + *out = new(int64) + **out = **in + } + if in.StartingDeadlineSeconds != nil { + in, out := &in.StartingDeadlineSeconds, &out.StartingDeadlineSeconds + *out = new(int64) + **out = **in + } if in.S3 != nil { in, out := &in.S3, &out.S3 *out = new(S3BackupSpec) diff --git a/config/crd/bases/memgraph.base14.io_memgraphclusters.yaml b/config/crd/bases/memgraph.base14.io_memgraphclusters.yaml index 917ceca..8703c05 100644 --- a/config/crd/bases/memgraph.base14.io_memgraphclusters.yaml +++ b/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 diff --git a/config/samples/memgraph_v1alpha1_memgraphcluster.yaml b/config/samples/memgraph_v1alpha1_memgraphcluster.yaml index f741d31..85002e6 100644 --- a/config/samples/memgraph_v1alpha1_memgraphcluster.yaml +++ b/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 diff --git a/internal/controller/snapshot.go b/internal/controller/snapshot.go index 112f23f..1c6f50e 100644 --- a/internal/controller/snapshot.go +++ b/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,7 +117,8 @@ 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{ @@ -99,6 +126,7 @@ func buildSnapshotCronJob(cluster *memgraphv1alpha1.MemgraphCluster) *batchv1.Cr 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), diff --git a/internal/controller/snapshot_test.go b/internal/controller/snapshot_test.go index 04fd539..2bd0d01 100644 --- a/internal/controller/snapshot_test.go +++ b/internal/controller/snapshot_test.go @@ -6,6 +6,7 @@ import ( "strings" "testing" + batchv1 "k8s.io/api/batch/v1" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -74,6 +75,50 @@ func TestBuildSnapshotCronJob(t *testing.T) { } } +func TestBuildSnapshotCronJobPropagatesScheduling(t *testing.T) { + cluster := &memgraphv1alpha1.MemgraphCluster{ + ObjectMeta: metav1.ObjectMeta{Name: "test-cluster", Namespace: "default"}, + Spec: memgraphv1alpha1.MemgraphClusterSpec{ + NodeSelector: map[string]string{"workload": "database"}, + Tolerations: []corev1.Toleration{{ + Key: "kubernetes.io/arch", + Operator: corev1.TolerationOpEqual, + Value: "arm64", + Effect: corev1.TaintEffectNoSchedule, + }}, + Affinity: &corev1.Affinity{ + NodeAffinity: &corev1.NodeAffinity{ + RequiredDuringSchedulingIgnoredDuringExecution: &corev1.NodeSelector{ + NodeSelectorTerms: []corev1.NodeSelectorTerm{{ + MatchExpressions: []corev1.NodeSelectorRequirement{{ + Key: "kubernetes.io/arch", + Operator: corev1.NodeSelectorOpIn, + Values: []string{"arm64"}, + }}, + }}, + }, + }, + }, + Snapshot: memgraphv1alpha1.SnapshotSpec{Enabled: true}, + }, + } + + podSpec := buildSnapshotCronJob(cluster).Spec.JobTemplate.Spec.Template.Spec + + if podSpec.NodeSelector["workload"] != "database" { + t.Errorf("NodeSelector not propagated, got %v", podSpec.NodeSelector) + } + if len(podSpec.Tolerations) != 1 { + t.Fatalf("expected 1 toleration, got %d", len(podSpec.Tolerations)) + } + if podSpec.Tolerations[0].Key != "kubernetes.io/arch" { + t.Errorf("expected toleration key 'kubernetes.io/arch', got %s", podSpec.Tolerations[0].Key) + } + if podSpec.Affinity == nil || podSpec.Affinity.NodeAffinity == nil { + t.Error("Affinity not propagated") + } +} + func TestBuildSnapshotCronJobWithS3(t *testing.T) { secretRef := &corev1.LocalObjectReference{Name: "s3-credentials"} cluster := &memgraphv1alpha1.MemgraphCluster{ @@ -209,6 +254,130 @@ func TestBuildSnapshotCronJobDefaults(t *testing.T) { } } +func TestBuildSnapshotCronJobWedgePreventionDefaults(t *testing.T) { + cluster := &memgraphv1alpha1.MemgraphCluster{ + ObjectMeta: metav1.ObjectMeta{Name: "test-cluster", Namespace: "default"}, + Spec: memgraphv1alpha1.MemgraphClusterSpec{ + Snapshot: memgraphv1alpha1.SnapshotSpec{Enabled: true}, + }, + } + + cj := buildSnapshotCronJob(cluster) + + if cj.Spec.ConcurrencyPolicy != batchv1.ForbidConcurrent { + t.Errorf("expected Forbid, got %s", cj.Spec.ConcurrencyPolicy) + } + if cj.Spec.StartingDeadlineSeconds == nil { + t.Fatal("StartingDeadlineSeconds must be set or a wedged CronJob trips the >100 missed-starts lockout") + } + if *cj.Spec.StartingDeadlineSeconds != defaultStartingDeadlineSeconds { + t.Errorf("expected %d, got %d", defaultStartingDeadlineSeconds, *cj.Spec.StartingDeadlineSeconds) + } + deadline := cj.Spec.JobTemplate.Spec.ActiveDeadlineSeconds + if deadline == nil { + t.Fatal("ActiveDeadlineSeconds must be set or an unschedulable job blocks the schedule forever") + } + if *deadline != defaultActiveDeadlineSeconds { + t.Errorf("expected %d, got %d", defaultActiveDeadlineSeconds, *deadline) + } +} + +func TestBuildSnapshotCronJobWedgePreventionOverrides(t *testing.T) { + cluster := &memgraphv1alpha1.MemgraphCluster{ + ObjectMeta: metav1.ObjectMeta{Name: "test-cluster", Namespace: "default"}, + Spec: memgraphv1alpha1.MemgraphClusterSpec{ + Snapshot: memgraphv1alpha1.SnapshotSpec{ + Enabled: true, + ConcurrencyPolicy: memgraphv1alpha1.SnapshotConcurrencyReplace, + ActiveDeadlineSeconds: ptr(int64(1800)), + StartingDeadlineSeconds: ptr(int64(120)), + }, + }, + } + + cj := buildSnapshotCronJob(cluster) + + if cj.Spec.ConcurrencyPolicy != batchv1.ReplaceConcurrent { + t.Errorf("expected Replace, got %s", cj.Spec.ConcurrencyPolicy) + } + if *cj.Spec.StartingDeadlineSeconds != 120 { + t.Errorf("expected 120, got %d", *cj.Spec.StartingDeadlineSeconds) + } + if *cj.Spec.JobTemplate.Spec.ActiveDeadlineSeconds != 1800 { + t.Errorf("expected 1800, got %d", *cj.Spec.JobTemplate.Spec.ActiveDeadlineSeconds) + } +} + +func snapshotTestCluster(tolerated bool) *memgraphv1alpha1.MemgraphCluster { + c := &memgraphv1alpha1.MemgraphCluster{ + ObjectMeta: metav1.ObjectMeta{Name: "test-cluster", Namespace: "default"}, + Spec: memgraphv1alpha1.MemgraphClusterSpec{ + Snapshot: memgraphv1alpha1.SnapshotSpec{Enabled: true, Schedule: "0 * * * *"}, + }, + } + if tolerated { + c.Spec.Tolerations = []corev1.Toleration{{ + Key: "kubernetes.io/arch", + Operator: corev1.TolerationOpEqual, + Value: "arm64", + Effect: corev1.TaintEffectNoSchedule, + }} + } + return c +} + +func TestSnapshotCronJobNeedsUpdateStableRoundTrip(t *testing.T) { + cluster := snapshotTestCluster(true) + if snapshotCronJobNeedsUpdate(buildSnapshotCronJob(cluster), buildSnapshotCronJob(cluster)) { + t.Error("identical CronJobs reported as drifted - this would cause an infinite update loop") + } +} + +// The API server defaults many PodSpec fields the operator never sets. If those +// defaults were compared, every reconcile would report drift and the operator +// would rewrite the CronJob forever. +func TestSnapshotCronJobNeedsUpdateIgnoresServerDefaults(t *testing.T) { + cluster := snapshotTestCluster(true) + desired := buildSnapshotCronJob(cluster) + + existing := buildSnapshotCronJob(cluster) + podSpec := &existing.Spec.JobTemplate.Spec.Template.Spec + podSpec.DNSPolicy = corev1.DNSClusterFirst + podSpec.SchedulerName = "default-scheduler" + podSpec.TerminationGracePeriodSeconds = ptr(int64(30)) + podSpec.RestartPolicy = corev1.RestartPolicyOnFailure + for i := range podSpec.Containers { + podSpec.Containers[i].ImagePullPolicy = corev1.PullIfNotPresent + podSpec.Containers[i].TerminationMessagePath = "/dev/termination-log" + } + existing.Spec.JobTemplate.Spec.BackoffLimit = ptr(int32(6)) + existing.Spec.Suspend = ptr(false) + + if snapshotCronJobNeedsUpdate(existing, desired) { + t.Error("server-defaulted fields reported as drift - the operator would rewrite the CronJob on every reconcile") + } +} + +func TestSnapshotCronJobNeedsUpdateDetectsTolerationDrift(t *testing.T) { + // existing = what a pre-fix operator created; desired = what we build now. + existing := buildSnapshotCronJob(snapshotTestCluster(false)) + desired := buildSnapshotCronJob(snapshotTestCluster(true)) + + if !snapshotCronJobNeedsUpdate(existing, desired) { + t.Error("toleration drift not detected - already-deployed CronJobs would never be repaired") + } +} + +func TestSnapshotCronJobNeedsUpdateDetectsScheduleDrift(t *testing.T) { + existing := buildSnapshotCronJob(snapshotTestCluster(true)) + desired := buildSnapshotCronJob(snapshotTestCluster(true)) + desired.Spec.Schedule = "*/5 * * * *" + + if !snapshotCronJobNeedsUpdate(existing, desired) { + t.Error("schedule drift not detected") + } +} + func TestBuildSnapshotInitContainers(t *testing.T) { cluster := &memgraphv1alpha1.MemgraphCluster{ ObjectMeta: metav1.ObjectMeta{