chore: validate sharding restore and incremental backups (#9944)

Author: loomts

controllers/apps/cluster/transformer_cluster_restore.go

@@ -20,6 +20,8 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
 package cluster
 
 import (
+	"sort"
+
 	"k8s.io/apimachinery/pkg/util/json"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
@@ -72,43 +74,56 @@ func (c *clusterRestoreTransformer) Transform(ctx graph.TransformContext, dag *g
 		if err != nil {
 			return err
 		}
+
+		targets := backup.Status.Targets
 		// obtain components that have already been assigned targets.
 		allocateTargetMap := map[string]string{}
 		restoreDoneForShardComponents := true
 		for _, v := range shardComponents {
 			if model.IsObjectDeleting(&v) {
 				continue
 			}
-			if v.Annotations[constant.RestoreDoneAnnotationKey] != "true" {
+
+			compName := v.Labels[constant.KBAppComponentLabelKey]
+			compAnnotations := c.initClusterAnnotations(compName)
+
+			if v.Annotations[constant.BackupSourceTargetAnnotationKey] != "" && v.Annotations[constant.RestoreDoneAnnotationKey] != "true" {
 				restoreDoneForShardComponents = false
 			}
 			if targetName, ok := v.Annotations[constant.BackupSourceTargetAnnotationKey]; ok {
-				compName := v.Labels[constant.KBAppComponentLabelKey]
 				allocateTargetMap[targetName] = compName
-				c.initClusterAnnotations(compName)
-				c.annotations[compName][constant.BackupSourceTargetAnnotationKey] = targetName
+				compAnnotations[constant.BackupSourceTargetAnnotationKey] = targetName
 			}
 		}
-		if len(allocateTargetMap) == len(backup.Status.Targets) {
+		if len(allocateTargetMap) == len(targets) {
 			// check if the restore is completed when all source target have allocated.
 			if err = c.cleanupRestoreAnnotationForSharding(dag, spec.Name, restoreDoneForShardComponents); err != nil {
 				return err
 			}
-			continue
 		}
-		for _, target := range backup.Status.Targets {
+		// guarantee that when available targets are fewer than shards, the first shards are prioritized for restore.
+		sort.Slice(c.shardingComps[spec.Name], func(i, j int) bool {
+			return c.shardingComps[spec.Name][i].Name < c.shardingComps[spec.Name][j].Name
+		})
+		for _, target := range targets {
 			if _, ok = allocateTargetMap[target.Name]; ok {
 				continue
 			}
 			for _, compSpec := range c.shardingComps[spec.Name] {
-				if _, ok = c.annotations[compSpec.Name][constant.BackupSourceTargetAnnotationKey]; ok {
+				compAnnotations := c.initClusterAnnotations(compSpec.Name)
+				if _, ok = compAnnotations[constant.BackupSourceTargetAnnotationKey]; ok {
 					continue
 				}
-				c.initClusterAnnotations(compSpec.Name)
-				c.annotations[compSpec.Name][constant.BackupSourceTargetAnnotationKey] = target.Name
+				compAnnotations[constant.BackupSourceTargetAnnotationKey] = target.Name
 				break
 			}
 		}
+		for _, compSpec := range c.shardingComps[spec.Name] {
+			compAnnotations := c.initClusterAnnotations(compSpec.Name)
+			if compAnnotations[constant.BackupSourceTargetAnnotationKey] == "" {
+				compAnnotations[constant.SkipRestoreAnnotationKey] = "true"
+			}
+		}
 	}
 	// if component needs to do post ready restore after cluster is running, annotate component
 	if c.Cluster.Status.Phase == appsv1.RunningClusterPhase {
@@ -132,13 +147,14 @@ func (c *clusterRestoreTransformer) Transform(ctx graph.TransformContext, dag *g
 	return nil
 }
 
-func (c *clusterRestoreTransformer) initClusterAnnotations(compName string) {
+func (c *clusterRestoreTransformer) initClusterAnnotations(compName string) map[string]string {
 	if c.annotations == nil {
-		c.annotations = map[string]map[string]string{}
+		c.annotations = make(map[string]map[string]string)
 	}
 	if c.annotations[compName] == nil {
-		c.annotations[compName] = map[string]string{}
+		c.annotations[compName] = make(map[string]string)
 	}
+	return c.annotations[compName]
 }
 
 func (c *clusterRestoreTransformer) cleanupRestoreAnnotationForSharding(dag *graph.DAG,

controllers/dataprotection/backup_controller.go

@@ -1078,7 +1078,7 @@ func prepare4Incremental(request *dpbackup.Request) (*dpbackup.Request, error) {
 		return nil, fmt.Errorf("backupRepo for incremental backup can't be empty")
 	}
 	// get and validate parent backup
-	parentBackup, err := GetParentBackup(request.Ctx, request.Client, request.Backup, request.BackupMethod, request.BackupRepo.Name)
+	parentBackup, err := GetParentBackup(request.Ctx, request.Client, request.Backup, request.BackupMethod, request.BackupPolicy, request.BackupRepo.Name)
 	if err != nil {
 		return nil, err
 	}

controllers/dataprotection/backup_controller_test.go

@@ -681,6 +681,13 @@ var _ = Describe("Backup Controller test", func() {
 					Phase:               dpv1alpha1.BackupPhaseCompleted,
 					CompletionTimestamp: &metav1.Time{Time: time.Now().Add(-time.Hour * 24)},
 					BackupRepoName:      testdp.BackupRepoName,
+					Target: &dpv1alpha1.BackupStatusTarget{
+						BackupTarget: dpv1alpha1.BackupTarget{
+							PodSelector: &dpv1alpha1.PodSelector{
+								Strategy: dpv1alpha1.PodSelectionStrategyAny,
+							},
+						},
+					},
 					EncryptionConfig: &dpv1alpha1.EncryptionConfig{
 						Algorithm: "AES-256-CFB",
 						PassPhraseSecretKeyRef: &corev1.SecretKeySelector{
@@ -915,6 +922,13 @@ var _ = Describe("Backup Controller test", func() {
 					BackupRepoName:   repoName,
 					ParentBackupName: parentBackup,
 					BaseBackupName:   baseBackup,
+					Target: &dpv1alpha1.BackupStatusTarget{
+						BackupTarget: dpv1alpha1.BackupTarget{
+							PodSelector: &dpv1alpha1.PodSelector{
+								Strategy: dpv1alpha1.PodSelectionStrategyAny,
+							},
+						},
+					},
 					TimeRange: &dpv1alpha1.BackupTimeRange{
 						Start: step(),
 						End:   step(),

controllers/dataprotection/backuppolicydriver_controller.go

@@ -218,13 +218,17 @@ func (r *backupPolicyAndScheduleBuilder) transformBackupPolicy() (*dpv1alpha1.Ba
 				Annotations: r.buildAnnotations(),
 			},
 		}
-		r.buildBackupPolicy(backupPolicy)
+		if err := r.buildBackupPolicy(backupPolicy); err != nil {
+			return nil, err
+		}
 		if err := controllerutil.SetControllerReference(r.Cluster, backupPolicy, r.schema); err != nil {
 			return nil, err
 		}
 		return backupPolicy, r.Client.Create(r.Context, backupPolicy)
 	}
-	r.buildBackupPolicy(backupPolicy)
+	if err := r.buildBackupPolicy(backupPolicy); err != nil {
+		return nil, err
+	}
 	return backupPolicy, r.Client.Update(r.Context, backupPolicy)
 }
 
@@ -325,7 +329,7 @@ func (r *backupPolicyAndScheduleBuilder) syncBackupSchedule(backupSchedule *dpv1
 }
 
 // buildBackupPolicy builds a new backup policy by the backup policy template.
-func (r *backupPolicyAndScheduleBuilder) buildBackupPolicy(backupPolicy *dpv1alpha1.BackupPolicy) {
+func (r *backupPolicyAndScheduleBuilder) buildBackupPolicy(backupPolicy *dpv1alpha1.BackupPolicy) error {
 	bpSpec := &backupPolicy.Spec
 	// if cluster have backup repo, set backup repo name to backup policy.
 	if r.Cluster.Spec.Backup != nil && r.Cluster.Spec.Backup.RepoName != "" {
@@ -338,17 +342,24 @@ func (r *backupPolicyAndScheduleBuilder) buildBackupPolicy(backupPolicy *dpv1alp
 	intctrlutil.MergeMetadataMapInplace(r.buildAnnotations(), &backupPolicy.Annotations)
 	intctrlutil.MergeMetadataMapInplace(r.buildLabels(), &backupPolicy.Labels)
 
-	r.buildBackupMethods(backupPolicy)
+	if err := r.buildBackupMethods(backupPolicy); err != nil {
+		return err
+	}
 
 	if needSyncFromTemplate(backupPolicy) {
 		bpSpec.BackoffLimit = r.backupPolicyTPL.Spec.BackoffLimit
 		bpSpec.RetentionPolicy = r.backupPolicyTPL.Spec.RetentionPolicy
 		if r.isSharding {
-			bpSpec.Targets = r.buildBackupTargets(backupPolicy.Spec.Targets)
+			targets, err := r.buildBackupTargets(backupPolicy.Spec.Targets)
+			if err != nil {
+				return err
+			}
+			bpSpec.Targets = targets
 		} else {
 			bpSpec.Target = r.buildBackupTarget(backupPolicy.Spec.Target, r.backupPolicyTPL.Spec.Target, r.componentName)
 		}
 	}
+	return nil
 }
 
 func (r *backupPolicyAndScheduleBuilder) setDefaultEncryptionConfig(backupPolicy *dpv1alpha1.BackupPolicy) {
@@ -400,7 +411,7 @@ func (r *backupPolicyAndScheduleBuilder) syncRoleLabelSelectorWhenReplicaChanges
 }
 
 // buildBackupMethods build the backupMethod of tpl to backupPolicy.
-func (r *backupPolicyAndScheduleBuilder) buildBackupMethods(backupPolicy *dpv1alpha1.BackupPolicy) {
+func (r *backupPolicyAndScheduleBuilder) buildBackupMethods(backupPolicy *dpv1alpha1.BackupPolicy) error {
 	var backupMethods []dpv1alpha1.BackupMethod
 	oldBackupMethodMap := map[string]dpv1alpha1.BackupMethod{}
 	for _, v := range backupPolicy.Spec.BackupMethods {
@@ -421,7 +432,11 @@ func (r *backupPolicyAndScheduleBuilder) buildBackupMethods(backupPolicy *dpv1al
 		}
 		if backupMethodTPL.Target != nil {
 			if r.isSharding {
-				backupMethod.Targets = r.buildBackupTargets(backupMethod.Targets)
+				targets, err := r.buildBackupTargets(backupMethod.Targets)
+				if err != nil {
+					return err
+				}
+				backupMethod.Targets = targets
 			} else {
 				backupMethod.Target = r.buildBackupTarget(backupMethod.Target, *backupMethodTPL.Target, r.componentName)
 			}
@@ -430,6 +445,7 @@ func (r *backupPolicyAndScheduleBuilder) buildBackupMethods(backupPolicy *dpv1al
 		backupMethods = append(backupMethods, backupMethod)
 	}
 	backupPolicy.Spec.BackupMethods = backupMethods
+	return nil
 }
 
 func (r *backupPolicyAndScheduleBuilder) resolveBackupMethodEnv(compSpec *appsv1.ClusterComponentSpec, envs []dpv1alpha1.EnvVar) []corev1.EnvVar {
@@ -460,8 +476,15 @@ func (r *backupPolicyAndScheduleBuilder) matchMappingName(names []string, target
 	return false
 }
 
-func (r *backupPolicyAndScheduleBuilder) buildBackupTargets(targets []dpv1alpha1.BackupTarget) []dpv1alpha1.BackupTarget {
-	shardComponents, _ := sharding.ListShardingComponents(r.Context, r.Client, r.Cluster, r.componentName)
+func (r *backupPolicyAndScheduleBuilder) buildBackupTargets(targets []dpv1alpha1.BackupTarget) ([]dpv1alpha1.BackupTarget, error) {
+	shardComponents, err := sharding.ListShardingComponents(r.Context, r.Client, r.Cluster, r.componentName)
+	if err != nil {
+		return nil, err
+	}
+	if len(shardComponents) == 0 {
+		return nil, fmt.Errorf("sharding components %s not found", r.componentName)
+	}
+
 	sourceTargetMap := map[string]*dpv1alpha1.BackupTarget{}
 	for i := range targets {
 		sourceTargetMap[targets[i].Name] = &targets[i]
@@ -474,7 +497,7 @@ func (r *backupPolicyAndScheduleBuilder) buildBackupTargets(targets []dpv1alpha1
 			backupTargets = append(backupTargets, *target)
 		}
 	}
-	return backupTargets
+	return backupTargets, nil
 }
 
 func (r *backupPolicyAndScheduleBuilder) buildBackupTarget(

controllers/dataprotection/gc_controller_test.go

@@ -88,6 +88,13 @@ var _ = Describe("Data Protection Garbage Collection Controller", func() {
 				Expiration:          &now,
 				StartTimestamp:      &now,
 				CompletionTimestamp: &now,
+				Target: &dpv1alpha1.BackupStatusTarget{
+					BackupTarget: dpv1alpha1.BackupTarget{
+						PodSelector: &dpv1alpha1.PodSelector{
+							Strategy: dpv1alpha1.PodSelectionStrategyAny,
+						},
+					},
+				},
 			}
 			autoBackupLabel = map[string]string{
 				dptypes.AutoBackupLabelKey:     "true",
@@ -199,6 +206,13 @@ var _ = Describe("Data Protection Garbage Collection Controller", func() {
 			olderBackup.Status.CompletionTimestamp = &metav1.Time{Time: expiredTime.Time.Add(-time.Hour * 3)}
 			olderBackup.Status.Phase = dpv1alpha1.BackupPhaseCompleted
 			olderBackup.Status.BackupRepoName = testdp.BackupRepoName
+			olderBackup.Status.Target = &dpv1alpha1.BackupStatusTarget{
+				BackupTarget: dpv1alpha1.BackupTarget{
+					PodSelector: &dpv1alpha1.PodSelector{
+						Strategy: dpv1alpha1.PodSelectionStrategyAny,
+					},
+				},
+			}
 			testdp.PatchBackupStatus(&testCtx, olderKey, olderBackup.Status)
 
 			By("the older full backup should be not deleted, it is the latest backup for now")
@@ -221,6 +235,13 @@ var _ = Describe("Data Protection Garbage Collection Controller", func() {
 			incrementalBackup.Status.StartTimestamp = &metav1.Time{Time: expiredTime.Time.Add(-time.Hour * 2)}
 			incrementalBackup.Status.CompletionTimestamp = &metav1.Time{Time: expiredTime.Time.Add(-time.Hour * 2)}
 			incrementalBackup.Status.Phase = dpv1alpha1.BackupPhaseCompleted
+			incrementalBackup.Status.Target = &dpv1alpha1.BackupStatusTarget{
+				BackupTarget: dpv1alpha1.BackupTarget{
+					PodSelector: &dpv1alpha1.PodSelector{
+						Strategy: dpv1alpha1.PodSelectionStrategyAny,
+					},
+				},
+			}
 			testdp.PatchBackupStatus(&testCtx, incrementalKey, incrementalBackup.Status)
 
 			By("the incremental backup should be not deleted, its parent is the latest backup for now")
@@ -243,6 +264,13 @@ var _ = Describe("Data Protection Garbage Collection Controller", func() {
 			latestBackup.Status.StartTimestamp = &metav1.Time{Time: expiredTime.Time.Add(-time.Hour)}
 			latestBackup.Status.CompletionTimestamp = &metav1.Time{Time: expiredTime.Time.Add(-time.Hour)}
 			latestBackup.Status.Phase = dpv1alpha1.BackupPhaseCompleted
+			latestBackup.Status.Target = &dpv1alpha1.BackupStatusTarget{
+				BackupTarget: dpv1alpha1.BackupTarget{
+					PodSelector: &dpv1alpha1.PodSelector{
+						Strategy: dpv1alpha1.PodSelectionStrategyAny,
+					},
+				},
+			}
 			testdp.PatchBackupStatus(&testCtx, latestKey, latestBackup.Status)
 
 			By("verify the latest full backup is retained while older is deleted")

controllers/dataprotection/utils.go

@@ -590,7 +590,7 @@ func fromFlattenName(flatten string) (name string, namespace string) {
 // then validate and return the parent backup.
 // If parentBackupName is not specified, find the latest valid parent backup.
 func GetParentBackup(ctx context.Context, cli client.Client, backup *dpv1alpha1.Backup,
-	backupMethod *dpv1alpha1.BackupMethod, backupRepoName string) (*dpv1alpha1.Backup, error) {
+	backupMethod *dpv1alpha1.BackupMethod, backupPolicy *dpv1alpha1.BackupPolicy, backupRepoName string) (*dpv1alpha1.Backup, error) {
 	if backup == nil || backupMethod == nil {
 		return nil, fmt.Errorf("backup or backupMethod is nil")
 	}
@@ -616,16 +616,23 @@ func GetParentBackup(ctx context.Context, cli client.Client, backup *dpv1alpha1.
 		}, parentBackup); err != nil {
 			return nil, err
 		}
-		if err := ValidateParentBackup(ctx, cli, backup, parentBackup, backupMethod, backupRepoName); err != nil {
+		if err := ValidateParentBackup(ctx, cli, backup, parentBackup, backupMethod, backupPolicy, backupRepoName); err != nil {
 			return nil, fmt.Errorf("failed to validate parent backup %s: %w", parentBackupName, err)
 		}
 		return parentBackup, nil
 	}
-	parentBackup, err := FindParentBackupIfNotSet(ctx, cli, backup, backupMethod, backupRepoName)
+	parentBackup, err := FindParentBackupIfNotSet(ctx, cli, backup, backupMethod, backupPolicy, backupRepoName)
 	if err != nil {
 		return nil, fmt.Errorf("failed to find parent backup: %w", err)
 	}
 	if parentBackup == nil {
+		// output sharding message
+		expectedTargets := dputils.GetBackupTargets(backupPolicy, backupMethod)
+		if len(expectedTargets) > 0 {
+			return nil, fmt.Errorf("failed to find a valid parent backup for backup %s/%s: "+
+				"current backup expects %d shards, this may be due to shard count mismatch with existing backups",
+				backup.Namespace, backup.Name, len(expectedTargets))
+		}
 		return nil, fmt.Errorf("failed to find a valid parent backup for backup %s/%s", backup.Namespace, backup.Name)
 	}
 	return parentBackup, nil
@@ -637,7 +644,7 @@ func GetParentBackup(ctx context.Context, cli client.Client, backup *dpv1alpha1.
 // b. return the latest incremental backup.
 // c. return the latest full backup if incremental backups are not found.
 func FindParentBackupIfNotSet(ctx context.Context, cli client.Client, backup *dpv1alpha1.Backup,
-	backupMethod *dpv1alpha1.BackupMethod, backupRepoName string) (*dpv1alpha1.Backup, error) {
+	backupMethod *dpv1alpha1.BackupMethod, backupPolicy *dpv1alpha1.BackupPolicy, backupRepoName string) (*dpv1alpha1.Backup, error) {
 	getLatestBackup := func(backupList []*dpv1alpha1.Backup) *dpv1alpha1.Backup {
 		if len(backupList) == 0 {
 			return nil
@@ -655,7 +662,7 @@ func FindParentBackupIfNotSet(ctx context.Context, cli client.Client, backup *dp
 			client.MatchingLabels(labels)); err != nil && !apierrors.IsNotFound(err) {
 			return nil, err
 		}
-		filteredbackupList := FilterParentBackups(ctx, cli, backupList, backup, backupMethod, incremental, backupRepoName)
+		filteredbackupList := FilterParentBackups(ctx, cli, backupList, backup, backupMethod, backupPolicy, incremental, backupRepoName)
 		return getLatestBackup(filteredbackupList), nil
 	}
 
@@ -701,13 +708,13 @@ func FindParentBackupIfNotSet(ctx context.Context, cli client.Client, backup *dp
 
 // FilterParentBackups filters the parent backups by backup phase, backup method and end time.
 func FilterParentBackups(ctx context.Context, cli client.Client, backupList *dpv1alpha1.BackupList, targetBackup *dpv1alpha1.Backup,
-	backupMethod *dpv1alpha1.BackupMethod, incremental bool, backupRepoName string) []*dpv1alpha1.Backup {
+	backupMethod *dpv1alpha1.BackupMethod, backupPolicy *dpv1alpha1.BackupPolicy, incremental bool, backupRepoName string) []*dpv1alpha1.Backup {
 	var res []*dpv1alpha1.Backup
 	if backupList == nil || len(backupList.Items) == 0 {
 		return res
 	}
 	for i, backup := range backupList.Items {
-		if err := ValidateParentBackup(ctx, cli, targetBackup, &backup, backupMethod, backupRepoName); err != nil {
+		if err := ValidateParentBackup(ctx, cli, targetBackup, &backup, backupMethod, backupPolicy, backupRepoName); err != nil {
 			continue
 		}
 		// backups are listed by backup type label, validate if the backup method matches
@@ -728,7 +735,7 @@ func FilterParentBackups(ctx context.Context, cli client.Client, backupList *dpv
 
 // ValidateParentBackup validates the parent backup.
 func ValidateParentBackup(ctx context.Context, cli client.Client, backup *dpv1alpha1.Backup, parentBackup *dpv1alpha1.Backup,
-	backupMethod *dpv1alpha1.BackupMethod, backupRepoName string) error {
+	backupMethod *dpv1alpha1.BackupMethod, backupPolicy *dpv1alpha1.BackupPolicy, backupRepoName string) error {
 	// validate parent backup is completed
 	if parentBackup.Status.Phase != dpv1alpha1.BackupPhaseCompleted {
 		return fmt.Errorf("parent backup %s/%s is not completed", parentBackup.Namespace, parentBackup.Name)
@@ -759,6 +766,17 @@ func ValidateParentBackup(ctx context.Context, cli client.Client, backup *dpv1al
 				parentBackup.Namespace, parentBackup.Name, err)
 		}
 	}
+	// validate target count consistency for incremental backup
+	expectedTargets := dputils.GetBackupTargets(backupPolicy, backupMethod)
+	parentTargetCount := len(parentBackup.Status.Targets)
+	if parentBackup.Status.Target != nil {
+		parentTargetCount = 1
+	}
+	if parentTargetCount != len(expectedTargets) {
+		return fmt.Errorf("parent backup %s/%s has %d targets, but current backup expects %d targets; "+
+			"incremental backup requires consistent targets, it may happen by shards changed",
+			parentBackup.Namespace, parentBackup.Name, parentTargetCount, len(expectedTargets))
+	}
 	return nil
 }