Make the TTL of pgBackRest backups configurable

The default retention of one failed backup Job can leave a Job and its
Pods in a failed state indefinitely. The TTL setting lets someone choose
how long they want Jobs, Pods, and their logs to be available.

This field is functional in Kubernetes 1.21 and OpenShift 4.8 where the
TTLAfterFinished feature gate is enabled by default.

Issue: [sc-14014]
Issue: #3444

config/crd/bases/postgres-operator.crunchydata.com_postgresclusters.yaml

@@ -1325,6 +1325,12 @@ spec:
                                   type: string
                               type: object
                             type: array
+                          ttlSecondsAfterFinished:
+                            description: 'Limit the lifetime of a Job that has finished.
+                              More info: https://kubernetes.io/docs/concepts/workloads/controllers/job'
+                            format: int32
+                            minimum: 60
+                            type: integer
                         type: object
                       manual:
                         description: Defines details for manual pgBackRest backup

internal/controller/postgrescluster/pgbackrest.go

@@ -720,6 +720,10 @@ func generateBackupJobSpecIntent(postgresCluster *v1beta1.PostgresCluster,
 		},
 	}
 
+	if jobs := postgresCluster.Spec.Backups.PGBackRest.Jobs; jobs != nil {
+		jobSpec.TTLSecondsAfterFinished = jobs.TTLSecondsAfterFinished
+	}
+
 	// set the priority class name, tolerations, and affinity, if they exist
 	if postgresCluster.Spec.Backups.PGBackRest.Jobs != nil {
 		if postgresCluster.Spec.Backups.PGBackRest.Jobs.PriorityClassName != nil {

internal/controller/postgrescluster/pgbackrest_test.go

@@ -2560,9 +2560,8 @@ volumes:
 	})
 
 	t.Run("Resources", func(t *testing.T) {
-		cluster := &v1beta1.PostgresCluster{
-			Spec: v1beta1.PostgresClusterSpec{},
-		}
+		cluster := &v1beta1.PostgresCluster{}
+
 		t.Run("Resources not defined in jobs", func(t *testing.T) {
 			cluster.Spec.Backups = v1beta1.Backups{
 				PGBackRest: v1beta1.PGBackRestArchive{},
@@ -2635,16 +2634,9 @@ volumes:
 	})
 
 	t.Run("PriorityClassName", func(t *testing.T) {
-		cluster := &v1beta1.PostgresCluster{
-			Spec: v1beta1.PostgresClusterSpec{
-				Backups: v1beta1.Backups{
-					PGBackRest: v1beta1.PGBackRestArchive{
-						Jobs: &v1beta1.BackupJobs{
-							PriorityClassName: initialize.String("some-priority-class"),
-						},
-					},
-				},
-			},
+		cluster := &v1beta1.PostgresCluster{}
+		cluster.Spec.Backups.PGBackRest.Jobs = &v1beta1.BackupJobs{
+			PriorityClassName: initialize.String("some-priority-class"),
 		}
 		job, err := generateBackupJobSpecIntent(
 			cluster, v1beta1.PGBackRestRepo{},
@@ -2661,16 +2653,9 @@ volumes:
 			Operator: "Exist",
 		}}
 
-		cluster := &v1beta1.PostgresCluster{
-			Spec: v1beta1.PostgresClusterSpec{
-				Backups: v1beta1.Backups{
-					PGBackRest: v1beta1.PGBackRestArchive{
-						Jobs: &v1beta1.BackupJobs{
-							Tolerations: tolerations,
-						},
-					},
-				},
-			},
+		cluster := &v1beta1.PostgresCluster{}
+		cluster.Spec.Backups.PGBackRest.Jobs = &v1beta1.BackupJobs{
+			Tolerations: tolerations,
 		}
 		job, err := generateBackupJobSpecIntent(
 			cluster, v1beta1.PGBackRestRepo{},
@@ -2680,6 +2665,56 @@ volumes:
 		assert.NilError(t, err)
 		assert.DeepEqual(t, job.Template.Spec.Tolerations, tolerations)
 	})
+
+	t.Run("TTLSecondsAfterFinished", func(t *testing.T) {
+		cluster := &v1beta1.PostgresCluster{}
+
+		t.Run("Undefined", func(t *testing.T) {
+			cluster.Spec.Backups.PGBackRest.Jobs = nil
+
+			spec, err := generateBackupJobSpecIntent(
+				cluster, v1beta1.PGBackRestRepo{}, "", nil, nil,
+			)
+			assert.NilError(t, err)
+			assert.Assert(t, spec.TTLSecondsAfterFinished == nil)
+
+			cluster.Spec.Backups.PGBackRest.Jobs = &v1beta1.BackupJobs{}
+
+			spec, err = generateBackupJobSpecIntent(
+				cluster, v1beta1.PGBackRestRepo{}, "", nil, nil,
+			)
+			assert.NilError(t, err)
+			assert.Assert(t, spec.TTLSecondsAfterFinished == nil)
+		})
+
+		t.Run("Zero", func(t *testing.T) {
+			cluster.Spec.Backups.PGBackRest.Jobs = &v1beta1.BackupJobs{
+				TTLSecondsAfterFinished: initialize.Int32(0),
+			}
+
+			spec, err := generateBackupJobSpecIntent(
+				cluster, v1beta1.PGBackRestRepo{}, "", nil, nil,
+			)
+			assert.NilError(t, err)
+			if assert.Check(t, spec.TTLSecondsAfterFinished != nil) {
+				assert.Equal(t, *spec.TTLSecondsAfterFinished, int32(0))
+			}
+		})
+
+		t.Run("Positive", func(t *testing.T) {
+			cluster.Spec.Backups.PGBackRest.Jobs = &v1beta1.BackupJobs{
+				TTLSecondsAfterFinished: initialize.Int32(100),
+			}
+
+			spec, err := generateBackupJobSpecIntent(
+				cluster, v1beta1.PGBackRestRepo{}, "", nil, nil,
+			)
+			assert.NilError(t, err)
+			if assert.Check(t, spec.TTLSecondsAfterFinished != nil) {
+				assert.Equal(t, *spec.TTLSecondsAfterFinished, int32(100))
+			}
+		})
+	})
 }
 
 func TestGenerateRepoHostIntent(t *testing.T) {

pkg/apis/postgres-operator.crunchydata.com/v1beta1/pgbackrest_types.go

@@ -180,6 +180,12 @@ type BackupJobs struct {
 	// More info: https://kubernetes.io/docs/concepts/scheduling-eviction/taint-and-toleration
 	// +optional
 	Tolerations []corev1.Toleration `json:"tolerations,omitempty"`
+
+	// Limit the lifetime of a Job that has finished.
+	// More info: https://kubernetes.io/docs/concepts/workloads/controllers/job
+	// +optional
+	// +kubebuilder:validation:Minimum=60
+	TTLSecondsAfterFinished *int32 `json:"ttlSecondsAfterFinished,omitempty"`
 }
 
 // PGBackRestManualBackup contains information that is used for creating a