Merge branch 'main' into issue-fix-7535

pkg/backup/backup_pv_action_test.go

@@ -50,14 +50,14 @@ func TestBackupPVAction(t *testing.T) {
 	// and no additional items
 	_, additional, err := a.Execute(pvc, backup)
 	assert.NoError(t, err)
-	assert.Len(t, additional, 0)
+	assert.Empty(t, additional)
 
 	// empty spec.volumeName should result in no error
 	// and no additional items
 	pvc.Object["spec"].(map[string]interface{})["volumeName"] = ""
 	_, additional, err = a.Execute(pvc, backup)
 	assert.NoError(t, err)
-	assert.Len(t, additional, 0)
+	assert.Empty(t, additional)
 
 	// Action should clean the spec.Selector when the StorageClassName is not set.
 	input := builder.ForPersistentVolumeClaim("abc", "abc").VolumeName("pv").Selector(&metav1.LabelSelector{MatchLabels: map[string]string{"abc": "abc"}}).Phase(corev1.ClaimBound).Result()
@@ -119,21 +119,21 @@ func TestBackupPVAction(t *testing.T) {
 	pvc.Object["spec"].(map[string]interface{})["volumeName"] = "myVolume"
 	_, additional, err = a.Execute(pvc, backup)
 	require.NoError(t, err)
-	require.Len(t, additional, 0)
+	require.Empty(t, additional)
 
 	// non-empty spec.volumeName when status.phase is 'Pending'
 	// should result in no error and no additional items
 	pvc.Object["status"].(map[string]interface{})["phase"] = corev1api.ClaimPending
 	_, additional, err = a.Execute(pvc, backup)
 	require.NoError(t, err)
-	require.Len(t, additional, 0)
+	require.Empty(t, additional)
 
 	// non-empty spec.volumeName when status.phase is 'Lost'
 	// should result in no error and no additional items
 	pvc.Object["status"].(map[string]interface{})["phase"] = corev1api.ClaimLost
 	_, additional, err = a.Execute(pvc, backup)
 	require.NoError(t, err)
-	require.Len(t, additional, 0)
+	require.Empty(t, additional)
 
 	// non-empty spec.volumeName when status.phase is 'Bound'
 	// should result in no error and one additional item for the PV
@@ -148,5 +148,5 @@ func TestBackupPVAction(t *testing.T) {
 	pvc.Object["spec"].(map[string]interface{})["volumeName"] = ""
 	_, additional, err = a.Execute(pvc, backup)
 	assert.NoError(t, err)
-	assert.Len(t, additional, 0)
+	assert.Empty(t, additional)
 }

pkg/backup/pv_skip_tracker_test.go

@@ -60,5 +60,5 @@ func TestTrackUntrack(t *testing.T) {
 	tracker.Track("pv3", podVolumeApproach, "it's set to opt-out")
 	tracker.Untrack("pv3")
 	tracker.Track("pv3", csiSnapshotApproach, "not applicable for CSI ")
-	assert.Equal(t, 0, len(tracker.Summary()))
+	assert.Empty(t, tracker.Summary())
 }

pkg/client/config_test.go

@@ -91,7 +91,7 @@ func TestConfigOperations(t *testing.T) {
 
 	// Test Features
 	feature := savedConfig.Features()
-	assert.Equal(t, 1, len(feature))
+	assert.Len(t, feature, 1)
 	assert.Equal(t, expectedFeature, feature[0])
 
 	// Test Colorized

pkg/cmd/cli/backuplocation/delete_test.go

@@ -89,13 +89,13 @@ func TestDeleteFunctions(t *testing.T) {
 	bkrepoList := velerov1api.BackupRepositoryList{}
 	t.Run("findAssociatedBackups", func(t *testing.T) {
 		bkList, e := findAssociatedBackups(kbclient, "bk-loc-1", "ns1")
-		assert.Equal(t, 0, len(bkList.Items))
+		assert.Empty(t, bkList.Items)
 		assert.NoError(t, e)
 	})
 
 	t.Run("findAssociatedBackupRepos", func(t *testing.T) {
 		bkrepoList, e := findAssociatedBackupRepos(kbclient, "bk-loc-1", "ns1")
-		assert.Equal(t, 0, len(bkrepoList.Items))
+		assert.Empty(t, bkrepoList.Items)
 		assert.NoError(t, e)
 	})
 
@@ -104,7 +104,7 @@ func TestDeleteFunctions(t *testing.T) {
 		bk.Name = "bk-name-last"
 		bkList.Items = append(bkList.Items, bk)
 		errList := deleteBackups(kbclient, bkList)
-		assert.Equal(t, 1, len(errList))
+		assert.Len(t, errList, 1)
 		assert.Contains(t, errList[0].Error(), fmt.Sprintf("delete backup \"%s\" associated with deleted BSL: backups.velero.io \"%s\" not found", bk.Name, bk.Name))
 	})
 	t.Run("deleteBackupRepos", func(t *testing.T) {
@@ -112,7 +112,7 @@ func TestDeleteFunctions(t *testing.T) {
 		bkrepo.Name = "bk-repo-name-last"
 		bkrepoList.Items = append(bkrepoList.Items, bkrepo)
 		errList := deleteBackupRepos(kbclient, bkrepoList)
-		assert.Equal(t, 1, len(errList))
+		assert.Len(t, errList, 1)
 		assert.Contains(t, errList[0].Error(), fmt.Sprintf("delete backup repository \"%s\" associated with deleted BSL: backuprepositories.velero.io \"%s\" not found", bkrepo.Name, bkrepo.Name))
 	})
 }

pkg/cmd/server/server.go

@@ -799,6 +799,7 @@ func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string
 			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 @@ type backupDeletionReconciler struct {
 	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 @@ func NewBackupDeletionReconciler(
 	newPluginManager func(logrus.FieldLogger) clientmgmt.Manager,
 	backupStoreGetter persistence.ObjectBackupStoreGetter,
 	credentialStore credentials.FileStore,
+	repoEnsurer *repository.Ensurer,
 ) *backupDeletionReconciler {
 	return &backupDeletionReconciler{
 		Client:            client,
@@ -93,6 +95,7 @@ func NewBackupDeletionReconciler(
 		newPluginManager:  newPluginManager,
 		backupStoreGetter: backupStoreGetter,
 		credentialStore:   credentialStore,
+		repoEnsurer:       repoEnsurer,
 	}
 }
 
@@ -502,20 +505,16 @@ func (r *backupDeletionReconciler) deletePodVolumeSnapshots(ctx context.Context,
 		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 @@ func (r *backupDeletionReconciler) deleteMovedSnapshots(ctx context.Context, bac
 		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
 		}
-		r.logger.Infof("Deleted snapshot %s, namespace: %s, repo type: %s", snapshot.SnapshotID, snapshot.VolumeNamespace, snapshot.RepositoryType)
+
+		if directSnapshots[snapshot.VolumeNamespace] == nil {
+			directSnapshots[snapshot.VolumeNamespace] = []repository.SnapshotIdentifier{}
+		}
+
+		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 @@ func (r *backupDeletionReconciler) patchBackup(ctx context.Context, backup *vele
 
 // 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 @@ func getSnapshotsInBackup(ctx context.Context, backup *velerov1api.Backup, kbCli
 
 	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
+}

pkg/controller/backup_deletion_controller_test.go

@@ -18,9 +18,11 @@ package controller
 
 import (
 	"bytes"
+	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
-	"sort"
+	"reflect"
 	"time"
 
 	"context"
@@ -53,6 +55,7 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
 	"github.com/vmware-tanzu/velero/pkg/plugin/velero/mocks"
 	"github.com/vmware-tanzu/velero/pkg/repository"
+	repomocks "github.com/vmware-tanzu/velero/pkg/repository/mocks"
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
 )
 
@@ -94,6 +97,7 @@ func setupBackupDeletionControllerTest(t *testing.T, req *velerov1api.DeleteBack
 			func(logrus.FieldLogger) clientmgmt.Manager { return pluginManager },
 			NewFakeSingleObjectBackupStoreGetter(backupStore),
 			velerotest.NewFakeCredentialsFileStore("", nil),
+			nil,
 		),
 		req: ctrl.Request{NamespacedName: types.NamespacedName{Namespace: req.Namespace, Name: req.Name}},
 	}
@@ -138,7 +142,7 @@ func TestBackupDeletionControllerReconcile(t *testing.T) {
 		err = td.fakeClient.Get(ctx, td.req.NamespacedName, res)
 		require.NoError(t, err)
 		assert.Equal(t, "Processed", string(res.Status.Phase))
-		assert.Equal(t, 1, len(res.Status.Errors))
+		assert.Len(t, res.Status.Errors, 1)
 		assert.Equal(t, "spec.backupName is required", res.Status.Errors[0])
 	})
 
@@ -206,7 +210,7 @@ func TestBackupDeletionControllerReconcile(t *testing.T) {
 		err = td.fakeClient.Get(ctx, td.req.NamespacedName, res)
 		require.NoError(t, err)
 		assert.Equal(t, "Processed", string(res.Status.Phase))
-		assert.Equal(t, 1, len(res.Status.Errors))
+		assert.Len(t, res.Status.Errors, 1)
 		assert.Equal(t, "backup is still in progress", res.Status.Errors[0])
 	})
 
@@ -221,7 +225,7 @@ func TestBackupDeletionControllerReconcile(t *testing.T) {
 		err = td.fakeClient.Get(ctx, td.req.NamespacedName, res)
 		require.NoError(t, err)
 		assert.Equal(t, "Processed", string(res.Status.Phase))
-		assert.Equal(t, 1, len(res.Status.Errors))
+		assert.Len(t, res.Status.Errors, 1)
 		assert.Equal(t, "backup not found", res.Status.Errors[0])
 	})
 	t.Run("unable to find backup storage location", func(t *testing.T) {
@@ -236,7 +240,7 @@ func TestBackupDeletionControllerReconcile(t *testing.T) {
 		err = td.fakeClient.Get(ctx, td.req.NamespacedName, res)
 		require.NoError(t, err)
 		assert.Equal(t, "Processed", string(res.Status.Phase))
-		assert.Equal(t, 1, len(res.Status.Errors))
+		assert.Len(t, res.Status.Errors, 1)
 		assert.Equal(t, "backup storage location default not found", res.Status.Errors[0])
 	})
 
@@ -253,7 +257,7 @@ func TestBackupDeletionControllerReconcile(t *testing.T) {
 		err = td.fakeClient.Get(ctx, td.req.NamespacedName, res)
 		require.NoError(t, err)
 		assert.Equal(t, "Processed", string(res.Status.Phase))
-		assert.Equal(t, 1, len(res.Status.Errors))
+		assert.Len(t, res.Status.Errors, 1)
 		assert.Equal(t, "cannot delete backup because backup storage location default is currently in read-only mode", res.Status.Errors[0])
 	})
 	t.Run("full delete, no errors", func(t *testing.T) {
@@ -684,7 +688,7 @@ func TestBackupDeletionControllerReconcile(t *testing.T) {
 		err = td.fakeClient.Get(ctx, td.req.NamespacedName, res)
 		require.NoError(t, err)
 		assert.Equal(t, "Processed", string(res.Status.Phase))
-		assert.Equal(t, 1, len(res.Status.Errors))
+		assert.Len(t, res.Status.Errors, 1)
 		assert.Equal(t, "backup not found", res.Status.Errors[0])
 
 	})
@@ -694,13 +698,13 @@ func TestGetSnapshotsInBackup(t *testing.T) {
 	tests := []struct {
 		name                  string
 		podVolumeBackups      []velerov1api.PodVolumeBackup
-		expected              []repository.SnapshotIdentifier
+		expected              map[string][]repository.SnapshotIdentifier
 		longBackupNameEnabled bool
 	}{
 		{
 			name:             "no pod volume backups",
 			podVolumeBackups: nil,
-			expected:         nil,
+			expected:         map[string][]repository.SnapshotIdentifier{},
 		},
 		{
 			name: "no pod volume backups with matching label",
@@ -720,7 +724,7 @@ func TestGetSnapshotsInBackup(t *testing.T) {
 					Status: velerov1api.PodVolumeBackupStatus{SnapshotID: "snap-2"},
 				},
 			},
-			expected: nil,
+			expected: map[string][]repository.SnapshotIdentifier{},
 		},
 		{
 			name: "some pod volume backups with matching label",
@@ -761,16 +765,18 @@ func TestGetSnapshotsInBackup(t *testing.T) {
 					Status: velerov1api.PodVolumeBackupStatus{SnapshotID: ""},
 				},
 			},
-			expected: []repository.SnapshotIdentifier{
-				{
-					VolumeNamespace: "ns-1",
-					SnapshotID:      "snap-3",
-					RepositoryType:  "restic",
-				},
-				{
-					VolumeNamespace: "ns-1",
-					SnapshotID:      "snap-4",
-					RepositoryType:  "restic",
+			expected: map[string][]repository.SnapshotIdentifier{
+				"ns-1": {
+					{
+						VolumeNamespace: "ns-1",
+						SnapshotID:      "snap-3",
+						RepositoryType:  "restic",
+					},
+					{
+						VolumeNamespace: "ns-1",
+						SnapshotID:      "snap-4",
+						RepositoryType:  "restic",
+					},
 				},
 			},
 		},
@@ -814,11 +820,13 @@ func TestGetSnapshotsInBackup(t *testing.T) {
 					Status: velerov1api.PodVolumeBackupStatus{SnapshotID: ""},
 				},
 			},
-			expected: []repository.SnapshotIdentifier{
-				{
-					VolumeNamespace: "ns-1",
-					SnapshotID:      "snap-3",
-					RepositoryType:  "restic",
+			expected: map[string][]repository.SnapshotIdentifier{
+				"ns-1": {
+					{
+						VolumeNamespace: "ns-1",
+						SnapshotID:      "snap-3",
+						RepositoryType:  "restic",
+					},
 				},
 			},
 		},
@@ -843,21 +851,179 @@ func TestGetSnapshotsInBackup(t *testing.T) {
 			res, err := getSnapshotsInBackup(context.TODO(), veleroBackup, clientBuilder.Build())
 			assert.NoError(t, err)
 
-			// sort to ensure good compare of slices
-			less := func(snapshots []repository.SnapshotIdentifier) func(i, j int) bool {
-				return func(i, j int) bool {
-					if snapshots[i].VolumeNamespace == snapshots[j].VolumeNamespace {
-						return snapshots[i].SnapshotID < snapshots[j].SnapshotID
-					}
-					return snapshots[i].VolumeNamespace < snapshots[j].VolumeNamespace
+			assert.True(t, reflect.DeepEqual(res, test.expected))
+		})
+	}
+}
+
+func batchDeleteSucceed(ctx context.Context, repoEnsurer *repository.Ensurer, repoMgr repository.Manager, directSnapshots map[string][]repository.SnapshotIdentifier, backup *velerov1api.Backup, logger logrus.FieldLogger) []error {
+	return nil
+}
+
+func batchDeleteFail(ctx context.Context, repoEnsurer *repository.Ensurer, repoMgr repository.Manager, directSnapshots map[string][]repository.SnapshotIdentifier, backup *velerov1api.Backup, logger logrus.FieldLogger) []error {
+	return []error{
+		errors.New("fake-delete-1"),
+		errors.New("fake-delete-2"),
+	}
+}
+
+func generateSnapshotData(snapshot *repository.SnapshotIdentifier) (map[string]string, error) {
+	if snapshot == nil {
+		return nil, nil
+	}
+
+	b, err := json.Marshal(snapshot)
+	if err != nil {
+		return nil, err
+	}
+
+	data := make(map[string]string)
+	if err := json.Unmarshal(b, &data); err != nil {
+		return nil, err
+	}
+
+	return data, nil
+}
+
+func TestDeleteMovedSnapshots(t *testing.T) {
+	tests := []struct {
+		name               string
+		repoMgr            repository.Manager
+		batchDeleteSucceed bool
+		backupName         string
+		snapshots          []*repository.SnapshotIdentifier
+		expected           []string
+	}{
+		{
+			name: "repoMgr is nil",
+		},
+		{
+			name:    "no cm",
+			repoMgr: repomocks.NewManager(t),
+		},
+		{
+			name:       "bad cm info",
+			repoMgr:    repomocks.NewManager(t),
+			backupName: "backup-01",
+			snapshots:  []*repository.SnapshotIdentifier{nil},
+			expected:   []string{"no snapshot info in config"},
+		},
+		{
+			name:       "invalid snapshots",
+			repoMgr:    repomocks.NewManager(t),
+			backupName: "backup-01",
+			snapshots: []*repository.SnapshotIdentifier{
+				{
+					RepositoryType:  "repo-1",
+					VolumeNamespace: "ns-1",
+				},
+				{
+					SnapshotID:      "snapshot-1",
+					VolumeNamespace: "ns-1",
+				},
+				{
+					SnapshotID:     "snapshot-1",
+					RepositoryType: "repo-1",
+				},
+			},
+			batchDeleteSucceed: true,
+			expected: []string{
+				"invalid snapshot, ID , namespace ns-1, repository repo-1",
+				"invalid snapshot, ID snapshot-1, namespace ns-1, repository ",
+				"invalid snapshot, ID snapshot-1, namespace , repository repo-1",
+			},
+		},
+		{
+			name:       "batch delete succeed",
+			repoMgr:    repomocks.NewManager(t),
+			backupName: "backup-01",
+			snapshots: []*repository.SnapshotIdentifier{
+
+				{
+					SnapshotID:      "snapshot-1",
+					RepositoryType:  "repo-1",
+					VolumeNamespace: "ns-1",
+				},
+			},
+			batchDeleteSucceed: true,
+			expected:           []string{},
+		},
+		{
+			name:       "batch delete fail",
+			repoMgr:    repomocks.NewManager(t),
+			backupName: "backup-01",
+			snapshots: []*repository.SnapshotIdentifier{
+				{
+					RepositoryType:  "repo-1",
+					VolumeNamespace: "ns-1",
+				},
+				{
+					SnapshotID:      "snapshot-1",
+					RepositoryType:  "repo-1",
+					VolumeNamespace: "ns-1",
+				},
+			},
+			expected: []string{"invalid snapshot, ID , namespace ns-1, repository repo-1", "fake-delete-1", "fake-delete-2"},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			objs := []runtime.Object{}
+			for i, snapshot := range test.snapshots {
+				snapshotData, err := generateSnapshotData(snapshot)
+				require.NoError(t, err)
+
+				cm := corev1api.ConfigMap{
+					TypeMeta: metav1.TypeMeta{
+						APIVersion: corev1api.SchemeGroupVersion.String(),
+						Kind:       "ConfigMap",
+					},
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: "velero",
+						Name:      fmt.Sprintf("du-info-%d", i),
+						Labels: map[string]string{
+							velerov1api.BackupNameLabel:             test.backupName,
+							velerov1api.DataUploadSnapshotInfoLabel: "true",
+						},
+					},
+					Data: snapshotData,
 				}
 
+				objs = append(objs, &cm)
 			}
 
-			sort.Slice(test.expected, less(test.expected))
-			sort.Slice(res, less(res))
+			veleroBackup := &velerov1api.Backup{}
+			controller := NewBackupDeletionReconciler(
+				velerotest.NewLogger(),
+				velerotest.NewFakeControllerRuntimeClient(t, objs...),
+				NewBackupTracker(),
+				test.repoMgr,
+				metrics.NewServerMetrics(),
+				nil, // discovery helper
+				func(logrus.FieldLogger) clientmgmt.Manager { return pluginManager },
+				NewFakeSingleObjectBackupStoreGetter(backupStore),
+				velerotest.NewFakeCredentialsFileStore("", nil),
+				nil,
+			)
 
-			assert.Equal(t, test.expected, res)
+			veleroBackup.Name = test.backupName
+
+			if test.batchDeleteSucceed {
+				batchDeleteSnapshotFunc = batchDeleteSucceed
+			} else {
+				batchDeleteSnapshotFunc = batchDeleteFail
+			}
+
+			errs := controller.deleteMovedSnapshots(context.Background(), veleroBackup)
+			if test.expected == nil {
+				assert.Nil(t, errs)
+			} else {
+				assert.Equal(t, len(test.expected), len(errs))
+				for i := range test.expected {
+					assert.EqualError(t, errs[i], test.expected[i])
+				}
+			}
 		})
 	}
 }

pkg/controller/data_upload_controller.go

@@ -343,7 +343,7 @@ func (r *DataUploadReconciler) runCancelableDataUpload(ctx context.Context, fsBa
 		velerov1api.AsyncOperationIDLabel: du.Labels[velerov1api.AsyncOperationIDLabel],
 	}
 
-	if err := fsBackup.StartBackup(path, fmt.Sprintf("%s/%s", du.Spec.SourceNamespace, du.Spec.SourcePVC), "", false, tags, du.Spec.DataMoverConfig); err != nil {
+	if err := fsBackup.StartBackup(path, datamover.GetRealSource(du.Spec.SourceNamespace, du.Spec.SourcePVC), "", false, tags, du.Spec.DataMoverConfig); err != nil {
 		return r.errorOut(ctx, du, err, "error starting data path backup", log)
 	}
 

pkg/controller/pod_volume_restore_controller_test.go

@@ -492,7 +492,7 @@ func TestFindVolumeRestoresForPod(t *testing.T) {
 		logger: logrus.New(),
 	}
 	requests := reconciler.findVolumeRestoresForPod(context.Background(), pod)
-	assert.Len(t, requests, 0)
+	assert.Empty(t, requests)
 
 	// contain one matching PVR
 	reconciler.Client = clientBuilder.WithLists(&velerov1api.PodVolumeRestoreList{

pkg/controller/restore_controller_test.go

@@ -592,7 +592,7 @@ func TestRestoreReconcile(t *testing.T) {
 				return
 			}
 			if !test.addValidFinalizer {
-				assert.Equal(t, 1, len(restorer.Calls))
+				assert.Len(t, restorer.Calls, 1)
 			}
 
 			// validate Patch call 2 (setting phase)

pkg/controller/restore_finalizer_controller_test.go

@@ -442,7 +442,7 @@ func TestPatchDynamicPVWithVolumeInfo(t *testing.T) {
 
 		errs := ctx.patchDynamicPVWithVolumeInfo()
 		if tc.expectedErrNum > 0 {
-			assert.Equal(t, tc.expectedErrNum, len(errs.Namespaces))
+			assert.Len(t, errs.Namespaces, tc.expectedErrNum)
 		}
 
 		for pvName, expectedPVInfo := range tc.expectedPatch {

pkg/controller/schedule_controller_test.go

@@ -219,16 +219,16 @@ func TestReconcileOfSchedule(t *testing.T) {
 			// new backup shouldn't be submitted.
 			if test.backup != nil &&
 				(test.backup.Status.Phase == velerov1.BackupPhaseNew || test.backup.Status.Phase == velerov1.BackupPhaseInProgress) {
-				assert.Equal(t, 1, len(backups.Items))
+				assert.Len(t, backups.Items, 1)
 				require.Nil(t, client.Delete(ctx, test.backup))
 			}
 
 			require.Nil(t, client.List(ctx, backups))
 
 			if test.expectedBackupCreate == nil {
-				assert.Equal(t, 0, len(backups.Items))
+				assert.Empty(t, backups.Items)
 			} else {
-				assert.Equal(t, 1, len(backups.Items))
+				assert.Len(t, backups.Items, 1)
 			}
 		})
 	}

pkg/datamover/dataupload_delete_action.go

@@ -56,7 +56,9 @@ func genConfigmap(bak *velerov1.Backup, du velerov2alpha1.DataUpload) *corev1api
 		VolumeNamespace:       du.Spec.SourceNamespace,
 		BackupStorageLocation: bak.Spec.StorageLocation,
 		SnapshotID:            du.Status.SnapshotID,
-		RepositoryType:        GetUploaderType(du.Spec.DataMover),
+		RepositoryType:        velerov1.BackupRepositoryTypeKopia,
+		UploaderType:          GetUploaderType(du.Spec.DataMover),
+		Source:                GetRealSource(du.Spec.SourceNamespace, du.Spec.SourcePVC),
 	}
 	b, err := json.Marshal(snapshot)
 	if err != nil {

pkg/datamover/util.go

@@ -16,6 +16,8 @@ limitations under the License.
 
 package datamover
 
+import "fmt"
+
 func GetUploaderType(dataMover string) string {
 	if dataMover == "" || dataMover == "velero" {
 		return "kopia"
@@ -27,3 +29,7 @@ func GetUploaderType(dataMover string) string {
 func IsBuiltInUploader(dataMover string) bool {
 	return dataMover == "" || dataMover == "velero"
 }
+
+func GetRealSource(sourceNamespace string, pvcName string) string {
+	return fmt.Sprintf("%s/%s", sourceNamespace, pvcName)
+}

pkg/datapath/manager_test.go

@@ -42,10 +42,10 @@ func TestManager(t *testing.T) {
 	assert.Equal(t, async_job_1, ret)
 
 	m.RemoveAsyncBR("job-0")
-	assert.Equal(t, 2, len(m.tracker))
+	assert.Len(t, m.tracker, 2)
 
 	m.RemoveAsyncBR("job-1")
-	assert.Equal(t, 1, len(m.tracker))
+	assert.Len(t, m.tracker, 1)
 
 	ret = m.GetAsyncBR("job-1")
 	assert.Equal(t, nil, ret)

pkg/install/daemonset_test.go

@@ -34,8 +34,8 @@ func TestDaemonSet(t *testing.T) {
 	assert.Equal(t, corev1.PullIfNotPresent, ds.Spec.Template.Spec.Containers[0].ImagePullPolicy)
 
 	ds = DaemonSet("velero", WithSecret(true))
-	assert.Equal(t, 7, len(ds.Spec.Template.Spec.Containers[0].Env))
-	assert.Equal(t, 4, len(ds.Spec.Template.Spec.Volumes))
+	assert.Len(t, ds.Spec.Template.Spec.Containers[0].Env, 7)
+	assert.Len(t, ds.Spec.Template.Spec.Volumes, 4)
 
 	ds = DaemonSet("velero", WithFeatures([]string{"foo,bar,baz"}))
 	assert.Len(t, ds.Spec.Template.Spec.Containers[0].Args, 3)

pkg/install/deployment_test.go

@@ -43,8 +43,8 @@ func TestDeployment(t *testing.T) {
 	assert.Equal(t, corev1.PullIfNotPresent, deploy.Spec.Template.Spec.Containers[0].ImagePullPolicy)
 
 	deploy = Deployment("velero", WithSecret(true))
-	assert.Equal(t, 7, len(deploy.Spec.Template.Spec.Containers[0].Env))
-	assert.Equal(t, 3, len(deploy.Spec.Template.Spec.Volumes))
+	assert.Len(t, deploy.Spec.Template.Spec.Containers[0].Env, 7)
+	assert.Len(t, deploy.Spec.Template.Spec.Volumes, 3)
 
 	deploy = Deployment("velero", WithDefaultRepoMaintenanceFrequency(24*time.Hour))
 	assert.Len(t, deploy.Spec.Template.Spec.Containers[0].Args, 2)

pkg/metrics/metrics.go

@@ -69,7 +69,7 @@ const (
 	// data mover metrics
 	DataUploadSuccessTotal   = "data_upload_success_total"
 	DataUploadFailureTotal   = "data_upload_failure_total"
-	DataUploadCancelTotal    = "data_upload_cancel_total" //nolint:gosec // Not a hard code secret.
+	DataUploadCancelTotal    = "data_upload_cancel_total"
 	DataDownloadSuccessTotal = "data_download_success_total"
 	DataDownloadFailureTotal = "data_download_failure_total"
 	DataDownloadCancelTotal  = "data_download_cancel_total"

pkg/podvolume/backupper_test.go

@@ -642,22 +642,22 @@ func TestPVCBackupSummary(t *testing.T) {
 
 	// it won't be added if the volme is not in the pvc map.
 	pbs.addSkipped("vol-3", "whatever reason")
-	assert.Equal(t, 0, len(pbs.Skipped))
+	assert.Empty(t, pbs.Skipped)
 	pbs.addBackedup("vol-3")
-	assert.Equal(t, 0, len(pbs.Backedup))
+	assert.Empty(t, pbs.Backedup)
 
 	// only can be added as skipped when it's not in backedup set
 	pbs.addBackedup("vol-1")
-	assert.Equal(t, 1, len(pbs.Backedup))
+	assert.Len(t, pbs.Backedup, 1)
 	assert.Equal(t, "pvc-1", pbs.Backedup["vol-1"].Name)
 	pbs.addSkipped("vol-1", "whatever reason")
-	assert.Equal(t, 0, len(pbs.Skipped))
+	assert.Empty(t, pbs.Skipped)
 	pbs.addSkipped("vol-2", "vol-2 has to be skipped")
-	assert.Equal(t, 1, len(pbs.Skipped))
+	assert.Len(t, pbs.Skipped, 1)
 	assert.Equal(t, "pvc-2", pbs.Skipped["vol-2"].PVC.Name)
 
 	// adding a vol as backedup removes it from skipped set
 	pbs.addBackedup("vol-2")
-	assert.Equal(t, 0, len(pbs.Skipped))
-	assert.Equal(t, 2, len(pbs.Backedup))
+	assert.Empty(t, pbs.Skipped)
+	assert.Len(t, pbs.Backedup, 2)
 }

pkg/podvolume/util.go

@@ -122,19 +122,30 @@ func getVolumeBackupInfoForPod(podVolumeBackups []*velerov1api.PodVolumeBackup,
 }
 
 // GetSnapshotIdentifier returns the snapshots represented by SnapshotIdentifier for the given PVBs
-func GetSnapshotIdentifier(podVolumeBackups *velerov1api.PodVolumeBackupList) []repository.SnapshotIdentifier {
-	var res []repository.SnapshotIdentifier
+func GetSnapshotIdentifier(podVolumeBackups *velerov1api.PodVolumeBackupList) map[string][]repository.SnapshotIdentifier {
+	res := map[string][]repository.SnapshotIdentifier{}
 	for _, item := range podVolumeBackups.Items {
 		if item.Status.SnapshotID == "" {
 			continue
 		}
 
-		res = append(res, repository.SnapshotIdentifier{
+		if res[item.Spec.Pod.Namespace] == nil {
+			res[item.Spec.Pod.Namespace] = []repository.SnapshotIdentifier{}
+		}
+
+		snapshots := res[item.Spec.Pod.Namespace]
+
+		snapshots = append(snapshots, repository.SnapshotIdentifier{
 			VolumeNamespace:       item.Spec.Pod.Namespace,
 			BackupStorageLocation: item.Spec.BackupStorageLocation,
 			SnapshotID:            item.Status.SnapshotID,
 			RepositoryType:        getRepositoryType(item.Spec.UploaderType),
+			UploaderType:          item.Spec.UploaderType,
+			Source:                item.Status.Path,
+			RepoIdentifier:        item.Spec.RepoIdentifier,
 		})
+
+		res[item.Spec.Pod.Namespace] = snapshots
 	}
 
 	return res

pkg/repository/manager.go

@@ -48,6 +48,16 @@ type SnapshotIdentifier struct {
 	// RepositoryType is the type of the repository where the
 	// snapshot is stored
 	RepositoryType string `json:"repositoryType"`
+
+	// Source is the source of the data saved in the repo by the snapshot
+	Source string `json:"source"`
+
+	// UploaderType is the type of uploader which saved the snapshot data
+	UploaderType string `json:"uploaderType"`
+
+	// RepoIdentifier is the identifier of the repository where the
+	// snapshot is stored
+	RepoIdentifier string `json:"repoIdentifier"`
 }
 
 // Manager manages backup repositories.
@@ -71,7 +81,12 @@ type Manager interface {
 
 	// Forget removes a snapshot from the list of
 	// available snapshots in a repo.
-	Forget(context.Context, SnapshotIdentifier) error
+	Forget(context.Context, *velerov1api.BackupRepository, string) error
+
+	// BatchForget removes a list of snapshots from the list of
+	// available snapshots in a repo.
+	BatchForget(context.Context, *velerov1api.BackupRepository, []string) []error
+
 	// DefaultMaintenanceFrequency returns the default maintenance frequency from the specific repo
 	DefaultMaintenanceFrequency(repo *velerov1api.BackupRepository) (time.Duration, error)
 }
@@ -195,12 +210,7 @@ func (m *manager) UnlockRepo(repo *velerov1api.BackupRepository) error {
 	return prd.EnsureUnlockRepo(context.Background(), param)
 }
 
-func (m *manager) Forget(ctx context.Context, snapshot SnapshotIdentifier) error {
-	repo, err := m.repoEnsurer.EnsureRepo(ctx, m.namespace, snapshot.VolumeNamespace, snapshot.BackupStorageLocation, snapshot.RepositoryType)
-	if err != nil {
-		return err
-	}
-
+func (m *manager) Forget(ctx context.Context, repo *velerov1api.BackupRepository, snapshot string) error {
 	m.repoLocker.LockExclusive(repo.Name)
 	defer m.repoLocker.UnlockExclusive(repo.Name)
 
@@ -217,7 +227,27 @@ func (m *manager) Forget(ctx context.Context, snapshot SnapshotIdentifier) error
 		return errors.WithStack(err)
 	}
 
-	return prd.Forget(context.Background(), snapshot.SnapshotID, param)
+	return prd.Forget(context.Background(), snapshot, param)
+}
+
+func (m *manager) BatchForget(ctx context.Context, repo *velerov1api.BackupRepository, snapshots []string) []error {
+	m.repoLocker.LockExclusive(repo.Name)
+	defer m.repoLocker.UnlockExclusive(repo.Name)
+
+	prd, err := m.getRepositoryProvider(repo)
+	if err != nil {
+		return []error{errors.WithStack(err)}
+	}
+	param, err := m.assembleRepoParam(repo)
+	if err != nil {
+		return []error{errors.WithStack(err)}
+	}
+
+	if err := prd.BoostRepoConnect(context.Background(), param); err != nil {
+		return []error{errors.WithStack(err)}
+	}
+
+	return prd.BatchForget(context.Background(), snapshots, param)
 }
 
 func (m *manager) DefaultMaintenanceFrequency(repo *velerov1api.BackupRepository) (time.Duration, error) {

pkg/repository/mocks/Manager.go

@@ -1,4 +1,4 @@
-// Code generated by mockery v2.14.0. DO NOT EDIT.
+// Code generated by mockery v2.39.1. DO NOT EDIT.
 
 package mocks
 
@@ -6,7 +6,6 @@ import (
 	context "context"
 
 	mock "github.com/stretchr/testify/mock"
-	repository "github.com/vmware-tanzu/velero/pkg/repository"
 
 	time "time"
 
@@ -18,10 +17,34 @@ type Manager struct {
 	mock.Mock
 }
 
+// BatchForget provides a mock function with given fields: _a0, _a1, _a2
+func (_m *Manager) BatchForget(_a0 context.Context, _a1 *v1.BackupRepository, _a2 []string) []error {
+	ret := _m.Called(_a0, _a1, _a2)
+
+	if len(ret) == 0 {
+		panic("no return value specified for BatchForget")
+	}
+
+	var r0 []error
+	if rf, ok := ret.Get(0).(func(context.Context, *v1.BackupRepository, []string) []error); ok {
+		r0 = rf(_a0, _a1, _a2)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]error)
+		}
+	}
+
+	return r0
+}
+
 // ConnectToRepo provides a mock function with given fields: repo
 func (_m *Manager) ConnectToRepo(repo *v1.BackupRepository) error {
 	ret := _m.Called(repo)
 
+	if len(ret) == 0 {
+		panic("no return value specified for ConnectToRepo")
+	}
+
 	var r0 error
 	if rf, ok := ret.Get(0).(func(*v1.BackupRepository) error); ok {
 		r0 = rf(repo)
@@ -36,14 +59,21 @@ func (_m *Manager) ConnectToRepo(repo *v1.BackupRepository) error {
 func (_m *Manager) DefaultMaintenanceFrequency(repo *v1.BackupRepository) (time.Duration, error) {
 	ret := _m.Called(repo)
 
+	if len(ret) == 0 {
+		panic("no return value specified for DefaultMaintenanceFrequency")
+	}
+
 	var r0 time.Duration
+	var r1 error
+	if rf, ok := ret.Get(0).(func(*v1.BackupRepository) (time.Duration, error)); ok {
+		return rf(repo)
+	}
 	if rf, ok := ret.Get(0).(func(*v1.BackupRepository) time.Duration); ok {
 		r0 = rf(repo)
 	} else {
 		r0 = ret.Get(0).(time.Duration)
 	}
 
-	var r1 error
 	if rf, ok := ret.Get(1).(func(*v1.BackupRepository) error); ok {
 		r1 = rf(repo)
 	} else {
@@ -53,13 +83,17 @@ func (_m *Manager) DefaultMaintenanceFrequency(repo *v1.BackupRepository) (time.
 	return r0, r1
 }
 
-// Forget provides a mock function with given fields: _a0, _a1
-func (_m *Manager) Forget(_a0 context.Context, _a1 repository.SnapshotIdentifier) error {
-	ret := _m.Called(_a0, _a1)
+// Forget provides a mock function with given fields: _a0, _a1, _a2
+func (_m *Manager) Forget(_a0 context.Context, _a1 *v1.BackupRepository, _a2 string) error {
+	ret := _m.Called(_a0, _a1, _a2)
+
+	if len(ret) == 0 {
+		panic("no return value specified for Forget")
+	}
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, repository.SnapshotIdentifier) error); ok {
-		r0 = rf(_a0, _a1)
+	if rf, ok := ret.Get(0).(func(context.Context, *v1.BackupRepository, string) error); ok {
+		r0 = rf(_a0, _a1, _a2)
 	} else {
 		r0 = ret.Error(0)
 	}
@@ -71,6 +105,10 @@ func (_m *Manager) Forget(_a0 context.Context, _a1 repository.SnapshotIdentifier
 func (_m *Manager) InitRepo(repo *v1.BackupRepository) error {
 	ret := _m.Called(repo)
 
+	if len(ret) == 0 {
+		panic("no return value specified for InitRepo")
+	}
+
 	var r0 error
 	if rf, ok := ret.Get(0).(func(*v1.BackupRepository) error); ok {
 		r0 = rf(repo)
@@ -85,6 +123,10 @@ func (_m *Manager) InitRepo(repo *v1.BackupRepository) error {
 func (_m *Manager) PrepareRepo(repo *v1.BackupRepository) error {
 	ret := _m.Called(repo)
 
+	if len(ret) == 0 {
+		panic("no return value specified for PrepareRepo")
+	}
+
 	var r0 error
 	if rf, ok := ret.Get(0).(func(*v1.BackupRepository) error); ok {
 		r0 = rf(repo)
@@ -99,6 +141,10 @@ func (_m *Manager) PrepareRepo(repo *v1.BackupRepository) error {
 func (_m *Manager) PruneRepo(repo *v1.BackupRepository) error {
 	ret := _m.Called(repo)
 
+	if len(ret) == 0 {
+		panic("no return value specified for PruneRepo")
+	}
+
 	var r0 error
 	if rf, ok := ret.Get(0).(func(*v1.BackupRepository) error); ok {
 		r0 = rf(repo)
@@ -113,6 +159,10 @@ func (_m *Manager) PruneRepo(repo *v1.BackupRepository) error {
 func (_m *Manager) UnlockRepo(repo *v1.BackupRepository) error {
 	ret := _m.Called(repo)
 
+	if len(ret) == 0 {
+		panic("no return value specified for UnlockRepo")
+	}
+
 	var r0 error
 	if rf, ok := ret.Get(0).(func(*v1.BackupRepository) error); ok {
 		r0 = rf(repo)
@@ -123,13 +173,12 @@ func (_m *Manager) UnlockRepo(repo *v1.BackupRepository) error {
 	return r0
 }
 
-type mockConstructorTestingTNewManager interface {
+// NewManager creates a new instance of Manager. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+// The first argument is typically a *testing.T value.
+func NewManager(t interface {
 	mock.TestingT
 	Cleanup(func())
-}
-
-// NewManager creates a new instance of Manager. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
-func NewManager(t mockConstructorTestingTNewManager) *Manager {
+}) *Manager {
 	mock := &Manager{}
 	mock.Mock.Test(t)
 

pkg/repository/provider/provider.go

@@ -57,6 +57,9 @@ type Provider interface {
 	// Forget is to delete a snapshot from the repository
 	Forget(ctx context.Context, snapshotID string, param RepoParam) error
 
+	// BatchForget is to delete a list of snapshots from the repository
+	BatchForget(ctx context.Context, snapshotIDs []string, param RepoParam) []error
+
 	// DefaultMaintenanceFrequency returns the default frequency to run maintenance
 	DefaultMaintenanceFrequency(ctx context.Context, param RepoParam) time.Duration
 }

pkg/repository/provider/restic.go

@@ -78,6 +78,16 @@ func (r *resticRepositoryProvider) Forget(ctx context.Context, snapshotID string
 	return r.svc.Forget(param.BackupLocation, param.BackupRepo, snapshotID)
 }
 
+func (r *resticRepositoryProvider) BatchForget(ctx context.Context, snapshotIDs []string, param RepoParam) []error {
+	errs := []error{}
+	for _, snapshot := range snapshotIDs {
+		err := r.Forget(ctx, snapshot, param)
+		errs = append(errs, err)
+	}
+
+	return errs
+}
+
 func (r *resticRepositoryProvider) DefaultMaintenanceFrequency(ctx context.Context, param RepoParam) time.Duration {
 	return r.svc.DefaultMaintenanceFrequency()
 }

pkg/repository/provider/unified_repo.go

@@ -314,6 +314,56 @@ func (urp *unifiedRepoProvider) Forget(ctx context.Context, snapshotID string, p
 	return nil
 }
 
+func (urp *unifiedRepoProvider) BatchForget(ctx context.Context, snapshotIDs []string, param RepoParam) []error {
+	log := urp.log.WithFields(logrus.Fields{
+		"BSL name":    param.BackupLocation.Name,
+		"repo name":   param.BackupRepo.Name,
+		"repo UID":    param.BackupRepo.UID,
+		"snapshotIDs": snapshotIDs,
+	})
+
+	log.Debug("Start to batch forget snapshot")
+
+	repoOption, err := udmrepo.NewRepoOptions(
+		udmrepo.WithPassword(urp, param),
+		udmrepo.WithConfigFile(urp.workPath, string(param.BackupRepo.UID)),
+		udmrepo.WithDescription(repoOpDescForget),
+	)
+
+	if err != nil {
+		return []error{errors.Wrap(err, "error to get repo options")}
+	}
+
+	bkRepo, err := urp.repoService.Open(ctx, *repoOption)
+	if err != nil {
+		return []error{errors.Wrap(err, "error to open backup repo")}
+	}
+
+	defer func() {
+		c := bkRepo.Close(ctx)
+		if c != nil {
+			log.WithError(c).Error("Failed to close repo")
+		}
+	}()
+
+	errs := []error{}
+	for _, snapshotID := range snapshotIDs {
+		err = bkRepo.DeleteManifest(ctx, udmrepo.ID(snapshotID))
+		if err != nil {
+			errs = append(errs, errors.Wrapf(err, "error to delete manifest %s", snapshotID))
+		}
+	}
+
+	err = bkRepo.Flush(ctx)
+	if err != nil {
+		return []error{errors.Wrap(err, "error to flush repo")}
+	}
+
+	log.Debug("Forget snapshot complete")
+
+	return errs
+}
+
 func (urp *unifiedRepoProvider) DefaultMaintenanceFrequency(ctx context.Context, param RepoParam) time.Duration {
 	return urp.repoService.DefaultMaintenanceFrequency()
 }

pkg/repository/provider/unified_repo_test.go

@@ -857,6 +857,161 @@ func TestForget(t *testing.T) {
 	}
 }
 
+func TestBatchForget(t *testing.T) {
+	var backupRepo *reposervicenmocks.BackupRepo
+
+	testCases := []struct {
+		name            string
+		funcTable       localFuncTable
+		getter          *credmock.SecretStore
+		repoService     *reposervicenmocks.BackupRepoService
+		backupRepo      *reposervicenmocks.BackupRepo
+		retFuncOpen     []interface{}
+		retFuncDelete   interface{}
+		retFuncFlush    interface{}
+		credStoreReturn string
+		credStoreError  error
+		snapshots       []string
+		expectedErr     []string
+	}{
+		{
+			name:        "get repo option fail",
+			expectedErr: []string{"error to get repo options: error to get repo password: invalid credentials interface"},
+		},
+		{
+			name:            "repo open fail",
+			getter:          new(credmock.SecretStore),
+			credStoreReturn: "fake-password",
+			funcTable: localFuncTable{
+				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string) (map[string]string, error) {
+					return map[string]string{}, nil
+				},
+				getStorageCredentials: func(*velerov1api.BackupStorageLocation, velerocredentials.FileStore) (map[string]string, error) {
+					return map[string]string{}, nil
+				},
+			},
+			repoService: new(reposervicenmocks.BackupRepoService),
+			retFuncOpen: []interface{}{
+				func(context.Context, udmrepo.RepoOptions) udmrepo.BackupRepo {
+					return backupRepo
+				},
+
+				func(context.Context, udmrepo.RepoOptions) error {
+					return errors.New("fake-error-2")
+				},
+			},
+			expectedErr: []string{"error to open backup repo: fake-error-2"},
+		},
+		{
+			name:            "delete fail",
+			getter:          new(credmock.SecretStore),
+			credStoreReturn: "fake-password",
+			funcTable: localFuncTable{
+				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string) (map[string]string, error) {
+					return map[string]string{}, nil
+				},
+				getStorageCredentials: func(*velerov1api.BackupStorageLocation, velerocredentials.FileStore) (map[string]string, error) {
+					return map[string]string{}, nil
+				},
+			},
+			repoService: new(reposervicenmocks.BackupRepoService),
+			backupRepo:  new(reposervicenmocks.BackupRepo),
+			retFuncOpen: []interface{}{
+				func(context.Context, udmrepo.RepoOptions) udmrepo.BackupRepo {
+					return backupRepo
+				},
+
+				func(context.Context, udmrepo.RepoOptions) error {
+					return nil
+				},
+			},
+			retFuncDelete: func(context.Context, udmrepo.ID) error {
+				return errors.New("fake-error-3")
+			},
+			snapshots:   []string{"snapshot-1", "snapshot-2"},
+			expectedErr: []string{"error to delete manifest snapshot-1: fake-error-3", "error to delete manifest snapshot-2: fake-error-3"},
+		},
+		{
+			name:            "flush fail",
+			getter:          new(credmock.SecretStore),
+			credStoreReturn: "fake-password",
+			funcTable: localFuncTable{
+				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string) (map[string]string, error) {
+					return map[string]string{}, nil
+				},
+				getStorageCredentials: func(*velerov1api.BackupStorageLocation, velerocredentials.FileStore) (map[string]string, error) {
+					return map[string]string{}, nil
+				},
+			},
+			repoService: new(reposervicenmocks.BackupRepoService),
+			backupRepo:  new(reposervicenmocks.BackupRepo),
+			retFuncOpen: []interface{}{
+				func(context.Context, udmrepo.RepoOptions) udmrepo.BackupRepo {
+					return backupRepo
+				},
+
+				func(context.Context, udmrepo.RepoOptions) error {
+					return nil
+				},
+			},
+			retFuncDelete: func(context.Context, udmrepo.ID) error {
+				return nil
+			},
+			retFuncFlush: func(context.Context) error {
+				return errors.New("fake-error-4")
+			},
+			expectedErr: []string{"error to flush repo: fake-error-4"},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			funcTable = tc.funcTable
+
+			var secretStore velerocredentials.SecretStore
+			if tc.getter != nil {
+				tc.getter.On("Get", mock.Anything, mock.Anything).Return(tc.credStoreReturn, tc.credStoreError)
+				secretStore = tc.getter
+			}
+
+			urp := unifiedRepoProvider{
+				credentialGetter: velerocredentials.CredentialGetter{
+					FromSecret: secretStore,
+				},
+				repoService: tc.repoService,
+				log:         velerotest.NewLogger(),
+			}
+
+			backupRepo = tc.backupRepo
+
+			if tc.repoService != nil {
+				tc.repoService.On("Open", mock.Anything, mock.Anything).Return(tc.retFuncOpen[0], tc.retFuncOpen[1])
+			}
+
+			if tc.backupRepo != nil {
+				backupRepo.On("DeleteManifest", mock.Anything, mock.Anything).Return(tc.retFuncDelete)
+				backupRepo.On("Flush", mock.Anything).Return(tc.retFuncFlush)
+				backupRepo.On("Close", mock.Anything).Return(nil)
+			}
+
+			errs := urp.BatchForget(context.Background(), tc.snapshots, RepoParam{
+				BackupLocation: &velerov1api.BackupStorageLocation{},
+				BackupRepo:     &velerov1api.BackupRepository{},
+			})
+
+			if tc.expectedErr == nil {
+				assert.Equal(t, 0, len(errs))
+			} else {
+				assert.Equal(t, len(tc.expectedErr), len(errs))
+
+				for i := range tc.expectedErr {
+					assert.EqualError(t, errs[i], tc.expectedErr[i])
+				}
+			}
+		})
+	}
+}
+
 func TestInitRepo(t *testing.T) {
 	testCases := []struct {
 		name            string

pkg/repository/udmrepo/kopialib/backend/utils.go

@@ -30,7 +30,7 @@ func mustHaveString(key string, flags map[string]string) (string, error) {
 	if value, exist := flags[key]; exist {
 		return value, nil
 	}
-	return "", errors.New("key " + key + " not found")
+	return "", errors.Errorf("key %s not found", key)
 }
 
 func optionalHaveString(key string, flags map[string]string) string {

pkg/repository/udmrepo/kopialib/lib_repo.go

@@ -72,6 +72,10 @@ type kopiaObjectWriter struct {
 	rawWriter object.Writer
 }
 
+type openOptions struct {
+	allowIndexWriteOnLoad bool
+}
+
 const (
 	defaultLogInterval             = time.Second * 10
 	defaultMaintainCheckPeriod     = time.Hour
@@ -115,7 +119,7 @@ func (ks *kopiaRepoService) Open(ctx context.Context, repoOption udmrepo.RepoOpt
 
 	repoCtx := kopia.SetupKopiaLog(ctx, ks.logger)
 
-	r, err := openKopiaRepo(repoCtx, repoConfig, repoOption.RepoPassword)
+	r, err := openKopiaRepo(repoCtx, repoConfig, repoOption.RepoPassword, nil)
 	if err != nil {
 		return nil, err
 	}
@@ -158,11 +162,15 @@ func (ks *kopiaRepoService) Maintain(ctx context.Context, repoOption udmrepo.Rep
 
 	repoCtx := kopia.SetupKopiaLog(ctx, ks.logger)
 
-	r, err := openKopiaRepo(repoCtx, repoConfig, repoOption.RepoPassword)
+	ks.logger.Info("Start to open repo for maintenance, allow index write on load")
+
+	r, err := openKopiaRepo(repoCtx, repoConfig, repoOption.RepoPassword, &openOptions{allowIndexWriteOnLoad: true})
 	if err != nil {
 		return err
 	}
 
+	ks.logger.Info("Succeeded to open repo for maintenance")
+
 	defer func() {
 		c := r.Close(repoCtx)
 		if c != nil {
@@ -546,8 +554,13 @@ func (lt *logThrottle) shouldLog() bool {
 	return false
 }
 
-func openKopiaRepo(ctx context.Context, configFile string, password string) (repo.Repository, error) {
-	r, err := kopiaRepoOpen(ctx, configFile, password, &repo.Options{})
+func openKopiaRepo(ctx context.Context, configFile string, password string, options *openOptions) (repo.Repository, error) {
+	allowIndexWriteOnLoad := false
+	if options != nil {
+		allowIndexWriteOnLoad = options.allowIndexWriteOnLoad
+	}
+
+	r, err := kopiaRepoOpen(ctx, configFile, password, &repo.Options{AllowWriteOnIndexLoad: allowIndexWriteOnLoad})
 	if os.IsNotExist(err) {
 		return nil, errors.Wrap(err, "error to open repo, repo doesn't exist")
 	}
@@ -560,7 +573,7 @@ func openKopiaRepo(ctx context.Context, configFile string, password string) (rep
 }
 
 func writeInitParameters(ctx context.Context, repoOption udmrepo.RepoOptions, logger logrus.FieldLogger) error {
-	r, err := openKopiaRepo(ctx, repoOption.ConfigFilePath, repoOption.RepoPassword)
+	r, err := openKopiaRepo(ctx, repoOption.ConfigFilePath, repoOption.RepoPassword, nil)
 	if err != nil {
 		return err
 	}

pkg/util/azure/credential.go

@@ -81,7 +81,7 @@ type configCredentialOptions struct {
 	AdditionallyAllowedTenants []string
 }
 
-// newConfigCredential works same as the azidentity.EnvironmentCredential but reads the credentials from a map
+// newConfigCredential works similar as the azidentity.EnvironmentCredential but reads the credentials from a map
 // rather than environment variables. This is required for Velero to run B/R concurrently
 // https://github.com/Azure/azure-sdk-for-go/blob/sdk/azidentity/v1.3.0/sdk/azidentity/environment_credential.go#L80
 func newConfigCredential(creds map[string]string, options configCredentialOptions) (azcore.TokenCredential, error) {
@@ -102,19 +102,24 @@ func newConfigCredential(creds map[string]string, options configCredentialOption
 		})
 	}
 
-	// certificate
-	if certPath := creds[CredentialKeyClientCertificatePath]; certPath != "" {
-		certData, err := os.ReadFile(certPath)
-		if err != nil {
-			return nil, errors.Wrapf(err, "failed to read certificate file %s", certPath)
+	// raw certificate or certificate file
+	if rawCerts, certsPath := []byte(creds[CredentialKeyClientCertificate]), creds[CredentialKeyClientCertificatePath]; len(rawCerts) > 0 || len(certsPath) > 0 {
+		var err error
+		// raw certificate isn't specified while certificate path is specified
+		if len(rawCerts) == 0 {
+			rawCerts, err = os.ReadFile(certsPath)
+			if err != nil {
+				return nil, errors.Wrapf(err, "failed to read certificate file %s", certsPath)
+			}
 		}
+
 		var password []byte
 		if v := creds[CredentialKeyClientCertificatePassword]; v != "" {
 			password = []byte(v)
 		}
-		certs, key, err := azidentity.ParseCertificates(certData, password)
+		certs, key, err := azidentity.ParseCertificates(rawCerts, password)
 		if err != nil {
-			return nil, errors.Wrapf(err, "failed to load certificate from %s", certPath)
+			return nil, errors.Wrap(err, "failed to parse certificate")
 		}
 		o := &azidentity.ClientCertificateCredentialOptions{
 			AdditionallyAllowedTenants: options.AdditionallyAllowedTenants,

pkg/util/azure/util.go

@@ -43,6 +43,7 @@ const (
 	CredentialKeyTenantID                   = "AZURE_TENANT_ID"                     // #nosec
 	CredentialKeyClientID                   = "AZURE_CLIENT_ID"                     // #nosec
 	CredentialKeyClientSecret               = "AZURE_CLIENT_SECRET"                 // #nosec
+	CredentialKeyClientCertificate          = "AZURE_CLIENT_CERTIFICATE"            // #nosec
 	CredentialKeyClientCertificatePath      = "AZURE_CLIENT_CERTIFICATE_PATH"       // #nosec
 	CredentialKeyClientCertificatePassword  = "AZURE_CLIENT_CERTIFICATE_PASSWORD"   // #nosec
 	CredentialKeySendCertChain              = "AZURE_CLIENT_SEND_CERTIFICATE_CHAIN" // #nosec

pkg/util/stringslice/stringslice_test.go

@@ -36,7 +36,7 @@ func TestHas(t *testing.T) {
 func TestExcept(t *testing.T) {
 	items := []string{}
 	except := Except(items, "asdf")
-	assert.Len(t, except, 0)
+	assert.Empty(t, except)
 
 	items = []string{"a", "b", "c"}
 	assert.Equal(t, []string{"b", "c"}, Except(items, "a"))