Merge branch 'main' into report-incremental-fallback

pkg/cmd/cli/backup/describe.go

@@ -56,7 +56,7 @@ func NewDescribeCommand(f client.Factory, use string) *cobra.Command {
 			cmd.CheckError(err)
 
 			if outputFormat != "plaintext" && outputFormat != "json" {
-				cmd.CheckError(fmt.Errorf("invalid output format '%s'. valid value are 'plaintext, json'", outputFormat))
+				cmd.CheckError(fmt.Errorf("invalid output format '%s'. valid values are 'plaintext' and 'json'", outputFormat))
 			}
 
 			backups := new(velerov1api.BackupList)
@@ -118,7 +118,7 @@ func NewDescribeCommand(f client.Factory, use string) *cobra.Command {
 	c.Flags().BoolVar(&details, "details", details, "Display additional detail in the command output.")
 	c.Flags().BoolVar(&insecureSkipTLSVerify, "insecure-skip-tls-verify", insecureSkipTLSVerify, "If true, the object store's TLS certificate will not be checked for validity. This is insecure and susceptible to man-in-the-middle attacks. Not recommended for production.")
 	c.Flags().StringVar(&caCertFile, "cacert", caCertFile, "Path to a certificate bundle to use when verifying TLS connections. If not specified, the CA certificate from the BackupStorageLocation will be used if available.")
-	c.Flags().StringVarP(&outputFormat, "output", "o", outputFormat, "Output display format. Valid formats are 'plaintext, json'. 'json' only applies to a single backup")
+	c.Flags().StringVarP(&outputFormat, "output", "o", outputFormat, "Output display format. Valid formats are 'plaintext' and 'json'. 'json' only applies to a single backup")
 
 	return c
 }

pkg/cmd/cli/restore/describe.go

@@ -39,6 +39,7 @@ func NewDescribeCommand(f client.Factory, use string) *cobra.Command {
 		listOptions           metav1.ListOptions
 		details               bool
 		insecureSkipTLSVerify bool
+		outputFormat          = "plaintext"
 	)
 
 	config, err := client.LoadConfig()
@@ -54,6 +55,10 @@ func NewDescribeCommand(f client.Factory, use string) *cobra.Command {
 			kbClient, err := f.KubebuilderClient()
 			cmd.CheckError(err)
 
+			if outputFormat != "plaintext" && outputFormat != "json" {
+				cmd.CheckError(fmt.Errorf("invalid output format '%s'. valid values are 'plaintext' and 'json'", outputFormat))
+			}
+
 			restoreList := new(velerov1api.RestoreList)
 			if len(args) > 0 {
 				for _, name := range args {
@@ -81,12 +86,19 @@ func NewDescribeCommand(f client.Factory, use string) *cobra.Command {
 					fmt.Fprintf(os.Stderr, "error getting PodVolumeRestores for restore %s: %v\n", restore.Name, err)
 				}
 
-				s := output.DescribeRestore(context.Background(), kbClient, &restoreList.Items[i], podVolumeRestoreList.Items, details, insecureSkipTLSVerify, caCertFile)
-				if first {
-					first = false
+				// structured output only applies to a single restore in case of OOM
+				// To describe a list of restores in structured format, iterate and describe one at a time.
+				if len(restoreList.Items) == 1 && outputFormat != "plaintext" {
+					s := output.DescribeRestoreInSF(context.Background(), kbClient, &restoreList.Items[i], podVolumeRestoreList.Items, details, insecureSkipTLSVerify, caCertFile, outputFormat)
 					fmt.Print(s)
 				} else {
-					fmt.Printf("\n\n%s", s)
+					s := output.DescribeRestore(context.Background(), kbClient, &restoreList.Items[i], podVolumeRestoreList.Items, details, insecureSkipTLSVerify, caCertFile)
+					if first {
+						first = false
+						fmt.Print(s)
+					} else {
+						fmt.Printf("\n\n%s", s)
+					}
 				}
 			}
 			cmd.CheckError(err)
@@ -98,6 +110,7 @@ func NewDescribeCommand(f client.Factory, use string) *cobra.Command {
 	c.Flags().BoolVar(&details, "details", details, "Display additional detail in the command output.")
 	c.Flags().BoolVar(&insecureSkipTLSVerify, "insecure-skip-tls-verify", insecureSkipTLSVerify, "If true, the object store's TLS certificate will not be checked for validity. This is insecure and susceptible to man-in-the-middle attacks. Not recommended for production.")
 	c.Flags().StringVar(&caCertFile, "cacert", caCertFile, "Path to a certificate bundle to use when verifying TLS connections.")
+	c.Flags().StringVarP(&outputFormat, "output", "o", outputFormat, "Output display format. Valid formats are 'plaintext' and 'json'. 'json' only applies to a single restore")
 
 	return c
 }

pkg/cmd/util/output/restore_structured_describer.go

@@ -0,0 +1,542 @@
+/*
+Copyright the Velero contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package output
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"strings"
+
+	corev1api "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/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/cacert"
+	"github.com/vmware-tanzu/velero/pkg/cmd/util/downloadrequest"
+	"github.com/vmware-tanzu/velero/pkg/itemoperation"
+	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
+	"github.com/vmware-tanzu/velero/pkg/util/results"
+)
+
+// DescribeRestoreInSF describes a restore in structured format.
+func DescribeRestoreInSF(
+	ctx context.Context,
+	kbClient kbclient.Client,
+	restore *velerov1api.Restore,
+	podVolumeRestores []velerov1api.PodVolumeRestore,
+	details bool,
+	insecureSkipTLSVerify bool,
+	caCertFile string,
+	outputFormat string,
+) string {
+	return DescribeInSF(func(d *StructuredDescriber) {
+		d.DescribeMetadata(restore.ObjectMeta)
+
+		phase := restore.Status.Phase
+		if phase == "" {
+			phase = velerov1api.RestorePhaseNew
+		}
+		phaseString := string(phase)
+		if !restore.DeletionTimestamp.IsZero() {
+			phaseString += " (Deleting)"
+		}
+		d.Describe("phase", phaseString)
+
+		describeRestoreProgressInSF(d, restore)
+		describeRestoreTimestampsInSF(d, restore)
+
+		if len(restore.Status.ValidationErrors) > 0 {
+			d.Describe("validationErrors", restore.Status.ValidationErrors)
+		}
+
+		describeRestoreResultsInSF(ctx, kbClient, d, restore, insecureSkipTLSVerify, caCertFile)
+
+		describeRestoreSpecInSF(d, restore.Spec)
+
+		describePodVolumeRestoresInSF(d, podVolumeRestores, details)
+
+		describeRestoreCSISnapshotsInSF(ctx, kbClient, d, restore, details, insecureSkipTLSVerify, caCertFile)
+
+		describeRestoreItemOperationsInSF(ctx, kbClient, d, restore, details, insecureSkipTLSVerify, caCertFile)
+
+		if restore.Status.HookStatus != nil {
+			hookStatus := map[string]any{
+				"hooksAttempted": restore.Status.HookStatus.HooksAttempted,
+				"hooksFailed":    restore.Status.HookStatus.HooksFailed,
+			}
+			d.Describe("hookStatus", hookStatus)
+		}
+
+		if details {
+			describeRestoreResourceListInSF(ctx, kbClient, d, restore, insecureSkipTLSVerify, caCertFile)
+		}
+	}, outputFormat)
+}
+
+func describeRestoreProgressInSF(d *StructuredDescriber, restore *velerov1api.Restore) {
+	if restore.Status.Progress == nil {
+		return
+	}
+	progress := map[string]any{}
+	if restore.Status.Phase == velerov1api.RestorePhaseInProgress {
+		progress["estimatedTotalItemsToBeRestored"] = restore.Status.Progress.TotalItems
+		progress["itemsRestoredSoFar"] = restore.Status.Progress.ItemsRestored
+	} else {
+		progress["totalItemsToBeRestored"] = restore.Status.Progress.TotalItems
+		progress["itemsRestored"] = restore.Status.Progress.ItemsRestored
+	}
+	d.Describe("progress", progress)
+}
+
+func describeRestoreTimestampsInSF(d *StructuredDescriber, restore *velerov1api.Restore) {
+	timestamps := map[string]any{}
+	if restore.Status.StartTimestamp == nil || restore.Status.StartTimestamp.IsZero() {
+		timestamps["started"] = "<n/a>"
+	} else {
+		timestamps["started"] = restore.Status.StartTimestamp.String()
+	}
+	if restore.Status.CompletionTimestamp == nil || restore.Status.CompletionTimestamp.IsZero() {
+		timestamps["completed"] = "<n/a>"
+	} else {
+		timestamps["completed"] = restore.Status.CompletionTimestamp.String()
+	}
+	d.Describe("timestamps", timestamps)
+}
+
+func describeRestoreSpecInSF(d *StructuredDescriber, spec velerov1api.RestoreSpec) {
+	specInfo := map[string]any{}
+
+	specInfo["backupName"] = spec.BackupName
+
+	// namespaces
+	namespaceInfo := map[string]any{}
+	var s string
+	if len(spec.IncludedNamespaces) == 0 || (len(spec.IncludedNamespaces) == 1 && spec.IncludedNamespaces[0] == "*") {
+		s = "all namespaces found in the backup"
+	} else {
+		s = strings.Join(spec.IncludedNamespaces, ", ")
+	}
+	namespaceInfo["included"] = s
+	if len(spec.ExcludedNamespaces) == 0 {
+		s = emptyDisplay
+	} else {
+		s = strings.Join(spec.ExcludedNamespaces, ", ")
+	}
+	namespaceInfo["excluded"] = s
+	specInfo["namespaces"] = namespaceInfo
+
+	// resources
+	resourcesInfo := map[string]string{}
+	if len(spec.IncludedResources) == 0 {
+		s = "*"
+	} else {
+		s = strings.Join(spec.IncludedResources, ", ")
+	}
+	resourcesInfo["included"] = s
+	if len(spec.ExcludedResources) == 0 {
+		s = emptyDisplay
+	} else {
+		s = strings.Join(spec.ExcludedResources, ", ")
+	}
+	resourcesInfo["excluded"] = s
+	resourcesInfo["clusterScoped"] = BoolPointerString(spec.IncludeClusterResources, "excluded", "included", "auto")
+	specInfo["resources"] = resourcesInfo
+
+	// namespace mappings
+	if len(spec.NamespaceMapping) > 0 {
+		specInfo["namespaceMappings"] = spec.NamespaceMapping
+	} else {
+		specInfo["namespaceMappings"] = emptyDisplay
+	}
+
+	// label selector
+	s = emptyDisplay
+	if spec.LabelSelector != nil {
+		s = metav1.FormatLabelSelector(spec.LabelSelector)
+	}
+	specInfo["labelSelector"] = s
+
+	// or label selectors
+	if len(spec.OrLabelSelectors) == 0 {
+		specInfo["orLabelSelectors"] = emptyDisplay
+	} else {
+		orSelectors := make([]string, 0, len(spec.OrLabelSelectors))
+		for _, v := range spec.OrLabelSelectors {
+			orSelectors = append(orSelectors, metav1.FormatLabelSelector(v))
+		}
+		specInfo["orLabelSelectors"] = strings.Join(orSelectors, " or ")
+	}
+
+	specInfo["restorePVs"] = BoolPointerString(spec.RestorePVs, "false", "true", "auto")
+
+	// existing resource policy
+	if spec.ExistingResourcePolicy != "" {
+		specInfo["existingResourcePolicy"] = string(spec.ExistingResourcePolicy)
+	} else {
+		specInfo["existingResourcePolicy"] = emptyDisplay
+	}
+
+	specInfo["itemOperationTimeout"] = spec.ItemOperationTimeout.Duration.String()
+	specInfo["preserveNodePorts"] = BoolPointerString(spec.PreserveNodePorts, "false", "true", "auto")
+
+	// resource modifier
+	if spec.ResourceModifier != nil {
+		specInfo["resourceModifier"] = describeResourceModifierInSF(spec.ResourceModifier)
+	}
+
+	// resource policy
+	if spec.ResourcePolicy != nil {
+		specInfo["resourcePolicy"] = map[string]any{
+			"type": spec.ResourcePolicy.Kind,
+			"name": spec.ResourcePolicy.Name,
+		}
+	}
+
+	// uploader config
+	if spec.UploaderConfig != nil {
+		uploaderConfig := map[string]any{}
+		if boolptr.IsSetToTrue(spec.UploaderConfig.WriteSparseFiles) {
+			uploaderConfig["writeSparseFiles"] = true
+		}
+		if spec.UploaderConfig.ParallelFilesDownload > 0 {
+			uploaderConfig["parallelFilesDownload"] = spec.UploaderConfig.ParallelFilesDownload
+		}
+		specInfo["uploaderConfig"] = uploaderConfig
+	}
+
+	d.Describe("spec", specInfo)
+}
+
+func describeResourceModifierInSF(resModifier *corev1api.TypedLocalObjectReference) map[string]any {
+	return map[string]any{
+		"type": resModifier.Kind,
+		"name": resModifier.Name,
+	}
+}
+
+func describePodVolumeRestoresInSF(d *StructuredDescriber, restores []velerov1api.PodVolumeRestore, details bool) {
+	if len(restores) == 0 {
+		d.Describe("podVolumeRestores", "<none included>")
+		return
+	}
+
+	uploaderType := restores[0].Spec.UploaderType
+	podVolumeInfo := map[string]any{
+		"uploaderType": uploaderType,
+	}
+
+	restoresByPhase := groupRestoresByPhase(restores)
+
+	for _, phase := range []string{
+		string(velerov1api.PodVolumeRestorePhaseCompleted),
+		string(velerov1api.PodVolumeRestorePhaseCanceled),
+		string(velerov1api.PodVolumeRestorePhaseFailed),
+		"In Progress",
+		string(velerov1api.PodVolumeRestorePhasePrepared),
+		string(velerov1api.PodVolumeRestorePhaseAccepted),
+		string(velerov1api.PodVolumeRestorePhaseNew),
+	} {
+		if len(restoresByPhase[phase]) == 0 {
+			continue
+		}
+		if !details {
+			podVolumeInfo[phase] = len(restoresByPhase[phase])
+			continue
+		}
+
+		restoresByPod := new(volumesByPod)
+		for _, restore := range restoresByPhase[phase] {
+			restoresByPod.Add(restore.Spec.Pod.Namespace, restore.Spec.Pod.Name, restore.Spec.Volume, phase, restore.Status.Progress, nil)
+		}
+
+		podEntries := make([]map[string]string, 0)
+		for _, restoreGroup := range restoresByPod.Sorted() {
+			podEntries = append(podEntries, map[string]string{
+				restoreGroup.label: strings.Join(restoreGroup.volumes, ", "),
+			})
+		}
+		podVolumeInfo[phase] = podEntries
+	}
+
+	d.Describe("podVolumeRestores", podVolumeInfo)
+}
+
+func describeRestoreCSISnapshotsInSF(ctx context.Context, kbClient kbclient.Client, d *StructuredDescriber, restore *velerov1api.Restore, details bool, insecureSkipTLSVerify bool, caCertFile string) {
+	bslCACert, err := cacert.GetCACertFromRestore(ctx, kbClient, restore.Namespace, restore)
+	if err != nil {
+		bslCACert = ""
+	}
+
+	buf := new(bytes.Buffer)
+	if err := downloadrequest.StreamWithBSLCACert(ctx, kbClient, restore.Namespace, restore.Name, velerov1api.DownloadTargetKindRestoreVolumeInfo,
+		buf, downloadRequestTimeout, insecureSkipTLSVerify, caCertFile, bslCACert); err != nil {
+		if !errors.Is(err, downloadrequest.ErrNotFound) {
+			d.Describe("csiSnapshotRestores", fmt.Sprintf("<error getting restore volume info: %v>", err))
+		}
+		return
+	}
+
+	describeCSISnapshotsRestoresFromReader(d, buf, details)
+}
+
+func describeCSISnapshotsRestoresFromReader(d *StructuredDescriber, r io.Reader, details bool) {
+	var restoreVolInfo []volume.RestoreVolumeInfo
+	if err := json.NewDecoder(r).Decode(&restoreVolInfo); err != nil {
+		d.Describe("csiSnapshotRestores", fmt.Sprintf("<error reading restore volume info: %v>", err))
+		return
+	}
+	describeCSISnapshotsRestoresInSF(d, restoreVolInfo, details)
+}
+
+func describeCSISnapshotsRestoresInSF(d *StructuredDescriber, restoreVolInfo []volume.RestoreVolumeInfo, details bool) {
+	var nonDMInfoList, dmInfoList []volume.RestoreVolumeInfo
+	for _, info := range restoreVolInfo {
+		if info.RestoreMethod != volume.CSISnapshot {
+			continue
+		}
+		if info.SnapshotDataMoved {
+			dmInfoList = append(dmInfoList, info)
+		} else {
+			nonDMInfoList = append(nonDMInfoList, info)
+		}
+	}
+
+	if len(nonDMInfoList) == 0 && len(dmInfoList) == 0 {
+		d.Describe("csiSnapshotRestores", "<none included>")
+		return
+	}
+
+	csiRestores := map[string]any{}
+
+	for _, info := range nonDMInfoList {
+		key := fmt.Sprintf("%s/%s", info.PVCNamespace, info.PVCName)
+		if details {
+			if info.CSISnapshotInfo == nil {
+				csiRestores[key] = map[string]any{
+					"snapshot": "<CSI snapshot info not found>",
+				}
+				continue
+			}
+			csiRestores[key] = map[string]any{
+				"snapshot": map[string]any{
+					"snapshotContentName": info.CSISnapshotInfo.VSCName,
+					"storageSnapshotID":   info.CSISnapshotInfo.SnapshotHandle,
+					"csiDriver":           info.CSISnapshotInfo.Driver,
+				},
+			}
+		} else {
+			csiRestores[key] = map[string]any{
+				"snapshot": "specify --details for more information",
+			}
+		}
+	}
+
+	for _, info := range dmInfoList {
+		key := fmt.Sprintf("%s/%s", info.PVCNamespace, info.PVCName)
+		if details {
+			if info.SnapshotDataMovementInfo == nil {
+				csiRestores[key] = map[string]any{
+					"dataMovement": "<snapshot data movement info not found>",
+				}
+				continue
+			}
+			csiRestores[key] = map[string]any{
+				"dataMovement": map[string]any{
+					"operationID":  info.SnapshotDataMovementInfo.OperationID,
+					"dataMover":    info.SnapshotDataMovementInfo.DataMover,
+					"uploaderType": info.SnapshotDataMovementInfo.UploaderType,
+				},
+			}
+		} else {
+			csiRestores[key] = map[string]any{
+				"dataMovement": "specify --details for more information",
+			}
+		}
+	}
+
+	d.Describe("csiSnapshotRestores", csiRestores)
+}
+
+func describeRestoreResultsInSF(ctx context.Context, kbClient kbclient.Client, d *StructuredDescriber, restore *velerov1api.Restore, insecureSkipTLSVerify bool, caCertPath string) {
+	if restore.Status.Warnings == 0 && restore.Status.Errors == 0 {
+		return
+	}
+
+	bslCACert, err := cacert.GetCACertFromRestore(ctx, kbClient, restore.Namespace, restore)
+	if err != nil {
+		bslCACert = ""
+	}
+
+	var buf bytes.Buffer
+
+	warnings, errs := make(map[string]any), make(map[string]any)
+	defer func() {
+		if restore.Status.Warnings > 0 {
+			d.Describe("warnings", warnings)
+		}
+		if restore.Status.Errors > 0 {
+			d.Describe("errors", errs)
+		}
+	}()
+
+	if err := downloadrequest.StreamWithBSLCACert(ctx, kbClient, restore.Namespace, restore.Name, velerov1api.DownloadTargetKindRestoreResults, &buf, downloadRequestTimeout, insecureSkipTLSVerify, caCertPath, bslCACert); err != nil {
+		if restore.Status.Warnings > 0 {
+			warnings["errorGettingWarnings"] = fmt.Sprintf("<error getting warnings: %v>", err)
+		}
+		if restore.Status.Errors > 0 {
+			errs["errorGettingErrors"] = fmt.Sprintf("<error getting errors: %v>", err)
+		}
+		return
+	}
+
+	describeRestoreResultsFromReader(warnings, errs, &buf, restore)
+}
+
+func describeRestoreResultsFromReader(warnings, errs map[string]any, r io.Reader, restore *velerov1api.Restore) {
+	var resultMap map[string]results.Result
+	if err := json.NewDecoder(r).Decode(&resultMap); err != nil {
+		if restore.Status.Warnings > 0 {
+			warnings["errorDecodingWarnings"] = fmt.Sprintf("<error decoding warnings: %v>", err)
+		}
+		if restore.Status.Errors > 0 {
+			errs["errorDecodingErrors"] = fmt.Sprintf("<error decoding errors: %v>", err)
+		}
+		return
+	}
+
+	if restore.Status.Warnings > 0 {
+		describeResultInSF(warnings, resultMap["warnings"])
+	}
+	if restore.Status.Errors > 0 {
+		describeResultInSF(errs, resultMap["errors"])
+	}
+}
+
+func describeRestoreItemOperationsInSF(ctx context.Context, kbClient kbclient.Client, d *StructuredDescriber, restore *velerov1api.Restore, details bool, insecureSkipTLSVerify bool, caCertPath string) {
+	status := restore.Status
+	if status.RestoreItemOperationsAttempted == 0 {
+		return
+	}
+
+	opsInfo := map[string]any{
+		"attempted": status.RestoreItemOperationsAttempted,
+		"completed": status.RestoreItemOperationsCompleted,
+		"failed":    status.RestoreItemOperationsFailed,
+	}
+
+	if !details {
+		d.Describe("restoreItemOperations", opsInfo)
+		return
+	}
+
+	bslCACert, err := cacert.GetCACertFromRestore(ctx, kbClient, restore.Namespace, restore)
+	if err != nil {
+		bslCACert = ""
+	}
+
+	buf := new(bytes.Buffer)
+	if err := downloadrequest.StreamWithBSLCACert(ctx, kbClient, restore.Namespace, restore.Name, velerov1api.DownloadTargetKindRestoreItemOperations, buf, downloadRequestTimeout, insecureSkipTLSVerify, caCertPath, bslCACert); err != nil {
+		opsInfo["errorGettingOperations"] = fmt.Sprintf("<error getting operation info: %v>", err)
+		d.Describe("restoreItemOperations", opsInfo)
+		return
+	}
+
+	describeRestoreItemOperationsFromReader(d, opsInfo, buf)
+}
+
+func describeRestoreItemOperationsFromReader(d *StructuredDescriber, opsInfo map[string]any, r io.Reader) {
+	var operations []*itemoperation.RestoreOperation
+	if err := json.NewDecoder(r).Decode(&operations); err != nil {
+		opsInfo["errorReadingOperations"] = fmt.Sprintf("<error reading operation info: %v>", err)
+		d.Describe("restoreItemOperations", opsInfo)
+		return
+	}
+
+	opsList := make([]map[string]any, 0, len(operations))
+	for _, op := range operations {
+		opsList = append(opsList, describeRestoreItemOperationInSF(op))
+	}
+	opsInfo["operations"] = opsList
+	d.Describe("restoreItemOperations", opsInfo)
+}
+
+func describeRestoreItemOperationInSF(op *itemoperation.RestoreOperation) map[string]any {
+	opEntry := map[string]any{
+		"resource":                fmt.Sprintf("%s %s/%s", op.Spec.ResourceIdentifier, op.Spec.ResourceIdentifier.Namespace, op.Spec.ResourceIdentifier.Name),
+		"restoreItemActionPlugin": op.Spec.RestoreItemAction,
+		"operationID":             op.Spec.OperationID,
+		"phase":                   op.Status.Phase,
+	}
+	if op.Status.Error != "" {
+		opEntry["error"] = op.Status.Error
+	}
+	if op.Status.NTotal > 0 || op.Status.NCompleted > 0 {
+		opEntry["progress"] = map[string]any{
+			"completed": op.Status.NCompleted,
+			"total":     op.Status.NTotal,
+			"units":     op.Status.OperationUnits,
+		}
+	}
+	if op.Status.Description != "" {
+		opEntry["progressDescription"] = op.Status.Description
+	}
+	if op.Status.Created != nil {
+		opEntry["created"] = op.Status.Created.String()
+	}
+	if op.Status.Started != nil {
+		opEntry["started"] = op.Status.Started.String()
+	}
+	if op.Status.Updated != nil {
+		opEntry["updated"] = op.Status.Updated.String()
+	}
+	return opEntry
+}
+
+func describeRestoreResourceListInSF(ctx context.Context, kbClient kbclient.Client, d *StructuredDescriber, restore *velerov1api.Restore, insecureSkipTLSVerify bool, caCertPath string) {
+	bslCACert, err := cacert.GetCACertFromRestore(ctx, kbClient, restore.Namespace, restore)
+	if err != nil {
+		bslCACert = ""
+	}
+
+	buf := new(bytes.Buffer)
+	if err := downloadrequest.StreamWithBSLCACert(ctx, kbClient, restore.Namespace, restore.Name, velerov1api.DownloadTargetKindRestoreResourceList, buf, downloadRequestTimeout, insecureSkipTLSVerify, caCertPath, bslCACert); err != nil {
+		if errors.Is(err, downloadrequest.ErrNotFound) {
+			d.Describe("resourceList", "<restore resource list not found>")
+		} else {
+			d.Describe("resourceList", fmt.Sprintf("<error getting restore resource list: %v>", err))
+		}
+		return
+	}
+
+	describeRestoreResourceListFromReader(d, buf)
+}
+
+func describeRestoreResourceListFromReader(d *StructuredDescriber, r io.Reader) {
+	var resourceList map[string][]string
+	if err := json.NewDecoder(r).Decode(&resourceList); err != nil {
+		d.Describe("resourceList", fmt.Sprintf("<error reading restore resource list: %v>", err))
+		return
+	}
+
+	d.Describe("resourceList", resourceList)
+}

pkg/cmd/util/output/restore_structured_describer_test.go

@@ -0,0 +1,962 @@
+/*
+Copyright the Velero contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package output
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	corev1api "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"github.com/vmware-tanzu/velero/internal/volume"
+	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"github.com/vmware-tanzu/velero/pkg/builder"
+	"github.com/vmware-tanzu/velero/pkg/itemoperation"
+	"github.com/vmware-tanzu/velero/pkg/test"
+	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
+	"github.com/vmware-tanzu/velero/pkg/util/results"
+)
+
+func TestDescribeRestoreProgressInSF(t *testing.T) {
+	testcases := []struct {
+		name   string
+		input  *velerov1api.Restore
+		expect map[string]any
+	}{
+		{
+			name:   "nil progress — nothing added",
+			input:  builder.ForRestore("velero", "r1").Result(),
+			expect: map[string]any{},
+		},
+		{
+			name: "in-progress phase shows estimated labels",
+			input: func() *velerov1api.Restore {
+				r := builder.ForRestore("velero", "r2").Phase(velerov1api.RestorePhaseInProgress).Result()
+				r.Status.Progress = &velerov1api.RestoreProgress{TotalItems: 100, ItemsRestored: 50}
+				return r
+			}(),
+			expect: map[string]any{
+				"progress": map[string]any{
+					"estimatedTotalItemsToBeRestored": 100,
+					"itemsRestoredSoFar":              50,
+				},
+			},
+		},
+		{
+			name: "completed phase shows final labels",
+			input: func() *velerov1api.Restore {
+				r := builder.ForRestore("velero", "r3").Phase(velerov1api.RestorePhaseCompleted).Result()
+				r.Status.Progress = &velerov1api.RestoreProgress{TotalItems: 80, ItemsRestored: 80}
+				return r
+			}(),
+			expect: map[string]any{
+				"progress": map[string]any{
+					"totalItemsToBeRestored": 80,
+					"itemsRestored":          80,
+				},
+			},
+		},
+	}
+	for _, tc := range testcases {
+		t.Run(tc.name, func(tt *testing.T) {
+			sd := &StructuredDescriber{output: make(map[string]any), format: ""}
+			describeRestoreProgressInSF(sd, tc.input)
+			assert.Equal(tt, tc.expect, sd.output)
+		})
+	}
+}
+
+func TestDescribeRestoreTimestampsInSF(t *testing.T) {
+	t1 := time.Date(2024, 1, 10, 12, 0, 0, 0, time.UTC)
+	t2 := time.Date(2024, 1, 10, 13, 0, 0, 0, time.UTC)
+	mt1 := metav1.NewTime(t1)
+	mt2 := metav1.NewTime(t2)
+
+	testcases := []struct {
+		name   string
+		input  *velerov1api.Restore
+		expect map[string]any
+	}{
+		{
+			name:  "nil timestamps show <n/a>",
+			input: builder.ForRestore("velero", "r1").Result(),
+			expect: map[string]any{
+				"timestamps": map[string]any{
+					"started":   "<n/a>",
+					"completed": "<n/a>",
+				},
+			},
+		},
+		{
+			name: "both timestamps set",
+			input: func() *velerov1api.Restore {
+				r := builder.ForRestore("velero", "r2").Result()
+				r.Status.StartTimestamp = &mt1
+				r.Status.CompletionTimestamp = &mt2
+				return r
+			}(),
+			expect: map[string]any{
+				"timestamps": map[string]any{
+					"started":   mt1.String(),
+					"completed": mt2.String(),
+				},
+			},
+		},
+	}
+	for _, tc := range testcases {
+		t.Run(tc.name, func(tt *testing.T) {
+			sd := &StructuredDescriber{output: make(map[string]any), format: ""}
+			describeRestoreTimestampsInSF(sd, tc.input)
+			assert.Equal(tt, tc.expect, sd.output)
+		})
+	}
+}
+
+func TestDescribeRestoreSpecInSF(t *testing.T) {
+	testcases := []struct {
+		name   string
+		spec   velerov1api.RestoreSpec
+		expect map[string]any
+	}{
+		{
+			name: "minimal spec",
+			spec: velerov1api.RestoreSpec{
+				BackupName: "backup-1",
+			},
+			expect: map[string]any{
+				"spec": map[string]any{
+					"backupName": "backup-1",
+					"namespaces": map[string]any{
+						"included": "all namespaces found in the backup",
+						"excluded": emptyDisplay,
+					},
+					"resources": map[string]string{
+						"included":      "*",
+						"excluded":      emptyDisplay,
+						"clusterScoped": "auto",
+					},
+					"namespaceMappings":      emptyDisplay,
+					"labelSelector":          emptyDisplay,
+					"orLabelSelectors":       emptyDisplay,
+					"restorePVs":             "auto",
+					"existingResourcePolicy": emptyDisplay,
+					"itemOperationTimeout":   "0s",
+					"preserveNodePorts":      "auto",
+				},
+			},
+		},
+		{
+			name: "included namespaces wildcard treated as all",
+			spec: velerov1api.RestoreSpec{
+				BackupName:             "backup-2",
+				IncludedNamespaces:     []string{"*"},
+				ExcludedNamespaces:     []string{"kube-system"},
+				IncludedResources:      []string{"pods", "configmaps"},
+				ExcludedResources:      []string{"secrets"},
+				ExistingResourcePolicy: velerov1api.ResourcePolicyTypeUpdate,
+			},
+			expect: map[string]any{
+				"spec": map[string]any{
+					"backupName": "backup-2",
+					"namespaces": map[string]any{
+						"included": "all namespaces found in the backup",
+						"excluded": "kube-system",
+					},
+					"resources": map[string]string{
+						"included":      "pods, configmaps",
+						"excluded":      "secrets",
+						"clusterScoped": "auto",
+					},
+					"namespaceMappings":      emptyDisplay,
+					"labelSelector":          emptyDisplay,
+					"orLabelSelectors":       emptyDisplay,
+					"restorePVs":             "auto",
+					"existingResourcePolicy": string(velerov1api.ResourcePolicyTypeUpdate),
+					"itemOperationTimeout":   "0s",
+					"preserveNodePorts":      "auto",
+				},
+			},
+		},
+		{
+			name: "spec with resource modifier and uploader config",
+			spec: velerov1api.RestoreSpec{
+				BackupName: "backup-3",
+				ResourceModifier: &corev1api.TypedLocalObjectReference{
+					Kind: "ConfigMap",
+					Name: "my-modifier",
+				},
+				UploaderConfig: &velerov1api.UploaderConfigForRestore{
+					WriteSparseFiles:      boolptr.True(),
+					ParallelFilesDownload: 4,
+				},
+			},
+			expect: map[string]any{
+				"spec": map[string]any{
+					"backupName": "backup-3",
+					"namespaces": map[string]any{
+						"included": "all namespaces found in the backup",
+						"excluded": emptyDisplay,
+					},
+					"resources": map[string]string{
+						"included":      "*",
+						"excluded":      emptyDisplay,
+						"clusterScoped": "auto",
+					},
+					"namespaceMappings":      emptyDisplay,
+					"labelSelector":          emptyDisplay,
+					"orLabelSelectors":       emptyDisplay,
+					"restorePVs":             "auto",
+					"existingResourcePolicy": emptyDisplay,
+					"itemOperationTimeout":   "0s",
+					"preserveNodePorts":      "auto",
+					"resourceModifier": map[string]any{
+						"type": "ConfigMap",
+						"name": "my-modifier",
+					},
+					"uploaderConfig": map[string]any{
+						"writeSparseFiles":      true,
+						"parallelFilesDownload": 4,
+					},
+				},
+			},
+		},
+		{
+			name: "namespaces, mappings, selectors, resource policy and flags",
+			spec: velerov1api.RestoreSpec{
+				BackupName:              "backup-4",
+				IncludedNamespaces:      []string{"ns-a", "ns-b"},
+				NamespaceMapping:        map[string]string{"ns-a": "ns-a-new"},
+				LabelSelector:           &metav1.LabelSelector{MatchLabels: map[string]string{"app": "nginx"}},
+				OrLabelSelectors:        []*metav1.LabelSelector{{MatchLabels: map[string]string{"env": "prod"}}, {MatchLabels: map[string]string{"env": "stage"}}},
+				IncludeClusterResources: boolptr.True(),
+				RestorePVs:              boolptr.True(),
+				PreserveNodePorts:       boolptr.False(),
+				ResourcePolicy: &corev1api.TypedLocalObjectReference{
+					Kind: "configmap",
+					Name: "volume-policy",
+				},
+				UploaderConfig: &velerov1api.UploaderConfigForRestore{
+					WriteSparseFiles: boolptr.False(),
+				},
+			},
+			expect: map[string]any{
+				"spec": map[string]any{
+					"backupName": "backup-4",
+					"namespaces": map[string]any{
+						"included": "ns-a, ns-b",
+						"excluded": emptyDisplay,
+					},
+					"resources": map[string]string{
+						"included":      "*",
+						"excluded":      emptyDisplay,
+						"clusterScoped": "included",
+					},
+					"namespaceMappings":      map[string]string{"ns-a": "ns-a-new"},
+					"labelSelector":          "app=nginx",
+					"orLabelSelectors":       "env=prod or env=stage",
+					"restorePVs":             "true",
+					"existingResourcePolicy": emptyDisplay,
+					"itemOperationTimeout":   "0s",
+					"preserveNodePorts":      "false",
+					"resourcePolicy": map[string]any{
+						"type": "configmap",
+						"name": "volume-policy",
+					},
+					"uploaderConfig": map[string]any{},
+				},
+			},
+		},
+	}
+	for _, tc := range testcases {
+		t.Run(tc.name, func(tt *testing.T) {
+			sd := &StructuredDescriber{output: make(map[string]any), format: ""}
+			describeRestoreSpecInSF(sd, tc.spec)
+			assert.Equal(tt, tc.expect, sd.output)
+		})
+	}
+}
+
+func TestDescribePodVolumeRestoresInSF(t *testing.T) {
+	pvr1 := builder.ForPodVolumeRestore("velero", "pvr-1").
+		UploaderType("kopia").
+		Phase(velerov1api.PodVolumeRestorePhaseCompleted).
+		Volume("vol-1").
+		PodName("pod-1").
+		PodNamespace("ns-1").Result()
+
+	pvr2 := builder.ForPodVolumeRestore("velero", "pvr-2").
+		UploaderType("kopia").
+		Phase(velerov1api.PodVolumeRestorePhaseCompleted).
+		Volume("vol-2").
+		PodName("pod-2").
+		PodNamespace("ns-1").Result()
+
+	pvr3 := builder.ForPodVolumeRestore("velero", "pvr-3").
+		UploaderType("kopia").
+		Phase(velerov1api.PodVolumeRestorePhaseFailed).
+		Volume("vol-3").
+		PodName("pod-3").
+		PodNamespace("ns-1").Result()
+
+	testcases := []struct {
+		name     string
+		restores []velerov1api.PodVolumeRestore
+		details  bool
+		expect   map[string]any
+	}{
+		{
+			name:     "empty list",
+			restores: []velerov1api.PodVolumeRestore{},
+			details:  false,
+			expect: map[string]any{
+				"podVolumeRestores": "<none included>",
+			},
+		},
+		{
+			name:     "2 completed, no details",
+			restores: []velerov1api.PodVolumeRestore{*pvr1, *pvr2},
+			details:  false,
+			expect: map[string]any{
+				"podVolumeRestores": map[string]any{
+					"uploaderType": "kopia",
+					"Completed":    2,
+				},
+			},
+		},
+		{
+			name:     "2 completed with details",
+			restores: []velerov1api.PodVolumeRestore{*pvr1, *pvr2},
+			details:  true,
+			expect: map[string]any{
+				"podVolumeRestores": map[string]any{
+					"uploaderType": "kopia",
+					"Completed": []map[string]string{
+						{"ns-1/pod-1": "vol-1"},
+						{"ns-1/pod-2": "vol-2"},
+					},
+				},
+			},
+		},
+		{
+			name:     "completed and failed, no details",
+			restores: []velerov1api.PodVolumeRestore{*pvr1, *pvr2, *pvr3},
+			details:  false,
+			expect: map[string]any{
+				"podVolumeRestores": map[string]any{
+					"uploaderType": "kopia",
+					"Completed":    2,
+					"Failed":       1,
+				},
+			},
+		},
+	}
+	for _, tc := range testcases {
+		t.Run(tc.name, func(tt *testing.T) {
+			sd := &StructuredDescriber{output: make(map[string]any), format: ""}
+			describePodVolumeRestoresInSF(sd, tc.restores, tc.details)
+			assert.Equal(tt, tc.expect, sd.output)
+		})
+	}
+}
+
+func TestDescribeRestoreCSISnapshotsInSF_NoData(t *testing.T) {
+	testcases := []struct {
+		name             string
+		inputVolInfoList []volume.RestoreVolumeInfo
+		details          bool
+		expect           map[string]any
+	}{
+		{
+			name:             "no CSI entries — none included",
+			inputVolInfoList: []volume.RestoreVolumeInfo{},
+			details:          false,
+			expect: map[string]any{
+				"csiSnapshotRestores": "<none included>",
+			},
+		},
+		{
+			name: "only native snapshot entries — none included",
+			inputVolInfoList: []volume.RestoreVolumeInfo{
+				{
+					RestoreMethod: volume.NativeSnapshot,
+					PVCName:       "pvc-1",
+					PVCNamespace:  "ns-1",
+				},
+			},
+			details: false,
+			expect: map[string]any{
+				"csiSnapshotRestores": "<none included>",
+			},
+		},
+		{
+			name: "CSI snapshot, no details",
+			inputVolInfoList: []volume.RestoreVolumeInfo{
+				{
+					RestoreMethod: volume.CSISnapshot,
+					PVCName:       "pvc-1",
+					PVCNamespace:  "ns-1",
+					CSISnapshotInfo: &volume.CSISnapshotInfo{
+						VSCName:        "vsc-1",
+						SnapshotHandle: "snap-handle-1",
+						Driver:         "csi.test.driver",
+					},
+				},
+			},
+			details: false,
+			expect: map[string]any{
+				"csiSnapshotRestores": map[string]any{
+					"ns-1/pvc-1": map[string]any{
+						"snapshot": "specify --details for more information",
+					},
+				},
+			},
+		},
+		{
+			name: "CSI snapshot, with details",
+			inputVolInfoList: []volume.RestoreVolumeInfo{
+				{
+					RestoreMethod: volume.CSISnapshot,
+					PVCName:       "pvc-2",
+					PVCNamespace:  "ns-2",
+					CSISnapshotInfo: &volume.CSISnapshotInfo{
+						VSCName:        "vsc-2",
+						SnapshotHandle: "snap-handle-2",
+						Driver:         "csi.test.driver",
+					},
+				},
+			},
+			details: true,
+			expect: map[string]any{
+				"csiSnapshotRestores": map[string]any{
+					"ns-2/pvc-2": map[string]any{
+						"snapshot": map[string]any{
+							"snapshotContentName": "vsc-2",
+							"storageSnapshotID":   "snap-handle-2",
+							"csiDriver":           "csi.test.driver",
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "data movement entry, with details",
+			inputVolInfoList: []volume.RestoreVolumeInfo{
+				{
+					RestoreMethod:     volume.CSISnapshot,
+					SnapshotDataMoved: true,
+					PVCName:           "pvc-3",
+					PVCNamespace:      "ns-3",
+					SnapshotDataMovementInfo: &volume.SnapshotDataMovementInfo{
+						OperationID:  "op-3",
+						DataMover:    "velero",
+						UploaderType: "kopia",
+					},
+				},
+			},
+			details: true,
+			expect: map[string]any{
+				"csiSnapshotRestores": map[string]any{
+					"ns-3/pvc-3": map[string]any{
+						"dataMovement": map[string]any{
+							"operationID":  "op-3",
+							"dataMover":    "velero",
+							"uploaderType": "kopia",
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "data movement entry, no details",
+			inputVolInfoList: []volume.RestoreVolumeInfo{
+				{
+					RestoreMethod:     volume.CSISnapshot,
+					SnapshotDataMoved: true,
+					PVCName:           "pvc-3",
+					PVCNamespace:      "ns-3",
+					SnapshotDataMovementInfo: &volume.SnapshotDataMovementInfo{
+						OperationID:  "op-3",
+						DataMover:    "velero",
+						UploaderType: "kopia",
+					},
+				},
+			},
+			details: false,
+			expect: map[string]any{
+				"csiSnapshotRestores": map[string]any{
+					"ns-3/pvc-3": map[string]any{
+						"dataMovement": "specify --details for more information",
+					},
+				},
+			},
+		},
+		{
+			name: "CSI snapshot with details and nil CSISnapshotInfo",
+			inputVolInfoList: []volume.RestoreVolumeInfo{
+				{
+					RestoreMethod:   volume.CSISnapshot,
+					PVCName:         "pvc-4",
+					PVCNamespace:    "ns-4",
+					CSISnapshotInfo: nil,
+				},
+			},
+			details: true,
+			expect: map[string]any{
+				"csiSnapshotRestores": map[string]any{
+					"ns-4/pvc-4": map[string]any{
+						"snapshot": "<CSI snapshot info not found>",
+					},
+				},
+			},
+		},
+		{
+			name: "data movement with details and nil SnapshotDataMovementInfo",
+			inputVolInfoList: []volume.RestoreVolumeInfo{
+				{
+					RestoreMethod:            volume.CSISnapshot,
+					SnapshotDataMoved:        true,
+					PVCName:                  "pvc-5",
+					PVCNamespace:             "ns-5",
+					SnapshotDataMovementInfo: nil,
+				},
+			},
+			details: true,
+			expect: map[string]any{
+				"csiSnapshotRestores": map[string]any{
+					"ns-5/pvc-5": map[string]any{
+						"dataMovement": "<snapshot data movement info not found>",
+					},
+				},
+			},
+		},
+	}
+
+	for _, tc := range testcases {
+		t.Run(tc.name, func(tt *testing.T) {
+			sd := &StructuredDescriber{output: make(map[string]any), format: ""}
+			describeCSISnapshotsRestoresInSF(sd, tc.inputVolInfoList, tc.details)
+			assert.Equal(tt, tc.expect, sd.output)
+		})
+	}
+}
+
+func TestDescribeCSISnapshotsRestoresFromReader(t *testing.T) {
+	t.Run("invalid json", func(t *testing.T) {
+		sd := &StructuredDescriber{output: make(map[string]any), format: ""}
+		describeCSISnapshotsRestoresFromReader(sd, strings.NewReader("not-json"), false)
+		got, ok := sd.output["csiSnapshotRestores"].(string)
+		require.True(t, ok)
+		assert.Contains(t, got, "<error reading restore volume info:")
+	})
+
+	t.Run("valid json", func(t *testing.T) {
+		volInfo := []volume.RestoreVolumeInfo{
+			{
+				RestoreMethod: volume.CSISnapshot,
+				PVCName:       "pvc-1",
+				PVCNamespace:  "ns-1",
+				CSISnapshotInfo: &volume.CSISnapshotInfo{
+					VSCName:        "vsc-1",
+					SnapshotHandle: "snap-handle-1",
+					Driver:         "csi.test.driver",
+				},
+			},
+		}
+		data, err := json.Marshal(volInfo)
+		require.NoError(t, err)
+
+		sd := &StructuredDescriber{output: make(map[string]any), format: ""}
+		describeCSISnapshotsRestoresFromReader(sd, bytes.NewReader(data), false)
+		assert.Equal(t, map[string]any{
+			"csiSnapshotRestores": map[string]any{
+				"ns-1/pvc-1": map[string]any{
+					"snapshot": "specify --details for more information",
+				},
+			},
+		}, sd.output)
+	})
+}
+
+func TestDescribeRestoreItemOperationsInSF_NoDownload(t *testing.T) {
+	kbClient := test.NewFakeControllerRuntimeClient(t)
+	testcases := []struct {
+		name   string
+		status velerov1api.RestoreStatus
+		expect map[string]any
+	}{
+		{
+			name:   "zero operations — nothing added",
+			status: velerov1api.RestoreStatus{},
+			expect: map[string]any{},
+		},
+		{
+			name: "some operations, no details",
+			status: velerov1api.RestoreStatus{
+				RestoreItemOperationsAttempted: 5,
+				RestoreItemOperationsCompleted: 4,
+				RestoreItemOperationsFailed:    1,
+			},
+			expect: map[string]any{
+				"restoreItemOperations": map[string]any{
+					"attempted": 5,
+					"completed": 4,
+					"failed":    1,
+				},
+			},
+		},
+	}
+
+	for _, tc := range testcases {
+		t.Run(tc.name, func(tt *testing.T) {
+			restore := builder.ForRestore("velero", "r1").Result()
+			restore.Status = tc.status
+
+			sd := &StructuredDescriber{output: make(map[string]any), format: ""}
+			describeRestoreItemOperationsInSF(context.Background(), kbClient, sd, restore, false, false, "")
+			assert.Equal(tt, tc.expect, sd.output)
+		})
+	}
+}
+
+func TestDescribeRestoreItemOperationsInSF_DownloadError(t *testing.T) {
+	kbClient := test.NewFakeControllerRuntimeClient(t)
+	restore := builder.ForRestore("velero", "r1").Result()
+	restore.Status.RestoreItemOperationsAttempted = 2
+	restore.Status.RestoreItemOperationsCompleted = 1
+	restore.Status.RestoreItemOperationsFailed = 1
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	sd := &StructuredDescriber{output: make(map[string]any), format: ""}
+	describeRestoreItemOperationsInSF(ctx, kbClient, sd, restore, true, false, "")
+
+	ops, ok := sd.output["restoreItemOperations"].(map[string]any)
+	require.True(t, ok)
+	assert.Equal(t, 2, ops["attempted"])
+	assert.Equal(t, 1, ops["completed"])
+	assert.Equal(t, 1, ops["failed"])
+	_, hasErr := ops["errorGettingOperations"]
+	assert.True(t, hasErr)
+}
+
+func TestDescribeRestoreItemOperationInSF(t *testing.T) {
+	t1 := time.Date(2023, 6, 26, 0, 0, 0, 0, time.UTC)
+	t2 := time.Date(2023, 6, 25, 0, 0, 0, 0, time.UTC)
+	t3 := time.Date(2023, 6, 24, 0, 0, 0, 0, time.UTC)
+
+	input := builder.ForRestoreOperation().
+		RestoreName("restore-1").
+		OperationID("op-1").
+		RestoreItemAction("action-1").
+		ResourceIdentifier("group", "rs-type", "ns", "rs-name").
+		Status(*builder.ForOperationStatus().
+			Phase(itemoperation.OperationPhaseFailed).
+			Error("operation error").
+			Progress(50, 100, "bytes").
+			Description("operation description").
+			Created(t3).
+			Started(t2).
+			Updated(t1).
+			Result()).Result()
+
+	got := describeRestoreItemOperationInSF(input)
+	assert.Equal(t, "operation error", got["error"])
+	assert.Equal(t, "op-1", got["operationID"])
+	assert.Equal(t, "action-1", got["restoreItemActionPlugin"])
+	assert.Equal(t, itemoperation.OperationPhaseFailed, got["phase"])
+	assert.Equal(t, "operation description", got["progressDescription"])
+	assert.Equal(t, t3.String(), got["created"])
+	assert.Equal(t, t2.String(), got["started"])
+	assert.Equal(t, t1.String(), got["updated"])
+	assert.Equal(t, map[string]any{
+		"completed": int64(50),
+		"total":     int64(100),
+		"units":     "bytes",
+	}, got["progress"])
+}
+
+func TestDescribeRestoreItemOperationsFromReader(t *testing.T) {
+	t.Run("invalid json", func(t *testing.T) {
+		sd := &StructuredDescriber{output: make(map[string]any), format: ""}
+		opsInfo := map[string]any{"attempted": 1}
+		describeRestoreItemOperationsFromReader(sd, opsInfo, strings.NewReader("not-json"))
+		got, ok := sd.output["restoreItemOperations"].(map[string]any)
+		require.True(t, ok)
+		_, hasErr := got["errorReadingOperations"]
+		assert.True(t, hasErr)
+	})
+
+	t.Run("valid json", func(t *testing.T) {
+		op := builder.ForRestoreOperation().
+			RestoreName("restore-1").
+			OperationID("op-1").
+			RestoreItemAction("action-1").
+			ResourceIdentifier("group", "rs-type", "ns", "rs-name").
+			Status(*builder.ForOperationStatus().Phase(itemoperation.OperationPhaseCompleted).Result()).
+			Result()
+		data, err := json.Marshal([]*itemoperation.RestoreOperation{op})
+		require.NoError(t, err)
+
+		sd := &StructuredDescriber{output: make(map[string]any), format: ""}
+		opsInfo := map[string]any{"attempted": 1, "completed": 1, "failed": 0}
+		describeRestoreItemOperationsFromReader(sd, opsInfo, bytes.NewReader(data))
+
+		got, ok := sd.output["restoreItemOperations"].(map[string]any)
+		require.True(t, ok)
+		ops, ok := got["operations"].([]map[string]any)
+		require.True(t, ok)
+		require.Len(t, ops, 1)
+		assert.Equal(t, "op-1", ops[0]["operationID"])
+	})
+}
+
+func TestDescribeResourceModifierInSF(t *testing.T) {
+	input := &corev1api.TypedLocalObjectReference{
+		Kind: "ConfigMap",
+		Name: "my-modifier",
+	}
+	expect := map[string]any{
+		"type": "ConfigMap",
+		"name": "my-modifier",
+	}
+	assert.Equal(t, expect, describeResourceModifierInSF(input))
+}
+
+func TestDescribeRestoreResultsFromReader(t *testing.T) {
+	restoreBoth := builder.ForRestore("velero", "r1").Result()
+	restoreBoth.Status.Warnings = 2
+	restoreBoth.Status.Errors = 1
+
+	t.Run("invalid json", func(t *testing.T) {
+		warnings, errs := make(map[string]any), make(map[string]any)
+		describeRestoreResultsFromReader(warnings, errs, strings.NewReader("not-json"), restoreBoth)
+		_, hasWarn := warnings["errorDecodingWarnings"]
+		_, hasErr := errs["errorDecodingErrors"]
+		assert.True(t, hasWarn)
+		assert.True(t, hasErr)
+	})
+
+	t.Run("valid json", func(t *testing.T) {
+		payload := map[string]results.Result{
+			"warnings": {
+				Velero:  []string{"w1"},
+				Cluster: []string{"c1"},
+				Namespaces: map[string][]string{
+					"ns-1": {"n1"},
+				},
+			},
+			"errors": {
+				Velero: []string{"e1"},
+			},
+		}
+		data, err := json.Marshal(payload)
+		require.NoError(t, err)
+
+		warnings, errs := make(map[string]any), make(map[string]any)
+		describeRestoreResultsFromReader(warnings, errs, bytes.NewReader(data), restoreBoth)
+		assert.Equal(t, []string{"w1"}, warnings["velero"])
+		assert.Equal(t, []string{"c1"}, warnings["cluster"])
+		assert.Equal(t, map[string][]string{"ns-1": {"n1"}}, warnings["namespace"])
+		assert.Equal(t, []string{"e1"}, errs["velero"])
+	})
+
+	t.Run("warnings only decode error", func(t *testing.T) {
+		restore := builder.ForRestore("velero", "r2").Result()
+		restore.Status.Warnings = 1
+		warnings, errs := make(map[string]any), make(map[string]any)
+		describeRestoreResultsFromReader(warnings, errs, strings.NewReader("not-json"), restore)
+		_, hasWarn := warnings["errorDecodingWarnings"]
+		assert.True(t, hasWarn)
+		assert.Empty(t, errs)
+	})
+
+	t.Run("errors only decode error", func(t *testing.T) {
+		restore := builder.ForRestore("velero", "r3").Result()
+		restore.Status.Errors = 1
+		warnings, errs := make(map[string]any), make(map[string]any)
+		describeRestoreResultsFromReader(warnings, errs, strings.NewReader("not-json"), restore)
+		_, hasErr := errs["errorDecodingErrors"]
+		assert.True(t, hasErr)
+		assert.Empty(t, warnings)
+	})
+}
+
+func TestDescribeRestoreResultsInSF(t *testing.T) {
+	kbClient := test.NewFakeControllerRuntimeClient(t)
+
+	t.Run("no warnings or errors", func(t *testing.T) {
+		sd := &StructuredDescriber{output: make(map[string]any), format: ""}
+		restore := builder.ForRestore("velero", "r1").Result()
+		describeRestoreResultsInSF(context.Background(), kbClient, sd, restore, false, "")
+		assert.Empty(t, sd.output)
+	})
+
+	t.Run("download error", func(t *testing.T) {
+		restore := builder.ForRestore("velero", "r1").Result()
+		restore.Status.Warnings = 1
+		restore.Status.Errors = 1
+
+		ctx, cancel := context.WithCancel(context.Background())
+		cancel()
+
+		sd := &StructuredDescriber{output: make(map[string]any), format: ""}
+		describeRestoreResultsInSF(ctx, kbClient, sd, restore, false, "")
+
+		warnings, ok := sd.output["warnings"].(map[string]any)
+		require.True(t, ok)
+		_, hasWarn := warnings["errorGettingWarnings"]
+		assert.True(t, hasWarn)
+
+		errs, ok := sd.output["errors"].(map[string]any)
+		require.True(t, ok)
+		_, hasErr := errs["errorGettingErrors"]
+		assert.True(t, hasErr)
+	})
+
+	t.Run("download error warnings only", func(t *testing.T) {
+		restore := builder.ForRestore("velero", "r1").Result()
+		restore.Status.Warnings = 1
+		ctx, cancel := context.WithCancel(context.Background())
+		cancel()
+
+		sd := &StructuredDescriber{output: make(map[string]any), format: ""}
+		describeRestoreResultsInSF(ctx, kbClient, sd, restore, false, "")
+		_, hasWarn := sd.output["warnings"]
+		_, hasErr := sd.output["errors"]
+		assert.True(t, hasWarn)
+		assert.False(t, hasErr)
+	})
+
+	t.Run("download error errors only", func(t *testing.T) {
+		restore := builder.ForRestore("velero", "r1").Result()
+		restore.Status.Errors = 1
+		ctx, cancel := context.WithCancel(context.Background())
+		cancel()
+
+		sd := &StructuredDescriber{output: make(map[string]any), format: ""}
+		describeRestoreResultsInSF(ctx, kbClient, sd, restore, false, "")
+		_, hasWarn := sd.output["warnings"]
+		_, hasErr := sd.output["errors"]
+		assert.False(t, hasWarn)
+		assert.True(t, hasErr)
+	})
+}
+
+func TestDescribeRestoreResourceListFromReader(t *testing.T) {
+	t.Run("invalid json", func(t *testing.T) {
+		sd := &StructuredDescriber{output: make(map[string]any), format: ""}
+		describeRestoreResourceListFromReader(sd, strings.NewReader("not-json"))
+		got, ok := sd.output["resourceList"].(string)
+		require.True(t, ok)
+		assert.Contains(t, got, "<error reading restore resource list:")
+	})
+
+	t.Run("valid json", func(t *testing.T) {
+		sd := &StructuredDescriber{output: make(map[string]any), format: ""}
+		payload := map[string][]string{"v1/Pod": {"ns/pod-1"}}
+		data, err := json.Marshal(payload)
+		require.NoError(t, err)
+		describeRestoreResourceListFromReader(sd, bytes.NewReader(data))
+		assert.Equal(t, payload, sd.output["resourceList"])
+	})
+}
+
+func TestDescribeRestoreResourceListInSF_DownloadError(t *testing.T) {
+	kbClient := test.NewFakeControllerRuntimeClient(t)
+	restore := builder.ForRestore("velero", "r1").Result()
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	sd := &StructuredDescriber{output: make(map[string]any), format: ""}
+	describeRestoreResourceListInSF(ctx, kbClient, sd, restore, false, "")
+	got, ok := sd.output["resourceList"].(string)
+	require.True(t, ok)
+	assert.Contains(t, got, "<error getting restore resource list:")
+}
+
+func TestDescribeRestoreCSISnapshotsInSF_DownloadError(t *testing.T) {
+	kbClient := test.NewFakeControllerRuntimeClient(t)
+	restore := builder.ForRestore("velero", "r1").Result()
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	sd := &StructuredDescriber{output: make(map[string]any), format: ""}
+	describeRestoreCSISnapshotsInSF(ctx, kbClient, sd, restore, false, false, "")
+	got, ok := sd.output["csiSnapshotRestores"].(string)
+	require.True(t, ok)
+	assert.Contains(t, got, "<error getting restore volume info:")
+}
+
+func TestDescribeRestoreInSF(t *testing.T) {
+	kbClient := test.NewFakeControllerRuntimeClient(t)
+	started := metav1.NewTime(time.Date(2024, 1, 10, 12, 0, 0, 0, time.UTC))
+	completed := metav1.NewTime(time.Date(2024, 1, 10, 13, 0, 0, 0, time.UTC))
+	deletedAt := metav1.NewTime(time.Date(2024, 1, 10, 14, 0, 0, 0, time.UTC))
+
+	pvr := builder.ForPodVolumeRestore("velero", "pvr-1").
+		UploaderType("kopia").
+		Phase(velerov1api.PodVolumeRestorePhaseCompleted).
+		Volume("vol-1").
+		PodName("pod-1").
+		PodNamespace("ns-1").Result()
+
+	restore := builder.ForRestore("velero", "restore-1").
+		Backup("backup-1").
+		Phase(velerov1api.RestorePhaseCompleted).
+		ObjectMeta(builder.WithLabels("app", "velero"), builder.WithAnnotations("a", "b")).
+		Result()
+	restore.DeletionTimestamp = &deletedAt
+	restore.Status.StartTimestamp = &started
+	restore.Status.CompletionTimestamp = &completed
+	restore.Status.Progress = &velerov1api.RestoreProgress{TotalItems: 10, ItemsRestored: 10}
+	restore.Status.ValidationErrors = []string{"invalid include"}
+	restore.Status.Warnings = 1
+	restore.Status.Errors = 1
+	restore.Status.RestoreItemOperationsAttempted = 2
+	restore.Status.RestoreItemOperationsCompleted = 2
+	restore.Status.HookStatus = &velerov1api.HookStatus{HooksAttempted: 3, HooksFailed: 1}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	out := DescribeRestoreInSF(ctx, kbClient, restore, []velerov1api.PodVolumeRestore{*pvr}, true, false, "", "json")
+
+	var parsed map[string]any
+	require.NoError(t, json.Unmarshal([]byte(out), &parsed))
+	assert.Equal(t, "Completed (Deleting)", parsed["phase"])
+	assert.Contains(t, parsed, "metadata")
+	assert.Contains(t, parsed, "progress")
+	assert.Contains(t, parsed, "timestamps")
+	assert.Contains(t, parsed, "spec")
+	assert.Contains(t, parsed, "podVolumeRestores")
+	assert.Contains(t, parsed, "hookStatus")
+	assert.Equal(t, []any{"invalid include"}, parsed["validationErrors"])
+
+	t.Run("empty phase defaults to New", func(t *testing.T) {
+		r := builder.ForRestore("velero", "restore-2").Result()
+		out := DescribeRestoreInSF(ctx, kbClient, r, nil, false, false, "", "json")
+		require.Contains(t, out, `"phase": "New"`)
+	})
+}

pkg/exposer/generic_restore.go

@@ -255,27 +255,11 @@ func (e *genericRestoreExposer) Expose(ctx context.Context, ownerObject corev1ap
 	// Get volumeID before creating the restore pod because the existingPV may be deleted when creating the PVC if the volume policy is different
 	var volumeID string
 	if param.CSI != nil && param.CSI.Snapshot != nil {
-		vs := &snapshotv1api.VolumeSnapshot{}
-		if err := e.ctrlClient.Get(ctx, client.ObjectKey{
-			Namespace: param.CSI.Snapshot.VolumeSnapshotNamespace,
-			Name:      param.CSI.Snapshot.VolumeSnapshot,
-		}, vs); err != nil {
-			return errors.Wrapf(err, "error to get volume snapshot %s/%s", param.CSI.Snapshot.VolumeSnapshotNamespace, param.CSI.Snapshot.VolumeSnapshot)
-		}
-
-		var vsc *snapshotv1api.VolumeSnapshotContent
-		vsc, err = csi.GetVSCForVS(ctx, vs, e.ctrlClient)
+		volumeID, err = e.getVolumeID(ctx, param.CSI.Snapshot, param.TargetPVName)
 		if err != nil {
-			return errors.Wrapf(err, "error to get volume snapshot content for volume snapshot %s/%s", vs.Namespace, vs.Name)
+			// only log the error. Without the volume ID, exposer will fallback to full restore.
+			curLog.Errorf("failed to get volume ID from snapshot %s/%s, err: %v", param.CSI.Snapshot.VolumeSnapshotNamespace, param.CSI.Snapshot.VolumeSnapshot, err)
 		}
-
-		var cbtInfo csi.CBTInfo
-		cbtInfo, err = csi.GetCBTInfo(ctx, e.kubeClient, e.log, vs, vsc, param.TargetPVName)
-		if err != nil {
-			return errors.Wrap(err, "error to get CBT info")
-		}
-		curLog.Debugf("CBT info: %+v", cbtInfo)
-		volumeID = cbtInfo.VolumeID
 	}
 
 	curLog.Info("Creating restore PVC")
@@ -1082,3 +1066,25 @@ func (e *genericRestoreExposer) validateSelectedNode(ctx context.Context, node s
 
 	return true
 }
+
+func (e *genericRestoreExposer) getVolumeID(ctx context.Context, snapshot *velerov2alpha1api.CSISnapshotSpec, targetPVName string) (string, error) {
+	vs := &snapshotv1api.VolumeSnapshot{}
+	if err := e.ctrlClient.Get(ctx, client.ObjectKey{
+		Namespace: snapshot.VolumeSnapshotNamespace,
+		Name:      snapshot.VolumeSnapshot,
+	}, vs); err != nil {
+		return "", errors.Wrapf(err, "error to get volume snapshot %s/%s", snapshot.VolumeSnapshotNamespace, snapshot.VolumeSnapshot)
+	}
+
+	vsc, err := csi.GetVSCForVS(ctx, vs, e.ctrlClient)
+	if err != nil {
+		return "", errors.Wrapf(err, "error to get volume snapshot content for volume snapshot %s/%s", vs.Namespace, vs.Name)
+	}
+
+	var cbtInfo csi.CBTInfo
+	cbtInfo, err = csi.GetCBTInfo(ctx, e.kubeClient, e.log, vs, vsc, targetPVName)
+	if err != nil {
+		return "", errors.Wrap(err, "error to get CBT info")
+	}
+	return cbtInfo.VolumeID, nil
+}

pkg/exposer/generic_restore_test.go

@@ -21,6 +21,7 @@ import (
 	"time"
 
 	"github.com/cockroachdb/errors"
+	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v8/apis/volumesnapshot/v1"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	appsv1api "k8s.io/api/apps/v1"
@@ -33,8 +34,10 @@ import (
 	clientTesting "k8s.io/client-go/testing"
 
 	velerov1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	velerov2alpha1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
 	velerotypes "github.com/vmware-tanzu/velero/pkg/types"
+	"github.com/vmware-tanzu/velero/pkg/util"
 	"github.com/vmware-tanzu/velero/pkg/util/datamover"
 	"github.com/vmware-tanzu/velero/pkg/util/kube"
 )
@@ -716,6 +719,336 @@ func TestRestoreExpose_SecretCopy(t *testing.T) {
 	})
 }
 
+func TestGetVolumeID(t *testing.T) {
+	vscName := "fake-vsc"
+	snapshotHandle := "fake-snapshot-handle"
+
+	tests := []struct {
+		name          string
+		snapshot      *velerov2alpha1api.CSISnapshotSpec
+		targetPVName  string
+		ctrlClientObj []runtime.Object
+		kubeClientObj []runtime.Object
+		expectedID    string
+		expectedErr   string
+	}{
+		{
+			name: "VS not found in ctrlClient",
+			snapshot: &velerov2alpha1api.CSISnapshotSpec{
+				VolumeSnapshot:          "non-existent-vs",
+				VolumeSnapshotNamespace: "fake-ns",
+			},
+			expectedErr: "error to get volume snapshot fake-ns/non-existent-vs",
+		},
+		{
+			name: "GetVSCForVS error - VS has no bound VSC",
+			snapshot: &velerov2alpha1api.CSISnapshotSpec{
+				VolumeSnapshot:          "fake-vs",
+				VolumeSnapshotNamespace: "fake-ns",
+			},
+			ctrlClientObj: []runtime.Object{
+				&snapshotv1api.VolumeSnapshot{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: "fake-ns",
+						Name:      "fake-vs",
+					},
+					Status: nil,
+				},
+			},
+			expectedErr: "error to get volume snapshot content for volume snapshot fake-ns/fake-vs: invalid snapshot info in volume snapshot fake-vs",
+		},
+		{
+			name: "GetVSCForVS error - VSC not found in ctrlClient",
+			snapshot: &velerov2alpha1api.CSISnapshotSpec{
+				VolumeSnapshot:          "fake-vs",
+				VolumeSnapshotNamespace: "fake-ns",
+			},
+			ctrlClientObj: []runtime.Object{
+				&snapshotv1api.VolumeSnapshot{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: "fake-ns",
+						Name:      "fake-vs",
+					},
+					Status: &snapshotv1api.VolumeSnapshotStatus{
+						BoundVolumeSnapshotContentName: &vscName,
+					},
+				},
+			},
+			expectedErr: "error to get volume snapshot content for volume snapshot fake-ns/fake-vs: error getting volume snapshot content from API",
+		},
+		{
+			name: "GetCBTInfo error - target PV not found",
+			snapshot: &velerov2alpha1api.CSISnapshotSpec{
+				VolumeSnapshot:          "fake-vs",
+				VolumeSnapshotNamespace: "fake-ns",
+			},
+			targetPVName: "missing-pv",
+			ctrlClientObj: []runtime.Object{
+				&snapshotv1api.VolumeSnapshot{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: "fake-ns",
+						Name:      "fake-vs",
+					},
+					Status: &snapshotv1api.VolumeSnapshotStatus{
+						BoundVolumeSnapshotContentName: &vscName,
+					},
+				},
+				&snapshotv1api.VolumeSnapshotContent{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: vscName,
+					},
+					Status: &snapshotv1api.VolumeSnapshotContentStatus{
+						SnapshotHandle: &snapshotHandle,
+					},
+				},
+			},
+			expectedErr: "error to get CBT info: failed to get pv missing-pv",
+		},
+		{
+			name: "GetCBTInfo error - empty volumeID on PV",
+			snapshot: &velerov2alpha1api.CSISnapshotSpec{
+				VolumeSnapshot:          "fake-vs",
+				VolumeSnapshotNamespace: "fake-ns",
+			},
+			targetPVName: "fake-pv",
+			ctrlClientObj: []runtime.Object{
+				&snapshotv1api.VolumeSnapshot{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: "fake-ns",
+						Name:      "fake-vs",
+					},
+					Status: &snapshotv1api.VolumeSnapshotStatus{
+						BoundVolumeSnapshotContentName: &vscName,
+					},
+				},
+				&snapshotv1api.VolumeSnapshotContent{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: vscName,
+					},
+					Status: &snapshotv1api.VolumeSnapshotContentStatus{
+						SnapshotHandle: &snapshotHandle,
+					},
+				},
+			},
+			kubeClientObj: []runtime.Object{
+				&corev1api.PersistentVolume{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "fake-pv",
+					},
+				},
+			},
+			expectedErr: "error to get CBT info: volumeID must not be empty for CBT",
+		},
+		{
+			name: "success with VKS annotations",
+			snapshot: &velerov2alpha1api.CSISnapshotSpec{
+				VolumeSnapshot:          "fake-vs",
+				VolumeSnapshotNamespace: "fake-ns",
+			},
+			ctrlClientObj: []runtime.Object{
+				&snapshotv1api.VolumeSnapshot{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: "fake-ns",
+						Name:      "fake-vs",
+						Annotations: map[string]string{
+							util.VSphereCNSChangeIDAnno: "c-1",
+							util.VSphereCNSSnapshotAnno: "vol-vks+snap-1",
+						},
+					},
+					Status: &snapshotv1api.VolumeSnapshotStatus{
+						BoundVolumeSnapshotContentName: &vscName,
+					},
+				},
+				&snapshotv1api.VolumeSnapshotContent{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: vscName,
+					},
+				},
+			},
+			expectedID: "vol-vks",
+		},
+		{
+			name: "success with PV CSI volume handle",
+			snapshot: &velerov2alpha1api.CSISnapshotSpec{
+				VolumeSnapshot:          "fake-vs",
+				VolumeSnapshotNamespace: "fake-ns",
+			},
+			targetPVName: "fake-pv",
+			ctrlClientObj: []runtime.Object{
+				&snapshotv1api.VolumeSnapshot{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: "fake-ns",
+						Name:      "fake-vs",
+					},
+					Status: &snapshotv1api.VolumeSnapshotStatus{
+						BoundVolumeSnapshotContentName: &vscName,
+					},
+				},
+				&snapshotv1api.VolumeSnapshotContent{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: vscName,
+					},
+					Status: &snapshotv1api.VolumeSnapshotContentStatus{
+						SnapshotHandle: &snapshotHandle,
+					},
+				},
+			},
+			kubeClientObj: []runtime.Object{
+				&corev1api.PersistentVolume{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "fake-pv",
+					},
+					Spec: corev1api.PersistentVolumeSpec{
+						PersistentVolumeSource: corev1api.PersistentVolumeSource{
+							CSI: &corev1api.CSIPersistentVolumeSource{
+								VolumeHandle: "csi-vol-789",
+							},
+						},
+					},
+				},
+			},
+			expectedID: "csi-vol-789",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			fakeKubeClient := fake.NewSimpleClientset(test.kubeClientObj...)
+			fakeCtrlClient := velerotest.NewFakeControllerRuntimeClient(t, test.ctrlClientObj...)
+
+			exposer := genericRestoreExposer{
+				kubeClient: fakeKubeClient,
+				ctrlClient: fakeCtrlClient,
+				log:        velerotest.NewLogger(),
+			}
+
+			volID, err := exposer.getVolumeID(t.Context(), test.snapshot, test.targetPVName)
+			if test.expectedErr != "" {
+				require.Error(t, err)
+				assert.Contains(t, err.Error(), test.expectedErr)
+				assert.Empty(t, volID)
+			} else {
+				require.NoError(t, err)
+				assert.Equal(t, test.expectedID, volID)
+			}
+		})
+	}
+}
+
+func TestRestoreExpose_CSISnapshot(t *testing.T) {
+	scName := "fake-sc"
+	restore := &velerov1.Restore{
+		TypeMeta:   metav1.TypeMeta{APIVersion: velerov1.SchemeGroupVersion.String(), Kind: "Restore"},
+		ObjectMeta: metav1.ObjectMeta{Namespace: velerov1.DefaultNamespace, Name: "fake-restore", UID: "fake-uid"},
+	}
+	ownerObject := corev1api.ObjectReference{
+		Kind:       restore.Kind,
+		Namespace:  restore.Namespace,
+		Name:       restore.Name,
+		UID:        restore.UID,
+		APIVersion: restore.APIVersion,
+	}
+	targetPVCObj := &corev1api.PersistentVolumeClaim{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "fake-ns", Name: "fake-target-pvc"},
+		Spec:       corev1api.PersistentVolumeClaimSpec{StorageClassName: &scName},
+	}
+	storageClass := &storagev1api.StorageClass{ObjectMeta: metav1.ObjectMeta{Name: "fake-sc"}}
+	daemonSet := &appsv1api.DaemonSet{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "velero", Name: "node-agent"},
+		TypeMeta:   metav1.TypeMeta{Kind: "DaemonSet", APIVersion: appsv1api.SchemeGroupVersion.String()},
+		Spec: appsv1api.DaemonSetSpec{
+			Template: corev1api.PodTemplateSpec{
+				Spec: corev1api.PodSpec{Containers: []corev1api.Container{{Image: "fake-image"}}},
+			},
+		},
+	}
+
+	vscName := "fake-vsc"
+
+	t.Run("getVolumeID fails - falls back to full restore and creates pod without volume ID", func(t *testing.T) {
+		fakeKubeClient := fake.NewSimpleClientset(targetPVCObj, storageClass, daemonSet)
+		fakeCtrlClient := velerotest.NewFakeControllerRuntimeClient(t)
+		exposer := genericRestoreExposer{
+			kubeClient: fakeKubeClient,
+			ctrlClient: fakeCtrlClient,
+			log:        velerotest.NewLogger(),
+		}
+
+		err := exposer.Expose(t.Context(), ownerObject, GenericRestoreExposeParam{
+			TargetPVCName:    "fake-target-pvc",
+			TargetNamespace:  "fake-ns",
+			HostingPodLabels: map[string]string{},
+			Resources:        corev1api.ResourceRequirements{},
+			ExposeTimeout:    time.Millisecond,
+			CSI: &GenericRestoreExposeCSI{
+				Snapshot: &velerov2alpha1api.CSISnapshotSpec{
+					VolumeSnapshot:          "non-existent-vs",
+					VolumeSnapshotNamespace: "fake-ns",
+				},
+			},
+		})
+		require.NoError(t, err)
+
+		pod, err := fakeKubeClient.CoreV1().Pods(ownerObject.Namespace).Get(t.Context(), ownerObject.Name, metav1.GetOptions{})
+		require.NoError(t, err)
+		require.Len(t, pod.Spec.Containers, 1)
+		for _, arg := range pod.Spec.Containers[0].Args {
+			assert.NotContains(t, arg, "--volume-id=")
+			assert.NotContains(t, arg, "--vs-namespace=")
+		}
+	})
+
+	t.Run("getVolumeID succeeds - passes volume ID to restore pod", func(t *testing.T) {
+		fakeKubeClient := fake.NewSimpleClientset(targetPVCObj, storageClass, daemonSet)
+		fakeCtrlClient := velerotest.NewFakeControllerRuntimeClient(t,
+			&snapshotv1api.VolumeSnapshot{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "fake-ns",
+					Name:      "fake-vs",
+					Annotations: map[string]string{
+						util.VSphereCNSChangeIDAnno: "c-1",
+						util.VSphereCNSSnapshotAnno: "vol-123+snap-1",
+					},
+				},
+				Status: &snapshotv1api.VolumeSnapshotStatus{
+					BoundVolumeSnapshotContentName: &vscName,
+				},
+			},
+			&snapshotv1api.VolumeSnapshotContent{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: vscName,
+				},
+			},
+		)
+		exposer := genericRestoreExposer{
+			kubeClient: fakeKubeClient,
+			ctrlClient: fakeCtrlClient,
+			log:        velerotest.NewLogger(),
+		}
+
+		err := exposer.Expose(t.Context(), ownerObject, GenericRestoreExposeParam{
+			TargetPVCName:    "fake-target-pvc",
+			TargetNamespace:  "fake-ns",
+			HostingPodLabels: map[string]string{},
+			Resources:        corev1api.ResourceRequirements{},
+			ExposeTimeout:    time.Millisecond,
+			CSI: &GenericRestoreExposeCSI{
+				Snapshot: &velerov2alpha1api.CSISnapshotSpec{
+					VolumeSnapshot:          "fake-vs",
+					VolumeSnapshotNamespace: "fake-ns",
+				},
+			},
+		})
+		require.NoError(t, err)
+
+		pod, err := fakeKubeClient.CoreV1().Pods(ownerObject.Namespace).Get(t.Context(), ownerObject.Name, metav1.GetOptions{})
+		require.NoError(t, err)
+		require.Len(t, pod.Spec.Containers, 1)
+		assert.Contains(t, pod.Spec.Containers[0].Args, "--volume-id=vol-123")
+		assert.Contains(t, pod.Spec.Containers[0].Args, "--vs-namespace=fake-ns")
+	})
+}
+
 func TestRebindVolume(t *testing.T) {
 	restore := &velerov1.Restore{
 		TypeMeta: metav1.TypeMeta{

pkg/podvolume/backupper_test.go

@@ -333,7 +333,8 @@ func createPVObj(index int, withHostPath bool) *corev1api.PersistentVolume {
 }
 
 func createPVCObj(index int) *corev1api.PersistentVolumeClaim {
-	pvcObj := builder.ForPersistentVolumeClaim("fake-ns", fmt.Sprintf("fake-pvc-%d", index)).VolumeName(fmt.Sprintf("fake-pv-%d", index)).Result()
+	pvcObj := builder.ForPersistentVolumeClaim("fake-ns", fmt.Sprintf("fake-pvc-%d", index)).VolumeName(fmt.Sprintf("fake-pv-%d", index)).
+		Phase(corev1api.ClaimBound).Result()
 	return pvcObj
 }
 

pkg/podvolume/restorer.go

@@ -49,6 +49,9 @@ type RestoreData struct {
 	Pod                             *corev1api.Pod
 	PodVolumeBackups                []*velerov1api.PodVolumeBackup
 	SourceNamespace, BackupLocation string
+	// BackupVolumeInfos is the backup's volume info keyed by PV name, used by
+	// the in-place restore pre-flight checks.
+	BackupVolumeInfos map[string]volume.BackupVolumeInfo
 }
 
 // Restorer can execute pod volume restores of volumes in a pod.
@@ -186,6 +189,10 @@ func (r *restorer) RestorePodVolumes(data RestoreData, tracker *volume.RestoreVo
 		// to write into, and they cannot write to it themselves until this
 		// restore's PodVolumeRestores complete.
 		if data.Restore.IsVolumeDataInplaceRestore() && pvc != nil {
+			if err := inplace.CheckPVCBoundToBackedUpPV(pvc, backedUpPVName(data.BackupVolumeInfos, data.SourceNamespace, pvc.Name), data.SourceNamespace); err != nil {
+				errs = append(errs, err)
+				continue
+			}
 			if err := inplace.CheckPVCNotInUse(r.ctx, r.crClient, pvc, data.Restore.UID); err != nil {
 				errs = append(errs, err)
 				continue
@@ -317,6 +324,17 @@ func newPodVolumeRestore(restore *velerov1api.Restore, pod *corev1api.Pod, backu
 	return pvr
 }
 
+// backedUpPVName returns the name of the PV the given source-namespace PVC was
+// bound to at backup time, or "" if unknown.
+func backedUpPVName(infos map[string]volume.BackupVolumeInfo, pvcNamespace, pvcName string) string {
+	for pvName, info := range infos {
+		if info.PVCNamespace == pvcNamespace && info.PVCName == pvcName {
+			return pvName
+		}
+	}
+	return ""
+}
+
 func getVolumesRepositoryType(volumes map[string]volumeBackupInfo) (string, error) {
 	if len(volumes) == 0 {
 		return "", errors.New("empty volume list")

pkg/podvolume/restorer_test.go

@@ -197,6 +197,7 @@ func TestRestorePodVolumes(t *testing.T) {
 		pvbs            []*velerov1api.PodVolumeBackup
 		restoredPod     *corev1api.Pod
 		sourceNamespace string
+		volumeInfos     map[string]volume.BackupVolumeInfo
 		inplace         bool
 		errs            []expectError
 	}{
@@ -413,6 +414,31 @@ func TestRestorePodVolumes(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "in-place restore blocked when the PVC is bound to a different PV than at backup time",
+			pvbs: []*velerov1api.PodVolumeBackup{
+				createPVBObj(true, true, 1, "kopia"),
+			},
+			inplace: true,
+			kubeClientObj: []runtime.Object{
+				createNodeAgentDaemonset(),
+				createPVCObj(1),
+			},
+			ctlClientObj: []runtime.Object{
+				createBackupRepoObj(),
+			},
+			restoredPod:     createPodObj(true, true, true, 1),
+			sourceNamespace: "fake-ns",
+			bsl:             "fake-bsl",
+			volumeInfos:     map[string]volume.BackupVolumeInfo{"some-other-pv": {PVCNamespace: "fake-ns", PVCName: "fake-pvc-1"}},
+			runtimeScheme:   scheme,
+			errs: []expectError{
+				{
+					err:        "in-place restore pre-flight check failed",
+					prefixOnly: true,
+				},
+			},
+		},
 		{
 			name: "in-place restore proceeds when the PVC is only used by the gated restored pod",
 			pvbs: []*velerov1api.PodVolumeBackup{
@@ -481,11 +507,12 @@ func TestRestorePodVolumes(t *testing.T) {
 			}()
 
 			errs := rs.RestorePodVolumes(RestoreData{
-				Restore:          restoreObj,
-				Pod:              test.restoredPod,
-				PodVolumeBackups: test.pvbs,
-				SourceNamespace:  test.sourceNamespace,
-				BackupLocation:   test.bsl,
+				Restore:           restoreObj,
+				Pod:               test.restoredPod,
+				PodVolumeBackups:  test.pvbs,
+				SourceNamespace:   test.sourceNamespace,
+				BackupLocation:    test.bsl,
+				BackupVolumeInfos: test.volumeInfos,
 			}, volume.NewRestoreVolInfoTracker(restoreObj, logrus.New(), fakeCRClient))
 
 			if errs == nil {

pkg/restore/actions/csi/pvc_action.go

@@ -234,11 +234,10 @@ func (p *pvcRestoreItemAction) executeWithDataMove(logger *logrus.Entry, input *
 	var volumeSnapshot *snapshotv1api.VolumeSnapshot
 	restoreType := input.Restore.Spec.ExistingVolumeDataPolicy
 	if pvcExists {
-		if existingPVC.Status.Phase != corev1api.ClaimBound {
-			return nil, errors.New("ExistingVolumeDataPolicy is in-place restore, but the existing PVC is not bound.")
-		}
-
 		// Pre-flight checks must pass before any side effect on the existing PVC/PV.
+		if err := inplace.CheckPVCBoundToBackedUpPV(existingPVC, pvcFromBackup.Spec.VolumeName, pvcFromBackup.Namespace); err != nil {
+			return nil, errors.WithStack(err)
+		}
 		if err := inplace.CheckPVCNotInUse(ctx, p.crClient, existingPVC, input.Restore.UID); err != nil {
 			return nil, errors.WithStack(err)
 		}

pkg/restore/actions/csi/pvc_action_test.go

@@ -742,8 +742,8 @@ func TestExecuteInplaceRestore(t *testing.T) {
 }
 
 // TestExecuteInplaceRestorePreflight verifies the RIA fails the item without
-// side effects when the pre-flight check fails. The in-use semantics are
-// covered by the pkg/restore/inplace unit tests.
+// side effects when a pre-flight check fails. The check semantics themselves
+// are covered by the pkg/restore/inplace unit tests.
 func TestExecuteInplaceRestorePreflight(t *testing.T) {
 	newPodUsingPVC := func(phase corev1api.PodPhase) *corev1api.Pod {
 		pod := builder.ForPod("velero", "consumer-pod").
@@ -754,17 +754,25 @@ func TestExecuteInplaceRestorePreflight(t *testing.T) {
 	}
 
 	tests := []struct {
-		name        string
-		pod         *corev1api.Pod
-		expectBlock bool
+		name           string
+		pod            *corev1api.Pod
+		backedUpPVName string
+		expectBlock    string
 	}{
 		{
-			name: "no pod, restore proceeds",
+			name:           "checks pass, restore proceeds",
+			backedUpPVName: "testPV",
 		},
 		{
-			name:        "active pod blocks the restore",
-			pod:         newPodUsingPVC(corev1api.PodRunning),
-			expectBlock: true,
+			name:           "active pod blocks the restore",
+			pod:            newPodUsingPVC(corev1api.PodRunning),
+			backedUpPVName: "testPV",
+			expectBlock:    "consumer-pod",
+		},
+		{
+			name:           "PVC bound to a different PV blocks the restore",
+			backedUpPVName: "backupPV",
+			expectBlock:    "was bound to PV backupPV at backup time",
 		},
 	}
 
@@ -778,6 +786,7 @@ func TestExecuteInplaceRestorePreflight(t *testing.T) {
 			restore := builder.ForRestore("velero", "testRestore").Backup("testBackup").
 				ObjectMeta(builder.WithUID("uid")).ExistingVolumeDataPolicy("full").Result()
 			pvcFromBackup := builder.ForPersistentVolumeClaim("velero", "testPVC").
+				VolumeName(tc.backedUpPVName).
 				ObjectMeta(builder.WithAnnotations(
 					velerov1api.VolumeSnapshotLabel, "vsName",
 					velerov1api.DataUploadNameAnnotation, "velero/testDU",
@@ -817,10 +826,10 @@ func TestExecuteInplaceRestorePreflight(t *testing.T) {
 			dataDownloadList := new(velerov2alpha1.DataDownloadList)
 			require.NoError(t, pvcRIA.crClient.List(t.Context(), dataDownloadList, &crclient.ListOptions{}))
 
-			if tc.expectBlock {
+			if tc.expectBlock != "" {
 				require.Error(t, err)
 				require.Contains(t, err.Error(), "pre-flight check failed")
-				require.Contains(t, err.Error(), "consumer-pod")
+				require.Contains(t, err.Error(), tc.expectBlock)
 				// No side effects: PVC untouched with the original volumeName,
 				// PV reclaim policy not patched, no DataDownload created.
 				require.NoError(t, getErr)

pkg/restore/inplace/preflight.go

@@ -129,3 +129,23 @@ func gatedByThisRestore(pod *corev1api.Pod, restoreUID types.UID) bool {
 	// is still closed.
 	return true
 }
+
+// CheckPVCBoundToBackedUpPV verifies the existing PVC is still bound to the
+// same PV it was bound to at backup time. An in-place restore onto a
+// different volume is unsafe: an incremental (CBT) restore computes deltas
+// against a different volume lineage, and even a full restore would patch and
+// write into a volume unrelated to the backup. The PV comparison is skipped
+// when the PVC is restored into a different namespace, where it is necessarily
+// bound to a different PV (the documented cross-namespace clone-and-restore
+// workflow), and when the backed-up PV name is unknown.
+func CheckPVCBoundToBackedUpPV(existingPVC *corev1api.PersistentVolumeClaim, backedUpPVName, sourceNamespace string) error {
+	if existingPVC.Status.Phase != corev1api.ClaimBound {
+		return errors.Errorf("in-place restore pre-flight check failed, skipping volume data restore: PVC %s/%s is not bound (phase %s)",
+			existingPVC.Namespace, existingPVC.Name, existingPVC.Status.Phase)
+	}
+	if existingPVC.Namespace != sourceNamespace || backedUpPVName == "" || existingPVC.Spec.VolumeName == backedUpPVName {
+		return nil
+	}
+	return errors.Errorf("in-place restore pre-flight check failed, skipping volume data restore: PVC %s/%s is bound to PV %s, but was bound to PV %s at backup time",
+		existingPVC.Namespace, existingPVC.Name, existingPVC.Spec.VolumeName, backedUpPVName)
+}

pkg/restore/inplace/preflight_test.go

@@ -200,3 +200,66 @@ func TestCheckPVCNotInUse(t *testing.T) {
 		})
 	}
 }
+
+func TestCheckPVCBoundToBackedUpPV(t *testing.T) {
+	pvc := func(namespace, pvName string, phase corev1api.PersistentVolumeClaimPhase) *corev1api.PersistentVolumeClaim {
+		return &corev1api.PersistentVolumeClaim{
+			ObjectMeta: metav1.ObjectMeta{Name: "pvc-1", Namespace: namespace},
+			Spec:       corev1api.PersistentVolumeClaimSpec{VolumeName: pvName},
+			Status:     corev1api.PersistentVolumeClaimStatus{Phase: phase},
+		}
+	}
+
+	tests := []struct {
+		name           string
+		existingPVC    *corev1api.PersistentVolumeClaim
+		backedUpPVName string
+		expectError    string
+	}{
+		{
+			name:           "bound to the backed-up PV, check passes",
+			existingPVC:    pvc("default", "pv-1", corev1api.ClaimBound),
+			backedUpPVName: "pv-1",
+		},
+		{
+			name:           "bound to a different PV, check fails",
+			existingPVC:    pvc("default", "pv-other", corev1api.ClaimBound),
+			backedUpPVName: "pv-1",
+			expectError:    "is bound to PV pv-other, but was bound to PV pv-1 at backup time",
+		},
+		{
+			name:           "PVC not bound, check fails",
+			existingPVC:    pvc("default", "", corev1api.ClaimPending),
+			backedUpPVName: "pv-1",
+			expectError:    "is not bound (phase Pending)",
+		},
+		{
+			name:           "different PV in a different namespace, check passes",
+			existingPVC:    pvc("mapped-ns", "pv-other", corev1api.ClaimBound),
+			backedUpPVName: "pv-1",
+		},
+		{
+			name:           "backed-up PV name unknown, check passes",
+			existingPVC:    pvc("default", "pv-other", corev1api.ClaimBound),
+			backedUpPVName: "",
+		},
+		{
+			name:           "different namespace but PVC not bound, check still fails",
+			existingPVC:    pvc("mapped-ns", "", corev1api.ClaimLost),
+			backedUpPVName: "pv-1",
+			expectError:    "is not bound (phase Lost)",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			err := CheckPVCBoundToBackedUpPV(tc.existingPVC, tc.backedUpPVName, "default")
+			if tc.expectError == "" {
+				require.NoError(t, err)
+				return
+			}
+			require.Error(t, err)
+			assert.Contains(t, err.Error(), tc.expectError)
+		})
+	}
+}

pkg/restore/restore.go

@@ -2287,11 +2287,12 @@ func restorePodVolumeBackups(ctx *restoreContext, createdObj *unstructured.Unstr
 			}
 
 			data := podvolume.RestoreData{
-				Restore:          ctx.restore,
-				Pod:              pod,
-				PodVolumeBackups: ctx.podVolumeBackups,
-				SourceNamespace:  originalNamespace,
-				BackupLocation:   ctx.backup.Spec.StorageLocation,
+				Restore:           ctx.restore,
+				Pod:               pod,
+				PodVolumeBackups:  ctx.podVolumeBackups,
+				SourceNamespace:   originalNamespace,
+				BackupLocation:    ctx.backup.Spec.StorageLocation,
+				BackupVolumeInfos: ctx.backupVolumeInfoMap,
 			}
 			if errs := ctx.podVolumeRestorer.RestorePodVolumes(data, ctx.restoreVolumeInfoTracker); errs != nil {
 				ctx.log.WithError(kubeerrs.NewAggregate(errs)).Error("unable to successfully complete pod volume restores of pod's volumes")

pkg/uploader/block/snapshot.go

@@ -222,6 +222,8 @@ func Restore(ctx context.Context, blkUp Uploader, rep udmrepo.BackupRepo, snapsh
 			log.Warnf("No ChangeID tag from snapshot %s, fallback to full restore", snapshotID)
 		} else if snapshot.Tags[uploader.CBTVolumeIDTag] == "" {
 			log.Warnf("No VolumeID tag from snapshot %s, fallback to full restore", snapshotID)
+		} else if cbtSource.VolumeID == "" {
+			log.Warnf("No VolumeID in cbt source %v, fallback to full restore", cbtSource)
 		} else if snapshot.Tags[uploader.CBTVolumeIDTag] != cbtSource.VolumeID {
 			log.Warnf("VolumeID %s from snapshot %s is not expected as %s, fallback to full restore", snapshot.Tags[uploader.CBTVolumeIDTag], snapshotID, cbtSource.VolumeID)
 		} else {

pkg/uploader/block/snapshot_test.go

@@ -766,6 +766,27 @@ func TestRestore(t *testing.T) {
 			},
 			expectedSize: 4096,
 		},
+		{
+			name:        "incremental restore fallback - empty cbtSource VolumeID",
+			incremental: true,
+			cbtSource:   cbtservice.SourceInfo{Snapshot: "snap-cbt", VolumeID: ""},
+			setupMocks: func(blkup *mockUploader, repo *udmrepomocks.BackupRepo) {
+				snapWithTags := udmrepo.Snapshot{
+					Tags: map[string]string{
+						uploader.CBTChangeIDTag: "cid-1",
+						uploader.CBTVolumeIDTag: "vol-1",
+					},
+				}
+				repo.On("GetSnapshot", mock.Anything, udmrepo.ID("snap-001")).Return(snapWithTags, nil)
+				blkup.On("Restore", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
+					Return(int64(4096), int64(4096), nil)
+			},
+			setupOpenDev: func(t *testing.T) *os.File {
+				t.Helper()
+				return tempFile(t, "")
+			},
+			expectedSize: 4096,
+		},
 		{
 			name:        "incremental restore fallback - VolumeID mismatch",
 			incremental: true,