fix: check if continuous backup was created via backupschedule (#7228)

constrain continuous backup created via backupschedule to limit at most one pitr workload

(cherry picked from commit 0eb3636)

controllers/dataprotection/backup_controller.go

@@ -387,6 +387,22 @@ func (r *BackupReconciler) prepareBackupRequest(
  return nil, err
  }
  request.ActionSet = actionSet
+
+ // check continuous backups should have backupschedule label
+ if request.ActionSet.Spec.BackupType == dpv1alpha1.BackupTypeContinuous {
+ if _, ok := request.Labels[dptypes.BackupScheduleLabelKey]; !ok {
+ return nil, fmt.Errorf("continuous backup is only allowed to be created by backupSchedule")
+ }
+ backupSchedule := &dpv1alpha1.BackupSchedule{}
+ if err := request.Client.Get(reqCtx.Ctx, client.ObjectKey{Name: backup.Labels[dptypes.BackupScheduleLabelKey],
+ Namespace: backup.Namespace}, backupSchedule); err != nil {
+ return nil, err
+ }
+ if backupSchedule.Status.Phase != dpv1alpha1.BackupSchedulePhaseAvailable {
+ return nil, fmt.Errorf("create continuous backup by failed backupschedule %s/%s",
+ backupSchedule.Namespace, backupSchedule.Name)
+ }
+ }
  }
 
  // check encryption config

controllers/dataprotection/backup_controller_test.go

@@ -700,6 +700,28 @@ var _ = Describe("Backup Controller test", func() {
  })).Should(Succeed())
  })
  })
+
+ Context("create continuous backup", func() {
+ It("should fail when continuous backup don't have backupschedule label", func() {
+ By("create actionset with continuous backuptype")
+ actionSet := testdp.NewFakeActionSet(&testCtx)
+ actionSetKey := client.ObjectKeyFromObject(actionSet)
+ Eventually(testapps.GetAndChangeObj(&testCtx, actionSetKey, func(fetched *dpv1alpha1.ActionSet) {
+ fetched.Spec.BackupType = dpv1alpha1.BackupTypeContinuous
+ }))
+ By("create continuous backup without backupschedule label")
+ backupPolicy = testdp.NewFakeBackupPolicy(&testCtx, nil)
+ backup := testdp.NewFakeBackup(&testCtx, func(bp *dpv1alpha1.Backup) {
+ bp.ObjectMeta.Name = "continuousbackup"
+ bp.Labels = map[string]string{}
+ })
+ backupKey := client.ObjectKeyFromObject(backup)
+ By("check backup phase")
+ Eventually(testapps.CheckObj(&testCtx, backupKey, func(g Gomega, fetched *dpv1alpha1.Backup) {
+ g.Expect(fetched.Status.Phase).Should(Equal(dpv1alpha1.BackupPhaseFailed))
+ })).Should(Succeed())
+ })
+ })
  })
 
  When("with backup repo", func() {