From fd1940ad12e9b693ca02a550777ccb1266b9e8f0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Wenkai=20Yin=28=E5=B0=B9=E6=96=87=E5=BC=80=29?= Date: Wed, 26 Aug 2026 15:19:31 +0800 Subject: [PATCH] Update the control path to make the in-place incremental restore with block data mover work E2E (#10410) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Update the control path to make the in-place incremental restore with block data mover work E2E Signed-off-by: Wenkai Yin(尹文开) --- changelogs/unreleased/10331-chlins | 1 - .../bases/velero.io_datadownloads.yaml | 4 + .../v2alpha1/bases/velero.io_datauploads.yaml | 4 + pkg/apis/velero/v1/restore_types.go | 8 + pkg/apis/velero/v2alpha1/data_upload_types.go | 4 + pkg/cmd/cli/datamover/restore.go | 16 +- pkg/cmd/cli/nodeagent/server.go | 6 + pkg/controller/data_download_controller.go | 109 ++++++----- .../data_download_controller_test.go | 9 +- pkg/datamover/restore_micro_service.go | 19 +- pkg/datapath/data_path.go | 15 +- pkg/exposer/csi_snapshot.go | 62 +----- pkg/exposer/csi_snapshot_test.go | 25 +-- pkg/exposer/generic_restore.go | 73 ++++++- pkg/exposer/generic_restore_priority_test.go | 6 + pkg/exposer/generic_restore_test.go | 3 + pkg/exposer/mocks/GenericRestoreExposer.go | 18 +- pkg/restore/actions/csi/pvc_action.go | 182 +++++++++++++----- pkg/restore/actions/csi/pvc_action_test.go | 25 ++- pkg/uploader/provider/block_test.go | 4 +- pkg/util/csi/cbt.go | 80 ++++++++ pkg/util/kube/pvc_pv.go | 42 ++++ 22 files changed, 536 insertions(+), 179 deletions(-) delete mode 100644 changelogs/unreleased/10331-chlins create mode 100644 pkg/util/csi/cbt.go diff --git a/changelogs/unreleased/10331-chlins b/changelogs/unreleased/10331-chlins deleted file mode 100644 index b4bb1b1e3..000000000 --- a/changelogs/unreleased/10331-chlins +++ /dev/null @@ -1 +0,0 @@ -Preserve PVC selected-node annotation via carrier annotation for in-place restore diff --git a/config/crd/v2alpha1/bases/velero.io_datadownloads.yaml b/config/crd/v2alpha1/bases/velero.io_datadownloads.yaml index d3313eedc..71e662fe8 100644 --- a/config/crd/v2alpha1/bases/velero.io_datadownloads.yaml +++ b/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 diff --git a/config/crd/v2alpha1/bases/velero.io_datauploads.yaml b/config/crd/v2alpha1/bases/velero.io_datauploads.yaml index 8d03da279..a3d7dbe80 100644 --- a/config/crd/v2alpha1/bases/velero.io_datauploads.yaml +++ b/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 diff --git a/pkg/apis/velero/v1/restore_types.go b/pkg/apis/velero/v1/restore_types.go index 58b0dc423..312781e2a 100644 --- a/pkg/apis/velero/v1/restore_types.go +++ b/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. diff --git a/pkg/apis/velero/v2alpha1/data_upload_types.go b/pkg/apis/velero/v2alpha1/data_upload_types.go index 6f28d399b..db4c8d3a8 100644 --- a/pkg/apis/velero/v2alpha1/data_upload_types.go +++ b/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"` diff --git a/pkg/cmd/cli/datamover/restore.go b/pkg/cmd/cli/datamover/restore.go index ed6867e96..6b112a248 100644 --- a/pkg/cmd/cli/datamover/restore.go +++ b/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 } diff --git a/pkg/cmd/cli/nodeagent/server.go b/pkg/cmd/cli/nodeagent/server.go index 287e45591..c1442aab0 100644 --- a/pkg/cmd/cli/nodeagent/server.go +++ b/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 { diff --git a/pkg/controller/data_download_controller.go b/pkg/controller/data_download_controller.go index 568a19617..d8062bc72 100644 --- a/pkg/controller/data_download_controller.go +++ b/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 } diff --git a/pkg/controller/data_download_controller_test.go b/pkg/controller/data_download_controller_test.go index 4cc7a0a96..72d51167b 100644 --- a/pkg/controller/data_download_controller_test.go +++ b/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 diff --git a/pkg/datamover/restore_micro_service.go b/pkg/datamover/restore_micro_service.go index 373cc4946..7711fc503 100644 --- a/pkg/datamover/restore_micro_service.go +++ b/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") } diff --git a/pkg/datapath/data_path.go b/pkg/datapath/data_path.go index f07027ba3..1e7ae948e 100644 --- a/pkg/datapath/data_path.go +++ b/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) diff --git a/pkg/exposer/csi_snapshot.go b/pkg/exposer/csi_snapshot.go index 30e299380..a5639537c 100644 --- a/pkg/exposer/csi_snapshot.go +++ b/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...) diff --git a/pkg/exposer/csi_snapshot_test.go b/pkg/exposer/csi_snapshot_test.go index 7849e0d24..688c439a9 100644 --- a/pkg/exposer/csi_snapshot_test.go +++ b/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) } }) } diff --git a/pkg/exposer/generic_restore.go b/pkg/exposer/generic_restore.go index ace8b57fa..b19720389 100644 --- a/pkg/exposer/generic_restore.go +++ b/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...) diff --git a/pkg/exposer/generic_restore_priority_test.go b/pkg/exposer/generic_restore_priority_test.go index 642e0cc43..c8ca784ee 100644 --- a/pkg/exposer/generic_restore_priority_test.go +++ b/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 diff --git a/pkg/exposer/generic_restore_test.go b/pkg/exposer/generic_restore_test.go index 891d8adeb..c08c16b60 100644 --- a/pkg/exposer/generic_restore_test.go +++ b/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) diff --git a/pkg/exposer/mocks/GenericRestoreExposer.go b/pkg/exposer/mocks/GenericRestoreExposer.go index a1d8943d4..30639b6a8 100644 --- a/pkg/exposer/mocks/GenericRestoreExposer.go +++ b/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 } diff --git a/pkg/restore/actions/csi/pvc_action.go b/pkg/restore/actions/csi/pvc_action.go index 9498e63bd..a14b985a7 100644 --- a/pkg/restore/actions/csi/pvc_action.go +++ b/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 } } diff --git a/pkg/restore/actions/csi/pvc_action_test.go b/pkg/restore/actions/csi/pvc_action_test.go index d6350652b..47e8937a1 100644 --- a/pkg/restore/actions/csi/pvc_action_test.go +++ b/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) diff --git a/pkg/uploader/provider/block_test.go b/pkg/uploader/provider/block_test.go index e5044d536..970fc7cf6 100644 --- a/pkg/uploader/provider/block_test.go +++ b/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) diff --git a/pkg/util/csi/cbt.go b/pkg/util/csi/cbt.go new file mode 100644 index 000000000..00342996d --- /dev/null +++ b/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 +} diff --git a/pkg/util/kube/pvc_pv.go b/pkg/util/kube/pvc_pv.go index 1bdb0c44d..49d0bbc60 100644 --- a/pkg/util/kube/pvc_pv.go +++ b/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 {