feat: supports incremental backup

apis/dataprotection/v1alpha1/backup_types.go

@@ -194,6 +194,18 @@ type BackupStatus struct {
 	// +optional
 	VolumeSnapshots []VolumeSnapshotStatus `json:"volumeSnapshots,omitempty"`
 
+	// Records the parent backup name for incremental or differential backup.
+	// When the parent backup is deleted, the backup will also be deleted.
+	//
+	// +optional
+	ParentBackupName string `json:"parentBackupName,omitempty"`
+
+	// Records the base full backup name for incremental backup or differential backup.
+	// When the base backup is deleted, the backup will also be deleted.
+	//
+	// +optional
+	BaseBackupName string `json:"baseBackupName,omitempty"`
+
 	// Records any additional information for the backup.
 	//
 	// +optional

apis/dataprotection/v1alpha1/backuppolicy_types.go

@@ -221,6 +221,12 @@ type BackupMethod struct {
 	// +kubebuilder:validation:Pattern:=`^[a-z0-9]([a-z0-9\.\-]*[a-z0-9])?$`
 	Name string `json:"name"`
 
+	// The name of the compatible full backup method, used by incremental backups.
+	//
+	// +optional
+	// +kubebuilder:validation:Pattern:=`^[a-z0-9]([a-z0-9\.\-]*[a-z0-9])?$`
+	CompatibleMethod string `json:"compatibleMethod,omitempty"`
+
 	// Specifies whether to take snapshots of persistent volumes. If true,
 	// the ActionSetName is not required, the controller will use the CSI volume
 	// snapshotter to create the snapshot.

apis/dataprotection/v1alpha1/backuppolicytemplate_types.go

@@ -82,6 +82,12 @@ type BackupMethodTPL struct {
 	// +kubebuilder:validation:Pattern:=`^[a-z0-9]([a-z0-9\.\-]*[a-z0-9])?$`
 	Name string `json:"name"`
 
+	// The name of the compatible full backup method, used by incremental backups.
+	//
+	// +optional
+	// +kubebuilder:validation:Pattern:=`^[a-z0-9]([a-z0-9\.\-]*[a-z0-9])?$`
+	CompatibleMethod string `json:"compatibleMethod,omitempty"`
+
 	// Specifies whether to take snapshots of persistent volumes. If true,
 	// the ActionSetName is not required, the controller will use the CSI volume
 	// snapshotter to create the snapshot.

config/crd/bases/dataprotection.kubeblocks.io_backuppolicies.yaml

@@ -73,6 +73,11 @@ spec:
                         For volume snapshot backup, the actionSet is not required, the controller
                         will use the CSI volume snapshotter to create the snapshot.
                       type: string
+                    compatibleMethod:
+                      description: The name of the compatible full backup method,
+                        used by incremental backups.
+                      pattern: ^[a-z0-9]([a-z0-9\.\-]*[a-z0-9])?$
+                      type: string
                     env:
                       description: Specifies the environment variables for the backup
                         workload.

config/crd/bases/dataprotection.kubeblocks.io_backuppolicytemplates.yaml

@@ -76,6 +76,11 @@ spec:
                         For volume snapshot backup, the actionSet is not required, the controller
                         will use the CSI volume snapshotter to create the snapshot.
                       type: string
+                    compatibleMethod:
+                      description: The name of the compatible full backup method,
+                        used by incremental backups.
+                      pattern: ^[a-z0-9]([a-z0-9\.\-]*[a-z0-9])?$
+                      type: string
                     env:
                       description: Specifies the environment variables for the backup
                         workload.

config/crd/bases/dataprotection.kubeblocks.io_backups.yaml

@@ -298,6 +298,11 @@ spec:
                       For volume snapshot backup, the actionSet is not required, the controller
                       will use the CSI volume snapshotter to create the snapshot.
                     type: string
+                  compatibleMethod:
+                    description: The name of the compatible full backup method, used
+                      by incremental backups.
+                    pattern: ^[a-z0-9]([a-z0-9\.\-]*[a-z0-9])?$
+                    type: string
                   env:
                     description: Specifies the environment variables for the backup
                       workload.
@@ -1013,6 +1018,11 @@ spec:
               backupRepoName:
                 description: The name of the backup repository.
                 type: string
+              baseBackupName:
+                description: |-
+                  Records the base full backup name for incremental backup or differential backup.
+                  When the base backup is deleted, the backup will also be deleted.
+                type: string
               completionTimestamp:
                 description: |-
                   Records the time when the backup operation was completed.
@@ -1092,6 +1102,11 @@ spec:
               kopiaRepoPath:
                 description: Records the path of the Kopia repository.
                 type: string
+              parentBackupName:
+                description: |-
+                  Records the parent backup name for incremental or differential backup.
+                  When the parent backup is deleted, the backup will also be deleted.
+                type: string
               path:
                 description: |-
                   The directory within the backup repository where the backup data is stored.

controllers/apps/transformer_cluster_backup_policy.go

@@ -414,11 +414,12 @@ func (r *backupPolicyBuilder) syncBackupMethods(backupPolicy *dpv1alpha1.BackupP
 	}
 	for _, backupMethodTPL := range r.backupPolicyTPL.Spec.BackupMethods {
 		backupMethod := dpv1alpha1.BackupMethod{
-			Name:            backupMethodTPL.Name,
-			ActionSetName:   backupMethodTPL.ActionSetName,
-			SnapshotVolumes: backupMethodTPL.SnapshotVolumes,
-			TargetVolumes:   backupMethodTPL.TargetVolumes,
-			RuntimeSettings: backupMethodTPL.RuntimeSettings,
+			Name:             backupMethodTPL.Name,
+			CompatibleMethod: backupMethodTPL.CompatibleMethod,
+			ActionSetName:    backupMethodTPL.ActionSetName,
+			SnapshotVolumes:  backupMethodTPL.SnapshotVolumes,
+			TargetVolumes:    backupMethodTPL.TargetVolumes,
+			RuntimeSettings:  backupMethodTPL.RuntimeSettings,
 		}
 		if m, ok := oldBackupMethodMap[backupMethodTPL.Name]; ok {
 			backupMethod = m

controllers/dataprotection/actionset_controller_test.go

@@ -55,14 +55,14 @@ var _ = Describe("ActionSet Controller test", func() {
 
 	Context("create a actionSet", func() {
 		It("should be available", func() {
-			as := testdp.NewFakeActionSet(&testCtx)
+			as := testdp.NewFakeActionSet(&testCtx, nil)
 			Expect(as).ShouldNot(BeNil())
 		})
 	})
 
 	Context("validate a actionSet", func() {
 		It("validate withParameters", func() {
-			as := testdp.NewFakeActionSet(&testCtx)
+			as := testdp.NewFakeActionSet(&testCtx, nil)
 			Expect(as).ShouldNot(BeNil())
 			By("set invalid withParameters and schema")
 			Expect(testapps.ChangeObj(&testCtx, as, func(action *dpv1alpha1.ActionSet) {

controllers/dataprotection/backup_controller.go

@@ -245,6 +245,11 @@ func (r *BackupReconciler) deleteBackupFiles(reqCtx intctrlutil.RequestCtx, back
 // handleDeletingPhase handles the deletion of backup. It will delete the backup CR
 // and the backup workload(job).
 func (r *BackupReconciler) handleDeletingPhase(reqCtx intctrlutil.RequestCtx, backup *dpv1alpha1.Backup) (ctrl.Result, error) {
+	// delete related backups
+	if err := r.deleteRelatedBackups(reqCtx, backup); err != nil {
+		return intctrlutil.RequeueWithError(err, reqCtx.Log, "")
+	}
+
 	// if backup phase is Deleting, delete the backup reference workloads,
 	// backup data stored in backup repository and volume snapshots.
 	// TODO(ldm): if backup is being used by restore, do not delete it.
@@ -395,28 +400,11 @@ func (r *BackupReconciler) prepareBackupRequest(
 		if err != nil {
 			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)
-			}
-		}
-
 		// validate parameters
 		if err := dputils.ValidateParameters(actionSet, backup.Spec.Parameters, true); err != nil {
 			return nil, fmt.Errorf("fails to validate parameters with actionset %s: %v", actionSet.Name, err)
 		}
+		request.ActionSet = actionSet
 	}
 
 	// check encryption config
@@ -432,13 +420,25 @@ func (r *BackupReconciler) prepareBackupRequest(
 	}
 
 	request.BackupPolicy = backupPolicy
+	request.BackupMethod = backupMethod
+
+	switch dpv1alpha1.BackupType(request.GetBackupType()) {
+	case dpv1alpha1.BackupTypeIncremental:
+		request, err = prepare4Incremental(request)
+	case dpv1alpha1.BackupTypeContinuous:
+		err = validateContinuousBackup(backup, reqCtx, request.Client)
+	}
+	if err != nil {
+		return nil, err
+	}
+
 	if !snapshotVolumes {
 		// if use volume snapshot, ignore backup repo
 		if err = HandleBackupRepo(request); err != nil {
 			return nil, err
 		}
 	}
-	request.BackupMethod = backupMethod
+
 	return request, nil
 }
 
@@ -527,6 +527,14 @@ func (r *BackupReconciler) patchBackupStatus(
 	request.Status.Phase = dpv1alpha1.BackupPhaseRunning
 	request.Status.StartTimestamp = &metav1.Time{Time: r.clock.Now().UTC()}
 
+	// set status parent backup and base backup name
+	if request.ParentBackup != nil {
+		request.Status.ParentBackupName = request.ParentBackup.Name
+	}
+	if request.BaseBackup != nil {
+		request.Status.BaseBackupName = request.BaseBackup.Name
+	}
+
 	if err = dpbackup.SetExpirationByCreationTime(request.Backup); err != nil {
 		return err
 	}
@@ -751,6 +759,33 @@ func (r *BackupReconciler) deleteExternalResources(
 	return deleteRelatedObjectList(reqCtx, r.Client, &appsv1.StatefulSetList{}, namespaces, labels)
 }
 
+// deleteRelatedBackups deletes the related backups.
+func (r *BackupReconciler) deleteRelatedBackups(
+	reqCtx intctrlutil.RequestCtx,
+	backup *dpv1alpha1.Backup) error {
+	backupList := &dpv1alpha1.BackupList{}
+	labels := map[string]string{
+		dptypes.BackupPolicyLabelKey: backup.Spec.BackupPolicyName,
+	}
+	if err := r.Client.List(reqCtx.Ctx, backupList,
+		client.InNamespace(backup.Namespace), client.MatchingLabels(labels)); client.IgnoreNotFound(err) != nil {
+		return err
+	}
+	for i := range backupList.Items {
+		bp := &backupList.Items[i]
+		// delete backups related to the current backup
+		// files in the related backup's status.path will be deleted by its own associated deleter
+		if bp.Status.ParentBackupName != backup.Name && bp.Status.BaseBackupName != backup.Name {
+			continue
+		}
+		if err := intctrlutil.BackgroundDeleteObject(r.Client, reqCtx.Ctx, bp); err != nil {
+			return err
+		}
+		reqCtx.Log.Info("delete the related backup", "backup", fmt.Sprintf("%s/%s", bp.Namespace, bp.Name))
+	}
+	return nil
+}
+
 // PatchBackupObjectMeta patches backup object metaObject include cluster snapshot.
 func PatchBackupObjectMeta(
 	original *dpv1alpha1.Backup,
@@ -922,3 +957,56 @@ func setClusterSnapshotAnnotation(request *dpbackup.Request, cluster *kbappsv1.C
 	request.Backup.Annotations[constant.ClusterSnapshotAnnotationKey] = *clusterString
 	return nil
 }
+
+// validateContinuousBackup validates the continuous backup.
+func validateContinuousBackup(backup *dpv1alpha1.Backup, reqCtx intctrlutil.RequestCtx, cli client.Client) error {
+	// validate if the continuous backup is created by a backupSchedule.
+	if _, ok := backup.Labels[dptypes.BackupScheduleLabelKey]; !ok {
+		return fmt.Errorf("continuous backup is only allowed to be created by backupSchedule")
+	}
+	backupSchedule := &dpv1alpha1.BackupSchedule{}
+	if err := cli.Get(reqCtx.Ctx, client.ObjectKey{Name: backup.Labels[dptypes.BackupScheduleLabelKey],
+		Namespace: backup.Namespace}, backupSchedule); err != nil {
+		return err
+	}
+	if backupSchedule.Status.Phase != dpv1alpha1.BackupSchedulePhaseAvailable {
+		return fmt.Errorf("create continuous backup by failed backupschedule %s/%s",
+			backupSchedule.Namespace, backupSchedule.Name)
+	}
+	return nil
+}
+
+// prepare4Incremental prepares for incremental backup
+func prepare4Incremental(request *dpbackup.Request) (*dpbackup.Request, error) {
+	// get and validate parent backup
+	parentBackup, err := GetParentBackup(request.Ctx, request.Client, request.Backup, request.BackupMethod)
+	if err != nil {
+		return nil, err
+	}
+	parentBackupType, err := dputils.GetBackupTypeByMethodName(request.RequestCtx,
+		request.Client, parentBackup.Spec.BackupMethod, request.BackupPolicy)
+	if err != nil {
+		return nil, err
+	}
+	request.ParentBackup = parentBackup
+	// get and validate base backup
+	switch parentBackupType {
+	case dpv1alpha1.BackupTypeFull:
+		request.BaseBackup = request.ParentBackup
+	case dpv1alpha1.BackupTypeIncremental:
+		baseBackup := &dpv1alpha1.Backup{}
+		baseBackupName := request.ParentBackup.Status.BaseBackupName
+		if len(baseBackupName) == 0 {
+			return nil, fmt.Errorf("backup %s/%s base backup name is empty",
+				request.ParentBackup.Namespace, request.ParentBackup.Name)
+		}
+		if err := request.Client.Get(request.Ctx, client.ObjectKey{Name: baseBackupName,
+			Namespace: request.ParentBackup.Namespace}, baseBackup); err != nil {
+			return nil, fmt.Errorf("failed to get base backup %s/%s: %w", request.ParentBackup.Namespace, baseBackupName, err)
+		}
+		request.BaseBackup = baseBackup
+	default:
+		return nil, fmt.Errorf("parent backup type is %s, but only full and incremental backup are supported", parentBackupType)
+	}
+	return request, nil
+}