Issue 7281: batch delete snapshot

changelogs/unreleased/7438-Lyndon-Li

@@ -0,0 +1 @@
+Fix issue #7281, batch delete snapshots in the same repo

pkg/cmd/server/server.go

@@ -793,6 +793,7 @@
 			newPluginManager,
 			backupStoreGetter,
 			s.credentialFileStore,
+			s.repoEnsurer,
 		).SetupWithManager(s.mgr); err != nil {
 			s.logger.Fatal(err, "unable to create controller", "controller", controller.BackupDeletion)
 		}

pkg/controller/backup_deletion_controller.go

@@ -68,6 +68,7 @@
 	newPluginManager  func(logrus.FieldLogger) clientmgmt.Manager
 	backupStoreGetter persistence.ObjectBackupStoreGetter
 	credentialStore   credentials.FileStore
+	repoEnsurer       *repository.Ensurer
 }
 
 // NewBackupDeletionReconciler creates a new backup deletion reconciler.
@@ -81,6 +82,7 @@
 	newPluginManager func(logrus.FieldLogger) clientmgmt.Manager,
 	backupStoreGetter persistence.ObjectBackupStoreGetter,
 	credentialStore credentials.FileStore,
+	repoEnsurer *repository.Ensurer,
 ) *backupDeletionReconciler {
 	return &backupDeletionReconciler{
 		Client:            client,
@@ -93,6 +95,7 @@
 		newPluginManager:  newPluginManager,
 		backupStoreGetter: backupStoreGetter,
 		credentialStore:   credentialStore,
+		repoEnsurer:       repoEnsurer,
 	}
 }
 
@@ -502,20 +505,16 @@
 		return nil
 	}
 
-	snapshots, err := getSnapshotsInBackup(ctx, backup, r.Client)
+	directSnapshots, err := getSnapshotsInBackup(ctx, backup, r.Client)
 	if err != nil {
 		return []error{err}
 	}
 
-	var errs []error
-	for _, snapshot := range snapshots {
-		if err := r.repoMgr.Forget(ctx, snapshot); err != nil {
-			errs = append(errs, err)
-		}
-	}
-	return errs
+	return batchDeleteSnapshots(ctx, r.repoEnsurer, r.repoMgr, directSnapshots, backup, r.logger)
 }
 
+var batchDeleteSnapshotFunc = batchDeleteSnapshots
+
 func (r *backupDeletionReconciler) deleteMovedSnapshots(ctx context.Context, backup *velerov1api.Backup) []error {
 	if r.repoMgr == nil {
 		return nil
@@ -532,26 +531,52 @@
 		return []error{errors.Wrapf(err, "failed to retrieve config for snapshot info")}
 	}
 	var errs []error
+	directSnapshots := map[string][]repository.SnapshotIdentifier{}
 	for i := range list.Items {
 		cm := list.Items[i]
-		snapshot := repository.SnapshotIdentifier{}
+		if cm.Data == nil || len(cm.Data) == 0 {
+			errs = append(errs, errors.New("no snapshot info in config"))
+			continue
+		}
+
 		b, err := json.Marshal(cm.Data)
 		if err != nil {
 			errs = append(errs, errors.Wrapf(err, "fail to marshal the snapshot info into JSON"))
 			continue
 		}
+
+		snapshot := repository.SnapshotIdentifier{}
 		if err := json.Unmarshal(b, &snapshot); err != nil {
 			errs = append(errs, errors.Wrapf(err, "failed to unmarshal snapshot info"))
 			continue
 		}
-		if err := r.repoMgr.Forget(ctx, snapshot); err != nil {
-			errs = append(errs, errors.Wrapf(err, "failed to delete snapshot %s, namespace: %s", snapshot.SnapshotID, snapshot.VolumeNamespace))
+
+		if snapshot.SnapshotID == "" || snapshot.VolumeNamespace == "" || snapshot.RepositoryType == "" {
+			errs = append(errs, errors.Errorf("invalid snapshot, ID %s, namespace %s, repository %s", snapshot.SnapshotID, snapshot.VolumeNamespace, snapshot.RepositoryType))
+			continue
+		}
+
+		if directSnapshots[snapshot.VolumeNamespace] == nil {
+			directSnapshots[snapshot.VolumeNamespace] = []repository.SnapshotIdentifier{}
 		}
-		r.logger.Infof("Deleted snapshot %s, namespace: %s, repo type: %s", snapshot.SnapshotID, snapshot.VolumeNamespace, snapshot.RepositoryType)
+
+		directSnapshots[snapshot.VolumeNamespace] = append(directSnapshots[snapshot.VolumeNamespace], snapshot)
+
+		r.logger.Infof("Deleting snapshot %s, namespace: %s, repo type: %s", snapshot.SnapshotID, snapshot.VolumeNamespace, snapshot.RepositoryType)
+	}
+
+	for i := range list.Items {
+		cm := list.Items[i]
 		if err := r.Client.Delete(ctx, &cm); err != nil {
 			r.logger.Warnf("Failed to delete snapshot info configmap %s/%s: %v", cm.Namespace, cm.Name, err)
 		}
 	}
+
+	if len(directSnapshots) > 0 {
+		deleteErrs := batchDeleteSnapshotFunc(ctx, r.repoEnsurer, r.repoMgr, directSnapshots, backup, r.logger)
+		errs = append(errs, deleteErrs...)
+	}
+
 	return errs
 }
 
@@ -592,7 +617,7 @@
 
 // getSnapshotsInBackup returns a list of all pod volume snapshot ids associated with
 // a given Velero backup.
-func getSnapshotsInBackup(ctx context.Context, backup *velerov1api.Backup, kbClient client.Client) ([]repository.SnapshotIdentifier, error) {
+func getSnapshotsInBackup(ctx context.Context, backup *velerov1api.Backup, kbClient client.Client) (map[string][]repository.SnapshotIdentifier, error) {
 	podVolumeBackups := &velerov1api.PodVolumeBackupList{}
 	options := &client.ListOptions{
 		LabelSelector: labels.Set(map[string]string{
@@ -607,3 +632,31 @@
 
 	return podvolume.GetSnapshotIdentifier(podVolumeBackups), nil
 }
+
+func batchDeleteSnapshots(ctx context.Context, repoEnsurer *repository.Ensurer, repoMgr repository.Manager,
+	directSnapshots map[string][]repository.SnapshotIdentifier, backup *velerov1api.Backup, logger logrus.FieldLogger) []error {
+	var errs []error
+	for volumeNamespace, snapshots := range directSnapshots {
+		batchForget := []string{}
+		for _, snapshot := range snapshots {
+			batchForget = append(batchForget, snapshot.SnapshotID)
+		}
+
+		// For volumes in one backup, the BSL and repositoryType should always be the same
+		repoType := snapshots[0].RepositoryType
+		repo, err := repoEnsurer.EnsureRepo(ctx, backup.Namespace, volumeNamespace, backup.Spec.StorageLocation, repoType)
+		if err != nil {
+			errs = append(errs, errors.Wrapf(err, "error to ensure repo %s-%s-%s, skip deleting PVB snapshots %v", backup.Spec.StorageLocation, volumeNamespace, repoType, batchForget))
+			continue
+		}
+
+		if forgetErrs := repoMgr.BatchForget(ctx, repo, batchForget); len(forgetErrs) > 0 {
+			errs = append(errs, forgetErrs...)
+			continue
+		}
+
+		logger.Infof("Batch deleted snapshots %v", batchForget)
+	}
+
+	return errs
+}