Merge pull request #7125 from Lyndon-Li/issue-fix-6695

Issue fix 6695: add describe for data mover backups

changelogs/unreleased/7125-Lyndon-Li

@@ -0,0 +1 @@
+Fix issue #6695, add describe for data mover backups

pkg/cmd/cli/backup/describe.go

@@ -21,8 +21,6 @@ import (
 	"fmt"
 	"os"
 
-	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
-	snapshotv1client "github.com/kubernetes-csi/external-snapshotter/client/v4/clientset/versioned"
 	"github.com/spf13/cobra"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
@@ -32,7 +30,6 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/client"
 	"github.com/vmware-tanzu/velero/pkg/cmd"
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/output"
-	"github.com/vmware-tanzu/velero/pkg/features"
 	"github.com/vmware-tanzu/velero/pkg/label"
 )
 
@@ -57,14 +54,6 @@ func NewDescribeCommand(f client.Factory, use string) *cobra.Command {
 			kbClient, err := f.KubebuilderClient()
 			cmd.CheckError(err)
 
-			var csiClient *snapshotv1client.Clientset
-			if features.IsEnabled(velerov1api.CSIFeatureFlag) {
-				clientConfig, err := f.ClientConfig()
-				cmd.CheckError(err)
-				csiClient, err = snapshotv1client.NewForConfig(clientConfig)
-				cmd.CheckError(err)
-			}
-
 			if outputFormat != "plaintext" && outputFormat != "json" {
 				cmd.CheckError(fmt.Errorf("invalid output format '%s'. valid value are 'plaintext, json'", outputFormat))
 			}
@@ -104,23 +93,13 @@ func NewDescribeCommand(f client.Factory, use string) *cobra.Command {
 					fmt.Fprintf(os.Stderr, "error getting PodVolumeBackups for backup %s: %v\n", backup.Name, err)
 				}
 
-				// declare vscList up here since it may be empty and we'll pass the empty Items field into DescribeBackup
-				vscList := new(snapshotv1api.VolumeSnapshotContentList)
-				if features.IsEnabled(velerov1api.CSIFeatureFlag) {
-					opts := label.NewListOptionsForBackup(backup.Name)
-					vscList, err = csiClient.SnapshotV1().VolumeSnapshotContents().List(context.TODO(), opts)
-					if err != nil {
-						fmt.Fprintf(os.Stderr, "error getting VolumeSnapshotContent objects for backup %s: %v\n", backup.Name, err)
-					}
-				}
-
 				// structured output only applies to a single backup in case of OOM
 				// To describe the list of backups in structured format, users could iterate over the list and describe backup one after another.
 				if len(backups.Items) == 1 && outputFormat != "plaintext" {
-					s := output.DescribeBackupInSF(context.Background(), kbClient, &backups.Items[i], deleteRequestList.Items, podVolumeBackupList.Items, vscList.Items, details, insecureSkipTLSVerify, caCertFile, outputFormat)
+					s := output.DescribeBackupInSF(context.Background(), kbClient, &backups.Items[i], deleteRequestList.Items, podVolumeBackupList.Items, details, insecureSkipTLSVerify, caCertFile, outputFormat)
 					fmt.Print(s)
 				} else {
-					s := output.DescribeBackup(context.Background(), kbClient, &backups.Items[i], deleteRequestList.Items, podVolumeBackupList.Items, vscList.Items, details, insecureSkipTLSVerify, caCertFile)
+					s := output.DescribeBackup(context.Background(), kbClient, &backups.Items[i], deleteRequestList.Items, podVolumeBackupList.Items, details, insecureSkipTLSVerify, caCertFile)
 					if first {
 						first = false
 						fmt.Print(s)

pkg/cmd/cli/backup/describe_test.go

@@ -39,7 +39,7 @@ func TestNewDescribeCommand(t *testing.T) {
 	// create a factory
 	f := &factorymocks.Factory{}
 	backupName := "bk-describe-1"
-	testBackup := builder.ForBackup(cmdtest.VeleroNameSpace, backupName).Result()
+	testBackup := builder.ForBackup(cmdtest.VeleroNameSpace, backupName).SnapshotVolumes(false).Result()
 
 	clientConfig := rest.Config{}
 	kbClient := test.NewFakeControllerRuntimeClient(t)
@@ -68,7 +68,7 @@ func TestNewDescribeCommand(t *testing.T) {
 	stdout, _, err := veleroexec.RunCommand(cmd)
 
 	if err == nil {
-		assert.Contains(t, stdout, "Velero-Native Snapshots: <none included>")
+		assert.Contains(t, stdout, "Backup Volumes: <none included>")
 		assert.Contains(t, stdout, "Or label selector:  <none>")
 		assert.Contains(t, stdout, fmt.Sprintf("Name:         %s", backupName))
 		return

pkg/cmd/util/output/backup_describer.go

@@ -22,12 +22,14 @@ import (
 	"encoding/json"
 	"fmt"
 	"sort"
+	"strconv"
 	"strings"
 
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
+	"github.com/pkg/errors"
 
 	"github.com/fatih/color"
 	kbclient "sigs.k8s.io/controller-runtime/pkg/client"
@@ -38,9 +40,11 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/features"
 	"github.com/vmware-tanzu/velero/pkg/itemoperation"
 
+	"github.com/vmware-tanzu/velero/internal/volume"
+	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
 	"github.com/vmware-tanzu/velero/pkg/util/collections"
 	"github.com/vmware-tanzu/velero/pkg/util/results"
-	"github.com/vmware-tanzu/velero/pkg/volume"
+	nativesnap "github.com/vmware-tanzu/velero/pkg/volume"
 )
 
 // DescribeBackup describes a backup in human-readable format.
@@ -50,7 +54,6 @@ func DescribeBackup(
 	backup *velerov1api.Backup,
 	deleteRequests []velerov1api.DeleteBackupRequest,
 	podVolumeBackups []velerov1api.PodVolumeBackup,
-	volumeSnapshotContents []snapshotv1api.VolumeSnapshotContent,
 	details bool,
 	insecureSkipTLSVerify bool,
 	caCertFile string,
@@ -109,22 +112,12 @@ func DescribeBackup(
 		DescribeBackupSpec(d, backup.Spec)
 
 		d.Println()
-		DescribeBackupStatus(ctx, kbClient, d, backup, details, insecureSkipTLSVerify, caCertFile)
+		DescribeBackupStatus(ctx, kbClient, d, backup, details, insecureSkipTLSVerify, caCertFile, podVolumeBackups)
 
 		if len(deleteRequests) > 0 {
 			d.Println()
 			DescribeDeleteBackupRequests(d, deleteRequests)
 		}
-
-		if features.IsEnabled(velerov1api.CSIFeatureFlag) {
-			d.Println()
-			DescribeCSIVolumeSnapshots(d, details, volumeSnapshotContents)
-		}
-
-		if len(podVolumeBackups) > 0 {
-			d.Println()
-			DescribePodVolumeBackups(d, podVolumeBackups, details)
-		}
 	})
 }
 
@@ -321,7 +314,8 @@ func DescribeBackupSpec(d *Describer, spec velerov1api.BackupSpec) {
 }
 
 // DescribeBackupStatus describes a backup status in human-readable format.
-func DescribeBackupStatus(ctx context.Context, kbClient kbclient.Client, d *Describer, backup *velerov1api.Backup, details bool, insecureSkipTLSVerify bool, caCertPath string) {
+func DescribeBackupStatus(ctx context.Context, kbClient kbclient.Client, d *Describer, backup *velerov1api.Backup, details bool,
+	insecureSkipTLSVerify bool, caCertPath string, podVolumeBackups []velerov1api.PodVolumeBackup) {
 	status := backup.Status
 
 	// Status.Version has been deprecated, use Status.FormatVersion
@@ -366,32 +360,9 @@ func DescribeBackupStatus(ctx context.Context, kbClient kbclient.Client, d *Desc
 		d.Println()
 	}
 
-	if status.VolumeSnapshotsAttempted > 0 {
-		if !details {
-			d.Printf("Velero-Native Snapshots:\t%d of %d snapshots completed successfully (specify --details for more information)\n", status.VolumeSnapshotsCompleted, status.VolumeSnapshotsAttempted)
-			return
-		}
+	describeBackupVolumes(ctx, kbClient, d, backup, details, insecureSkipTLSVerify, caCertPath, podVolumeBackups)
 
-		buf := new(bytes.Buffer)
-		if err := downloadrequest.Stream(ctx, kbClient, backup.Namespace, backup.Name, velerov1api.DownloadTargetKindBackupVolumeSnapshots, buf, downloadRequestTimeout, insecureSkipTLSVerify, caCertPath); err != nil {
-			d.Printf("Velero-Native Snapshots:\t<error getting snapshot info: %v>\n", err)
-			return
-		}
-
-		var snapshots []*volume.Snapshot
-		if err := json.NewDecoder(buf).Decode(&snapshots); err != nil {
-			d.Printf("Velero-Native Snapshots:\t<error reading snapshot info: %v>\n", err)
-			return
-		}
-
-		d.Printf("Velero-Native Snapshots:\n")
-		for _, snap := range snapshots {
-			describeSnapshot(d, snap.Spec.PersistentVolumeName, snap.Status.ProviderSnapshotID, snap.Spec.VolumeType, snap.Spec.VolumeAZ, snap.Spec.VolumeIOPS)
-		}
-		return
-	}
-
-	d.Printf("Velero-Native Snapshots: <none included>\n")
+	d.Println()
 
 	if status.HookStatus != nil {
 		d.Println()
@@ -463,16 +434,259 @@ func describeBackupResourceList(ctx context.Context, kbClient kbclient.Client, d
 	}
 }
 
-func describeSnapshot(d *Describer, pvName, snapshotID, volumeType, volumeAZ string, iops *int64) {
-	d.Printf("\t%s:\n", pvName)
-	d.Printf("\t\tSnapshot ID:\t%s\n", snapshotID)
-	d.Printf("\t\tType:\t%s\n", volumeType)
-	d.Printf("\t\tAvailability Zone:\t%s\n", volumeAZ)
-	iopsString := "<N/A>"
-	if iops != nil {
-		iopsString = fmt.Sprintf("%d", *iops)
+func describeBackupVolumes(ctx context.Context, kbClient kbclient.Client, d *Describer, backup *velerov1api.Backup, details bool,
+	insecureSkipTLSVerify bool, caCertPath string, podVolumeBackupCRs []velerov1api.PodVolumeBackup) {
+	if boolptr.IsSetToFalse(backup.Spec.SnapshotVolumes) {
+		d.Println("Backup Volumes: <none included>")
+		return
+	}
+
+	d.Println("Backup Volumes:")
+
+	nativeSnapshots := []*volume.VolumeInfo{}
+	csiSnapshots := []*volume.VolumeInfo{}
+
+	buf := new(bytes.Buffer)
+	err := downloadrequest.Stream(ctx, kbClient, backup.Namespace, backup.Name, velerov1api.DownloadTargetKindBackupVolumeInfos, buf, downloadRequestTimeout, insecureSkipTLSVerify, caCertPath)
+	if err == downloadrequest.ErrNotFound {
+		nativeSnapshots, err = retrieveNativeSnapshotLegacy(ctx, kbClient, backup, insecureSkipTLSVerify, caCertPath)
+		if err != nil {
+			d.Printf("\t<error concluding native snapshot info: %v>\n", err)
+			return
+		}
+
+		csiSnapshots, err = retrieveCSISnapshotLegacy(ctx, kbClient, backup, insecureSkipTLSVerify, caCertPath)
+		if err != nil {
+			d.Printf("\t<error concluding CSI snapshot info: %v>\n", err)
+			return
+		}
+	} else if err != nil {
+		d.Printf("\t<error getting backup volume info: %v>\n", err)
+		return
+	} else {
+		var volumeInfos []volume.VolumeInfo
+		if err := json.NewDecoder(buf).Decode(&volumeInfos); err != nil {
+			d.Printf("\t<error reading backup volume info: %v>\n", err)
+			return
+		}
+
+		for i := range volumeInfos {
+			switch volumeInfos[i].BackupMethod {
+			case volume.NativeSnapshot:
+				nativeSnapshots = append(nativeSnapshots, &volumeInfos[i])
+			case volume.CSISnapshot:
+				csiSnapshots = append(csiSnapshots, &volumeInfos[i])
+			}
+		}
+	}
+
+	describeNativeSnapshots(d, details, nativeSnapshots)
+	d.Println()
+
+	describeCSISnapshots(d, details, csiSnapshots)
+	d.Println()
+
+	describePodVolumeBackups(d, details, podVolumeBackupCRs)
+}
+
+func retrieveNativeSnapshotLegacy(ctx context.Context, kbClient kbclient.Client, backup *velerov1api.Backup, insecureSkipTLSVerify bool, caCertPath string) ([]*volume.VolumeInfo, error) {
+	status := backup.Status
+	nativeSnapshots := []*volume.VolumeInfo{}
+
+	if status.VolumeSnapshotsAttempted == 0 {
+		return nativeSnapshots, nil
+	}
+
+	buf := new(bytes.Buffer)
+	if err := downloadrequest.Stream(ctx, kbClient, backup.Namespace, backup.Name, velerov1api.DownloadTargetKindBackupVolumeSnapshots, buf, downloadRequestTimeout, insecureSkipTLSVerify, caCertPath); err != nil {
+		return nativeSnapshots, errors.Wrapf(err, "error to download native snapshot info")
+	}
+
+	var snapshots []*nativesnap.Snapshot
+	if err := json.NewDecoder(buf).Decode(&snapshots); err != nil {
+		return nativeSnapshots, errors.Wrapf(err, "error to decode native snapshot info")
+	}
+
+	for _, snap := range snapshots {
+		volumeInfo := volume.VolumeInfo{
+			PVName: snap.Spec.PersistentVolumeName,
+			NativeSnapshotInfo: volume.NativeSnapshotInfo{
+				SnapshotHandle: snap.Status.ProviderSnapshotID,
+				VolumeType:     snap.Spec.VolumeType,
+				VolumeAZ:       snap.Spec.VolumeAZ,
+			},
+		}
+
+		if snap.Spec.VolumeIOPS != nil {
+			volumeInfo.NativeSnapshotInfo.IOPS = strconv.FormatInt(*snap.Spec.VolumeIOPS, 10)
+		}
+
+		nativeSnapshots = append(nativeSnapshots, &volumeInfo)
+	}
+
+	return nativeSnapshots, nil
+}
+
+func retrieveCSISnapshotLegacy(ctx context.Context, kbClient kbclient.Client, backup *velerov1api.Backup, insecureSkipTLSVerify bool, caCertPath string) ([]*volume.VolumeInfo, error) {
+	status := backup.Status
+	csiSnapshots := []*volume.VolumeInfo{}
+
+	if !features.IsEnabled(velerov1api.CSIFeatureFlag) {
+		return csiSnapshots, nil
+	}
+
+	if status.CSIVolumeSnapshotsAttempted == 0 {
+		return csiSnapshots, nil
+	}
+
+	vsBuf := new(bytes.Buffer)
+	err := downloadrequest.Stream(ctx, kbClient, backup.Namespace, backup.Name, velerov1api.DownloadTargetKindCSIBackupVolumeSnapshots, vsBuf, downloadRequestTimeout, insecureSkipTLSVerify, caCertPath)
+	if err != nil {
+		return csiSnapshots, errors.Wrapf(err, "error to download vs list")
+	}
+
+	var vsList []snapshotv1api.VolumeSnapshot
+	if err := json.NewDecoder(vsBuf).Decode(&vsList); err != nil {
+		return csiSnapshots, errors.Wrapf(err, "error to decode vs list")
+	}
+
+	vscBuf := new(bytes.Buffer)
+	err = downloadrequest.Stream(ctx, kbClient, backup.Namespace, backup.Name, velerov1api.DownloadTargetKindCSIBackupVolumeSnapshotContents, vscBuf, downloadRequestTimeout, insecureSkipTLSVerify, caCertPath)
+	if err != nil {
+		return csiSnapshots, errors.Wrapf(err, "error to download vsc list")
+	}
+
+	var vscList []snapshotv1api.VolumeSnapshotContent
+	if err := json.NewDecoder(vscBuf).Decode(&vscList); err != nil {
+		return csiSnapshots, errors.Wrapf(err, "error to decode vsc list")
+	}
+
+	for _, vsc := range vscList {
+		volInfo := volume.VolumeInfo{
+			PreserveLocalSnapshot: true,
+			CSISnapshotInfo: volume.CSISnapshotInfo{
+				VSCName: vsc.Name,
+				Driver:  vsc.Spec.Driver,
+			},
+		}
+
+		if vsc.Status != nil && vsc.Status.SnapshotHandle != nil {
+			volInfo.CSISnapshotInfo.SnapshotHandle = *vsc.Status.SnapshotHandle
+		}
+
+		if vsc.Status != nil && vsc.Status.RestoreSize != nil {
+			volInfo.CSISnapshotInfo.Size = *vsc.Status.RestoreSize
+		}
+
+		for _, vs := range vsList {
+			if vs.Status.BoundVolumeSnapshotContentName == nil {
+				continue
+			}
+
+			if vs.Spec.Source.PersistentVolumeClaimName == nil {
+				continue
+			}
+
+			if *vs.Status.BoundVolumeSnapshotContentName == vsc.Name {
+				volInfo.PVCName = *vs.Spec.Source.PersistentVolumeClaimName
+			}
+		}
+
+		if volInfo.PVCName == "" {
+			volInfo.PVCName = "<PVC name not found>"
+		}
+
+		csiSnapshots = append(csiSnapshots, &volInfo)
+	}
+
+	return csiSnapshots, nil
+}
+
+func describeNativeSnapshots(d *Describer, details bool, infos []*volume.VolumeInfo) {
+	if len(infos) == 0 {
+		d.Printf("\tVelero-Native Snapshots: <none included>\n")
+		return
+	}
+
+	d.Println("\tVelero-Native Snapshots:")
+	for _, info := range infos {
+		describNativeSnapshot(d, details, info)
+	}
+}
+
+func describNativeSnapshot(d *Describer, details bool, info *volume.VolumeInfo) {
+	if details {
+		d.Printf("\t\t%s:\n", info.PVName)
+		d.Printf("\t\t\tSnapshot ID:\t%s\n", info.NativeSnapshotInfo.SnapshotHandle)
+		d.Printf("\t\t\tType:\t%s\n", info.NativeSnapshotInfo.VolumeType)
+		d.Printf("\t\t\tAvailability Zone:\t%s\n", info.NativeSnapshotInfo.VolumeAZ)
+		d.Printf("\t\t\tIOPS:\t%s\n", info.NativeSnapshotInfo.IOPS)
+	} else {
+		d.Printf("\t\t%s: specify --details for more information\n", info.PVName)
+	}
+}
+
+func describeCSISnapshots(d *Describer, details bool, infos []*volume.VolumeInfo) {
+	if !features.IsEnabled(velerov1api.CSIFeatureFlag) {
+		return
+	}
+
+	if len(infos) == 0 {
+		d.Printf("\tCSI Snapshots: <none included>\n")
+		return
+	}
+
+	d.Println("\tCSI Snapshots:")
+	for _, info := range infos {
+		describeCSISnapshot(d, details, info)
+	}
+}
+
+func describeCSISnapshot(d *Describer, details bool, info *volume.VolumeInfo) {
+	d.Printf("\t\t%s:\n", info.PVCName)
+
+	describeLocalSnapshot(d, details, info)
+	describeDataMovement(d, details, info)
+}
+
+func describeLocalSnapshot(d *Describer, details bool, info *volume.VolumeInfo) {
+	if !info.PreserveLocalSnapshot {
+		return
+	}
+
+	if details {
+		d.Printf("\t\t\tSnapshot:\n")
+		if !info.SnapshotDataMoved && info.OperationID != "" {
+			d.Printf("\t\t\t\tOperation ID: %s\n", info.OperationID)
+		}
+
+		d.Printf("\t\t\t\tSnapshot Content Name: %s\n", info.CSISnapshotInfo.VSCName)
+		d.Printf("\t\t\t\tStorage Snapshot ID: %s\n", info.CSISnapshotInfo.SnapshotHandle)
+		d.Printf("\t\t\t\tSnapshot Size (bytes): %d\n", info.CSISnapshotInfo.Size)
+		d.Printf("\t\t\t\tCSI Driver: %s\n", info.CSISnapshotInfo.Driver)
+	} else {
+		d.Printf("\t\t\tSnapshot: %s\n", "included, specify --details for more information")
+	}
+}
+
+func describeDataMovement(d *Describer, details bool, info *volume.VolumeInfo) {
+	if !info.SnapshotDataMoved {
+		return
+	}
+
+	if details {
+		d.Printf("\t\t\tData Movement:\n")
+		d.Printf("\t\t\t\tOperation ID: %s\n", info.OperationID)
+
+		dataMover := "velero"
+		if info.SnapshotDataMovementInfo.DataMover != "" {
+			dataMover = info.SnapshotDataMovementInfo.DataMover
+		}
+		d.Printf("\t\t\t\tData Mover: %s\n", dataMover)
+		d.Printf("\t\t\t\tUploader Type: %s\n", info.SnapshotDataMovementInfo.UploaderType)
+	} else {
+		d.Printf("\t\t\tData Movement: %s\n", "included, specify --details for more information")
 	}
-	d.Printf("\t\tIOPS:\t%s\n", iopsString)
 }
 
 func describeBackupItemOperation(d *Describer, operation *itemoperation.BackupOperation) {
@@ -545,25 +759,26 @@ func failedDeletionCount(requests []velerov1api.DeleteBackupRequest) int {
 	return count
 }
 
-// DescribePodVolumeBackups describes pod volume backups in human-readable format.
-func DescribePodVolumeBackups(d *Describer, backups []velerov1api.PodVolumeBackup, details bool) {
+// describePodVolumeBackups describes pod volume backups in human-readable format.
+func describePodVolumeBackups(d *Describer, details bool, podVolumeBackups []velerov1api.PodVolumeBackup) {
 	// Get the type of pod volume uploader. Since the uploader only comes from a single source, we can
 	// take the uploader type from the first element of the array.
 	var uploaderType string
-	if len(backups) > 0 {
-		uploaderType = backups[0].Spec.UploaderType
+	if len(podVolumeBackups) > 0 {
+		uploaderType = podVolumeBackups[0].Spec.UploaderType
 	} else {
+		d.Printf("\tPod Volume Backups: <none included>\n")
 		return
 	}
 
 	if details {
-		d.Printf("%s Backups:\n", uploaderType)
+		d.Printf("\tPod Volume Backups - %s:\n", uploaderType)
 	} else {
-		d.Printf("%s Backups (specify --details for more information):\n", uploaderType)
+		d.Printf("\tPod Volume Backups - %s (specify --details for more information):\n", uploaderType)
 	}
 
 	// separate backups by phase (combining <none> and New into a single group)
-	backupsByPhase := groupByPhase(backups)
+	backupsByPhase := groupByPhase(podVolumeBackups)
 
 	// go through phases in a specific order
 	for _, phase := range []string{
@@ -578,7 +793,7 @@ func DescribePodVolumeBackups(d *Describer, backups []velerov1api.PodVolumeBacku
 
 		// if we're not printing details, just report the phase and count
 		if !details {
-			d.Printf("\t%s:\t%d\n", phase, len(backupsByPhase[phase]))
+			d.Printf("\t\t%s:\t%d\n", phase, len(backupsByPhase[phase]))
 			continue
 		}
 
@@ -589,12 +804,12 @@ func DescribePodVolumeBackups(d *Describer, backups []velerov1api.PodVolumeBacku
 			backupsByPod.Add(backup.Spec.Pod.Namespace, backup.Spec.Pod.Name, backup.Spec.Volume, phase, backup.Status.Progress)
 		}
 
-		d.Printf("\t%s:\n", phase)
+		d.Printf("\t\t%s:\n", phase)
 		for _, backupGroup := range backupsByPod.Sorted() {
 			sort.Strings(backupGroup.volumes)
 
 			// print volumes backed up for this pod
-			d.Printf("\t\t%s: %s\n", backupGroup.label, strings.Join(backupGroup.volumes, ", "))
+			d.Printf("\t\t\t%s: %s\n", backupGroup.label, strings.Join(backupGroup.volumes, ", "))
 		}
 	}
 }
@@ -667,49 +882,6 @@ func (v *volumesByPod) Sorted() []*podVolumeGroup {
 	return v.volumesByPodSlice
 }
 
-func DescribeCSIVolumeSnapshots(d *Describer, details bool, volumeSnapshotContents []snapshotv1api.VolumeSnapshotContent) {
-	if !features.IsEnabled(velerov1api.CSIFeatureFlag) {
-		return
-	}
-
-	if len(volumeSnapshotContents) == 0 {
-		d.Printf("CSI Volume Snapshots: <none included>\n")
-		return
-	}
-
-	if !details {
-		d.Printf("CSI Volume Snapshots:\t%d included (specify --details for more information)\n", len(volumeSnapshotContents))
-		return
-	}
-
-	d.Printf("CSI Volume Snapshots:\n")
-
-	for _, vsc := range volumeSnapshotContents {
-		DescribeVSC(d, details, vsc)
-	}
-}
-
-func DescribeVSC(d *Describer, details bool, vsc snapshotv1api.VolumeSnapshotContent) {
-	if vsc.Status == nil {
-		d.Printf("Volume Snapshot Content %s cannot be described because its status is nil\n", vsc.Name)
-		return
-	}
-
-	d.Printf("Snapshot Content Name: %s\n", vsc.Name)
-
-	if vsc.Status.SnapshotHandle != nil {
-		d.Printf("\tStorage Snapshot ID: %s\n", *vsc.Status.SnapshotHandle)
-	}
-
-	if vsc.Status.RestoreSize != nil {
-		d.Printf("\tSnapshot Size (bytes): %d\n", *vsc.Status.RestoreSize)
-	}
-
-	if vsc.Status.ReadyToUse != nil {
-		d.Printf("\tReady to use: %t\n", *vsc.Status.ReadyToUse)
-	}
-}
-
 // DescribeBackupResults describes errors and warnings in human-readable format.
 func DescribeBackupResults(ctx context.Context, kbClient kbclient.Client, d *Describer, backup *velerov1api.Backup, insecureSkipTLSVerify bool, caCertPath string) {
 	if backup.Status.Warnings == 0 && backup.Status.Errors == 0 {

pkg/cmd/util/output/backup_describer_test.go

@@ -6,16 +6,16 @@ import (
 	"text/tabwriter"
 	"time"
 
+	"github.com/vmware-tanzu/velero/internal/volume"
+	"github.com/vmware-tanzu/velero/pkg/features"
 	"github.com/vmware-tanzu/velero/pkg/itemoperation"
 
 	"github.com/stretchr/testify/require"
 
-	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
 	"github.com/stretchr/testify/assert"
 	v1 "k8s.io/api/core/v1"
 
 	"github.com/vmware-tanzu/velero/pkg/builder"
-	"github.com/vmware-tanzu/velero/pkg/features"
 
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 )
@@ -307,22 +307,213 @@ OrderedResources:
 	}
 }
 
-func TestDescribeSnapshot(t *testing.T) {
-	d := &Describer{
-		Prefix: "",
-		out:    &tabwriter.Writer{},
-		buf:    &bytes.Buffer{},
+func TestDescribeNativeSnapshots(t *testing.T) {
+	testcases := []struct {
+		name         string
+		volumeInfo   []*volume.VolumeInfo
+		inputDetails bool
+		expect       string
+	}{
+		{
+			name: "no details",
+			volumeInfo: []*volume.VolumeInfo{
+				{
+					BackupMethod: volume.NativeSnapshot,
+					PVName:       "pv-1",
+					NativeSnapshotInfo: volume.NativeSnapshotInfo{
+						SnapshotHandle: "snapshot-1",
+						VolumeType:     "ebs",
+						VolumeAZ:       "us-east-2",
+						IOPS:           "1000 mbps",
+					},
+				},
+			},
+			expect: `  Velero-Native Snapshots:
+    pv-1: specify --details for more information
+`,
+		},
+		{
+			name: "details",
+			volumeInfo: []*volume.VolumeInfo{
+				{
+					BackupMethod: volume.NativeSnapshot,
+					PVName:       "pv-1",
+					NativeSnapshotInfo: volume.NativeSnapshotInfo{
+						SnapshotHandle: "snapshot-1",
+						VolumeType:     "ebs",
+						VolumeAZ:       "us-east-2",
+						IOPS:           "1000 mbps",
+					},
+				},
+			},
+			inputDetails: true,
+			expect: `  Velero-Native Snapshots:
+    pv-1:
+      Snapshot ID:        snapshot-1
+      Type:               ebs
+      Availability Zone:  us-east-2
+      IOPS:               1000 mbps
+`,
+		},
+	}
+
+	for _, tc := range testcases {
+		t.Run(tc.name, func(tt *testing.T) {
+			d := &Describer{
+				Prefix: "",
+				out:    &tabwriter.Writer{},
+				buf:    &bytes.Buffer{},
+			}
+			d.out.Init(d.buf, 0, 8, 2, ' ', 0)
+			describeNativeSnapshots(d, tc.inputDetails, tc.volumeInfo)
+			d.out.Flush()
+			assert.Equal(t, tc.expect, d.buf.String())
+		})
+	}
+}
+
+func TestCSISnapshots(t *testing.T) {
+	features.Enable(velerov1api.CSIFeatureFlag)
+	defer func() {
+		features.Disable(velerov1api.CSIFeatureFlag)
+	}()
+
+	testcases := []struct {
+		name         string
+		volumeInfo   []*volume.VolumeInfo
+		inputDetails bool
+		expect       string
+	}{
+		{
+			name: "no details, local snapshot",
+			volumeInfo: []*volume.VolumeInfo{
+				{
+					BackupMethod:          volume.CSISnapshot,
+					PVCName:               "pvc-1",
+					PreserveLocalSnapshot: true,
+					OperationID:           "fake-operation-1",
+					CSISnapshotInfo: volume.CSISnapshotInfo{
+						SnapshotHandle: "snapshot-1",
+						Size:           1024,
+						Driver:         "fake-driver",
+						VSCName:        "vsc-1",
+					},
+				},
+			},
+			expect: `  CSI Snapshots:
+    pvc-1:
+      Snapshot: included, specify --details for more information
+`,
+		},
+		{
+			name: "details, local snapshot",
+			volumeInfo: []*volume.VolumeInfo{
+				{
+					BackupMethod:          volume.CSISnapshot,
+					PVCName:               "pvc-2",
+					PreserveLocalSnapshot: true,
+					OperationID:           "fake-operation-2",
+					CSISnapshotInfo: volume.CSISnapshotInfo{
+						SnapshotHandle: "snapshot-2",
+						Size:           1024,
+						Driver:         "fake-driver",
+						VSCName:        "vsc-2",
+					},
+				},
+			},
+			inputDetails: true,
+			expect: `  CSI Snapshots:
+    pvc-2:
+      Snapshot:
+        Operation ID: fake-operation-2
+        Snapshot Content Name: vsc-2
+        Storage Snapshot ID: snapshot-2
+        Snapshot Size (bytes): 1024
+        CSI Driver: fake-driver
+`,
+		},
+		{
+			name: "no details, data movement",
+			volumeInfo: []*volume.VolumeInfo{
+				{
+					BackupMethod:      volume.CSISnapshot,
+					PVCName:           "pvc-3",
+					SnapshotDataMoved: true,
+					OperationID:       "fake-operation-3",
+					SnapshotDataMovementInfo: volume.SnapshotDataMovementInfo{
+						DataMover:      "velero",
+						UploaderType:   "fake-uploader",
+						SnapshotHandle: "fake-repo-id-3",
+					},
+				},
+			},
+			expect: `  CSI Snapshots:
+    pvc-3:
+      Data Movement: included, specify --details for more information
+`,
+		},
+		{
+			name: "details, data movement",
+			volumeInfo: []*volume.VolumeInfo{
+				{
+					BackupMethod:      volume.CSISnapshot,
+					PVCName:           "pvc-4",
+					SnapshotDataMoved: true,
+					OperationID:       "fake-operation-4",
+					SnapshotDataMovementInfo: volume.SnapshotDataMovementInfo{
+						DataMover:      "velero",
+						UploaderType:   "fake-uploader",
+						SnapshotHandle: "fake-repo-id-4",
+					},
+				},
+			},
+			inputDetails: true,
+			expect: `  CSI Snapshots:
+    pvc-4:
+      Data Movement:
+        Operation ID: fake-operation-4
+        Data Mover: velero
+        Uploader Type: fake-uploader
+`,
+		},
+		{
+			name: "details, data movement, data mover is empty",
+			volumeInfo: []*volume.VolumeInfo{
+				{
+					BackupMethod:      volume.CSISnapshot,
+					PVCName:           "pvc-5",
+					SnapshotDataMoved: true,
+					OperationID:       "fake-operation-5",
+					SnapshotDataMovementInfo: volume.SnapshotDataMovementInfo{
+						UploaderType:   "fake-uploader",
+						SnapshotHandle: "fake-repo-id-5",
+					},
+				},
+			},
+			inputDetails: true,
+			expect: `  CSI Snapshots:
+    pvc-5:
+      Data Movement:
+        Operation ID: fake-operation-5
+        Data Mover: velero
+        Uploader Type: fake-uploader
+`,
+		},
+	}
+
+	for _, tc := range testcases {
+		t.Run(tc.name, func(tt *testing.T) {
+			d := &Describer{
+				Prefix: "",
+				out:    &tabwriter.Writer{},
+				buf:    &bytes.Buffer{},
+			}
+			d.out.Init(d.buf, 0, 8, 2, ' ', 0)
+			describeCSISnapshots(d, tc.inputDetails, tc.volumeInfo)
+			d.out.Flush()
+			assert.Equal(t, tc.expect, d.buf.String())
+		})
 	}
-	d.out.Init(d.buf, 0, 8, 2, ' ', 0)
-	describeSnapshot(d, "pv-1", "snapshot-1", "ebs", "us-east-2", nil)
-	expect1 := `  pv-1:
-    Snapshot ID:        snapshot-1
-    Type:               ebs
-    Availability Zone:  us-east-2
-    IOPS:               <N/A>
-`
-	d.out.Flush()
-	assert.Equal(t, expect1, d.buf.String())
 }
 
 func TestDescribePodVolumeBackups(t *testing.T) {
@@ -353,24 +544,25 @@ func TestDescribePodVolumeBackups(t *testing.T) {
 			name:         "empty list",
 			inputPVBList: []velerov1api.PodVolumeBackup{},
 			inputDetails: true,
-			expect:       ``,
+			expect: `  Pod Volume Backups: <none included>
+`,
 		},
 		{
 			name:         "2 completed pvbs no details",
 			inputPVBList: []velerov1api.PodVolumeBackup{*pvb1, *pvb2},
 			inputDetails: false,
-			expect: `kopia Backups (specify --details for more information):
-  Completed:  2
+			expect: `  Pod Volume Backups - kopia (specify --details for more information):
+    Completed:  2
 `,
 		},
 		{
 			name:         "2 completed pvbs with details",
 			inputPVBList: []velerov1api.PodVolumeBackup{*pvb1, *pvb2},
 			inputDetails: true,
-			expect: `kopia Backups:
-  Completed:
-    pod-ns-1/pod-1: vol-1
-    pod-ns-1/pod-2: vol-2
+			expect: `  Pod Volume Backups - kopia:
+    Completed:
+      pod-ns-1/pod-1: vol-1
+      pod-ns-1/pod-2: vol-2
 `,
 		},
 	}
@@ -383,68 +575,7 @@ func TestDescribePodVolumeBackups(t *testing.T) {
 				buf:    &bytes.Buffer{},
 			}
 			d.out.Init(d.buf, 0, 8, 2, ' ', 0)
-			DescribePodVolumeBackups(d, tc.inputPVBList, tc.inputDetails)
-			d.out.Flush()
-			assert.Equal(tt, tc.expect, d.buf.String())
-		})
-	}
-}
-
-func TestDescribeCSIVolumeSnapshots(t *testing.T) {
-	features.Enable(velerov1api.CSIFeatureFlag)
-	defer func() {
-		features.Disable(velerov1api.CSIFeatureFlag)
-	}()
-	handle := "handle-1"
-	readyToUse := true
-	size := int64(1024)
-	vsc1 := builder.ForVolumeSnapshotContent("vsc-1").
-		Status(&snapshotv1api.VolumeSnapshotContentStatus{
-			SnapshotHandle: &handle,
-			ReadyToUse:     &readyToUse,
-			RestoreSize:    &size,
-		}).Result()
-	testcases := []struct {
-		name         string
-		inputVSCList []snapshotv1api.VolumeSnapshotContent
-		inputDetails bool
-		expect       string
-	}{
-		{
-			name:         "empty list",
-			inputVSCList: []snapshotv1api.VolumeSnapshotContent{},
-			inputDetails: false,
-			expect: `CSI Volume Snapshots: <none included>
-`,
-		},
-		{
-			name:         "1 vsc no details",
-			inputVSCList: []snapshotv1api.VolumeSnapshotContent{*vsc1},
-			inputDetails: false,
-			expect: `CSI Volume Snapshots:  1 included (specify --details for more information)
-`,
-		},
-		{
-			name:         "1 vsc with details",
-			inputVSCList: []snapshotv1api.VolumeSnapshotContent{*vsc1},
-			inputDetails: true,
-			expect: `CSI Volume Snapshots:
-Snapshot Content Name: vsc-1
-  Storage Snapshot ID: handle-1
-  Snapshot Size (bytes): 1024
-  Ready to use: true
-`,
-		},
-	}
-	for _, tc := range testcases {
-		t.Run(tc.name, func(tt *testing.T) {
-			d := &Describer{
-				Prefix: "",
-				out:    &tabwriter.Writer{},
-				buf:    &bytes.Buffer{},
-			}
-			d.out.Init(d.buf, 0, 8, 2, ' ', 0)
-			DescribeCSIVolumeSnapshots(d, tc.inputDetails, tc.inputVSCList)
+			describePodVolumeBackups(d, tc.inputDetails, tc.inputPVBList)
 			d.out.Flush()
 			assert.Equal(tt, tc.expect, d.buf.String())
 		})

pkg/cmd/util/output/backup_structured_describer.go

@@ -26,15 +26,14 @@ import (
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
-	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
-
 	kbclient "sigs.k8s.io/controller-runtime/pkg/client"
 
+	"github.com/vmware-tanzu/velero/internal/volume"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/downloadrequest"
 	"github.com/vmware-tanzu/velero/pkg/features"
+	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
 	"github.com/vmware-tanzu/velero/pkg/util/results"
-	"github.com/vmware-tanzu/velero/pkg/volume"
 )
 
 // DescribeBackupInSF describes a backup in structured format.
@@ -44,7 +43,6 @@ func DescribeBackupInSF(
 	backup *velerov1api.Backup,
 	deleteRequests []velerov1api.DeleteBackupRequest,
 	podVolumeBackups []velerov1api.PodVolumeBackup,
-	volumeSnapshotContents []snapshotv1api.VolumeSnapshotContent,
 	details bool,
 	insecureSkipTLSVerify bool,
 	caCertFile string,
@@ -68,19 +66,11 @@ func DescribeBackupInSF(
 
 		DescribeBackupSpecInSF(d, backup.Spec)
 
-		DescribeBackupStatusInSF(ctx, kbClient, d, backup, details, insecureSkipTLSVerify, caCertFile)
+		DescribeBackupStatusInSF(ctx, kbClient, d, backup, details, insecureSkipTLSVerify, caCertFile, podVolumeBackups)
 
 		if len(deleteRequests) > 0 {
 			DescribeDeleteBackupRequestsInSF(d, deleteRequests)
 		}
-
-		if features.IsEnabled(velerov1api.CSIFeatureFlag) {
-			DescribeCSIVolumeSnapshotsInSF(d, details, volumeSnapshotContents)
-		}
-
-		if len(podVolumeBackups) > 0 {
-			DescribePodVolumeBackupsInSF(d, podVolumeBackups, details)
-		}
 	}, outputFormat)
 }
 
@@ -234,7 +224,8 @@ func DescribeBackupSpecInSF(d *StructuredDescriber, spec velerov1api.BackupSpec)
 }
 
 // DescribeBackupStatusInSF describes a backup status in structured format.
-func DescribeBackupStatusInSF(ctx context.Context, kbClient kbclient.Client, d *StructuredDescriber, backup *velerov1api.Backup, details bool, insecureSkipTLSVerify bool, caCertPath string) {
+func DescribeBackupStatusInSF(ctx context.Context, kbClient kbclient.Client, d *StructuredDescriber, backup *velerov1api.Backup, details bool,
+	insecureSkipTLSVerify bool, caCertPath string, podVolumeBackups []velerov1api.PodVolumeBackup) {
 	status := backup.Status
 	backupStatusInfo := make(map[string]interface{})
 
@@ -274,35 +265,7 @@ func DescribeBackupStatusInSF(ctx context.Context, kbClient kbclient.Client, d *
 		describeBackupResourceListInSF(ctx, kbClient, backupStatusInfo, backup, insecureSkipTLSVerify, caCertPath)
 	}
 
-	// In consideration of decoding structured output conveniently, the three separate fields were created here
-	// the field of "veleroNativeSnapshots" displays the brief snapshots info
-	// the field of "errorGettingSnapshots" displays the error message if it fails to get snapshot info
-	// the field of "veleroNativeSnapshotsDetail" displays the detailed snapshots info
-	if status.VolumeSnapshotsAttempted > 0 {
-		if !details {
-			backupStatusInfo["veleroNativeSnapshots"] = fmt.Sprintf("%d of %d snapshots completed successfully", status.VolumeSnapshotsCompleted, status.VolumeSnapshotsAttempted)
-			return
-		}
-
-		buf := new(bytes.Buffer)
-		if err := downloadrequest.Stream(ctx, kbClient, backup.Namespace, backup.Name, velerov1api.DownloadTargetKindBackupVolumeSnapshots, buf, downloadRequestTimeout, insecureSkipTLSVerify, caCertPath); err != nil {
-			backupStatusInfo["errorGettingSnapshots"] = fmt.Sprintf("<error getting snapshot info: %v>", err)
-			return
-		}
-
-		var snapshots []*volume.Snapshot
-		if err := json.NewDecoder(buf).Decode(&snapshots); err != nil {
-			backupStatusInfo["errorGettingSnapshots"] = fmt.Sprintf("<error reading snapshot info: %v>", err)
-			return
-		}
-
-		snapshotDetails := make(map[string]interface{})
-		for _, snap := range snapshots {
-			describeSnapshotInSF(snap.Spec.PersistentVolumeName, snap.Status.ProviderSnapshotID, snap.Spec.VolumeType, snap.Spec.VolumeAZ, snap.Spec.VolumeIOPS, snapshotDetails)
-		}
-		backupStatusInfo["veleroNativeSnapshotsDetail"] = snapshotDetails
-		return
-	}
+	describeBackupVolumesInSF(ctx, kbClient, backup, details, insecureSkipTLSVerify, caCertPath, podVolumeBackups, backupStatusInfo)
 
 	if status.HookStatus != nil {
 		backupStatusInfo["hooksAttempted"] = status.HookStatus.HooksAttempted
@@ -337,18 +300,159 @@ func describeBackupResourceListInSF(ctx context.Context, kbClient kbclient.Clien
 	backupStatusInfo["resourceList"] = resourceList
 }
 
-func describeSnapshotInSF(pvName, snapshotID, volumeType, volumeAZ string, iops *int64, snapshotDetails map[string]interface{}) {
-	snapshotInfo := make(map[string]string)
-	iopsString := "<N/A>"
-	if iops != nil {
-		iopsString = fmt.Sprintf("%d", *iops)
+func describeBackupVolumesInSF(ctx context.Context, kbClient kbclient.Client, backup *velerov1api.Backup, details bool,
+	insecureSkipTLSVerify bool, caCertPath string, podVolumeBackupCRs []velerov1api.PodVolumeBackup, backupStatusInfo map[string]interface{}) {
+	if boolptr.IsSetToFalse(backup.Spec.SnapshotVolumes) {
+		backupStatusInfo["backupVolumes"] = "<none included>"
+		return
 	}
 
-	snapshotInfo["snapshotID"] = snapshotID
-	snapshotInfo["type"] = volumeType
-	snapshotInfo["availabilityZone"] = volumeAZ
-	snapshotInfo["IOPS"] = iopsString
-	snapshotDetails[pvName] = snapshotInfo
+	backupVolumes := make(map[string]interface{})
+
+	nativeSnapshots := []*volume.VolumeInfo{}
+	csiSnapshots := []*volume.VolumeInfo{}
+
+	buf := new(bytes.Buffer)
+	err := downloadrequest.Stream(ctx, kbClient, backup.Namespace, backup.Name, velerov1api.DownloadTargetKindBackupVolumeInfos, buf, downloadRequestTimeout, insecureSkipTLSVerify, caCertPath)
+	if err == downloadrequest.ErrNotFound {
+		nativeSnapshots, err = retrieveNativeSnapshotLegacy(ctx, kbClient, backup, insecureSkipTLSVerify, caCertPath)
+		if err != nil {
+			backupVolumes["errorConcludeNativeSnapshot"] = fmt.Sprintf("error concluding native snapshot info: %v", err)
+			return
+		}
+
+		csiSnapshots, err = retrieveCSISnapshotLegacy(ctx, kbClient, backup, insecureSkipTLSVerify, caCertPath)
+		if err != nil {
+			backupVolumes["errorConcludeCSISnapshot"] = fmt.Sprintf("error concluding CSI snapshot info: %v", err)
+			return
+		}
+	} else if err != nil {
+		backupVolumes["errorGetBackupVolumeInfo"] = fmt.Sprintf("error getting backup volume info: %v", err)
+		return
+	} else {
+		var volumeInfos []volume.VolumeInfo
+		if err := json.NewDecoder(buf).Decode(&volumeInfos); err != nil {
+			backupVolumes["errorReadBackupVolumeInfo"] = fmt.Sprintf("error reading backup volume info: %v", err)
+			return
+		}
+
+		for i := range volumeInfos {
+			switch volumeInfos[i].BackupMethod {
+			case volume.NativeSnapshot:
+				nativeSnapshots = append(nativeSnapshots, &volumeInfos[i])
+			case volume.CSISnapshot:
+				csiSnapshots = append(csiSnapshots, &volumeInfos[i])
+			}
+		}
+	}
+
+	describeNativeSnapshotsInSF(details, nativeSnapshots, backupVolumes)
+
+	describeCSISnapshotsInSF(details, csiSnapshots, backupVolumes)
+
+	describePodVolumeBackupsInSF(podVolumeBackupCRs, details, backupVolumes)
+
+	backupStatusInfo["backupVolumes"] = backupVolumes
+}
+
+func describeNativeSnapshotsInSF(details bool, infos []*volume.VolumeInfo, backupVolumes map[string]interface{}) {
+	if len(infos) == 0 {
+		backupVolumes["nativeSnapshots"] = "<none included>"
+		return
+	}
+
+	snapshotDetails := make(map[string]interface{})
+	for _, info := range infos {
+		describNativeSnapshotInSF(details, info, snapshotDetails)
+	}
+	backupVolumes["nativeSnapshots"] = snapshotDetails
+}
+
+func describNativeSnapshotInSF(details bool, info *volume.VolumeInfo, snapshotDetails map[string]interface{}) {
+	if details {
+		snapshotInfo := make(map[string]string)
+		snapshotInfo["snapshotID"] = info.NativeSnapshotInfo.SnapshotHandle
+		snapshotInfo["type"] = info.NativeSnapshotInfo.VolumeType
+		snapshotInfo["availabilityZone"] = info.NativeSnapshotInfo.VolumeAZ
+		snapshotInfo["IOPS"] = info.NativeSnapshotInfo.IOPS
+
+		snapshotDetails[info.PVName] = snapshotInfo
+	} else {
+		snapshotDetails[info.PVName] = "specify --details for more information"
+	}
+}
+
+func describeCSISnapshotsInSF(details bool, infos []*volume.VolumeInfo, backupVolumes map[string]interface{}) {
+	if !features.IsEnabled(velerov1api.CSIFeatureFlag) {
+		return
+	}
+
+	if len(infos) == 0 {
+		backupVolumes["csiSnapshots"] = "<none included>"
+		return
+	}
+
+	snapshotDetails := make(map[string]interface{})
+	for _, info := range infos {
+		describeCSISnapshotInSF(details, info, snapshotDetails)
+	}
+	backupVolumes["csiSnapshots"] = snapshotDetails
+}
+
+func describeCSISnapshotInSF(details bool, info *volume.VolumeInfo, snapshotDetails map[string]interface{}) {
+	snapshotDetail := make(map[string]interface{})
+
+	describeLocalSnapshotInSF(details, info, snapshotDetail)
+	describeDataMovementInSF(details, info, snapshotDetail)
+
+	snapshotDetails[info.PVCName] = snapshotDetail
+}
+
+// describeVSCInSF describes CSI volume snapshot contents in structured format.
+func describeLocalSnapshotInSF(details bool, info *volume.VolumeInfo, snapshotDetail map[string]interface{}) {
+	if !info.PreserveLocalSnapshot {
+		return
+	}
+
+	if details {
+		localSnapshot := make(map[string]interface{})
+
+		if !info.SnapshotDataMoved {
+			localSnapshot["operationID"] = info.OperationID
+		}
+
+		localSnapshot["snapshotContentName"] = info.CSISnapshotInfo.VSCName
+		localSnapshot["storageSnapshotID"] = info.CSISnapshotInfo.SnapshotHandle
+		localSnapshot["snapshotSize(bytes)"] = info.CSISnapshotInfo.Size
+		localSnapshot["csiDriver"] = info.CSISnapshotInfo.Driver
+
+		snapshotDetail["snapshot"] = localSnapshot
+	} else {
+		snapshotDetail["snapshot"] = "included, specify --details for more information"
+	}
+}
+
+func describeDataMovementInSF(details bool, info *volume.VolumeInfo, snapshotDetail map[string]interface{}) {
+	if !info.SnapshotDataMoved {
+		return
+	}
+
+	if details {
+		dataMovement := make(map[string]interface{})
+		dataMovement["operationID"] = info.OperationID
+
+		dataMover := "velero"
+		if info.SnapshotDataMovementInfo.DataMover != "" {
+			dataMover = info.SnapshotDataMovementInfo.DataMover
+		}
+		dataMovement["dataMover"] = dataMover
+
+		dataMovement["uploaderType"] = info.SnapshotDataMovementInfo.UploaderType
+
+		snapshotDetail["dataMovement"] = dataMovement
+	} else {
+		snapshotDetail["dataMovement"] = "included, specify --details for more information"
+	}
 }
 
 // DescribeDeleteBackupRequestsInSF describes delete backup requests in structured format.
@@ -373,19 +477,20 @@ func DescribeDeleteBackupRequestsInSF(d *StructuredDescriber, requests []velerov
 	d.Describe("deletionAttempts", deletionAttempts)
 }
 
-// DescribePodVolumeBackupsInSF describes pod volume backups in structured format.
-func DescribePodVolumeBackupsInSF(d *StructuredDescriber, backups []velerov1api.PodVolumeBackup, details bool) {
-	PodVolumeBackupsInfo := make(map[string]interface{})
+// describePodVolumeBackupsInSF describes pod volume backups in structured format.
+func describePodVolumeBackupsInSF(backups []velerov1api.PodVolumeBackup, details bool, backupVolumes map[string]interface{}) {
+	podVolumeBackupsInfo := make(map[string]interface{})
 	// Get the type of pod volume uploader. Since the uploader only comes from a single source, we can
 	// take the uploader type from the first element of the array.
 	var uploaderType string
 	if len(backups) > 0 {
 		uploaderType = backups[0].Spec.UploaderType
 	} else {
+		backupVolumes["podVolumeBackups"] = "<none included>"
 		return
 	}
 	// type display the type of pod volume backups
-	PodVolumeBackupsInfo["type"] = uploaderType
+	podVolumeBackupsInfo["uploderType"] = uploaderType
 
 	podVolumeBackupsDetails := make(map[string]interface{})
 	// separate backups by phase (combining <none> and New into a single group)
@@ -420,56 +525,8 @@ func DescribePodVolumeBackupsInSF(d *StructuredDescriber, backups []velerov1api.
 		podVolumeBackupsDetails[phase] = backupsByPods
 	}
 	// Pod Volume Backups Details display the detailed pod volume backups info
-	PodVolumeBackupsInfo["podVolumeBackupsDetails"] = podVolumeBackupsDetails
-	d.Describe("podVolumeBackups", PodVolumeBackupsInfo)
-}
-
-// DescribeCSIVolumeSnapshotsInSF describes CSI volume snapshots in structured format.
-func DescribeCSIVolumeSnapshotsInSF(d *StructuredDescriber, details bool, volumeSnapshotContents []snapshotv1api.VolumeSnapshotContent) {
-	CSIVolumeSnapshotsInfo := make(map[string]interface{})
-	if !features.IsEnabled(velerov1api.CSIFeatureFlag) {
-		return
-	}
-
-	if len(volumeSnapshotContents) == 0 {
-		return
-	}
-
-	// In consideration of decoding structured output conveniently, the two separate fields were created here
-	// the field of 'CSI Volume Snapshots Count' displays the count of CSI Volume Snapshots
-	// the field of 'CSI Volume Snapshots Details' displays the content of CSI Volume Snapshots
-	if !details {
-		CSIVolumeSnapshotsInfo["CSIVolumeSnapshotsCount"] = len(volumeSnapshotContents)
-	} else {
-		vscDetails := make(map[string]interface{})
-		for _, vsc := range volumeSnapshotContents {
-			DescribeVSCInSF(details, vsc, vscDetails)
-		}
-		CSIVolumeSnapshotsInfo["CSIVolumeSnapshotsDetails"] = vscDetails
-	}
-	d.Describe("CSIVolumeSnapshots", CSIVolumeSnapshotsInfo)
-}
-
-// DescribeVSCInSF describes CSI volume snapshot contents in structured format.
-func DescribeVSCInSF(details bool, vsc snapshotv1api.VolumeSnapshotContent, vscDetails map[string]interface{}) {
-	content := make(map[string]interface{})
-	if vsc.Status == nil {
-		vscDetails[vsc.Name] = content
-		return
-	}
-
-	if vsc.Status.SnapshotHandle != nil {
-		content["storageSnapshotID"] = *vsc.Status.SnapshotHandle
-	}
-
-	if vsc.Status.RestoreSize != nil {
-		content["snapshotSize(bytes)"] = *vsc.Status.RestoreSize
-	}
-
-	if vsc.Status.ReadyToUse != nil {
-		content["readyToUse"] = *vsc.Status.ReadyToUse
-	}
-	vscDetails[vsc.Name] = content
+	podVolumeBackupsInfo["podVolumeBackupsDetails"] = podVolumeBackupsDetails
+	backupVolumes["podVolumeBackups"] = podVolumeBackupsInfo
 }
 
 // DescribeBackupResultsInSF describes errors and warnings in structured format.

pkg/cmd/util/output/backup_structured_describer_test.go

@@ -9,6 +9,7 @@ import (
 
 	v1 "k8s.io/api/core/v1"
 
+	"github.com/vmware-tanzu/velero/internal/volume"
 	"github.com/vmware-tanzu/velero/pkg/features"
 	"github.com/vmware-tanzu/velero/pkg/util/results"
 
@@ -16,8 +17,6 @@ import (
 
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/builder"
-
-	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
 )
 
 func TestDescribeBackupInSF(t *testing.T) {
@@ -239,7 +238,7 @@ func TestDescribePodVolumeBackupsInSF(t *testing.T) {
 			name:         "empty list",
 			inputPVBList: []velerov1api.PodVolumeBackup{},
 			inputDetails: false,
-			expect:       map[string]interface{}{},
+			expect:       map[string]interface{}{"podVolumeBackups": "<none included>"},
 		},
 		{
 			name:         "2 completed pvbs",
@@ -253,72 +252,224 @@ func TestDescribePodVolumeBackupsInSF(t *testing.T) {
 							{"pod-ns-1/pod-2": "vol-2"},
 						},
 					},
-					"type": "kopia",
+					"uploderType": "kopia",
 				},
 			},
 		},
 	}
 	for _, tc := range testcases {
 		t.Run(tc.name, func(tt *testing.T) {
-			sd := &StructuredDescriber{
-				output: make(map[string]interface{}),
-				format: "",
-			}
-			DescribePodVolumeBackupsInSF(sd, tc.inputPVBList, tc.inputDetails)
-			assert.True(tt, reflect.DeepEqual(sd.output, tc.expect))
+			output := make(map[string]interface{})
+			describePodVolumeBackupsInSF(tc.inputPVBList, tc.inputDetails, output)
+			assert.True(tt, reflect.DeepEqual(output, tc.expect))
 		})
 	}
 }
 
-func TestDescribeCSIVolumeSnapshotsInSF(t *testing.T) {
+func TestDescribeNativeSnapshotsInSF(t *testing.T) {
+	testcases := []struct {
+		name         string
+		volumeInfo   []*volume.VolumeInfo
+		inputDetails bool
+		expect       map[string]interface{}
+	}{
+		{
+			name: "no details",
+			volumeInfo: []*volume.VolumeInfo{
+				{
+					BackupMethod: volume.NativeSnapshot,
+					PVName:       "pv-1",
+					NativeSnapshotInfo: volume.NativeSnapshotInfo{
+						SnapshotHandle: "snapshot-1",
+						VolumeType:     "ebs",
+						VolumeAZ:       "us-east-2",
+						IOPS:           "1000 mbps",
+					},
+				},
+			},
+			expect: map[string]interface{}{
+				"nativeSnapshots": map[string]interface{}{
+					"pv-1": "specify --details for more information",
+				},
+			},
+		},
+		{
+			name: "details",
+			volumeInfo: []*volume.VolumeInfo{
+				{
+					BackupMethod: volume.NativeSnapshot,
+					PVName:       "pv-1",
+					NativeSnapshotInfo: volume.NativeSnapshotInfo{
+						SnapshotHandle: "snapshot-1",
+						VolumeType:     "ebs",
+						VolumeAZ:       "us-east-2",
+						IOPS:           "1000 mbps",
+					},
+				},
+			},
+			inputDetails: true,
+			expect: map[string]interface{}{
+				"nativeSnapshots": map[string]interface{}{
+					"pv-1": map[string]string{
+						"snapshotID":       "snapshot-1",
+						"type":             "ebs",
+						"availabilityZone": "us-east-2",
+						"IOPS":             "1000 mbps",
+					},
+				},
+			},
+		},
+	}
+
+	for _, tc := range testcases {
+		t.Run(tc.name, func(tt *testing.T) {
+			output := make(map[string]interface{})
+			describeNativeSnapshotsInSF(tc.inputDetails, tc.volumeInfo, output)
+			assert.True(tt, reflect.DeepEqual(output, tc.expect))
+		})
+	}
+}
+
+func TestDescribeCSISnapshotsInSF(t *testing.T) {
 	features.Enable(velerov1api.CSIFeatureFlag)
 	defer func() {
 		features.Disable(velerov1api.CSIFeatureFlag)
 	}()
 
-	vscBuilder1 := builder.ForVolumeSnapshotContent("vsc-1")
-	handle := "handle-1"
-	readyToUse := true
-	size := int64(1024)
-	vsc1 := vscBuilder1.Status(&snapshotv1api.VolumeSnapshotContentStatus{
-		SnapshotHandle: &handle,
-		ReadyToUse:     &readyToUse,
-		RestoreSize:    &size,
-	}).Result()
-
 	testcases := []struct {
 		name         string
-		inputVSCList []snapshotv1api.VolumeSnapshotContent
+		volumeInfo   []*volume.VolumeInfo
 		inputDetails bool
 		expect       map[string]interface{}
 	}{
 		{
-			name:         "empty list",
-			inputVSCList: []snapshotv1api.VolumeSnapshotContent{},
-			inputDetails: false,
-			expect:       map[string]interface{}{},
-		},
-		{
-			name:         "1 vsc no detail",
-			inputVSCList: []snapshotv1api.VolumeSnapshotContent{*vsc1},
-			inputDetails: false,
+			name: "no details, local snapshot",
+			volumeInfo: []*volume.VolumeInfo{
+				{
+					BackupMethod:          volume.CSISnapshot,
+					PVCName:               "pvc-1",
+					PreserveLocalSnapshot: true,
+					OperationID:           "fake-operation-1",
+					CSISnapshotInfo: volume.CSISnapshotInfo{
+						SnapshotHandle: "snapshot-1",
+						Size:           1024,
+						Driver:         "fake-driver",
+						VSCName:        "vsc-1",
+					},
+				},
+			},
 			expect: map[string]interface{}{
-				"CSIVolumeSnapshots": map[string]interface{}{
-					"CSIVolumeSnapshotsCount": 1,
+				"csiSnapshots": map[string]interface{}{
+					"pvc-1": map[string]interface{}{
+						"snapshot": "included, specify --details for more information",
+					},
 				},
 			},
 		},
 		{
-			name:         "1 vsc with detail",
-			inputVSCList: []snapshotv1api.VolumeSnapshotContent{*vsc1},
+			name: "details, local snapshot",
+			volumeInfo: []*volume.VolumeInfo{
+				{
+					BackupMethod:          volume.CSISnapshot,
+					PVCName:               "pvc-2",
+					PreserveLocalSnapshot: true,
+					OperationID:           "fake-operation-2",
+					CSISnapshotInfo: volume.CSISnapshotInfo{
+						SnapshotHandle: "snapshot-2",
+						Size:           1024,
+						Driver:         "fake-driver",
+						VSCName:        "vsc-2",
+					},
+				},
+			},
 			inputDetails: true,
 			expect: map[string]interface{}{
-				"CSIVolumeSnapshots": map[string]interface{}{
-					"CSIVolumeSnapshotsDetails": map[string]interface{}{
-						"vsc-1": map[string]interface{}{
-							"readyToUse":          true,
+				"csiSnapshots": map[string]interface{}{
+					"pvc-2": map[string]interface{}{
+						"snapshot": map[string]interface{}{
+							"operationID":         "fake-operation-2",
+							"snapshotContentName": "vsc-2",
+							"storageSnapshotID":   "snapshot-2",
 							"snapshotSize(bytes)": int64(1024),
-							"storageSnapshotID":   "handle-1",
+							"csiDriver":           "fake-driver",
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "no details, data movement",
+			volumeInfo: []*volume.VolumeInfo{
+				{
+					BackupMethod:      volume.CSISnapshot,
+					PVCName:           "pvc-3",
+					SnapshotDataMoved: true,
+					OperationID:       "fake-operation-3",
+					SnapshotDataMovementInfo: volume.SnapshotDataMovementInfo{
+						DataMover:      "velero",
+						UploaderType:   "fake-uploader",
+						SnapshotHandle: "fake-repo-id-3",
+					},
+				},
+			},
+			expect: map[string]interface{}{
+				"csiSnapshots": map[string]interface{}{
+					"pvc-3": map[string]interface{}{
+						"dataMovement": "included, specify --details for more information",
+					},
+				},
+			},
+		},
+		{
+			name: "details, data movement",
+			volumeInfo: []*volume.VolumeInfo{
+				{
+					BackupMethod:      volume.CSISnapshot,
+					PVCName:           "pvc-4",
+					SnapshotDataMoved: true,
+					OperationID:       "fake-operation-4",
+					SnapshotDataMovementInfo: volume.SnapshotDataMovementInfo{
+						DataMover:      "velero",
+						UploaderType:   "fake-uploader",
+						SnapshotHandle: "fake-repo-id-4",
+					},
+				},
+			},
+			inputDetails: true,
+			expect: map[string]interface{}{
+				"csiSnapshots": map[string]interface{}{
+					"pvc-4": map[string]interface{}{
+						"dataMovement": map[string]interface{}{
+							"operationID":  "fake-operation-4",
+							"dataMover":    "velero",
+							"uploaderType": "fake-uploader",
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "details, data movement, data mover is empty",
+			volumeInfo: []*volume.VolumeInfo{
+				{
+					BackupMethod:      volume.CSISnapshot,
+					PVCName:           "pvc-4",
+					SnapshotDataMoved: true,
+					OperationID:       "fake-operation-4",
+					SnapshotDataMovementInfo: volume.SnapshotDataMovementInfo{
+						UploaderType:   "fake-uploader",
+						SnapshotHandle: "fake-repo-id-4",
+					},
+				},
+			},
+			inputDetails: true,
+			expect: map[string]interface{}{
+				"csiSnapshots": map[string]interface{}{
+					"pvc-4": map[string]interface{}{
+						"dataMovement": map[string]interface{}{
+							"operationID":  "fake-operation-4",
+							"dataMover":    "velero",
+							"uploaderType": "fake-uploader",
 						},
 					},
 				},
@@ -328,12 +479,9 @@ func TestDescribeCSIVolumeSnapshotsInSF(t *testing.T) {
 
 	for _, tc := range testcases {
 		t.Run(tc.name, func(tt *testing.T) {
-			sd := &StructuredDescriber{
-				output: make(map[string]interface{}),
-				format: "",
-			}
-			DescribeCSIVolumeSnapshotsInSF(sd, tc.inputDetails, tc.inputVSCList)
-			assert.True(tt, reflect.DeepEqual(sd.output, tc.expect))
+			output := make(map[string]interface{})
+			describeCSISnapshotsInSF(tc.inputDetails, tc.volumeInfo, output)
+			assert.True(tt, reflect.DeepEqual(output, tc.expect))
 		})
 	}
 }
@@ -441,18 +589,3 @@ func TestDescribeDeleteBackupRequestsInSF(t *testing.T) {
 	}
 
 }
-
-func TestDescribeSnapshotInSF(t *testing.T) {
-	res := map[string]interface{}{}
-	iops := int64(100)
-	describeSnapshotInSF("pv-1", "snapshot-1", "ebs", "us-east-2", &iops, res)
-	expect := map[string]interface{}{
-		"pv-1": map[string]string{
-			"snapshotID":       "snapshot-1",
-			"type":             "ebs",
-			"availabilityZone": "us-east-2",
-			"IOPS":             "100",
-		},
-	}
-	assert.True(t, reflect.DeepEqual(expect, res))
-}