Merge branch 'main' into update-du-dd-progress-when-terminal-event-is-missing

.golangci.yaml

@@ -227,6 +227,12 @@ linters-settings:
     require-explanation: true
     # Enable to require nolint directives to mention the specific linter being suppressed. Default is false.
     require-specific: true
+  perfsprint:
+    strconcat: false
+    sprintf1: false
+    errorf: false
+    int-conversion: true
+    fiximports: true
   revive:
     rules:
       - name: unexported-return
@@ -310,6 +316,7 @@ linters:
     - nilerr
     - noctx
     - nolintlint
+    - perfsprint
     - revive
     - staticcheck
     - stylecheck
@@ -331,7 +338,6 @@ issues:
       text: "DefaultVolumesToRestic" # No need to report deprecate for DefaultVolumesToRestic.
     - path: ".*_test.go$"
       linters:
-        - dupword
         - errcheck
         - goconst
         - gosec
@@ -342,7 +348,6 @@ issues:
         - unused
     - path: test/
       linters:
-        - dupword
         - errcheck
         - goconst
         - gosec
@@ -351,6 +356,14 @@ issues:
         - stylecheck
         - unparam
         - unused
+    - path: ".*data_upload_controller_test.go$"
+      linters:
+        - dupword
+      text: "type"
+    - path: ".*config_test.go$"
+      linters:
+        - dupword
+      text: "bucket"
 
   # The list of ids of default excludes to include or disable. By default it's empty.
   include:

changelogs/unreleased/8550-Lyndon-Li

@@ -0,0 +1 @@
+Fix issue #8044, allow users to ignore delay binding the restorePVC of data mover when it is in WaitForFirstConsumer mode

changelogs/unreleased/8590-Lyndon-Li

@@ -0,0 +1 @@
+Fix issue #8579, set event burst to block event broadcaster from filtering events

changelogs/unreleased/8591-reasonerjt

@@ -0,0 +1 @@
+Skip patching the PV in finalization for failed operation

changelogs/unreleased/8594-Lyndon-Li

@@ -0,0 +1 @@
+Data mover restore for Windows

changelogs/unreleased/8602-Lyndon-Li

@@ -0,0 +1 @@
+Fix issue #8067, add tmp folder (/tmp for linux, C:\Windows\Temp for Windows) as an alternative of udmrepo's config file location

changelogs/unreleased/8603-ywk253100

@@ -0,0 +1 @@
+Check the PVB status via podvolume Backupper rather than calling API server to avoid API server issue

changelogs/unreleased/8606-Lyndon-Li

@@ -0,0 +1 @@
+Fix issue #8418, add Windows toleration to data mover pods

config/crd/v2alpha1/bases/velero.io_datadownloads.yaml

@@ -92,6 +92,13 @@ spec:
                   DataMover specifies the data mover to be used by the backup.
                   If DataMover is "" or "velero", the built-in data mover will be used.
                 type: string
+              nodeOS:
+                description: NodeOS is OS of the node where the DataDownload is processed.
+                enum:
+                - auto
+                - linux
+                - windows
+                type: string
               operationTimeout:
                 description: |-
                   OperationTimeout specifies the time used to wait internal operations,

config/crd/v2alpha1/bases/velero.io_datauploads.yaml

@@ -144,7 +144,8 @@ spec:
             description: DataUploadStatus is the current status of a DataUpload.
             properties:
               acceptedByNode:
-                description: Node is name of the node where the DataUpload is prepared.
+                description: AcceptedByNode is name of the node where the DataUpload
+                  is prepared.
                 type: string
               acceptedTimestamp:
                 description: |-
@@ -175,6 +176,13 @@ spec:
               node:
                 description: Node is name of the node where the DataUpload is processed.
                 type: string
+              nodeOS:
+                description: NodeOS is OS of the node where the DataUpload is processed.
+                enum:
+                - auto
+                - linux
+                - windows
+                type: string
               path:
                 description: Path is the full path of the snapshot volume being backed
                   up.

internal/delete/delete_item_action_handler_test.go

@@ -230,7 +230,7 @@ func (h *harness) addResource(t *testing.T, resource *test.APIResource) {
 }
 
 // recordResourcesAction is a delete item action that can be configured to run
-// for specific resources/namespaces and simply record the items that is is
+// for specific resources/namespaces and simply record the items that is
 // executed for.
 type recordResourcesAction struct {
 	selector velero.ResourceSelector

pkg/apis/velero/v2alpha1/data_download_types.go

@@ -54,6 +54,10 @@ type DataDownloadSpec struct {
 	// OperationTimeout specifies the time used to wait internal operations,
 	// before returning error as timeout.
 	OperationTimeout metav1.Duration `json:"operationTimeout"`
+
+	// NodeOS is OS of the node where the DataDownload is processed.
+	// +optional
+	NodeOS NodeOS `json:"nodeOS,omitempty"`
 }
 
 // TargetVolumeSpec is the specification for a target PVC.

pkg/apis/velero/v2alpha1/data_upload_types.go

@@ -96,6 +96,16 @@ const (
 	DataUploadPhaseFailed     DataUploadPhase = "Failed"
 )
 
+// NodeOS represents OS of a node.
+// +kubebuilder:validation:Enum=auto;linux;windows
+type NodeOS string
+
+const (
+	NodeOSLinux   NodeOS = "linux"
+	NodeOSWindows NodeOS = "windows"
+	NodeOSAuto    NodeOS = "auto"
+)
+
 // DataUploadStatus is the current status of a DataUpload.
 type DataUploadStatus struct {
 	// Phase is the current state of the DataUpload.
@@ -144,7 +154,12 @@ type DataUploadStatus struct {
 	// Node is name of the node where the DataUpload is processed.
 	// +optional
 	Node string `json:"node,omitempty"`
-	// Node is name of the node where the DataUpload is prepared.
+
+	// NodeOS is OS of the node where the DataUpload is processed.
+	// +optional
+	NodeOS NodeOS `json:"nodeOS,omitempty"`
+
+	// AcceptedByNode is name of the node where the DataUpload is prepared.
 	// +optional
 	AcceptedByNode string `json:"acceptedByNode,omitempty"`
 
@@ -221,4 +236,8 @@ type DataUploadResult struct {
 	// +optional
 	// +nullable
 	DataMoverResult *map[string]string `json:"dataMoverResult,omitempty"`
+
+	// NodeOS is OS of the node where the DataUpload is processed.
+	// +optional
+	NodeOS NodeOS `json:"nodeOS,omitempty"`
 }

pkg/backup/actions/csi/pvc_action.go

@@ -19,6 +19,7 @@ package csi
 import (
 	"context"
 	"fmt"
+	"strconv"
 
 	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
 	"github.com/pkg/errors"
@@ -484,9 +485,7 @@ func newDataUpload(
 	if backup.Spec.UploaderConfig != nil &&
 		backup.Spec.UploaderConfig.ParallelFilesUpload > 0 {
 		dataUpload.Spec.DataMoverConfig = make(map[string]string)
-		dataUpload.Spec.DataMoverConfig[uploaderUtil.ParallelFilesUpload] = fmt.Sprintf(
-			"%d", backup.Spec.UploaderConfig.ParallelFilesUpload,
-		)
+		dataUpload.Spec.DataMoverConfig[uploaderUtil.ParallelFilesUpload] = strconv.Itoa(backup.Spec.UploaderConfig.ParallelFilesUpload)
 	}
 
 	return dataUpload

pkg/backup/backup.go

@@ -35,10 +35,8 @@ import (
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
-	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
-	"k8s.io/apimachinery/pkg/selection"
 	kubeerrs "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/apimachinery/pkg/util/wait"
 	kbclient "sigs.k8s.io/controller-runtime/pkg/client"
@@ -315,7 +313,7 @@ func (kb *kubernetesBackupper) BackupWithResolvers(
 
 	var podVolumeBackupper podvolume.Backupper
 	if kb.podVolumeBackupperFactory != nil {
-		podVolumeBackupper, err = kb.podVolumeBackupperFactory.NewBackupper(ctx, backupRequest.Backup, kb.uploaderType)
+		podVolumeBackupper, err = kb.podVolumeBackupperFactory.NewBackupper(ctx, log, backupRequest.Backup, kb.uploaderType)
 		if err != nil {
 			log.WithError(errors.WithStack(err)).Debugf("Error from NewBackupper")
 			return errors.WithStack(err)
@@ -745,6 +743,7 @@ func (kb *kubernetesBackupper) handleItemBlockPostHooks(ctx context.Context, ite
 	log := itemBlock.Log
 	defer itemBlock.itemBackupper.hookTracker.AsyncItemBlocks.Done()
 
+	// the post hooks will not execute until all PVBs of the item block pods are processed
 	if err := kb.waitUntilPVBsProcessed(ctx, log, itemBlock, hookPods); err != nil {
 		log.WithError(err).Error("failed to wait PVBs processed for the ItemBlock")
 		return
@@ -758,36 +757,19 @@ func (kb *kubernetesBackupper) handleItemBlockPostHooks(ctx context.Context, ite
 	}
 }
 
+// wait all PVBs of the item block pods to be processed
 func (kb *kubernetesBackupper) waitUntilPVBsProcessed(ctx context.Context, log logrus.FieldLogger, itemBlock BackupItemBlock, pods []itemblock.ItemBlockItem) error {
-	requirement, err := labels.NewRequirement(velerov1api.BackupUIDLabel, selection.Equals, []string{string(itemBlock.itemBackupper.backupRequest.UID)})
-	if err != nil {
-		return errors.Wrapf(err, "failed to create label requirement")
-	}
-	options := &kbclient.ListOptions{
-		LabelSelector: labels.NewSelector().Add(*requirement),
-	}
-	pvbList := &velerov1api.PodVolumeBackupList{}
-	if err := kb.kbClient.List(context.Background(), pvbList, options); err != nil {
-		return errors.Wrap(err, "failed to list PVBs")
-	}
-
-	podMap := map[string]struct{}{}
-	for _, pod := range pods {
-		podMap[string(pod.Item.GetUID())] = struct{}{}
-	}
-
 	pvbMap := map[*velerov1api.PodVolumeBackup]bool{}
-	for i, pvb := range pvbList.Items {
-		if _, exist := podMap[string(pvb.Spec.Pod.UID)]; !exist {
-			continue
+	for _, pod := range pods {
+		namespace, name := pod.Item.GetNamespace(), pod.Item.GetName()
+		pvbs, err := itemBlock.itemBackupper.podVolumeBackupper.ListPodVolumeBackupsByPod(namespace, name)
+		if err != nil {
+			return errors.Wrapf(err, "failed to list PodVolumeBackups for pod %s/%s", namespace, name)
 		}
-
-		processed := false
-		if pvb.Status.Phase == velerov1api.PodVolumeBackupPhaseCompleted ||
-			pvb.Status.Phase == velerov1api.PodVolumeBackupPhaseFailed {
-			processed = true
+		for _, pvb := range pvbs {
+			pvbMap[pvb] = pvb.Status.Phase == velerov1api.PodVolumeBackupPhaseCompleted ||
+				pvb.Status.Phase == velerov1api.PodVolumeBackupPhaseFailed
 		}
-		pvbMap[&pvbList.Items[i]] = processed
 	}
 
 	checkFunc := func(context.Context) (done bool, err error) {
@@ -796,8 +778,8 @@ func (kb *kubernetesBackupper) waitUntilPVBsProcessed(ctx context.Context, log l
 			if processed {
 				continue
 			}
-			updatedPVB := &velerov1api.PodVolumeBackup{}
-			if err := kb.kbClient.Get(ctx, kbclient.ObjectKeyFromObject(pvb), updatedPVB); err != nil {
+			updatedPVB, err := itemBlock.itemBackupper.podVolumeBackupper.GetPodVolumeBackup(pvb.Namespace, pvb.Name)
+			if err != nil {
 				allProcessed = false
 				log.Infof("failed to get PVB: %v", err)
 				continue

pkg/backup/backup_test.go

@@ -3945,7 +3945,7 @@ func TestBackupWithHooks(t *testing.T) {
 
 type fakePodVolumeBackupperFactory struct{}
 
-func (f *fakePodVolumeBackupperFactory) NewBackupper(context.Context, *velerov1.Backup, string) (podvolume.Backupper, error) {
+func (f *fakePodVolumeBackupperFactory) NewBackupper(context.Context, logrus.FieldLogger, *velerov1.Backup, string) (podvolume.Backupper, error) {
 	return &fakePodVolumeBackupper{}, nil
 }
 
@@ -3978,6 +3978,24 @@ func (b *fakePodVolumeBackupper) WaitAllPodVolumesProcessed(log logrus.FieldLogg
 	return b.pvbs
 }
 
+func (b *fakePodVolumeBackupper) GetPodVolumeBackup(namespace, name string) (*velerov1.PodVolumeBackup, error) {
+	for _, pvb := range b.pvbs {
+		if pvb.Namespace == namespace && pvb.Name == name {
+			return pvb, nil
+		}
+	}
+	return nil, nil
+}
+func (b *fakePodVolumeBackupper) ListPodVolumeBackupsByPod(podNamespace, podName string) ([]*velerov1.PodVolumeBackup, error) {
+	var pvbs []*velerov1.PodVolumeBackup
+	for _, pvb := range b.pvbs {
+		if pvb.Spec.Pod.Namespace == podNamespace && pvb.Spec.Pod.Name == podName {
+			pvbs = append(pvbs, pvb)
+		}
+	}
+	return pvbs, nil
+}
+
 // TestBackupWithPodVolume runs backups of pods that are annotated for PodVolume backup,
 // and ensures that the pod volume backupper is called, that the returned PodVolumeBackups
 // are added to the Request object, and that when PVCs are backed up with PodVolume, the

pkg/builder/data_download_builder.go

@@ -148,6 +148,12 @@ func (d *DataDownloadBuilder) Node(node string) *DataDownloadBuilder {
 	return d
 }
 
+// NodeOS sets the DataDownload's Node OS.
+func (d *DataDownloadBuilder) NodeOS(nodeOS velerov2alpha1api.NodeOS) *DataDownloadBuilder {
+	d.object.Spec.NodeOS = nodeOS
+	return d
+}
+
 // AcceptedByNode sets the DataDownload's AcceptedByNode.
 func (d *DataDownloadBuilder) AcceptedByNode(node string) *DataDownloadBuilder {
 	d.object.Status.AcceptedByNode = node

pkg/builder/data_upload_builder.go

@@ -151,6 +151,12 @@ func (d *DataUploadBuilder) Node(node string) *DataUploadBuilder {
 	return d
 }
 
+// NodeOS sets the DataUpload's Node OS.
+func (d *DataUploadBuilder) NodeOS(nodeOS velerov2alpha1api.NodeOS) *DataUploadBuilder {
+	d.object.Status.NodeOS = nodeOS
+	return d
+}
+
 // AcceptedByNode sets the DataUpload's AcceptedByNode.
 func (d *DataUploadBuilder) AcceptedByNode(node string) *DataUploadBuilder {
 	d.object.Status.AcceptedByNode = node

pkg/cmd/cli/backup/create_test.go

@@ -19,6 +19,7 @@ package backup
 import (
 	"context"
 	"fmt"
+	"strconv"
 	"strings"
 	"testing"
 	"time"
@@ -222,7 +223,7 @@ func TestCreateCommand(t *testing.T) {
 		flags.Parse([]string{"--default-volumes-to-fs-backup", defaultVolumesToFsBackup})
 		flags.Parse([]string{"--resource-policies-configmap", resPoliciesConfigmap})
 		flags.Parse([]string{"--data-mover", dataMover})
-		flags.Parse([]string{"--parallel-files-upload", fmt.Sprintf("%d", parallelFilesUpload)})
+		flags.Parse([]string{"--parallel-files-upload", strconv.Itoa(parallelFilesUpload)})
 		//flags.Parse([]string{"--wait"})
 
 		client := velerotest.NewFakeControllerRuntimeClient(t).(kbclient.WithWatch)

pkg/cmd/cli/datamover/restore.go

@@ -160,7 +160,24 @@ func newdataMoverRestore(logger logrus.FieldLogger, factory client.Factory, conf
 		return nil, errors.Wrap(err, "error to create client")
 	}
 
-	cache, err := ctlcache.New(clientConfig, cacheOption)
+	var cache ctlcache.Cache
+	retry := 10
+	for {
+		cache, err = ctlcache.New(clientConfig, cacheOption)
+		if err == nil {
+			break
+		}
+
+		retry--
+		if retry == 0 {
+			break
+		}
+
+		logger.WithError(err).Warn("Failed to create client cache, need retry")
+
+		time.Sleep(time.Second)
+	}
+
 	if err != nil {
 		cancelFunc()
 		return nil, errors.Wrap(err, "error to create client cache")

pkg/cmd/cli/nodeagent/server.go

@@ -353,7 +353,13 @@ func (s *nodeAgentServer) run() {
 		s.logger.WithError(err).Fatal("Unable to create the data upload controller")
 	}
 
-	dataDownloadReconciler := controller.NewDataDownloadReconciler(s.mgr.GetClient(), s.mgr, s.kubeClient, s.dataPathMgr, podResources, s.nodeName, s.config.dataMoverPrepareTimeout, s.logger, s.metrics)
+	var restorePVCConfig nodeagent.RestorePVC
+	if s.dataPathConfigs != nil && s.dataPathConfigs.RestorePVCConfig != nil {
+		restorePVCConfig = *s.dataPathConfigs.RestorePVCConfig
+		s.logger.Infof("Using customized restorePVC config %v", restorePVCConfig)
+	}
+
+	dataDownloadReconciler := controller.NewDataDownloadReconciler(s.mgr.GetClient(), s.mgr, s.kubeClient, s.dataPathMgr, restorePVCConfig, podResources, s.nodeName, s.config.dataMoverPrepareTimeout, s.logger, s.metrics)
 	if err = dataDownloadReconciler.SetupWithManager(s.mgr); err != nil {
 		s.logger.WithError(err).Fatal("Unable to create the data download controller")
 	}

pkg/cmd/server/plugin/plugin.go

@@ -21,6 +21,7 @@ import (
 	"github.com/spf13/cobra"
 	apiextensions "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
 
+	"github.com/vmware-tanzu/velero/pkg/constant"
 	"github.com/vmware-tanzu/velero/pkg/datamover"
 
 	dia "github.com/vmware-tanzu/velero/internal/delete/actions/csi"
@@ -162,7 +163,7 @@ func NewCommand(f client.Factory) *cobra.Command {
 					newVolumeSnapshotClassBackupItemAction,
 				).
 				RegisterRestoreItemActionV2(
-					"velero.io/csi-pvc-restorer",
+					constant.PluginCSIPVCRestoreRIA,
 					newPvcRestoreItemAction(f),
 				).
 				RegisterRestoreItemActionV2(

pkg/constant/constant.go

@@ -19,4 +19,6 @@ const (
 	ControllerSchedule              = "schedule"
 	ControllerServerStatusRequest   = "server-status-request"
 	ControllerRestoreFinalizer      = "restore-finalizer"
+
+	PluginCSIPVCRestoreRIA = "velero.io/csi-pvc-restorer"
 )

pkg/controller/data_download_controller.go

@@ -64,13 +64,15 @@ type DataDownloadReconciler struct {
 	restoreExposer   exposer.GenericRestoreExposer
 	nodeName         string
 	dataPathMgr      *datapath.Manager
+	restorePVCConfig nodeagent.RestorePVC
 	podResources     v1.ResourceRequirements
 	preparingTimeout time.Duration
 	metrics          *metrics.ServerMetrics
 }
 
 func NewDataDownloadReconciler(client client.Client, mgr manager.Manager, kubeClient kubernetes.Interface, dataPathMgr *datapath.Manager,
-	podResources v1.ResourceRequirements, nodeName string, preparingTimeout time.Duration, logger logrus.FieldLogger, metrics *metrics.ServerMetrics) *DataDownloadReconciler {
+	restorePVCConfig nodeagent.RestorePVC, podResources v1.ResourceRequirements, nodeName string, preparingTimeout time.Duration,
+	logger logrus.FieldLogger, metrics *metrics.ServerMetrics) *DataDownloadReconciler {
 	return &DataDownloadReconciler{
 		client:           client,
 		kubeClient:       kubeClient,
@@ -79,6 +81,7 @@ func NewDataDownloadReconciler(client client.Client, mgr manager.Manager, kubeCl
 		Clock:            &clock.RealClock{},
 		nodeName:         nodeName,
 		restoreExposer:   exposer.NewGenericRestoreExposer(kubeClient, logger),
+		restorePVCConfig: restorePVCConfig,
 		dataPathMgr:      dataPathMgr,
 		podResources:     podResources,
 		preparingTimeout: preparingTimeout,
@@ -180,21 +183,15 @@ func (r *DataDownloadReconciler) Reconcile(ctx context.Context, req ctrl.Request
 			return ctrl.Result{}, nil
 		}
 
-		hostingPodLabels := map[string]string{velerov1api.DataDownloadLabel: dd.Name}
-		for _, k := range util.ThirdPartyLabels {
-			if v, err := nodeagent.GetLabelValue(ctx, r.kubeClient, dd.Namespace, k, kube.NodeOSLinux); err != nil {
-				if err != nodeagent.ErrNodeAgentLabelNotFound {
-					log.WithError(err).Warnf("Failed to check node-agent label, skip adding host pod label %s", k)
-				}
-			} else {
-				hostingPodLabels[k] = v
-			}
+		exposeParam, err := r.setupExposeParam(dd)
+		if err != nil {
+			return r.errorOut(ctx, dd, err, "failed to set exposer parameters", log)
 		}
 
 		// Expose() will trigger to create one pod whose volume is restored by a given volume snapshot,
 		// but the pod maybe is not in the same node of the current controller, so we need to return it here.
 		// And then only the controller who is in the same node could do the rest work.
-		err = r.restoreExposer.Expose(ctx, getDataDownloadOwnerObject(dd), dd.Spec.TargetVolume.PVC, dd.Spec.TargetVolume.Namespace, hostingPodLabels, r.podResources, dd.Spec.OperationTimeout.Duration)
+		err = r.restoreExposer.Expose(ctx, getDataDownloadOwnerObject(dd), exposeParam)
 		if err != nil {
 			if err := r.client.Get(ctx, req.NamespacedName, dd); err != nil {
 				if !apierrors.IsNotFound(err) {
@@ -233,7 +230,7 @@ func (r *DataDownloadReconciler) Reconcile(ctx context.Context, req ctrl.Request
 			log.Debugf("Data download is been canceled %s in Phase %s", dd.GetName(), dd.Status.Phase)
 			r.tryCancelAcceptedDataDownload(ctx, dd, "")
 		} else if peekErr := r.restoreExposer.PeekExposed(ctx, getDataDownloadOwnerObject(dd)); peekErr != nil {
-			r.tryCancelAcceptedDataDownload(ctx, dd, fmt.Sprintf("found a dataupload %s/%s with expose error: %s. mark it as cancel", dd.Namespace, dd.Name, peekErr))
+			r.tryCancelAcceptedDataDownload(ctx, dd, fmt.Sprintf("found a datadownload %s/%s with expose error: %s. mark it as cancel", dd.Namespace, dd.Name, peekErr))
 			log.Errorf("Cancel dd %s/%s because of expose error %s", dd.Namespace, dd.Name, peekErr)
 		} else if dd.Status.AcceptedTimestamp != nil {
 			if time.Since(dd.Status.AcceptedTimestamp.Time) >= r.preparingTimeout {
@@ -727,6 +724,42 @@ func (r *DataDownloadReconciler) closeDataPath(ctx context.Context, ddName strin
 	r.dataPathMgr.RemoveAsyncBR(ddName)
 }
 
+func (r *DataDownloadReconciler) setupExposeParam(dd *velerov2alpha1api.DataDownload) (exposer.GenericRestoreExposeParam, error) {
+	log := r.logger.WithField("datadownload", dd.Name)
+
+	nodeOS := string(dd.Spec.NodeOS)
+	if nodeOS == "" {
+		log.Info("nodeOS is empty in DD, fallback to linux")
+		nodeOS = kube.NodeOSLinux
+	}
+
+	if err := kube.HasNodeWithOS(context.Background(), nodeOS, r.kubeClient.CoreV1()); err != nil {
+		return exposer.GenericRestoreExposeParam{}, errors.Wrapf(err, "no appropriate node to run datadownload %s/%s", dd.Namespace, dd.Name)
+	}
+
+	hostingPodLabels := map[string]string{velerov1api.DataDownloadLabel: dd.Name}
+	for _, k := range util.ThirdPartyLabels {
+		if v, err := nodeagent.GetLabelValue(context.Background(), r.kubeClient, dd.Namespace, k, nodeOS); err != nil {
+			if err != nodeagent.ErrNodeAgentLabelNotFound {
+				log.WithError(err).Warnf("Failed to check node-agent label, skip adding host pod label %s", k)
+			}
+		} else {
+			hostingPodLabels[k] = v
+		}
+	}
+
+	return exposer.GenericRestoreExposeParam{
+		TargetPVCName:    dd.Spec.TargetVolume.PVC,
+		TargetNamespace:  dd.Spec.TargetVolume.Namespace,
+		HostingPodLabels: hostingPodLabels,
+		Resources:        r.podResources,
+		OperationTimeout: dd.Spec.OperationTimeout.Duration,
+		ExposeTimeout:    r.preparingTimeout,
+		NodeOS:           nodeOS,
+		RestorePVCConfig: r.restorePVCConfig,
+	}, nil
+}
+
 func getDataDownloadOwnerObject(dd *velerov2alpha1api.DataDownload) v1.ObjectReference {
 	return v1.ObjectReference{
 		Kind:       dd.Kind,

pkg/controller/data_download_controller_test.go

@@ -50,8 +50,10 @@ import (
 	datapathmockes "github.com/vmware-tanzu/velero/pkg/datapath/mocks"
 	"github.com/vmware-tanzu/velero/pkg/exposer"
 	"github.com/vmware-tanzu/velero/pkg/metrics"
+	"github.com/vmware-tanzu/velero/pkg/nodeagent"
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
 	"github.com/vmware-tanzu/velero/pkg/uploader"
+	"github.com/vmware-tanzu/velero/pkg/util/kube"
 
 	exposermockes "github.com/vmware-tanzu/velero/pkg/exposer/mocks"
 )
@@ -66,7 +68,7 @@ func dataDownloadBuilder() *builder.DataDownloadBuilder {
 		PV:        "test-pv",
 		PVC:       "test-pvc",
 		Namespace: "test-ns",
-	})
+	}).NodeOS(velerov2alpha1api.NodeOS("linux"))
 }
 
 func initDataDownloadReconciler(objects []runtime.Object, needError ...bool) (*DataDownloadReconciler, error) {
@@ -140,7 +142,7 @@ func initDataDownloadReconcilerWithError(objects []runtime.Object, needError ...
 
 	dataPathMgr := datapath.NewManager(1)
 
-	return NewDataDownloadReconciler(fakeClient, nil, fakeKubeClient, dataPathMgr, corev1.ResourceRequirements{}, "test-node", time.Minute*5, velerotest.NewLogger(), metrics.NewServerMetrics()), nil
+	return NewDataDownloadReconciler(fakeClient, nil, fakeKubeClient, dataPathMgr, nodeagent.RestorePVC{}, corev1.ResourceRequirements{}, "test-node", time.Minute*5, velerotest.NewLogger(), metrics.NewServerMetrics()), nil
 }
 
 func TestDataDownloadReconcile(t *testing.T) {
@@ -166,6 +168,8 @@ func TestDataDownloadReconcile(t *testing.T) {
 		},
 	}
 
+	node := builder.ForNode("fake-node").Labels(map[string]string{kube.NodeOSLabel: kube.NodeOSLinux}).Result()
+
 	tests := []struct {
 		name              string
 		dd                *velerov2alpha1api.DataDownload
@@ -325,9 +329,15 @@ func TestDataDownloadReconcile(t *testing.T) {
 		},
 		{
 			name:      "Restore is exposed",
-			dd:        dataDownloadBuilder().Result(),
+			dd:        dataDownloadBuilder().NodeOS(velerov2alpha1api.NodeOSLinux).Result(),
 			targetPVC: builder.ForPersistentVolumeClaim("test-ns", "test-pvc").Result(),
 		},
+		{
+			name:              "Expected node doesn't exist",
+			dd:                dataDownloadBuilder().NodeOS(velerov2alpha1api.NodeOSWindows).Result(),
+			targetPVC:         builder.ForPersistentVolumeClaim("test-ns", "test-pvc").Result(),
+			expectedStatusMsg: "no appropriate node to run datadownload",
+		},
 		{
 			name:      "Get empty restore exposer",
 			dd:        dataDownloadBuilder().Phase(velerov2alpha1api.DataDownloadPhasePrepared).Result(),
@@ -387,9 +397,9 @@ func TestDataDownloadReconcile(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			var objs []runtime.Object
+			objs := []runtime.Object{daemonSet, node}
 			if test.targetPVC != nil {
-				objs = []runtime.Object{test.targetPVC, daemonSet}
+				objs = append(objs, test.targetPVC)
 			}
 			r, err := initDataDownloadReconciler(objs, test.needErrs...)
 			require.NoError(t, err)
@@ -959,7 +969,7 @@ func (dt *ddResumeTestHelper) resumeCancellableDataPath(_ *DataUploadReconciler,
 	return dt.resumeErr
 }
 
-func (dt *ddResumeTestHelper) Expose(context.Context, corev1.ObjectReference, string, string, map[string]string, corev1.ResourceRequirements, time.Duration) error {
+func (dt *ddResumeTestHelper) Expose(context.Context, corev1.ObjectReference, exposer.GenericRestoreExposeParam) error {
 	return nil
 }
 

pkg/controller/data_upload_controller.go

@@ -285,6 +285,10 @@ func (r *DataUploadReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 			return ctrl.Result{}, nil
 		}
 
+		if res.ByPod.NodeOS == nil {
+			return r.errorOut(ctx, du, errors.New("unsupported ambiguous node OS"), "invalid expose result", log)
+		}
+
 		log.Info("Exposed snapshot is ready and creating data path routine")
 
 		// Need to first create file system BR and get data path instance then update data upload status
@@ -317,6 +321,7 @@ func (r *DataUploadReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 		original := du.DeepCopy()
 		du.Status.Phase = velerov2alpha1api.DataUploadPhaseInProgress
 		du.Status.StartTimestamp = &metav1.Time{Time: r.Clock.Now()}
+		du.Status.NodeOS = velerov2alpha1api.NodeOS(*res.ByPod.NodeOS)
 		if err := r.client.Patch(ctx, du, client.MergeFrom(original)); err != nil {
 			log.WithError(err).Warnf("Failed to update dataupload %s to InProgress, will data path close and retry", du.Name)
 
@@ -792,6 +797,8 @@ func (r *DataUploadReconciler) closeDataPath(ctx context.Context, duName string)
 }
 
 func (r *DataUploadReconciler) setupExposeParam(du *velerov2alpha1api.DataUpload) (interface{}, error) {
+	log := r.logger.WithField("dataupload", du.Name)
+
 	if du.Spec.SnapshotType == velerov2alpha1api.SnapshotTypeCSI {
 		pvc := &corev1.PersistentVolumeClaim{}
 		err := r.client.Get(context.Background(), types.NamespacedName{
@@ -803,7 +810,7 @@ func (r *DataUploadReconciler) setupExposeParam(du *velerov2alpha1api.DataUpload
 			return nil, errors.Wrapf(err, "failed to get PVC %s/%s", du.Spec.SourceNamespace, du.Spec.SourcePVC)
 		}
 
-		nodeOS, err := kube.GetPVCAttachingNodeOS(pvc, r.kubeClient.CoreV1(), r.kubeClient.StorageV1(), r.logger)
+		nodeOS, err := kube.GetPVCAttachingNodeOS(pvc, r.kubeClient.CoreV1(), r.kubeClient.StorageV1(), log)
 		if err != nil {
 			return nil, errors.Wrapf(err, "failed to get attaching node OS for PVC %s/%s", du.Spec.SourceNamespace, du.Spec.SourcePVC)
 		}
@@ -821,7 +828,7 @@ func (r *DataUploadReconciler) setupExposeParam(du *velerov2alpha1api.DataUpload
 		for _, k := range util.ThirdPartyLabels {
 			if v, err := nodeagent.GetLabelValue(context.Background(), r.kubeClient, du.Namespace, k, nodeOS); err != nil {
 				if err != nodeagent.ErrNodeAgentLabelNotFound {
-					r.logger.WithError(err).Warnf("Failed to check node-agent label, skip adding host pod label %s", k)
+					log.WithError(err).Warnf("Failed to check node-agent label, skip adding host pod label %s", k)
 				}
 			} else {
 				hostingPodLabels[k] = v
@@ -843,6 +850,7 @@ func (r *DataUploadReconciler) setupExposeParam(du *velerov2alpha1api.DataUpload
 			NodeOS:           nodeOS,
 		}, nil
 	}
+
 	return nil, nil
 }
 

pkg/controller/data_upload_controller_test.go

@@ -166,6 +166,7 @@ func initDataUploaderReconcilerWithError(needError ...error) (*DataUploadReconci
 			RestoreSize: &restoreSize,
 		},
 	}
+
 	daemonSet := &appsv1.DaemonSet{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: "velero",
@@ -265,9 +266,10 @@ func dataUploadBuilder() *builder.DataUploadBuilder {
 }
 
 type fakeSnapshotExposer struct {
-	kubeClient kbclient.Client
-	clock      clock.WithTickerAndDelayedExecution
-	peekErr    error
+	kubeClient      kbclient.Client
+	clock           clock.WithTickerAndDelayedExecution
+	ambiguousNodeOS bool
+	peekErr         error
 }
 
 func (f *fakeSnapshotExposer) Expose(ctx context.Context, ownerObject corev1.ObjectReference, param interface{}) error {
@@ -296,7 +298,13 @@ func (f *fakeSnapshotExposer) GetExposed(ctx context.Context, du corev1.ObjectRe
 	if err != nil {
 		return nil, err
 	}
-	return &exposer.ExposeResult{ByPod: exposer.ExposeByPod{HostingPod: pod, VolumeName: dataUploadName}}, nil
+
+	nodeOS := "linux"
+	pNodeOS := &nodeOS
+	if f.ambiguousNodeOS {
+		pNodeOS = nil
+	}
+	return &exposer.ExposeResult{ByPod: exposer.ExposeByPod{HostingPod: pod, VolumeName: dataUploadName, NodeOS: pNodeOS}}, nil
 }
 
 func (f *fakeSnapshotExposer) PeekExposed(ctx context.Context, ownerObject corev1.ObjectReference) error {
@@ -350,6 +358,8 @@ func TestReconcile(t *testing.T) {
 		expectedRequeue     ctrl.Result
 		expectedErrMsg      string
 		needErrs            []bool
+		removeNode          bool
+		ambiguousNodeOS     bool
 		peekErr             error
 		notCreateFSBR       bool
 		fsBRInitErr         error
@@ -359,25 +369,29 @@ func TestReconcile(t *testing.T) {
 			name:            "Dataupload is not initialized",
 			du:              builder.ForDataUpload("unknown-ns", "unknown-name").Result(),
 			expectedRequeue: ctrl.Result{},
-		}, {
+		},
+		{
 			name:            "Error get Dataupload",
 			du:              builder.ForDataUpload(velerov1api.DefaultNamespace, "unknown-name").Result(),
 			expectedRequeue: ctrl.Result{},
 			expectedErrMsg:  "getting DataUpload: Get error",
 			needErrs:        []bool{true, false, false, false},
-		}, {
+		},
+		{
 			name:            "Unsupported data mover type",
 			du:              dataUploadBuilder().DataMover("unknown type").Result(),
 			expected:        dataUploadBuilder().Phase("").Result(),
 			expectedRequeue: ctrl.Result{},
-		}, {
+		},
+		{
 			name:              "Unknown type of snapshot exposer is not initialized",
 			du:                dataUploadBuilder().SnapshotType("unknown type").Result(),
 			expectedProcessed: true,
 			expected:          dataUploadBuilder().Phase(velerov2alpha1api.DataUploadPhaseFailed).Result(),
 			expectedRequeue:   ctrl.Result{},
 			expectedErrMsg:    "unknown type type of snapshot exposer is not exist",
-		}, {
+		},
+		{
 			name:            "Dataupload should be accepted",
 			du:              dataUploadBuilder().Result(),
 			pod:             builder.ForPod("fake-ns", dataUploadName).Volumes(&corev1.Volume{Name: "test-pvc"}).Result(),
@@ -394,6 +408,27 @@ func TestReconcile(t *testing.T) {
 			expectedRequeue:   ctrl.Result{},
 			expectedErrMsg:    "failed to get PVC",
 		},
+		{
+			name:              "Dataupload should fail to get PVC attaching node",
+			du:                dataUploadBuilder().Result(),
+			pod:               builder.ForPod("fake-ns", dataUploadName).Volumes(&corev1.Volume{Name: "test-pvc"}).Result(),
+			pvc:               builder.ForPersistentVolumeClaim("fake-ns", "test-pvc").StorageClass("fake-sc").Result(),
+			expectedProcessed: true,
+			expected:          dataUploadBuilder().Phase(velerov2alpha1api.DataUploadPhaseFailed).Result(),
+			expectedRequeue:   ctrl.Result{},
+			expectedErrMsg:    "error to get storage class",
+		},
+		{
+			name:              "Dataupload should fail because expected node doesn't exist",
+			du:                dataUploadBuilder().Result(),
+			pod:               builder.ForPod("fake-ns", dataUploadName).Volumes(&corev1.Volume{Name: "test-pvc"}).Result(),
+			pvc:               builder.ForPersistentVolumeClaim("fake-ns", "test-pvc").Result(),
+			removeNode:        true,
+			expectedProcessed: true,
+			expected:          dataUploadBuilder().Phase(velerov2alpha1api.DataUploadPhaseFailed).Result(),
+			expectedRequeue:   ctrl.Result{},
+			expectedErrMsg:    "no appropriate node to run data upload",
+		},
 		{
 			name:            "Dataupload should be prepared",
 			du:              dataUploadBuilder().SnapshotType(fakeSnapshotType).Result(),
@@ -407,6 +442,15 @@ func TestReconcile(t *testing.T) {
 			expected:        dataUploadBuilder().Phase(velerov2alpha1api.DataUploadPhaseInProgress).Result(),
 			expectedRequeue: ctrl.Result{},
 		},
+		{
+			name:              "Dataupload should fail if expose returns ambiguous nodeOS",
+			pod:               builder.ForPod(velerov1api.DefaultNamespace, dataUploadName).Volumes(&corev1.Volume{Name: "dataupload-1"}).Result(),
+			du:                dataUploadBuilder().Phase(velerov2alpha1api.DataUploadPhasePrepared).SnapshotType(fakeSnapshotType).Result(),
+			ambiguousNodeOS:   true,
+			expectedProcessed: true,
+			expected:          dataUploadBuilder().Phase(velerov2alpha1api.DataUploadPhaseFailed).Result(),
+			expectedErrMsg:    "unsupported ambiguous node OS",
+		},
 		{
 			name:            "Dataupload with not enabled cancel",
 			pod:             builder.ForPod(velerov1api.DefaultNamespace, dataUploadName).Volumes(&corev1.Volume{Name: "dataupload-1"}).Result(),
@@ -557,6 +601,11 @@ func TestReconcile(t *testing.T) {
 				require.NoError(t, err)
 			}
 
+			if test.removeNode {
+				err = r.kubeClient.CoreV1().Nodes().Delete(ctx, "fake-node", metav1.DeleteOptions{})
+				require.NoError(t, err)
+			}
+
 			if test.dataMgr != nil {
 				r.dataPathMgr = test.dataMgr
 			} else {
@@ -564,7 +613,7 @@ func TestReconcile(t *testing.T) {
 			}
 
 			if test.du.Spec.SnapshotType == fakeSnapshotType {
-				r.snapshotExposerList = map[velerov2alpha1api.SnapshotType]exposer.SnapshotExposer{fakeSnapshotType: &fakeSnapshotExposer{r.client, r.Clock, test.peekErr}}
+				r.snapshotExposerList = map[velerov2alpha1api.SnapshotType]exposer.SnapshotExposer{fakeSnapshotType: &fakeSnapshotExposer{r.client, r.Clock, test.ambiguousNodeOS, test.peekErr}}
 			} else if test.du.Spec.SnapshotType == velerov2alpha1api.SnapshotTypeCSI {
 				r.snapshotExposerList = map[velerov2alpha1api.SnapshotType]exposer.SnapshotExposer{velerov2alpha1api.SnapshotTypeCSI: exposer.NewCSISnapshotExposer(r.kubeClient, r.csiSnapshotClient, velerotest.NewLogger())}
 			}

pkg/controller/restore_finalizer_controller.go

@@ -22,6 +22,12 @@ import (
 	"sync"
 	"time"
 
+	"k8s.io/apimachinery/pkg/runtime/schema"
+
+	"github.com/vmware-tanzu/velero/pkg/constant"
+	"github.com/vmware-tanzu/velero/pkg/itemoperation"
+	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+
 	storagev1api "k8s.io/api/storage/v1"
 
 	"github.com/pkg/errors"
@@ -155,6 +161,12 @@ func (r *restoreFinalizerReconciler) Reconcile(ctx context.Context, req ctrl.Req
 
 	restoredPVCList := volume.RestoredPVCFromRestoredResourceList(restoredResourceList)
 
+	restoreItemOperations, err := backupStore.GetRestoreItemOperations(restore.Name)
+	if err != nil {
+		log.WithError(err).Error("error getting itemOperationList")
+		return ctrl.Result{}, errors.Wrap(err, "error getting itemOperationList")
+	}
+
 	finalizerCtx := &finalizerContext{
 		logger:           log,
 		restore:          restore,
@@ -163,6 +175,9 @@ func (r *restoreFinalizerReconciler) Reconcile(ctx context.Context, req ctrl.Req
 		restoredPVCList:  restoredPVCList,
 		multiHookTracker: r.multiHookTracker,
 		resourceTimeout:  r.resourceTimeout,
+		restoreItemOperationList: restoreItemOperationList{
+			items: restoreItemOperations,
+		},
 	}
 	warnings, errs := finalizerCtx.execute()
 
@@ -239,16 +254,44 @@ func (r *restoreFinalizerReconciler) finishProcessing(restorePhase velerov1api.R
 	return kubeutil.PatchResourceWithRetriesOnErrors(r.resourceTimeout, original, restore, r.Client)
 }
 
+type restoreItemOperationList struct {
+	items []*itemoperation.RestoreOperation
+}
+
+func (r *restoreItemOperationList) selectByResource(group, resource, ns, name string) []*itemoperation.RestoreOperation {
+	var res []*itemoperation.RestoreOperation
+	rid := velero.ResourceIdentifier{
+		GroupResource: schema.GroupResource{
+			Group:    group,
+			Resource: resource,
+		},
+		Namespace: ns,
+		Name:      name,
+	}
+	for _, item := range r.items {
+		if item != nil && item.Spec.ResourceIdentifier == rid {
+			res = append(res, item)
+		}
+	}
+	return res
+}
+
+// SelectByPVC filters the restore item operation list by PVC namespace and name.
+func (r *restoreItemOperationList) SelectByPVC(ns, name string) []*itemoperation.RestoreOperation {
+	return r.selectByResource("", "persistentvolumeclaims", ns, name)
+}
+
 // finalizerContext includes all the dependencies required by finalization tasks and
 // a function execute() to orderly implement task logic.
 type finalizerContext struct {
-	logger           logrus.FieldLogger
-	restore          *velerov1api.Restore
-	crClient         client.Client
-	volumeInfo       []*volume.BackupVolumeInfo
-	restoredPVCList  map[string]struct{}
-	multiHookTracker *hook.MultiHookTracker
-	resourceTimeout  time.Duration
+	logger                   logrus.FieldLogger
+	restore                  *velerov1api.Restore
+	crClient                 client.Client
+	volumeInfo               []*volume.BackupVolumeInfo
+	restoredPVCList          map[string]struct{}
+	restoreItemOperationList restoreItemOperationList
+	multiHookTracker         *hook.MultiHookTracker
+	resourceTimeout          time.Duration
 }
 
 func (ctx *finalizerContext) execute() (results.Result, results.Result) { //nolint:unparam //temporarily ignore the lint report: result 0 is always nil (unparam)
@@ -310,6 +353,17 @@ func (ctx *finalizerContext) patchDynamicPVWithVolumeInfo() (errs results.Result
 						return false, err
 					}
 
+					// Check whether the async operation to populate the PVC is successful.  If it's not, will skip patching the PV, instead of waiting.
+					operations := ctx.restoreItemOperationList.SelectByPVC(pvc.Namespace, pvc.Name)
+					for _, op := range operations {
+						if op.Spec.RestoreItemAction == constant.PluginCSIPVCRestoreRIA &&
+							op.Status.Phase != itemoperation.OperationPhaseCompleted {
+							log.Warnf("skipping PV patch, because the operation to restore the PVC is not completed, "+
+								"operation: %s, phase: %s", op.Spec.OperationID, op.Status.Phase)
+							return true, nil
+						}
+					}
+
 					// We are handling a common but specific scenario where a PVC is in a pending state and uses a storage class with
 					// VolumeBindingMode set to WaitForFirstConsumer. In this case, the PV patch step is skipped to avoid
 					// failures due to the PVC not being bound, which could cause a timeout and result in a failed restore.

pkg/controller/restore_finalizer_controller_test.go

@@ -23,6 +23,11 @@ import (
 	"testing"
 	"time"
 
+	"k8s.io/apimachinery/pkg/runtime/schema"
+
+	"github.com/vmware-tanzu/velero/pkg/itemoperation"
+	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
@@ -145,6 +150,7 @@ func TestRestoreFinalizerReconcile(t *testing.T) {
 			if test.restore != nil && test.restore.Namespace == velerov1api.DefaultNamespace {
 				require.NoError(t, r.Client.Create(context.Background(), test.restore))
 				backupStore.On("GetRestoredResourceList", test.restore.Name).Return(map[string][]string{}, nil)
+				backupStore.On("GetRestoreItemOperations", test.restore.Name).Return([]*itemoperation.RestoreOperation{}, nil)
 			}
 			if test.backup != nil {
 				assert.NoError(t, r.Client.Create(context.Background(), test.backup))
@@ -627,3 +633,112 @@ func Test_restoreFinalizerReconciler_finishProcessing(t *testing.T) {
 		})
 	}
 }
+
+func TestRestoreOperationList(t *testing.T) {
+	var empty []*itemoperation.RestoreOperation
+	tests := []struct {
+		name         string
+		items        []*itemoperation.RestoreOperation
+		inputPVCNS   string
+		inputPVCName string
+		expected     []*itemoperation.RestoreOperation
+	}{
+		{
+			name:         "no restore operations",
+			items:        []*itemoperation.RestoreOperation{},
+			inputPVCNS:   "ns-1",
+			inputPVCName: "pvc-1",
+			expected:     empty,
+		},
+		{
+			name: "one operation with matched info and a nil element",
+			items: []*itemoperation.RestoreOperation{
+				nil,
+				{
+					Spec: itemoperation.RestoreOperationSpec{
+						RestoreName:       "restore-1",
+						RestoreUID:        "uid-1",
+						RestoreItemAction: "velero.io/csi-pvc-restorer",
+						OperationID:       "dd-abbb048d-7036-4855-bf50-ebba978b59a6.2426dd0e-b863-4222b5b2b",
+						ResourceIdentifier: velero.ResourceIdentifier{
+							GroupResource: schema.GroupResource{
+								Group:    "",
+								Resource: "persistentvolumeclaims",
+							},
+							Namespace: "ns-1",
+							Name:      "pvc-1",
+						},
+					},
+					Status: itemoperation.OperationStatus{
+						Phase:          itemoperation.OperationPhaseCompleted,
+						OperationUnits: "Byte",
+						Description:    "Completed",
+					},
+				},
+			},
+			inputPVCNS:   "ns-1",
+			inputPVCName: "pvc-1",
+			expected: []*itemoperation.RestoreOperation{
+				{
+					Spec: itemoperation.RestoreOperationSpec{
+						RestoreName:       "restore-1",
+						RestoreUID:        "uid-1",
+						RestoreItemAction: "velero.io/csi-pvc-restorer",
+						OperationID:       "dd-abbb048d-7036-4855-bf50-ebba978b59a6.2426dd0e-b863-4222b5b2b",
+						ResourceIdentifier: velero.ResourceIdentifier{
+							GroupResource: schema.GroupResource{
+								Group:    "",
+								Resource: "persistentvolumeclaims",
+							},
+							Namespace: "ns-1",
+							Name:      "pvc-1",
+						},
+					},
+					Status: itemoperation.OperationStatus{
+						Phase:          itemoperation.OperationPhaseCompleted,
+						OperationUnits: "Byte",
+						Description:    "Completed",
+					},
+				},
+			},
+		},
+		{
+			name: "one operation with incorrect resource type",
+			items: []*itemoperation.RestoreOperation{
+				{
+					Spec: itemoperation.RestoreOperationSpec{
+						RestoreName:       "restore-1",
+						RestoreUID:        "uid-1",
+						RestoreItemAction: "velero.io/csi-pvc-restorer",
+						OperationID:       "dd-abbb048d-7036-4855-bf50-ebba978b59a6.2426dd0e-b863-4222b5b2b",
+						ResourceIdentifier: velero.ResourceIdentifier{
+							GroupResource: schema.GroupResource{
+								Group:    "",
+								Resource: "configmaps",
+							},
+							Namespace: "ns-1",
+							Name:      "pvc-1",
+						},
+					},
+					Status: itemoperation.OperationStatus{
+						Phase:          itemoperation.OperationPhaseCompleted,
+						OperationUnits: "Byte",
+						Description:    "Completed",
+					},
+				},
+			},
+			inputPVCNS:   "ns-1",
+			inputPVCName: "pvc-1",
+			expected:     empty,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			l := restoreItemOperationList{
+				items: tt.items,
+			}
+			assert.Equal(t, tt.expected, l.SelectByPVC(tt.inputPVCNS, tt.inputPVCName))
+		})
+	}
+}

pkg/datamover/backup_micro_service_test.go

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

pkg/datamover/restore_micro_service_test.go

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

pkg/datapath/file_system_test.go

@@ -85,6 +85,7 @@ func TestAsyncBackup(t *testing.T) {
 					SnapshotID:    "fake-snapshot",
 					EmptySnapshot: false,
 					Source:        AccessPoint{ByPath: "fake-path"},
+					TotalBytes:    1000,
 				},
 			},
 			path: "fake-path",
@@ -95,7 +96,7 @@ func TestAsyncBackup(t *testing.T) {
 		t.Run(test.name, func(t *testing.T) {
 			fs := newFileSystemBR("job-1", "test", nil, "velero", Callbacks{}, velerotest.NewLogger()).(*fileSystemBR)
 			mockProvider := providerMock.NewProvider(t)
-			mockProvider.On("RunBackup", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(test.result.Backup.SnapshotID, test.result.Backup.EmptySnapshot, test.err)
+			mockProvider.On("RunBackup", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(test.result.Backup.SnapshotID, test.result.Backup.EmptySnapshot, test.result.Backup.TotalBytes, test.err)
 			mockProvider.On("Close", mock.Anything).Return(nil)
 			fs.uploaderProv = mockProvider
 			fs.initialized = true
@@ -167,7 +168,8 @@ func TestAsyncRestore(t *testing.T) {
 			},
 			result: Result{
 				Restore: RestoreResult{
-					Target: AccessPoint{ByPath: "fake-path"},
+					Target:     AccessPoint{ByPath: "fake-path"},
+					TotalBytes: 1000,
 				},
 			},
 			path:     "fake-path",
@@ -179,7 +181,7 @@ func TestAsyncRestore(t *testing.T) {
 		t.Run(test.name, func(t *testing.T) {
 			fs := newFileSystemBR("job-1", "test", nil, "velero", Callbacks{}, velerotest.NewLogger()).(*fileSystemBR)
 			mockProvider := providerMock.NewProvider(t)
-			mockProvider.On("RunRestore", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(test.err)
+			mockProvider.On("RunRestore", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(test.result.Restore.TotalBytes, test.err)
 			mockProvider.On("Close", mock.Anything).Return(nil)
 			fs.uploaderProv = mockProvider
 			fs.initialized = true

pkg/datapath/micro_service_watcher_test.go

@@ -203,6 +203,7 @@ func TestStartWatch(t *testing.T) {
 			expectStartEvent:     true,
 			expectTerminateEvent: true,
 			expectComplete:       true,
+			expectProgress:       1,
 		},
 		{
 			name:          "completed",
@@ -223,6 +224,7 @@ func TestStartWatch(t *testing.T) {
 			expectStartEvent:     true,
 			expectTerminateEvent: true,
 			expectComplete:       true,
+			expectProgress:       1,
 		},
 		{
 			name:          "completed with redirect error",
@@ -244,6 +246,7 @@ func TestStartWatch(t *testing.T) {
 			expectStartEvent:     true,
 			expectTerminateEvent: true,
 			expectComplete:       true,
+			expectProgress:       1,
 		},
 		{
 			name:          "complete but terminated event not received in time",
@@ -263,6 +266,7 @@ func TestStartWatch(t *testing.T) {
 			},
 			expectStartEvent: true,
 			expectComplete:   true,
+			expectProgress:   1,
 		},
 		{
 			name:          "complete but terminated event not received immediately",
@@ -286,6 +290,7 @@ func TestStartWatch(t *testing.T) {
 			expectStartEvent:     true,
 			expectTerminateEvent: true,
 			expectComplete:       true,
+			expectProgress:       1,
 		},
 		{
 			name:          "completed with progress",
@@ -313,7 +318,7 @@ func TestStartWatch(t *testing.T) {
 			expectStartEvent:     true,
 			expectTerminateEvent: true,
 			expectComplete:       true,
-			expectProgress:       2,
+			expectProgress:       3,
 		},
 		{
 			name:          "failed",

pkg/exposer/csi_snapshot.go

@@ -281,10 +281,16 @@ func (e *csiSnapshotExposer) GetExposed(ctx context.Context, ownerObject corev1.
 
 	curLog.WithField("pod", pod.Name).Infof("Backup volume is found in pod at index %v", i)
 
+	var nodeOS *string
+	if os, found := pod.Spec.NodeSelector[kube.NodeOSLabel]; found {
+		nodeOS = &os
+	}
+
 	return &ExposeResult{ByPod: ExposeByPod{
 		HostingPod:       pod,
 		HostingContainer: containerName,
 		VolumeName:       volumeName,
+		NodeOS:           nodeOS,
 	}}, nil
 }
 
@@ -580,6 +586,7 @@ func (e *csiSnapshotExposer) createBackupPod(
 	var securityCtx *corev1.PodSecurityContext
 	nodeSelector := map[string]string{}
 	podOS := corev1.PodOS{}
+	toleration := []corev1.Toleration{}
 	if nodeOS == kube.NodeOSWindows {
 		userID := "ContainerAdministrator"
 		securityCtx = &corev1.PodSecurityContext{
@@ -590,6 +597,13 @@ func (e *csiSnapshotExposer) createBackupPod(
 
 		nodeSelector[kube.NodeOSLabel] = kube.NodeOSWindows
 		podOS.Name = kube.NodeOSWindows
+
+		toleration = append(toleration, corev1.Toleration{
+			Key:      "os",
+			Operator: "Equal",
+			Effect:   "NoSchedule",
+			Value:    "windows",
+		})
 	} else {
 		userID := int64(0)
 		securityCtx = &corev1.PodSecurityContext{
@@ -660,6 +674,7 @@ func (e *csiSnapshotExposer) createBackupPod(
 			Volumes:                       volumes,
 			RestartPolicy:                 corev1.RestartPolicyNever,
 			SecurityContext:               securityCtx,
+			Tolerations:                   toleration,
 		},
 	}
 

pkg/exposer/generic_restore.go

@@ -35,10 +35,37 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/util/kube"
 )
 
+// GenericRestoreExposeParam define the input param for Generic Restore Expose
+type GenericRestoreExposeParam struct {
+	// TargetPVCName is the target volume name to be restored
+	TargetPVCName string
+
+	// TargetNamespace is the namespace of the volume to be restored
+	TargetNamespace string
+
+	// HostingPodLabels is the labels that are going to apply to the hosting pod
+	HostingPodLabels map[string]string
+
+	// Resources defines the resource requirements of the hosting pod
+	Resources corev1.ResourceRequirements
+
+	// ExposeTimeout specifies the timeout for the entire expose process
+	ExposeTimeout time.Duration
+
+	// OperationTimeout specifies the time wait for resources operations in Expose
+	OperationTimeout time.Duration
+
+	// NodeOS specifies the OS of node that the volume should be attached
+	NodeOS string
+
+	// RestorePVCConfig is the config for restorePVC (intermediate PVC) of generic restore
+	RestorePVCConfig nodeagent.RestorePVC
+}
+
 // GenericRestoreExposer is the interfaces for a generic restore exposer
 type GenericRestoreExposer interface {
 	// Expose starts the process to a restore expose, the expose process may take long time
-	Expose(context.Context, corev1.ObjectReference, string, string, map[string]string, corev1.ResourceRequirements, time.Duration) error
+	Expose(context.Context, corev1.ObjectReference, GenericRestoreExposeParam) error
 
 	// GetExposed polls the status of the expose.
 	// If the expose is accessible by the current caller, it waits the expose ready and returns the expose result.
@@ -74,25 +101,25 @@ type genericRestoreExposer struct {
 	log        logrus.FieldLogger
 }
 
-func (e *genericRestoreExposer) Expose(ctx context.Context, ownerObject corev1.ObjectReference, targetPVCName string, sourceNamespace string, hostingPodLabels map[string]string, resources corev1.ResourceRequirements, timeout time.Duration) error {
+func (e *genericRestoreExposer) Expose(ctx context.Context, ownerObject corev1.ObjectReference, param GenericRestoreExposeParam) error {
 	curLog := e.log.WithFields(logrus.Fields{
 		"owner":            ownerObject.Name,
-		"target PVC":       targetPVCName,
-		"source namespace": sourceNamespace,
+		"target PVC":       param.TargetPVCName,
+		"target namespace": param.TargetNamespace,
 	})
 
-	selectedNode, targetPVC, err := kube.WaitPVCConsumed(ctx, e.kubeClient.CoreV1(), targetPVCName, sourceNamespace, e.kubeClient.StorageV1(), timeout)
+	selectedNode, targetPVC, err := kube.WaitPVCConsumed(ctx, e.kubeClient.CoreV1(), param.TargetPVCName, param.TargetNamespace, e.kubeClient.StorageV1(), param.ExposeTimeout, param.RestorePVCConfig.IgnoreDelayBinding)
 	if err != nil {
-		return errors.Wrapf(err, "error to wait target PVC consumed, %s/%s", sourceNamespace, targetPVCName)
+		return errors.Wrapf(err, "error to wait target PVC consumed, %s/%s", param.TargetNamespace, param.TargetPVCName)
 	}
 
-	curLog.WithField("target PVC", targetPVCName).WithField("selected node", selectedNode).Info("Target PVC is consumed")
+	curLog.WithField("target PVC", param.TargetPVCName).WithField("selected node", selectedNode).Info("Target PVC is consumed")
 
 	if kube.IsPVCBound(targetPVC) {
-		return errors.Errorf("Target PVC %s/%s has already been bound, abort", sourceNamespace, targetPVCName)
+		return errors.Errorf("Target PVC %s/%s has already been bound, abort", param.TargetNamespace, param.TargetPVCName)
 	}
 
-	restorePod, err := e.createRestorePod(ctx, ownerObject, targetPVC, timeout, hostingPodLabels, selectedNode, resources)
+	restorePod, err := e.createRestorePod(ctx, ownerObject, targetPVC, param.OperationTimeout, param.HostingPodLabels, selectedNode, param.Resources, param.NodeOS)
 	if err != nil {
 		return errors.Wrapf(err, "error to create restore pod")
 	}
@@ -253,19 +280,19 @@ func (e *genericRestoreExposer) CleanUp(ctx context.Context, ownerObject corev1.
 	kube.DeletePVAndPVCIfAny(ctx, e.kubeClient.CoreV1(), restorePVCName, ownerObject.Namespace, 0, e.log)
 }
 
-func (e *genericRestoreExposer) RebindVolume(ctx context.Context, ownerObject corev1.ObjectReference, targetPVCName string, sourceNamespace string, timeout time.Duration) error {
+func (e *genericRestoreExposer) RebindVolume(ctx context.Context, ownerObject corev1.ObjectReference, targetPVCName string, targetNamespace string, timeout time.Duration) error {
 	restorePodName := ownerObject.Name
 	restorePVCName := ownerObject.Name
 
 	curLog := e.log.WithFields(logrus.Fields{
 		"owner":            ownerObject.Name,
 		"target PVC":       targetPVCName,
-		"source namespace": sourceNamespace,
+		"target namespace": targetNamespace,
 	})
 
-	targetPVC, err := e.kubeClient.CoreV1().PersistentVolumeClaims(sourceNamespace).Get(ctx, targetPVCName, metav1.GetOptions{})
+	targetPVC, err := e.kubeClient.CoreV1().PersistentVolumeClaims(targetNamespace).Get(ctx, targetPVCName, metav1.GetOptions{})
 	if err != nil {
-		return errors.Wrapf(err, "error to get target PVC %s/%s", sourceNamespace, targetPVCName)
+		return errors.Wrapf(err, "error to get target PVC %s/%s", targetNamespace, targetPVCName)
 	}
 
 	restorePV, err := kube.WaitPVCBound(ctx, e.kubeClient.CoreV1(), e.kubeClient.CoreV1(), restorePVCName, ownerObject.Namespace, timeout)
@@ -347,7 +374,7 @@ func (e *genericRestoreExposer) RebindVolume(ctx context.Context, ownerObject co
 }
 
 func (e *genericRestoreExposer) createRestorePod(ctx context.Context, ownerObject corev1.ObjectReference, targetPVC *corev1.PersistentVolumeClaim,
-	operationTimeout time.Duration, label map[string]string, selectedNode string, resources corev1.ResourceRequirements) (*corev1.Pod, error) {
+	operationTimeout time.Duration, label map[string]string, selectedNode string, resources corev1.ResourceRequirements, nodeType string) (*corev1.Pod, error) {
 	restorePodName := ownerObject.Name
 	restorePVCName := ownerObject.Name
 
@@ -388,7 +415,36 @@ func (e *genericRestoreExposer) createRestorePod(ctx context.Context, ownerObjec
 	args = append(args, podInfo.logFormatArgs...)
 	args = append(args, podInfo.logLevelArgs...)
 
-	userID := int64(0)
+	var securityCtx *corev1.PodSecurityContext
+	nodeSelector := map[string]string{}
+	podOS := corev1.PodOS{}
+	toleration := []corev1.Toleration{}
+	if nodeType == kube.NodeOSWindows {
+		userID := "ContainerAdministrator"
+		securityCtx = &corev1.PodSecurityContext{
+			WindowsOptions: &corev1.WindowsSecurityContextOptions{
+				RunAsUserName: &userID,
+			},
+		}
+
+		nodeSelector[kube.NodeOSLabel] = kube.NodeOSWindows
+		podOS.Name = kube.NodeOSWindows
+
+		toleration = append(toleration, corev1.Toleration{
+			Key:      "os",
+			Operator: "Equal",
+			Effect:   "NoSchedule",
+			Value:    "windows",
+		})
+	} else {
+		userID := int64(0)
+		securityCtx = &corev1.PodSecurityContext{
+			RunAsUser: &userID,
+		}
+
+		nodeSelector[kube.NodeOSLabel] = kube.NodeOSLinux
+		podOS.Name = kube.NodeOSLinux
+	}
 
 	pod := &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{
@@ -406,6 +462,8 @@ func (e *genericRestoreExposer) createRestorePod(ctx context.Context, ownerObjec
 			Labels: label,
 		},
 		Spec: corev1.PodSpec{
+			NodeSelector: nodeSelector,
+			OS:           &podOS,
 			Containers: []corev1.Container{
 				{
 					Name:            containerName,
@@ -429,9 +487,8 @@ func (e *genericRestoreExposer) createRestorePod(ctx context.Context, ownerObjec
 			Volumes:                       volumes,
 			NodeName:                      selectedNode,
 			RestartPolicy:                 corev1.RestartPolicyNever,
-			SecurityContext: &corev1.PodSecurityContext{
-				RunAsUser: &userID,
-			},
+			SecurityContext:               securityCtx,
+			Tolerations:                   toleration,
 		},
 	}
 

pkg/exposer/generic_restore_test.go

@@ -31,7 +31,6 @@ import (
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
 
 	appsv1 "k8s.io/api/apps/v1"
-	corev1 "k8s.io/api/core/v1"
 	corev1api "k8s.io/api/core/v1"
 	clientTesting "k8s.io/client-go/testing"
 )
@@ -76,9 +75,9 @@ func TestRestoreExpose(t *testing.T) {
 			APIVersion: appsv1.SchemeGroupVersion.String(),
 		},
 		Spec: appsv1.DaemonSetSpec{
-			Template: corev1.PodTemplateSpec{
-				Spec: corev1.PodSpec{
-					Containers: []corev1.Container{
+			Template: corev1api.PodTemplateSpec{
+				Spec: corev1api.PodSpec{
+					Containers: []corev1api.Container{
 						{
 							Image: "fake-image",
 						},
@@ -93,21 +92,21 @@ func TestRestoreExpose(t *testing.T) {
 		kubeClientObj   []runtime.Object
 		ownerRestore    *velerov1.Restore
 		targetPVCName   string
-		sourceNamespace string
+		targetNamespace string
 		kubeReactors    []reactor
 		err             string
 	}{
 		{
 			name:            "wait target pvc consumed fail",
 			targetPVCName:   "fake-target-pvc",
-			sourceNamespace: "fake-ns",
+			targetNamespace: "fake-ns",
 			ownerRestore:    restore,
 			err:             "error to wait target PVC consumed, fake-ns/fake-target-pvc: error to wait for PVC: error to get pvc fake-ns/fake-target-pvc: persistentvolumeclaims \"fake-target-pvc\" not found",
 		},
 		{
 			name:            "target pvc is already bound",
 			targetPVCName:   "fake-target-pvc",
-			sourceNamespace: "fake-ns",
+			targetNamespace: "fake-ns",
 			ownerRestore:    restore,
 			kubeClientObj: []runtime.Object{
 				targetPVCObjBound,
@@ -117,7 +116,7 @@ func TestRestoreExpose(t *testing.T) {
 		{
 			name:            "create restore pod fail",
 			targetPVCName:   "fake-target-pvc",
-			sourceNamespace: "fake-ns",
+			targetNamespace: "fake-ns",
 			ownerRestore:    restore,
 			kubeClientObj: []runtime.Object{
 				targetPVCObj,
@@ -137,7 +136,7 @@ func TestRestoreExpose(t *testing.T) {
 		{
 			name:            "create restore pvc fail",
 			targetPVCName:   "fake-target-pvc",
-			sourceNamespace: "fake-ns",
+			targetNamespace: "fake-ns",
 			ownerRestore:    restore,
 			kubeClientObj: []runtime.Object{
 				targetPVCObj,
@@ -180,7 +179,12 @@ func TestRestoreExpose(t *testing.T) {
 				}
 			}
 
-			err := exposer.Expose(context.Background(), ownerObject, test.targetPVCName, test.sourceNamespace, map[string]string{}, corev1.ResourceRequirements{}, time.Millisecond)
+			err := exposer.Expose(context.Background(), ownerObject, GenericRestoreExposeParam{
+				TargetPVCName:    test.targetPVCName,
+				TargetNamespace:  test.targetNamespace,
+				HostingPodLabels: map[string]string{},
+				Resources:        corev1api.ResourceRequirements{},
+				ExposeTimeout:    time.Millisecond})
 			assert.EqualError(t, err, test.err)
 		})
 	}
@@ -239,21 +243,21 @@ func TestRebindVolume(t *testing.T) {
 		kubeClientObj   []runtime.Object
 		ownerRestore    *velerov1.Restore
 		targetPVCName   string
-		sourceNamespace string
+		targetNamespace string
 		kubeReactors    []reactor
 		err             string
 	}{
 		{
 			name:            "get target pvc fail",
 			targetPVCName:   "fake-target-pvc",
-			sourceNamespace: "fake-ns",
+			targetNamespace: "fake-ns",
 			ownerRestore:    restore,
 			err:             "error to get target PVC fake-ns/fake-target-pvc: persistentvolumeclaims \"fake-target-pvc\" not found",
 		},
 		{
 			name:            "wait restore pvc bound fail",
 			targetPVCName:   "fake-target-pvc",
-			sourceNamespace: "fake-ns",
+			targetNamespace: "fake-ns",
 			ownerRestore:    restore,
 			kubeClientObj: []runtime.Object{
 				targetPVCObj,
@@ -263,7 +267,7 @@ func TestRebindVolume(t *testing.T) {
 		{
 			name:            "retain target pv fail",
 			targetPVCName:   "fake-target-pvc",
-			sourceNamespace: "fake-ns",
+			targetNamespace: "fake-ns",
 			ownerRestore:    restore,
 			kubeClientObj: []runtime.Object{
 				targetPVCObj,
@@ -284,7 +288,7 @@ func TestRebindVolume(t *testing.T) {
 		{
 			name:            "delete restore pod fail",
 			targetPVCName:   "fake-target-pvc",
-			sourceNamespace: "fake-ns",
+			targetNamespace: "fake-ns",
 			ownerRestore:    restore,
 			kubeClientObj: []runtime.Object{
 				targetPVCObj,
@@ -306,7 +310,7 @@ func TestRebindVolume(t *testing.T) {
 		{
 			name:            "delete restore pvc fail",
 			targetPVCName:   "fake-target-pvc",
-			sourceNamespace: "fake-ns",
+			targetNamespace: "fake-ns",
 			ownerRestore:    restore,
 			kubeClientObj: []runtime.Object{
 				targetPVCObj,
@@ -328,7 +332,7 @@ func TestRebindVolume(t *testing.T) {
 		{
 			name:            "rebind target pvc fail",
 			targetPVCName:   "fake-target-pvc",
-			sourceNamespace: "fake-ns",
+			targetNamespace: "fake-ns",
 			ownerRestore:    restore,
 			kubeClientObj: []runtime.Object{
 				targetPVCObj,
@@ -350,7 +354,7 @@ func TestRebindVolume(t *testing.T) {
 		{
 			name:            "reset pv binding fail",
 			targetPVCName:   "fake-target-pvc",
-			sourceNamespace: "fake-ns",
+			targetNamespace: "fake-ns",
 			ownerRestore:    restore,
 			kubeClientObj: []runtime.Object{
 				targetPVCObj,
@@ -377,7 +381,7 @@ func TestRebindVolume(t *testing.T) {
 		{
 			name:            "wait restore PV bound fail",
 			targetPVCName:   "fake-target-pvc",
-			sourceNamespace: "fake-ns",
+			targetNamespace: "fake-ns",
 			ownerRestore:    restore,
 			kubeClientObj: []runtime.Object{
 				targetPVCObj,
@@ -415,7 +419,7 @@ func TestRebindVolume(t *testing.T) {
 
 			hookCount = 0
 
-			err := exposer.RebindVolume(context.Background(), ownerObject, test.targetPVCName, test.sourceNamespace, time.Millisecond)
+			err := exposer.RebindVolume(context.Background(), ownerObject, test.targetPVCName, test.targetNamespace, time.Millisecond)
 			assert.EqualError(t, err, test.err)
 		})
 	}
@@ -521,7 +525,7 @@ func Test_ReastoreDiagnoseExpose(t *testing.T) {
 		},
 	}
 
-	restorePodWithoutNodeName := corev1.Pod{
+	restorePodWithoutNodeName := corev1api.Pod{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: velerov1.DefaultNamespace,
 			Name:      "fake-restore",
@@ -534,19 +538,19 @@ func Test_ReastoreDiagnoseExpose(t *testing.T) {
 				},
 			},
 		},
-		Status: corev1.PodStatus{
-			Phase: corev1.PodPending,
-			Conditions: []corev1.PodCondition{
+		Status: corev1api.PodStatus{
+			Phase: corev1api.PodPending,
+			Conditions: []corev1api.PodCondition{
 				{
-					Type:    corev1.PodInitialized,
-					Status:  corev1.ConditionTrue,
+					Type:    corev1api.PodInitialized,
+					Status:  corev1api.ConditionTrue,
 					Message: "fake-pod-message",
 				},
 			},
 		},
 	}
 
-	restorePodWithNodeName := corev1.Pod{
+	restorePodWithNodeName := corev1api.Pod{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: velerov1.DefaultNamespace,
 			Name:      "fake-restore",
@@ -559,22 +563,22 @@ func Test_ReastoreDiagnoseExpose(t *testing.T) {
 				},
 			},
 		},
-		Spec: corev1.PodSpec{
+		Spec: corev1api.PodSpec{
 			NodeName: "fake-node",
 		},
-		Status: corev1.PodStatus{
-			Phase: corev1.PodPending,
-			Conditions: []corev1.PodCondition{
+		Status: corev1api.PodStatus{
+			Phase: corev1api.PodPending,
+			Conditions: []corev1api.PodCondition{
 				{
-					Type:    corev1.PodInitialized,
-					Status:  corev1.ConditionTrue,
+					Type:    corev1api.PodInitialized,
+					Status:  corev1api.ConditionTrue,
 					Message: "fake-pod-message",
 				},
 			},
 		},
 	}
 
-	restorePVCWithoutVolumeName := corev1.PersistentVolumeClaim{
+	restorePVCWithoutVolumeName := corev1api.PersistentVolumeClaim{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: velerov1.DefaultNamespace,
 			Name:      "fake-restore",
@@ -587,12 +591,12 @@ func Test_ReastoreDiagnoseExpose(t *testing.T) {
 				},
 			},
 		},
-		Status: corev1.PersistentVolumeClaimStatus{
-			Phase: corev1.ClaimPending,
+		Status: corev1api.PersistentVolumeClaimStatus{
+			Phase: corev1api.ClaimPending,
 		},
 	}
 
-	restorePVCWithVolumeName := corev1.PersistentVolumeClaim{
+	restorePVCWithVolumeName := corev1api.PersistentVolumeClaim{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: velerov1.DefaultNamespace,
 			Name:      "fake-restore",
@@ -605,35 +609,35 @@ func Test_ReastoreDiagnoseExpose(t *testing.T) {
 				},
 			},
 		},
-		Spec: corev1.PersistentVolumeClaimSpec{
+		Spec: corev1api.PersistentVolumeClaimSpec{
 			VolumeName: "fake-pv",
 		},
-		Status: corev1.PersistentVolumeClaimStatus{
-			Phase: corev1.ClaimPending,
+		Status: corev1api.PersistentVolumeClaimStatus{
+			Phase: corev1api.ClaimPending,
 		},
 	}
 
-	restorePV := corev1.PersistentVolume{
+	restorePV := corev1api.PersistentVolume{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: "fake-pv",
 		},
-		Status: corev1.PersistentVolumeStatus{
-			Phase:   corev1.VolumePending,
+		Status: corev1api.PersistentVolumeStatus{
+			Phase:   corev1api.VolumePending,
 			Message: "fake-pv-message",
 		},
 	}
 
-	nodeAgentPod := corev1.Pod{
+	nodeAgentPod := corev1api.Pod{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: velerov1.DefaultNamespace,
 			Name:      "node-agent-pod-1",
 			Labels:    map[string]string{"role": "node-agent"},
 		},
-		Spec: corev1.PodSpec{
+		Spec: corev1api.PodSpec{
 			NodeName: "fake-node",
 		},
-		Status: corev1.PodStatus{
-			Phase: corev1.PodRunning,
+		Status: corev1api.PodStatus{
+			Phase: corev1api.PodRunning,
 		},
 	}
 

pkg/exposer/mocks/GenericRestoreExposer.go

@@ -44,17 +44,17 @@ func (_m *GenericRestoreExposer) DiagnoseExpose(_a0 context.Context, _a1 v1.Obje
 	return r0
 }
 
-// Expose provides a mock function with given fields: _a0, _a1, _a2, _a3, _a4, _a5, _a6
-func (_m *GenericRestoreExposer) Expose(_a0 context.Context, _a1 v1.ObjectReference, _a2 string, _a3 string, _a4 map[string]string, _a5 v1.ResourceRequirements, _a6 time.Duration) error {
-	ret := _m.Called(_a0, _a1, _a2, _a3, _a4, _a5, _a6)
+// Expose provides a mock function with given fields: _a0, _a1, _a2
+func (_m *GenericRestoreExposer) Expose(_a0 context.Context, _a1 v1.ObjectReference, _a2 exposer.GenericRestoreExposeParam) error {
+	ret := _m.Called(_a0, _a1, _a2)
 
 	if len(ret) == 0 {
 		panic("no return value specified for Expose")
 	}
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, v1.ObjectReference, string, string, map[string]string, v1.ResourceRequirements, time.Duration) error); ok {
-		r0 = rf(_a0, _a1, _a2, _a3, _a4, _a5, _a6)
+	if rf, ok := ret.Get(0).(func(context.Context, v1.ObjectReference, exposer.GenericRestoreExposeParam) error); ok {
+		r0 = rf(_a0, _a1, _a2)
 	} else {
 		r0 = ret.Error(0)
 	}

pkg/exposer/types.go

@@ -38,4 +38,5 @@ type ExposeByPod struct {
 	HostingPod       *corev1.Pod
 	HostingContainer string
 	VolumeName       string
+	NodeOS           *string
 }

pkg/label/label.go

@@ -18,6 +18,7 @@ package label
 
 import (
 	"crypto/sha256"
+	"encoding/hex"
 	"fmt"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -38,7 +39,7 @@ func GetValidName(label string) string {
 	}
 
 	sha := sha256.Sum256([]byte(label))
-	strSha := fmt.Sprintf("%x", sha)
+	strSha := hex.EncodeToString(sha[:])
 	charsFromLabel := validation.DNS1035LabelMaxLength - 6
 	if charsFromLabel < 0 {
 		// Derive the label name from sha hash in case the DNS1035LabelMaxLength is less than 6

pkg/nodeagent/node_agent.go

@@ -81,6 +81,11 @@ type BackupPVC struct {
 	SPCNoRelabeling bool `json:"spcNoRelabeling,omitempty"`
 }
 
+type RestorePVC struct {
+	// IgnoreDelayBinding indicates to ignore delay binding the restorePVC when it is in WaitForFirstConsumer mode
+	IgnoreDelayBinding bool `json:"ignoreDelayBinding,omitempty"`
+}
+
 type Configs struct {
 	// LoadConcurrency is the config for data path load concurrency per node.
 	LoadConcurrency *LoadConcurrency `json:"loadConcurrency,omitempty"`
@@ -91,6 +96,9 @@ type Configs struct {
 	// BackupPVCConfig is the config for backupPVC (intermediate PVC) of snapshot data movement
 	BackupPVCConfig map[string]BackupPVC `json:"backupPVC,omitempty"`
 
+	// RestoreVCConfig is the config for restorePVC (intermediate PVC) of generic restore
+	RestorePVCConfig *RestorePVC `json:"restorePVC,omitempty"`
+
 	// PodResources is the resource config for various types of pods launched by node-agent, i.e., data mover pods.
 	PodResources *kube.PodResources `json:"podResources,omitempty"`
 }

pkg/persistence/mocks/backup_store.go

@@ -13,7 +13,7 @@ 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.
 */
-// Code generated by mockery v2.16.0. DO NOT EDIT.
+// Code generated by mockery v2.42.2. DO NOT EDIT.
 
 package mocks
 
@@ -21,15 +21,17 @@ import (
 	io "io"
 
 	mock "github.com/stretchr/testify/mock"
-	volumesnapshotv1 "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
-
 	itemoperation "github.com/vmware-tanzu/velero/pkg/itemoperation"
-	"github.com/vmware-tanzu/velero/pkg/persistence"
+
+	persistence "github.com/vmware-tanzu/velero/pkg/persistence"
+
+	results "github.com/vmware-tanzu/velero/pkg/util/results"
+
 	v1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
-	"github.com/vmware-tanzu/velero/internal/volume"
-	"github.com/vmware-tanzu/velero/pkg/util/results"
 
+	volume "github.com/vmware-tanzu/velero/internal/volume"
 
+	volumesnapshotv1 "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
 )
 
 // BackupStore is an autogenerated mock type for the BackupStore type
@@ -41,14 +43,21 @@ type BackupStore struct {
 func (_m *BackupStore) BackupExists(bucket string, backupName string) (bool, error) {
 	ret := _m.Called(bucket, backupName)
 
+	if len(ret) == 0 {
+		panic("no return value specified for BackupExists")
+	}
+
 	var r0 bool
+	var r1 error
+	if rf, ok := ret.Get(0).(func(string, string) (bool, error)); ok {
+		return rf(bucket, backupName)
+	}
 	if rf, ok := ret.Get(0).(func(string, string) bool); ok {
 		r0 = rf(bucket, backupName)
 	} else {
 		r0 = ret.Get(0).(bool)
 	}
 
-	var r1 error
 	if rf, ok := ret.Get(1).(func(string, string) error); ok {
 		r1 = rf(bucket, backupName)
 	} else {
@@ -62,6 +71,10 @@ func (_m *BackupStore) BackupExists(bucket string, backupName string) (bool, err
 func (_m *BackupStore) DeleteBackup(name string) error {
 	ret := _m.Called(name)
 
+	if len(ret) == 0 {
+		panic("no return value specified for DeleteBackup")
+	}
+
 	var r0 error
 	if rf, ok := ret.Get(0).(func(string) error); ok {
 		r0 = rf(name)
@@ -76,6 +89,10 @@ func (_m *BackupStore) DeleteBackup(name string) error {
 func (_m *BackupStore) DeleteRestore(name string) error {
 	ret := _m.Called(name)
 
+	if len(ret) == 0 {
+		panic("no return value specified for DeleteRestore")
+	}
+
 	var r0 error
 	if rf, ok := ret.Get(0).(func(string) error); ok {
 		r0 = rf(name)
@@ -90,7 +107,15 @@ func (_m *BackupStore) DeleteRestore(name string) error {
 func (_m *BackupStore) GetBackupContents(name string) (io.ReadCloser, error) {
 	ret := _m.Called(name)
 
+	if len(ret) == 0 {
+		panic("no return value specified for GetBackupContents")
+	}
+
 	var r0 io.ReadCloser
+	var r1 error
+	if rf, ok := ret.Get(0).(func(string) (io.ReadCloser, error)); ok {
+		return rf(name)
+	}
 	if rf, ok := ret.Get(0).(func(string) io.ReadCloser); ok {
 		r0 = rf(name)
 	} else {
@@ -99,7 +124,6 @@ func (_m *BackupStore) GetBackupContents(name string) (io.ReadCloser, error) {
 		}
 	}
 
-	var r1 error
 	if rf, ok := ret.Get(1).(func(string) error); ok {
 		r1 = rf(name)
 	} else {
@@ -113,7 +137,15 @@ func (_m *BackupStore) GetBackupContents(name string) (io.ReadCloser, error) {
 func (_m *BackupStore) GetBackupItemOperations(name string) ([]*itemoperation.BackupOperation, error) {
 	ret := _m.Called(name)
 
+	if len(ret) == 0 {
+		panic("no return value specified for GetBackupItemOperations")
+	}
+
 	var r0 []*itemoperation.BackupOperation
+	var r1 error
+	if rf, ok := ret.Get(0).(func(string) ([]*itemoperation.BackupOperation, error)); ok {
+		return rf(name)
+	}
 	if rf, ok := ret.Get(0).(func(string) []*itemoperation.BackupOperation); ok {
 		r0 = rf(name)
 	} else {
@@ -122,7 +154,6 @@ func (_m *BackupStore) GetBackupItemOperations(name string) ([]*itemoperation.Ba
 		}
 	}
 
-	var r1 error
 	if rf, ok := ret.Get(1).(func(string) error); ok {
 		r1 = rf(name)
 	} else {
@@ -136,7 +167,15 @@ func (_m *BackupStore) GetBackupItemOperations(name string) ([]*itemoperation.Ba
 func (_m *BackupStore) GetBackupMetadata(name string) (*v1.Backup, error) {
 	ret := _m.Called(name)
 
+	if len(ret) == 0 {
+		panic("no return value specified for GetBackupMetadata")
+	}
+
 	var r0 *v1.Backup
+	var r1 error
+	if rf, ok := ret.Get(0).(func(string) (*v1.Backup, error)); ok {
+		return rf(name)
+	}
 	if rf, ok := ret.Get(0).(func(string) *v1.Backup); ok {
 		r0 = rf(name)
 	} else {
@@ -145,7 +184,36 @@ func (_m *BackupStore) GetBackupMetadata(name string) (*v1.Backup, error) {
 		}
 	}
 
+	if rf, ok := ret.Get(1).(func(string) error); ok {
+		r1 = rf(name)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// GetBackupVolumeInfos provides a mock function with given fields: name
+func (_m *BackupStore) GetBackupVolumeInfos(name string) ([]*volume.BackupVolumeInfo, error) {
+	ret := _m.Called(name)
+
+	if len(ret) == 0 {
+		panic("no return value specified for GetBackupVolumeInfos")
+	}
+
+	var r0 []*volume.BackupVolumeInfo
 	var r1 error
+	if rf, ok := ret.Get(0).(func(string) ([]*volume.BackupVolumeInfo, error)); ok {
+		return rf(name)
+	}
+	if rf, ok := ret.Get(0).(func(string) []*volume.BackupVolumeInfo); ok {
+		r0 = rf(name)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]*volume.BackupVolumeInfo)
+		}
+	}
+
 	if rf, ok := ret.Get(1).(func(string) error); ok {
 		r1 = rf(name)
 	} else {
@@ -159,7 +227,15 @@ func (_m *BackupStore) GetBackupMetadata(name string) (*v1.Backup, error) {
 func (_m *BackupStore) GetBackupVolumeSnapshots(name string) ([]*volume.Snapshot, error) {
 	ret := _m.Called(name)
 
+	if len(ret) == 0 {
+		panic("no return value specified for GetBackupVolumeSnapshots")
+	}
+
 	var r0 []*volume.Snapshot
+	var r1 error
+	if rf, ok := ret.Get(0).(func(string) ([]*volume.Snapshot, error)); ok {
+		return rf(name)
+	}
 	if rf, ok := ret.Get(0).(func(string) []*volume.Snapshot); ok {
 		r0 = rf(name)
 	} else {
@@ -168,7 +244,6 @@ func (_m *BackupStore) GetBackupVolumeSnapshots(name string) ([]*volume.Snapshot
 		}
 	}
 
-	var r1 error
 	if rf, ok := ret.Get(1).(func(string) error); ok {
 		r1 = rf(name)
 	} else {
@@ -182,7 +257,15 @@ func (_m *BackupStore) GetBackupVolumeSnapshots(name string) ([]*volume.Snapshot
 func (_m *BackupStore) GetCSIVolumeSnapshotClasses(name string) ([]*volumesnapshotv1.VolumeSnapshotClass, error) {
 	ret := _m.Called(name)
 
+	if len(ret) == 0 {
+		panic("no return value specified for GetCSIVolumeSnapshotClasses")
+	}
+
 	var r0 []*volumesnapshotv1.VolumeSnapshotClass
+	var r1 error
+	if rf, ok := ret.Get(0).(func(string) ([]*volumesnapshotv1.VolumeSnapshotClass, error)); ok {
+		return rf(name)
+	}
 	if rf, ok := ret.Get(0).(func(string) []*volumesnapshotv1.VolumeSnapshotClass); ok {
 		r0 = rf(name)
 	} else {
@@ -191,7 +274,6 @@ func (_m *BackupStore) GetCSIVolumeSnapshotClasses(name string) ([]*volumesnapsh
 		}
 	}
 
-	var r1 error
 	if rf, ok := ret.Get(1).(func(string) error); ok {
 		r1 = rf(name)
 	} else {
@@ -205,7 +287,15 @@ func (_m *BackupStore) GetCSIVolumeSnapshotClasses(name string) ([]*volumesnapsh
 func (_m *BackupStore) GetCSIVolumeSnapshotContents(name string) ([]*volumesnapshotv1.VolumeSnapshotContent, error) {
 	ret := _m.Called(name)
 
+	if len(ret) == 0 {
+		panic("no return value specified for GetCSIVolumeSnapshotContents")
+	}
+
 	var r0 []*volumesnapshotv1.VolumeSnapshotContent
+	var r1 error
+	if rf, ok := ret.Get(0).(func(string) ([]*volumesnapshotv1.VolumeSnapshotContent, error)); ok {
+		return rf(name)
+	}
 	if rf, ok := ret.Get(0).(func(string) []*volumesnapshotv1.VolumeSnapshotContent); ok {
 		r0 = rf(name)
 	} else {
@@ -214,7 +304,6 @@ func (_m *BackupStore) GetCSIVolumeSnapshotContents(name string) ([]*volumesnaps
 		}
 	}
 
-	var r1 error
 	if rf, ok := ret.Get(1).(func(string) error); ok {
 		r1 = rf(name)
 	} else {
@@ -228,7 +317,15 @@ func (_m *BackupStore) GetCSIVolumeSnapshotContents(name string) ([]*volumesnaps
 func (_m *BackupStore) GetCSIVolumeSnapshots(name string) ([]*volumesnapshotv1.VolumeSnapshot, error) {
 	ret := _m.Called(name)
 
+	if len(ret) == 0 {
+		panic("no return value specified for GetCSIVolumeSnapshots")
+	}
+
 	var r0 []*volumesnapshotv1.VolumeSnapshot
+	var r1 error
+	if rf, ok := ret.Get(0).(func(string) ([]*volumesnapshotv1.VolumeSnapshot, error)); ok {
+		return rf(name)
+	}
 	if rf, ok := ret.Get(0).(func(string) []*volumesnapshotv1.VolumeSnapshot); ok {
 		r0 = rf(name)
 	} else {
@@ -237,7 +334,6 @@ func (_m *BackupStore) GetCSIVolumeSnapshots(name string) ([]*volumesnapshotv1.V
 		}
 	}
 
-	var r1 error
 	if rf, ok := ret.Get(1).(func(string) error); ok {
 		r1 = rf(name)
 	} else {
@@ -251,14 +347,21 @@ func (_m *BackupStore) GetCSIVolumeSnapshots(name string) ([]*volumesnapshotv1.V
 func (_m *BackupStore) GetDownloadURL(target v1.DownloadTarget) (string, error) {
 	ret := _m.Called(target)
 
+	if len(ret) == 0 {
+		panic("no return value specified for GetDownloadURL")
+	}
+
 	var r0 string
+	var r1 error
+	if rf, ok := ret.Get(0).(func(v1.DownloadTarget) (string, error)); ok {
+		return rf(target)
+	}
 	if rf, ok := ret.Get(0).(func(v1.DownloadTarget) string); ok {
 		r0 = rf(target)
 	} else {
 		r0 = ret.Get(0).(string)
 	}
 
-	var r1 error
 	if rf, ok := ret.Get(1).(func(v1.DownloadTarget) error); ok {
 		r1 = rf(target)
 	} else {
@@ -272,7 +375,15 @@ func (_m *BackupStore) GetDownloadURL(target v1.DownloadTarget) (string, error)
 func (_m *BackupStore) GetPodVolumeBackups(name string) ([]*v1.PodVolumeBackup, error) {
 	ret := _m.Called(name)
 
+	if len(ret) == 0 {
+		panic("no return value specified for GetPodVolumeBackups")
+	}
+
 	var r0 []*v1.PodVolumeBackup
+	var r1 error
+	if rf, ok := ret.Get(0).(func(string) ([]*v1.PodVolumeBackup, error)); ok {
+		return rf(name)
+	}
 	if rf, ok := ret.Get(0).(func(string) []*v1.PodVolumeBackup); ok {
 		r0 = rf(name)
 	} else {
@@ -281,7 +392,6 @@ func (_m *BackupStore) GetPodVolumeBackups(name string) ([]*v1.PodVolumeBackup,
 		}
 	}
 
-	var r1 error
 	if rf, ok := ret.Get(1).(func(string) error); ok {
 		r1 = rf(name)
 	} else {
@@ -295,7 +405,15 @@ func (_m *BackupStore) GetPodVolumeBackups(name string) ([]*v1.PodVolumeBackup,
 func (_m *BackupStore) GetRestoreItemOperations(name string) ([]*itemoperation.RestoreOperation, error) {
 	ret := _m.Called(name)
 
+	if len(ret) == 0 {
+		panic("no return value specified for GetRestoreItemOperations")
+	}
+
 	var r0 []*itemoperation.RestoreOperation
+	var r1 error
+	if rf, ok := ret.Get(0).(func(string) ([]*itemoperation.RestoreOperation, error)); ok {
+		return rf(name)
+	}
 	if rf, ok := ret.Get(0).(func(string) []*itemoperation.RestoreOperation); ok {
 		r0 = rf(name)
 	} else {
@@ -304,7 +422,6 @@ func (_m *BackupStore) GetRestoreItemOperations(name string) ([]*itemoperation.R
 		}
 	}
 
-	var r1 error
 	if rf, ok := ret.Get(1).(func(string) error); ok {
 		r1 = rf(name)
 	} else {
@@ -314,48 +431,19 @@ func (_m *BackupStore) GetRestoreItemOperations(name string) ([]*itemoperation.R
 	return r0, r1
 }
 
-// GetRestoreItemOperations provides a mock function with given fields: name
-func (_m *BackupStore) GetBackupVolumeInfos(name string) ([]*volume.BackupVolumeInfo, error) {
-	ret := _m.Called(name)
-
-	var r0 []*volume.BackupVolumeInfo
-	if rf, ok := ret.Get(0).(func(string) []*volume.BackupVolumeInfo); ok {
-		r0 = rf(name)
-	} else {
-		if ret.Get(0) != nil {
-			r0 = ret.Get(0).([]*volume.BackupVolumeInfo)
-		}
-	}
-
-	var r1 error
-	if rf, ok := ret.Get(1).(func(string) error); ok {
-		r1 = rf(name)
-	} else {
-		r1 = ret.Error(1)
-	}
-
-	return r0, r1
-}
-
-// PutBackupVolumeInfos provides a mock function with given fields: name, volumeInfo
-func (_m *BackupStore) PutBackupVolumeInfos(name string, volumeInfo io.Reader) error {
-	ret := _m.Called(name, volumeInfo)
-
-	var r0 error
-	if rf, ok := ret.Get(0).(func(string, io.Reader) error); ok {
-		r0 = rf(name, volumeInfo)
-	} else {
-		r0 = ret.Error(0)
-	}
-
-	return r0
-}
-
 // GetRestoreResults provides a mock function with given fields: name
 func (_m *BackupStore) GetRestoreResults(name string) (map[string]results.Result, error) {
 	ret := _m.Called(name)
 
-	r0 := make(map[string]results.Result)
+	if len(ret) == 0 {
+		panic("no return value specified for GetRestoreResults")
+	}
+
+	var r0 map[string]results.Result
+	var r1 error
+	if rf, ok := ret.Get(0).(func(string) (map[string]results.Result, error)); ok {
+		return rf(name)
+	}
 	if rf, ok := ret.Get(0).(func(string) map[string]results.Result); ok {
 		r0 = rf(name)
 	} else {
@@ -364,7 +452,6 @@ func (_m *BackupStore) GetRestoreResults(name string) (map[string]results.Result
 		}
 	}
 
-	var r1 error
 	if rf, ok := ret.Get(1).(func(string) error); ok {
 		r1 = rf(name)
 	} else {
@@ -378,7 +465,15 @@ func (_m *BackupStore) GetRestoreResults(name string) (map[string]results.Result
 func (_m *BackupStore) GetRestoredResourceList(name string) (map[string][]string, error) {
 	ret := _m.Called(name)
 
-	r0 := make(map[string][]string)
+	if len(ret) == 0 {
+		panic("no return value specified for GetRestoredResourceList")
+	}
+
+	var r0 map[string][]string
+	var r1 error
+	if rf, ok := ret.Get(0).(func(string) (map[string][]string, error)); ok {
+		return rf(name)
+	}
 	if rf, ok := ret.Get(0).(func(string) map[string][]string); ok {
 		r0 = rf(name)
 	} else {
@@ -387,7 +482,6 @@ func (_m *BackupStore) GetRestoredResourceList(name string) (map[string][]string
 		}
 	}
 
-	var r1 error
 	if rf, ok := ret.Get(1).(func(string) error); ok {
 		r1 = rf(name)
 	} else {
@@ -401,6 +495,10 @@ func (_m *BackupStore) GetRestoredResourceList(name string) (map[string][]string
 func (_m *BackupStore) IsValid() error {
 	ret := _m.Called()
 
+	if len(ret) == 0 {
+		panic("no return value specified for IsValid")
+	}
+
 	var r0 error
 	if rf, ok := ret.Get(0).(func() error); ok {
 		r0 = rf()
@@ -415,7 +513,15 @@ func (_m *BackupStore) IsValid() error {
 func (_m *BackupStore) ListBackups() ([]string, error) {
 	ret := _m.Called()
 
+	if len(ret) == 0 {
+		panic("no return value specified for ListBackups")
+	}
+
 	var r0 []string
+	var r1 error
+	if rf, ok := ret.Get(0).(func() ([]string, error)); ok {
+		return rf()
+	}
 	if rf, ok := ret.Get(0).(func() []string); ok {
 		r0 = rf()
 	} else {
@@ -424,7 +530,6 @@ func (_m *BackupStore) ListBackups() ([]string, error) {
 		}
 	}
 
-	var r1 error
 	if rf, ok := ret.Get(1).(func() error); ok {
 		r1 = rf()
 	} else {
@@ -438,6 +543,10 @@ func (_m *BackupStore) ListBackups() ([]string, error) {
 func (_m *BackupStore) PutBackup(info persistence.BackupInfo) error {
 	ret := _m.Called(info)
 
+	if len(ret) == 0 {
+		panic("no return value specified for PutBackup")
+	}
+
 	var r0 error
 	if rf, ok := ret.Get(0).(func(persistence.BackupInfo) error); ok {
 		r0 = rf(info)
@@ -452,6 +561,10 @@ func (_m *BackupStore) PutBackup(info persistence.BackupInfo) error {
 func (_m *BackupStore) PutBackupContents(backup string, backupContents io.Reader) error {
 	ret := _m.Called(backup, backupContents)
 
+	if len(ret) == 0 {
+		panic("no return value specified for PutBackupContents")
+	}
+
 	var r0 error
 	if rf, ok := ret.Get(0).(func(string, io.Reader) error); ok {
 		r0 = rf(backup, backupContents)
@@ -466,6 +579,10 @@ func (_m *BackupStore) PutBackupContents(backup string, backupContents io.Reader
 func (_m *BackupStore) PutBackupItemOperations(backup string, backupItemOperations io.Reader) error {
 	ret := _m.Called(backup, backupItemOperations)
 
+	if len(ret) == 0 {
+		panic("no return value specified for PutBackupItemOperations")
+	}
+
 	var r0 error
 	if rf, ok := ret.Get(0).(func(string, io.Reader) error); ok {
 		r0 = rf(backup, backupItemOperations)
@@ -480,6 +597,10 @@ func (_m *BackupStore) PutBackupItemOperations(backup string, backupItemOperatio
 func (_m *BackupStore) PutBackupMetadata(backup string, backupMetadata io.Reader) error {
 	ret := _m.Called(backup, backupMetadata)
 
+	if len(ret) == 0 {
+		panic("no return value specified for PutBackupMetadata")
+	}
+
 	var r0 error
 	if rf, ok := ret.Get(0).(func(string, io.Reader) error); ok {
 		r0 = rf(backup, backupMetadata)
@@ -490,10 +611,32 @@ func (_m *BackupStore) PutBackupMetadata(backup string, backupMetadata io.Reader
 	return r0
 }
 
+// PutBackupVolumeInfos provides a mock function with given fields: name, volumeInfo
+func (_m *BackupStore) PutBackupVolumeInfos(name string, volumeInfo io.Reader) error {
+	ret := _m.Called(name, volumeInfo)
+
+	if len(ret) == 0 {
+		panic("no return value specified for PutBackupVolumeInfos")
+	}
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(string, io.Reader) error); ok {
+		r0 = rf(name, volumeInfo)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
 // PutRestoreItemOperations provides a mock function with given fields: restore, restoreItemOperations
 func (_m *BackupStore) PutRestoreItemOperations(restore string, restoreItemOperations io.Reader) error {
 	ret := _m.Called(restore, restoreItemOperations)
 
+	if len(ret) == 0 {
+		panic("no return value specified for PutRestoreItemOperations")
+	}
+
 	var r0 error
 	if rf, ok := ret.Get(0).(func(string, io.Reader) error); ok {
 		r0 = rf(restore, restoreItemOperations)
@@ -508,6 +651,10 @@ func (_m *BackupStore) PutRestoreItemOperations(restore string, restoreItemOpera
 func (_m *BackupStore) PutRestoreLog(backup string, restore string, log io.Reader) error {
 	ret := _m.Called(backup, restore, log)
 
+	if len(ret) == 0 {
+		panic("no return value specified for PutRestoreLog")
+	}
+
 	var r0 error
 	if rf, ok := ret.Get(0).(func(string, string, io.Reader) error); ok {
 		r0 = rf(backup, restore, log)
@@ -518,13 +665,17 @@ func (_m *BackupStore) PutRestoreLog(backup string, restore string, log io.Reade
 	return r0
 }
 
-// PutRestoreResults provides a mock function with given fields: backup, restore, results
-func (_m *BackupStore) PutRestoreResults(backup string, restore string, results io.Reader) error {
-	ret := _m.Called(backup, restore, results)
+// PutRestoreResults provides a mock function with given fields: backup, restore, _a2
+func (_m *BackupStore) PutRestoreResults(backup string, restore string, _a2 io.Reader) error {
+	ret := _m.Called(backup, restore, _a2)
+
+	if len(ret) == 0 {
+		panic("no return value specified for PutRestoreResults")
+	}
 
 	var r0 error
 	if rf, ok := ret.Get(0).(func(string, string, io.Reader) error); ok {
-		r0 = rf(backup, restore, results)
+		r0 = rf(backup, restore, _a2)
 	} else {
 		r0 = ret.Error(0)
 	}
@@ -532,13 +683,17 @@ func (_m *BackupStore) PutRestoreResults(backup string, restore string, results
 	return r0
 }
 
-// PutRestoredResourceList provides a mock function with given fields: restore, results
-func (_m *BackupStore) PutRestoredResourceList(restore string, results io.Reader) error {
-	ret := _m.Called(restore, results)
+// PutRestoreVolumeInfo provides a mock function with given fields: restore, volumeInfo
+func (_m *BackupStore) PutRestoreVolumeInfo(restore string, volumeInfo io.Reader) error {
+	ret := _m.Called(restore, volumeInfo)
+
+	if len(ret) == 0 {
+		panic("no return value specified for PutRestoreVolumeInfo")
+	}
 
 	var r0 error
 	if rf, ok := ret.Get(0).(func(string, io.Reader) error); ok {
-		r0 = rf(restore, results)
+		r0 = rf(restore, volumeInfo)
 	} else {
 		r0 = ret.Error(0)
 	}
@@ -546,26 +701,30 @@ func (_m *BackupStore) PutRestoredResourceList(restore string, results io.Reader
 	return r0
 }
 
-// PutRestoreVolumeInfo provides a mock function with given fields: restore, results
-func (_m *BackupStore) 	PutRestoreVolumeInfo(restore string, results io.Reader) error {
-	ret := _m.Called(restore, results)
+// PutRestoredResourceList provides a mock function with given fields: restore, _a1
+func (_m *BackupStore) PutRestoredResourceList(restore string, _a1 io.Reader) error {
+	ret := _m.Called(restore, _a1)
+
+	if len(ret) == 0 {
+		panic("no return value specified for PutRestoredResourceList")
+	}
 
 	var r0 error
 	if rf, ok := ret.Get(0).(func(string, io.Reader) error); ok {
-		r0 = rf(restore, results)
+		r0 = rf(restore, _a1)
 	} else {
 		r0 = ret.Error(0)
 	}
 
 	return r0
 }
-type mockConstructorTestingTNewBackupStore interface {
-	mock.TestingT
-	Cleanup(func())
-}
 
 // NewBackupStore creates a new instance of BackupStore. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
-func NewBackupStore(t mockConstructorTestingTNewBackupStore) *BackupStore {
+// The first argument is typically a *testing.T value.
+func NewBackupStore(t interface {
+	mock.TestingT
+	Cleanup(func())
+}) *BackupStore {
 	mock := &BackupStore{}
 	mock.Mock.Test(t)
 

pkg/podvolume/backupper.go

@@ -48,6 +48,8 @@ type Backupper interface {
 	// BackupPodVolumes backs up all specified volumes in a pod.
 	BackupPodVolumes(backup *velerov1api.Backup, pod *corev1api.Pod, volumesToBackup []string, resPolicies *resourcepolicies.Policies, log logrus.FieldLogger) ([]*velerov1api.PodVolumeBackup, *PVCBackupSummary, []error)
 	WaitAllPodVolumesProcessed(log logrus.FieldLogger) []*velerov1api.PodVolumeBackup
+	GetPodVolumeBackup(namespace, name string) (*velerov1api.PodVolumeBackup, error)
+	ListPodVolumeBackupsByPod(podNamespace, podName string) ([]*velerov1api.PodVolumeBackup, error)
 }
 
 type backupper struct {
@@ -59,7 +61,10 @@ type backupper struct {
 	pvbInformer         ctrlcache.Informer
 	handlerRegistration cache.ResourceEventHandlerRegistration
 	wg                  sync.WaitGroup
-	result              []*velerov1api.PodVolumeBackup
+	// pvbIndexer holds all PVBs created by this backuper and is capable to search
+	// the PVBs based on specific properties quickly because of the embedded indexes.
+	// The statuses of the PVBs are got updated when Informer receives update events.
+	pvbIndexer cache.Indexer
 }
 
 type skippedPVC struct {
@@ -101,8 +106,22 @@ func (pbs *PVCBackupSummary) addSkipped(volumeName string, reason string) {
 	}
 }
 
+const indexNamePod = "POD"
+
+func podIndexFunc(obj interface{}) ([]string, error) {
+	pvb, ok := obj.(*velerov1api.PodVolumeBackup)
+	if !ok {
+		return nil, errors.Errorf("expected PodVolumeBackup, but got %T", obj)
+	}
+	if pvb == nil {
+		return nil, errors.New("PodVolumeBackup is nil")
+	}
+	return []string{cache.NewObjectName(pvb.Spec.Pod.Namespace, pvb.Spec.Pod.Name).String()}, nil
+}
+
 func newBackupper(
 	ctx context.Context,
+	log logrus.FieldLogger,
 	repoLocker *repository.RepoLocker,
 	repoEnsurer *repository.Ensurer,
 	pvbInformer ctrlcache.Informer,
@@ -118,13 +137,19 @@ func newBackupper(
 		uploaderType: uploaderType,
 		pvbInformer:  pvbInformer,
 		wg:           sync.WaitGroup{},
-		result:       []*velerov1api.PodVolumeBackup{},
+		pvbIndexer: cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{
+			indexNamePod: podIndexFunc,
+		}),
 	}
 
 	b.handlerRegistration, _ = pvbInformer.AddEventHandler(
 		cache.ResourceEventHandlerFuncs{
 			UpdateFunc: func(_, obj interface{}) {
-				pvb := obj.(*velerov1api.PodVolumeBackup)
+				pvb, ok := obj.(*velerov1api.PodVolumeBackup)
+				if !ok {
+					log.Errorf("expected PodVolumeBackup, but got %T", obj)
+					return
+				}
 
 				if pvb.GetLabels()[velerov1api.BackupUIDLabel] != string(backup.UID) {
 					return
@@ -135,7 +160,10 @@ func newBackupper(
 					return
 				}
 
-				b.result = append(b.result, pvb)
+				// the Indexer inserts PVB directly if the PVB to be updated doesn't exist
+				if err := b.pvbIndexer.Update(pvb); err != nil {
+					log.WithError(err).Errorf("failed to update PVB %s/%s in indexer", pvb.Namespace, pvb.Name)
+				}
 				b.wg.Done()
 			},
 		},
@@ -318,6 +346,12 @@ func (b *backupper) BackupPodVolumes(backup *velerov1api.Backup, pod *corev1api.
 			continue
 		}
 		b.wg.Add(1)
+
+		if err := b.pvbIndexer.Add(volumeBackup); err != nil {
+			errs = append(errs, errors.Wrapf(err, "failed to add PodVolumeBackup %s/%s to indexer", volumeBackup.Namespace, volumeBackup.Name))
+			continue
+		}
+
 		podVolumeBackups = append(podVolumeBackups, volumeBackup)
 		pvcSummary.addBackedup(volumeName)
 	}
@@ -343,7 +377,12 @@ func (b *backupper) WaitAllPodVolumesProcessed(log logrus.FieldLogger) []*velero
 	case <-b.ctx.Done():
 		log.Error("timed out waiting for all PodVolumeBackups to complete")
 	case <-done:
-		for _, pvb := range b.result {
+		for _, obj := range b.pvbIndexer.List() {
+			pvb, ok := obj.(*velerov1api.PodVolumeBackup)
+			if !ok {
+				log.Errorf("expected PodVolumeBackup, but got %T", obj)
+				continue
+			}
 			podVolumeBackups = append(podVolumeBackups, pvb)
 			if pvb.Status.Phase == velerov1api.PodVolumeBackupPhaseFailed {
 				log.Errorf("pod volume backup failed: %s", pvb.Status.Message)
@@ -353,6 +392,37 @@ func (b *backupper) WaitAllPodVolumesProcessed(log logrus.FieldLogger) []*velero
 	return podVolumeBackups
 }
 
+func (b *backupper) GetPodVolumeBackup(namespace, name string) (*velerov1api.PodVolumeBackup, error) {
+	obj, exist, err := b.pvbIndexer.GetByKey(cache.NewObjectName(namespace, name).String())
+	if err != nil {
+		return nil, err
+	}
+	if !exist {
+		return nil, nil
+	}
+	pvb, ok := obj.(*velerov1api.PodVolumeBackup)
+	if !ok {
+		return nil, errors.Errorf("expected PodVolumeBackup, but got %T", obj)
+	}
+	return pvb, nil
+}
+
+func (b *backupper) ListPodVolumeBackupsByPod(podNamespace, podName string) ([]*velerov1api.PodVolumeBackup, error) {
+	objs, err := b.pvbIndexer.ByIndex(indexNamePod, cache.NewObjectName(podNamespace, podName).String())
+	if err != nil {
+		return nil, err
+	}
+	var pvbs []*velerov1api.PodVolumeBackup
+	for _, obj := range objs {
+		pvb, ok := obj.(*velerov1api.PodVolumeBackup)
+		if !ok {
+			return nil, errors.Errorf("expected PodVolumeBackup, but got %T", obj)
+		}
+		pvbs = append(pvbs, pvb)
+	}
+	return pvbs, nil
+}
+
 func skipAllPodVolumes(pod *corev1api.Pod, volumesToBackup []string, err error, pvcSummary *PVCBackupSummary, log logrus.FieldLogger) {
 	for _, volumeName := range volumesToBackup {
 		log.WithError(err).Warnf("Skip pod volume %s", volumeName)

pkg/podvolume/backupper_factory.go

@@ -32,7 +32,7 @@ import (
 // BackupperFactory can construct pod volumes backuppers.
 type BackupperFactory interface {
 	// NewBackupper returns a pod volumes backupper for use during a single Velero backup.
-	NewBackupper(context.Context, *velerov1api.Backup, string) (Backupper, error)
+	NewBackupper(context.Context, logrus.FieldLogger, *velerov1api.Backup, string) (Backupper, error)
 }
 
 func NewBackupperFactory(
@@ -59,8 +59,8 @@ type backupperFactory struct {
 	log         logrus.FieldLogger
 }
 
-func (bf *backupperFactory) NewBackupper(ctx context.Context, backup *velerov1api.Backup, uploaderType string) (Backupper, error) {
-	b := newBackupper(ctx, bf.repoLocker, bf.repoEnsurer, bf.pvbInformer, bf.crClient, uploaderType, backup)
+func (bf *backupperFactory) NewBackupper(ctx context.Context, log logrus.FieldLogger, backup *velerov1api.Backup, uploaderType string) (Backupper, error) {
+	b := newBackupper(ctx, log, bf.repoLocker, bf.repoEnsurer, bf.pvbInformer, bf.crClient, uploaderType, backup)
 
 	if !cache.WaitForCacheSync(ctx.Done(), bf.pvbInformer.HasSynced) {
 		return nil, errors.New("timed out waiting for caches to sync")

pkg/podvolume/backupper_test.go

@@ -315,6 +315,7 @@ func TestBackupPodVolumes(t *testing.T) {
 	scheme := runtime.NewScheme()
 	velerov1api.AddToScheme(scheme)
 	corev1api.AddToScheme(scheme)
+	log := logrus.New()
 
 	tests := []struct {
 		name                  string
@@ -594,7 +595,7 @@ func TestBackupPodVolumes(t *testing.T) {
 			backupObj.Spec.StorageLocation = test.bsl
 
 			factory := NewBackupperFactory(repository.NewRepoLocker(), ensurer, fakeCtrlClient, pvbInformer, velerotest.NewLogger())
-			bp, err := factory.NewBackupper(ctx, backupObj, test.uploaderType)
+			bp, err := factory.NewBackupper(ctx, log, backupObj, test.uploaderType)
 
 			require.NoError(t, err)
 
@@ -619,6 +620,91 @@ func TestBackupPodVolumes(t *testing.T) {
 	}
 }
 
+func TestGetPodVolumeBackup(t *testing.T) {
+	backupper := &backupper{
+		pvbIndexer: cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{
+			indexNamePod: podIndexFunc,
+		}),
+	}
+
+	obj := &velerov1api.PodVolumeBackup{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "velero",
+			Name:      "pvb",
+		},
+		Spec: velerov1api.PodVolumeBackupSpec{
+			Pod: corev1api.ObjectReference{
+				Kind:      "Pod",
+				Namespace: "default",
+				Name:      "pod",
+			},
+		},
+	}
+
+	err := backupper.pvbIndexer.Add(obj)
+	require.NoError(t, err)
+
+	// not exist PVB
+	pvb, err := backupper.GetPodVolumeBackup("invalid-namespace", "invalid-name")
+	require.NoError(t, err)
+	assert.Nil(t, pvb)
+
+	// exist PVB
+	pvb, err = backupper.GetPodVolumeBackup("velero", "pvb")
+	require.NoError(t, err)
+	assert.NotNil(t, pvb)
+}
+
+func TestListPodVolumeBackupsByPodp(t *testing.T) {
+	backupper := &backupper{
+		pvbIndexer: cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{
+			indexNamePod: podIndexFunc,
+		}),
+	}
+
+	obj1 := &velerov1api.PodVolumeBackup{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "velero",
+			Name:      "pvb1",
+		},
+		Spec: velerov1api.PodVolumeBackupSpec{
+			Pod: corev1api.ObjectReference{
+				Kind:      "Pod",
+				Namespace: "default",
+				Name:      "pod",
+			},
+		},
+	}
+	obj2 := &velerov1api.PodVolumeBackup{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "velero",
+			Name:      "pvb2",
+		},
+		Spec: velerov1api.PodVolumeBackupSpec{
+			Pod: corev1api.ObjectReference{
+				Kind:      "Pod",
+				Namespace: "default",
+				Name:      "pod",
+			},
+		},
+	}
+
+	err := backupper.pvbIndexer.Add(obj1)
+	require.NoError(t, err)
+	err = backupper.pvbIndexer.Add(obj2)
+	require.NoError(t, err)
+
+	// not exist PVBs
+	pvbs, err := backupper.ListPodVolumeBackupsByPod("invalid-namespace", "invalid-name")
+	require.NoError(t, err)
+	assert.Empty(t, pvbs)
+
+	// exist PVBs
+	pvbs, err = backupper.ListPodVolumeBackupsByPod("default", "pod")
+	require.NoError(t, err)
+	assert.Len(t, pvbs, 2)
+}
+
 type logHook struct {
 	entry *logrus.Entry
 }
@@ -636,6 +722,7 @@ func TestWaitAllPodVolumesProcessed(t *testing.T) {
 	defer func() {
 		cancelFunc()
 	}()
+	log := logrus.New()
 	cases := []struct {
 		name              string
 		ctx               context.Context
@@ -691,7 +778,7 @@ func TestWaitAllPodVolumesProcessed(t *testing.T) {
 		logHook := &logHook{}
 		logger.Hooks.Add(logHook)
 
-		backuper := newBackupper(c.ctx, nil, nil, informer, nil, "", &velerov1api.Backup{})
+		backuper := newBackupper(c.ctx, log, nil, nil, informer, nil, "", &velerov1api.Backup{})
 		backuper.wg.Add(1)
 
 		if c.statusToBeUpdated != nil {

pkg/repository/udmrepo/repo_options.go

@@ -200,7 +200,12 @@ func GetRepoDomain() string {
 
 func getRepoConfigFile(workPath string, repoID string) string {
 	if workPath == "" {
-		workPath = filepath.Join(os.Getenv("HOME"), "udmrepo")
+		home := os.Getenv("HOME")
+		if home != "" {
+			workPath = filepath.Join(home, "udmrepo")
+		} else {
+			workPath = filepath.Join(os.TempDir(), "udmrepo")
+		}
 	}
 
 	name := "repo-" + strings.ToLower(repoID) + ".conf"

pkg/restore/actions/csi/pvc_action.go

@@ -478,6 +478,7 @@ func newDataDownload(
 			SnapshotID:            dataUploadResult.SnapshotID,
 			SourceNamespace:       dataUploadResult.SourceNamespace,
 			OperationTimeout:      backup.Spec.CSISnapshotTimeout,
+			NodeOS:                dataUploadResult.NodeOS,
 		},
 	}
 	if restore.Spec.UploaderConfig != nil {

pkg/restore/actions/dataupload_retrieve_action.go

@@ -78,6 +78,7 @@ func (d *DataUploadRetrieveAction) Execute(input *velero.RestoreItemActionExecut
 		SnapshotID:            dataUpload.Status.SnapshotID,
 		SourceNamespace:       dataUpload.Spec.SourceNamespace,
 		DataMoverResult:       dataUpload.Status.DataMoverResult,
+		NodeOS:                dataUpload.Status.NodeOS,
 	}
 
 	jsonBytes, err := json.Marshal(dataUploadResult)

pkg/restore/restore_test.go

@@ -1709,7 +1709,7 @@ func (a *pluggableAction) AreAdditionalItemsReady(additionalItems []velero.Resou
 }
 
 // TestRestoreActionModifications runs restores with restore item actions that modify resources, and
-// verifies that that the modified item is correctly created in the API. Verification is done by looking
+// verifies that the modified item is correctly created in the API. Verification is done by looking
 // at the full object in the API.
 func TestRestoreActionModifications(t *testing.T) {
 	// modifyingActionGetter is a helper function that returns a *pluggableAction, whose Execute(...)
@@ -2006,7 +2006,7 @@ func TestRestoreWithAsyncOperations(t *testing.T) {
 }
 
 // TestRestoreActionAdditionalItems runs restores with restore item actions that return additional items
-// to be restored, and verifies that that the correct set of items is created in the API. Verification is
+// to be restored, and verifies that the correct set of items is created in the API. Verification is
 // done by looking at the namespaces/names of the items in the API; contents are not checked.
 func TestRestoreActionAdditionalItems(t *testing.T) {
 	tests := []struct {

pkg/util/csi/volume_snapshot_test.go

@@ -1684,28 +1684,28 @@ func TestWaitUntilVSCHandleIsReady(t *testing.T) {
 			},
 		},
 		{
-			name:        "waitDisabled should not find volumesnapshotcontent volumesnapshot status is nil",
+			name:        "waitDisabled should not find volumesnapshotcontent when volumesnapshot status is nil",
 			wait:        false,
 			expectError: false,
 			exepctedVSC: nil,
 			volSnap:     vsWithNilStatus,
 		},
 		{
-			name:        "waitDisabled should not find volumesnapshotcontent volumesnapshot status.BoundVolumeSnapshotContentName is nil",
+			name:        "waitDisabled should not find volumesnapshotcontent when volumesnapshot status.BoundVolumeSnapshotContentName is nil",
 			wait:        false,
 			expectError: false,
 			exepctedVSC: nil,
 			volSnap:     vsWithNilStatusField,
 		},
 		{
-			name:        "waitDisabled should find volumesnapshotcontent volumesnapshotcontent status is nil",
+			name:        "waitDisabled should find volumesnapshotcontent when volumesnapshotcontent status is nil",
 			wait:        false,
 			expectError: false,
 			exepctedVSC: vscWithNilStatus,
 			volSnap:     vsForNilStatusVsc,
 		},
 		{
-			name:        "waitDisabled should find volumesnapshotcontent volumesnapshotcontent status.SnapshotHandle is nil",
+			name:        "waitDisabled should find volumesnapshotcontent when volumesnapshotcontent status.SnapshotHandle is nil",
 			wait:        false,
 			expectError: false,
 			exepctedVSC: vscWithNilStatusField,

pkg/util/kube/event.go

@@ -16,6 +16,7 @@ limitations under the License.
 package kube
 
 import (
+	"math"
 	"sync"
 	"time"
 
@@ -60,6 +61,9 @@ func NewEventRecorder(kubeClient kubernetes.Interface, scheme *runtime.Scheme, e
 	}
 
 	res.broadcaster = record.NewBroadcasterWithCorrelatorOptions(record.CorrelatorOptions{
+		// Bypass the built-in EventCorrelator's rate filtering, otherwise, the event will be abandoned if the rate exceeds.
+		// The callers (i.e., data mover pods) have controlled the rate and total number outside. E.g., the progress is designed to be updated every 10 seconds and is changeable.
+		BurstSize: math.MaxInt32,
 		MaxEvents: 1,
 		MessageFunc: func(event *v1.Event) string {
 			return event.Message

pkg/util/kube/event_test.go

@@ -43,9 +43,12 @@ func TestEvent(t *testing.T) {
 	}
 
 	cases := []struct {
-		name     string
-		events   []testEvent
-		expected int
+		name           string
+		events         []testEvent
+		generateDiff   int
+		generateSame   int
+		generateEnding bool
+		expected       int
 	}{
 		{
 			name: "update events, different message",
@@ -116,6 +119,18 @@ func TestEvent(t *testing.T) {
 			},
 			expected: -1,
 		},
+		{
+			name:           "auto generate 200",
+			generateDiff:   200,
+			generateEnding: true,
+			expected:       201,
+		},
+		{
+			name:           "auto generate 200, update",
+			generateSame:   200,
+			generateEnding: true,
+			expected:       2,
+		},
 	}
 
 	shutdownTimeout = time.Second * 5
@@ -143,6 +158,28 @@ func TestEvent(t *testing.T) {
 			_, err = client.CoreV1().Pods("fake-ns").Create(context.Background(), pod, metav1.CreateOptions{})
 			require.NoError(t, err)
 
+			for i := 0; i < tc.generateDiff; i++ {
+				tc.events = append(tc.events, testEvent{
+					reason:  fmt.Sprintf("fake-reason-%v", i),
+					message: fmt.Sprintf("fake-message-%v", i),
+				})
+			}
+
+			for i := 0; i < tc.generateSame; i++ {
+				tc.events = append(tc.events, testEvent{
+					reason:  "fake-reason",
+					message: fmt.Sprintf("fake-message-%v", i),
+				})
+			}
+
+			if tc.generateEnding {
+				tc.events = append(tc.events, testEvent{
+					reason:  "fake-ending-reason",
+					message: "fake-ending-message",
+					ending:  true,
+				})
+			}
+
 			for _, e := range tc.events {
 				if e.ending {
 					recorder.EndingEvent(pod, e.warning, e.reason, e.message)

pkg/util/kube/pvc_pv.go

@@ -271,7 +271,7 @@ func SetPVReclaimPolicy(ctx context.Context, pvGetter corev1client.CoreV1Interfa
 // nothing if the consuming doesn't affect the PV provision.
 // The latest PVC and the selected node will be returned.
 func WaitPVCConsumed(ctx context.Context, pvcGetter corev1client.CoreV1Interface, pvc string, namespace string,
-	storageClient storagev1.StorageV1Interface, timeout time.Duration) (string, *corev1api.PersistentVolumeClaim, error) {
+	storageClient storagev1.StorageV1Interface, timeout time.Duration, ignoreConsume bool) (string, *corev1api.PersistentVolumeClaim, error) {
 	selectedNode := ""
 	var updated *corev1api.PersistentVolumeClaim
 	var storageClass *storagev1api.StorageClass
@@ -282,18 +282,20 @@ func WaitPVCConsumed(ctx context.Context, pvcGetter corev1client.CoreV1Interface
 			return false, errors.Wrapf(err, "error to get pvc %s/%s", namespace, pvc)
 		}
 
-		if tmpPVC.Spec.StorageClassName != nil && storageClass == nil {
-			storageClass, err = storageClient.StorageClasses().Get(ctx, *tmpPVC.Spec.StorageClassName, metav1.GetOptions{})
-			if err != nil {
-				return false, errors.Wrapf(err, "error to get storage class %s", *tmpPVC.Spec.StorageClassName)
+		if !ignoreConsume {
+			if tmpPVC.Spec.StorageClassName != nil && storageClass == nil {
+				storageClass, err = storageClient.StorageClasses().Get(ctx, *tmpPVC.Spec.StorageClassName, metav1.GetOptions{})
+				if err != nil {
+					return false, errors.Wrapf(err, "error to get storage class %s", *tmpPVC.Spec.StorageClassName)
+				}
 			}
-		}
 
-		if storageClass != nil {
-			if storageClass.VolumeBindingMode != nil && *storageClass.VolumeBindingMode == storagev1api.VolumeBindingWaitForFirstConsumer {
-				selectedNode = tmpPVC.Annotations[KubeAnnSelectedNode]
-				if selectedNode == "" {
-					return false, nil
+			if storageClass != nil {
+				if storageClass.VolumeBindingMode != nil && *storageClass.VolumeBindingMode == storagev1api.VolumeBindingWaitForFirstConsumer {
+					selectedNode = tmpPVC.Annotations[KubeAnnSelectedNode]
+					if selectedNode == "" {
+						return false, nil
+					}
 				}
 			}
 		}

pkg/util/kube/pvc_pv_test.go

@@ -189,14 +189,15 @@ func TestWaitPVCConsumed(t *testing.T) {
 	}
 
 	tests := []struct {
-		name          string
-		pvcName       string
-		pvcNamespace  string
-		kubeClientObj []runtime.Object
-		kubeReactors  []reactor
-		expectedPVC   *corev1api.PersistentVolumeClaim
-		selectedNode  string
-		err           string
+		name                       string
+		pvcName                    string
+		pvcNamespace               string
+		kubeClientObj              []runtime.Object
+		kubeReactors               []reactor
+		expectedPVC                *corev1api.PersistentVolumeClaim
+		selectedNode               string
+		ignoreWaitForFirstConsumer bool
+		err                        string
 	}{
 		{
 			name:         "get pvc error",
@@ -213,6 +214,16 @@ func TestWaitPVCConsumed(t *testing.T) {
 			},
 			expectedPVC: pvcObject,
 		},
+		{
+			name:                       "success when ignore wait for first consumer",
+			pvcName:                    "fake-pvc-2",
+			pvcNamespace:               "fake-namespace",
+			ignoreWaitForFirstConsumer: true,
+			kubeClientObj: []runtime.Object{
+				pvcObjectWithSC,
+			},
+			expectedPVC: pvcObjectWithSC,
+		},
 		{
 			name:         "get sc fail",
 			pvcName:      "fake-pvc-2",
@@ -275,7 +286,7 @@ func TestWaitPVCConsumed(t *testing.T) {
 
 			var kubeClient kubernetes.Interface = fakeKubeClient
 
-			selectedNode, pvc, err := WaitPVCConsumed(context.Background(), kubeClient.CoreV1(), test.pvcName, test.pvcNamespace, kubeClient.StorageV1(), time.Millisecond)
+			selectedNode, pvc, err := WaitPVCConsumed(context.Background(), kubeClient.CoreV1(), test.pvcName, test.pvcNamespace, kubeClient.StorageV1(), time.Millisecond, test.ignoreWaitForFirstConsumer)
 
 			if err != nil {
 				assert.EqualError(t, err, test.err)

test/util/velero/velero_utils.go

@@ -786,19 +786,18 @@ func WaitForVSphereUploadCompletion(ctx context.Context, timeout time.Duration,
 		for _, curLine := range lines {
 			fmt.Printf("%s %s\n", curLine, time.Now().Format("2006-01-02 15:04:05"))
 			comps := strings.Split(curLine, "=")
-			// SnapshotPhase represents the lifecycle phase of a Snapshot.
-			// New - No work yet, next phase is InProgress
-			// InProgress - snapshot being taken
-			// Snapshotted - local snapshot complete, next phase is Protecting or SnapshotFailed
-			// SnapshotFailed - end state, snapshot was not able to be taken
-			// Uploading - snapshot is being moved to durable storage
-			// Uploaded - end state, snapshot has been protected
-			// UploadFailed - end state, unable to move to durable storage
-			// Canceling - when the SanpshotCancel flag is set, if the Snapshot has not already moved into a terminal state, the
-			//             status will move to Canceling.  The snapshot ID will be removed from the status status if has been filled in
-			//             and the snapshot ID will not longer be valid for a Clone operation
-			// Canceled - the operation was canceled, the snapshot ID is not valid
-			// Canceled - the operation was canceled, the snapshot ID is not valid
+			// SnapshotPhase represents the lifecycle phase of a Snapshot:
+			// * New: No work yet, next phase is InProgress
+			// * InProgress:     snapshot being taken
+			// * Snapshotted:    local snapshot complete, next phase is Protecting or SnapshotFailed
+			// * SnapshotFailed: end state, snapshot was not able to be taken
+			// * Uploading:      snapshot is being moved to durable storage
+			// * Uploaded:       end state, snapshot has been protected
+			// * UploadFailed:   end state, unable to move to durable storage
+			// * Canceling:      when the SanpshotCancel flag is set, if the Snapshot has not already moved into a terminal state, the
+			//                   status will move to Canceling.  The snapshot ID will be removed from the status if the status has been filled in
+			//                   and the snapshot ID will not longer be valid for a Clone operation
+			// * Canceled:       the operation was canceled, the snapshot ID is not valid
 			if len(comps) == 2 {
 				phase := comps[1]
 				actualCount++