Merge branch 'main' into use-pv-topology-for-block-data-mover-for-windows

pkg/apis/velero/v1/pod_volume_restore_type.go

@@ -111,6 +111,10 @@ type PodVolumeRestoreStatus struct {
 	// +optional
 	Progress shared.DataMoveOperationProgress `json:"progress,omitempty"`
 
+	// IncrementalBytes holds the number of bytes restored incrementally
+	// +optional
+	IncrementalBytes *int64 `json:"incrementalBytes,omitempty"`
+
 	// AcceptedTimestamp records the time the pod volume restore is to be prepared.
 	// The server's time is used for AcceptedTimestamp
 	// +optional
@@ -129,9 +133,13 @@ type PodVolumeRestoreStatus struct {
 // +kubebuilder:object:root=true
 // +kubebuilder:storageversion
 // +kubebuilder:printcolumn:name="Status",type="string",JSONPath=".status.phase",description="PodVolumeRestore status such as New/InProgress"
+// The "Restore Type" column is hidden by default to align with PVB.
+// +kubebuilder:printcolumn:name="Restore Type",type="string",JSONPath=".spec.restoreType",description="Restore type such as Full/Incremental",priority=10
 // +kubebuilder:printcolumn:name="Started",type="date",JSONPath=".status.startTimestamp",description="Time duration since this PodVolumeRestore was started"
 // +kubebuilder:printcolumn:name="Bytes Done",type="integer",format="int64",JSONPath=".status.progress.bytesDone",description="Completed bytes"
 // +kubebuilder:printcolumn:name="Total Bytes",type="integer",format="int64",JSONPath=".status.progress.totalBytes",description="Total bytes"
+// The "Incremental Bytes" column is hidden by default to align with PVB.
+// +kubebuilder:printcolumn:name="Incremental Bytes",type="integer",format="int64",JSONPath=".status.incrementalBytes",description="Incremental bytes",priority=10
 // +kubebuilder:printcolumn:name="Storage Location",type="string",JSONPath=".spec.backupStorageLocation",description="Name of the Backup Storage Location where the backup data is stored"
 // +kubebuilder:printcolumn:name="Age",type="date",JSONPath=".metadata.creationTimestamp",description="Time duration since this PodVolumeRestore was created"
 // +kubebuilder:printcolumn:name="Node",type="string",JSONPath=".status.node",description="Name of the node where the PodVolumeRestore is processed"

pkg/apis/velero/v1/zz_generated.deepcopy.go

@@ -1170,6 +1170,11 @@ func (in *PodVolumeRestoreStatus) DeepCopyInto(out *PodVolumeRestoreStatus) {
 		*out = (*in).DeepCopy()
 	}
 	out.Progress = in.Progress
+	if in.IncrementalBytes != nil {
+		in, out := &in.IncrementalBytes, &out.IncrementalBytes
+		*out = new(int64)
+		**out = **in
+	}
 	if in.AcceptedTimestamp != nil {
 		in, out := &in.AcceptedTimestamp, &out.AcceptedTimestamp
 		*out = (*in).DeepCopy()

pkg/apis/velero/v2alpha1/data_download_types.go

@@ -132,6 +132,10 @@ type DataDownloadStatus struct {
 	// +optional
 	Progress shared.DataMoveOperationProgress `json:"progress,omitempty"`
 
+	// IncrementalBytes holds the number of bytes restored incrementally since the last snapshot
+	// +optional
+	IncrementalBytes *int64 `json:"incrementalBytes,omitempty"`
+
 	// Node is name of the node where the DataDownload is processed.
 	// +optional
 	Node string `json:"node,omitempty"`
@@ -154,9 +158,13 @@ type DataDownloadStatus struct {
 // +kubebuilder:object:root=true
 // +kubebuilder:storageversion
 // +kubebuilder:printcolumn:name="Status",type="string",JSONPath=".status.phase",description="DataDownload status such as New/InProgress"
+// The "Restore Type" column is hidden by default to align with DataUpload.
+// +kubebuilder:printcolumn:name="Restore Type",type="string",JSONPath=".spec.restoreType",description="Restore type such as Full/Incremental",priority=10
 // +kubebuilder:printcolumn:name="Started",type="date",JSONPath=".status.startTimestamp",description="Time duration since this DataDownload was started"
 // +kubebuilder:printcolumn:name="Bytes Done",type="integer",format="int64",JSONPath=".status.progress.bytesDone",description="Completed bytes"
 // +kubebuilder:printcolumn:name="Total Bytes",type="integer",format="int64",JSONPath=".status.progress.totalBytes",description="Total bytes"
+// The "Incremental Bytes" column is hidden by default to align with DataUpload.
+// +kubebuilder:printcolumn:name="Incremental Bytes",type="integer",format="int64",JSONPath=".status.incrementalBytes",description="Incremental bytes",priority=10
 // +kubebuilder:printcolumn:name="Storage Location",type="string",JSONPath=".spec.backupStorageLocation",description="Name of the Backup Storage Location where the backup data is stored"
 // +kubebuilder:printcolumn:name="Age",type="date",JSONPath=".metadata.creationTimestamp",description="Time duration since this DataDownload was created"
 // +kubebuilder:printcolumn:name="Node",type="string",JSONPath=".status.node",description="Name of the node where the DataDownload is processed"

pkg/apis/velero/v2alpha1/zz_generated.deepcopy.go

@@ -123,6 +123,11 @@ func (in *DataDownloadStatus) DeepCopyInto(out *DataDownloadStatus) {
 		*out = (*in).DeepCopy()
 	}
 	out.Progress = in.Progress
+	if in.IncrementalBytes != nil {
+		in, out := &in.IncrementalBytes, &out.IncrementalBytes
+		*out = new(int64)
+		**out = **in
+	}
 	if in.AcceptedTimestamp != nil {
 		in, out := &in.AcceptedTimestamp, &out.AcceptedTimestamp
 		*out = (*in).DeepCopy()

pkg/backup/item_collector.go

@@ -478,12 +478,15 @@ func (r *itemCollector) getResourceItems(
 		namespacesToList = []string{""}
 	}
 
+	grString := gr.String()
+	hasNamespacedPolicies := len(r.backupRequest.NamespacedFilterMap) > 0
+
 	var items []*kubernetesResource
 
 	for _, namespace := range namespacesToList {
 		// Check per-namespace resource type filter from ResourcePolicy
 		if nsFilter := r.backupRequest.GetNamespaceFilter(namespace); nsFilter != nil {
-			_, hasSpecific := nsFilter.ResourceFilterMap[gr.String()]
+			_, hasSpecific := nsFilter.ResourceFilterMap[grString]
 			if !hasSpecific && nsFilter.CatchAllFilter == nil {
 				log.Debugf("Skipping resource %s in namespace %s: not in resourceFilters",
 					gr, namespace)
@@ -497,9 +500,66 @@ func (r *itemCollector) getResourceItems(
 			continue
 		}
 
+		var lastNS string
+		var lastNSFilter *ResolvedNamespaceFilter
+
 		// Collect items in included Namespaces
 		for i := range unstructuredItems {
 			item := &unstructuredItems[i]
+			itemNS := item.GetNamespace()
+
+			// Apply namespace inclusion/exclusion and fine-grained filter policies in-memory for cluster-wide queries.
+			if itemNS != "" && namespace == "" {
+				if r.backupRequest.NamespaceIncludesExcludes != nil &&
+					!r.backupRequest.NamespaceIncludesExcludes.ShouldInclude(itemNS) {
+					log.Debugf("Skipping resource %s in namespace %s: namespace excluded",
+						gr, itemNS)
+					continue
+				}
+
+				if hasNamespacedPolicies {
+					if itemNS != lastNS {
+						lastNS = itemNS
+						lastNSFilter = r.backupRequest.GetNamespaceFilter(itemNS)
+					}
+
+					if lastNSFilter != nil {
+						rf := lastNSFilter.ResourceFilterMap[grString]
+						if rf == nil {
+							rf = lastNSFilter.CatchAllFilter
+						}
+						if rf == nil {
+							log.Debugf("Skipping resource %s in namespace %s: not in resourceFilters",
+								gr, itemNS)
+							continue
+						}
+
+						// In-memory label selector checks for fine-grained filters
+						if rf.LabelSelector != nil || len(rf.OrLabelSelectors) > 0 {
+							itemLabels := labels.Set(item.GetLabels())
+							if rf.LabelSelector != nil && !rf.LabelSelector.Matches(itemLabels) {
+								log.Debugf("Skipping resource %s in namespace %s: does not match labelSelector",
+									gr, itemNS)
+								continue
+							}
+							if len(rf.OrLabelSelectors) > 0 {
+								matched := false
+								for _, s := range rf.OrLabelSelectors {
+									if s.Matches(itemLabels) {
+										matched = true
+										break
+									}
+								}
+								if !matched {
+									log.Debugf("Skipping resource %s in namespace %s: does not match orLabelSelectors",
+										gr, itemNS)
+									continue
+								}
+							}
+						}
+					}
+				}
+			}
 
 			path, err := r.writeToFile(item)
 			if err != nil {

pkg/backup/item_collector_test.go

@@ -466,3 +466,190 @@ func TestGetOrderedResourcesForTypeTrimsSpaces(t *testing.T) {
 		{namespace: "ns1", name: "pod3"},
 	}, sorted)
 }
+
+func TestGetResourceItems_NamespacedFilterPolicies_ClusterWideListing(t *testing.T) {
+	// Simulate cluster-wide API calls where client is called with namespace=""
+	prodSA := unstructured.Unstructured{
+		Object: map[string]any{
+			"apiVersion": "v1",
+			"kind":       "ServiceAccount",
+			"metadata": map[string]any{
+				"name":      "default",
+				"namespace": "production",
+			},
+		},
+	}
+	defaultSA := unstructured.Unstructured{
+		Object: map[string]any{
+			"apiVersion": "v1",
+			"kind":       "ServiceAccount",
+			"metadata": map[string]any{
+				"name":      "default",
+				"namespace": "default",
+			},
+		},
+	}
+	kubeSystemSA := unstructured.Unstructured{
+		Object: map[string]any{
+			"apiVersion": "v1",
+			"kind":       "ServiceAccount",
+			"metadata": map[string]any{
+				"name":      "default",
+				"namespace": "kube-system",
+			},
+		},
+	}
+
+	prodCM1 := unstructured.Unstructured{
+		Object: map[string]any{
+			"apiVersion": "v1",
+			"kind":       "ConfigMap",
+			"metadata": map[string]any{
+				"name":      "app-config",
+				"namespace": "production",
+				"labels": map[string]any{
+					"app": "frontend",
+				},
+			},
+		},
+	}
+	prodCM2 := unstructured.Unstructured{
+		Object: map[string]any{
+			"apiVersion": "v1",
+			"kind":       "ConfigMap",
+			"metadata": map[string]any{
+				"name":      "other-config",
+				"namespace": "production",
+				"labels": map[string]any{
+					"app": "backend",
+				},
+			},
+		},
+	}
+	defaultCM := unstructured.Unstructured{
+		Object: map[string]any{
+			"apiVersion": "v1",
+			"kind":       "ConfigMap",
+			"metadata": map[string]any{
+				"name":      "my-config",
+				"namespace": "default",
+			},
+		},
+	}
+
+	dcClusterWideSA := &test.FakeDynamicClient{}
+	dcClusterWideSA.On("List", mock.Anything).Return(&unstructured.UnstructuredList{Items: []unstructured.Unstructured{prodSA, defaultSA, kubeSystemSA}}, nil)
+
+	dcClusterWideCM := &test.FakeDynamicClient{}
+	dcClusterWideCM.On("List", mock.Anything).Return(&unstructured.UnstructuredList{Items: []unstructured.Unstructured{prodCM1, prodCM2, defaultCM}}, nil)
+
+	factory := &test.FakeDynamicFactory{}
+	factory.On("ClientForGroupVersionResource", schema.GroupVersion{Version: "v1"}, metav1.APIResource{Name: "serviceaccounts", Namespaced: true, Kind: "ServiceAccount"}, "").Return(dcClusterWideSA, nil)
+	factory.On("ClientForGroupVersionResource", schema.GroupVersion{Version: "v1"}, metav1.APIResource{Name: "configmaps", Namespaced: true, Kind: "ConfigMap"}, "").Return(dcClusterWideCM, nil)
+
+	frontendSelector, err := metav1.LabelSelectorAsSelector(&metav1.LabelSelector{
+		MatchLabels: map[string]string{"app": "frontend"},
+	})
+	require.NoError(t, err)
+
+	req := &Request{
+		Backup: builder.ForBackup("velero", "backup").Result(),
+		NamespaceIncludesExcludes: collections.NewNamespaceIncludesExcludes().
+			Includes("*").
+			Excludes("kube-system"),
+		NamespacedFilterMap: map[string]*ResolvedNamespaceFilter{
+			"production": {
+				ResourceFilterMap: map[string]*ResolvedResourceFilter{
+					"configmaps": {
+						LabelSelector: frontendSelector,
+					},
+				},
+			},
+		},
+		ResourceIncludesExcludes: includeAllIE{},
+	}
+
+	tempDir := t.TempDir()
+	r := &itemCollector{
+		backupRequest:   req,
+		dynamicFactory:  factory,
+		discoveryHelper: test.NewFakeDiscoveryHelper(true, nil),
+		log:             test.NewLogger(),
+		dir:             tempDir,
+	}
+
+	// 1. ServiceAccounts:
+	// - production should be skipped because ServiceAccount is not in production's resourceFilters
+	// - kube-system should be skipped because kube-system is in excludedNamespaces
+	// - default should be included
+	saResource := metav1.APIResource{Name: "serviceaccounts", Namespaced: true, Kind: "ServiceAccount"}
+	saItems, err := r.getResourceItems(test.NewLogger(), schema.GroupVersion{Version: "v1"}, saResource, nil)
+	require.NoError(t, err)
+	require.Len(t, saItems, 1)
+	assert.Equal(t, "default", saItems[0].namespace)
+	assert.Equal(t, "default", saItems[0].name)
+
+	// 2. ConfigMaps:
+	// - production/app-config matches label app=frontend and should be included
+	// - production/other-config has label app=backend and should be skipped by labelSelector
+	// - default/my-config has no filter policy and should be included
+	cmResource := metav1.APIResource{Name: "configmaps", Namespaced: true, Kind: "ConfigMap"}
+	cmItems, err := r.getResourceItems(test.NewLogger(), schema.GroupVersion{Version: "v1"}, cmResource, nil)
+	require.NoError(t, err)
+	require.Len(t, cmItems, 2)
+	var cmNames []string
+	for _, it := range cmItems {
+		cmNames = append(cmNames, it.namespace+"/"+it.name)
+	}
+	assert.ElementsMatch(t, []string{"production/app-config", "default/my-config"}, cmNames)
+}
+
+func TestGetResourceItems_NamespacedFilterPolicies_SpecificNamespaces(t *testing.T) {
+	defaultSA := unstructured.Unstructured{
+		Object: map[string]any{
+			"apiVersion": "v1",
+			"kind":       "ServiceAccount",
+			"metadata": map[string]any{
+				"name":      "default",
+				"namespace": "default",
+			},
+		},
+	}
+
+	dcDefaultSA := &test.FakeDynamicClient{}
+	dcDefaultSA.On("List", mock.Anything).Return(&unstructured.UnstructuredList{Items: []unstructured.Unstructured{defaultSA}}, nil)
+
+	factory := &test.FakeDynamicFactory{}
+	// Note: production client is never even requested because production skips ServiceAccount at the loop top!
+	factory.On("ClientForGroupVersionResource", schema.GroupVersion{Version: "v1"}, metav1.APIResource{Name: "serviceaccounts", Namespaced: true, Kind: "ServiceAccount"}, "default").Return(dcDefaultSA, nil)
+
+	req := &Request{
+		Backup: builder.ForBackup("velero", "backup").Result(),
+		NamespaceIncludesExcludes: collections.NewNamespaceIncludesExcludes().
+			Includes("production", "default"),
+		NamespacedFilterMap: map[string]*ResolvedNamespaceFilter{
+			"production": {
+				ResourceFilterMap: map[string]*ResolvedResourceFilter{
+					"configmaps": {},
+				},
+			},
+		},
+		ResourceIncludesExcludes: includeAllIE{},
+	}
+
+	tempDir := t.TempDir()
+	r := &itemCollector{
+		backupRequest:   req,
+		dynamicFactory:  factory,
+		discoveryHelper: test.NewFakeDiscoveryHelper(true, nil),
+		log:             test.NewLogger(),
+		dir:             tempDir,
+	}
+
+	saResource := metav1.APIResource{Name: "serviceaccounts", Namespaced: true, Kind: "ServiceAccount"}
+	saItems, err := r.getResourceItems(test.NewLogger(), schema.GroupVersion{Version: "v1"}, saResource, nil)
+	require.NoError(t, err)
+	require.Len(t, saItems, 1)
+	assert.Equal(t, "default", saItems[0].namespace)
+	assert.Equal(t, "default", saItems[0].name)
+}

pkg/cmd/cli/backup/create.go

@@ -25,6 +25,7 @@ import (
 	"github.com/spf13/cobra"
 	"github.com/spf13/pflag"
 	kubeerrs "k8s.io/apimachinery/pkg/util/errors"
+	"k8s.io/apimachinery/pkg/util/validation"
 	"k8s.io/client-go/tools/cache"
 	kbclient "sigs.k8s.io/controller-runtime/pkg/client"
 
@@ -45,7 +46,21 @@ func NewCreateCommand(f client.Factory, use string) *cobra.Command {
 	c := &cobra.Command{
 		Use:   use + " NAME",
 		Short: "Create a backup",
-		Args:  cobra.MaximumNArgs(1),
+		Args: func(c *cobra.Command, args []string) error {
+			if err := cobra.MaximumNArgs(1)(c, args); err != nil {
+				return err
+			}
+			fromSchedule, _ := c.Flags().GetString("from-schedule")
+			if fromSchedule == "" && len(args) == 0 {
+				return fmt.Errorf("a backup name is required, unless you are creating based on a schedule")
+			}
+			if len(args) == 1 {
+				if errs := validation.IsDNS1123Subdomain(args[0]); len(errs) > 0 {
+					return fmt.Errorf("invalid backup name %q: %s", args[0], strings.Join(errs, "; "))
+				}
+			}
+			return nil
+		},
 		Run: func(c *cobra.Command, args []string) {
 			cmd.CheckError(o.Complete(args, f))
 			cmd.CheckError(o.Validate(c, args, f))
@@ -191,11 +206,27 @@ func (o *CreateOptions) Validate(c *cobra.Command, args []string, f client.Facto
 		return err
 	}
 
-	// Ensure that unless FromSchedule is set, args contains a backup name
-	if o.FromSchedule == "" && len(args) != 1 {
+	// Ensure that unless FromSchedule is set, a backup name is required
+	if o.FromSchedule == "" && o.Name == "" {
 		return fmt.Errorf("a backup name is required, unless you are creating based on a schedule")
 	}
-
+	// Validate the backup name format whenever a name is provided
+	if o.Name != "" {
+		if errs := validation.IsDNS1123Subdomain(o.Name); len(errs) > 0 {
+			return fmt.Errorf("invalid backup name %q: %s", o.Name, strings.Join(errs, "; "))
+		}
+	}
+	// When a backup name will be generated from the schedule (i.e. FromSchedule
+	// is set and no explicit name was given), ensure the schedule name leaves
+	// enough room for the generated timestamp suffix ("-" + 14-digit timestamp,
+	// 15 characters total) within the DNS1123 subdomain length limit.
+	if o.FromSchedule != "" && o.Name == "" {
+		const timestampSuffixLen = 15 // "-" + "20060102150405"
+		maxScheduleNameLen := validation.DNS1123SubdomainMaxLength - timestampSuffixLen
+		if len(o.FromSchedule) > maxScheduleNameLen {
+			return fmt.Errorf("schedule name %q is too long: must be %d characters or fewer to leave room for the generated timestamp suffix", o.FromSchedule, maxScheduleNameLen)
+		}
+	}
 	errs := collections.ValidateNamespaceIncludesExcludes(o.IncludeNamespaces, o.ExcludeNamespaces)
 	if len(errs) > 0 {
 		return kubeerrs.NewAggregate(errs)

pkg/cmd/cli/backup/create_test.go

@@ -245,7 +245,7 @@ func TestCreateOptions_OrderedResources(t *testing.T) {
 }
 
 func TestCreateCommand(t *testing.T) {
-	name := "nameToBeCreated"
+	name := "name-to-be-created"
 	args := []string{name}
 
 	t.Run("create a backup create command with full options except fromSchedule and wait, then run by create option", func(t *testing.T) {
@@ -457,3 +457,127 @@ func TestCreateCommand(t *testing.T) {
 		assert.NoError(t, e)
 	})
 }
+func TestCreateCommand_Args(t *testing.T) {
+	testCases := []struct {
+		name         string
+		args         []string
+		fromSchedule string
+		expectError  bool
+	}{
+		{
+			name:        "should error when no name and no from-schedule",
+			args:        []string{},
+			expectError: true,
+		},
+		{
+			name:        "should pass when a valid name is provided",
+			args:        []string{"my-backup"},
+			expectError: false,
+		},
+		{
+			name:        "should error when the name is not a valid DNS1123 subdomain",
+			args:        []string{"Invalid_Name!"},
+			expectError: true,
+		},
+		{
+			name:         "should pass with no name when from-schedule is set",
+			args:         []string{},
+			fromSchedule: "daily-backup",
+			expectError:  false,
+		},
+		{
+			name:        "should error when more than one arg is given",
+			args:        []string{"name1", "name2"},
+			expectError: true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			f := &factorymocks.Factory{}
+			cmd := NewCreateCommand(f, "")
+			if tc.fromSchedule != "" {
+				err := cmd.Flags().Set("from-schedule", tc.fromSchedule)
+				require.NoError(t, err)
+			}
+
+			err := cmd.Args(cmd, tc.args)
+
+			if tc.expectError {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+			}
+		})
+	}
+}
+
+func TestCreateOptions_Validate(t *testing.T) {
+	testCases := []struct {
+		name         string
+		optName      string
+		fromSchedule string
+		args         []string
+		expectError  bool
+	}{
+		{
+			name:        "should error when no name and no from-schedule",
+			optName:     "",
+			args:        []string{},
+			expectError: true,
+		},
+		{
+			name:        "should pass with a valid name and no from-schedule",
+			optName:     "my-backup",
+			args:        []string{"my-backup"},
+			expectError: false,
+		},
+		{
+			name:         "should error when name is invalid, regardless of from-schedule",
+			optName:      "Invalid_Name!",
+			fromSchedule: "daily-backup",
+			args:         []string{"Invalid_Name!"},
+			expectError:  true,
+		},
+		{
+			name:         "should pass when from-schedule is set and no name given",
+			optName:      "",
+			fromSchedule: "daily-backup",
+			args:         []string{},
+			expectError:  false,
+		},
+		{
+			name:         "should pass when schedule name leaves room for timestamp suffix",
+			optName:      "",
+			fromSchedule: strings.Repeat("a", 238), // exactly at the 238-char limit
+			args:         []string{},
+			expectError:  false,
+		},
+		{
+			name:         "should error when schedule name is too long to leave room for timestamp suffix",
+			optName:      "",
+			fromSchedule: strings.Repeat("a", 239), // one over the 238-char limit
+			args:         []string{},
+			expectError:  true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			f := &factorymocks.Factory{}
+			cmd := NewCreateCommand(f, "")
+
+			o := NewCreateOptions()
+			o.Name = tc.optName
+			o.FromSchedule = tc.fromSchedule
+
+			err := o.Validate(cmd, tc.args, f)
+
+			if tc.expectError {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+			}
+		})
+	}
+}

pkg/cmd/errors.go

@@ -27,7 +27,7 @@ import (
 func CheckError(err error) {
 	if err != nil {
 		if err != context.Canceled {
-			fmt.Fprintf(os.Stderr, "An error occurred: %v\n", err)
+			fmt.Fprintf(os.Stderr, "velero: %v\n", err)
 		}
 		os.Exit(1)
 	}

pkg/controller/backup_controller.go

@@ -362,7 +362,7 @@ func (b *backupReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr
 		// result in the backup being Failed.
 		log.WithError(err).Error("backup failed")
 		request.Status.Phase = velerov1api.BackupPhaseFailed
-		request.Status.FailureReason = err.Error()
+		request.Status.FailureReason = fmt.Sprintf("backup execution failed: %v", err)
 	}
 
 	switch request.Status.Phase {
@@ -619,7 +619,7 @@ func (b *backupReconciler) prepareBackupRequest(ctx context.Context, backup *vel
 	resourcePolicies, err := resourcepolicies.GetResourcePoliciesFromBackupWithGlobal(
 		*request.Backup, b.kbClient, b.globalVolumePoliciesConfigMap, request.Namespace, logger)
 	if err != nil {
-		request.Status.ValidationErrors = append(request.Status.ValidationErrors, err.Error())
+		request.Status.ValidationErrors = append(request.Status.ValidationErrors, fmt.Sprintf("invalid resource policies: %v", err))
 	} else if b.globalVolumePoliciesConfigMap != "" {
 		// Record the contributing global volume policies ConfigMap so `velero backup describe` can surface it.
 		request.Annotations[velerov1api.GlobalBackupVolumePolicyConfigMapAnnotation] = b.globalVolumePoliciesConfigMap

pkg/controller/backup_controller_test.go

@@ -1205,7 +1205,7 @@ func TestProcessBackupCompletions(t *testing.T) {
 				},
 				Status: velerov1api.BackupStatus{
 					Phase:               velerov1api.BackupPhaseFailed,
-					FailureReason:       "backup already exists in object storage",
+					FailureReason:       "backup execution failed: backup already exists in object storage",
 					Version:             1,
 					FormatVersion:       "1.1.0",
 					StartTimestamp:      &timestamp,
@@ -1250,7 +1250,7 @@ func TestProcessBackupCompletions(t *testing.T) {
 				},
 				Status: velerov1api.BackupStatus{
 					Phase:               velerov1api.BackupPhaseFailed,
-					FailureReason:       "error checking if backup already exists in object storage: Backup already exists in object storage",
+					FailureReason:       "backup execution failed: error checking if backup already exists in object storage: Backup already exists in object storage",
 					Version:             1,
 					FormatVersion:       "1.1.0",
 					StartTimestamp:      &timestamp,

pkg/controller/backup_deletion_controller.go

@@ -315,7 +315,7 @@ func (r *backupDeletionReconciler) Reconcile(ctx context.Context, req ctrl.Reque
 				volumeSnapshotter, ok := volumeSnapshotters[snapshot.Spec.Location]
 				if !ok {
 					if volumeSnapshotter, err = r.volumeSnapshottersForVSL(ctx, backup.Namespace, snapshot.Spec.Location, pluginManager); err != nil {
-						errs = append(errs, err.Error())
+						errs = append(errs, errors.Wrapf(err, "error getting volume snapshotter for location %q", snapshot.Spec.Location).Error())
 						continue
 					}
 					volumeSnapshotters[snapshot.Spec.Location] = volumeSnapshotter
@@ -334,7 +334,7 @@ func (r *backupDeletionReconciler) Reconcile(ctx context.Context, req ctrl.Reque
 	log.Info("Removing pod volume snapshots")
 	if deleteErrs := r.deletePodVolumeSnapshots(ctx, backup); len(deleteErrs) > 0 {
 		for _, err := range deleteErrs {
-			errs = append(errs, err.Error())
+			errs = append(errs, errors.Wrap(err, "error deleting pod volume snapshots").Error())
 		}
 	}
 
@@ -342,7 +342,7 @@ func (r *backupDeletionReconciler) Reconcile(ctx context.Context, req ctrl.Reque
 		log.Info("Removing snapshot data by data mover")
 		if deleteErrs := r.deleteMovedSnapshots(ctx, backup); len(deleteErrs) > 0 {
 			for _, err := range deleteErrs {
-				errs = append(errs, err.Error())
+				errs = append(errs, errors.Wrap(err, "error deleting moved snapshot data").Error())
 			}
 		}
 		duList := &velerov2alpha1.DataUploadList{}
@@ -354,12 +354,12 @@ func (r *backupDeletionReconciler) Reconcile(ctx context.Context, req ctrl.Reque
 			}),
 		}); err != nil {
 			log.WithError(err).Error("Error listing datauploads")
-			errs = append(errs, err.Error())
+			errs = append(errs, errors.Wrap(err, "error listing datauploads for backup").Error())
 		} else {
 			for i := range duList.Items {
 				du := duList.Items[i]
 				if err := r.Delete(ctx, &du); err != nil {
-					errs = append(errs, err.Error())
+					errs = append(errs, errors.Wrapf(err, "error deleting dataupload %q", du.Name).Error())
 				}
 			}
 		}
@@ -368,7 +368,7 @@ func (r *backupDeletionReconciler) Reconcile(ctx context.Context, req ctrl.Reque
 	if backupStore != nil && len(errs) == 0 {
 		log.Info("Removing backup from backup storage")
 		if err := backupStore.DeleteBackup(backup.Name); err != nil {
-			errs = append(errs, err.Error())
+			errs = append(errs, errors.Wrap(err, "error removing backup from backup storage").Error())
 		}
 	} else if len(errs) > 0 {
 		log.Info("Skipping removal of backup from backup storage due to previous errors")
@@ -631,7 +631,7 @@ func (r *backupDeletionReconciler) patchDeleteBackupRequest(ctx context.Context,
 func (r *backupDeletionReconciler) patchDeleteBackupRequestWithError(ctx context.Context, req *velerov1api.DeleteBackupRequest, err error) error {
 	_, err = r.patchDeleteBackupRequest(ctx, req, func(r *velerov1api.DeleteBackupRequest) {
 		r.Status.Phase = velerov1api.DeleteBackupRequestPhaseProcessed
-		r.Status.Errors = []string{err.Error()}
+		r.Status.Errors = []string{errors.WithMessage(err, "backup deletion failed").Error()}
 	})
 	return err
 }

pkg/controller/backup_deletion_controller_test.go

@@ -137,7 +137,7 @@ func TestBackupDeletionControllerReconcile(t *testing.T) {
 		td.fakeClient.Get(ctx, td.req.NamespacedName, res)
 		assert.Equal(t, "Processed", string(res.Status.Phase))
 		assert.Len(t, res.Status.Errors, 1)
-		assert.True(t, strings.HasPrefix(res.Status.Errors[0], "error getting the backup store"))
+		assert.True(t, strings.HasPrefix(res.Status.Errors[0], "backup deletion failed: error getting the backup store"))
 	})
 
 	t.Run("missing spec.backupName", func(t *testing.T) {
@@ -153,7 +153,7 @@ func TestBackupDeletionControllerReconcile(t *testing.T) {
 		require.NoError(t, err)
 		assert.Equal(t, "Processed", string(res.Status.Phase))
 		assert.Len(t, res.Status.Errors, 1)
-		assert.Equal(t, "spec.backupName is required", res.Status.Errors[0])
+		assert.Equal(t, "backup deletion failed: spec.backupName is required", res.Status.Errors[0])
 	})
 
 	t.Run("existing deletion requests for the backup are deleted", func(t *testing.T) {
@@ -221,7 +221,7 @@ func TestBackupDeletionControllerReconcile(t *testing.T) {
 		require.NoError(t, err)
 		assert.Equal(t, "Processed", string(res.Status.Phase))
 		assert.Len(t, res.Status.Errors, 1)
-		assert.Equal(t, "backup is still in progress", res.Status.Errors[0])
+		assert.Equal(t, "backup deletion failed: backup is still in progress", res.Status.Errors[0])
 	})
 
 	t.Run("unable to find backup", func(t *testing.T) {
@@ -235,7 +235,7 @@ func TestBackupDeletionControllerReconcile(t *testing.T) {
 		require.NoError(t, err)
 		assert.Equal(t, "Processed", string(res.Status.Phase))
 		assert.Len(t, res.Status.Errors, 1)
-		assert.Equal(t, "backup not found", res.Status.Errors[0])
+		assert.Equal(t, "backup deletion failed: backup not found", res.Status.Errors[0])
 	})
 	t.Run("unable to find backup storage location", func(t *testing.T) {
 		backup := builder.ForBackup(velerov1api.DefaultNamespace, "foo").StorageLocation("default").Result()
@@ -250,7 +250,7 @@ func TestBackupDeletionControllerReconcile(t *testing.T) {
 		require.NoError(t, err)
 		assert.Equal(t, "Processed", string(res.Status.Phase))
 		assert.Len(t, res.Status.Errors, 1)
-		assert.Equal(t, "backup storage location default not found", res.Status.Errors[0])
+		assert.Equal(t, "backup deletion failed: backup storage location default not found", res.Status.Errors[0])
 	})
 
 	t.Run("backup storage location is in read-only mode", func(t *testing.T) {
@@ -267,7 +267,7 @@ func TestBackupDeletionControllerReconcile(t *testing.T) {
 		require.NoError(t, err)
 		assert.Equal(t, "Processed", string(res.Status.Phase))
 		assert.Len(t, res.Status.Errors, 1)
-		assert.Equal(t, "cannot delete backup because backup storage location default is currently in read-only mode", res.Status.Errors[0])
+		assert.Equal(t, "backup deletion failed: cannot delete backup because backup storage location default is currently in read-only mode", res.Status.Errors[0])
 	})
 
 	t.Run("backup storage location is in unavailable state", func(t *testing.T) {
@@ -284,7 +284,7 @@ func TestBackupDeletionControllerReconcile(t *testing.T) {
 		require.NoError(t, err)
 		assert.Equal(t, "Processed", string(res.Status.Phase))
 		assert.Len(t, res.Status.Errors, 1)
-		assert.Equal(t, "cannot delete backup because backup storage location default is currently in Unavailable state", res.Status.Errors[0])
+		assert.Equal(t, "backup deletion failed: cannot delete backup because backup storage location default is currently in Unavailable state", res.Status.Errors[0])
 	})
 
 	t.Run("full delete, no errors", func(t *testing.T) {
@@ -877,7 +877,7 @@ func TestBackupDeletionControllerReconcile(t *testing.T) {
 		require.NoError(t, err)
 		assert.Equal(t, "Processed", string(res.Status.Phase))
 		assert.Len(t, res.Status.Errors, 1)
-		assert.Equal(t, "backup not found", res.Status.Errors[0])
+		assert.Equal(t, "backup deletion failed: backup not found", res.Status.Errors[0])
 	})
 }
 

pkg/controller/backup_sync_controller_test.go

@@ -439,7 +439,7 @@ var _ = Describe("Backup Sync Reconciler", func() {
 			}
 
 			if test.location != nil {
-				Expect(r.client.Create(ctx, test.location)).ShouldNot(HaveOccurred())
+				Expect(r.client.Create(ctx, test.location)).ToNot(HaveOccurred())
 				backupStores[test.location.Name] = &persistencemocks.BackupStore{}
 
 				backupStore, ok := backupStores[test.location.Name]
@@ -457,12 +457,12 @@ var _ = Describe("Backup Sync Reconciler", func() {
 
 			for _, existingBackup := range test.existingBackups {
 				err := client.Create(context.TODO(), existingBackup, &ctrlClient.CreateOptions{})
-				Expect(err).ShouldNot(HaveOccurred())
+				Expect(err).ToNot(HaveOccurred())
 			}
 
 			for _, existingPodVolumeBackup := range test.existingPodVolumeBackups {
 				err := client.Create(context.TODO(), existingPodVolumeBackup, &ctrlClient.CreateOptions{})
-				Expect(err).ShouldNot(HaveOccurred())
+				Expect(err).ToNot(HaveOccurred())
 			}
 
 			actualResult, err := r.Reconcile(ctx, ctrl.Request{
@@ -534,7 +534,7 @@ var _ = Describe("Backup Sync Reconciler", func() {
 							cloudBackupData.backup.Status.Expiration.After(fakeClock.Now())) {
 						Expect(apierrors.IsNotFound(err)).To(BeTrue())
 					} else {
-						Expect(err).ShouldNot(HaveOccurred())
+						Expect(err).ToNot(HaveOccurred())
 
 						// did this cloud pod volume backup already exist in the cluster?
 						var existingPodVolumeBackup *velerov1api.PodVolumeBackup
@@ -673,7 +673,7 @@ var _ = Describe("Backup Sync Reconciler", func() {
 		}
 
 		queueScheme := runtime.NewScheme()
-		Expect(velerov1api.AddToScheme(queueScheme)).ShouldNot(HaveOccurred())
+		Expect(velerov1api.AddToScheme(queueScheme)).ToNot(HaveOccurred())
 
 		for _, test := range tests {
 			var (
@@ -693,7 +693,7 @@ var _ = Describe("Backup Sync Reconciler", func() {
 				logger:                  velerotest.NewLogger(),
 			}
 
-			Expect(client.Create(ctx, location)).ShouldNot(HaveOccurred(), test.name)
+			Expect(client.Create(ctx, location)).ToNot(HaveOccurred(), test.name)
 			backupStore := &persistencemocks.BackupStore{}
 			backupStores[location.Name] = backupStore
 			backupStore.On("ListBackups").Return([]string{test.cloudBackup.Name}, nil)
@@ -704,7 +704,7 @@ var _ = Describe("Backup Sync Reconciler", func() {
 			_, err := syncReconciler.Reconcile(ctx, ctrl.Request{
 				NamespacedName: types.NamespacedName{Namespace: location.Namespace, Name: location.Name},
 			})
-			Expect(err).ShouldNot(HaveOccurred(), test.name)
+			Expect(err).ToNot(HaveOccurred(), test.name)
 
 			backupKey := types.NamespacedName{Namespace: "ns-1", Name: test.cloudBackup.Name}
 			synced := &velerov1api.Backup{}
@@ -714,7 +714,7 @@ var _ = Describe("Backup Sync Reconciler", func() {
 				Expect(apierrors.IsNotFound(err)).To(BeTrue(), test.name)
 				continue
 			}
-			Expect(err).ShouldNot(HaveOccurred(), test.name)
+			Expect(err).ToNot(HaveOccurred(), test.name)
 
 			// Reconcile the synced backup with the queue controller twice: the first
 			// reconcile would move a New/empty-phase backup to Queued, the second one
@@ -723,11 +723,11 @@ var _ = Describe("Backup Sync Reconciler", func() {
 			queueReconciler := NewBackupQueueReconciler(client, queueScheme, velerotest.NewLogger(), 1, NewBackupTracker())
 			for range 2 {
 				_, err = queueReconciler.Reconcile(ctx, ctrl.Request{NamespacedName: backupKey})
-				Expect(err).ShouldNot(HaveOccurred(), test.name)
+				Expect(err).ToNot(HaveOccurred(), test.name)
 			}
 
 			after := &velerov1api.Backup{}
-			Expect(client.Get(ctx, backupKey, after)).ShouldNot(HaveOccurred(), test.name)
+			Expect(client.Get(ctx, backupKey, after)).ToNot(HaveOccurred(), test.name)
 			Expect(after.Status.Phase).To(BeEquivalentTo(test.expectPhase), test.name)
 			// Hooks are dropped on sync, so the stored metadata cannot carry a payload
 			// that a later code path could execute.
@@ -880,7 +880,7 @@ var _ = Describe("Backup Sync Reconciler", func() {
 			for _, backup := range test.k8sBackups {
 				// add test backup to client
 				err := client.Create(context.TODO(), backup, &ctrlClient.CreateOptions{})
-				Expect(err).ShouldNot(HaveOccurred())
+				Expect(err).ToNot(HaveOccurred())
 			}
 
 			bslName := "default"
@@ -890,7 +890,7 @@ var _ = Describe("Backup Sync Reconciler", func() {
 			r.deleteOrphanedBackups(ctx, bslName, test.cloudBackups, velerotest.NewLogger())
 
 			numBackups, err := numBackups(client)
-			Expect(err).ShouldNot(HaveOccurred())
+			Expect(err).ToNot(HaveOccurred())
 
 			fmt.Println("")
 
@@ -913,7 +913,7 @@ var _ = Describe("Backup Sync Reconciler", func() {
 
 		testObjList := backupSyncSourceOrderFunc(locationList)
 		testObjArray, err := meta.ExtractList(testObjList)
-		Expect(err).ShouldNot(HaveOccurred())
+		Expect(err).ToNot(HaveOccurred())
 
 		expectLocation := testObjArray[0].(*velerov1api.BackupStorageLocation)
 		Expect(expectLocation.Spec.Default).To(BeEquivalentTo(true))
@@ -1085,7 +1085,7 @@ var _ = Describe("Backup Sync Reconciler", func() {
 				//create all required schedules as needed.
 				for _, creatable := range test.toCreate {
 					err := b.client.Create(context.Background(), creatable)
-					Expect(err).ShouldNot(HaveOccurred())
+					Expect(err).ToNot(HaveOccurred())
 				}
 
 				references := b.filterBackupOwnerReferences(context.Background(), test.backup, logger)

pkg/controller/data_download_controller.go

@@ -32,6 +32,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/utils/clock"
+	"k8s.io/utils/ptr"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -501,6 +502,7 @@ func (r *DataDownloadReconciler) OnDataDownloadCompleted(ctx context.Context, na
 		}
 
 		dd.Status.Phase = velerov2alpha1api.DataDownloadPhaseCompleted
+		dd.Status.IncrementalBytes = ptr.To(result.Restore.IncrementalBytes)
 		dd.Status.CompletionTimestamp = &metav1.Time{Time: r.Clock.Now()}
 
 		delete(dd.Labels, exposer.ExposeOnGoingLabel)

pkg/controller/pod_volume_backup_controller.go

@@ -784,10 +784,9 @@ func UpdatePVBStatusToFailed(ctx context.Context, c client.Client, pvb *velerov1
 				pvb.Status.SnapshotID = dataPathError.GetSnapshotID()
 			}
 			if len(strings.TrimSpace(msg)) == 0 {
-				pvb.Status.Message = errOut.Error()
-			} else {
-				pvb.Status.Message = errors.WithMessage(errOut, msg).Error()
+				msg = "pod volume backup failed"
 			}
+			pvb.Status.Message = errors.WithMessage(errOut, msg).Error()
 			if pvb.Status.StartTimestamp.IsZero() {
 				pvb.Status.StartTimestamp = &metav1.Time{Time: time}
 			}

pkg/controller/pod_volume_restore_controller.go

@@ -33,6 +33,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/kubernetes"
 	clocks "k8s.io/utils/clock"
+	"k8s.io/utils/ptr"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -837,6 +838,7 @@ func (r *PodVolumeRestoreReconciler) OnDataPathCompleted(ctx context.Context, na
 
 		pvr.Status.Phase = velerov1api.PodVolumeRestorePhaseCompleted
 		pvr.Status.CompletionTimestamp = &metav1.Time{Time: r.clock.Now()}
+		pvr.Status.IncrementalBytes = ptr.To(result.Restore.IncrementalBytes)
 
 		delete(pvr.Labels, exposer.ExposeOnGoingLabel)
 

pkg/controller/restore_controller.go

@@ -275,7 +275,7 @@ func (r *restoreReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 	if err := r.runValidatedRestore(restore, info, resourceModifiers, restoreResPolicies); err != nil {
 		log.WithError(err).Debug("Restore failed")
 		restore.Status.Phase = api.RestorePhaseFailed
-		restore.Status.FailureReason = err.Error()
+		restore.Status.FailureReason = fmt.Sprintf("restore execution failed: %v", err)
 		r.metrics.RegisterRestoreFailed(backupScheduleName)
 	}
 
@@ -349,7 +349,7 @@ func (r *restoreReconciler) validateAndComplete(ctx context.Context, restore *ap
 	// validate Restore Init Hook's InitContainers
 	restoreHooks, err := hook.GetRestoreHooksFromSpec(&restore.Spec.Hooks)
 	if err != nil {
-		restore.Status.ValidationErrors = append(restore.Status.ValidationErrors, err.Error())
+		restore.Status.ValidationErrors = append(restore.Status.ValidationErrors, fmt.Sprintf("invalid restore hooks: %v", err))
 	}
 	for _, resource := range restoreHooks {
 		for _, h := range resource.RestoreHooks {
@@ -357,7 +357,7 @@ func (r *restoreReconciler) validateAndComplete(ctx context.Context, restore *ap
 				for _, container := range h.Init.InitContainers {
 					err = hook.ValidateContainer(container.Raw)
 					if err != nil {
-						restore.Status.ValidationErrors = append(restore.Status.ValidationErrors, err.Error())
+						restore.Status.ValidationErrors = append(restore.Status.ValidationErrors, fmt.Sprintf("invalid init container in restore hook %q: %v", resource.Name, err))
 					}
 				}
 			}
@@ -433,7 +433,7 @@ func (r *restoreReconciler) validateAndComplete(ctx context.Context, restore *ap
 		)
 		if err != nil {
 			restore.Status.ValidationErrors = append(
-				restore.Status.ValidationErrors, err.Error(),
+				restore.Status.ValidationErrors, fmt.Sprintf("invalid restore resource policies: %v", err),
 			)
 			return backupInfo{}, nil, nil
 		}

pkg/controller/server_status_request_controller_test.go

@@ -92,7 +92,7 @@ var _ = Describe("Server Status Request Reconciler", func() {
 				Expect(apierrors.IsNotFound(err)).To(BeTrue())
 			} else {
 				Expect(err).ToNot(HaveOccurred())
-				Eventually(instance.Status.Phase == test.expected.Status.Phase, timeout).Should(BeTrue())
+				Expect(instance.Status.Phase).To(Equal(test.expected.Status.Phase))
 			}
 		},
 		Entry("with phase=empty will be processed and phased successfully patched", request{

pkg/datamover/restore_micro_service_test.go

@@ -102,7 +102,7 @@ func TestOnDataDownloadCompleted(t *testing.T) {
 		{
 			name:        "marshal fail",
 			marshalErr:  errors.New("fake-marshal-error"),
-			expectedErr: "Failed to marshal restore result {{ } 0}: fake-marshal-error",
+			expectedErr: "Failed to marshal restore result {{ } 0 0}: fake-marshal-error",
 		},
 		{
 			name:                "succeed",

pkg/datapath/data_path.go

@@ -250,7 +250,7 @@ func (dp *generalDataPath) StartRestore(snapshotID string, target AccessPoint, u
 			dp.wgDataPath.Done()
 		}()
 
-		totalBytes, err := dp.uploaderProv.RunRestore(dp.ctx, snapshotID, target.ByPath, restoreParam.Incremental,
+		incrementalBytes, totalBytes, err := dp.uploaderProv.RunRestore(dp.ctx, snapshotID, target.ByPath, restoreParam.Incremental,
 			provider.CBTParam{
 				Source: cbtservice.SourceInfo{
 					Snapshot: restoreParam.VolumeSnapshotName,
@@ -268,7 +268,7 @@ func (dp *generalDataPath) StartRestore(snapshotID string, target AccessPoint, u
 			}
 			dp.callbacks.OnFailed(context.Background(), dp.namespace, dp.jobName, dataPathErr)
 		} else {
-			dp.callbacks.OnCompleted(context.Background(), dp.namespace, dp.jobName, Result{Restore: RestoreResult{Target: target, TotalBytes: totalBytes}})
+			dp.callbacks.OnCompleted(context.Background(), dp.namespace, dp.jobName, Result{Restore: RestoreResult{Target: target, TotalBytes: totalBytes, IncrementalBytes: incrementalBytes}})
 		}
 	}()
 

pkg/datapath/data_path_test.go

@@ -190,7 +190,7 @@ func TestAsyncRestore(t *testing.T) {
 		t.Run(test.name, func(t *testing.T) {
 			dp := newGeneralDataPath("job-1", "test", nil, "velero", Callbacks{}, velerotest.NewLogger()).(*generalDataPath)
 			mockProvider := providerMock.NewProvider(t)
-			mockProvider.On("RunRestore", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(test.result.Restore.TotalBytes, test.err)
+			mockProvider.On("RunRestore", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(test.result.Restore.IncrementalBytes, test.result.Restore.TotalBytes, test.err)
 			mockProvider.On("Close", mock.Anything).Return(nil)
 			dp.uploaderProv = mockProvider
 			dp.initialized = true

pkg/datapath/types.go

@@ -43,8 +43,9 @@ type BackupResult struct {
 
 // RestoreResult represents the result of a restore
 type RestoreResult struct {
-	Target     AccessPoint `json:"target,omitempty"`
-	TotalBytes int64       `json:"totalBytes,omitempty"`
+	Target           AccessPoint `json:"target,omitempty"`
+	TotalBytes       int64       `json:"totalBytes,omitempty"`
+	IncrementalBytes int64       `json:"incrementalBytes,omitempty"`
 }
 
 // Callbacks defines the collection of callbacks during backup/restore

pkg/exposer/generic_restore.go

@@ -252,6 +252,31 @@ 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)
+		}
+
+		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, 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")
 
 	var targetPV *corev1api.PersistentVolume
@@ -300,30 +325,6 @@ func (e *genericRestoreExposer) Expose(ctx context.Context, ownerObject corev1ap
 	}
 
 	curLog.Info("Creating restore pod")
-	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)
-		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, param.TargetPVName)
-		if err != nil {
-			return errors.Wrap(err, "error to get CBT info")
-		}
-		curLog.Debugf("CBT info: %+v", cbtInfo)
-		volumeID = cbtInfo.VolumeID
-	}
 	var csiSnapshotMetadataServiceConfigs *velerotypes.CSISnapshotMetadataService
 	if param.CSI != nil {
 		csiSnapshotMetadataServiceConfigs = param.CSI.SnapshotMetadataServiceConfigs
@@ -1026,7 +1027,7 @@ func (e *genericRestoreExposer) createRestorePVC(ctx context.Context, ownerObjec
 				Spec: *targetPV.Spec.DeepCopy(),
 			}
 			tmpPV.Spec.VolumeMode = restorePVC.Spec.VolumeMode
-			e.log.Infof("the volume mode is different, creating temporary PV %s with volume mode %s", tmpPV.Name, tmpPV.Spec.VolumeMode)
+			e.log.Infof("the volume mode is different, creating temporary PV %s with volume mode %v", tmpPV.Name, tmpPV.Spec.VolumeMode)
 			tmpPV, err = e.kubeClient.CoreV1().PersistentVolumes().Create(ctx, tmpPV, metav1.CreateOptions{})
 			if err != nil {
 				return nil, errors.Wrapf(err, "fail to create the temporary PV %s", volumeName)

pkg/itemblock/itemblock_test.go

@@ -0,0 +1,357 @@
+/*
+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 itemblock
+
+import (
+	"testing"
+
+	"github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+// newUnstructuredItem creates an unstructured object with the given apiVersion,
+// kind, namespace, and name for use in tests.
+func newUnstructuredItem(apiVersion, kind, namespace, name string) *unstructured.Unstructured {
+	return &unstructured.Unstructured{
+		Object: map[string]any{
+			"apiVersion": apiVersion,
+			"kind":       kind,
+			"metadata": map[string]any{
+				"namespace": namespace,
+				"name":      name,
+			},
+		},
+	}
+}
+
+func TestAddUnstructured(t *testing.T) {
+	tests := []struct {
+		name        string
+		items       []ItemBlockItem
+		expectedLen int
+	}{
+		{
+			name: "add single item to empty block",
+			items: []ItemBlockItem{
+				{
+					Gr:           schema.GroupResource{Group: "apps", Resource: "deployments"},
+					Item:         newUnstructuredItem("apps/v1", "Deployment", "default", "nginx"),
+					PreferredGVR: schema.GroupVersionResource{Group: "apps", Version: "v1", Resource: "deployments"},
+				},
+			},
+			expectedLen: 1,
+		},
+		{
+			name: "add multiple items",
+			items: []ItemBlockItem{
+				{
+					Gr:           schema.GroupResource{Group: "", Resource: "pods"},
+					Item:         newUnstructuredItem("v1", "Pod", "default", "pod-1"),
+					PreferredGVR: schema.GroupVersionResource{Group: "", Version: "v1", Resource: "pods"},
+				},
+				{
+					Gr:           schema.GroupResource{Group: "", Resource: "services"},
+					Item:         newUnstructuredItem("v1", "Service", "kube-system", "kube-dns"),
+					PreferredGVR: schema.GroupVersionResource{Group: "", Version: "v1", Resource: "services"},
+				},
+				{
+					Gr:           schema.GroupResource{Group: "apps", Resource: "deployments"},
+					Item:         newUnstructuredItem("apps/v1", "Deployment", "default", "web"),
+					PreferredGVR: schema.GroupVersionResource{Group: "apps", Version: "v1", Resource: "deployments"},
+				},
+			},
+			expectedLen: 3,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			ib := &ItemBlock{
+				Log: logrus.New(),
+			}
+
+			for _, item := range tc.items {
+				ib.AddUnstructured(item.Gr, item.Item, item.PreferredGVR)
+			}
+
+			require.Len(t, ib.Items, tc.expectedLen)
+
+			// Verify each added item matches what was provided
+			for i, item := range tc.items {
+				assert.Equal(t, item.Gr, ib.Items[i].Gr)
+				assert.Equal(t, item.Item, ib.Items[i].Item)
+				assert.Equal(t, item.PreferredGVR, ib.Items[i].PreferredGVR)
+			}
+		})
+	}
+}
+
+func TestFindItem(t *testing.T) {
+	podsGR := schema.GroupResource{Group: "", Resource: "pods"}
+	deploymentsGR := schema.GroupResource{Group: "apps", Resource: "deployments"}
+	servicesGR := schema.GroupResource{Group: "", Resource: "services"}
+
+	podsGVR := schema.GroupVersionResource{Group: "", Version: "v1", Resource: "pods"}
+	deploymentsGVR := schema.GroupVersionResource{Group: "apps", Version: "v1", Resource: "deployments"}
+
+	tests := []struct {
+		name            string
+		existingItems   []ItemBlockItem
+		searchGR        schema.GroupResource
+		searchNamespace string
+		searchName      string
+		expectedCount   int
+		// If set, verifies the names in the returned items are in this exact order
+		expectedNames []string
+	}{
+		{
+			name:            "find item in empty block returns nil",
+			existingItems:   nil,
+			searchGR:        podsGR,
+			searchNamespace: "default",
+			searchName:      "pod-1",
+			expectedCount:   0,
+		},
+		{
+			name: "find matching item by GR, namespace, and name",
+			existingItems: []ItemBlockItem{
+				{
+					Gr:           podsGR,
+					Item:         newUnstructuredItem("v1", "Pod", "default", "pod-1"),
+					PreferredGVR: podsGVR,
+				},
+			},
+			searchGR:        podsGR,
+			searchNamespace: "default",
+			searchName:      "pod-1",
+			expectedCount:   1,
+			expectedNames:   []string{"pod-1"},
+		},
+		{
+			name: "no match when GR differs",
+			existingItems: []ItemBlockItem{
+				{
+					Gr:           podsGR,
+					Item:         newUnstructuredItem("v1", "Pod", "default", "pod-1"),
+					PreferredGVR: podsGVR,
+				},
+			},
+			searchGR:        deploymentsGR,
+			searchNamespace: "default",
+			searchName:      "pod-1",
+			expectedCount:   0,
+		},
+		{
+			name: "no match when namespace differs",
+			existingItems: []ItemBlockItem{
+				{
+					Gr:           podsGR,
+					Item:         newUnstructuredItem("v1", "Pod", "default", "pod-1"),
+					PreferredGVR: podsGVR,
+				},
+			},
+			searchGR:        podsGR,
+			searchNamespace: "kube-system",
+			searchName:      "pod-1",
+			expectedCount:   0,
+		},
+		{
+			name: "no match when name differs",
+			existingItems: []ItemBlockItem{
+				{
+					Gr:           podsGR,
+					Item:         newUnstructuredItem("v1", "Pod", "default", "pod-1"),
+					PreferredGVR: podsGVR,
+				},
+			},
+			searchGR:        podsGR,
+			searchNamespace: "default",
+			searchName:      "pod-2",
+			expectedCount:   0,
+		},
+		{
+			name: "nil item is skipped",
+			existingItems: []ItemBlockItem{
+				{
+					Gr:           podsGR,
+					Item:         nil,
+					PreferredGVR: podsGVR,
+				},
+			},
+			searchGR:        podsGR,
+			searchNamespace: "default",
+			searchName:      "pod-1",
+			expectedCount:   0,
+		},
+		{
+			name: "preferred GVR match is returned first",
+			existingItems: []ItemBlockItem{
+				{
+					Gr:           deploymentsGR,
+					Item:         newUnstructuredItem("apps/v1beta1", "Deployment", "default", "web"),
+					PreferredGVR: deploymentsGVR, // preferred is v1, item is v1beta1 → non-preferred
+				},
+				{
+					Gr:           deploymentsGR,
+					Item:         newUnstructuredItem("apps/v1", "Deployment", "default", "web"),
+					PreferredGVR: deploymentsGVR, // preferred is v1, item is v1 → preferred match
+				},
+			},
+			searchGR:        deploymentsGR,
+			searchNamespace: "default",
+			searchName:      "web",
+			expectedCount:   2,
+			expectedNames:   []string{"web", "web"},
+		},
+		{
+			name: "multiple non-preferred items returned when no preferred match",
+			existingItems: []ItemBlockItem{
+				{
+					Gr:           deploymentsGR,
+					Item:         newUnstructuredItem("apps/v1beta1", "Deployment", "default", "web"),
+					PreferredGVR: deploymentsGVR,
+				},
+				{
+					Gr:           deploymentsGR,
+					Item:         newUnstructuredItem("apps/v1beta2", "Deployment", "default", "web"),
+					PreferredGVR: deploymentsGVR,
+				},
+			},
+			searchGR:        deploymentsGR,
+			searchNamespace: "default",
+			searchName:      "web",
+			expectedCount:   2,
+		},
+		{
+			name: "item with unparsable apiVersion is treated as non-preferred",
+			existingItems: []ItemBlockItem{
+				{
+					Gr:           deploymentsGR,
+					Item:         newUnstructuredItem("not/a/valid/version", "Deployment", "default", "web"),
+					PreferredGVR: deploymentsGVR,
+				},
+			},
+			searchGR:        deploymentsGR,
+			searchNamespace: "default",
+			searchName:      "web",
+			expectedCount:   1,
+		},
+		{
+			name: "only matching GR items are returned from mixed block",
+			existingItems: []ItemBlockItem{
+				{
+					Gr:           podsGR,
+					Item:         newUnstructuredItem("v1", "Pod", "default", "app"),
+					PreferredGVR: podsGVR,
+				},
+				{
+					Gr:           deploymentsGR,
+					Item:         newUnstructuredItem("apps/v1", "Deployment", "default", "app"),
+					PreferredGVR: deploymentsGVR,
+				},
+				{
+					Gr:           servicesGR,
+					Item:         newUnstructuredItem("v1", "Service", "default", "app"),
+					PreferredGVR: schema.GroupVersionResource{Group: "", Version: "v1", Resource: "services"},
+				},
+			},
+			searchGR:        deploymentsGR,
+			searchNamespace: "default",
+			searchName:      "app",
+			expectedCount:   1,
+			expectedNames:   []string{"app"},
+		},
+		{
+			name: "cluster-scoped item found with empty namespace",
+			existingItems: []ItemBlockItem{
+				{
+					Gr:   schema.GroupResource{Group: "", Resource: "namespaces"},
+					Item: newUnstructuredItem("v1", "Namespace", "", "production"),
+					PreferredGVR: schema.GroupVersionResource{
+						Group: "", Version: "v1", Resource: "namespaces",
+					},
+				},
+			},
+			searchGR:        schema.GroupResource{Group: "", Resource: "namespaces"},
+			searchNamespace: "",
+			searchName:      "production",
+			expectedCount:   1,
+			expectedNames:   []string{"production"},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			ib := &ItemBlock{
+				Log:   logrus.New(),
+				Items: tc.existingItems,
+			}
+
+			result := ib.FindItem(tc.searchGR, tc.searchNamespace, tc.searchName)
+
+			require.Len(t, result, tc.expectedCount)
+
+			if tc.expectedNames != nil {
+				for i, expectedName := range tc.expectedNames {
+					assert.Equal(t, expectedName, result[i].Item.GetName())
+				}
+			}
+		})
+	}
+}
+
+func TestFindItemPreferredOrdering(t *testing.T) {
+	deploymentsGR := schema.GroupResource{Group: "apps", Resource: "deployments"}
+	deploymentsGVR := schema.GroupVersionResource{Group: "apps", Version: "v1", Resource: "deployments"}
+
+	// Insert non-preferred first, preferred second to verify ordering
+	ib := &ItemBlock{
+		Log: logrus.New(),
+		Items: []ItemBlockItem{
+			{
+				Gr:           deploymentsGR,
+				Item:         newUnstructuredItem("apps/v1beta1", "Deployment", "ns", "deploy"),
+				PreferredGVR: deploymentsGVR,
+			},
+			{
+				Gr:           deploymentsGR,
+				Item:         newUnstructuredItem("apps/v1", "Deployment", "ns", "deploy"),
+				PreferredGVR: deploymentsGVR,
+			},
+			{
+				Gr:           deploymentsGR,
+				Item:         newUnstructuredItem("apps/v1beta2", "Deployment", "ns", "deploy"),
+				PreferredGVR: deploymentsGVR,
+			},
+		},
+	}
+
+	result := ib.FindItem(deploymentsGR, "ns", "deploy")
+
+	require.Len(t, result, 3)
+
+	// The preferred match (apps/v1) should be first, regardless of insertion order
+	assert.Equal(t, "apps/v1", result[0].Item.GetAPIVersion(),
+		"preferred GVR match should be returned first")
+
+	// Non-preferred items follow in insertion order
+	assert.Equal(t, "apps/v1beta1", result[1].Item.GetAPIVersion())
+	assert.Equal(t, "apps/v1beta2", result[2].Item.GetAPIVersion())
+}

pkg/nodeagent/node_agent.go

@@ -82,13 +82,17 @@ func KbClientIsRunningInNode(ctx context.Context, namespace string, nodeName str
 }
 
 // IsReady checks whether the node-agent daemonset has at least one ready pod
-// by inspecting the DaemonSet status.
+// by inspecting the DaemonSet status. Both the linux and windows daemonsets
+// are checked before returning any non-NotFound lookup error, so that a
+// transient error fetching one daemonset does not mask the other daemonset
+// being ready.
 func IsReady(ctx context.Context, namespace string, crClient ctrlclient.Client) error {
 	dsLinux := new(appsv1api.DaemonSet)
+	var lookupErr error
 	if err := crClient.Get(ctx, ctrlclient.ObjectKey{Namespace: namespace, Name: daemonSet}, dsLinux); err != nil {
 		dsLinux = nil
 		if !apierrors.IsNotFound(err) {
-			return errors.Wrap(err, "failed to get linux node-agent daemonset")
+			lookupErr = errors.Wrap(err, "failed to get linux node-agent daemonset")
 		}
 	}
 
@@ -96,7 +100,7 @@ func IsReady(ctx context.Context, namespace string, crClient ctrlclient.Client)
 	if err := crClient.Get(ctx, ctrlclient.ObjectKey{Namespace: namespace, Name: daemonsetWindows}, dsWindows); err != nil {
 		dsWindows = nil
 		if !apierrors.IsNotFound(err) {
-			return errors.Wrap(err, "failed to get windows node-agent daemonset")
+			lookupErr = errors.CombineErrors(lookupErr, errors.Wrap(err, "failed to get windows node-agent daemonset"))
 		}
 	}
 
@@ -108,6 +112,10 @@ func IsReady(ctx context.Context, namespace string, crClient ctrlclient.Client)
 		return nil
 	}
 
+	if lookupErr != nil {
+		return lookupErr
+	}
+
 	return errors.New("node-agent is not ready: no ready pods found")
 }
 

pkg/nodeagent/node_agent_test.go

@@ -18,6 +18,7 @@ package nodeagent
 
 import (
 	"context"
+	"fmt"
 	"testing"
 
 	"github.com/cockroachdb/errors"
@@ -275,6 +276,52 @@ func TestIsReady(t *testing.T) {
 			},
 			expectErr: "failed to get windows node-agent daemonset: fake-get-error",
 		},
+		{
+			name:      "linux daemonset get error but windows ready",
+			namespace: "fake-ns",
+			kubeClientObj: []runtime.Object{
+				dsWindowsReady,
+			},
+			interceptor: &interceptor.Funcs{
+				Get: func(ctx context.Context, c ctrlclient.WithWatch, key ctrlclient.ObjectKey, obj ctrlclient.Object, opts ...ctrlclient.GetOption) error {
+					if key.Name == "node-agent" {
+						return errors.New("fake-get-error")
+					}
+					return c.Get(ctx, key, obj, opts...)
+				},
+			},
+		},
+		{
+			name:      "windows daemonset get error but linux ready",
+			namespace: "fake-ns",
+			kubeClientObj: []runtime.Object{
+				dsLinuxReady,
+			},
+			interceptor: &interceptor.Funcs{
+				Get: func(ctx context.Context, c ctrlclient.WithWatch, key ctrlclient.ObjectKey, obj ctrlclient.Object, opts ...ctrlclient.GetOption) error {
+					if key.Name == "node-agent-windows" {
+						return errors.New("fake-get-error")
+					}
+					return c.Get(ctx, key, obj, opts...)
+				},
+			},
+		},
+		{
+			name:      "linux daemonset get error and windows not ready",
+			namespace: "fake-ns",
+			kubeClientObj: []runtime.Object{
+				dsWindowsNotReady,
+			},
+			interceptor: &interceptor.Funcs{
+				Get: func(ctx context.Context, c ctrlclient.WithWatch, key ctrlclient.ObjectKey, obj ctrlclient.Object, opts ...ctrlclient.GetOption) error {
+					if key.Name == "node-agent" {
+						return errors.New("fake-get-error")
+					}
+					return c.Get(ctx, key, obj, opts...)
+				},
+			},
+			expectErr: "failed to get linux node-agent daemonset: fake-get-error",
+		},
 		{
 			name:      "linux ds exist but no ready pods",
 			namespace: "fake-ns",
@@ -362,6 +409,34 @@ func TestIsReady(t *testing.T) {
 	}
 }
 
+// TestIsReadyBothDaemonsetsGetError ensures that when both daemonset lookups
+// return a non-NotFound error, the linux error is returned as the primary
+// error while the windows error is retained as a secondary/attached error
+// rather than being silently discarded.
+func TestIsReadyBothDaemonsetsGetError(t *testing.T) {
+	scheme := runtime.NewScheme()
+	appsv1api.AddToScheme(scheme)
+
+	fakeClient := clientFake.NewClientBuilder().
+		WithScheme(scheme).
+		WithInterceptorFuncs(interceptor.Funcs{
+			Get: func(ctx context.Context, c ctrlclient.WithWatch, key ctrlclient.ObjectKey, obj ctrlclient.Object, opts ...ctrlclient.GetOption) error {
+				if key.Name == "node-agent" {
+					return errors.New("fake-linux-get-error")
+				}
+				if key.Name == "node-agent-windows" {
+					return errors.New("fake-windows-get-error")
+				}
+				return c.Get(ctx, key, obj, opts...)
+			},
+		}).
+		Build()
+
+	err := IsReady(t.Context(), "fake-ns", fakeClient)
+	require.EqualError(t, err, "failed to get linux node-agent daemonset: fake-linux-get-error")
+	assert.Contains(t, fmt.Sprintf("%+v", err), "failed to get windows node-agent daemonset: fake-windows-get-error")
+}
+
 func TestGetPodSpec(t *testing.T) {
 	podSpec := corev1api.PodSpec{
 		NodeName: "fake-node",

pkg/podvolume/restore_micro_service_test.go

@@ -165,7 +165,7 @@ func TestOnPvrCompleted(t *testing.T) {
 		{
 			name:        "marshal fail",
 			marshalErr:  errors.New("fake-marshal-error"),
-			expectedErr: "error marshaling restore result {{ } 0}: fake-marshal-error",
+			expectedErr: "error marshaling restore result {{ } 0 0}: fake-marshal-error",
 		},
 		{
 			name:                "succeed",

pkg/podvolume/restorer.go

@@ -38,6 +38,7 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/label"
 	"github.com/vmware-tanzu/velero/pkg/nodeagent"
 	"github.com/vmware-tanzu/velero/pkg/repository"
+	"github.com/vmware-tanzu/velero/pkg/restore/inplace"
 	uploaderutil "github.com/vmware-tanzu/velero/pkg/uploader/util"
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
 	"github.com/vmware-tanzu/velero/pkg/util/kube"
@@ -179,6 +180,18 @@ func (r *restorer) RestorePodVolumes(data RestoreData, tracker *volume.RestoreVo
 			}
 		}
 
+		// Pre-flight checks for in-place restore. Pods gated by this
+		// restore's restore-wait init container are excluded: they must mount
+		// the PVC so the volume gets mounted on the node for the node-agent
+		// 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.CheckPVCNotInUse(r.ctx, r.crClient, pvc, data.Restore.UID); err != nil {
+				errs = append(errs, err)
+				continue
+			}
+		}
+
 		volumeRestore := newPodVolumeRestore(data.Restore, data.Pod, data.BackupLocation, volume, backupInfo.snapshotID, backupInfo.snapshotSize, "", backupInfo.uploaderType, data.SourceNamespace, pvc)
 		if err := veleroclient.CreateRetryGenerateName(r.crClient, r.ctx, volumeRestore); err != nil {
 			errs = append(errs, errors.WithStack(err))

pkg/podvolume/restorer_test.go

@@ -37,6 +37,7 @@ import (
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/builder"
 	"github.com/vmware-tanzu/velero/pkg/repository"
+	"github.com/vmware-tanzu/velero/pkg/restorehelper"
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
 )
 
@@ -119,6 +120,21 @@ func TestGetVolumesRepositoryType(t *testing.T) {
 	}
 }
 
+// createGatedPodObj returns the restored pod as it exists in the cluster:
+// running with the restore-wait init container injected by the given restore.
+func createGatedPodObj(restoreUID string, volumeNum int) *corev1api.Pod {
+	pod := createPodObj(true, true, true, volumeNum)
+	pod.Spec.InitContainers = append([]corev1api.Container{{
+		Name: restorehelper.WaitInitContainer,
+		Args: []string{restoreUID},
+	}}, pod.Spec.InitContainers...)
+	pod.Status.InitContainerStatuses = []corev1api.ContainerStatus{{
+		Name:  restorehelper.WaitInitContainer,
+		State: corev1api.ContainerState{Running: &corev1api.ContainerStateRunning{}},
+	}}
+	return pod
+}
+
 func createNodeAgentDaemonset() *appsv1api.DaemonSet {
 	ds := &appsv1api.DaemonSet{
 		ObjectMeta: metav1.ObjectMeta{
@@ -181,6 +197,7 @@ func TestRestorePodVolumes(t *testing.T) {
 		pvbs            []*velerov1api.PodVolumeBackup
 		restoredPod     *corev1api.Pod
 		sourceNamespace string
+		inplace         bool
 		errs            []expectError
 	}{
 		{
@@ -340,6 +357,86 @@ func TestRestorePodVolumes(t *testing.T) {
 				completedPVR,
 			},
 		},
+		{
+			name: "in-place restore blocked when the PVC is used by another running pod",
+			pvbs: []*velerov1api.PodVolumeBackup{
+				createPVBObj(true, true, 1, "kopia"),
+			},
+			inplace: true,
+			kubeClientObj: []runtime.Object{
+				createNodeAgentDaemonset(),
+				createPVCObj(1),
+				func() *corev1api.Pod {
+					pod := builder.ForPod("fake-ns", "other-pod").
+						Volumes(builder.ForVolume("fake-volume-1").PersistentVolumeClaimSource("fake-pvc-1").Result()).
+						Result()
+					pod.Status.Phase = corev1api.PodRunning
+					return pod
+				}(),
+			},
+			ctlClientObj: []runtime.Object{
+				createBackupRepoObj(),
+			},
+			restoredPod:     createPodObj(true, true, true, 1),
+			sourceNamespace: "fake-ns",
+			bsl:             "fake-bsl",
+			runtimeScheme:   scheme,
+			errs: []expectError{
+				{
+					err:        "in-place restore pre-flight check failed",
+					prefixOnly: true,
+				},
+			},
+		},
+		{
+			name: "in-place restore blocked when the pod is gated by a different restore",
+			pvbs: []*velerov1api.PodVolumeBackup{
+				createPVBObj(true, true, 1, "kopia"),
+			},
+			inplace: true,
+			kubeClientObj: []runtime.Object{
+				createNodeAgentDaemonset(),
+				createPVCObj(1),
+				createGatedPodObj("old-restore-uid", 1),
+			},
+			ctlClientObj: []runtime.Object{
+				createBackupRepoObj(),
+			},
+			restoredPod:     createPodObj(true, true, true, 1),
+			sourceNamespace: "fake-ns",
+			bsl:             "fake-bsl",
+			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{
+				createPVBObj(true, true, 1, "kopia"),
+			},
+			inplace: true,
+			kubeClientObj: []runtime.Object{
+				createNodeAgentDaemonset(),
+				createNodeObj(),
+				createPVCObj(1),
+				createGatedPodObj("fake-restore-uid", 1),
+				createNodeAgentPodObj(true),
+			},
+			ctlClientObj: []runtime.Object{
+				createBackupRepoObj(),
+			},
+			restoredPod:     createPodObj(true, true, true, 1),
+			sourceNamespace: "fake-ns",
+			bsl:             "fake-bsl",
+			runtimeScheme:   scheme,
+			retPVRs: []*velerov1api.PodVolumeRestore{
+				completedPVR,
+			},
+		},
 	}
 
 	for _, test := range tests {
@@ -362,7 +459,12 @@ func TestRestorePodVolumes(t *testing.T) {
 
 			ensurer := repository.NewEnsurer(fakeCRClient, velerotest.NewLogger(), time.Millisecond)
 
-			restoreObj := builder.ForRestore(velerov1api.DefaultNamespace, "fake-restore").Result()
+			restoreBuilder := builder.ForRestore(velerov1api.DefaultNamespace, "fake-restore").
+				ObjectMeta(builder.WithUID("fake-restore-uid"))
+			if test.inplace {
+				restoreBuilder = restoreBuilder.ExistingVolumeDataPolicy(string(velerov1api.VolumeDataPolicyTypeFull))
+			}
+			restoreObj := restoreBuilder.Result()
 
 			rs := newRestorer(ctx, repository.NewRepoLocker(), ensurer, pvrInformer, kubeClient, fakeCRClient, restoreObj, velerotest.NewLogger())
 

pkg/restore/actions/csi/pvc_action.go

@@ -44,6 +44,7 @@ import (
 	plugincommon "github.com/vmware-tanzu/velero/pkg/plugin/framework/common"
 	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
 	riav2 "github.com/vmware-tanzu/velero/pkg/plugin/velero/restoreitemaction/v2"
+	"github.com/vmware-tanzu/velero/pkg/restore/inplace"
 	uploaderUtil "github.com/vmware-tanzu/velero/pkg/uploader/util"
 	"github.com/vmware-tanzu/velero/pkg/util"
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
@@ -236,6 +237,12 @@ func (p *pvcRestoreItemAction) executeWithDataMove(logger *logrus.Entry, input *
 		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.CheckPVCNotInUse(ctx, p.crClient, existingPVC, input.Restore.UID); err != nil {
+			return nil, errors.WithStack(err)
+		}
+
 		// take a CSI snapshot of the existing PVC as the baseline of CBT
 		if input.Restore.IsVolumeDataInplaceIncrementalRestore() && datamover.IsVeleroBlockDataMover(dataUploadResult.DataMover) {
 			logger.Info("ExistingVolumeDataPolicy is in-place incremental restore and data mover is velero-block. Taking a CSI snapshot of the existing PVC as the baseline of CBT...")

pkg/restore/actions/csi/pvc_action_test.go

@@ -741,6 +741,105 @@ func TestExecuteInplaceRestore(t *testing.T) {
 	require.Equal(t, "testPV", dataDownloadList.Items[0].Spec.TargetVolume.PV)
 }
 
+// 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.
+func TestExecuteInplaceRestorePreflight(t *testing.T) {
+	newPodUsingPVC := func(phase corev1api.PodPhase) *corev1api.Pod {
+		pod := builder.ForPod("velero", "consumer-pod").
+			Volumes(builder.ForVolume("data").PersistentVolumeClaimSource("testPVC").Result()).
+			Result()
+		pod.Status.Phase = phase
+		return pod
+	}
+
+	tests := []struct {
+		name        string
+		pod         *corev1api.Pod
+		expectBlock bool
+	}{
+		{
+			name: "no pod, restore proceeds",
+		},
+		{
+			name:        "active pod blocks the restore",
+			pod:         newPodUsingPVC(corev1api.PodRunning),
+			expectBlock: true,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			existingPVC := builder.ForPersistentVolumeClaim("velero", "testPVC").
+				VolumeName("testPV").
+				Phase(corev1api.ClaimBound).Result()
+			existingPV := builder.ForPersistentVolume("testPV").Result()
+			backup := builder.ForBackup("velero", "testBackup").SnapshotMoveData(true).Result()
+			restore := builder.ForRestore("velero", "testRestore").Backup("testBackup").
+				ObjectMeta(builder.WithUID("uid")).ExistingVolumeDataPolicy("full").Result()
+			pvcFromBackup := builder.ForPersistentVolumeClaim("velero", "testPVC").
+				ObjectMeta(builder.WithAnnotations(
+					velerov1api.VolumeSnapshotLabel, "vsName",
+					velerov1api.DataUploadNameAnnotation, "velero/testDU",
+				)).Result()
+			dataUploadResult := builder.ForConfigMap("velero", "testCM").Data("uid", "{}").
+				ObjectMeta(builder.WithLabels(
+					velerov1api.RestoreUIDLabel, "uid",
+					velerov1api.PVCNamespaceNameLabel, "velero.testPVC",
+					velerov1api.ResourceUsageLabel, label.GetValidName(string(velerov1api.VeleroResourceUsageDataUploadResult)),
+				)).Result()
+
+			crObjects := []runtime.Object{existingPVC, existingPV, backup, dataUploadResult}
+			kubeObjects := []runtime.Object{existingPVC, existingPV}
+			if tc.pod != nil {
+				crObjects = append(crObjects, tc.pod)
+				kubeObjects = append(kubeObjects, tc.pod)
+			}
+
+			pvcRIA := pvcRestoreItemAction{
+				log:        logrus.New(),
+				crClient:   velerotest.NewFakeControllerRuntimeClient(t, crObjects...),
+				kubeClient: fake.NewSimpleClientset(kubeObjects...),
+			}
+
+			pvcMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(pvcFromBackup.DeepCopy())
+			require.NoError(t, err)
+			pvcFromBackupMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(pvcFromBackup)
+			require.NoError(t, err)
+
+			_, err = pvcRIA.Execute(&velero.RestoreItemActionExecuteInput{
+				Item:           &unstructured.Unstructured{Object: pvcMap},
+				ItemFromBackup: &unstructured.Unstructured{Object: pvcFromBackupMap},
+				Restore:        restore,
+			})
+
+			gotPVC, getErr := pvcRIA.kubeClient.CoreV1().PersistentVolumeClaims("velero").Get(t.Context(), "testPVC", metav1.GetOptions{})
+			dataDownloadList := new(velerov2alpha1.DataDownloadList)
+			require.NoError(t, pvcRIA.crClient.List(t.Context(), dataDownloadList, &crclient.ListOptions{}))
+
+			if tc.expectBlock {
+				require.Error(t, err)
+				require.Contains(t, err.Error(), "pre-flight check failed")
+				require.Contains(t, err.Error(), "consumer-pod")
+				// No side effects: PVC untouched with the original volumeName,
+				// PV reclaim policy not patched, no DataDownload created.
+				require.NoError(t, getErr)
+				require.Equal(t, "testPV", gotPVC.Spec.VolumeName)
+				gotPV, pvErr := pvcRIA.kubeClient.CoreV1().PersistentVolumes().Get(t.Context(), "testPV", metav1.GetOptions{})
+				require.NoError(t, pvErr)
+				require.Equal(t, existingPV.Spec.PersistentVolumeReclaimPolicy, gotPV.Spec.PersistentVolumeReclaimPolicy)
+				require.Empty(t, dataDownloadList.Items)
+			} else {
+				require.NoError(t, err)
+				// The in-place restore proceeded: the existing PVC is deleted
+				// and a DataDownload is created.
+				require.True(t, apierrors.IsNotFound(getErr))
+				require.Len(t, dataDownloadList.Items, 1)
+			}
+		})
+	}
+}
+
 func TestPVCAppliesTo(t *testing.T) {
 	p := pvcRestoreItemAction{
 		log: logrus.StandardLogger(),

pkg/restore/inplace/preflight.go

@@ -0,0 +1,131 @@
+/*
+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 inplace holds the pre-flight checks for in-place volume data
+// restores. The checks must pass before Velero performs any side effect on
+// the existing PVC/PV.
+package inplace
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"github.com/cockroachdb/errors"
+	corev1api "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
+	crclient "sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/vmware-tanzu/velero/pkg/restorehelper"
+)
+
+// CheckPVCNotInUse verifies the target PVC is not used by any active pod,
+// aligned with the pvc-protection controller semantics: terminal-phase pods
+// (Succeeded/Failed) don't block; all other phases do, and terminating pods
+// are flagged so the message can hint the user to wait.
+//
+// Pods gated by this restore's restore-wait init container are exempted: on
+// the file system restore path the restored pods must mount the PVC for the
+// node-agent to restore the data, and an RWX PVC may be mounted by several of
+// them. Such a pod cannot write to the volume since its workload containers
+// are blocked until this restore's PodVolumeRestores complete (see
+// gatedByThisRestore). Any other pod, including one gated by a different
+// restore whose release timing is out of our control, still blocks.
+func CheckPVCNotInUse(
+	ctx context.Context,
+	cli crclient.Client,
+	pvc *corev1api.PersistentVolumeClaim,
+	restoreUID types.UID,
+) error {
+	podList := new(corev1api.PodList)
+	if err := cli.List(ctx, podList, &crclient.ListOptions{Namespace: pvc.Namespace}); err != nil {
+		return errors.Wrapf(err, "failed to check whether PVC %s/%s is in use: failed to list pods in namespace %s", pvc.Namespace, pvc.Name, pvc.Namespace)
+	}
+
+	podsInUse := []string{}
+	terminatingOnly := true
+	for i := range podList.Items {
+		pod := &podList.Items[i]
+		if !podUsesPVC(pod, pvc.Name) ||
+			pod.Status.Phase == corev1api.PodSucceeded || pod.Status.Phase == corev1api.PodFailed ||
+			gatedByThisRestore(pod, restoreUID) {
+			continue
+		}
+		state := string(pod.Status.Phase)
+		if pod.DeletionTimestamp != nil {
+			state += ", terminating"
+		} else {
+			terminatingOnly = false
+		}
+		podsInUse = append(podsInUse, fmt.Sprintf("%s (%s)", pod.Name, state))
+	}
+	if len(podsInUse) == 0 {
+		return nil
+	}
+
+	hint := "delete the workloads consuming the PVC and retry"
+	if terminatingOnly {
+		hint = "the pod(s) are terminating; retry after they are fully removed"
+	}
+	return errors.Errorf("in-place restore pre-flight check failed, skipping volume data restore: PVC %s/%s is still in use by pod(s) [%s]: %s",
+		pvc.Namespace, pvc.Name, strings.Join(podsInUse, ", "), hint)
+}
+
+func podUsesPVC(pod *corev1api.Pod, pvcName string) bool {
+	for _, vol := range pod.Spec.Volumes {
+		if vol.PersistentVolumeClaim != nil && vol.PersistentVolumeClaim.ClaimName == pvcName {
+			return true
+		}
+	}
+	return false
+}
+
+// gatedByThisRestore reports whether the pod is blocked by the restore-wait
+// init container injected by this restore, identified by the restore UID in
+// the init container's args. Such a pod cannot write to the volume: its
+// workload containers won't start until this restore's PodVolumeRestores
+// complete and write the done signal. A terminated init container means the
+// gate is already open, so the pod is no longer exempted. Pods gated by a
+// different restore are not exempted either, since their release timing is
+// unrelated to this restore.
+func gatedByThisRestore(pod *corev1api.Pod, restoreUID types.UID) bool {
+	if restoreUID == "" {
+		return false
+	}
+
+	idx := -1
+	for i, c := range pod.Spec.InitContainers {
+		if c.Name == restorehelper.WaitInitContainer {
+			if len(c.Args) == 0 || c.Args[0] != string(restoreUID) {
+				return false
+			}
+			idx = i
+			break
+		}
+	}
+	if idx < 0 {
+		return false
+	}
+
+	for _, cs := range pod.Status.InitContainerStatuses {
+		if cs.Name == restorehelper.WaitInitContainer {
+			return cs.State.Terminated == nil
+		}
+	}
+	// Statuses not populated yet: the init container hasn't run, so the gate
+	// is still closed.
+	return true
+}

pkg/restore/inplace/preflight_test.go

@@ -0,0 +1,202 @@
+/*
+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 inplace
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	corev1api "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+
+	"github.com/vmware-tanzu/velero/pkg/restorehelper"
+	velerotest "github.com/vmware-tanzu/velero/pkg/test"
+)
+
+func podUsingPVC(name, pvcName string, phase corev1api.PodPhase, terminating bool) *corev1api.Pod {
+	pod := &corev1api.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: "default",
+			UID:       types.UID(name + "-uid"),
+		},
+		Spec: corev1api.PodSpec{
+			Volumes: []corev1api.Volume{
+				{
+					Name: "data",
+					VolumeSource: corev1api.VolumeSource{
+						PersistentVolumeClaim: &corev1api.PersistentVolumeClaimVolumeSource{
+							ClaimName: pvcName,
+						},
+					},
+				},
+			},
+		},
+		Status: corev1api.PodStatus{Phase: phase},
+	}
+	if terminating {
+		now := metav1.Now()
+		pod.DeletionTimestamp = &now
+		pod.Finalizers = []string{"fake-finalizer"}
+	}
+	return pod
+}
+
+// gatedPod adds the restore-wait init container (as injected by the
+// PodVolumeRestoreAction RIA, carrying the restore UID in args) to the pod.
+// terminated simulates the gate having already been released.
+func gatedPod(pod *corev1api.Pod, restoreUID string, terminated bool) *corev1api.Pod {
+	pod.Spec.InitContainers = append([]corev1api.Container{{
+		Name: restorehelper.WaitInitContainer,
+		Args: []string{restoreUID},
+	}}, pod.Spec.InitContainers...)
+	status := corev1api.ContainerStatus{
+		Name:  restorehelper.WaitInitContainer,
+		State: corev1api.ContainerState{Running: &corev1api.ContainerStateRunning{}},
+	}
+	if terminated {
+		status.State = corev1api.ContainerState{Terminated: &corev1api.ContainerStateTerminated{}}
+	}
+	pod.Status.InitContainerStatuses = []corev1api.ContainerStatus{status}
+	return pod
+}
+
+func TestCheckPVCNotInUse(t *testing.T) {
+	tests := []struct {
+		name          string
+		pods          []*corev1api.Pod
+		restoreUID    types.UID
+		expectPass    bool
+		expectMessage []string
+	}{
+		{
+			name:       "no pods, check passes",
+			expectPass: true,
+		},
+		{
+			name:          "active pod blocks with the delete hint",
+			pods:          []*corev1api.Pod{podUsingPVC("pod-1", "pvc-1", corev1api.PodRunning, false)},
+			expectMessage: []string{"pod-1 (Running)", "delete the workloads"},
+		},
+		{
+			name:          "unknown-phase pod blocks (node may be unreachable)",
+			pods:          []*corev1api.Pod{podUsingPVC("pod-1", "pvc-1", corev1api.PodUnknown, false)},
+			expectMessage: []string{"pod-1 (Unknown)"},
+		},
+		{
+			name:          "terminating pod blocks with the wait hint",
+			pods:          []*corev1api.Pod{podUsingPVC("pod-1", "pvc-1", corev1api.PodRunning, true)},
+			expectMessage: []string{"pod-1 (Running, terminating)", "retry after they are fully removed"},
+		},
+		{
+			name: "terminal-phase pods do not block",
+			pods: []*corev1api.Pod{
+				podUsingPVC("pod-1", "pvc-1", corev1api.PodSucceeded, false),
+				podUsingPVC("pod-2", "pvc-1", corev1api.PodFailed, false),
+			},
+			expectPass: true,
+		},
+		{
+			name:       "pod using another PVC does not block",
+			pods:       []*corev1api.Pod{podUsingPVC("pod-1", "other-pvc", corev1api.PodRunning, false)},
+			expectPass: true,
+		},
+		{
+			name: "pods gated by this restore do not block, other pods still do",
+			pods: []*corev1api.Pod{
+				gatedPod(podUsingPVC("restored-pod-1", "pvc-1", corev1api.PodPending, false), "restore-uid", false),
+				gatedPod(podUsingPVC("restored-pod-2", "pvc-1", corev1api.PodPending, false), "restore-uid", false),
+				podUsingPVC("other-pod", "pvc-1", corev1api.PodRunning, false),
+			},
+			restoreUID:    "restore-uid",
+			expectMessage: []string{"[other-pod (Running)]"},
+		},
+		{
+			name: "multiple pods gated by this restore pass the check",
+			pods: []*corev1api.Pod{
+				gatedPod(podUsingPVC("restored-pod-1", "pvc-1", corev1api.PodPending, false), "restore-uid", false),
+				gatedPod(podUsingPVC("restored-pod-2", "pvc-1", corev1api.PodPending, false), "restore-uid", false),
+			},
+			restoreUID: "restore-uid",
+			expectPass: true,
+		},
+		{
+			name: "pod gated by a different restore still blocks",
+			pods: []*corev1api.Pod{
+				gatedPod(podUsingPVC("old-restored-pod", "pvc-1", corev1api.PodPending, false), "old-restore-uid", false),
+			},
+			restoreUID:    "restore-uid",
+			expectMessage: []string{"[old-restored-pod (Pending)]"},
+		},
+		{
+			name: "pod whose restore-wait already terminated still blocks",
+			pods: []*corev1api.Pod{
+				gatedPod(podUsingPVC("released-pod", "pvc-1", corev1api.PodRunning, false), "restore-uid", true),
+			},
+			restoreUID:    "restore-uid",
+			expectMessage: []string{"[released-pod (Running)]"},
+		},
+		{
+			name: "gated pod without init container status yet passes the check",
+			pods: []*corev1api.Pod{
+				func() *corev1api.Pod {
+					pod := gatedPod(podUsingPVC("new-pod", "pvc-1", corev1api.PodPending, false), "restore-uid", false)
+					pod.Status.InitContainerStatuses = nil
+					return pod
+				}(),
+			},
+			restoreUID: "restore-uid",
+			expectPass: true,
+		},
+		{
+			name: "empty restore UID exempts nothing",
+			pods: []*corev1api.Pod{
+				gatedPod(podUsingPVC("restored-pod", "pvc-1", corev1api.PodPending, false), "", false),
+			},
+			restoreUID:    "",
+			expectMessage: []string{"[restored-pod (Pending)]"},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			objs := []runtime.Object{}
+			for _, pod := range tc.pods {
+				objs = append(objs, pod)
+			}
+			cli := velerotest.NewFakeControllerRuntimeClient(t, objs...)
+
+			pvc := &corev1api.PersistentVolumeClaim{
+				ObjectMeta: metav1.ObjectMeta{Name: "pvc-1", Namespace: "default"},
+			}
+
+			err := CheckPVCNotInUse(t.Context(), cli, pvc, tc.restoreUID)
+			if tc.expectPass {
+				require.NoError(t, err)
+				return
+			}
+			require.Error(t, err)
+			assert.Contains(t, err.Error(), "pre-flight check failed")
+			for _, fragment := range tc.expectMessage {
+				assert.Contains(t, err.Error(), fragment)
+			}
+		})
+	}
+}

pkg/uploader/block/snapshot.go

@@ -205,12 +205,12 @@ func getParentBackupInfo(ctx context.Context, rep udmrepo.BackupRepo, forceFull
 }
 
 // Restore restore specific sourcePath with given snapshotID and update progress
-func Restore(ctx context.Context, blkUp Uploader, rep udmrepo.BackupRepo, snapshotID, dest string, incremental bool, cbtSource cbtservice.SourceInfo, cbtService cbtservice.Service, uploaderCfg map[string]string, log logrus.FieldLogger) (int64, error) {
+func Restore(ctx context.Context, blkUp Uploader, rep udmrepo.BackupRepo, snapshotID, dest string, incremental bool, cbtSource cbtservice.SourceInfo, cbtService cbtservice.Service, uploaderCfg map[string]string, log logrus.FieldLogger) (int64, int64, error) {
 	log.Info("Start to restore...")
 
 	snapshot, err := rep.GetSnapshot(ctx, udmrepo.ID(snapshotID))
 	if err != nil {
-		return 0, errors.Wrapf(err, "Unable to load snapshot %v", snapshotID)
+		return 0, 0, errors.Wrapf(err, "Unable to load snapshot %v", snapshotID)
 	}
 	log.Infof("Restore from snapshot %s, incremental %v, cbt source %v, description %s, created time %v, tags %v", snapshotID, incremental, cbtSource, snapshot.Description, snapshot.EndTime, snapshot.Tags)
 
@@ -246,34 +246,34 @@ func Restore(ctx context.Context, blkUp Uploader, rep udmrepo.BackupRepo, snapsh
 
 	destPath, err := filepath.Abs(dest)
 	if err != nil {
-		return 0, errors.Wrapf(err, "invalid dest path '%s'", dest)
+		return 0, 0, errors.Wrapf(err, "invalid dest path '%s'", dest)
 	}
 
 	destPath = filepath.Clean(destPath)
 
 	destDev, err := openBlockDeviceFunc(destPath, false)
 	if err != nil {
-		return 0, errors.Wrapf(err, "error opening block device '%s'", destPath)
+		return 0, 0, errors.Wrapf(err, "error opening block device '%s'", destPath)
 	}
 
 	defer destDev.Close()
 
 	destSize, err := destDev.Seek(0, io.SeekEnd)
 	if err != nil {
-		return 0, errors.Wrapf(err, "error getting length of block device %s", dest)
+		return 0, 0, errors.Wrapf(err, "error getting length of block device %s", dest)
 	}
 
 	_, err = destDev.Seek(0, io.SeekStart)
 	if err != nil {
-		return 0, errors.Wrapf(err, "error reset pos of block device %s", dest)
+		return 0, 0, errors.Wrapf(err, "error reset pos of block device %s", dest)
 	}
 
-	_, totalSize, err := blkUp.Restore(snapshot, destInfo{dev: destDev, path: destPath, size: destSize}, bitmap.Iterator(), uploaderCfg)
+	incrementalBytes, totalSize, err := blkUp.Restore(snapshot, destInfo{dev: destDev, path: destPath, size: destSize}, bitmap.Iterator(), uploaderCfg)
 	if err != nil {
-		return 0, errors.Wrapf(err, "error restoring to block dev %s", destPath)
+		return 0, 0, errors.Wrapf(err, "error restoring to block dev %s", destPath)
 	}
 
-	return totalSize, nil
+	return incrementalBytes, totalSize, nil
 }
 
 func findPreviousSnapshot(ctx context.Context, rep udmrepo.BackupRepo, path string, snapshotTags map[string]string, noLaterThan *time.Time, log logrus.FieldLogger) (udmrepo.Snapshot, error) {

pkg/uploader/block/snapshot_test.go

@@ -842,7 +842,7 @@ func TestRestore(t *testing.T) {
 				cbtSvc = tc.cbtService(t)
 			}
 
-			size, err := Restore(ctx, mockBlkup, mockRepo, "snap-001", "/dev/sdb", tc.incremental, tc.cbtSource, cbtSvc, map[string]string{}, testLog())
+			_, size, err := Restore(ctx, mockBlkup, mockRepo, "snap-001", "/dev/sdb", tc.incremental, tc.cbtSource, cbtSvc, map[string]string{}, testLog())
 
 			if tc.expectedErrStr != "" {
 				require.Error(t, err)

pkg/uploader/provider/block.go

@@ -167,9 +167,9 @@ func (bp *blockProvider) RunRestore(
 	cbtParam CBTParam,
 	volMode uploader.PersistentVolumeMode,
 	uploaderCfg map[string]string,
-	updater uploader.ProgressUpdater) (int64, error) {
+	updater uploader.ProgressUpdater) (int64, int64, error) {
 	if updater == nil {
-		return 0, errors.New("restore progress updater is invalid")
+		return 0, 0, errors.New("restore progress updater is invalid")
 	}
 
 	log := bp.log.WithFields(logrus.Fields{
@@ -180,24 +180,24 @@ func (bp *blockProvider) RunRestore(
 
 	blkUploader := block.NewUploader(ctx, bp.bkRepo, updater, log)
 
-	size, err := blockRestoreFunc(ctx, blkUploader, bp.bkRepo, snapshotID, volumePath, incremental, cbtParam.Source, cbtParam.Service, uploaderCfg, log)
+	incrementalBytes, totalBytes, err := blockRestoreFunc(ctx, blkUploader, bp.bkRepo, snapshotID, volumePath, incremental, cbtParam.Source, cbtParam.Service, uploaderCfg, log)
 
 	// errors.Is, not ==: see the equivalent comment on the backup path above.
 	if errors.Is(err, block.ErrCanceled) {
 		log.Warn("Block restore is canceled")
-		return 0, ErrorCanceled
+		return 0, 0, ErrorCanceled
 	}
 
 	if err != nil {
-		return 0, errors.Wrapf(err, "Failed to run block restore")
+		return 0, 0, errors.Wrapf(err, "Failed to run block restore")
 	}
 
 	updater.UpdateProgress(&uploader.Progress{
-		TotalBytes: size,
-		BytesDone:  size,
+		TotalBytes: totalBytes,
+		BytesDone:  totalBytes,
 	})
 
-	log.Infof("Block restore finished, restore size %v", size)
+	log.Infof("Block restore finished, restore incremental size %v, total size %v", incrementalBytes, totalBytes)
 
-	return size, nil
+	return incrementalBytes, totalBytes, nil
 }

pkg/uploader/provider/block_test.go

@@ -412,8 +412,8 @@ func TestBlockProviderCancelThroughWrappedError(t *testing.T) {
 	t.Run("restore", func(t *testing.T) {
 		orig := blockRestoreFunc
 		defer func() { blockRestoreFunc = orig }()
-		blockRestoreFunc = func(_ context.Context, _ block.Uploader, _ udmrepo.BackupRepo, _ string, _ string, _ bool, _ cbtservice.SourceInfo, _ cbtservice.Service, _ map[string]string, _ logrus.FieldLogger) (int64, error) {
-			return 0, errors.Wrap(block.ErrCanceled, "error restoring bdev")
+		blockRestoreFunc = func(_ context.Context, _ block.Uploader, _ udmrepo.BackupRepo, _ string, _ string, _ bool, _ cbtservice.SourceInfo, _ cbtservice.Service, _ map[string]string, _ logrus.FieldLogger) (int64, int64, error) {
+			return 0, 0, errors.Wrap(block.ErrCanceled, "error restoring bdev")
 		}
 
 		bp := &blockProvider{
@@ -422,7 +422,7 @@ func TestBlockProviderCancelThroughWrappedError(t *testing.T) {
 			log:           logrus.New(),
 		}
 
-		_, err := bp.RunRestore(t.Context(), "snap-1", "/dev/sda", false, CBTParam{},
+		_, _, err := bp.RunRestore(t.Context(), "snap-1", "/dev/sda", false, CBTParam{},
 			uploader.PersistentVolumeBlock, map[string]string{}, &blockMockProgressUpdater{})
 
 		require.ErrorIs(t, err, ErrorCanceled)
@@ -496,10 +496,10 @@ func TestBlockProviderRunRestore(t *testing.T) {
 			var capturedSnapshotID string
 			var capturedVolumePath string
 
-			blockRestoreFunc = func(ctx context.Context, blkUp block.Uploader, rep udmrepo.BackupRepo, snapshotID string, dest string, incremental bool, cbtSource cbtservice.SourceInfo, cbtService cbtservice.Service, uploaderCfg map[string]string, log logrus.FieldLogger) (int64, error) {
+			blockRestoreFunc = func(ctx context.Context, blkUp block.Uploader, rep udmrepo.BackupRepo, snapshotID string, dest string, incremental bool, cbtSource cbtservice.SourceInfo, cbtService cbtservice.Service, uploaderCfg map[string]string, log logrus.FieldLogger) (int64, int64, error) {
 				capturedSnapshotID = snapshotID
 				capturedVolumePath = dest
-				return tc.mockRestoreSize, tc.mockRestoreErr
+				return tc.mockRestoreSize, tc.mockRestoreSize, tc.mockRestoreErr
 			}
 
 			bp := &blockProvider{
@@ -507,7 +507,7 @@ func TestBlockProviderRunRestore(t *testing.T) {
 				log:    logrus.New(),
 			}
 
-			size, err := bp.RunRestore(
+			_, size, err := bp.RunRestore(
 				t.Context(),
 				tc.snapshotID,
 				tc.volumePath,

pkg/uploader/provider/kopia.go

@@ -215,7 +215,7 @@ func (kp *kopiaProvider) RunRestore(
 	_ CBTParam,
 	volMode uploader.PersistentVolumeMode,
 	uploaderCfg map[string]string,
-	updater uploader.ProgressUpdater) (int64, error) {
+	updater uploader.ProgressUpdater) (int64, int64, error) {
 	log := kp.log.WithFields(logrus.Fields{
 		"snapshotID": snapshotID,
 		"volumePath": volumePath,
@@ -239,12 +239,12 @@ func (kp *kopiaProvider) RunRestore(
 	size, fileCount, err := kopiaRestoreFunc(context.Background(), repoWriter, progress, snapshotID, volumePath, incremental, volMode, uploaderCfg, log, restoreCancel)
 
 	if err != nil {
-		return 0, errors.Wrapf(err, "Failed to run kopia restore")
+		return 0, 0, errors.Wrapf(err, "Failed to run kopia restore")
 	}
 
 	if atomic.LoadInt32(&kp.canceling) == 1 {
 		log.Error("Kopia restore is canceled")
-		return 0, ErrorCanceled
+		return 0, 0, ErrorCanceled
 	}
 
 	// which ensure that the statistic data of TotalBytes equal to BytesDone when finished
@@ -257,5 +257,6 @@ func (kp *kopiaProvider) RunRestore(
 
 	log.Info(output)
 
-	return size, nil
+	// the incremental bytes is the same as the total bytes because total bytes is the size of actual data Kopia writes
+	return size, size, nil
 }

pkg/uploader/provider/kopia_test.go

@@ -157,8 +157,10 @@ func TestRunRestore(t *testing.T) {
 			if tc.volMode == "" {
 				tc.volMode = uploader.PersistentVolumeFilesystem
 			}
-			kopiaRestoreFunc = tc.hookRestoreFunc
-			_, err := kp.RunRestore(t.Context(), "", "/var", tc.incremental, CBTParam{}, tc.volMode, map[string]string{}, &updater)
+			kopiaRestoreFunc = func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, incremental bool, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) {
+				return tc.hookRestoreFunc(ctx, rep, progress, snapshotID, dest, incremental, volMode, uploaderCfg, log, cancleCh)
+			}
+			_, _, err := kp.RunRestore(t.Context(), "", "/var", tc.incremental, CBTParam{}, tc.volMode, map[string]string{}, &updater)
 			if tc.notError {
 				assert.NoError(t, err)
 			} else {

pkg/uploader/provider/mocks/Provider.go

@@ -223,7 +223,7 @@ func (_c *Provider_RunBackup_Call) RunAndReturn(run func(ctx context.Context, pa
 }
 
 // RunRestore provides a mock function for the type Provider
-func (_mock *Provider) RunRestore(ctx context.Context, snapshotID string, volumePath string, incremental bool, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater) (int64, error) {
+func (_mock *Provider) RunRestore(ctx context.Context, snapshotID string, volumePath string, incremental bool, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater) (int64, int64, error) {
 	ret := _mock.Called(ctx, snapshotID, volumePath, incremental, cbtParam, volMode, uploaderConfig, updater)
 
 	if len(ret) == 0 {
@@ -231,8 +231,9 @@ func (_mock *Provider) RunRestore(ctx context.Context, snapshotID string, volume
 	}
 
 	var r0 int64
-	var r1 error
-	if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, bool, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) (int64, error)); ok {
+	var r1 int64
+	var r2 error
+	if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, bool, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) (int64, int64, error)); ok {
 		return returnFunc(ctx, snapshotID, volumePath, incremental, cbtParam, volMode, uploaderConfig, updater)
 	}
 	if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, bool, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) int64); ok {
@@ -240,12 +241,20 @@ func (_mock *Provider) RunRestore(ctx context.Context, snapshotID string, volume
 	} else {
 		r0 = ret.Get(0).(int64)
 	}
-	if returnFunc, ok := ret.Get(1).(func(context.Context, string, string, bool, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) error); ok {
+
+	if returnFunc, ok := ret.Get(1).(func(context.Context, string, string, bool, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) int64); ok {
 		r1 = returnFunc(ctx, snapshotID, volumePath, incremental, cbtParam, volMode, uploaderConfig, updater)
 	} else {
-		r1 = ret.Error(1)
+		r1 = ret.Get(1).(int64)
 	}
-	return r0, r1
+
+	if returnFunc, ok := ret.Get(2).(func(context.Context, string, string, bool, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) error); ok {
+		r2 = returnFunc(ctx, snapshotID, volumePath, incremental, cbtParam, volMode, uploaderConfig, updater)
+	} else {
+		r2 = ret.Error(2)
+	}
+
+	return r0, r1, r2
 }
 
 // Provider_RunRestore_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'RunRestore'
@@ -314,12 +323,12 @@ func (_c *Provider_RunRestore_Call) Run(run func(ctx context.Context, snapshotID
 	return _c
 }
 
-func (_c *Provider_RunRestore_Call) Return(n int64, err error) *Provider_RunRestore_Call {
-	_c.Call.Return(n, err)
+func (_c *Provider_RunRestore_Call) Return(_a0 int64, _a1 int64, _a2 error) *Provider_RunRestore_Call {
+	_c.Call.Return(_a0, _a1, _a2)
 	return _c
 }
 
-func (_c *Provider_RunRestore_Call) RunAndReturn(run func(ctx context.Context, snapshotID string, volumePath string, incremental bool, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater) (int64, error)) *Provider_RunRestore_Call {
+func (_c *Provider_RunRestore_Call) RunAndReturn(run func(ctx context.Context, snapshotID string, volumePath string, incremental bool, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater) (int64, int64, error)) *Provider_RunRestore_Call {
 	_c.Call.Return(run)
 	return _c
 }

pkg/uploader/provider/provider.go

@@ -68,7 +68,7 @@ type Provider interface {
 		cbtParam CBTParam,
 		volMode uploader.PersistentVolumeMode,
 		uploaderConfig map[string]string,
-		updater uploader.ProgressUpdater) (int64, error)
+		updater uploader.ProgressUpdater) (int64, int64, error)
 	// Close which will close related repository
 	Close(ctx context.Context) error
 }

pkg/util/csi/volume_snapshot_test.go

@@ -2179,3 +2179,107 @@ func TestGetVSCForVS(t *testing.T) {
 		})
 	}
 }
+
+func TestCleanupVolumeSnapshot(t *testing.T) {
+	retainVSCName := "retain-vsc"
+
+	testCases := []struct {
+		name          string
+		volSnap       *snapshotv1api.VolumeSnapshot
+		objs          []runtime.Object
+		expectDeleted bool
+		// name of the VolumeSnapshotContent expected to have been patched to
+		// DeletionPolicy=Delete; empty when no VSC should be touched.
+		expectedVSC string
+	}{
+		{
+			name: "should be a no-op if the VolumeSnapshot no longer exists",
+			volSnap: &snapshotv1api.VolumeSnapshot{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "does-not-exist",
+					Namespace: "velero",
+				},
+			},
+			objs:          []runtime.Object{},
+			expectDeleted: false,
+		},
+		{
+			name: "should delete a VolumeSnapshot with no bound VolumeSnapshotContent",
+			volSnap: &snapshotv1api.VolumeSnapshot{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "vs-no-vsc",
+					Namespace: "velero",
+				},
+			},
+			objs: []runtime.Object{
+				&snapshotv1api.VolumeSnapshot{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "vs-no-vsc",
+						Namespace: "velero",
+					},
+				},
+			},
+			expectDeleted: true,
+		},
+		{
+			name: "should patch bound VSC DeletionPolicy to Delete and delete the VolumeSnapshot",
+			volSnap: &snapshotv1api.VolumeSnapshot{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "vs-with-vsc",
+					Namespace: "velero",
+				},
+			},
+			objs: []runtime.Object{
+				&snapshotv1api.VolumeSnapshot{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "vs-with-vsc",
+						Namespace: "velero",
+					},
+					Status: &snapshotv1api.VolumeSnapshotStatus{
+						BoundVolumeSnapshotContentName: &retainVSCName,
+					},
+				},
+				&snapshotv1api.VolumeSnapshotContent{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "retain-vsc",
+					},
+					Spec: snapshotv1api.VolumeSnapshotContentSpec{
+						DeletionPolicy: snapshotv1api.VolumeSnapshotContentRetain,
+					},
+				},
+			},
+			expectDeleted: true,
+			expectedVSC:   "retain-vsc",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			fakeClient := velerotest.NewFakeControllerRuntimeClient(t, tc.objs...)
+
+			CleanupVolumeSnapshot(t.Context(), tc.volSnap, fakeClient, velerotest.NewLogger())
+
+			actual := new(snapshotv1api.VolumeSnapshot)
+			err := fakeClient.Get(
+				t.Context(),
+				crclient.ObjectKey{Name: tc.volSnap.Name, Namespace: tc.volSnap.Namespace},
+				actual,
+			)
+
+			if tc.expectDeleted {
+				assert.True(t, apierrors.IsNotFound(err), "expected VolumeSnapshot to be deleted")
+			}
+
+			if tc.expectedVSC != "" {
+				actualVSC := new(snapshotv1api.VolumeSnapshotContent)
+				err := fakeClient.Get(
+					t.Context(),
+					crclient.ObjectKey{Name: tc.expectedVSC},
+					actualVSC,
+				)
+				require.NoError(t, err)
+				assert.Equal(t, snapshotv1api.VolumeSnapshotContentDelete, actualVSC.Spec.DeletionPolicy)
+			}
+		})
+	}
+}