Update the control path to make the in-place incremental restore with block data mover work E2E (#10410)

Update the control path to make the in-place incremental restore with block data mover work E2E

Signed-off-by: Wenkai Yin(尹文开) <yinw@vmware.com>

changelogs/unreleased/10331-chlins

@@ -1 +0,0 @@
-Preserve PVC selected-node annotation via carrier annotation for in-place restore

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

@@ -103,6 +103,10 @@ spec:
                     description: VolumeSnapshot is the name of the volume snapshot
                       to be backed up
                     type: string
+                  volumeSnapshotNamespace:
+                    description: VolumeSnapshotNamespace is the namespece of the volume
+                      snapshot to be backed up
+                    type: string
                 required:
                 - storageClass
                 - volumeSnapshot

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

@@ -110,6 +110,10 @@ spec:
                     description: VolumeSnapshot is the name of the volume snapshot
                       to be backed up
                     type: string
+                  volumeSnapshotNamespace:
+                    description: VolumeSnapshotNamespace is the namespece of the volume
+                      snapshot to be backed up
+                    type: string
                 required:
                 - storageClass
                 - volumeSnapshot

pkg/apis/velero/v1/restore_types.go

@@ -464,6 +464,14 @@ func (r *Restore) IsVolumeDataInplaceRestore() bool {
 	return r.Spec.ExistingVolumeDataPolicy == VolumeDataPolicyTypeFull || r.Spec.ExistingVolumeDataPolicy == VolumeDataPolicyTypeIncremental
 }
 
+func (r *Restore) IsVolumeDataInplaceFullRestore() bool {
+	return r.Spec.ExistingVolumeDataPolicy == VolumeDataPolicyTypeFull
+}
+
+func (r *Restore) IsVolumeDataInplaceIncrementalRestore() bool {
+	return r.Spec.ExistingVolumeDataPolicy == VolumeDataPolicyTypeIncremental
+}
+
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
 
 // RestoreList is a list of Restores.

pkg/apis/velero/v2alpha1/data_upload_types.go

@@ -80,6 +80,10 @@ const (
 
 // CSISnapshotSpec is the specification for a CSI snapshot.
 type CSISnapshotSpec struct {
+	// VolumeSnapshotNamespace is the namespece of the volume snapshot to be backed up
+	// +optional
+	VolumeSnapshotNamespace string `json:"volumeSnapshotNamespace"`
+
 	// VolumeSnapshot is the name of the volume snapshot to be backed up
 	VolumeSnapshot string `json:"volumeSnapshot"`
 

pkg/cmd/cli/datamover/restore.go

@@ -36,6 +36,7 @@ import (
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	velerov2alpha1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
 	"github.com/vmware-tanzu/velero/pkg/buildinfo"
+	"github.com/vmware-tanzu/velero/pkg/cbtservice"
 	"github.com/vmware-tanzu/velero/pkg/client"
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/signals"
 	"github.com/vmware-tanzu/velero/pkg/datamover"
@@ -56,6 +57,9 @@ type dataMoverRestoreConfig struct {
 	ddName          string
 	cacheDir        string
 	resourceTimeout time.Duration
+	cbtSAName       string
+	vsNamespace     string
+	volumeID        string
 }
 
 func NewRestoreCommand(f client.Factory) *cobra.Command {
@@ -96,6 +100,9 @@ func NewRestoreCommand(f client.Factory) *cobra.Command {
 	command.Flags().StringVar(&config.ddName, "data-download", config.ddName, "The data download name")
 	command.Flags().StringVar(&config.cacheDir, "cache-volume-path", config.cacheDir, "The full path of the cache volume")
 	command.Flags().DurationVar(&config.resourceTimeout, "resource-timeout", config.resourceTimeout, "How long to wait for resource processes which are not covered by other specific timeout parameters.")
+	command.Flags().StringVar(&config.cbtSAName, "cbt-sa-name", config.cbtSAName, "The name of the service account used by CSI's CBT service")
+	command.Flags().StringVar(&config.vsNamespace, "vs-namespace", config.vsNamespace, "The namespace of the VolumeSnapshot")
+	command.Flags().StringVar(&config.volumeID, "volume-id", config.volumeID, "The volume ID of the snapshot")
 
 	_ = command.MarkFlagRequired("volume-path")
 	_ = command.MarkFlagRequired("volume-mode")
@@ -116,6 +123,7 @@ type dataMoverRestore struct {
 	config      dataMoverRestoreConfig
 	kubeClient  kubernetes.Interface
 	dataPathMgr *datapath.Manager
+	cbtService  cbtservice.Service
 }
 
 func newdataMoverRestore(logger logrus.FieldLogger, factory client.Factory, config dataMoverRestoreConfig) (*dataMoverRestore, error) {
@@ -201,6 +209,12 @@ func newdataMoverRestore(logger logrus.FieldLogger, factory client.Factory, conf
 		config:     config,
 		namespace:  factory.Namespace(),
 		nodeName:   nodeName,
+		cbtService: cbtservice.NewService(
+			logger,
+			config.vsNamespace,
+			config.cbtSAName,
+			clientConfig,
+		),
 	}
 
 	s.kubeClient, err = factory.KubeClient()
@@ -294,5 +308,5 @@ func (s *dataMoverRestore) createDataPathService() (dataPathService, error) {
 	return datamover.NewRestoreMicroService(s.ctx, s.client, s.kubeClient, s.config.ddName, s.namespace, s.nodeName, datapath.AccessPoint{
 		ByPath:  s.config.volumePath,
 		VolMode: uploader.PersistentVolumeMode(s.config.volumeMode),
-	}, s.dataPathMgr, repoEnsurer, credGetter, duInformer, s.config.cacheDir, s.logger), nil
+	}, s.dataPathMgr, repoEnsurer, credGetter, duInformer, s.config.cacheDir, s.config.volumeID, s.cbtService, s.logger), nil
 }

pkg/cmd/cli/nodeagent/server.go

@@ -27,6 +27,7 @@ import (
 
 	"github.com/bombsimon/logrusr/v3"
 	"github.com/cockroachdb/errors"
+	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v8/apis/volumesnapshot/v1"
 	snapshotv1client "github.com/kubernetes-csi/external-snapshotter/client/v8/clientset/versioned"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/sirupsen/logrus"
@@ -175,6 +176,10 @@ func newNodeAgentServer(logger logrus.FieldLogger, factory client.Factory, confi
 		cancelFunc()
 		return nil, err
 	}
+	if err := snapshotv1api.AddToScheme(scheme); err != nil {
+		cancelFunc()
+		return nil, err
+	}
 
 	nodeName := os.Getenv("NODE_NAME")
 
@@ -484,6 +489,7 @@ func (s *nodeAgentServer) run() {
 		s.repoConfigMgr,
 		podLabels,
 		podAnnotations,
+		csiSnapshotMetadataServiceConfigs,
 	)
 
 	if err := dataDownloadReconciler.SetupWithManager(s.mgr); err != nil {

pkg/controller/data_download_controller.go

@@ -59,27 +59,28 @@ import (
 
 // DataDownloadReconciler reconciles a DataDownload object
 type DataDownloadReconciler struct {
-	client                client.Client
-	kubeClient            kubernetes.Interface
-	mgr                   manager.Manager
-	logger                logrus.FieldLogger
-	Clock                 clock.WithTickerAndDelayedExecution
-	restoreExposer        exposer.GenericRestoreExposer
-	nodeName              string
-	dataPathMgr           *datapath.Manager
-	vgdpCounter           *exposer.VgdpCounter
-	loadAffinity          []*kube.LoadAffinity
-	restorePVCConfig      velerotypes.RestorePVC
-	backupRepoConfigs     map[string]string
-	cacheVolumeConfigs    *velerotypes.CachePVC
-	podResources          corev1api.ResourceRequirements
-	preparingTimeout      time.Duration
-	metrics               *metrics.ServerMetrics
-	cancelledDataDownload sync.Map
-	dataMovePriorityClass string
-	repoConfigMgr         repository.ConfigManager
-	podLabels             map[string]string
-	podAnnotations        map[string]string
+	client                         client.Client
+	kubeClient                     kubernetes.Interface
+	mgr                            manager.Manager
+	logger                         logrus.FieldLogger
+	Clock                          clock.WithTickerAndDelayedExecution
+	restoreExposer                 exposer.GenericRestoreExposer
+	nodeName                       string
+	dataPathMgr                    *datapath.Manager
+	vgdpCounter                    *exposer.VgdpCounter
+	loadAffinity                   []*kube.LoadAffinity
+	restorePVCConfig               velerotypes.RestorePVC
+	backupRepoConfigs              map[string]string
+	cacheVolumeConfigs             *velerotypes.CachePVC
+	podResources                   corev1api.ResourceRequirements
+	preparingTimeout               time.Duration
+	metrics                        *metrics.ServerMetrics
+	cancelledDataDownload          sync.Map
+	dataMovePriorityClass          string
+	repoConfigMgr                  repository.ConfigManager
+	podLabels                      map[string]string
+	podAnnotations                 map[string]string
+	snapshotMetadataServiceConfigs *velerotypes.CSISnapshotMetadataService
 }
 
 func NewDataDownloadReconciler(
@@ -101,28 +102,30 @@ func NewDataDownloadReconciler(
 	repoConfigMgr repository.ConfigManager,
 	podLabels map[string]string,
 	podAnnotations map[string]string,
+	snapshotMetadataServiceConfigs *velerotypes.CSISnapshotMetadataService,
 ) *DataDownloadReconciler {
 	return &DataDownloadReconciler{
-		client:                client,
-		kubeClient:            kubeClient,
-		mgr:                   mgr,
-		logger:                logger.WithField("controller", "DataDownload"),
-		Clock:                 &clock.RealClock{},
-		nodeName:              nodeName,
-		restoreExposer:        exposer.NewGenericRestoreExposer(kubeClient, logger),
-		restorePVCConfig:      restorePVCConfig,
-		backupRepoConfigs:     backupRepoConfigs,
-		cacheVolumeConfigs:    cacheVolumeConfigs,
-		dataPathMgr:           dataPathMgr,
-		vgdpCounter:           counter,
-		loadAffinity:          loadAffinity,
-		podResources:          podResources,
-		preparingTimeout:      preparingTimeout,
-		metrics:               metrics,
-		dataMovePriorityClass: dataMovePriorityClass,
-		repoConfigMgr:         repoConfigMgr,
-		podLabels:             podLabels,
-		podAnnotations:        podAnnotations,
+		client:                         client,
+		kubeClient:                     kubeClient,
+		mgr:                            mgr,
+		logger:                         logger.WithField("controller", "DataDownload"),
+		Clock:                          &clock.RealClock{},
+		nodeName:                       nodeName,
+		restoreExposer:                 exposer.NewGenericRestoreExposer(kubeClient, client, logger),
+		restorePVCConfig:               restorePVCConfig,
+		backupRepoConfigs:              backupRepoConfigs,
+		cacheVolumeConfigs:             cacheVolumeConfigs,
+		dataPathMgr:                    dataPathMgr,
+		vgdpCounter:                    counter,
+		loadAffinity:                   loadAffinity,
+		podResources:                   podResources,
+		preparingTimeout:               preparingTimeout,
+		metrics:                        metrics,
+		dataMovePriorityClass:          dataMovePriorityClass,
+		repoConfigMgr:                  repoConfigMgr,
+		podLabels:                      podLabels,
+		podAnnotations:                 podAnnotations,
+		snapshotMetadataServiceConfigs: snapshotMetadataServiceConfigs,
 	}
 }
 
@@ -488,7 +491,9 @@ func (r *DataDownloadReconciler) OnDataDownloadCompleted(ctx context.Context, na
 	}
 
 	log.Info("Cleaning up exposed environment")
-	r.restoreExposer.CleanUp(ctx, objRef)
+	r.restoreExposer.CleanUp(ctx, objRef, &exposer.GenericRestoreCleanUpParam{
+		Snapshot: dd.Spec.CSISnapshot,
+	})
 
 	if err := UpdateDataDownloadWithRetry(ctx, r.client, types.NamespacedName{Namespace: dd.Namespace, Name: dd.Name}, log, func(dd *velerov2alpha1api.DataDownload) bool {
 		if isDataDownloadInFinalState(dd) {
@@ -537,7 +542,9 @@ func (r *DataDownloadReconciler) OnDataDownloadCancelled(ctx context.Context, na
 		return
 	}
 	// cleans up any objects generated during the snapshot expose
-	r.restoreExposer.CleanUp(ctx, getDataDownloadOwnerObject(&dd))
+	r.restoreExposer.CleanUp(ctx, getDataDownloadOwnerObject(&dd), &exposer.GenericRestoreCleanUpParam{
+		Snapshot: dd.Spec.CSISnapshot,
+	})
 
 	if err := UpdateDataDownloadWithRetry(ctx, r.client, types.NamespacedName{Namespace: dd.Namespace, Name: dd.Name}, log, func(dd *velerov2alpha1api.DataDownload) bool {
 		if isDataDownloadInFinalState(dd) {
@@ -587,7 +594,9 @@ func (r *DataDownloadReconciler) tryCancelDataDownload(ctx context.Context, dd *
 
 	// success update
 	r.metrics.RegisterDataDownloadCancel(r.nodeName)
-	r.restoreExposer.CleanUp(ctx, getDataDownloadOwnerObject(dd))
+	r.restoreExposer.CleanUp(ctx, getDataDownloadOwnerObject(dd), &exposer.GenericRestoreCleanUpParam{
+		Snapshot: dd.Spec.CSISnapshot,
+	})
 
 	log.Warn("data download is canceled")
 
@@ -735,7 +744,9 @@ func (r *DataDownloadReconciler) prepareDataDownload(ssb *velerov2alpha1api.Data
 
 func (r *DataDownloadReconciler) errorOut(ctx context.Context, dd *velerov2alpha1api.DataDownload, err error, msg string, log logrus.FieldLogger) (ctrl.Result, error) {
 	if r.restoreExposer != nil {
-		r.restoreExposer.CleanUp(ctx, getDataDownloadOwnerObject(dd))
+		r.restoreExposer.CleanUp(ctx, getDataDownloadOwnerObject(dd), &exposer.GenericRestoreCleanUpParam{
+			Snapshot: dd.Spec.CSISnapshot,
+		})
 	}
 	return ctrl.Result{}, r.updateStatusToFailed(ctx, dd, err, msg, log)
 }
@@ -825,7 +836,9 @@ func (r *DataDownloadReconciler) onPrepareTimeout(ctx context.Context, dd *veler
 		log.Warnf("[Diagnose DD expose]%s", diag)
 	}
 
-	r.restoreExposer.CleanUp(ctx, getDataDownloadOwnerObject(dd))
+	r.restoreExposer.CleanUp(ctx, getDataDownloadOwnerObject(dd), &exposer.GenericRestoreCleanUpParam{
+		Snapshot: dd.Spec.CSISnapshot,
+	})
 
 	log.Info("Datadownload has been cleaned up")
 
@@ -952,6 +965,10 @@ func (r *DataDownloadReconciler) setupExposeParam(dd *velerov2alpha1api.DataDown
 		RestoreSize:           dd.Spec.SnapshotSize,
 		CacheVolume:           cacheVolume,
 		DataMover:             dd.Spec.DataMover,
+		CSI: &exposer.GenericRestoreExposeCSI{
+			Snapshot:                       dd.Spec.CSISnapshot,
+			SnapshotMetadataServiceConfigs: r.snapshotMetadataServiceConfigs,
+		},
 	}, nil
 }
 

pkg/controller/data_download_controller_test.go

@@ -150,6 +150,7 @@ func initDataDownloadReconcilerWithError(t *testing.T, objects []any, needError
 		nil,
 		nil, // podLabels
 		nil, // podAnnotations
+		nil, // snapshotMetadataServiceConfigs
 	), nil
 }
 
@@ -585,7 +586,7 @@ func TestDataDownloadReconcile(t *testing.T) {
 						}
 
 						if !test.notMockCleanUp {
-							ep.On("CleanUp", mock.Anything, mock.Anything).Return()
+							ep.On("CleanUp", mock.Anything, mock.Anything, mock.Anything).Return()
 						}
 						return ep
 					}()
@@ -744,7 +745,7 @@ func TestOnDataDownloadCompleted(t *testing.T) {
 				} else {
 					ep.On("RebindVolume", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
 				}
-				ep.On("CleanUp", mock.Anything, mock.Anything).Return()
+				ep.On("CleanUp", mock.Anything, mock.Anything, mock.Anything).Return()
 				return ep
 			}()
 
@@ -1122,7 +1123,8 @@ func (dt *ddResumeTestHelper) RebindVolume(context.Context, corev1api.ObjectRefe
 	return nil
 }
 
-func (dt *ddResumeTestHelper) CleanUp(context.Context, corev1api.ObjectReference) {}
+func (dt *ddResumeTestHelper) CleanUp(context.Context, corev1api.ObjectReference, *exposer.GenericRestoreCleanUpParam) {
+}
 
 func (dt *ddResumeTestHelper) newMicroServiceBRWatcher(kbclient.Client, kubernetes.Interface, manager.Manager, string, string, string, string, string, string,
 	datapath.Callbacks, logrus.FieldLogger) datapath.AsyncBR {
@@ -1445,6 +1447,7 @@ func TestDataDownloadSetupExposeParam(t *testing.T) {
 				nil, // repoConfigMgr (unused when cacheVolumeConfigs is nil)
 				tt.args.customLabels,
 				tt.args.customAnnotations,
+				nil,
 			)
 
 			// Act

pkg/datamover/restore_micro_service.go

@@ -32,6 +32,7 @@ import (
 	"github.com/vmware-tanzu/velero/internal/credentials"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	velerov2alpha1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
+	"github.com/vmware-tanzu/velero/pkg/cbtservice"
 	"github.com/vmware-tanzu/velero/pkg/datapath"
 	"github.com/vmware-tanzu/velero/pkg/repository"
 	"github.com/vmware-tanzu/velero/pkg/uploader"
@@ -62,11 +63,14 @@ type RestoreMicroService struct {
 	ddHandler  cachetool.ResourceEventHandlerRegistration
 	nodeName   string
 	cacheDir   string
+
+	volumeID   string
+	cbtService cbtservice.Service
 }
 
 func NewRestoreMicroService(ctx context.Context, client client.Client, kubeClient kubernetes.Interface, dataDownloadName string, namespace string, nodeName string,
 	sourceTargetPath datapath.AccessPoint, dataPathMgr *datapath.Manager, repoEnsurer *repository.Ensurer, cred *credentials.CredentialGetter,
-	ddInformer cache.Informer, cacheDir string, log logrus.FieldLogger) *RestoreMicroService {
+	ddInformer cache.Informer, cacheDir string, volumeID string, cbtService cbtservice.Service, log logrus.FieldLogger) *RestoreMicroService {
 	return &RestoreMicroService{
 		ctx:              ctx,
 		client:           client,
@@ -82,6 +86,8 @@ func NewRestoreMicroService(ctx context.Context, client client.Client, kubeClien
 		resultSignal:     make(chan dataPathResult),
 		ddInformer:       ddInformer,
 		cacheDir:         cacheDir,
+		volumeID:         volumeID,
+		cbtService:       cbtService,
 	}
 }
 
@@ -180,9 +186,16 @@ func (r *RestoreMicroService) RunCancelableDataPath(ctx context.Context) (string
 	}
 	log.Info("Async br init")
 
-	if err := dp.StartRestore(dd.Spec.SnapshotID, r.sourceTargetPath, dd.Spec.DataMoverConfig, &datapath.RestoreStartParam{
+	param := &datapath.RestoreStartParam{
 		Incremental: dd.Spec.RestoreType == string(velerov1api.VolumeDataPolicyTypeIncremental),
-	}); err != nil {
+		CBTService:  r.cbtService,
+	}
+	if dd.Spec.CSISnapshot != nil {
+		param.VolumeSnapshotNamespace = dd.Spec.CSISnapshot.VolumeSnapshotNamespace
+		param.VolumeSnapshotName = dd.Spec.CSISnapshot.VolumeSnapshot
+		param.VolumeID = r.volumeID
+	}
+	if err := dp.StartRestore(dd.Spec.SnapshotID, r.sourceTargetPath, dd.Spec.DataMoverConfig, param); err != nil {
 		return "", errors.Wrap(err, "error starting data path restore")
 	}
 

pkg/datapath/data_path.go

@@ -63,7 +63,11 @@ type BackupStartParam struct {
 
 // RestoreStartParam define the input param for restore start
 type RestoreStartParam struct {
-	Incremental bool
+	Incremental             bool
+	VolumeSnapshotNamespace string
+	VolumeSnapshotName      string
+	VolumeID                string
+	CBTService              cbtservice.Service
 }
 
 type generalDataPath struct {
@@ -246,7 +250,14 @@ func (dp *generalDataPath) StartRestore(snapshotID string, target AccessPoint, u
 			dp.wgDataPath.Done()
 		}()
 
-		totalBytes, err := dp.uploaderProv.RunRestore(dp.ctx, snapshotID, target.ByPath, restoreParam.Incremental, provider.CBTParam{}, target.VolMode, uploaderConfigs, dp)
+		totalBytes, err := dp.uploaderProv.RunRestore(dp.ctx, snapshotID, target.ByPath, restoreParam.Incremental,
+			provider.CBTParam{
+				Source: cbtservice.SourceInfo{
+					Snapshot: restoreParam.VolumeSnapshotName,
+					VolumeID: restoreParam.VolumeID,
+				},
+				Service: restoreParam.CBTService,
+			}, target.VolMode, uploaderConfigs, dp)
 
 		if err == provider.ErrorCanceled {
 			dp.callbacks.OnCancelled(context.Background(), dp.namespace, dp.jobName)

pkg/exposer/csi_snapshot.go

@@ -20,7 +20,6 @@ import (
 	"context"
 	"fmt"
 	"maps"
-	"strings"
 	"time"
 
 	"github.com/cockroachdb/errors"
@@ -116,12 +115,6 @@ type CSISnapshotExposeWaitParam struct {
 	NodeName   string
 }
 
-type cbtInfo struct {
-	changeID   string
-	volumeID   string
-	snapshotID string
-}
-
 // NewCSISnapshotExposer create a new instance of CSI snapshot exposer
 func NewCSISnapshotExposer(kubeClient kubernetes.Interface, csiSnapshotClient snapshotter.SnapshotV1Interface, log logrus.FieldLogger) SnapshotExposer {
 	return &csiSnapshotExposer{
@@ -299,9 +292,9 @@ func (e *csiSnapshotExposer) Expose(ctx context.Context, ownerObject corev1api.O
 
 	affinity := kube.GetLoadAffinityByStorageClass(csiExposeParam.Affinity, backupPVCStorageClass, curLog)
 
-	var cbtInfo cbtInfo
+	var cbtInfo csi.CBTInfo
 	if csiExposeParam.DataMover == datamover.DataMoverTypeVeleroBlock {
-		cbtInfo, err = e.getCBTInfo(ctx, backupVS, backupVSC, csiExposeParam.SourcePVName)
+		cbtInfo, err = csi.GetCBTInfo(ctx, e.kubeClient, e.log, backupVS, backupVSC, csiExposeParam.SourcePVName)
 		if err != nil {
 			return errors.Wrap(err, "error to get CBT info")
 		}
@@ -341,49 +334,6 @@ func (e *csiSnapshotExposer) Expose(ctx context.Context, ownerObject corev1api.O
 	return nil
 }
 
-func (e *csiSnapshotExposer) getCBTInfo(ctx context.Context, vs *snapshotv1api.VolumeSnapshot, vsc *snapshotv1api.VolumeSnapshotContent, sourcePVName string) (cbtInfo, error) {
-	cbtInfo := cbtInfo{}
-	if vs == nil || vsc == nil {
-		return cbtInfo, errors.New("vs or vsc is nil")
-	}
-
-	cbtInfo.snapshotID = vs.Name
-
-	if vs.Annotations != nil &&
-		(vs.Annotations[util.VSphereCNSChangeIDAnno] != "" ||
-			vs.Annotations[util.VSphereCNSSnapshotAnno] != "") {
-		cbtInfo.changeID = vs.Annotations[util.VSphereCNSChangeIDAnno]
-
-		splitSnapshotAnno := strings.Split(vs.Annotations[util.VSphereCNSSnapshotAnno], "+")
-		if len(splitSnapshotAnno) >= 2 {
-			cbtInfo.volumeID = splitSnapshotAnno[0]
-		}
-
-		e.log.Debugf("volumeID %s and changeID %s are read from VKS annotations.", cbtInfo.volumeID, cbtInfo.changeID)
-	} else {
-		pv, err := e.kubeClient.CoreV1().PersistentVolumes().Get(ctx, sourcePVName, metav1.GetOptions{})
-		if err != nil {
-			return cbtInfo, fmt.Errorf("failed to get pv %s: %w", sourcePVName, err)
-		}
-
-		if vsc.Status != nil && vsc.Status.SnapshotHandle != nil {
-			cbtInfo.changeID = *vsc.Status.SnapshotHandle
-		}
-
-		if pv.Spec.CSI != nil && pv.Spec.CSI.VolumeHandle != "" {
-			cbtInfo.volumeID = pv.Spec.CSI.VolumeHandle
-		}
-
-		e.log.Debugf("volumeID %s and changeID %s are read from PV and VS's handles.", cbtInfo.volumeID, cbtInfo.changeID)
-	}
-
-	if cbtInfo.volumeID == "" {
-		return cbtInfo, fmt.Errorf("volumeID must not be empty for CBT")
-	}
-
-	return cbtInfo, nil
-}
-
 func (e *csiSnapshotExposer) GetExposed(ctx context.Context, ownerObject corev1api.ObjectReference, timeout time.Duration, param any) (*ExposeResult, error) {
 	exposeWaitParam := param.(*CSISnapshotExposeWaitParam)
 
@@ -720,7 +670,7 @@ func (e *csiSnapshotExposer) createBackupPod(
 	intoleratableNodes []string,
 	volumeTopology *corev1api.NodeSelector,
 	csiSnapshotMetadataServiceConfigs *velerotypes.CSISnapshotMetadataService,
-	cbtInfo *cbtInfo,
+	cbtInfo *csi.CBTInfo,
 ) (*corev1api.Pod, error) {
 	podName := ownerObject.Name
 
@@ -776,9 +726,9 @@ func (e *csiSnapshotExposer) createBackupPod(
 	}
 
 	if cbtInfo != nil {
-		args = append(args, fmt.Sprintf("--change-id=%s", cbtInfo.changeID))
-		args = append(args, fmt.Sprintf("--volume-id=%s", cbtInfo.volumeID))
-		args = append(args, fmt.Sprintf("--snapshot-id=%s", cbtInfo.snapshotID))
+		args = append(args, fmt.Sprintf("--change-id=%s", cbtInfo.ChangeID))
+		args = append(args, fmt.Sprintf("--volume-id=%s", cbtInfo.VolumeID))
+		args = append(args, fmt.Sprintf("--snapshot-id=%s", cbtInfo.SnapshotID))
 	}
 
 	args = append(args, podInfo.logFormatArgs...)

pkg/exposer/csi_snapshot_test.go

@@ -47,6 +47,7 @@ import (
 	velerotypes "github.com/vmware-tanzu/velero/pkg/types"
 	"github.com/vmware-tanzu/velero/pkg/util"
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
+	"github.com/vmware-tanzu/velero/pkg/util/csi"
 	"github.com/vmware-tanzu/velero/pkg/util/datamover"
 	"github.com/vmware-tanzu/velero/pkg/util/kube"
 )
@@ -2216,7 +2217,7 @@ func TestGetCBTInfo(t *testing.T) {
 		vsc           *snapshotv1api.VolumeSnapshotContent
 		pv            *corev1api.PersistentVolume
 		sourcePVName  string
-		want          cbtInfo
+		want          csi.CBTInfo
 		wantErrSubstr string
 	}{
 		{
@@ -2239,10 +2240,10 @@ func TestGetCBTInfo(t *testing.T) {
 			},
 			vsc:          &snapshotv1api.VolumeSnapshotContent{},
 			sourcePVName: "pv-ignored",
-			want: cbtInfo{
-				changeID:   "change-id-1",
-				volumeID:   "volume-id-1",
-				snapshotID: "vs-anno",
+			want: csi.CBTInfo{
+				ChangeID:   "change-id-1",
+				VolumeID:   "volume-id-1",
+				SnapshotID: "vs-anno",
 			},
 		},
 		{
@@ -2266,10 +2267,10 @@ func TestGetCBTInfo(t *testing.T) {
 				},
 			},
 			sourcePVName: "pv-1",
-			want: cbtInfo{
-				changeID:   "snapshot-handle-1",
-				volumeID:   "csi-volume-handle-1",
-				snapshotID: "vs-fallback",
+			want: csi.CBTInfo{
+				ChangeID:   "snapshot-handle-1",
+				VolumeID:   "csi-volume-handle-1",
+				SnapshotID: "vs-fallback",
 			},
 		},
 		{
@@ -2332,7 +2333,7 @@ func TestGetCBTInfo(t *testing.T) {
 				log:        logrus.StandardLogger(),
 			}
 
-			got, err := exposer.getCBTInfo(context.Background(), tc.vs, tc.vsc, tc.sourcePVName)
+			got, err := csi.GetCBTInfo(context.Background(), exposer.kubeClient, exposer.log, tc.vs, tc.vsc, tc.sourcePVName)
 
 			if tc.wantErrSubstr != "" {
 				if err == nil {
@@ -2347,8 +2348,8 @@ func TestGetCBTInfo(t *testing.T) {
 			if err != nil {
 				t.Fatalf("unexpected error: %v", err)
 			}
-			if got.changeID != tc.want.changeID || got.volumeID != tc.want.volumeID || got.snapshotID != tc.want.snapshotID {
-				t.Fatalf("unexpected cbtInfo, want %+v, got %+v", tc.want, got)
+			if got.ChangeID != tc.want.ChangeID || got.VolumeID != tc.want.VolumeID || got.SnapshotID != tc.want.SnapshotID {
+				t.Fatalf("unexpected CBTInfo, want %+v, got %+v", tc.want, got)
 			}
 		})
 	}

pkg/exposer/generic_restore.go

@@ -23,6 +23,7 @@ import (
 
 	"github.com/cockroachdb/errors"
 	"github.com/google/uuid"
+	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v8/apis/volumesnapshot/v1"
 	"github.com/sirupsen/logrus"
 	corev1api "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -31,13 +32,23 @@ import (
 	"k8s.io/client-go/kubernetes"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
+	velerov2alpha1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
 	"github.com/vmware-tanzu/velero/pkg/nodeagent"
 	velerotypes "github.com/vmware-tanzu/velero/pkg/types"
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
+	"github.com/vmware-tanzu/velero/pkg/util/csi"
 	"github.com/vmware-tanzu/velero/pkg/util/datamover"
 	"github.com/vmware-tanzu/velero/pkg/util/kube"
 )
 
+// GenericRestoreExposeCSI define the CSI specific input param for Generic Restore Expose
+type GenericRestoreExposeCSI struct {
+	// Snapshot is the CSI snapshot spec
+	Snapshot *velerov2alpha1api.CSISnapshotSpec
+	// SnapshotMetadataServiceConfigs is the config for CSI snapshot metadata service
+	SnapshotMetadataServiceConfigs *velerotypes.CSISnapshotMetadataService
+}
+
 // GenericRestoreExposeParam define the input param for Generic Restore Expose
 type GenericRestoreExposeParam struct {
 	// TargetPVCName is the target volume name to be restored
@@ -87,6 +98,9 @@ type GenericRestoreExposeParam struct {
 
 	// DataMover is the data mover type, e.g., velero-fs, velero-block
 	DataMover string
+
+	// SnapshotMetadataServiceConfigs is the config for CSI snapshot metadata service
+	CSI *GenericRestoreExposeCSI
 }
 
 // GenericRestoreRebindVolumeParam define the input param for Generic Restore Rebind Volume
@@ -104,6 +118,11 @@ type GenericRestoreRebindVolumeParam struct {
 	TargetFSType string
 }
 
+// GenericRestoreCleanUpParam define the input param for Generic Restore CleanUp
+type GenericRestoreCleanUpParam struct {
+	Snapshot *velerov2alpha1api.CSISnapshotSpec
+}
+
 // 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
@@ -127,19 +146,21 @@ type GenericRestoreExposer interface {
 	RebindVolume(context.Context, corev1api.ObjectReference, GenericRestoreRebindVolumeParam) error
 
 	// CleanUp cleans up any objects generated during the restore expose
-	CleanUp(context.Context, corev1api.ObjectReference)
+	CleanUp(context.Context, corev1api.ObjectReference, *GenericRestoreCleanUpParam)
 }
 
 // NewGenericRestoreExposer creates a new instance of generic restore exposer
-func NewGenericRestoreExposer(kubeClient kubernetes.Interface, log logrus.FieldLogger) GenericRestoreExposer {
+func NewGenericRestoreExposer(kubeClient kubernetes.Interface, ctrlClient client.Client, log logrus.FieldLogger) GenericRestoreExposer {
 	return &genericRestoreExposer{
 		kubeClient: kubeClient,
+		ctrlClient: ctrlClient,
 		log:        log,
 	}
 }
 
 type genericRestoreExposer struct {
 	kubeClient kubernetes.Interface
+	ctrlClient client.Client
 	log        logrus.FieldLogger
 }
 
@@ -260,6 +281,33 @@ func (e *genericRestoreExposer) Expose(ctx context.Context, ownerObject corev1ap
 	}()
 
 	curLog.Info("Creating restore pod")
+	var volumeID string
+	if param.CSI != nil && param.CSI.Snapshot != nil {
+		vs := &snapshotv1api.VolumeSnapshot{}
+		if err := e.ctrlClient.Get(ctx, client.ObjectKey{
+			Namespace: param.CSI.Snapshot.VolumeSnapshotNamespace,
+			Name:      param.CSI.Snapshot.VolumeSnapshot,
+		}, vs); err != nil {
+			return errors.Wrapf(err, "error to get volume snapshot %s/%s", param.CSI.Snapshot.VolumeSnapshotNamespace, param.CSI.Snapshot.VolumeSnapshot)
+		}
+
+		vsc, err := csi.GetVSCForVS(ctx, vs, e.ctrlClient)
+		if err != nil {
+			return errors.Wrapf(err, "error to get volume snapshot content for volume snapshot %s/%s", vs.Namespace, vs.Name)
+		}
+
+		var cbtInfo csi.CBTInfo
+		cbtInfo, err = csi.GetCBTInfo(ctx, e.kubeClient, e.log, vs, vsc, param.TargetPVName)
+		if err != nil {
+			return errors.Wrap(err, "error to get CBT info")
+		}
+		curLog.Debugf("CBT info: %+v", cbtInfo)
+		volumeID = cbtInfo.VolumeID
+	}
+	var csiSnapshotMetadataServiceConfigs *velerotypes.CSISnapshotMetadataService
+	if param.CSI != nil {
+		csiSnapshotMetadataServiceConfigs = param.CSI.SnapshotMetadataServiceConfigs
+	}
 	restorePod, err := e.createRestorePod(
 		ctx,
 		ownerObject,
@@ -274,6 +322,9 @@ func (e *genericRestoreExposer) Expose(ctx context.Context, ownerObject corev1ap
 		affinity,
 		param.PriorityClassName,
 		cachePVC,
+		param.TargetNamespace,
+		volumeID,
+		csiSnapshotMetadataServiceConfigs,
 	)
 	if err != nil {
 		return errors.Wrapf(err, "error to create restore pod")
@@ -440,7 +491,7 @@ func (e *genericRestoreExposer) DiagnoseExpose(ctx context.Context, ownerObject
 	return diag
 }
 
-func (e *genericRestoreExposer) CleanUp(ctx context.Context, ownerObject corev1api.ObjectReference) {
+func (e *genericRestoreExposer) CleanUp(ctx context.Context, ownerObject corev1api.ObjectReference, param *GenericRestoreCleanUpParam) {
 	restorePodName := ownerObject.Name
 	restorePVCName := ownerObject.Name
 	cachePVCName := getCachePVCName(ownerObject)
@@ -453,6 +504,11 @@ func (e *genericRestoreExposer) CleanUp(ctx context.Context, ownerObject corev1a
 		BackupPVCSecretLabel, string(ownerObject.UID), e.log)
 	kube.DeleteConfigMapsWithLabel(ctx, e.kubeClient.CoreV1(), ownerObject.Namespace,
 		BackupPVCSecretLabel, string(ownerObject.UID), e.log)
+
+	if param.Snapshot != nil {
+		kube.EnsureDeleteVolumeSnapshotIfAny(ctx, e.ctrlClient, param.Snapshot.VolumeSnapshotNamespace,
+			param.Snapshot.VolumeSnapshot, 0, e.log)
+	}
 }
 
 func (e *genericRestoreExposer) RebindVolume(ctx context.Context, ownerObject corev1api.ObjectReference, param GenericRestoreRebindVolumeParam) error {
@@ -666,6 +722,9 @@ func (e *genericRestoreExposer) createRestorePod(
 	affinity *kube.LoadAffinity,
 	priorityClassName string,
 	cachePVC *corev1api.PersistentVolumeClaim,
+	volumeSnapshotNamespace string,
+	volumeID string,
+	csiSnapshotMetadataServiceConfigs *velerotypes.CSISnapshotMetadataService,
 ) (*corev1api.Pod, error) {
 	restorePodName := ownerObject.Name
 	restorePVCName := ownerObject.Name
@@ -746,6 +805,14 @@ func (e *genericRestoreExposer) createRestorePod(
 		fmt.Sprintf("--cache-volume-path=%s", cacheVolumePath),
 	}
 
+	if len(volumeID) > 0 {
+		args = append(args, fmt.Sprintf("--vs-namespace=%s", volumeSnapshotNamespace))
+		args = append(args, fmt.Sprintf("--volume-id=%s", volumeID))
+	}
+	if csiSnapshotMetadataServiceConfigs != nil && csiSnapshotMetadataServiceConfigs.SAName != "" {
+		args = append(args, fmt.Sprintf("--cbt-sa-name=%s", csiSnapshotMetadataServiceConfigs.SAName))
+	}
+
 	args = append(args, podInfo.logFormatArgs...)
 	args = append(args, podInfo.logLevelArgs...)
 

pkg/exposer/generic_restore_priority_test.go

@@ -149,6 +149,9 @@ func TestCreateRestorePodWithPriorityClass(t *testing.T) {
 				nil, // affinity
 				tc.expectedPriorityClass,
 				nil,
+				"", // volumeSnapshotNamespace
+				"", // volumeID
+				nil,
 			)
 
 			require.NoError(t, err, tc.description)
@@ -229,6 +232,9 @@ func TestCreateRestorePodWithMissingConfigMap(t *testing.T) {
 		nil, // affinity
 		"",  // empty priority class since config map is missing
 		nil,
+		"", // volumeSnapshotNamespace
+		"", // volumeID
+		nil,
 	)
 
 	// Should succeed even when config map is missing

pkg/exposer/generic_restore_test.go

@@ -1675,6 +1675,9 @@ func TestCreateRestorePod(t *testing.T) {
 				test.affinity,
 				"", // priority class name
 				nil,
+				"", // volumeSnapshotNamespace
+				"", // volumeID
+				nil,
 			)
 
 			require.NoError(t, err)

pkg/exposer/mocks/GenericRestoreExposer.go

@@ -42,8 +42,8 @@ func (_m *GenericRestoreExposer) EXPECT() *GenericRestoreExposer_Expecter {
 }
 
 // CleanUp provides a mock function for the type GenericRestoreExposer
-func (_mock *GenericRestoreExposer) CleanUp(context1 context.Context, objectReference v1.ObjectReference) {
-	_mock.Called(context1, objectReference)
+func (_mock *GenericRestoreExposer) CleanUp(context1 context.Context, objectReference v1.ObjectReference, param *exposer.GenericRestoreCleanUpParam) {
+	_mock.Called(context1, objectReference, param)
 	return
 }
 
@@ -55,11 +55,12 @@ type GenericRestoreExposer_CleanUp_Call struct {
 // CleanUp is a helper method to define mock.On call
 //   - context1 context.Context
 //   - objectReference v1.ObjectReference
-func (_e *GenericRestoreExposer_Expecter) CleanUp(context1 interface{}, objectReference interface{}) *GenericRestoreExposer_CleanUp_Call {
-	return &GenericRestoreExposer_CleanUp_Call{Call: _e.mock.On("CleanUp", context1, objectReference)}
+//   - param *exposer.GenericRestoreCleanUpParam
+func (_e *GenericRestoreExposer_Expecter) CleanUp(context1 interface{}, objectReference interface{}, param interface{}) *GenericRestoreExposer_CleanUp_Call {
+	return &GenericRestoreExposer_CleanUp_Call{Call: _e.mock.On("CleanUp", context1, objectReference, param)}
 }
 
-func (_c *GenericRestoreExposer_CleanUp_Call) Run(run func(context1 context.Context, objectReference v1.ObjectReference)) *GenericRestoreExposer_CleanUp_Call {
+func (_c *GenericRestoreExposer_CleanUp_Call) Run(run func(context1 context.Context, objectReference v1.ObjectReference, param *exposer.GenericRestoreCleanUpParam)) *GenericRestoreExposer_CleanUp_Call {
 	_c.Call.Run(func(args mock.Arguments) {
 		var arg0 context.Context
 		if args[0] != nil {
@@ -69,9 +70,14 @@ func (_c *GenericRestoreExposer_CleanUp_Call) Run(run func(context1 context.Cont
 		if args[1] != nil {
 			arg1 = args[1].(v1.ObjectReference)
 		}
+		var arg2 *exposer.GenericRestoreCleanUpParam
+		if args[2] != nil {
+			arg2 = args[2].(*exposer.GenericRestoreCleanUpParam)
+		}
 		run(
 			arg0,
 			arg1,
+			arg2,
 		)
 	})
 	return _c
@@ -82,7 +88,7 @@ func (_c *GenericRestoreExposer_CleanUp_Call) Return() *GenericRestoreExposer_Cl
 	return _c
 }
 
-func (_c *GenericRestoreExposer_CleanUp_Call) RunAndReturn(run func(context1 context.Context, objectReference v1.ObjectReference)) *GenericRestoreExposer_CleanUp_Call {
+func (_c *GenericRestoreExposer_CleanUp_Call) RunAndReturn(run func(context1 context.Context, objectReference v1.ObjectReference, param *exposer.GenericRestoreCleanUpParam)) *GenericRestoreExposer_CleanUp_Call {
 	_c.Run(run)
 	return _c
 }

pkg/restore/actions/csi/pvc_action.go

@@ -22,9 +22,9 @@ import (
 	"fmt"
 	"time"
 
-	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v8/apis/volumesnapshot/v1"
-
 	"github.com/cockroachdb/errors"
+	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v8/apis/volumesnapshot/v1"
+	snapshotter "github.com/kubernetes-csi/external-snapshotter/client/v8/clientset/versioned/typed/volumesnapshot/v1"
 	"github.com/sirupsen/logrus"
 	corev1api "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -47,6 +47,8 @@ import (
 	uploaderUtil "github.com/vmware-tanzu/velero/pkg/uploader/util"
 	"github.com/vmware-tanzu/velero/pkg/util"
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
+	"github.com/vmware-tanzu/velero/pkg/util/csi"
+	"github.com/vmware-tanzu/velero/pkg/util/datamover"
 	"github.com/vmware-tanzu/velero/pkg/util/kube"
 )
 
@@ -57,9 +59,10 @@ const (
 
 // pvcRestoreItemAction is a restore item action plugin for Velero
 type pvcRestoreItemAction struct {
-	log        logrus.FieldLogger
-	crClient   crclient.Client
-	kubeClient kubernetes.Interface
+	log               logrus.FieldLogger
+	crClient          crclient.Client
+	kubeClient        kubernetes.Interface
+	csiSnapshotClient snapshotter.SnapshotV1Interface
 }
 
 // AppliesTo returns information indicating that the
@@ -197,23 +200,16 @@ func (p *pvcRestoreItemAction) executeWithoutDataMove(logger *logrus.Entry, inpu
 	}, nil
 }
 
-func (p *pvcRestoreItemAction) executeWithDataMove(logger *logrus.Entry, input *velero.RestoreItemActionExecuteInput, backup *velerov1api.Backup, pvcExists bool, existingPVC, pvc, pvcFromBackup *corev1api.PersistentVolumeClaim) (*velero.RestoreItemActionExecuteOutput, error) {
+func (p *pvcRestoreItemAction) executeWithDataMove(logger *logrus.Entry, input *velero.RestoreItemActionExecuteInput, backup *velerov1api.Backup, pvcExists bool, existingPVC, pvc, pvcFromBackup *corev1api.PersistentVolumeClaim) (out *velero.RestoreItemActionExecuteOutput, err error) {
+	ctx := context.Background()
 	var existingPV *corev1api.PersistentVolume
-	var err error
-	if pvcExists {
-		// If PVC already exists and is not in-place restore, returns early.
-		if !input.Restore.IsVolumeDataInplaceRestore() {
-			logger.Warnf("PVC already exists and ExistingVolumeDataPolicy is not in-place restore. Skip restore this PVC.")
-			return &velero.RestoreItemActionExecuteOutput{
-				UpdatedItem: input.Item,
-			}, nil
-		}
 
-		// the existing PVC should be deleted here rather than in the Exposer, otherwise the target PVC cannot be restored
-		existingPV, err = p.prepareForInplaceRestore(context.Background(), logger, pvc, existingPVC, backup.Spec.CSISnapshotTimeout.Duration)
-		if err != nil {
-			return nil, errors.WithStack(err)
-		}
+	// If PVC already exists and is not in-place restore, returns early.
+	if pvcExists && !input.Restore.IsVolumeDataInplaceRestore() {
+		logger.Warnf("PVC already exists and ExistingVolumeDataPolicy is not in-place restore. Skip restore this PVC.")
+		return &velero.RestoreItemActionExecuteOutput{
+			UpdatedItem: input.Item,
+		}, nil
 	}
 
 	logger.Info("Start DataMover restore.")
@@ -228,6 +224,41 @@ func (p *pvcRestoreItemAction) executeWithDataMove(logger *logrus.Entry, input *
 		}, nil
 	}
 
+	var dataUploadResult *velerov2alpha1.DataUploadResult
+	dataUploadResult, err = getDataUploadResult(ctx, input.Restore, pvc, p.crClient)
+	if err != nil {
+		return nil, errors.Wrapf(err, "fail get DataUploadResult for restore: %s", input.Restore.Name)
+	}
+
+	var volumeSnapshot *snapshotv1api.VolumeSnapshot
+	restoreType := input.Restore.Spec.ExistingVolumeDataPolicy
+	if pvcExists {
+		if existingPVC.Status.Phase != corev1api.ClaimBound {
+			return nil, errors.New("ExistingVolumeDataPolicy is in-place restore, but the existing PVC is not bound.")
+		}
+		// take a CSI snapshot of the existing PVC as the baseline of CBT
+		if input.Restore.IsVolumeDataInplaceIncrementalRestore() && datamover.IsVeleroBlockDataMover(dataUploadResult.DataMover) {
+			logger.Info("ExistingVolumeDataPolicy is in-place incremental restore and data mover is velero-block. Taking a CSI snapshot of the existing PVC as the baseline of CBT...")
+			volumeSnapshot, err = p.createVolumeSnapshot(ctx, logger, input.Restore, *existingPVC, dataUploadResult.SnapshotClass, backup.Spec.CSISnapshotTimeout.Duration)
+			if err != nil {
+				logger.Warnf("fail to create VolumeSnapshot for existing PVC %s/%s: %s, fallback to in-place full restore", existingPVC.Namespace, existingPVC.Name, err.Error())
+				restoreType = velerov1api.VolumeDataPolicyTypeFull
+			} else {
+				defer func() {
+					if err != nil {
+						csi.CleanupVolumeSnapshot(ctx, volumeSnapshot, p.crClient, logger)
+					}
+				}()
+			}
+		}
+
+		// delete the existing PVC, otherwise the target PVC cannot be restored
+		existingPV, err = p.deleteExistingPVC(ctx, logger, pvc, existingPVC, backup.Spec.CSISnapshotTimeout.Duration)
+		if err != nil {
+			return nil, errors.WithStack(err)
+		}
+	}
+
 	operationID := label.GetValidName(
 		string(velerov1api.AsyncOperationIDPrefixDataDownload) +
 			string(input.Restore.UID) + "." + string(pvcFromBackup.UID))
@@ -240,9 +271,10 @@ func (p *pvcRestoreItemAction) executeWithDataMove(logger *logrus.Entry, input *
 		newNamespace = pvc.Namespace
 	}
 
-	dataDownload, err := restoreFromDataUploadResult(
-		context.Background(), input.Restore, backup, pvc, existingPV, newNamespace,
-		operationID, p.crClient)
+	var dataDownload *velerov2alpha1.DataDownload
+	dataDownload, err = restoreFromDataUploadResult(
+		context.Background(), dataUploadResult, input.Restore, backup, pvc, existingPV, newNamespace,
+		operationID, string(restoreType), volumeSnapshot, p.crClient)
 	if err != nil {
 		logger.Errorf("Fail to restore from DataUploadResult: %s", err.Error())
 		return nil, errors.WithStack(err)
@@ -250,7 +282,8 @@ func (p *pvcRestoreItemAction) executeWithDataMove(logger *logrus.Entry, input *
 	logger.Infof("DataDownload %s/%s is created successfully.",
 		dataDownload.Namespace, dataDownload.Name)
 
-	unstructuredPVC, err := runtime.DefaultUnstructuredConverter.ToUnstructured(pvc)
+	var unstructuredPVC map[string]any
+	unstructuredPVC, err = runtime.DefaultUnstructuredConverter.ToUnstructured(pvc)
 	if err != nil {
 		return nil, errors.WithStack(err)
 	}
@@ -468,7 +501,8 @@ func newDataDownload(
 	dataUploadResult *velerov2alpha1.DataUploadResult,
 	pvc *corev1api.PersistentVolumeClaim,
 	pv *corev1api.PersistentVolume,
-	newNamespace, operationID string,
+	newNamespace, operationID, restoreType string,
+	volumeSnapshot *snapshotv1api.VolumeSnapshot,
 ) *velerov2alpha1.DataDownload {
 	pvName := ""
 	if pv != nil {
@@ -511,31 +545,32 @@ func newDataDownload(
 			SourceNamespace:       dataUploadResult.SourceNamespace,
 			OperationTimeout:      backup.Spec.CSISnapshotTimeout,
 			NodeOS:                dataUploadResult.NodeOS,
+			RestoreType:           restoreType,
 		},
 	}
+	if volumeSnapshot != nil {
+		dataDownload.Spec.CSISnapshot = &velerov2alpha1.CSISnapshotSpec{
+			VolumeSnapshot:          volumeSnapshot.Name,
+			VolumeSnapshotNamespace: volumeSnapshot.Namespace,
+		}
+	}
 	if restore.Spec.UploaderConfig != nil {
 		dataDownload.Spec.DataMoverConfig = uploaderUtil.StoreRestoreConfig(restore.Spec.UploaderConfig)
 	}
-	if restore.IsVolumeDataInplaceRestore() {
-		dataDownload.Spec.RestoreType = string(restore.Spec.ExistingVolumeDataPolicy)
-	}
 	return dataDownload
 }
 
 func restoreFromDataUploadResult(
 	ctx context.Context,
+	dataUploadResult *velerov2alpha1.DataUploadResult,
 	restore *velerov1api.Restore,
 	backup *velerov1api.Backup,
 	pvc *corev1api.PersistentVolumeClaim,
 	pv *corev1api.PersistentVolume,
-	newNamespace, operationID string,
+	newNamespace, operationID, restoreType string,
+	volumeSnapshot *snapshotv1api.VolumeSnapshot,
 	crClient crclient.Client,
 ) (*velerov2alpha1.DataDownload, error) {
-	dataUploadResult, err := getDataUploadResult(ctx, restore, pvc, crClient)
-	if err != nil {
-		return nil, errors.Wrapf(err, "fail get DataUploadResult for restore: %s",
-			restore.Name)
-	}
 	pvc.Spec.VolumeName = ""
 	if pvc.Spec.Selector == nil {
 		pvc.Spec.Selector = &metav1.LabelSelector{}
@@ -555,8 +590,10 @@ func restoreFromDataUploadResult(
 		pv,
 		newNamespace,
 		operationID,
+		restoreType,
+		volumeSnapshot,
 	)
-	err = crClient.Create(ctx, dataDownload)
+	err := crClient.Create(ctx, dataDownload)
 	if err != nil {
 		return nil, errors.Wrapf(err, "fail to create DataDownload")
 	}
@@ -592,11 +629,7 @@ func (p *pvcRestoreItemAction) isResourceExist(
 	return false, nil, errors.Wrapf(err, "fail to get PVC %s in namespace %s", pvc.Name, targetNamespace)
 }
 
-func (p *pvcRestoreItemAction) prepareForInplaceRestore(ctx context.Context, logger *logrus.Entry, targetPVC *corev1api.PersistentVolumeClaim, existingPVC *corev1api.PersistentVolumeClaim, operationTimeout time.Duration) (*corev1api.PersistentVolume, error) {
-	if existingPVC.Status.Phase != corev1api.ClaimBound {
-		return nil, errors.New("ExistingVolumeDataPolicy is in-place restore, but the existing PVC is not bound.")
-	}
-
+func (p *pvcRestoreItemAction) deleteExistingPVC(ctx context.Context, logger *logrus.Entry, targetPVC *corev1api.PersistentVolumeClaim, existingPVC *corev1api.PersistentVolumeClaim, operationTimeout time.Duration) (*corev1api.PersistentVolume, error) {
 	// Capture the "selected-node" annotation from the existing PVC before it is deleted below,
 	// and carry it on the target PVC via a Velero-internal carrier annotation. The restore
 	// engine translates the carrier back to the Kubernetes "selected-node" annotation after
@@ -637,6 +670,59 @@ func (p *pvcRestoreItemAction) prepareForInplaceRestore(ctx context.Context, log
 	return pv, nil
 }
 
+func (p *pvcRestoreItemAction) createVolumeSnapshot(ctx context.Context, logger *logrus.Entry, restore *velerov1api.Restore, pvc corev1api.PersistentVolumeClaim, vsClass string, operationTimeout time.Duration) (vs *snapshotv1api.VolumeSnapshot, err error) {
+	logger.Infof("creating VolumeSnapshot for PVC %s/%s with VolumeSnapshotClass %s", pvc.Namespace, pvc.Name, vsClass)
+
+	labels := map[string]string{
+		velerov1api.RestoreNameLabel: label.GetValidName(restore.Name),
+	}
+	for k, v := range pvc.ObjectMeta.Labels {
+		labels[k] = v
+	}
+
+	vs = &snapshotv1api.VolumeSnapshot{
+		ObjectMeta: metav1.ObjectMeta{
+			GenerateName: "velero-" + pvc.Name + "-",
+			Namespace:    pvc.Namespace,
+			Labels:       labels,
+		},
+		Spec: snapshotv1api.VolumeSnapshotSpec{
+			Source: snapshotv1api.VolumeSnapshotSource{
+				PersistentVolumeClaimName: &pvc.Name,
+			},
+			VolumeSnapshotClassName: &vsClass,
+		},
+	}
+
+	if err := p.crClient.Create(ctx, vs); err != nil {
+		return nil, errors.Wrapf(err, "failed to create the VolumeSnapshot for PVC %s/%s", pvc.Namespace, pvc.Name)
+	}
+
+	logger.Infof("VolumeSnapshot %s for PVC %s/%s created", vs.Name, pvc.Namespace, pvc.Name)
+	vsName := vs.Name
+	vsNamespace := vs.Namespace
+
+	_, err = csi.WaitUntilVSCHandleIsReady(vs, p.crClient, logger, operationTimeout)
+	if err != nil {
+		csi.CleanupVolumeSnapshot(ctx, vs, p.crClient, logger)
+		return nil, errors.Wrapf(err, "failed to wait for VolumeSnapshotContent of VolumeSnapshot %s/%s to be ready within timeout %v",
+			vsNamespace, vsName, operationTimeout)
+	}
+
+	var updatedVS *snapshotv1api.VolumeSnapshot
+	updatedVS, err = csi.WaitVolumeSnapshotReady(ctx, p.csiSnapshotClient, vs.Name, vs.Namespace, operationTimeout, logger)
+	if err != nil {
+		csi.CleanupVolumeSnapshot(ctx, vs, p.crClient, logger)
+		return nil, errors.Wrapf(err, "failed to wait for VolumeSnapshot %s/%s to become Ready within timeout %v",
+			vsNamespace, vsName, operationTimeout)
+	}
+	vs = updatedVS
+
+	logger.Infof("VolumeSnapshot %s for PVC %s/%s is ready to use", vs.Name, pvc.Namespace, pvc.Name)
+
+	return vs, nil
+}
+
 func NewPvcRestoreItemAction(f client.Factory) plugincommon.HandlerInitializer {
 	return func(logger logrus.FieldLogger) (any, error) {
 		crClient, err := f.KubebuilderClient()
@@ -649,10 +735,20 @@ func NewPvcRestoreItemAction(f client.Factory) plugincommon.HandlerInitializer {
 			return nil, err
 		}
 
+		clientConfig, err := f.ClientConfig()
+		if err != nil {
+			return nil, err
+		}
+		csiSnapshotClient, err := snapshotter.NewForConfig(clientConfig)
+		if err != nil {
+			return nil, err
+		}
+
 		return &pvcRestoreItemAction{
-			log:        logger,
-			crClient:   crClient,
-			kubeClient: kubeClient,
+			log:               logger,
+			crClient:          crClient,
+			kubeClient:        kubeClient,
+			csiSnapshotClient: csiSnapshotClient,
 		}, nil
 	}
 }

pkg/restore/actions/csi/pvc_action_test.go

@@ -36,6 +36,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/validation"
 	"k8s.io/client-go/kubernetes/fake"
+	"k8s.io/client-go/rest"
 	crclient "sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/vmware-tanzu/velero/pkg/apis/velero/shared"
@@ -511,6 +512,27 @@ func TestExecute(t *testing.T) {
 				return d
 			}(),
 		},
+		{
+			name:             "PVC exists and in-place incremental restore set, createVolumeSnapshot fails",
+			backup:           builder.ForBackup("velero", "testBackup").SnapshotMoveData(true).Result(),
+			restore:          builder.ForRestore("velero", "testRestore").Backup("testBackup").ExistingVolumeDataPolicy(string(velerov1api.VolumeDataPolicyTypeIncremental)).ItemOperationTimeout(time.Minute * 10).ObjectMeta(builder.WithUID("uid")).Result(),
+			pvc:              builder.ForPersistentVolumeClaim("velero", "testPVC").VolumeName("testPV").Phase(corev1api.ClaimBound).ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotLabel, "vsName", velerov1api.VolumeSnapshotRestoreSize, "10Gi", velerov1api.DataUploadNameAnnotation, "velero/")).Result(),
+			pv:               builder.ForPersistentVolume("testPV").ReclaimPolicy(corev1api.PersistentVolumeReclaimRetain).Result(),
+			dataUploadResult: builder.ForConfigMap("velero", "testCM").Data("uid", "{\"DataMover\":\"velero-block\", \"SnapshotClass\":\"test-snapclass\"}").ObjectMeta(builder.WithLabels(velerov1api.RestoreUIDLabel, "uid", velerov1api.PVCNamespaceNameLabel, "velero.testPVC", velerov1api.ResourceUsageLabel, label.GetValidName(string(velerov1api.VeleroResourceUsageDataUploadResult)))).Result(),
+			preCreatePVC:     true,
+			kubeClientObj: []runtime.Object{
+				builder.ForPersistentVolumeClaim("velero", "testPVC").VolumeName("testPV").Phase(corev1api.ClaimBound).ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotLabel, "vsName", velerov1api.VolumeSnapshotRestoreSize, "10Gi", velerov1api.DataUploadNameAnnotation, "velero/")).Result(),
+			},
+			expectedDataDownload: func() *velerov2alpha1.DataDownload {
+				d := builder.ForDataDownload("velero", "name").TargetVolume(velerov2alpha1.TargetVolumeSpec{PVC: "testPVC", Namespace: "velero", PV: "testPV"}).
+					ObjectMeta(builder.WithOwnerReference([]metav1.OwnerReference{{APIVersion: velerov1api.SchemeGroupVersion.String(), Kind: "Restore", Name: "testRestore", UID: "uid", Controller: boolptr.True()}}),
+						builder.WithLabelsMap(map[string]string{velerov1api.AsyncOperationIDLabel: "dd-uid.", velerov1api.RestoreNameLabel: "testRestore", velerov1api.RestoreUIDLabel: "uid"}),
+						builder.WithGenerateName("testRestore-")).Result()
+				d.Spec.RestoreType = "full"
+				d.Spec.DataMover = "velero-block"
+				return d
+			}(),
+		},
 	}
 
 	for _, tc := range tests {
@@ -639,7 +661,7 @@ func TestPrepareForInplaceRestoreSelectedNode(t *testing.T) {
 			}
 
 			targetPVC := builder.ForPersistentVolumeClaim("ns-1", "pvc-1").Result()
-			returnedPV, err := pvcRIA.prepareForInplaceRestore(
+			returnedPV, err := pvcRIA.deleteExistingPVC(
 				t.Context(), logrus.New().WithField("test", tc.name),
 				targetPVC, tc.existingPVC, time.Minute)
 			require.NoError(t, err)
@@ -749,6 +771,7 @@ func TestNewPvcRestoreItemAction(t *testing.T) {
 	f1 := &factorymocks.Factory{}
 	f1.On("KubebuilderClient").Return(crClient, nil)
 	f1.On("KubeClient").Return(nil, nil)
+	f1.On("ClientConfig").Return(&rest.Config{}, nil)
 	plugin1 := NewPvcRestoreItemAction(f1)
 	_, err1 := plugin1(logger)
 	require.NoError(t, err1)

pkg/uploader/provider/block_test.go

@@ -412,7 +412,7 @@ func TestBlockProviderCancelThroughWrappedError(t *testing.T) {
 	t.Run("restore", func(t *testing.T) {
 		orig := blockRestoreFunc
 		defer func() { blockRestoreFunc = orig }()
-		blockRestoreFunc = func(_ context.Context, _ block.Uploader, _ udmrepo.BackupRepo, _ string, _ string, _ map[string]string, _ logrus.FieldLogger) (int64, error) {
+		blockRestoreFunc = func(_ context.Context, _ block.Uploader, _ udmrepo.BackupRepo, _ string, _ string, _ bool, _ cbtservice.SourceInfo, _ cbtservice.Service, _ map[string]string, _ logrus.FieldLogger) (int64, error) {
 			return 0, errors.Wrap(block.ErrCanceled, "error restoring bdev")
 		}
 
@@ -422,7 +422,7 @@ func TestBlockProviderCancelThroughWrappedError(t *testing.T) {
 			log:           logrus.New(),
 		}
 
-		_, err := bp.RunRestore(t.Context(), "snap-1", "/dev/sda",
+		_, err := bp.RunRestore(t.Context(), "snap-1", "/dev/sda", false, CBTParam{},
 			uploader.PersistentVolumeBlock, map[string]string{}, &blockMockProgressUpdater{})
 
 		require.ErrorIs(t, err, ErrorCanceled)

pkg/util/csi/cbt.go

@@ -0,0 +1,80 @@
+/*
+Copyright The Velero Contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package csi
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"github.com/cockroachdb/errors"
+	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v8/apis/volumesnapshot/v1"
+	"github.com/sirupsen/logrus"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+
+	"github.com/vmware-tanzu/velero/pkg/util"
+)
+
+// CBTInfo define the info for CBT
+type CBTInfo struct {
+	ChangeID   string
+	VolumeID   string
+	SnapshotID string
+}
+
+// GetCBTInfo returns the CBT info for a snapshot
+func GetCBTInfo(ctx context.Context, kubeClient kubernetes.Interface, log logrus.FieldLogger, vs *snapshotv1api.VolumeSnapshot, vsc *snapshotv1api.VolumeSnapshotContent, sourcePVName string) (CBTInfo, error) {
+	cbtInfo := CBTInfo{}
+	if vs == nil || vsc == nil {
+		return cbtInfo, errors.New("vs or vsc is nil")
+	}
+
+	cbtInfo.SnapshotID = vs.Name
+
+	if vs.Annotations != nil &&
+		(vs.Annotations[util.VSphereCNSChangeIDAnno] != "" ||
+			vs.Annotations[util.VSphereCNSSnapshotAnno] != "") {
+		cbtInfo.ChangeID = vs.Annotations[util.VSphereCNSChangeIDAnno]
+
+		splitSnapshotAnno := strings.Split(vs.Annotations[util.VSphereCNSSnapshotAnno], "+")
+		if len(splitSnapshotAnno) >= 2 {
+			cbtInfo.VolumeID = splitSnapshotAnno[0]
+		}
+		log.Debugf("volumeID %s and changeID %s are read from VKS annotations.", cbtInfo.VolumeID, cbtInfo.ChangeID)
+	} else {
+		pv, err := kubeClient.CoreV1().PersistentVolumes().Get(ctx, sourcePVName, metav1.GetOptions{})
+		if err != nil {
+			return cbtInfo, fmt.Errorf("failed to get pv %s: %w", sourcePVName, err)
+		}
+
+		if vsc.Status != nil && vsc.Status.SnapshotHandle != nil {
+			cbtInfo.ChangeID = *vsc.Status.SnapshotHandle
+		}
+
+		if pv.Spec.CSI != nil && pv.Spec.CSI.VolumeHandle != "" {
+			cbtInfo.VolumeID = pv.Spec.CSI.VolumeHandle
+		}
+		log.Debugf("volumeID %s and changeID %s are read from PV and VS's handles.", cbtInfo.VolumeID, cbtInfo.ChangeID)
+	}
+
+	if cbtInfo.VolumeID == "" {
+		return cbtInfo, fmt.Errorf("volumeID must not be empty for CBT")
+	}
+
+	return cbtInfo, nil
+}

pkg/util/kube/pvc_pv.go

@@ -35,6 +35,7 @@ import (
 	corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
 	crclient "sigs.k8s.io/controller-runtime/pkg/client"
 
+	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v8/apis/volumesnapshot/v1"
 	storagev1api "k8s.io/api/storage/v1"
 	storagev1 "k8s.io/client-go/kubernetes/typed/storage/v1"
 )
@@ -138,6 +139,47 @@ func DeletePVIfAny(ctx context.Context, pvGetter corev1client.CoreV1Interface, p
 	}
 }
 
+// EnsureDeleteVolumeSnapshotIfAny deletes a VolumeSnapshot by namespace and name if it exists, and log an error when the deletion fails
+func EnsureDeleteVolumeSnapshotIfAny(ctx context.Context, client crclient.Client, namespace, name string, ensureTimeout time.Duration, log logrus.FieldLogger) {
+	if err := client.Delete(ctx, &snapshotv1api.VolumeSnapshot{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: namespace,
+		},
+	}); err != nil && !apierrors.IsNotFound(err) {
+		log.WithError(err).Errorf("Failed to delete the VolumeSnapshot %s/%s", namespace, name)
+	}
+
+	if ensureTimeout == 0 {
+		return
+	}
+
+	var updated *snapshotv1api.VolumeSnapshot
+	err := wait.PollUntilContextTimeout(ctx, waitInternal, ensureTimeout, true, func(ctx context.Context) (bool, error) {
+		if err := client.Get(ctx, types.NamespacedName{Namespace: namespace, Name: name}, updated); err != nil {
+			if apierrors.IsNotFound(err) {
+				return true, nil
+			}
+
+			return false, errors.Wrapf(err, "error to get VolumeSnapshot %s/%s", namespace, name)
+		}
+
+		return false, nil
+	})
+
+	if err != nil {
+		if errors.Is(err, context.DeadlineExceeded) {
+			if updated == nil {
+				log.WithError(err).Errorf("Timeout to assure VolumeSnapshot %s/%s is deleted", namespace, name)
+			} else {
+				log.WithError(err).Errorf("Timeout to assure VolumeSnapshot %s/%s is deleted, finalizers in VolumeSnapshot %v", namespace, name, updated.Finalizers)
+			}
+		} else {
+			log.WithError(err).Errorf("Error to assure VolumeSnapshot %s/%s is deleted", namespace, name)
+		}
+	}
+}
+
 // EnsureDeletePVC asserts the existence of a PVC by name, deletes it and waits for its disappearance and returns errors on any failure
 // If timeout is 0, it doesn't wait and return nil
 func EnsureDeletePVC(ctx context.Context, pvcGetter corev1client.CoreV1Interface, pvcName string, namespace string, timeout time.Duration) error {