Merge branch 'main' of https://github.com/qiuming-best/velero into kopia-parallelism

2023-11-22 03:52:20 +00:00 · 2023-11-22 03:52:20 +00:00 · c2d4495efe
parent fea22bbbc9 5c958d820d
commit c2d4495efe
26 changed files with 1495 additions and 191 deletions
--- a/.github/workflows/stale-issues.yml
+++ b/.github/workflows/stale-issues.yml
@ -7,7 +7,7 @@ jobs:
  stale:
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/stale@v3
+      - uses: actions/stale@v6.0.1
        with:
          repo-token: ${{ secrets.GITHUB_TOKEN }}
          stale-issue-message: "This issue is stale because it has been open 60 days with no activity. Remove stale label or comment or this will be closed in 14 days. If a Velero team member has requested log or more information, please provide the output of the shared commands."
--- a/changelogs/unreleased/7100-blackpiglet
+++ b/changelogs/unreleased/7100-blackpiglet
@ -0,0 +1 @@
+Generate VolumeInfo for backup.
--- a/config/crd/v1/bases/velero.io_downloadrequests.yaml
+++ b/config/crd/v1/bases/velero.io_downloadrequests.yaml
@ -53,6 +53,7 @@ spec:
                    - RestoreItemOperations
                    - CSIBackupVolumeSnapshots
                    - CSIBackupVolumeSnapshotContents
+                    - BackupVolumeInfos
                    type: string
                  name:
                    description: Name is the name of the Kubernetes resource with
--- a/config/crd/v1/crds/crds.go
+++ b/config/crd/v1/crds/crds.go
--- a/design/pv_backup_info.md
+++ b/design/pv_backup_info.md
@ -61,7 +61,7 @@ type VolumeInfo struct {
 // CSISnapshotInfo is used for displaying the CSI snapshot status
 type CSISnapshotInfo struct {
    SnapshotHandle  string       // It's the storage provider's snapshot ID for CSI.
-    Size            int64        // The snapshot corresponding volume size. Some of the volume backup methods cannot retrieve the data by current design, for example, the Velero native snapshot.
+    Size            int64        // The snapshot corresponding volume size.

    Driver          string  // The name of the CSI driver.
    VSCName         string // The name of the VolumeSnapshotContent. 
@ -70,7 +70,7 @@ type CSISnapshotInfo struct {
 // SnapshotDataMovementInfo is used for displaying the snapshot data mover status.
 type SnapshotDataMovementInfo struct {
    DataMover        string    // The data mover used by the backup. The valid values are `velero` and ``(equals to `velero`).
-    UploaderType     string    // The type of the uploader that uploads the snapshot data. The valid values are `kopia` and `restic`. It's useful for file-system backup and snapshot data mover.
+    UploaderType     string    // The type of the uploader that uploads the snapshot data. The valid values are `kopia` and `restic`.
    RetainedSnapshot string    // The name or ID of the snapshot associated object(SAO). SAO is used to support local snapshots for the snapshot data mover, e.g. it could be a VolumeSnapshot for CSI snapshot data moign/pv_backup_info.
    SnapshotHandle string  	   // It's the filesystem repository's snapshot ID.
 	
@ -79,7 +79,6 @@ type SnapshotDataMovementInfo struct {
 // VeleroNativeSnapshotInfo is used for displaying the Velero native snapshot status.
 type VeleroNativeSnapshotInfo struct {
    SnapshotHandle      string       // It's the storage provider's snapshot ID for the Velero-native snapshot.
-    Size                int64        // The snapshot corresponding volume size. Some of the volume backup methods cannot retrieve the data by current design, for example, the Velero native snapshot.

    VolumeType string    // The cloud provider snapshot volume type.
    VolumeAZ   string    // The cloud provider snapshot volume's availability zones.
@ -89,11 +88,12 @@ type VeleroNativeSnapshotInfo struct {
 // PodVolumeBackupInfo is used for displaying the PodVolumeBackup snapshot status.
 type PodVolumeBackupInfo struct {
    SnapshotHandle      string       // It's the file-system uploader's snapshot ID for PodVolumeBackup.
-    Size                int64        // The snapshot corresponding volume size. Some of the volume backup methods cannot retrieve the data by current design, for example, the Velero native snapshot.
+    Size                int64        // The snapshot corresponding volume size.

-    UploaderType  string    // The type of the uploader that uploads the data. The valid values are `kopia` and `restic`. It's useful for file-system backup and snapshot data mover.
+    UploaderType  string    // The type of the uploader that uploads the data. The valid values are `kopia` and `restic`.
    VolumeName    string   // The PVC's corresponding volume name used by Pod: https://github.com/kubernetes/kubernetes/blob/e4b74dd12fa8cb63c174091d5536a10b8ec19d34/pkg/apis/core/types.go#L48
-    PodName       string   // The Pod name mounting this PVC. The format should be <namespace-name>/<pod-name>.
+    PodName       string   // The Pod name mounting this PVC.
+    PodNamespace  string   // The Pod namespace.
    NodeName      string   // The PVB-taken k8s node's name.
 }

--- a/internal/resourcemodifiers/json_patch.go
+++ b/internal/resourcemodifiers/json_patch.go
@ -19,30 +19,37 @@ type JSONPatch struct {
 }

 func (p *JSONPatch) ToString() string {
-	if addQuotes(p.Value) {
+	if addQuotes(&p.Value) {
 		return fmt.Sprintf(`{"op": "%s", "from": "%s", "path": "%s", "value": "%s"}`, p.Operation, p.From, p.Path, p.Value)
 	}
 	return fmt.Sprintf(`{"op": "%s", "from": "%s", "path": "%s", "value": %s}`, p.Operation, p.From, p.Path, p.Value)
 }

-func addQuotes(value string) bool {
-	if value == "" {
+func addQuotes(value *string) bool {
+	if *value == "" {
+		return true
+	}
+	// if value is escaped, remove escape and add quotes
+	// this is useful for scenarios where boolean, null and numbers are required to be set as string.
+	if strings.HasPrefix(*value, "\"") && strings.HasSuffix(*value, "\"") {
+		*value = strings.TrimPrefix(*value, "\"")
+		*value = strings.TrimSuffix(*value, "\"")
 		return true
 	}
 	// if value is null, then don't add quotes
-	if value == "null" {
+	if *value == "null" {
 		return false
 	}
 	// if value is a boolean, then don't add quotes
-	if _, err := strconv.ParseBool(value); err == nil {
+	if strings.ToLower(*value) == "true" || strings.ToLower(*value) == "false" {
 		return false
 	}
 	// if value is a json object or array, then don't add quotes.
-	if strings.HasPrefix(value, "{") || strings.HasPrefix(value, "[") {
+	if strings.HasPrefix(*value, "{") || strings.HasPrefix(*value, "[") {
 		return false
 	}
 	// if value is a number, then don't add quotes
-	if _, err := strconv.ParseFloat(value, 64); err == nil {
+	if _, err := strconv.ParseFloat(*value, 64); err == nil {
 		return false
 	}
 	return true
--- a/internal/resourcemodifiers/resource_modifiers_test.go
+++ b/internal/resourcemodifiers/resource_modifiers_test.go
@ -256,6 +256,64 @@ func TestGetResourceModifiersFromConfig(t *testing.T) {
 			},
 		},
 	}
+	cm9 := &v1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-configmap",
+			Namespace: "test-namespace",
+		},
+		Data: map[string]string{
+			"sub.yml": "version: v1\nresourceModifierRules:\n- conditions:\n    groupResource: deployments.apps\n    resourceNameRegex: \"^test-.*$\"\n    namespaces:\n    - bar\n    - foo\n  patches:\n  - operation: replace\n    path: \"/value/bool\"\n    value: \"\\\"true\\\"\"\n\n\n",
+		},
+	}
+
+	rules9 := &ResourceModifiers{
+		Version: "v1",
+		ResourceModifierRules: []ResourceModifierRule{
+			{
+				Conditions: Conditions{
+					GroupResource:     "deployments.apps",
+					ResourceNameRegex: "^test-.*$",
+					Namespaces:        []string{"bar", "foo"},
+				},
+				Patches: []JSONPatch{
+					{
+						Operation: "replace",
+						Path:      "/value/bool",
+						Value:     `"true"`,
+					},
+				},
+			},
+		},
+	}
+	cm10 := &v1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-configmap",
+			Namespace: "test-namespace",
+		},
+		Data: map[string]string{
+			"sub.yml": "version: v1\nresourceModifierRules:\n- conditions:\n    groupResource: deployments.apps\n    resourceNameRegex: \"^test-.*$\"\n    namespaces:\n    - bar\n    - foo\n  patches:\n  - operation: replace\n    path: \"/value/bool\"\n    value: \"true\"\n\n\n",
+		},
+	}
+
+	rules10 := &ResourceModifiers{
+		Version: "v1",
+		ResourceModifierRules: []ResourceModifierRule{
+			{
+				Conditions: Conditions{
+					GroupResource:     "deployments.apps",
+					ResourceNameRegex: "^test-.*$",
+					Namespaces:        []string{"bar", "foo"},
+				},
+				Patches: []JSONPatch{
+					{
+						Operation: "replace",
+						Path:      "/value/bool",
+						Value:     "true",
+					},
+				},
+			},
+		},
+	}

 	type args struct {
 		cm *v1.ConfigMap
@ -338,6 +396,22 @@ func TestGetResourceModifiersFromConfig(t *testing.T) {
 			want:    rules8,
 			wantErr: false,
 		},
+		{
+			name: "bool value as string",
+			args: args{
+				cm: cm9,
+			},
+			want:    rules9,
+			wantErr: false,
+		},
+		{
+			name: "bool value as bool",
+			args: args{
+				cm: cm10,
+			},
+			want:    rules10,
+			wantErr: false,
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@ -480,7 +554,24 @@ func TestResourceModifiers_ApplyResourceModifierRules(t *testing.T) {
 			},
 		},
 	}
-
+	cmTrue := &unstructured.Unstructured{
+		Object: map[string]interface{}{
+			"apiVersion": "v1",
+			"kind":       "ConfigMap",
+			"data": map[string]interface{}{
+				"test": "true",
+			},
+		},
+	}
+	cmFalse := &unstructured.Unstructured{
+		Object: map[string]interface{}{
+			"apiVersion": "v1",
+			"kind":       "ConfigMap",
+			"data": map[string]interface{}{
+				"test": "false",
+			},
+		},
+	}
 	type fields struct {
 		Version               string
 		ResourceModifierRules []ResourceModifierRule
@ -496,6 +587,33 @@ func TestResourceModifiers_ApplyResourceModifierRules(t *testing.T) {
 		wantErr bool
 		wantObj *unstructured.Unstructured
 	}{
+		{
+			name: "configmap true false string",
+			fields: fields{
+				Version: "v1",
+				ResourceModifierRules: []ResourceModifierRule{
+					{
+						Conditions: Conditions{
+							GroupResource:     "configmaps",
+							ResourceNameRegex: ".*",
+						},
+						Patches: []JSONPatch{
+							{
+								Operation: "replace",
+								Path:      "/data/test",
+								Value:     `"false"`,
+							},
+						},
+					},
+				},
+			},
+			args: args{
+				obj:           cmTrue.DeepCopy(),
+				groupResource: "configmaps",
+			},
+			wantErr: false,
+			wantObj: cmFalse.DeepCopy(),
+		},
 		{
 			name: "Invalid Regex throws error",
 			fields: fields{
--- a/pkg/apis/velero/v1/download_request_types.go
+++ b/pkg/apis/velero/v1/download_request_types.go
@ -25,7 +25,7 @@ type DownloadRequestSpec struct {
 }

 // DownloadTargetKind represents what type of file to download.
-// +kubebuilder:validation:Enum=BackupLog;BackupContents;BackupVolumeSnapshots;BackupItemOperations;BackupResourceList;BackupResults;RestoreLog;RestoreResults;RestoreResourceList;RestoreItemOperations;CSIBackupVolumeSnapshots;CSIBackupVolumeSnapshotContents
+// +kubebuilder:validation:Enum=BackupLog;BackupContents;BackupVolumeSnapshots;BackupItemOperations;BackupResourceList;BackupResults;RestoreLog;RestoreResults;RestoreResourceList;RestoreItemOperations;CSIBackupVolumeSnapshots;CSIBackupVolumeSnapshotContents;BackupVolumeInfos
 type DownloadTargetKind string

 const (
@ -41,6 +41,7 @@ const (
 	DownloadTargetKindRestoreItemOperations           DownloadTargetKind = "RestoreItemOperations"
 	DownloadTargetKindCSIBackupVolumeSnapshots        DownloadTargetKind = "CSIBackupVolumeSnapshots"
 	DownloadTargetKindCSIBackupVolumeSnapshotContents DownloadTargetKind = "CSIBackupVolumeSnapshotContents"
+	DownloadTargetKindBackupVolumeInfos               DownloadTargetKind = "BackupVolumeInfos"
 )

 // DownloadTarget is the specification for what kind of file to download, and the name of the
--- a/pkg/backup/backup_test.go
+++ b/pkg/backup/backup_test.go
@ -71,6 +71,7 @@ func TestBackedUpItemsMatchesTarballContents(t *testing.T) {
 	req := &Request{
 		Backup:           defaultBackup().Result(),
 		SkippedPVTracker: NewSkipPVTracker(),
+		PVMap:            map[string]PvcPvInfo{},
 	}
 	backupFile := bytes.NewBuffer([]byte{})

@ -84,8 +85,8 @@ func TestBackedUpItemsMatchesTarballContents(t *testing.T) {
 			builder.ForDeployment("zoo", "raz").Result(),
 		),
 		test.PVs(
-			builder.ForPersistentVolume("bar").Result(),
-			builder.ForPersistentVolume("baz").Result(),
+			builder.ForPersistentVolume("bar").ClaimRef("foo", "pvc1").Result(),
+			builder.ForPersistentVolume("baz").ClaimRef("bar", "pvc2").Result(),
 		),
 	}
 	for _, resource := range apiResources {
--- a/pkg/backup/item_backupper.go
+++ b/pkg/backup/item_backupper.go
@ -250,6 +250,10 @@ func (ib *itemBackupper) backupItemInternal(logger logrus.FieldLogger, obj runti
 	namespace = metadata.GetNamespace()

 	if groupResource == kuberesource.PersistentVolumes {
+		if err := ib.addVolumeInfo(obj, log); err != nil {
+			backupErrs = append(backupErrs, err)
+		}
+
 		if err := ib.takePVSnapshot(obj, log); err != nil {
 			backupErrs = append(backupErrs, err)
 		}
@ -685,6 +689,39 @@ func (ib *itemBackupper) unTrackSkippedPV(obj runtime.Unstructured, groupResourc
 	}
 }

+func (ib *itemBackupper) addVolumeInfo(obj runtime.Unstructured, log logrus.FieldLogger) error {
+	pv := new(corev1api.PersistentVolume)
+	err := runtime.DefaultUnstructuredConverter.FromUnstructured(obj.UnstructuredContent(), pv)
+	if err != nil {
+		log.WithError(err).Warnf("Fail to convert PV")
+		return err
+	}
+
+	if ib.backupRequest.PVMap == nil {
+		ib.backupRequest.PVMap = make(map[string]PvcPvInfo)
+	}
+
+	pvcName := ""
+	pvcNamespace := ""
+	if pv.Spec.ClaimRef != nil {
+		pvcName = pv.Spec.ClaimRef.Name
+		pvcNamespace = pv.Spec.ClaimRef.Namespace
+
+		ib.backupRequest.PVMap[pvcNamespace+"/"+pvcName] = PvcPvInfo{
+			PVCName:      pvcName,
+			PVCNamespace: pvcNamespace,
+			PV:           *pv,
+		}
+	}
+
+	ib.backupRequest.PVMap[pv.Name] = PvcPvInfo{
+		PVCName:      pvcName,
+		PVCNamespace: pvcNamespace,
+		PV:           *pv,
+	}
+	return nil
+}
+
 // convert the input object to PV/PVC and get the PV name
 func getPVName(obj runtime.Unstructured, groupResource schema.GroupResource) (string, error) {
 	if groupResource == kuberesource.PersistentVolumes {
--- a/pkg/backup/item_backupper_test.go
+++ b/pkg/backup/item_backupper_test.go
@ -19,10 +19,12 @@ package backup
 import (
 	"testing"

+	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/require"
 	"k8s.io/apimachinery/pkg/runtime/schema"

 	"github.com/vmware-tanzu/velero/pkg/kuberesource"
+	"github.com/vmware-tanzu/velero/pkg/volume"

 	"github.com/stretchr/testify/assert"
 	corev1api "k8s.io/api/core/v1"
@ -237,3 +239,55 @@ func TestRandom(t *testing.T) {
 	err2 := runtime.DefaultUnstructuredConverter.FromUnstructured(o, pvc)
 	t.Logf("err1: %v, err2: %v", err1, err2)
 }
+
+func TestAddVolumeInfo(t *testing.T) {
+	tests := []struct {
+		name               string
+		pv                 *corev1api.PersistentVolume
+		expectedVolumeInfo map[string]PvcPvInfo
+	}{
+		{
+			name: "PV has ClaimRef",
+			pv:   builder.ForPersistentVolume("testPV").ClaimRef("testNS", "testPVC").Result(),
+			expectedVolumeInfo: map[string]PvcPvInfo{
+				"testPV": {
+					PVCName:      "testPVC",
+					PVCNamespace: "testNS",
+					PV:           *builder.ForPersistentVolume("testPV").ClaimRef("testNS", "testPVC").Result(),
+				},
+				"testNS/testPVC": {
+					PVCName:      "testPVC",
+					PVCNamespace: "testNS",
+					PV:           *builder.ForPersistentVolume("testPV").ClaimRef("testNS", "testPVC").Result(),
+				},
+			},
+		},
+		{
+			name: "PV has no ClaimRef",
+			pv:   builder.ForPersistentVolume("testPV").Result(),
+			expectedVolumeInfo: map[string]PvcPvInfo{
+				"testPV": {
+					PVCName:      "",
+					PVCNamespace: "",
+					PV:           *builder.ForPersistentVolume("testPV").Result(),
+				},
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			ib := itemBackupper{}
+			ib.backupRequest = new(Request)
+			ib.backupRequest.VolumeInfos.VolumeInfos = make([]volume.VolumeInfo, 0)
+
+			pvObj, err := runtime.DefaultUnstructuredConverter.ToUnstructured(tc.pv)
+			require.NoError(t, err)
+			logger := logrus.StandardLogger()
+
+			err = ib.addVolumeInfo(&unstructured.Unstructured{Object: pvObj}, logger)
+			require.NoError(t, err)
+			require.Equal(t, tc.expectedVolumeInfo, ib.backupRequest.PVMap)
+		})
+	}
+}
--- a/pkg/backup/pv_skip_tracker.go
+++ b/pkg/backup/pv_skip_tracker.go
@ -10,6 +10,14 @@ type SkippedPV struct {
 	Reasons []PVSkipReason `json:"reasons"`
 }

+func (s *SkippedPV) SerializeSkipReasons() string {
+	ret := ""
+	for _, reason := range s.Reasons {
+		ret = ret + reason.Approach + ": " + reason.Reason + ";"
+	}
+	return ret
+}
+
 type PVSkipReason struct {
 	Approach string `json:"approach"`
 	Reason   string `json:"reason"`
--- a/pkg/backup/pv_skip_tracker_test.go
+++ b/pkg/backup/pv_skip_tracker_test.go
@ -4,6 +4,7 @@ import (
 	"testing"

 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )

 func TestSummary(t *testing.T) {
@ -41,3 +42,14 @@ func TestSummary(t *testing.T) {
 	}
 	assert.Equal(t, expected, tracker.Summary())
 }
+
+func TestSerializeSkipReasons(t *testing.T) {
+	tracker := NewSkipPVTracker()
+	//tracker.Track("pv5", "", "skipped due to policy")
+	tracker.Track("pv3", podVolumeApproach, "it's set to opt-out")
+	tracker.Track("pv3", csiSnapshotApproach, "not applicable for CSI ")
+
+	for _, skippedPV := range tracker.Summary() {
+		require.Equal(t, "csiSnapshot: not applicable for CSI ;podvolume: it's set to opt-out;", skippedPV.SerializeSkipReasons())
+	}
+}
--- a/pkg/backup/request.go
+++ b/pkg/backup/request.go
@ -20,6 +20,8 @@ import (
 	"fmt"
 	"sort"

+	corev1api "k8s.io/api/core/v1"
+
 	"github.com/vmware-tanzu/velero/internal/hook"
 	"github.com/vmware-tanzu/velero/internal/resourcepolicies"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
@ -52,6 +54,16 @@ type Request struct {
 	itemOperationsList        *[]*itemoperation.BackupOperation
 	ResPolicies               *resourcepolicies.Policies
 	SkippedPVTracker          *skipPVTracker
+	// A map contains the backup-included PV detail content.
+	// The key is PV name or PVC name(The format is PVC-namespace/PVC-name)
+	PVMap       map[string]PvcPvInfo
+	VolumeInfos volume.VolumeInfos
+}
+
+type PvcPvInfo struct {
+	PVCName      string
+	PVCNamespace string
+	PV           corev1api.PersistentVolume
 }

 // GetItemOperationsList returns ItemOperationsList, initializing it if necessary
--- a/pkg/backup/snapshots.go
+++ b/pkg/backup/snapshots.go
@ -4,7 +4,6 @@ import (
 	"context"

 	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
-	snapshotv1listers "github.com/kubernetes-csi/external-snapshotter/client/v4/listers/volumesnapshot/v1"
 	"github.com/sirupsen/logrus"
 	"k8s.io/apimachinery/pkg/util/sets"
 	kbclient "sigs.k8s.io/controller-runtime/pkg/client"
@ -17,27 +16,25 @@ import (

 // Common function to update the status of CSI snapshots
 // returns VolumeSnapshot, VolumeSnapshotContent, VolumeSnapshotClasses referenced
-func UpdateBackupCSISnapshotsStatus(client kbclient.Client, volumeSnapshotLister snapshotv1listers.VolumeSnapshotLister, backup *velerov1api.Backup, backupLog logrus.FieldLogger) (volumeSnapshots []snapshotv1api.VolumeSnapshot, volumeSnapshotContents []snapshotv1api.VolumeSnapshotContent, volumeSnapshotClasses []snapshotv1api.VolumeSnapshotClass) {
+func UpdateBackupCSISnapshotsStatus(client kbclient.Client, globalCRClient kbclient.Client, backup *velerov1api.Backup, backupLog logrus.FieldLogger) (volumeSnapshots []snapshotv1api.VolumeSnapshot, volumeSnapshotContents []snapshotv1api.VolumeSnapshotContent, volumeSnapshotClasses []snapshotv1api.VolumeSnapshotClass) {
 	if boolptr.IsSetToTrue(backup.Spec.SnapshotMoveData) {
 		backupLog.Info("backup SnapshotMoveData is set to true, skip VolumeSnapshot resource persistence.")
 	} else if features.IsEnabled(velerov1api.CSIFeatureFlag) {
 		selector := label.NewSelectorForBackup(backup.Name)
 		vscList := &snapshotv1api.VolumeSnapshotContentList{}

-		if volumeSnapshotLister != nil {
-			tmpVSs, err := volumeSnapshotLister.List(label.NewSelectorForBackup(backup.Name))
-			if err != nil {
-				backupLog.Error(err)
-			}
-			for _, vs := range tmpVSs {
-				volumeSnapshots = append(volumeSnapshots, *vs)
-			}
-		}
-
-		err := client.List(context.Background(), vscList, &kbclient.ListOptions{LabelSelector: selector})
+		vsList := new(snapshotv1api.VolumeSnapshotList)
+		err := globalCRClient.List(context.TODO(), vsList, &kbclient.ListOptions{
+			LabelSelector: label.NewSelectorForBackup(backup.Name),
+		})
 		if err != nil {
 			backupLog.Error(err)
 		}
+		volumeSnapshots = append(volumeSnapshots, vsList.Items...)
+
+		if err := client.List(context.Background(), vscList, &kbclient.ListOptions{LabelSelector: selector}); err != nil {
+			backupLog.Error(err)
+		}
 		if len(vscList.Items) >= 0 {
 			volumeSnapshotContents = vscList.Items
 		}
--- a/pkg/builder/volume_snapshot_builder.go
+++ b/pkg/builder/volume_snapshot_builder.go
@ -18,6 +18,7 @@ package builder

 import (
 	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )

@ -68,7 +69,21 @@ func (v *VolumeSnapshotBuilder) BoundVolumeSnapshotContentName(vscName string) *
 	return v
 }

+// SourcePVC set the built VolumeSnapshot's spec.Source.PersistentVolumeClaimName.
 func (v *VolumeSnapshotBuilder) SourcePVC(name string) *VolumeSnapshotBuilder {
 	v.object.Spec.Source.PersistentVolumeClaimName = &name
 	return v
 }
+
+// RestoreSize set the built VolumeSnapshot's status.RestoreSize.
+func (v *VolumeSnapshotBuilder) RestoreSize(size string) *VolumeSnapshotBuilder {
+	resourceSize := resource.MustParse(size)
+	v.object.Status.RestoreSize = &resourceSize
+	return v
+}
+
+// VolumeSnapshotClass set the built VolumeSnapshot's spec.VolumeSnapshotClassName value.
+func (v *VolumeSnapshotBuilder) VolumeSnapshotClass(name string) *VolumeSnapshotBuilder {
+	v.object.Spec.VolumeSnapshotClassName = &name
+	return v
+}
--- a/pkg/builder/volume_snapshot_content_builder.go
+++ b/pkg/builder/volume_snapshot_content_builder.go
@ -59,6 +59,7 @@ func (v *VolumeSnapshotContentBuilder) DeletionPolicy(policy snapshotv1api.Delet
 	return v
 }

+// VolumeSnapshotRef sets the built VolumeSnapshotContent's spec.VolumeSnapshotRef value.
 func (v *VolumeSnapshotContentBuilder) VolumeSnapshotRef(namespace, name string) *VolumeSnapshotContentBuilder {
 	v.object.Spec.VolumeSnapshotRef = v1.ObjectReference{
 		APIVersion: "snapshot.storage.k8s.io/v1",
@ -68,3 +69,18 @@ func (v *VolumeSnapshotContentBuilder) VolumeSnapshotRef(namespace, name string)
 	}
 	return v
 }
+
+// VolumeSnapshotClassName sets the built VolumeSnapshotContent's spec.VolumeSnapshotClassName value.
+func (v *VolumeSnapshotContentBuilder) VolumeSnapshotClassName(name string) *VolumeSnapshotContentBuilder {
+	v.object.Spec.VolumeSnapshotClassName = &name
+	return v
+}
+
+// ObjectMeta applies functional options to the VolumeSnapshotContent's ObjectMeta.
+func (v *VolumeSnapshotContentBuilder) ObjectMeta(opts ...ObjectMetaOpt) *VolumeSnapshotContentBuilder {
+	for _, opt := range opts {
+		opt(v.object)
+	}
+
+	return v
+}
--- a/pkg/client/factory.go
+++ b/pkg/client/factory.go
@ -24,6 +24,7 @@ import (
 	k8scheme "k8s.io/client-go/kubernetes/scheme"
 	kbclient "sigs.k8s.io/controller-runtime/pkg/client"

+	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
 	"github.com/pkg/errors"
 	"github.com/spf13/pflag"
 	"k8s.io/apimachinery/pkg/runtime"
@ -158,6 +159,9 @@ func (f *factory) KubebuilderClient() (kbclient.Client, error) {
 	if err := apiextv1.AddToScheme(scheme); err != nil {
 		return nil, err
 	}
+	if err := snapshotv1api.AddToScheme(scheme); err != nil {
+		return nil, err
+	}
 	kubebuilderClient, err := kbclient.New(clientConfig, kbclient.Options{
 		Scheme: scheme,
 	})
--- a/pkg/cmd/server/server.go
+++ b/pkg/cmd/server/server.go
@ -23,21 +23,16 @@ import (
 	"net/http"
 	"net/http/pprof"
 	"os"
-	"reflect"
 	"strings"
 	"time"

 	logrusr "github.com/bombsimon/logrusr/v3"
 	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
-	snapshotv1client "github.com/kubernetes-csi/external-snapshotter/client/v4/clientset/versioned"
-	snapshotv1informers "github.com/kubernetes-csi/external-snapshotter/client/v4/informers/externalversions"
-	snapshotv1listers "github.com/kubernetes-csi/external-snapshotter/client/v4/listers/volumesnapshot/v1"
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 	corev1api "k8s.io/api/core/v1"
-	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
@ -244,15 +239,17 @@ func NewCommand(f client.Factory) *cobra.Command {
 }

 type server struct {
-	namespace             string
-	metricsAddress        string
-	kubeClientConfig      *rest.Config
-	kubeClient            kubernetes.Interface
-	discoveryClient       discovery.DiscoveryInterface
-	discoveryHelper       velerodiscovery.Helper
-	dynamicClient         dynamic.Interface
-	csiSnapshotClient     *snapshotv1client.Clientset
-	csiSnapshotLister     snapshotv1listers.VolumeSnapshotLister
+	namespace        string
+	metricsAddress   string
+	kubeClientConfig *rest.Config
+	kubeClient       kubernetes.Interface
+	discoveryClient  discovery.DiscoveryInterface
+	discoveryHelper  velerodiscovery.Helper
+	dynamicClient    dynamic.Interface
+	// controller-runtime client. the difference from the controller-manager's client
+	// is that the the controller-manager's client is limited to list namespaced-scoped
+	// resources in the namespace where Velero is installed, or the cluster-scoped
+	// resources. The crClient doesn't have the limitation.
 	crClient              ctrlclient.Client
 	ctx                   context.Context
 	cancelFunc            context.CancelFunc
@ -399,23 +396,6 @@ func newServer(f client.Factory, config serverConfig, logger *logrus.Logger) (*s
 		featureVerifier:       featureVerifier,
 	}

-	// Setup CSI snapshot client and lister
-	var csiSnapClient *snapshotv1client.Clientset
-	if features.IsEnabled(velerov1api.CSIFeatureFlag) {
-		csiSnapClient, err = snapshotv1client.NewForConfig(clientConfig)
-		if err != nil {
-			cancelFunc()
-			return nil, err
-		}
-		s.csiSnapshotClient = csiSnapClient
-
-		s.csiSnapshotLister, err = s.getCSIVolumeSnapshotListers()
-		if err != nil {
-			cancelFunc()
-			return nil, err
-		}
-	}
-
 	return s, nil
 }

@ -615,40 +595,6 @@ func (s *server) initRepoManager() error {
 	return nil
 }

-func (s *server) getCSIVolumeSnapshotListers() (vsLister snapshotv1listers.VolumeSnapshotLister, err error) {
-	_, err = s.discoveryClient.ServerResourcesForGroupVersion(snapshotv1api.SchemeGroupVersion.String())
-	switch {
-	case apierrors.IsNotFound(err):
-		// CSI is enabled, but the required CRDs aren't installed, so halt.
-		s.logger.Warnf("The '%s' feature flag was specified, but CSI API group [%s] was not found.", velerov1api.CSIFeatureFlag, snapshotv1api.SchemeGroupVersion.String())
-		err = nil
-	case err == nil:
-		wrapper := NewCSIInformerFactoryWrapper(s.csiSnapshotClient)
-
-		s.logger.Debug("Creating CSI listers")
-		// Access the wrapped factory directly here since we've already done the feature flag check above to know it's safe.
-		vsLister = wrapper.factory.Snapshot().V1().VolumeSnapshots().Lister()
-
-		// start the informers & and wait for the caches to sync
-		wrapper.Start(s.ctx.Done())
-		s.logger.Info("Waiting for informer caches to sync")
-		csiCacheSyncResults := wrapper.WaitForCacheSync(s.ctx.Done())
-		s.logger.Info("Done waiting for informer caches to sync")
-
-		for informer, synced := range csiCacheSyncResults {
-			if !synced {
-				err = errors.Errorf("cache was not synced for informer %v", informer)
-				return
-			}
-			s.logger.WithField("informer", informer).Info("Informer cache synced")
-		}
-	case err != nil:
-		s.logger.Errorf("fail to find snapshot v1 schema: %s", err)
-	}
-
-	return vsLister, err
-}
-
 func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string) error {
 	s.logger.Info("Starting controllers")

@ -775,10 +721,10 @@ func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string
 			s.metrics,
 			backupStoreGetter,
 			s.config.formatFlag.Parse(),
-			s.csiSnapshotLister,
 			s.credentialFileStore,
 			s.config.maxConcurrentK8SConnections,
 			s.config.defaultSnapshotMoveData,
+			s.crClient,
 		).SetupWithManager(s.mgr); err != nil {
 			s.logger.Fatal(err, "unable to create controller", "controller", controller.Backup)
 		}
@ -837,7 +783,7 @@ func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string
 		cmd.CheckError(err)
 		r := controller.NewBackupFinalizerReconciler(
 			s.mgr.GetClient(),
-			s.csiSnapshotLister,
+			s.crClient,
 			clock.RealClock{},
 			backupper,
 			newPluginManager,
@ -1027,37 +973,6 @@ func (s *server) runProfiler() {
 	}
 }

-// CSIInformerFactoryWrapper is a proxy around the CSI SharedInformerFactory that checks the CSI feature flag before performing operations.
-type CSIInformerFactoryWrapper struct {
-	factory snapshotv1informers.SharedInformerFactory
-}
-
-func NewCSIInformerFactoryWrapper(c snapshotv1client.Interface) *CSIInformerFactoryWrapper {
-	// If no namespace is specified, all namespaces are watched.
-	// This is desirable for VolumeSnapshots, as we want to query for all VolumeSnapshots across all namespaces using this informer
-	w := &CSIInformerFactoryWrapper{}
-
-	if features.IsEnabled(velerov1api.CSIFeatureFlag) {
-		w.factory = snapshotv1informers.NewSharedInformerFactoryWithOptions(c, 0)
-	}
-	return w
-}
-
-// Start proxies the Start call to the CSI SharedInformerFactory.
-func (w *CSIInformerFactoryWrapper) Start(stopCh <-chan struct{}) {
-	if features.IsEnabled(velerov1api.CSIFeatureFlag) {
-		w.factory.Start(stopCh)
-	}
-}
-
-// WaitForCacheSync proxies the WaitForCacheSync call to the CSI SharedInformerFactory.
-func (w *CSIInformerFactoryWrapper) WaitForCacheSync(stopCh <-chan struct{}) map[reflect.Type]bool {
-	if features.IsEnabled(velerov1api.CSIFeatureFlag) {
-		return w.factory.WaitForCacheSync(stopCh)
-	}
-	return nil
-}
-
 // if there is a restarting during the reconciling of backups/restores/etc, these CRs may be stuck in progress status
 // markInProgressCRsFailed tries to mark the in progress CRs as failed when starting the server to avoid the issue
 func markInProgressCRsFailed(ctx context.Context, cfg *rest.Config, scheme *runtime.Scheme, namespace string, log logrus.FieldLogger) {
--- a/pkg/controller/backup_controller.go
+++ b/pkg/controller/backup_controller.go
@ -21,12 +21,11 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"strconv"
 	"strings"
 	"time"

 	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
-	snapshotterClientSet "github.com/kubernetes-csi/external-snapshotter/client/v4/clientset/versioned"
-	snapshotv1listers "github.com/kubernetes-csi/external-snapshotter/client/v4/listers/volumesnapshot/v1"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	corev1api "k8s.io/api/core/v1"
@ -43,14 +42,18 @@ import (
 	"github.com/vmware-tanzu/velero/internal/resourcepolicies"
 	"github.com/vmware-tanzu/velero/internal/storage"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	velerov2alpha1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
 	pkgbackup "github.com/vmware-tanzu/velero/pkg/backup"
 	"github.com/vmware-tanzu/velero/pkg/discovery"
 	"github.com/vmware-tanzu/velero/pkg/features"
+	"github.com/vmware-tanzu/velero/pkg/itemoperation"
+	"github.com/vmware-tanzu/velero/pkg/kuberesource"
 	"github.com/vmware-tanzu/velero/pkg/label"
 	"github.com/vmware-tanzu/velero/pkg/metrics"
 	"github.com/vmware-tanzu/velero/pkg/persistence"
 	"github.com/vmware-tanzu/velero/pkg/plugin/clientmgmt"
 	"github.com/vmware-tanzu/velero/pkg/plugin/framework"
+	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
 	"github.com/vmware-tanzu/velero/pkg/util/collections"
 	"github.com/vmware-tanzu/velero/pkg/util/encode"
@ -84,11 +87,10 @@ type backupReconciler struct {
 	metrics                     *metrics.ServerMetrics
 	backupStoreGetter           persistence.ObjectBackupStoreGetter
 	formatFlag                  logging.Format
-	volumeSnapshotLister        snapshotv1listers.VolumeSnapshotLister
-	volumeSnapshotClient        snapshotterClientSet.Interface
 	credentialFileStore         credentials.FileStore
 	maxConcurrentK8SConnections int
 	defaultSnapshotMoveData     bool
+	globalCRClient              kbclient.Client
 }

 func NewBackupReconciler(
@ -110,10 +112,10 @@ func NewBackupReconciler(
 	metrics *metrics.ServerMetrics,
 	backupStoreGetter persistence.ObjectBackupStoreGetter,
 	formatFlag logging.Format,
-	volumeSnapshotLister snapshotv1listers.VolumeSnapshotLister,
 	credentialStore credentials.FileStore,
 	maxConcurrentK8SConnections int,
 	defaultSnapshotMoveData bool,
+	globalCRClient kbclient.Client,
 ) *backupReconciler {
 	b := &backupReconciler{
 		ctx:                         ctx,
@ -135,10 +137,10 @@ func NewBackupReconciler(
 		metrics:                     metrics,
 		backupStoreGetter:           backupStoreGetter,
 		formatFlag:                  formatFlag,
-		volumeSnapshotLister:        volumeSnapshotLister,
 		credentialFileStore:         credentialStore,
 		maxConcurrentK8SConnections: maxConcurrentK8SConnections,
 		defaultSnapshotMoveData:     defaultSnapshotMoveData,
+		globalCRClient:              globalCRClient,
 	}
 	b.updateTotalBackupMetric()
 	return b
@ -317,6 +319,7 @@ func (b *backupReconciler) prepareBackupRequest(backup *velerov1api.Backup, logg
 	request := &pkgbackup.Request{
 		Backup:           backup.DeepCopy(), // don't modify items in the cache
 		SkippedPVTracker: pkgbackup.NewSkipPVTracker(),
+		PVMap:            map[string]pkgbackup.PvcPvInfo{},
 	}

 	// set backup major version - deprecated, use Status.FormatVersion
@ -665,7 +668,7 @@ func (b *backupReconciler) runBackup(backup *pkgbackup.Request) error {
 			backup.Status.VolumeSnapshotsCompleted++
 		}
 	}
-	volumeSnapshots, volumeSnapshotContents, volumeSnapshotClasses := pkgbackup.UpdateBackupCSISnapshotsStatus(b.kbClient, b.volumeSnapshotLister, backup.Backup, backupLog)
+	volumeSnapshots, volumeSnapshotContents, volumeSnapshotClasses := pkgbackup.UpdateBackupCSISnapshotsStatus(b.kbClient, b.globalCRClient, backup.Backup, backupLog)

 	// Iterate over backup item operations and update progress.
 	// Any errors on operations at this point should be added to backup errors.
@ -734,6 +737,8 @@ func (b *backupReconciler) runBackup(backup *pkgbackup.Request) error {
 	if logFile, err := backupLog.GetPersistFile(); err != nil {
 		fatalErrs = append(fatalErrs, errors.Wrap(err, "error getting backup log file"))
 	} else {
+		backup.VolumeInfos.VolumeInfos = generateVolumeInfo(backup, volumeSnapshots, volumeSnapshotContents, volumeSnapshotClasses, b.globalCRClient, backupLog)
+
 		if errs := persistBackup(backup, backupFile, logFile, backupStore, volumeSnapshots, volumeSnapshotContents, volumeSnapshotClasses, results); len(errs) > 0 {
 			fatalErrs = append(fatalErrs, errs...)
 		}
@ -796,7 +801,6 @@ func persistBackup(backup *pkgbackup.Request,
 ) []error {
 	persistErrs := []error{}
 	backupJSON := new(bytes.Buffer)
-	volumeInfos := make([]volume.VolumeInfo, 0)

 	if err := encode.To(backup.Backup, "json", backupJSON); err != nil {
 		persistErrs = append(persistErrs, errors.Wrap(err, "error encoding backup"))
@ -843,7 +847,7 @@ func persistBackup(backup *pkgbackup.Request,
 		persistErrs = append(persistErrs, errs...)
 	}

-	volumeInfoJSON, errs := encode.ToJSONGzip(volumeInfos, "backup volumes information")
+	volumeInfoJSON, errs := encode.ToJSONGzip(backup.VolumeInfos, "backup volumes information")
 	if errs != nil {
 		persistErrs = append(persistErrs, errs...)
 	}
@ -908,3 +912,328 @@ func oldAndNewFilterParametersUsedTogether(backupSpec velerov1api.BackupSpec) bo

 	return haveOldResourceFilterParameters && haveNewResourceFilterParameters
 }
+
+func generateVolumeInfo(backup *pkgbackup.Request, csiVolumeSnapshots []snapshotv1api.VolumeSnapshot,
+	csiVolumeSnapshotContents []snapshotv1api.VolumeSnapshotContent, csiVolumesnapshotClasses []snapshotv1api.VolumeSnapshotClass,
+	crClient kbclient.Client, logger logrus.FieldLogger) []volume.VolumeInfo {
+	volumeInfos := make([]volume.VolumeInfo, 0)
+
+	skippedVolumeInfos := generateVolumeInfoForSkippedPV(backup, logger)
+	volumeInfos = append(volumeInfos, skippedVolumeInfos...)
+
+	nativeSnapshotVolumeInfos := generateVolumeInfoForVeleroNativeSnapshot(backup, logger)
+	volumeInfos = append(volumeInfos, nativeSnapshotVolumeInfos...)
+
+	csiVolumeInfos := generateVolumeInfoForCSIVolumeSnapshot(backup, csiVolumeSnapshots, csiVolumeSnapshotContents, csiVolumesnapshotClasses, logger)
+	volumeInfos = append(volumeInfos, csiVolumeInfos...)
+
+	pvbVolumeInfos := generateVolumeInfoFromPVB(backup, crClient, logger)
+	volumeInfos = append(volumeInfos, pvbVolumeInfos...)
+
+	dataUploadVolumeInfos := generateVolumeInfoFromDataUpload(backup, crClient, logger)
+	volumeInfos = append(volumeInfos, dataUploadVolumeInfos...)
+
+	return volumeInfos
+}
+
+// generateVolumeInfoForSkippedPV generate VolumeInfos for SkippedPV.
+func generateVolumeInfoForSkippedPV(backup *pkgbackup.Request, logger logrus.FieldLogger) []volume.VolumeInfo {
+	tmpVolumeInfos := make([]volume.VolumeInfo, 0)
+
+	for _, skippedPV := range backup.SkippedPVTracker.Summary() {
+		if pvcPVInfo, ok := backup.PVMap[skippedPV.Name]; ok {
+			volumeInfo := volume.VolumeInfo{
+				PVCName:           pvcPVInfo.PVCName,
+				PVCNamespace:      pvcPVInfo.PVCNamespace,
+				PVName:            skippedPV.Name,
+				SnapshotDataMoved: false,
+				Skipped:           true,
+				SkippedReason:     skippedPV.SerializeSkipReasons(),
+				PVInfo: volume.PVInfo{
+					ReclaimPolicy: string(pvcPVInfo.PV.Spec.PersistentVolumeReclaimPolicy),
+					Labels:        pvcPVInfo.PV.Labels,
+				},
+			}
+			tmpVolumeInfos = append(tmpVolumeInfos, volumeInfo)
+		} else {
+			logger.Warnf("Cannot find info for PV %s", skippedPV.Name)
+			continue
+		}
+	}
+
+	return tmpVolumeInfos
+}
+
+// generateVolumeInfoForVeleroNativeSnapshot generate VolumeInfos for Velero native snapshot
+func generateVolumeInfoForVeleroNativeSnapshot(backup *pkgbackup.Request, logger logrus.FieldLogger) []volume.VolumeInfo {
+	tmpVolumeInfos := make([]volume.VolumeInfo, 0)
+
+	for _, nativeSnapshot := range backup.VolumeSnapshots {
+		var iops int64
+		if nativeSnapshot.Spec.VolumeIOPS != nil {
+			iops = *nativeSnapshot.Spec.VolumeIOPS
+		}
+
+		if pvcPVInfo, ok := backup.PVMap[nativeSnapshot.Spec.PersistentVolumeName]; ok {
+			volumeInfo := volume.VolumeInfo{
+				BackupMethod:      volume.NativeSnapshot,
+				PVCName:           pvcPVInfo.PVCName,
+				PVCNamespace:      pvcPVInfo.PVCNamespace,
+				PVName:            pvcPVInfo.PV.Name,
+				SnapshotDataMoved: false,
+				Skipped:           false,
+				NativeSnapshotInfo: volume.NativeSnapshotInfo{
+					SnapshotHandle: nativeSnapshot.Status.ProviderSnapshotID,
+					VolumeType:     nativeSnapshot.Spec.VolumeType,
+					VolumeAZ:       nativeSnapshot.Spec.VolumeAZ,
+					IOPS:           strconv.FormatInt(iops, 10),
+				},
+				PVInfo: volume.PVInfo{
+					ReclaimPolicy: string(pvcPVInfo.PV.Spec.PersistentVolumeReclaimPolicy),
+					Labels:        pvcPVInfo.PV.Labels,
+				},
+			}
+
+			tmpVolumeInfos = append(tmpVolumeInfos, volumeInfo)
+		} else {
+			logger.Warnf("cannot find info for PV %s", nativeSnapshot.Spec.PersistentVolumeName)
+			continue
+		}
+	}
+
+	return tmpVolumeInfos
+}
+
+// generateVolumeInfoForCSIVolumeSnapshot generate VolumeInfos for CSI VolumeSnapshot
+func generateVolumeInfoForCSIVolumeSnapshot(backup *pkgbackup.Request, csiVolumeSnapshots []snapshotv1api.VolumeSnapshot,
+	csiVolumeSnapshotContents []snapshotv1api.VolumeSnapshotContent, csiVolumesnapshotClasses []snapshotv1api.VolumeSnapshotClass,
+	logger logrus.FieldLogger) []volume.VolumeInfo {
+	tmpVolumeInfos := make([]volume.VolumeInfo, 0)
+
+	for _, volumeSnapshot := range csiVolumeSnapshots {
+		var volumeSnapshotClass *snapshotv1api.VolumeSnapshotClass
+		var volumeSnapshotContent *snapshotv1api.VolumeSnapshotContent
+
+		// This is protective logic. The passed-in VS should be all related
+		// to this backup.
+		if volumeSnapshot.Labels[velerov1api.BackupNameLabel] != backup.Name {
+			continue
+		}
+
+		if volumeSnapshot.Spec.VolumeSnapshotClassName == nil {
+			logger.Warnf("Cannot find VolumeSnapshotClass for VolumeSnapshot %s/%s", volumeSnapshot.Namespace, volumeSnapshot.Name)
+			continue
+		}
+
+		if volumeSnapshot.Status == nil || volumeSnapshot.Status.BoundVolumeSnapshotContentName == nil {
+			logger.Warnf("Cannot fine VolumeSnapshotContent for VolumeSnapshot %s/%s", volumeSnapshot.Namespace, volumeSnapshot.Name)
+			continue
+		}
+
+		if volumeSnapshot.Spec.Source.PersistentVolumeClaimName == nil {
+			logger.Warnf("VolumeSnapshot %s/%s doesn't have a source PVC", volumeSnapshot.Namespace, volumeSnapshot.Name)
+			continue
+		}
+
+		for index := range csiVolumesnapshotClasses {
+			if *volumeSnapshot.Spec.VolumeSnapshotClassName == csiVolumesnapshotClasses[index].Name {
+				volumeSnapshotClass = &csiVolumesnapshotClasses[index]
+			}
+		}
+
+		for index := range csiVolumeSnapshotContents {
+			if *volumeSnapshot.Status.BoundVolumeSnapshotContentName == csiVolumeSnapshotContents[index].Name {
+				volumeSnapshotContent = &csiVolumeSnapshotContents[index]
+			}
+		}
+
+		if volumeSnapshotClass == nil || volumeSnapshotContent == nil {
+			logger.Warnf("fail to get VolumeSnapshotContent or VolumeSnapshotClass for VolumeSnapshot: %s/%s",
+				volumeSnapshot.Namespace, volumeSnapshot.Name)
+			continue
+		}
+
+		var operation itemoperation.BackupOperation
+		for _, op := range *backup.GetItemOperationsList() {
+			if op.Spec.ResourceIdentifier.GroupResource.String() == kuberesource.VolumeSnapshots.String() &&
+				op.Spec.ResourceIdentifier.Name == volumeSnapshot.Name &&
+				op.Spec.ResourceIdentifier.Namespace == volumeSnapshot.Namespace {
+				operation = *op
+			}
+		}
+
+		var size int64
+		if volumeSnapshot.Status.RestoreSize != nil {
+			size = volumeSnapshot.Status.RestoreSize.Value()
+		}
+		snapshotHandle := ""
+		if volumeSnapshotContent.Status.SnapshotHandle != nil {
+			snapshotHandle = *volumeSnapshotContent.Status.SnapshotHandle
+		}
+		if pvcPVInfo, ok := backup.PVMap[volumeSnapshot.Namespace+"/"+*volumeSnapshot.Spec.Source.PersistentVolumeClaimName]; ok {
+			volumeInfo := volume.VolumeInfo{
+				BackupMethod:          volume.CSISnapshot,
+				PVCName:               pvcPVInfo.PVCName,
+				PVCNamespace:          pvcPVInfo.PVCNamespace,
+				PVName:                pvcPVInfo.PV.Name,
+				Skipped:               false,
+				SnapshotDataMoved:     false,
+				PreserveLocalSnapshot: true,
+				OperationID:           operation.Spec.OperationID,
+				StartTimestamp:        &volumeSnapshot.CreationTimestamp,
+				CSISnapshotInfo: volume.CSISnapshotInfo{
+					VSCName:        *volumeSnapshot.Status.BoundVolumeSnapshotContentName,
+					Size:           size,
+					Driver:         volumeSnapshotClass.Driver,
+					SnapshotHandle: snapshotHandle,
+				},
+				PVInfo: volume.PVInfo{
+					ReclaimPolicy: string(pvcPVInfo.PV.Spec.PersistentVolumeReclaimPolicy),
+					Labels:        pvcPVInfo.PV.Labels,
+				},
+			}
+
+			tmpVolumeInfos = append(tmpVolumeInfos, volumeInfo)
+		} else {
+			logger.Warnf("cannot find info for PVC %s/%s", volumeSnapshot.Namespace, volumeSnapshot.Spec.Source.PersistentVolumeClaimName)
+			continue
+		}
+	}
+
+	return tmpVolumeInfos
+}
+
+// generateVolumeInfoFromPVB generate VolumeInfo for PVB.
+func generateVolumeInfoFromPVB(backup *pkgbackup.Request, crClient kbclient.Client, logger logrus.FieldLogger) []volume.VolumeInfo {
+	tmpVolumeInfos := make([]volume.VolumeInfo, 0)
+
+	for _, pvb := range backup.PodVolumeBackups {
+		volumeInfo := volume.VolumeInfo{
+			BackupMethod:      volume.PodVolumeBackup,
+			SnapshotDataMoved: false,
+			Skipped:           false,
+			StartTimestamp:    pvb.Status.StartTimestamp,
+			PVBInfo: volume.PodVolumeBackupInfo{
+				SnapshotHandle: pvb.Status.SnapshotID,
+				Size:           pvb.Status.Progress.TotalBytes,
+				UploaderType:   pvb.Spec.UploaderType,
+				VolumeName:     pvb.Spec.Volume,
+				PodName:        pvb.Spec.Pod.Name,
+				PodNamespace:   pvb.Spec.Pod.Namespace,
+				NodeName:       pvb.Spec.Node,
+			},
+		}
+
+		pod := new(corev1api.Pod)
+		pvcName := ""
+		err := crClient.Get(context.TODO(), kbclient.ObjectKey{Namespace: pvb.Spec.Pod.Namespace, Name: pvb.Spec.Pod.Name}, pod)
+		if err != nil {
+			logger.WithError(err).Warn("Fail to get pod for PodVolumeBackup: ", pvb.Name)
+			continue
+		}
+		for _, volume := range pod.Spec.Volumes {
+			if volume.Name == pvb.Spec.Volume && volume.PersistentVolumeClaim != nil {
+				pvcName = volume.PersistentVolumeClaim.ClaimName
+			}
+		}
+
+		if pvcName != "" {
+			if pvcPVInfo, ok := backup.PVMap[pod.Namespace+"/"+pvcName]; ok {
+				volumeInfo.PVCName = pvcPVInfo.PVCName
+				volumeInfo.PVCNamespace = pvcPVInfo.PVCNamespace
+				volumeInfo.PVName = pvcPVInfo.PV.Name
+				volumeInfo.PVInfo = volume.PVInfo{
+					ReclaimPolicy: string(pvcPVInfo.PV.Spec.PersistentVolumeReclaimPolicy),
+					Labels:        pvcPVInfo.PV.Labels,
+				}
+			} else {
+				logger.Warnf("Cannot find info for PVC %s/%s", pod.Namespace, pvcName)
+				continue
+			}
+		} else {
+			logger.Debug("The PVB %s doesn't have a corresponding PVC", pvb.Name)
+		}
+
+		tmpVolumeInfos = append(tmpVolumeInfos, volumeInfo)
+	}
+
+	return tmpVolumeInfos
+}
+
+// generateVolumeInfoFromDataUpload generate VolumeInfo for DataUpload.
+func generateVolumeInfoFromDataUpload(backup *pkgbackup.Request, crClient kbclient.Client, logger logrus.FieldLogger) []volume.VolumeInfo {
+	tmpVolumeInfos := make([]volume.VolumeInfo, 0)
+	vsClassList := new(snapshotv1api.VolumeSnapshotClassList)
+	if err := crClient.List(context.TODO(), vsClassList); err != nil {
+		logger.WithError(err).Errorf("cannot list VolumeSnapshotClass %s", err.Error())
+		return tmpVolumeInfos
+	}
+
+	for _, operation := range *backup.GetItemOperationsList() {
+		if operation.Spec.ResourceIdentifier.GroupResource.String() == kuberesource.PersistentVolumeClaims.String() {
+			var duIdentifier velero.ResourceIdentifier
+
+			for _, identifier := range operation.Spec.PostOperationItems {
+				if identifier.GroupResource.String() == "datauploads.velero.io" {
+					duIdentifier = identifier
+				}
+			}
+			if duIdentifier.Empty() {
+				logger.Warnf("cannot find DataUpload for PVC %s/%s backup async operation",
+					operation.Spec.ResourceIdentifier.Namespace, operation.Spec.ResourceIdentifier.Name)
+				continue
+			}
+
+			dataUpload := new(velerov2alpha1.DataUpload)
+			err := crClient.Get(
+				context.TODO(),
+				kbclient.ObjectKey{
+					Namespace: duIdentifier.Namespace,
+					Name:      duIdentifier.Name},
+				dataUpload,
+			)
+			if err != nil {
+				logger.Warnf("fail to get DataUpload for operation %s: %s", operation.Spec.OperationID, err.Error())
+				continue
+			}
+
+			driverUsedByVSClass := ""
+			for index := range vsClassList.Items {
+				if vsClassList.Items[index].Name == dataUpload.Spec.CSISnapshot.SnapshotClass {
+					driverUsedByVSClass = vsClassList.Items[index].Driver
+				}
+			}
+
+			if pvcPVInfo, ok := backup.PVMap[operation.Spec.ResourceIdentifier.Namespace+"/"+operation.Spec.ResourceIdentifier.Name]; ok {
+				volumeInfo := volume.VolumeInfo{
+					BackupMethod:      volume.CSISnapshot,
+					PVCName:           pvcPVInfo.PVCName,
+					PVCNamespace:      pvcPVInfo.PVCNamespace,
+					PVName:            pvcPVInfo.PV.Name,
+					SnapshotDataMoved: true,
+					Skipped:           false,
+					OperationID:       operation.Spec.OperationID,
+					StartTimestamp:    operation.Status.Created,
+					CSISnapshotInfo: volume.CSISnapshotInfo{
+						Driver: driverUsedByVSClass,
+					},
+					SnapshotDataMovementInfo: volume.SnapshotDataMovementInfo{
+						DataMover:    dataUpload.Spec.DataMover,
+						UploaderType: "kopia",
+					},
+					PVInfo: volume.PVInfo{
+						ReclaimPolicy: string(pvcPVInfo.PV.Spec.PersistentVolumeReclaimPolicy),
+						Labels:        pvcPVInfo.PV.Labels,
+					},
+				}
+
+				tmpVolumeInfos = append(tmpVolumeInfos, volumeInfo)
+			} else {
+				logger.Warnf("Cannot find info for PVC %s/%s", operation.Spec.ResourceIdentifier.Namespace, operation.Spec.ResourceIdentifier.Name)
+				continue
+			}
+		}
+	}
+
+	return tmpVolumeInfos
+}
--- a/pkg/controller/backup_controller_test.go
+++ b/pkg/controller/backup_controller_test.go
@ -29,15 +29,16 @@ import (

 	"github.com/google/go-cmp/cmp"
 	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
-	snapshotfake "github.com/kubernetes-csi/external-snapshotter/client/v4/clientset/versioned/fake"
-	snapshotinformers "github.com/kubernetes-csi/external-snapshotter/client/v4/informers/externalversions"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
+	corev1api "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/version"
 	"k8s.io/utils/clock"
@ -45,11 +46,14 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 	kbclient "sigs.k8s.io/controller-runtime/pkg/client"

+	"github.com/vmware-tanzu/velero/pkg/backup"
 	kubeutil "github.com/vmware-tanzu/velero/pkg/util/kube"
+	"github.com/vmware-tanzu/velero/pkg/volume"

 	fakeClient "sigs.k8s.io/controller-runtime/pkg/client/fake"

 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	velerov2alpha1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
 	pkgbackup "github.com/vmware-tanzu/velero/pkg/backup"
 	"github.com/vmware-tanzu/velero/pkg/builder"
 	"github.com/vmware-tanzu/velero/pkg/discovery"
@ -61,6 +65,7 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/plugin/clientmgmt"
 	"github.com/vmware-tanzu/velero/pkg/plugin/framework"
 	pluginmocks "github.com/vmware-tanzu/velero/pkg/plugin/mocks"
+	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
 	biav2 "github.com/vmware-tanzu/velero/pkg/plugin/velero/backupitemaction/v2"
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
@ -1062,12 +1067,11 @@ func TestProcessBackupCompletions(t *testing.T) {
 					CSIVolumeSnapshotsCompleted: 0,
 				},
 			},
-			volumeSnapshot: builder.ForVolumeSnapshot("velero", "testVS").ObjectMeta(builder.WithLabels(velerov1api.BackupNameLabel, "backup-1")).Result(),
+			volumeSnapshot: builder.ForVolumeSnapshot("velero", "testVS").VolumeSnapshotClass("testClass").Status().BoundVolumeSnapshotContentName("testVSC").RestoreSize("10G").SourcePVC("testPVC").ObjectMeta(builder.WithLabels(velerov1api.BackupNameLabel, "backup-1")).Result(),
 		},
 		{
-			name:   "backup with snapshot data movement set to false when CSI feature is enabled",
-			backup: defaultBackup().SnapshotMoveData(false).Result(),
-			//backup:                   defaultBackup().Result(),
+			name:                     "backup with snapshot data movement set to false when CSI feature is enabled",
+			backup:                   defaultBackup().SnapshotMoveData(false).Result(),
 			backupLocation:           defaultBackupLocation,
 			defaultVolumesToFsBackup: false,
 			expectedResult: &velerov1api.Backup{
@ -1103,7 +1107,7 @@ func TestProcessBackupCompletions(t *testing.T) {
 					CSIVolumeSnapshotsCompleted: 0,
 				},
 			},
-			volumeSnapshot: builder.ForVolumeSnapshot("velero", "testVS").ObjectMeta(builder.WithLabels(velerov1api.BackupNameLabel, "backup-1")).Result(),
+			volumeSnapshot: builder.ForVolumeSnapshot("velero", "testVS").VolumeSnapshotClass("testClass").Status().BoundVolumeSnapshotContentName("testVSC").RestoreSize("10G").SourcePVC("testPVC").ObjectMeta(builder.WithLabels(velerov1api.BackupNameLabel, "backup-1")).Result(),
 		},
 		{
 			name:                     "backup with snapshot data movement not set when CSI feature is enabled",
@ -1143,7 +1147,7 @@ func TestProcessBackupCompletions(t *testing.T) {
 					CSIVolumeSnapshotsCompleted: 0,
 				},
 			},
-			volumeSnapshot: builder.ForVolumeSnapshot("velero", "testVS").ObjectMeta(builder.WithLabels(velerov1api.BackupNameLabel, "backup-1")).Result(),
+			volumeSnapshot: builder.ForVolumeSnapshot("velero", "testVS").VolumeSnapshotClass("testClass").Status().BoundVolumeSnapshotContentName("testVSC").RestoreSize("10G").SourcePVC("testPVC").ObjectMeta(builder.WithLabels(velerov1api.BackupNameLabel, "backup-1")).Result(),
 		},
 		{
 			name:                     "backup with snapshot data movement set to true and defaultSnapshotMoveData set to false",
@ -1184,7 +1188,7 @@ func TestProcessBackupCompletions(t *testing.T) {
 					CSIVolumeSnapshotsCompleted: 0,
 				},
 			},
-			volumeSnapshot: builder.ForVolumeSnapshot("velero", "testVS").ObjectMeta(builder.WithLabels(velerov1api.BackupNameLabel, "backup-1")).Result(),
+			volumeSnapshot: builder.ForVolumeSnapshot("velero", "testVS").VolumeSnapshotClass("testClass").Status().BoundVolumeSnapshotContentName("testVSC").RestoreSize("10G").SourcePVC("testPVC").ObjectMeta(builder.WithLabels(velerov1api.BackupNameLabel, "backup-1")).Result(),
 		},
 		{
 			name:                     "backup with snapshot data movement set to false and defaultSnapshotMoveData set to true",
@ -1225,7 +1229,7 @@ func TestProcessBackupCompletions(t *testing.T) {
 					CSIVolumeSnapshotsCompleted: 0,
 				},
 			},
-			volumeSnapshot: builder.ForVolumeSnapshot("velero", "testVS").ObjectMeta(builder.WithLabels(velerov1api.BackupNameLabel, "backup-1")).Result(),
+			volumeSnapshot: builder.ForVolumeSnapshot("velero", "testVS").VolumeSnapshotClass("testClass").Status().BoundVolumeSnapshotContentName("testVSC").RestoreSize("10G").SourcePVC("testPVC").ObjectMeta(builder.WithLabels(velerov1api.BackupNameLabel, "backup-1")).Result(),
 		},
 		{
 			name:                     "backup with snapshot data movement not set and defaultSnapshotMoveData set to true",
@ -1266,35 +1270,43 @@ func TestProcessBackupCompletions(t *testing.T) {
 					CSIVolumeSnapshotsCompleted: 0,
 				},
 			},
-			volumeSnapshot: builder.ForVolumeSnapshot("velero", "testVS").ObjectMeta(builder.WithLabels(velerov1api.BackupNameLabel, "backup-1")).Result(),
+			volumeSnapshot: builder.ForVolumeSnapshot("velero", "testVS").VolumeSnapshotClass("testClass").Status().BoundVolumeSnapshotContentName("testVSC").RestoreSize("10G").SourcePVC("testPVC").ObjectMeta(builder.WithLabels(velerov1api.BackupNameLabel, "backup-1")).Result(),
 		},
 	}

+	snapshotHandle := "testSnapshotID"
+
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			formatFlag := logging.FormatText
 			var (
-				logger         = logging.DefaultLogger(logrus.DebugLevel, formatFlag)
-				pluginManager  = new(pluginmocks.Manager)
-				backupStore    = new(persistencemocks.BackupStore)
-				backupper      = new(fakeBackupper)
-				snapshotClient = snapshotfake.NewSimpleClientset()
-				sharedInformer = snapshotinformers.NewSharedInformerFactory(snapshotClient, 0)
-				snapshotLister = sharedInformer.Snapshot().V1().VolumeSnapshots().Lister()
+				logger           = logging.DefaultLogger(logrus.DebugLevel, formatFlag)
+				pluginManager    = new(pluginmocks.Manager)
+				backupStore      = new(persistencemocks.BackupStore)
+				backupper        = new(fakeBackupper)
+				fakeGlobalClient = velerotest.NewFakeControllerRuntimeClient(t)
 			)

 			var fakeClient kbclient.Client
 			// add the test's backup storage location if it's different than the default
 			if test.backupLocation != nil && test.backupLocation != defaultBackupLocation {
-				fakeClient = velerotest.NewFakeControllerRuntimeClient(t, test.backupLocation)
+				fakeClient = velerotest.NewFakeControllerRuntimeClient(t, test.backupLocation,
+					builder.ForVolumeSnapshotClass("testClass").Driver("testDriver").Result(),
+					builder.ForVolumeSnapshotContent("testVSC").ObjectMeta(builder.WithLabels(velerov1api.BackupNameLabel, "backup-1")).VolumeSnapshotClassName("testClass").Status(&snapshotv1api.VolumeSnapshotContentStatus{
+						SnapshotHandle: &snapshotHandle,
+					}).Result(),
+				)
 			} else {
-				fakeClient = velerotest.NewFakeControllerRuntimeClient(t)
+				fakeClient = velerotest.NewFakeControllerRuntimeClient(t,
+					builder.ForVolumeSnapshotClass("testClass").Driver("testDriver").Result(),
+					builder.ForVolumeSnapshotContent("testVSC").ObjectMeta(builder.WithLabels(velerov1api.BackupNameLabel, "backup-1")).VolumeSnapshotClassName("testClass").Status(&snapshotv1api.VolumeSnapshotContentStatus{
+						SnapshotHandle: &snapshotHandle,
+					}).Result(),
+				)
 			}

 			if test.volumeSnapshot != nil {
-				snapshotClient.SnapshotV1().VolumeSnapshots(test.volumeSnapshot.Namespace).Create(context.Background(), test.volumeSnapshot, metav1.CreateOptions{})
-				sharedInformer.Snapshot().V1().VolumeSnapshots().Informer().GetStore().Add(test.volumeSnapshot)
-				sharedInformer.WaitForCacheSync(make(chan struct{}))
+				require.NoError(t, fakeGlobalClient.Create(context.TODO(), test.volumeSnapshot))
 			}

 			apiServer := velerotest.NewAPIServer(t)
@ -1328,8 +1340,7 @@ func TestProcessBackupCompletions(t *testing.T) {
 				backupStoreGetter:        NewFakeSingleObjectBackupStoreGetter(backupStore),
 				backupper:                backupper,
 				formatFlag:               formatFlag,
-				volumeSnapshotClient:     snapshotClient,
-				volumeSnapshotLister:     snapshotLister,
+				globalCRClient:           fakeGlobalClient,
 			}

 			pluginManager.On("GetBackupItemActionsV2").Return(nil, nil)
@ -1731,3 +1742,749 @@ func TestPatchResourceWorksWithStatus(t *testing.T) {

 	}
 }
+func TestGenerateVolumeInfoForSkippedPV(t *testing.T) {
+	tests := []struct {
+		name                string
+		skippedPVName       string
+		pvMap               map[string]backup.PvcPvInfo
+		expectedVolumeInfos []volume.VolumeInfo
+	}{
+		{
+			name:                "Cannot find info for PV",
+			skippedPVName:       "testPV",
+			expectedVolumeInfos: []volume.VolumeInfo{},
+		},
+		{
+			name:          "Normal Skipped PV info",
+			skippedPVName: "testPV",
+			pvMap: map[string]backup.PvcPvInfo{
+				"velero/testPVC": {
+					PVCName:      "testPVC",
+					PVCNamespace: "velero",
+					PV: corev1api.PersistentVolume{
+						ObjectMeta: metav1.ObjectMeta{
+							Name:   "testPV",
+							Labels: map[string]string{"a": "b"},
+						},
+						Spec: corev1api.PersistentVolumeSpec{
+							PersistentVolumeReclaimPolicy: corev1api.PersistentVolumeReclaimDelete,
+						},
+					},
+				},
+				"testPV": {
+					PVCName:      "testPVC",
+					PVCNamespace: "velero",
+					PV: corev1api.PersistentVolume{
+						ObjectMeta: metav1.ObjectMeta{
+							Name:   "testPV",
+							Labels: map[string]string{"a": "b"},
+						},
+						Spec: corev1api.PersistentVolumeSpec{
+							PersistentVolumeReclaimPolicy: corev1api.PersistentVolumeReclaimDelete,
+						},
+					},
+				},
+			},
+			expectedVolumeInfos: []volume.VolumeInfo{
+				{
+					PVCName:       "testPVC",
+					PVCNamespace:  "velero",
+					PVName:        "testPV",
+					Skipped:       true,
+					SkippedReason: "CSI: skipped for PodVolumeBackup;",
+					PVInfo: volume.PVInfo{
+						ReclaimPolicy: "Delete",
+						Labels: map[string]string{
+							"a": "b",
+						},
+					},
+				},
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			request := new(backup.Request)
+			request.SkippedPVTracker = backup.NewSkipPVTracker()
+			if tc.skippedPVName != "" {
+				request.SkippedPVTracker.Track(tc.skippedPVName, "CSI", "skipped for PodVolumeBackup")
+			}
+			if tc.pvMap != nil {
+				request.PVMap = tc.pvMap
+			}
+			logger := logging.DefaultLogger(logrus.DebugLevel, logging.FormatJSON)
+
+			volumeInfos := generateVolumeInfoForSkippedPV(request, logger)
+			require.Equal(t, tc.expectedVolumeInfos, volumeInfos)
+		})
+	}
+}
+
+func TestGenerateVolumeInfoForCSIVolumeSnapshot(t *testing.T) {
+	resourceQuantity := resource.MustParse("100Gi")
+	now := metav1.Now()
+	tests := []struct {
+		name                  string
+		volumeSnapshot        snapshotv1api.VolumeSnapshot
+		volumeSnapshotContent snapshotv1api.VolumeSnapshotContent
+		volumeSnapshotClass   snapshotv1api.VolumeSnapshotClass
+		pvMap                 map[string]backup.PvcPvInfo
+		operation             *itemoperation.BackupOperation
+		expectedVolumeInfos   []volume.VolumeInfo
+	}{
+		{
+			name: "VS doesn't have VolumeSnapshotClass name",
+			volumeSnapshot: snapshotv1api.VolumeSnapshot{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "testVS",
+					Namespace: "velero",
+				},
+				Spec: snapshotv1api.VolumeSnapshotSpec{},
+			},
+			expectedVolumeInfos: []volume.VolumeInfo{},
+		},
+		{
+			name: "VS doesn't have status",
+			volumeSnapshot: snapshotv1api.VolumeSnapshot{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "testVS",
+					Namespace: "velero",
+				},
+				Spec: snapshotv1api.VolumeSnapshotSpec{
+					VolumeSnapshotClassName: stringPtr("testClass"),
+				},
+			},
+			expectedVolumeInfos: []volume.VolumeInfo{},
+		},
+		{
+			name: "VS doesn't have PVC",
+			volumeSnapshot: snapshotv1api.VolumeSnapshot{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "testVS",
+					Namespace: "velero",
+				},
+				Spec: snapshotv1api.VolumeSnapshotSpec{
+					VolumeSnapshotClassName: stringPtr("testClass"),
+				},
+				Status: &snapshotv1api.VolumeSnapshotStatus{
+					BoundVolumeSnapshotContentName: stringPtr("testContent"),
+				},
+			},
+			expectedVolumeInfos: []volume.VolumeInfo{},
+		},
+		{
+			name: "Cannot find VSC for VS",
+			volumeSnapshot: snapshotv1api.VolumeSnapshot{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "testVS",
+					Namespace: "velero",
+				},
+				Spec: snapshotv1api.VolumeSnapshotSpec{
+					VolumeSnapshotClassName: stringPtr("testClass"),
+					Source: snapshotv1api.VolumeSnapshotSource{
+						PersistentVolumeClaimName: stringPtr("testPVC"),
+					},
+				},
+				Status: &snapshotv1api.VolumeSnapshotStatus{
+					BoundVolumeSnapshotContentName: stringPtr("testContent"),
+				},
+			},
+			expectedVolumeInfos: []volume.VolumeInfo{},
+		},
+		{
+			name: "Cannot find VolumeInfo for PVC",
+			volumeSnapshot: snapshotv1api.VolumeSnapshot{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "testVS",
+					Namespace: "velero",
+				},
+				Spec: snapshotv1api.VolumeSnapshotSpec{
+					VolumeSnapshotClassName: stringPtr("testClass"),
+					Source: snapshotv1api.VolumeSnapshotSource{
+						PersistentVolumeClaimName: stringPtr("testPVC"),
+					},
+				},
+				Status: &snapshotv1api.VolumeSnapshotStatus{
+					BoundVolumeSnapshotContentName: stringPtr("testContent"),
+				},
+			},
+			volumeSnapshotClass:   *builder.ForVolumeSnapshotClass("testClass").Driver("pd.csi.storage.gke.io").Result(),
+			volumeSnapshotContent: *builder.ForVolumeSnapshotContent("testContent").Status(&snapshotv1api.VolumeSnapshotContentStatus{SnapshotHandle: stringPtr("testSnapshotHandle")}).Result(),
+			expectedVolumeInfos:   []volume.VolumeInfo{},
+		},
+		{
+			name: "Normal VolumeSnapshot case",
+			volumeSnapshot: snapshotv1api.VolumeSnapshot{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:              "testVS",
+					Namespace:         "velero",
+					CreationTimestamp: now,
+				},
+				Spec: snapshotv1api.VolumeSnapshotSpec{
+					VolumeSnapshotClassName: stringPtr("testClass"),
+					Source: snapshotv1api.VolumeSnapshotSource{
+						PersistentVolumeClaimName: stringPtr("testPVC"),
+					},
+				},
+				Status: &snapshotv1api.VolumeSnapshotStatus{
+					BoundVolumeSnapshotContentName: stringPtr("testContent"),
+					RestoreSize:                    &resourceQuantity,
+				},
+			},
+			volumeSnapshotClass:   *builder.ForVolumeSnapshotClass("testClass").Driver("pd.csi.storage.gke.io").Result(),
+			volumeSnapshotContent: *builder.ForVolumeSnapshotContent("testContent").Status(&snapshotv1api.VolumeSnapshotContentStatus{SnapshotHandle: stringPtr("testSnapshotHandle")}).Result(),
+			pvMap: map[string]backup.PvcPvInfo{
+				"velero/testPVC": {
+					PVCName:      "testPVC",
+					PVCNamespace: "velero",
+					PV: corev1api.PersistentVolume{
+						ObjectMeta: metav1.ObjectMeta{
+							Name:   "testPV",
+							Labels: map[string]string{"a": "b"},
+						},
+						Spec: corev1api.PersistentVolumeSpec{
+							PersistentVolumeReclaimPolicy: corev1api.PersistentVolumeReclaimDelete,
+						},
+					},
+				},
+			},
+			operation: &itemoperation.BackupOperation{
+				Spec: itemoperation.BackupOperationSpec{
+					OperationID: "testID",
+					ResourceIdentifier: velero.ResourceIdentifier{
+						GroupResource: schema.GroupResource{
+							Group:    "snapshot.storage.k8s.io",
+							Resource: "volumesnapshots",
+						},
+						Namespace: "velero",
+						Name:      "testVS",
+					},
+				},
+			},
+			expectedVolumeInfos: []volume.VolumeInfo{
+				{
+					PVCName:               "testPVC",
+					PVCNamespace:          "velero",
+					PVName:                "testPV",
+					BackupMethod:          volume.CSISnapshot,
+					OperationID:           "testID",
+					StartTimestamp:        &now,
+					PreserveLocalSnapshot: true,
+					CSISnapshotInfo: volume.CSISnapshotInfo{
+						Driver:         "pd.csi.storage.gke.io",
+						SnapshotHandle: "testSnapshotHandle",
+						Size:           107374182400,
+						VSCName:        "testContent",
+					},
+					PVInfo: volume.PVInfo{
+						ReclaimPolicy: "Delete",
+						Labels: map[string]string{
+							"a": "b",
+						},
+					},
+				},
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			request := new(backup.Request)
+			request.Backup = new(velerov1api.Backup)
+			if tc.pvMap != nil {
+				request.PVMap = tc.pvMap
+			}
+			operationList := request.GetItemOperationsList()
+			if tc.operation != nil {
+				*operationList = append(*operationList, tc.operation)
+			}
+			logger := logging.DefaultLogger(logrus.DebugLevel, logging.FormatJSON)
+
+			volumeInfos := generateVolumeInfoForCSIVolumeSnapshot(request, []snapshotv1api.VolumeSnapshot{tc.volumeSnapshot}, []snapshotv1api.VolumeSnapshotContent{tc.volumeSnapshotContent}, []snapshotv1api.VolumeSnapshotClass{tc.volumeSnapshotClass}, logger)
+			require.Equal(t, tc.expectedVolumeInfos, volumeInfos)
+		})
+	}
+
+}
+
+func TestGenerateVolumeInfoForVeleroNativeSnapshot(t *testing.T) {
+	tests := []struct {
+		name                string
+		nativeSnapshot      volume.Snapshot
+		pvMap               map[string]backup.PvcPvInfo
+		expectedVolumeInfos []volume.VolumeInfo
+	}{
+		{
+			name: "Native snapshot's IPOS pointer is nil",
+			nativeSnapshot: volume.Snapshot{
+				Spec: volume.SnapshotSpec{
+					PersistentVolumeName: "testPV",
+					VolumeIOPS:           nil,
+				},
+			},
+			expectedVolumeInfos: []volume.VolumeInfo{},
+		},
+		{
+			name: "Cannot find info for the PV",
+			nativeSnapshot: volume.Snapshot{
+				Spec: volume.SnapshotSpec{
+					PersistentVolumeName: "testPV",
+					VolumeIOPS:           int64Ptr(100),
+				},
+			},
+			expectedVolumeInfos: []volume.VolumeInfo{},
+		},
+		{
+			name: "Normal native snapshot",
+			pvMap: map[string]backup.PvcPvInfo{
+				"testPV": {
+					PVCName:      "testPVC",
+					PVCNamespace: "velero",
+					PV: corev1api.PersistentVolume{
+						ObjectMeta: metav1.ObjectMeta{
+							Name:   "testPV",
+							Labels: map[string]string{"a": "b"},
+						},
+						Spec: corev1api.PersistentVolumeSpec{
+							PersistentVolumeReclaimPolicy: corev1api.PersistentVolumeReclaimDelete,
+						},
+					},
+				},
+			},
+			nativeSnapshot: volume.Snapshot{
+				Spec: volume.SnapshotSpec{
+					PersistentVolumeName: "testPV",
+					VolumeIOPS:           int64Ptr(100),
+					VolumeType:           "ssd",
+					VolumeAZ:             "us-central1-a",
+				},
+				Status: volume.SnapshotStatus{
+					ProviderSnapshotID: "pvc-b31e3386-4bbb-4937-95d-7934cd62-b0a1-494b-95d7-0687440e8d0c",
+				},
+			},
+			expectedVolumeInfos: []volume.VolumeInfo{
+				{
+					PVCName:      "testPVC",
+					PVCNamespace: "velero",
+					PVName:       "testPV",
+					BackupMethod: volume.NativeSnapshot,
+					PVInfo: volume.PVInfo{
+						ReclaimPolicy: "Delete",
+						Labels: map[string]string{
+							"a": "b",
+						},
+					},
+					NativeSnapshotInfo: volume.NativeSnapshotInfo{
+						SnapshotHandle: "pvc-b31e3386-4bbb-4937-95d-7934cd62-b0a1-494b-95d7-0687440e8d0c",
+						VolumeType:     "ssd",
+						VolumeAZ:       "us-central1-a",
+						IOPS:           "100",
+					},
+				},
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			request := new(backup.Request)
+			request.VolumeSnapshots = append(request.VolumeSnapshots, &tc.nativeSnapshot)
+			if tc.pvMap != nil {
+				request.PVMap = tc.pvMap
+			}
+			logger := logging.DefaultLogger(logrus.DebugLevel, logging.FormatJSON)
+
+			volumeInfos := generateVolumeInfoForVeleroNativeSnapshot(request, logger)
+			require.Equal(t, tc.expectedVolumeInfos, volumeInfos)
+		})
+	}
+}
+
+func TestGenerateVolumeInfoFromPVB(t *testing.T) {
+	tests := []struct {
+		name                string
+		pvb                 *velerov1api.PodVolumeBackup
+		pod                 *corev1api.Pod
+		pvMap               map[string]backup.PvcPvInfo
+		expectedVolumeInfos []volume.VolumeInfo
+	}{
+		{
+			name:                "cannot find PVB's pod, should fail",
+			pvb:                 builder.ForPodVolumeBackup("velero", "testPVB").PodName("testPod").PodNamespace("velero").Result(),
+			expectedVolumeInfos: []volume.VolumeInfo{},
+		},
+		{
+			name: "PVB doesn't have a related PVC",
+			pvb:  builder.ForPodVolumeBackup("velero", "testPVB").PodName("testPod").PodNamespace("velero").Result(),
+			pod: builder.ForPod("velero", "testPod").Containers(&corev1api.Container{
+				Name: "test",
+				VolumeMounts: []corev1api.VolumeMount{
+					{
+						Name:      "testVolume",
+						MountPath: "/data",
+					},
+				},
+			}).Volumes(
+				&corev1api.Volume{
+					Name: "",
+					VolumeSource: corev1api.VolumeSource{
+						HostPath: &corev1api.HostPathVolumeSource{},
+					},
+				},
+			).Result(),
+			expectedVolumeInfos: []volume.VolumeInfo{
+				{
+					PVCName:      "",
+					PVCNamespace: "",
+					PVName:       "",
+					BackupMethod: volume.PodVolumeBackup,
+					PVBInfo: volume.PodVolumeBackupInfo{
+						PodName:      "testPod",
+						PodNamespace: "velero",
+					},
+				},
+			},
+		},
+		{
+			name: "Backup doesn't have information for PVC",
+			pvb:  builder.ForPodVolumeBackup("velero", "testPVB").PodName("testPod").PodNamespace("velero").Result(),
+			pod: builder.ForPod("velero", "testPod").Containers(&corev1api.Container{
+				Name: "test",
+				VolumeMounts: []corev1api.VolumeMount{
+					{
+						Name:      "testVolume",
+						MountPath: "/data",
+					},
+				},
+			}).Volumes(
+				&corev1api.Volume{
+					Name: "",
+					VolumeSource: corev1api.VolumeSource{
+						PersistentVolumeClaim: &corev1api.PersistentVolumeClaimVolumeSource{
+							ClaimName: "testPVC",
+						},
+					},
+				},
+			).Result(),
+			expectedVolumeInfos: []volume.VolumeInfo{},
+		},
+		{
+			name: "PVB's volume has a PVC",
+			pvMap: map[string]backup.PvcPvInfo{
+				"velero/testPVC": {
+					PVCName:      "testPVC",
+					PVCNamespace: "velero",
+					PV: corev1api.PersistentVolume{
+						ObjectMeta: metav1.ObjectMeta{
+							Name:   "testPV",
+							Labels: map[string]string{"a": "b"},
+						},
+						Spec: corev1api.PersistentVolumeSpec{
+							PersistentVolumeReclaimPolicy: corev1api.PersistentVolumeReclaimDelete,
+						},
+					},
+				},
+			},
+			pvb: builder.ForPodVolumeBackup("velero", "testPVB").PodName("testPod").PodNamespace("velero").Result(),
+			pod: builder.ForPod("velero", "testPod").Containers(&corev1api.Container{
+				Name: "test",
+				VolumeMounts: []corev1api.VolumeMount{
+					{
+						Name:      "testVolume",
+						MountPath: "/data",
+					},
+				},
+			}).Volumes(
+				&corev1api.Volume{
+					Name: "",
+					VolumeSource: corev1api.VolumeSource{
+						PersistentVolumeClaim: &corev1api.PersistentVolumeClaimVolumeSource{
+							ClaimName: "testPVC",
+						},
+					},
+				},
+			).Result(),
+			expectedVolumeInfos: []volume.VolumeInfo{
+				{
+					PVCName:      "testPVC",
+					PVCNamespace: "velero",
+					PVName:       "testPV",
+					BackupMethod: volume.PodVolumeBackup,
+					PVBInfo: volume.PodVolumeBackupInfo{
+						PodName:      "testPod",
+						PodNamespace: "velero",
+					},
+					PVInfo: volume.PVInfo{
+						ReclaimPolicy: string(corev1api.PersistentVolumeReclaimDelete),
+						Labels:        map[string]string{"a": "b"},
+					},
+				},
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			crClient := velerotest.NewFakeControllerRuntimeClient(t)
+			logger := logging.DefaultLogger(logrus.DebugLevel, logging.FormatJSON)
+			request := new(pkgbackup.Request)
+			request.PodVolumeBackups = append(request.PodVolumeBackups, tc.pvb)
+			if tc.pvMap != nil {
+				request.PVMap = tc.pvMap
+			}
+			if tc.pod != nil {
+				require.NoError(t, crClient.Create(context.TODO(), tc.pod))
+			}
+
+			volumeInfos := generateVolumeInfoFromPVB(request, crClient, logger)
+			require.Equal(t, tc.expectedVolumeInfos, volumeInfos)
+		})
+	}
+}
+
+func TestGenerateVolumeInfoFromDataUpload(t *testing.T) {
+	now := metav1.Now()
+	tests := []struct {
+		name                string
+		volumeSnapshotClass *snapshotv1api.VolumeSnapshotClass
+		dataUpload          *velerov2alpha1.DataUpload
+		operation           *itemoperation.BackupOperation
+		pvMap               map[string]backup.PvcPvInfo
+		expectedVolumeInfos []volume.VolumeInfo
+	}{
+		{
+			name: "Operation is not for PVC",
+			operation: &itemoperation.BackupOperation{
+				Spec: itemoperation.BackupOperationSpec{
+					ResourceIdentifier: velero.ResourceIdentifier{
+						GroupResource: schema.GroupResource{
+							Group:    "",
+							Resource: "configmaps",
+						},
+					},
+				},
+			},
+			expectedVolumeInfos: []volume.VolumeInfo{},
+		},
+		{
+			name: "Operation doesn't have DataUpload PostItemOperation",
+			operation: &itemoperation.BackupOperation{
+				Spec: itemoperation.BackupOperationSpec{
+					ResourceIdentifier: velero.ResourceIdentifier{
+						GroupResource: schema.GroupResource{
+							Group:    "",
+							Resource: "persistentvolumeclaims",
+						},
+						Namespace: "velero",
+						Name:      "testPVC",
+					},
+					PostOperationItems: []velero.ResourceIdentifier{
+						{
+							GroupResource: schema.GroupResource{
+								Group:    "",
+								Resource: "configmaps",
+							},
+						},
+					},
+				},
+			},
+			expectedVolumeInfos: []volume.VolumeInfo{},
+		},
+		{
+			name: "DataUpload cannot be found for operation",
+			operation: &itemoperation.BackupOperation{
+				Spec: itemoperation.BackupOperationSpec{
+					OperationID: "testOperation",
+					ResourceIdentifier: velero.ResourceIdentifier{
+						GroupResource: schema.GroupResource{
+							Group:    "",
+							Resource: "persistentvolumeclaims",
+						},
+						Namespace: "velero",
+						Name:      "testPVC",
+					},
+					PostOperationItems: []velero.ResourceIdentifier{
+						{
+							GroupResource: schema.GroupResource{
+								Group:    "velero.io",
+								Resource: "datauploads",
+							},
+							Namespace: "velero",
+							Name:      "testDU",
+						},
+					},
+				},
+			},
+			expectedVolumeInfos: []volume.VolumeInfo{},
+		},
+		{
+			name: "VolumeSnapshotClass cannot be found for operation",
+			dataUpload: builder.ForDataUpload("velero", "testDU").DataMover("velero").CSISnapshot(&velerov2alpha1.CSISnapshotSpec{
+				VolumeSnapshot: "testVS",
+			}).SnapshotID("testSnapshotHandle").Result(),
+			operation: &itemoperation.BackupOperation{
+				Spec: itemoperation.BackupOperationSpec{
+					OperationID: "testOperation",
+					ResourceIdentifier: velero.ResourceIdentifier{
+						GroupResource: schema.GroupResource{
+							Group:    "",
+							Resource: "persistentvolumeclaims",
+						},
+						Namespace: "velero",
+						Name:      "testPVC",
+					},
+					PostOperationItems: []velero.ResourceIdentifier{
+						{
+							GroupResource: schema.GroupResource{
+								Group:    "velero.io",
+								Resource: "datauploads",
+							},
+							Namespace: "velero",
+							Name:      "testDU",
+						},
+					},
+				},
+			},
+			pvMap: map[string]backup.PvcPvInfo{
+				"velero/testPVC": {
+					PVCName:      "testPVC",
+					PVCNamespace: "velero",
+					PV: corev1api.PersistentVolume{
+						ObjectMeta: metav1.ObjectMeta{
+							Name:   "testPV",
+							Labels: map[string]string{"a": "b"},
+						},
+						Spec: corev1api.PersistentVolumeSpec{
+							PersistentVolumeReclaimPolicy: corev1api.PersistentVolumeReclaimDelete,
+						},
+					},
+				},
+			},
+			expectedVolumeInfos: []volume.VolumeInfo{
+				{
+					PVCName:           "testPVC",
+					PVCNamespace:      "velero",
+					PVName:            "testPV",
+					BackupMethod:      volume.CSISnapshot,
+					SnapshotDataMoved: true,
+					OperationID:       "testOperation",
+					SnapshotDataMovementInfo: volume.SnapshotDataMovementInfo{
+						DataMover:    "velero",
+						UploaderType: "kopia",
+					},
+					PVInfo: volume.PVInfo{
+						ReclaimPolicy: string(corev1api.PersistentVolumeReclaimDelete),
+						Labels:        map[string]string{"a": "b"},
+					},
+				},
+			},
+		},
+		{
+			name: "Normal DataUpload case",
+			dataUpload: builder.ForDataUpload("velero", "testDU").DataMover("velero").CSISnapshot(&velerov2alpha1.CSISnapshotSpec{
+				VolumeSnapshot: "testVS",
+				SnapshotClass:  "testClass",
+			}).SnapshotID("testSnapshotHandle").Result(),
+			volumeSnapshotClass: builder.ForVolumeSnapshotClass("testClass").Driver("pd.csi.storage.gke.io").Result(),
+			operation: &itemoperation.BackupOperation{
+				Spec: itemoperation.BackupOperationSpec{
+					OperationID: "testOperation",
+					ResourceIdentifier: velero.ResourceIdentifier{
+						GroupResource: schema.GroupResource{
+							Group:    "",
+							Resource: "persistentvolumeclaims",
+						},
+						Namespace: "velero",
+						Name:      "testPVC",
+					},
+					PostOperationItems: []velero.ResourceIdentifier{
+						{
+							GroupResource: schema.GroupResource{
+								Group:    "velero.io",
+								Resource: "datauploads",
+							},
+							Namespace: "velero",
+							Name:      "testDU",
+						},
+					},
+				},
+				Status: itemoperation.OperationStatus{
+					Created: &now,
+				},
+			},
+			pvMap: map[string]backup.PvcPvInfo{
+				"velero/testPVC": {
+					PVCName:      "testPVC",
+					PVCNamespace: "velero",
+					PV: corev1api.PersistentVolume{
+						ObjectMeta: metav1.ObjectMeta{
+							Name:   "testPV",
+							Labels: map[string]string{"a": "b"},
+						},
+						Spec: corev1api.PersistentVolumeSpec{
+							PersistentVolumeReclaimPolicy: corev1api.PersistentVolumeReclaimDelete,
+						},
+					},
+				},
+			},
+			expectedVolumeInfos: []volume.VolumeInfo{
+				{
+					PVCName:           "testPVC",
+					PVCNamespace:      "velero",
+					PVName:            "testPV",
+					BackupMethod:      volume.CSISnapshot,
+					SnapshotDataMoved: true,
+					OperationID:       "testOperation",
+					StartTimestamp:    &now,
+					CSISnapshotInfo: volume.CSISnapshotInfo{
+						Driver: "pd.csi.storage.gke.io",
+					},
+					SnapshotDataMovementInfo: volume.SnapshotDataMovementInfo{
+						DataMover:    "velero",
+						UploaderType: "kopia",
+					},
+					PVInfo: volume.PVInfo{
+						ReclaimPolicy: string(corev1api.PersistentVolumeReclaimDelete),
+						Labels:        map[string]string{"a": "b"},
+					},
+				},
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			request := new(backup.Request)
+			operationList := request.GetItemOperationsList()
+			if tc.operation != nil {
+				*operationList = append(*operationList, tc.operation)
+			}
+			if tc.pvMap != nil {
+				request.PVMap = tc.pvMap
+			}
+			logger := logging.DefaultLogger(logrus.DebugLevel, logging.FormatJSON)
+
+			crClient := velerotest.NewFakeControllerRuntimeClient(t)
+			if tc.dataUpload != nil {
+				crClient.Create(context.TODO(), tc.dataUpload)
+			}
+
+			if tc.volumeSnapshotClass != nil {
+				crClient.Create(context.TODO(), tc.volumeSnapshotClass)
+			}
+
+			volumeInfos := generateVolumeInfoFromDataUpload(request, crClient, logger)
+			require.Equal(t, tc.expectedVolumeInfos, volumeInfos)
+		})
+	}
+}
+
+func int64Ptr(val int) *int64 {
+	i := int64(val)
+	return &i
+}
+
+func stringPtr(str string) *string {
+	return &str
+}
--- a/pkg/controller/backup_finalizer_controller.go
+++ b/pkg/controller/backup_finalizer_controller.go
@ -29,8 +29,6 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 	kbclient "sigs.k8s.io/controller-runtime/pkg/client"

-	snapshotv1listers "github.com/kubernetes-csi/external-snapshotter/client/v4/listers/volumesnapshot/v1"
-
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	pkgbackup "github.com/vmware-tanzu/velero/pkg/backup"
 	"github.com/vmware-tanzu/velero/pkg/metrics"
@ -42,21 +40,21 @@ import (

 // backupFinalizerReconciler reconciles a Backup object
 type backupFinalizerReconciler struct {
-	client               kbclient.Client
-	volumeSnapshotLister snapshotv1listers.VolumeSnapshotLister
-	clock                clocks.WithTickerAndDelayedExecution
-	backupper            pkgbackup.Backupper
-	newPluginManager     func(logrus.FieldLogger) clientmgmt.Manager
-	backupTracker        BackupTracker
-	metrics              *metrics.ServerMetrics
-	backupStoreGetter    persistence.ObjectBackupStoreGetter
-	log                  logrus.FieldLogger
+	client            kbclient.Client
+	globalCRClient    kbclient.Client
+	clock             clocks.WithTickerAndDelayedExecution
+	backupper         pkgbackup.Backupper
+	newPluginManager  func(logrus.FieldLogger) clientmgmt.Manager
+	backupTracker     BackupTracker
+	metrics           *metrics.ServerMetrics
+	backupStoreGetter persistence.ObjectBackupStoreGetter
+	log               logrus.FieldLogger
 }

 // NewBackupFinalizerReconciler initializes and returns backupFinalizerReconciler struct.
 func NewBackupFinalizerReconciler(
 	client kbclient.Client,
-	volumeSnapshotLister snapshotv1listers.VolumeSnapshotLister,
+	globalCRClient kbclient.Client,
 	clock clocks.WithTickerAndDelayedExecution,
 	backupper pkgbackup.Backupper,
 	newPluginManager func(logrus.FieldLogger) clientmgmt.Manager,
@ -67,6 +65,7 @@ func NewBackupFinalizerReconciler(
 ) *backupFinalizerReconciler {
 	return &backupFinalizerReconciler{
 		client:            client,
+		globalCRClient:    globalCRClient,
 		clock:             clock,
 		backupper:         backupper,
 		newPluginManager:  newPluginManager,
@ -191,7 +190,7 @@ func (r *backupFinalizerReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 	backup.Status.CompletionTimestamp = &metav1.Time{Time: r.clock.Now()}
 	recordBackupMetrics(log, backup, outBackupFile, r.metrics, true)

-	pkgbackup.UpdateBackupCSISnapshotsStatus(r.client, r.volumeSnapshotLister, backup, log)
+	pkgbackup.UpdateBackupCSISnapshotsStatus(r.client, r.globalCRClient, backup, log)
 	// update backup metadata in object store
 	backupJSON := new(bytes.Buffer)
 	if err := encode.To(backup, "json", backupJSON); err != nil {
--- a/pkg/controller/backup_finalizer_controller_test.go
+++ b/pkg/controller/backup_finalizer_controller_test.go
@ -23,7 +23,6 @@ import (
 	"testing"
 	"time"

-	snapshotv1listers "github.com/kubernetes-csi/external-snapshotter/client/v4/listers/volumesnapshot/v1"
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
@ -44,14 +43,13 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/plugin/framework"
 	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
-	velerotestmocks "github.com/vmware-tanzu/velero/pkg/test/mocks"
 )

-func mockBackupFinalizerReconciler(fakeClient kbclient.Client, fakeVolumeSnapshotLister snapshotv1listers.VolumeSnapshotLister, fakeClock *testclocks.FakeClock) (*backupFinalizerReconciler, *fakeBackupper) {
+func mockBackupFinalizerReconciler(fakeClient kbclient.Client, fakeGlobalClient kbclient.Client, fakeClock *testclocks.FakeClock) (*backupFinalizerReconciler, *fakeBackupper) {
 	backupper := new(fakeBackupper)
 	return NewBackupFinalizerReconciler(
 		fakeClient,
-		fakeVolumeSnapshotLister,
+		fakeGlobalClient,
 		fakeClock,
 		backupper,
 		func(logrus.FieldLogger) clientmgmt.Manager { return pluginManager },
@ -164,9 +162,9 @@ func TestBackupFinalizerReconcile(t *testing.T) {

 			fakeClient := velerotest.NewFakeControllerRuntimeClient(t, initObjs...)

-			fakeVolumeSnapshotLister := velerotestmocks.NewVolumeSnapshotLister(t)
+			fakeGlobalClient := velerotest.NewFakeControllerRuntimeClient(t, initObjs...)

-			reconciler, backupper := mockBackupFinalizerReconciler(fakeClient, fakeVolumeSnapshotLister, fakeClock)
+			reconciler, backupper := mockBackupFinalizerReconciler(fakeClient, fakeGlobalClient, fakeClock)
 			pluginManager.On("CleanupClients").Return(nil)
 			backupStore.On("GetBackupItemOperations", test.backup.Name).Return(test.backupOperations, nil)
 			backupStore.On("GetBackupContents", mock.Anything).Return(io.NopCloser(bytes.NewReader([]byte("hello world"))), nil)
--- a/pkg/persistence/object_store.go
+++ b/pkg/persistence/object_store.go
@ -608,6 +608,8 @@ func (s *objectBackupStore) GetDownloadURL(target velerov1api.DownloadTarget) (s
 		return s.objectStore.CreateSignedURL(s.bucket, s.layout.getCSIVolumeSnapshotContentsKey(target.Name), DownloadURLTTL)
 	case velerov1api.DownloadTargetKindBackupResults:
 		return s.objectStore.CreateSignedURL(s.bucket, s.layout.getBackupResultsKey(target.Name), DownloadURLTTL)
+	case velerov1api.DownloadTargetKindBackupVolumeInfos:
+		return s.objectStore.CreateSignedURL(s.bucket, s.layout.getBackupVolumeInfoKey(target.Name), DownloadURLTTL)
 	default:
 		return "", errors.Errorf("unsupported download target kind %q", target.Kind)
 	}
--- a/pkg/persistence/object_store_test.go
+++ b/pkg/persistence/object_store_test.go
@ -768,6 +768,13 @@ func TestGetDownloadURL(t *testing.T) {
 				velerov1api.DownloadTargetKindRestoreResourceList:   "restores/b-cool-20170913154901-20170913154902/restore-b-cool-20170913154901-20170913154902-resource-list.json.gz",
 			},
 		},
+		{
+			name:       "",
+			targetName: "my-backup",
+			expectedKeyByKind: map[velerov1api.DownloadTargetKind]string{
+				velerov1api.DownloadTargetKindBackupVolumeInfos: "backups/my-backup/my-backup-volumeinfos.json.gz",
+			},
+		},
 	}

 	for _, test := range tests {
--- a/pkg/volume/volume_info_common.go
+++ b/pkg/volume/volume_info_common.go
@ -51,6 +51,8 @@ type VolumeInfo struct {
 	SnapshotDataMoved bool `json:"snapshotDataMoved"`

 	// Whether the local snapshot is preserved after snapshot is moved.
+	// The local snapshot may be a result of CSI snapshot backup(no data movement)
+	// or a CSI snapshot data movement plus preserve local snapshot.
 	PreserveLocalSnapshot bool `json:"preserveLocalSnapshot"`

 	// Whether the Volume is skipped in this backup.
@ -69,6 +71,7 @@ type VolumeInfo struct {
 	SnapshotDataMovementInfo SnapshotDataMovementInfo `json:"snapshotDataMovementInfo,omitempty"`
 	NativeSnapshotInfo       NativeSnapshotInfo       `json:"nativeSnapshotInfo,omitempty"`
 	PVBInfo                  PodVolumeBackupInfo      `json:"pvbInfo,omitempty"`
+	PVInfo                   PVInfo                   `json:"pvInfo,omitempty"`
 }

 // CSISnapshotInfo is used for displaying the CSI snapshot status
@ -76,7 +79,7 @@ type CSISnapshotInfo struct {
 	// It's the storage provider's snapshot ID for CSI.
 	SnapshotHandle string `json:"snapshotHandle"`

-	// The snapshot corresponding volume size. Some of the volume backup methods cannot retrieve the data by current design, for example, the Velero native snapshot.
+	// The snapshot corresponding volume size.
 	Size int64 `json:"size"`

 	// The name of the CSI driver.
@ -91,7 +94,7 @@ type SnapshotDataMovementInfo struct {
 	// The data mover used by the backup. The valid values are `velero` and ``(equals to `velero`).
 	DataMover string `json:"dataMover"`

-	// The type of the uploader that uploads the snapshot data. The valid values are `kopia` and `restic`. It's useful for file-system backup and snapshot data mover.
+	// The type of the uploader that uploads the snapshot data. The valid values are `kopia` and `restic`.
 	UploaderType string `json:"uploaderType"`

 	// The name or ID of the snapshot associated object(SAO).
@ -111,9 +114,6 @@ type NativeSnapshotInfo struct {
 	// It's the storage provider's snapshot ID for the Velero-native snapshot.
 	SnapshotHandle string `json:"snapshotHandle"`

-	// The snapshot corresponding volume size. Some of the volume backup methods cannot retrieve the data by current design, for example, the Velero native snapshot.
-	Size int64 `json:"size"`
-
 	// The cloud provider snapshot volume type.
 	VolumeType string `json:"volumeType"`

@ -129,19 +129,32 @@ type PodVolumeBackupInfo struct {
 	// It's the file-system uploader's snapshot ID for PodVolumeBackup.
 	SnapshotHandle string `json:"snapshotHandle"`

-	// The snapshot corresponding volume size. Some of the volume backup methods cannot retrieve the data by current design, for example, the Velero native snapshot.
+	// The snapshot corresponding volume size.
 	Size int64 `json:"size"`

-	// The type of the uploader that uploads the data. The valid values are `kopia` and `restic`. It's useful for file-system backup and snapshot data mover.
+	// The type of the uploader that uploads the data. The valid values are `kopia` and `restic`.
 	UploaderType string `json:"uploaderType"`

 	// The PVC's corresponding volume name used by Pod
 	// https://github.com/kubernetes/kubernetes/blob/e4b74dd12fa8cb63c174091d5536a10b8ec19d34/pkg/apis/core/types.go#L48
 	VolumeName string `json:"volumeName"`

-	// The Pod name mounting this PVC. The format should be <namespace-name>/<pod-name>.
+	// The Pod name mounting this PVC.
 	PodName string `json:"podName"`

+	// The Pod namespace
+	PodNamespace string `json:"podNamespace"`
+
 	// The PVB-taken k8s node's name.
 	NodeName string `json:"nodeName"`
 }
+
+// PVInfo is used to store some PV information modified after creation.
+// Those information are lost after PV recreation.
+type PVInfo struct {
+	// ReclaimPolicy of PV. It could be different from the referenced StorageClass.
+	ReclaimPolicy string `json:"reclaimPolicy"`
+
+	// The PV's labels should be kept after recreation.
+	Labels map[string]string `json:"labels"`
+}