chore: support to specify the continuous backup method for cluster.spec

apis/apps/v1alpha1/cluster_types.go

@@ -298,6 +298,11 @@ type ClusterBackup struct {
 	// +optional
 	PITREnabled *bool `json:"pitrEnabled,omitempty"`
 
+	// Specifies the backup method to use, if not set, use the first continuous method.
+	//
+	// +optional
+	ContinuousMethod string `json:"continuousMethod,omitempty"`
+
 	// Specifies whether to enable incremental backup.
 	//
 	// +kubebuilder:default=false

config/crd/bases/apps.kubeblocks.io_clusters.yaml

@@ -200,6 +200,10 @@ spec:
               backup:
                 description: Specifies the backup configuration of the Cluster.
                 properties:
+                  continuousMethod:
+                    description: Specifies the backup method to use, if not set, use
+                      the first continuous method.
+                    type: string
                   cronExpression:
                     description: The cron expression for the schedule. The timezone
                       is in UTC. See https://en.wikipedia.org/wiki/Cron.

controllers/apps/cluster_controller_test.go

@@ -40,6 +40,7 @@ import (
 	"github.com/apecloud/kubeblocks/pkg/controller/builder"
 	"github.com/apecloud/kubeblocks/pkg/controller/component"
 	"github.com/apecloud/kubeblocks/pkg/controller/scheduling"
+	"github.com/apecloud/kubeblocks/pkg/dataprotection/utils/boolptr"
 	"github.com/apecloud/kubeblocks/pkg/generics"
 	testapps "github.com/apecloud/kubeblocks/pkg/testutil/apps"
 	testdp "github.com/apecloud/kubeblocks/pkg/testutil/dataprotection"
@@ -161,7 +162,7 @@ var _ = Describe("Cluster Controller", func() {
 			GetObject()
 
 		By("Create a bpt obj")
-		createBackupPolicyTpl(clusterDefObj, compDefObj.Name, clusterVersionName)
+		createBackupPolicyTpl(clusterDefObj, compDefObj.Name, false, clusterVersionName)
 
 		By("Create a componentVersion obj")
 		compVersionObj = testapps.NewComponentVersionFactory(compVersionName).
@@ -1368,6 +1369,19 @@ var _ = Describe("Cluster Controller", func() {
 						RepoName:                backupRepoName,
 					},
 				},
+				{
+					desc: "backup with snapshot method and specified continuous method",
+					backup: &appsv1alpha1.ClusterBackup{
+						Enabled:                 &boolTrue,
+						RetentionPeriod:         retention("1d"),
+						Method:                  vsBackupMethodName,
+						CronExpression:          "*/1 * * * *",
+						StartingDeadlineMinutes: int64Ptr(int64(10)),
+						ContinuousMethod:        continuousMethodName1,
+						PITREnabled:             &boolTrue,
+						RepoName:                backupRepoName,
+					},
+				},
 				{
 					desc: "disable backup",
 					backup: &appsv1alpha1.ClusterBackup{
@@ -1405,19 +1419,36 @@ var _ = Describe("Cluster Controller", func() {
 
 				checkSchedule := func(g Gomega, schedule *dpv1alpha1.BackupSchedule) {
 					var policy *dpv1alpha1.SchedulePolicy
-					enableOtherFullMethod := false
-					for i, s := range schedule.Spec.Schedules {
+					hasCheckPITRMethod := false
+					for i := range schedule.Spec.Schedules {
+						s := &schedule.Spec.Schedules[i]
 						if s.BackupMethod == backup.Method {
 							Expect(*s.Enabled).Should(BeEquivalentTo(*backup.Enabled))
-							policy = &schedule.Spec.Schedules[i]
-							if *backup.Enabled {
-								enableOtherFullMethod = true
+							policy = s
+							continue
+						}
+						if !slices.Contains([]string{continuousMethodName, continuousMethodName1}, s.BackupMethod) {
+							if boolptr.IsSetToTrue(backup.Enabled) {
+								// another full backup method should be disabled.
+								Expect(*s.Enabled).Should(BeFalse())
 							}
 							continue
 						}
-						if enableOtherFullMethod {
-							// another full backup method should be disabled.
-							Expect(*s.Enabled).Should(BeFalse())
+						if len(backup.ContinuousMethod) == 0 {
+							// first continuous backup method should be equal to "PITREnabled", another is disabled.
+							if !hasCheckPITRMethod {
+								Expect(*s.Enabled).Should(BeEquivalentTo(*backup.PITREnabled))
+								hasCheckPITRMethod = true
+							} else {
+								Expect(*s.Enabled).Should(BeFalse())
+							}
+						} else {
+							// specified continuous backup method should be equal to "PITREnabled", another is disabled.
+							if backup.ContinuousMethod == s.BackupMethod {
+								Expect(*s.Enabled).Should(BeEquivalentTo(*backup.PITREnabled))
+							} else {
+								Expect(*s.Enabled).Should(BeFalse())
+							}
 						}
 					}
 					if backup.Enabled != nil && *backup.Enabled {
@@ -1593,9 +1624,10 @@ var _ = Describe("Cluster Controller", func() {
 	})
 })
 
-func createBackupPolicyTpl(clusterDefObj *appsv1alpha1.ClusterDefinition, compDef string, mappingClusterVersions ...string) {
+func createBackupPolicyTpl(clusterDefObj *appsv1alpha1.ClusterDefinition, compDef string, forHScale bool, mappingClusterVersions ...string) {
 	By("create actionSet")
-	fakeActionSet(clusterDefObj.Name)
+	fakeActionSet(actionSetName, clusterDefObj.Name, dpv1alpha1.BackupTypeFull)
+	fakeActionSet(continuousActionSetName, clusterDefObj.Name, dpv1alpha1.BackupTypeContinuous)
 
 	By("Creating a BackupPolicyTemplate")
 	bpt := testapps.NewBackupPolicyTemplateFactory(backupPolicyTPLName).
@@ -1618,6 +1650,16 @@ func createBackupPolicyTpl(clusterDefObj *appsv1alpha1.ClusterDefinition, compDe
 		case appsv1alpha1.Replication:
 			bpt.SetTargetRole("primary")
 		}
+		if !forHScale {
+			bpt.AddBackupMethod(continuousMethodName, false, continuousActionSetName).
+				SetComponentDef(compDef).
+				SetBackupMethodVolumeMounts("data", "/data").
+				AddBackupMethod(continuousMethodName1, false, continuousActionSetName).
+				SetComponentDef(compDef).
+				SetBackupMethodVolumeMounts("data", "/data").
+				AddSchedule(continuousMethodName, "0 0 * * *", ttl, false).
+				AddSchedule(continuousMethodName1, "0 0 * * *", ttl, false)
+		}
 	}
 	bpt.Create(&testCtx)
 }

controllers/apps/component_controller_test.go

@@ -66,10 +66,13 @@ import (
 )
 
 const (
-	backupPolicyTPLName = "test-backup-policy-template-mysql"
-	backupMethodName    = "test-backup-method"
-	vsBackupMethodName  = "test-vs-backup-method"
-	actionSetName       = "test-action-set"
+	backupPolicyTPLName     = "test-backup-policy-template-mysql"
+	backupMethodName        = "test-backup-method"
+	continuousMethodName    = "continuous-backup-method"
+	continuousMethodName1   = "continuous-backup-method1"
+	vsBackupMethodName      = "test-vs-backup-method"
+	actionSetName           = "test-action-set"
+	continuousActionSetName = "test-continuous-action-set"
 )
 
 var (
@@ -700,7 +703,7 @@ var _ = Describe("Component Controller", func() {
 
 					if policyType == appsv1alpha1.HScaleDataClonePolicyCloneVolume {
 						By("creating actionSet if backup policy is backup")
-						fakeActionSet(clusterDef.Name)
+						fakeActionSet(actionSetName, clusterDef.Name, dpv1alpha1.BackupTypeFull)
 					}
 				}
 			})()).ShouldNot(HaveOccurred())
@@ -2052,7 +2055,7 @@ var _ = Describe("Component Controller", func() {
 	Context("provisioning", func() {
 		BeforeEach(func() {
 			createAllWorkloadTypesClusterDef()
-			createBackupPolicyTpl(clusterDefObj, compDefName)
+			createBackupPolicyTpl(clusterDefObj, compDefName, true)
 		})
 
 		AfterEach(func() {
@@ -2123,7 +2126,7 @@ var _ = Describe("Component Controller", func() {
 
 		BeforeEach(func() {
 			createAllWorkloadTypesClusterDef()
-			createBackupPolicyTpl(clusterDefObj, compDefName)
+			createBackupPolicyTpl(clusterDefObj, compDefName, true)
 		})
 
 		AfterEach(func() {
@@ -2143,7 +2146,7 @@ var _ = Describe("Component Controller", func() {
 		BeforeEach(func() {
 			cleanEnv()
 			createAllWorkloadTypesClusterDef()
-			createBackupPolicyTpl(clusterDefObj, compDefName)
+			createBackupPolicyTpl(clusterDefObj, compDefName, true)
 		})
 
 		createNWaitClusterObj := func(components map[string]string,
@@ -2220,7 +2223,7 @@ var _ = Describe("Component Controller", func() {
 
 		BeforeEach(func() {
 			createAllWorkloadTypesClusterDef()
-			createBackupPolicyTpl(clusterDefObj, compDefName)
+			createBackupPolicyTpl(clusterDefObj, compDefName, true)
 			mockStorageClass = testk8s.CreateMockStorageClass(&testCtx, testk8s.DefaultStorageClassName)
 		})
 
@@ -2289,7 +2292,7 @@ var _ = Describe("Component Controller", func() {
 	When("creating cluster with workloadType=consensus component", func() {
 		BeforeEach(func() {
 			createAllWorkloadTypesClusterDef()
-			createBackupPolicyTpl(clusterDefObj, compDefName)
+			createBackupPolicyTpl(clusterDefObj, compDefName, true)
 		})
 
 		AfterEach(func() {
@@ -2544,7 +2547,7 @@ func checkRestoreAndSetCompleted(clusterKey types.NamespacedName, compName strin
 	mockRestoreCompleted(ml)
 }
 
-func fakeActionSet(clusterDefName string) *dpv1alpha1.ActionSet {
+func fakeActionSet(actionSetName, clusterDefName string, backupType dpv1alpha1.BackupType) *dpv1alpha1.ActionSet {
 	actionSet := &dpv1alpha1.ActionSet{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: actionSetName,
@@ -2559,7 +2562,7 @@ func fakeActionSet(clusterDefName string) *dpv1alpha1.ActionSet {
 					Value: "test-value",
 				},
 			},
-			BackupType: dpv1alpha1.BackupTypeFull,
+			BackupType: backupType,
 			Backup: &dpv1alpha1.BackupActionSpec{
 				BackupData: &dpv1alpha1.BackupDataActionSpec{
 					JobActionSpec: dpv1alpha1.JobActionSpec{

controllers/apps/opsrequest_controller_test.go

@@ -319,7 +319,7 @@ var _ = Describe("OpsRequest Controller", func() {
 		}
 
 		createMysqlCluster := func(replicas int32) {
-			createBackupPolicyTpl(clusterDefObj, mysqlCompDefName)
+			createBackupPolicyTpl(clusterDefObj, mysqlCompDefName, true)
 
 			By("set component to horizontal with snapshot policy and create a cluster")
 			testk8s.MockEnableVolumeSnapshot(&testCtx, testk8s.DefaultStorageClassName)

controllers/apps/transformer_cluster_backup_policy.go

@@ -647,7 +647,8 @@ func (r *clusterBackupPolicyTransformer) mergeClusterBackup(
 	hasSyncPITRMethod := false
 	hasSyncIncMethod := false
 	enableAutoBackup := boolptr.IsSetToTrue(backup.Enabled)
-	for i, s := range backupSchedule.Spec.Schedules {
+	for i := range backupSchedule.Spec.Schedules {
+		s := &backupSchedule.Spec.Schedules[i]
 		if s.BackupMethod == backup.Method {
 			mergeSchedulePolicy(sp, &backupSchedule.Spec.Schedules[i])
 			exist = true
@@ -671,15 +672,23 @@ func (r *clusterBackupPolicyTransformer) mergeClusterBackup(
 			r.Error(err, "failed to get ActionSet for backup.", "ActionSet", as.Name)
 			continue
 		}
-		if as.Spec.BackupType == dpv1alpha1.BackupTypeContinuous && backup.PITREnabled != nil && !hasSyncPITRMethod {
-			// auto-sync the first continuous backup for the 'pirtEnable' option.
-			backupSchedule.Spec.Schedules[i].Enabled = backup.PITREnabled
+		switch as.Spec.BackupType {
+		case dpv1alpha1.BackupTypeContinuous:
+			if backup.PITREnabled == nil {
+				continue
+			}
+			if boolptr.IsSetToFalse(backup.PITREnabled) || hasSyncPITRMethod ||
+				(len(backup.ContinuousMethod) > 0 && backup.ContinuousMethod != s.BackupMethod) {
+				s.Enabled = boolptr.False()
+				continue
+			}
+			// auto-sync the first or specified continuous backup for the 'pirtEnable' option.
+			s.Enabled = backup.PITREnabled
 			if backup.RetentionPeriod.String() != "" {
-				backupSchedule.Spec.Schedules[i].RetentionPeriod = backup.RetentionPeriod
+				s.RetentionPeriod = backup.RetentionPeriod
 			}
 			hasSyncPITRMethod = true
-		}
-		if as.Spec.BackupType == dpv1alpha1.BackupTypeIncremental {
+		case dpv1alpha1.BackupTypeIncremental:
 			if len(backup.Method) == 0 || m.CompatibleMethod != backup.Method {
 				// disable other incremental backup schedules
 				backupSchedule.Spec.Schedules[i].Enabled = boolptr.False()
@@ -692,10 +701,11 @@ func (r *clusterBackupPolicyTransformer) mergeClusterBackup(
 				}, &backupSchedule.Spec.Schedules[i])
 				hasSyncIncMethod = true
 			}
-		}
-		if as.Spec.BackupType == dpv1alpha1.BackupTypeFull && enableAutoBackup {
-			// disable the automatic backup for other full backup method
-			backupSchedule.Spec.Schedules[i].Enabled = boolptr.False()
+		case dpv1alpha1.BackupTypeFull:
+			if enableAutoBackup {
+				// disable the automatic backup for other full backup method
+				s.Enabled = boolptr.False()
+			}
 		}
 	}
 	if !exist {

deploy/helm/crds/apps.kubeblocks.io_clusters.yaml

@@ -200,6 +200,10 @@ spec:
               backup:
                 description: Specifies the backup configuration of the Cluster.
                 properties:
+                  continuousMethod:
+                    description: Specifies the backup method to use, if not set, use
+                      the first continuous method.
+                    type: string
                   cronExpression:
                     description: The cron expression for the schedule. The timezone
                       is in UTC. See https://en.wikipedia.org/wiki/Cron.

docs/developer_docs/api-reference/cluster.md

@@ -4548,6 +4548,18 @@ bool
 </tr>
 <tr>
 <td>
+<code>continuousMethod</code><br/>
+<em>
+string
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>Specifies the backup method to use, if not set, use the first continuous method.</p>
+</td>
+</tr>
+<tr>
+<td>
 <code>incrementalBackupEnabled</code><br/>
 <em>
 bool

pkg/dataprotection/backup/scheduler.go

@@ -365,18 +365,42 @@ type backupReconfigureRef struct {
 
 type parameterPairs map[string][]appsv1alpha1.ParameterPair
 
+func (s *Scheduler) convertLastAppliedConfigs(continuousMethod string) {
+	if _, ok := s.BackupSchedule.Annotations[dptypes.LastAppliedConfigsAnnotationKey]; ok {
+		return
+	}
+	lastAppliedConfig := s.BackupSchedule.Annotations[constant.LastAppliedConfigAnnotationKey]
+	if lastAppliedConfig == "" {
+		return
+	}
+	lastAppliedConfigMap := map[string]string{}
+	lastAppliedConfigMap[continuousMethod] = lastAppliedConfig
+	str, _ := json.Marshal(lastAppliedConfigMap)
+	s.BackupSchedule.Annotations[dptypes.LastAppliedConfigsAnnotationKey] = string(str)
+}
+
+func (s *Scheduler) getLastAppliedConfigsMap() (map[string]string, error) {
+	resMap := map[string]string{}
+	if err := json.Unmarshal([]byte(s.BackupSchedule.Annotations[dptypes.LastAppliedConfigsAnnotationKey]), &resMap); err != nil {
+		return nil, err
+	}
+	return resMap, nil
+}
+
 func (s *Scheduler) reconfigure(schedulePolicy *dpv1alpha1.SchedulePolicy) error {
 	reCfgRef := s.BackupSchedule.Annotations[dptypes.ReconfigureRefAnnotationKey]
 	if reCfgRef == "" {
 		return nil
 	}
+	// convert deprecated "lastAppliedConfig "to "lastAppliedConfigs"
+	s.convertLastAppliedConfigs(schedulePolicy.BackupMethod)
 	configRef := backupReconfigureRef{}
 	if err := json.Unmarshal([]byte(reCfgRef), &configRef); err != nil {
 		return err
 	}
 
 	enable := boolptr.IsSetToTrue(schedulePolicy.Enabled)
-	if s.BackupSchedule.Annotations[constant.LastAppliedConfigAnnotationKey] == "" && !enable {
+	if s.BackupSchedule.Annotations[dptypes.LastAppliedConfigsAnnotationKey] == "" && !enable {
 		// disable in the first policy created, no need reconfigure because default configs had been set.
 		return nil
 	}
@@ -392,9 +416,13 @@ func (s *Scheduler) reconfigure(schedulePolicy *dpv1alpha1.SchedulePolicy) error
 		// skip reconfigure if not found parameters.
 		return nil
 	}
+	lastAppliedConfigsMap, err := s.getLastAppliedConfigsMap()
+	if err != nil {
+		return err
+	}
 	updateParameterPairsBytes, _ := json.Marshal(parameters)
 	updateParameterPairs := string(updateParameterPairsBytes)
-	if updateParameterPairs == s.BackupSchedule.Annotations[constant.LastAppliedConfigAnnotationKey] {
+	if updateParameterPairs == lastAppliedConfigsMap[schedulePolicy.BackupMethod] {
 		// reconcile the config job if finished
 		return s.reconcileReconfigure(s.BackupSchedule)
 	}
@@ -450,7 +478,10 @@ func (s *Scheduler) reconfigure(schedulePolicy *dpv1alpha1.SchedulePolicy) error
 	if s.BackupSchedule.Annotations == nil {
 		s.BackupSchedule.Annotations = map[string]string{}
 	}
-	s.BackupSchedule.Annotations[constant.LastAppliedConfigAnnotationKey] = updateParameterPairs
+	lastAppliedConfigsMap[schedulePolicy.BackupMethod] = updateParameterPairs
+	updateParameterPairsBytes, _ = json.Marshal(lastAppliedConfigsMap)
+	s.BackupSchedule.Annotations[dptypes.LastAppliedConfigsAnnotationKey] = string(updateParameterPairsBytes)
+	delete(s.BackupSchedule.Annotations, constant.LastAppliedConfigAnnotationKey)
 	if err := s.Client.Patch(s.Ctx, s.BackupSchedule, patch); err != nil {
 		return err
 	}