From 0bc06323bf400f8b17956b6da60af5c4360be162 Mon Sep 17 00:00:00 2001 From: Xun Jiang Date: Thu, 28 May 2026 17:47:33 +0800 Subject: [PATCH] Support change-id and volume-id in backup workflow. * Add change-id and volume-id retrieve logic for both vks and vanilla k8s environment. * Add change-id and volume-id support code in exposer. Signed-off-by: Xun Jiang --- changelogs/unreleased/9863-blackpiglet | 1 + pkg/backup/actions/csi/pvc_action.go | 10 +- pkg/cbtservice/csi_service_impl.go | 4 +- pkg/cbtservice/csi_service_impl_test.go | 2 +- pkg/cmd/cli/datamover/backup.go | 34 ++- pkg/controller/data_upload_controller.go | 10 +- pkg/controller/data_upload_controller_test.go | 31 ++- pkg/datamover/backup_micro_service.go | 12 +- pkg/datamover/backup_micro_service_test.go | 22 +- pkg/datapath/data_path.go | 24 +- pkg/exposer/csi_snapshot.go | 66 ++++++ pkg/exposer/csi_snapshot_priority_test.go | 2 + pkg/exposer/csi_snapshot_test.go | 208 +++++++++++++++++- pkg/uploader/provider/kopia.go | 3 +- pkg/util/third_party.go | 2 + 15 files changed, 393 insertions(+), 38 deletions(-) create mode 100644 changelogs/unreleased/9863-blackpiglet diff --git a/changelogs/unreleased/9863-blackpiglet b/changelogs/unreleased/9863-blackpiglet new file mode 100644 index 000000000..49bae8d36 --- /dev/null +++ b/changelogs/unreleased/9863-blackpiglet @@ -0,0 +1 @@ +Support change-id and volume-id in backup workflow. \ No newline at end of file diff --git a/pkg/backup/actions/csi/pvc_action.go b/pkg/backup/actions/csi/pvc_action.go index 073ea4965..66c14b820 100644 --- a/pkg/backup/actions/csi/pvc_action.go +++ b/pkg/backup/actions/csi/pvc_action.go @@ -22,8 +22,6 @@ import ( "strconv" "time" - "k8s.io/client-go/util/retry" - "github.com/cockroachdb/errors" volumegroupsnapshotv1beta2 "github.com/kubernetes-csi/external-snapshotter/client/v8/apis/volumegroupsnapshot/v1beta2" snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v8/apis/volumesnapshot/v1" @@ -31,6 +29,7 @@ import ( corev1api "k8s.io/api/core/v1" storagev1api "k8s.io/api/storage/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/labels" @@ -39,11 +38,10 @@ import ( "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/wait" _ "k8s.io/client-go/plugin/pkg/client/auth/gcp" + "k8s.io/client-go/util/retry" crclient "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" - "k8s.io/apimachinery/pkg/api/resource" - velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1" velerov2alpha1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1" veleroclient "github.com/vmware-tanzu/velero/pkg/client" @@ -160,7 +158,7 @@ func (p *pvcBackupItemAction) getOrCreateVolumeHelper(backup *velerov1api.Backup return p.getVolumeHelperWithCache(backup) } -func (p *pvcBackupItemAction) validatePVCandPV( +func (p *pvcBackupItemAction) validatePVCAndPV( pvc corev1api.PersistentVolumeClaim, item runtime.Unstructured, ) ( @@ -304,7 +302,7 @@ func (p *pvcBackupItemAction) Execute( return nil, nil, "", nil, errors.WithStack(err) } - valid, item, fsType, err := p.validatePVCandPV( + valid, item, fsType, err := p.validatePVCAndPV( pvc, item, ) diff --git a/pkg/cbtservice/csi_service_impl.go b/pkg/cbtservice/csi_service_impl.go index 8918d36fa..4d0ea3fca 100644 --- a/pkg/cbtservice/csi_service_impl.go +++ b/pkg/cbtservice/csi_service_impl.go @@ -111,8 +111,8 @@ func (s *ServiceImpl) GetChangedBlocks(ctx context.Context, snapshot string, cha } args := iterator.Args{ - SnapshotName: snapshot, - PrevSnapshotName: changeID, + SnapshotName: snapshot, + PrevSnapshotID: changeID, Emitter: &emitterImpl{ logger: s.logger, recordCallBack: record, diff --git a/pkg/cbtservice/csi_service_impl_test.go b/pkg/cbtservice/csi_service_impl_test.go index 6ecad0850..cb6b311a9 100644 --- a/pkg/cbtservice/csi_service_impl_test.go +++ b/pkg/cbtservice/csi_service_impl_test.go @@ -234,7 +234,7 @@ func TestServiceImplGetChangedBlocks(t *testing.T) { require.NoError(t, err) assert.Equal(t, "snap-2", capturedArgs.SnapshotName) - assert.Equal(t, "snap-1", capturedArgs.PrevSnapshotName) + assert.Equal(t, "snap-1", capturedArgs.PrevSnapshotID) assert.Equal(t, "velero-ns", capturedArgs.Namespace) assert.Equal(t, iterator.DefaultTokenExpirySeconds, capturedArgs.TokenExpirySecs) assert.Zero(t, capturedArgs.MaxResults) diff --git a/pkg/cmd/cli/datamover/backup.go b/pkg/cmd/cli/datamover/backup.go index 2da71879c..f352c0aad 100644 --- a/pkg/cmd/cli/datamover/backup.go +++ b/pkg/cmd/cli/datamover/backup.go @@ -58,6 +58,9 @@ type dataMoverBackupConfig struct { duName string resourceTimeout time.Duration cbtSAName string + changeID string + volumeID string + snapshotID string } func NewBackupCommand(f client.Factory) *cobra.Command { @@ -79,7 +82,7 @@ func NewBackupCommand(f client.Factory) *cobra.Command { logger.Infof("Starting Velero data-mover backup %s (%s)", buildinfo.Version, buildinfo.FormattedGitSHA()) f.SetBasename(fmt.Sprintf("%s-%s", c.Parent().Name(), c.Name())) - s, err := newdataMoverBackup(logger, f, config) + s, err := newDataMoverBackup(logger, f, config) if err != nil { kube.ExitPodWithMessage(logger, false, "Failed to create data mover backup, %v", err) } @@ -95,6 +98,9 @@ func NewBackupCommand(f client.Factory) *cobra.Command { command.Flags().StringVar(&config.duName, "data-upload", config.duName, "The data upload name") 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.changeID, "change-id", config.changeID, "The change ID of the snapshot") + command.Flags().StringVar(&config.volumeID, "volume-id", config.volumeID, "The volume ID of the snapshot") + command.Flags().StringVar(&config.snapshotID, "snapshot-id", config.snapshotID, "The ID of the snapshot") _ = command.MarkFlagRequired("volume-path") _ = command.MarkFlagRequired("volume-mode") @@ -118,7 +124,7 @@ type dataMoverBackup struct { cbtService cbtservice.Service } -func newdataMoverBackup(logger logrus.FieldLogger, factory client.Factory, config dataMoverBackupConfig) (*dataMoverBackup, error) { +func newDataMoverBackup(logger logrus.FieldLogger, factory client.Factory, config dataMoverBackupConfig) (*dataMoverBackup, error) { ctx, cancelFunc := context.WithCancel(context.Background()) clientConfig, err := factory.ClientConfig() @@ -303,8 +309,24 @@ func (s *dataMoverBackup) createDataPathService() (dataPathService, error) { repoEnsurer := repository.NewEnsurer(s.client, s.logger, s.config.resourceTimeout) - return datamover.NewBackupMicroService(s.ctx, s.client, s.kubeClient, s.config.duName, s.namespace, s.nodeName, datapath.AccessPoint{ - ByPath: s.config.volumePath, - VolMode: uploader.PersistentVolumeMode(s.config.volumeMode), - }, s.dataPathMgr, repoEnsurer, credGetter, duInformer, s.logger), nil + return datamover.NewBackupMicroService( + s.ctx, + s.client, + s.kubeClient, + s.config.duName, + s.namespace, + s.nodeName, + datapath.AccessPoint{ + ByPath: s.config.volumePath, + VolMode: uploader.PersistentVolumeMode(s.config.volumeMode), + }, + s.dataPathMgr, + repoEnsurer, + credGetter, + duInformer, + s.config.changeID, + s.config.volumeID, + s.config.snapshotID, + s.logger, + ), nil } diff --git a/pkg/controller/data_upload_controller.go b/pkg/controller/data_upload_controller.go index c7bf07f89..9b2d9a2e3 100644 --- a/pkg/controller/data_upload_controller.go +++ b/pkg/controller/data_upload_controller.go @@ -463,9 +463,13 @@ func (r *DataUploadReconciler) initCancelableDataPath(ctx context.Context, async func (r *DataUploadReconciler) startCancelableDataPath(asyncBR datapath.AsyncBR, du *velerov2alpha1api.DataUpload, res *exposer.ExposeResult, log logrus.FieldLogger) error { log.Info("Start cancelable dataUpload") - if err := asyncBR.StartBackup(datapath.AccessPoint{ - ByPath: res.ByPod.VolumeName, - }, du.Spec.DataMoverConfig, nil); err != nil { + if err := asyncBR.StartBackup( + datapath.AccessPoint{ + ByPath: res.ByPod.VolumeName, + }, + du.Spec.DataMoverConfig, + nil, + ); err != nil { return errors.Wrapf(err, "error starting async backup for pod %s, volume %s", res.ByPod.HostingPod.Name, res.ByPod.VolumeName) } diff --git a/pkg/controller/data_upload_controller_test.go b/pkg/controller/data_upload_controller_test.go index d17ed527d..9703abe92 100644 --- a/pkg/controller/data_upload_controller_test.go +++ b/pkg/controller/data_upload_controller_test.go @@ -72,6 +72,7 @@ type FakeClient struct { patchError error updateConflict error listError error + getErrorMap map[string]error // key: object kind or name } func (c *FakeClient) Get(ctx context.Context, key kbclient.ObjectKey, obj kbclient.Object, opts ...kbclient.GetOption) error { @@ -79,6 +80,19 @@ func (c *FakeClient) Get(ctx context.Context, key kbclient.ObjectKey, obj kbclie return c.getError } + // Check if there's a specific error for this object type + if c.getErrorMap != nil { + objType := fmt.Sprintf("%T", obj) + if err, ok := c.getErrorMap[objType]; ok { + return err + } + + // Check if there's a specific error for this object name + if err, ok := c.getErrorMap[key.Name]; ok { + return err + } + } + return c.Client.Get(ctx, key, obj) } @@ -209,9 +223,13 @@ func initDataUploaderReconcilerWithError(needError ...error) (*DataUploadReconci if err != nil { return nil, err } + err = snapshotv1api.AddToScheme(scheme) + if err != nil { + return nil, err + } fakeClient := &FakeClient{ - Client: fake.NewClientBuilder().WithScheme(scheme).Build(), + Client: fake.NewClientBuilder().WithScheme(scheme).WithObjects(vsObject, node).Build(), } for k := range needError { @@ -505,7 +523,7 @@ func TestReconcile(t *testing.T) { { name: "du succeeds for accepted", du: dataUploadBuilder().Finalizers([]string{DataUploadDownloadFinalizer}).SnapshotType(fakeSnapshotType).Result(), - pvc: builder.ForPersistentVolumeClaim("fake-ns", "test-pvc").Result(), + pvc: builder.ForPersistentVolumeClaim("fake-ns", "test-pvc").VolumeName("test-pv").Result(), expected: dataUploadBuilder().Finalizers([]string{DataUploadDownloadFinalizer}).Phase(velerov2alpha1api.DataUploadPhaseAccepted).Result(), }, { @@ -636,6 +654,15 @@ func TestReconcile(t *testing.T) { if test.pvc != nil { err = r.client.Create(ctx, test.pvc) require.NoError(t, err) + + // Create the corresponding PV if PVC references one + if test.pvc.Spec.VolumeName != "" { + pv := builder.ForPersistentVolume(test.pvc.Spec.VolumeName). + CSI("csi.driver", "test-volume-id"). + ClaimRef(test.pvc.Namespace, test.pvc.Name).Result() + err = r.client.Create(ctx, pv) + require.NoError(t, err) + } } if test.dataMgr != nil { diff --git a/pkg/datamover/backup_micro_service.go b/pkg/datamover/backup_micro_service.go index 6b719c792..08a005217 100644 --- a/pkg/datamover/backup_micro_service.go +++ b/pkg/datamover/backup_micro_service.go @@ -67,6 +67,10 @@ type BackupMicroService struct { duInformer cache.Informer duHandler cachetool.ResourceEventHandlerRegistration nodeName string + + changeID string + volumeID string + snapshotID string } type dataPathResult struct { @@ -76,7 +80,7 @@ type dataPathResult struct { func NewBackupMicroService(ctx context.Context, client client.Client, kubeClient kubernetes.Interface, dataUploadName string, namespace string, nodeName string, sourceTargetPath datapath.AccessPoint, dataPathMgr *datapath.Manager, repoEnsurer *repository.Ensurer, cred *credentials.CredentialGetter, - duInformer cache.Informer, log logrus.FieldLogger) *BackupMicroService { + duInformer cache.Informer, changeID string, volumeID string, snapshotID string, log logrus.FieldLogger) *BackupMicroService { return &BackupMicroService{ ctx: ctx, client: client, @@ -91,6 +95,9 @@ func NewBackupMicroService(ctx context.Context, client client.Client, kubeClient nodeName: nodeName, resultSignal: make(chan dataPathResult), duInformer: duInformer, + changeID: changeID, + volumeID: volumeID, + snapshotID: snapshotID, } } @@ -200,6 +207,9 @@ func (r *BackupMicroService) RunCancelableDataPath(ctx context.Context) (string, ParentSnapshot: "", ForceFull: false, Tags: tags, + VolumeID: r.volumeID, + ChangeID: r.changeID, + SnapshotID: r.snapshotID, }); err != nil { return "", errors.Wrap(err, "error starting data path backup") } diff --git a/pkg/datamover/backup_micro_service_test.go b/pkg/datamover/backup_micro_service_test.go index ab664df71..e6291244b 100644 --- a/pkg/datamover/backup_micro_service_test.go +++ b/pkg/datamover/backup_micro_service_test.go @@ -29,21 +29,16 @@ import ( "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" "k8s.io/apimachinery/pkg/runtime" - - "github.com/vmware-tanzu/velero/pkg/builder" - "github.com/vmware-tanzu/velero/pkg/datapath" - "github.com/vmware-tanzu/velero/pkg/uploader" - - velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1" - + kbclient "sigs.k8s.io/controller-runtime/pkg/client" clientFake "sigs.k8s.io/controller-runtime/pkg/client/fake" + velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1" velerov2alpha1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1" - velerotest "github.com/vmware-tanzu/velero/pkg/test" - - kbclient "sigs.k8s.io/controller-runtime/pkg/client" - + "github.com/vmware-tanzu/velero/pkg/builder" + "github.com/vmware-tanzu/velero/pkg/datapath" datapathmockes "github.com/vmware-tanzu/velero/pkg/datapath/mocks" + velerotest "github.com/vmware-tanzu/velero/pkg/test" + "github.com/vmware-tanzu/velero/pkg/uploader" ) type backupMsTestHelper struct { @@ -294,7 +289,10 @@ func TestCancelDataUpload(t *testing.T) { func TestRunCancelableDataPath(t *testing.T) { dataUploadName := "fake-data-upload" du := builder.ForDataUpload(velerov1api.DefaultNamespace, dataUploadName).Phase(velerov2alpha1api.DataUploadPhaseNew).Result() - duInProgress := builder.ForDataUpload(velerov1api.DefaultNamespace, dataUploadName).Phase(velerov2alpha1api.DataUploadPhaseInProgress).Result() + duInProgress := builder.ForDataUpload(velerov1api.DefaultNamespace, dataUploadName).Phase(velerov2alpha1api.DataUploadPhaseInProgress).CSISnapshot( + &velerov2alpha1api.CSISnapshotSpec{ + VolumeSnapshot: "fake-snapshot", + }).Result() ctxTimeout, cancel := context.WithTimeout(t.Context(), time.Second) tests := []struct { diff --git a/pkg/datapath/data_path.go b/pkg/datapath/data_path.go index 71b8e0690..6cef1af26 100644 --- a/pkg/datapath/data_path.go +++ b/pkg/datapath/data_path.go @@ -26,6 +26,7 @@ import ( "github.com/vmware-tanzu/velero/internal/credentials" velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1" + "github.com/vmware-tanzu/velero/pkg/cbtservice" "github.com/vmware-tanzu/velero/pkg/repository" repokey "github.com/vmware-tanzu/velero/pkg/repository/keys" repoProvider "github.com/vmware-tanzu/velero/pkg/repository/provider" @@ -53,6 +54,9 @@ type BackupStartParam struct { ParentSnapshot string ForceFull bool Tags map[string]string + VolumeID string + ChangeID string + SnapshotID string } type generalDataPath struct { @@ -182,8 +186,24 @@ func (dp *generalDataPath) StartBackup(source AccessPoint, uploaderConfig map[st dp.wgDataPath.Done() }() - snapshotID, emptySnapshot, totalBytes, incrementalBytes, err := dp.uploaderProv.RunBackup(dp.ctx, source.ByPath, backupParam.RealSource, backupParam.Tags, backupParam.ForceFull, - backupParam.ParentSnapshot, provider.CBTParam{}, source.VolMode, uploaderConfig, dp) + snapshotID, emptySnapshot, totalBytes, incrementalBytes, err := dp.uploaderProv.RunBackup( + dp.ctx, + source.ByPath, + backupParam.RealSource, + backupParam.Tags, + backupParam.ForceFull, + backupParam.ParentSnapshot, + provider.CBTParam{ + Source: cbtservice.SourceInfo{ + Snapshot: backupParam.SnapshotID, + VolumeID: backupParam.VolumeID, + ChangeID: backupParam.ChangeID, + }, + }, + source.VolMode, + uploaderConfig, + 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 4582c1e62..6c92a6973 100644 --- a/pkg/exposer/csi_snapshot.go +++ b/pkg/exposer/csi_snapshot.go @@ -20,6 +20,7 @@ import ( "context" "fmt" "maps" + "strings" "time" "github.com/cockroachdb/errors" @@ -110,6 +111,12 @@ 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{ @@ -256,6 +263,14 @@ func (e *csiSnapshotExposer) Expose(ctx context.Context, ownerObject corev1api.O affinity := kube.GetLoadAffinityByStorageClass(csiExposeParam.Affinity, backupPVCStorageClass, curLog) + var cbtInfo cbtInfo + if csiExposeParam.DataMover == datamover.DataMoverTypeVeleroBlock { + cbtInfo, err = e.getCBTInfo(ctx, backupVS, backupVSC, csiExposeParam.SourcePVName) + if err != nil { + return errors.Wrap(err, "error to get CBT info") + } + } + backupPod, err := e.createBackupPod( ctx, ownerObject, @@ -273,6 +288,7 @@ func (e *csiSnapshotExposer) Expose(ctx context.Context, ownerObject corev1api.O intoleratableNodes, volumeTopology, csiExposeParam.SnapshotMetadataServiceConfigs, + &cbtInfo, ) if err != nil { return errors.Wrap(err, "error to create backup pod") @@ -289,6 +305,49 @@ 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) @@ -618,6 +677,7 @@ func (e *csiSnapshotExposer) createBackupPod( intoleratableNodes []string, volumeTopology *corev1api.NodeSelector, csiSnapshotMetadataServiceConfigs *velerotypes.CSISnapshotMetadataService, + cbtInfo *cbtInfo, ) (*corev1api.Pod, error) { podName := ownerObject.Name @@ -670,6 +730,12 @@ func (e *csiSnapshotExposer) createBackupPod( fmt.Sprintf("--resource-timeout=%s", operationTimeout.String()), } + 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, podInfo.logFormatArgs...) args = append(args, podInfo.logLevelArgs...) diff --git a/pkg/exposer/csi_snapshot_priority_test.go b/pkg/exposer/csi_snapshot_priority_test.go index 8c3086f76..f05ab6007 100644 --- a/pkg/exposer/csi_snapshot_priority_test.go +++ b/pkg/exposer/csi_snapshot_priority_test.go @@ -156,6 +156,7 @@ func TestCreateBackupPodWithPriorityClass(t *testing.T) { nil, nil, nil, + nil, ) require.NoError(t, err, tc.description) @@ -243,6 +244,7 @@ func TestCreateBackupPodWithMissingConfigMap(t *testing.T) { nil, nil, nil, + nil, ) // Should succeed even when config map is missing diff --git a/pkg/exposer/csi_snapshot_test.go b/pkg/exposer/csi_snapshot_test.go index bf3b08066..e1512e633 100644 --- a/pkg/exposer/csi_snapshot_test.go +++ b/pkg/exposer/csi_snapshot_test.go @@ -17,34 +17,38 @@ limitations under the License. package exposer import ( + "context" "fmt" "maps" + "strings" "testing" "time" "github.com/cockroachdb/errors" snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v8/apis/volumesnapshot/v1" snapshotFake "github.com/kubernetes-csi/external-snapshotter/client/v8/clientset/versioned/fake" + "github.com/sirupsen/logrus" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" appsv1api "k8s.io/api/apps/v1" corev1api "k8s.io/api/core/v1" + storagev1api "k8s.io/api/storage/v1" "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/client-go/kubernetes/fake" + kubefake "k8s.io/client-go/kubernetes/fake" clientTesting "k8s.io/client-go/testing" "k8s.io/utils/ptr" clientFake "sigs.k8s.io/controller-runtime/pkg/client/fake" velerov1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1" + "github.com/vmware-tanzu/velero/pkg/datamover" velerotest "github.com/vmware-tanzu/velero/pkg/test" velerotypes "github.com/vmware-tanzu/velero/pkg/types" "github.com/vmware-tanzu/velero/pkg/util" "github.com/vmware-tanzu/velero/pkg/util/boolptr" "github.com/vmware-tanzu/velero/pkg/util/kube" - - storagev1api "k8s.io/api/storage/v1" ) type reactor struct { @@ -191,6 +195,19 @@ func TestExpose(t *testing.T) { }, } + sourcePV := &corev1api.PersistentVolume{ + ObjectMeta: metav1.ObjectMeta{ + Name: "fake-pv", + }, + Spec: corev1api.PersistentVolumeSpec{ + PersistentVolumeSource: corev1api.PersistentVolumeSource{ + CSI: &corev1api.CSIPersistentVolumeSource{ + VolumeHandle: "csi-volume-handle", + }, + }, + }, + } + tests := []struct { name string snapshotClientObj []runtime.Object @@ -1015,6 +1032,46 @@ func TestExpose(t *testing.T) { }, expectedPVCAnnotation: map[string]string{util.VSphereCNSFastCloneAnno: "true"}, }, + { + name: "block data mover success", + ownerBackup: backup, + exposeParam: CSISnapshotExposeParam{ + SnapshotName: "fake-vs", + SourceNamespace: "fake-ns", + AccessMode: AccessModeFileSystem, + OperationTimeout: time.Millisecond, + ExposeTimeout: time.Millisecond, + StorageClass: "fake-sc", + SourcePVName: "fake-pv", + DataMover: datamover.DataMoverTypeVeleroBlock, + }, + snapshotClientObj: []runtime.Object{ + vsObject, + vscObj, + }, + kubeClientObj: []runtime.Object{ + daemonSet, + scObj, + sourcePV, + }, + expectedAffinity: &corev1api.Affinity{ + NodeAffinity: &corev1api.NodeAffinity{ + RequiredDuringSchedulingIgnoredDuringExecution: &corev1api.NodeSelector{ + NodeSelectorTerms: []corev1api.NodeSelectorTerm{ + { + MatchExpressions: []corev1api.NodeSelectorRequirement{ + { + Key: "kubernetes.io/os", + Operator: corev1api.NodeSelectorOpNotIn, + Values: []string{"windows"}, + }, + }, + }, + }, + }, + }, + }, + }, } for _, test := range tests { @@ -1994,3 +2051,150 @@ end diagnose CSI exposer`, }) } } + +func TestGetCBTInfo(t *testing.T) { + handle := "snapshot-handle-1" + + tests := []struct { + name string + vs *snapshotv1api.VolumeSnapshot + vsc *snapshotv1api.VolumeSnapshotContent + pv *corev1api.PersistentVolume + sourcePVName string + want cbtInfo + wantErrSubstr string + }{ + { + name: "return error when vs is nil", + vs: nil, + vsc: &snapshotv1api.VolumeSnapshotContent{}, + sourcePVName: "pv-1", + wantErrSubstr: "vs or vsc is nil", + }, + { + name: "use annotations when change-id and snapshot annotation exist", + vs: &snapshotv1api.VolumeSnapshot{ + ObjectMeta: metav1.ObjectMeta{ + Name: "vs-anno", + Annotations: map[string]string{ + util.VSphereCNSChangeIDAnno: "change-id-1", + util.VSphereCNSSnapshotAnno: "volume-id-1+snapshot-id-1", + }, + }, + }, + vsc: &snapshotv1api.VolumeSnapshotContent{}, + sourcePVName: "pv-ignored", + want: cbtInfo{ + changeID: "change-id-1", + volumeID: "volume-id-1", + snapshotID: "vs-anno", + }, + }, + { + name: "fallback to pv and vsc snapshot handle", + vs: &snapshotv1api.VolumeSnapshot{ + ObjectMeta: metav1.ObjectMeta{Name: "vs-fallback"}, + }, + vsc: &snapshotv1api.VolumeSnapshotContent{ + Status: &snapshotv1api.VolumeSnapshotContentStatus{ + SnapshotHandle: &handle, + }, + }, + pv: &corev1api.PersistentVolume{ + ObjectMeta: metav1.ObjectMeta{Name: "pv-1"}, + Spec: corev1api.PersistentVolumeSpec{ + PersistentVolumeSource: corev1api.PersistentVolumeSource{ + CSI: &corev1api.CSIPersistentVolumeSource{ + VolumeHandle: "csi-volume-handle-1", + }, + }, + }, + }, + sourcePVName: "pv-1", + want: cbtInfo{ + changeID: "snapshot-handle-1", + volumeID: "csi-volume-handle-1", + snapshotID: "vs-fallback", + }, + }, + { + name: "return error when pv not found in fallback path", + vs: &snapshotv1api.VolumeSnapshot{ + ObjectMeta: metav1.ObjectMeta{Name: "vs-no-pv"}, + }, + vsc: &snapshotv1api.VolumeSnapshotContent{}, + sourcePVName: "pv-not-found", + wantErrSubstr: "failed to get pv pv-not-found", + }, + { + name: "return error when pv has no csi volume handle", + vs: &snapshotv1api.VolumeSnapshot{ + ObjectMeta: metav1.ObjectMeta{Name: "vs-no-volume-handle"}, + }, + vsc: &snapshotv1api.VolumeSnapshotContent{}, + pv: &corev1api.PersistentVolume{ + ObjectMeta: metav1.ObjectMeta{Name: "pv-no-handle"}, + Spec: corev1api.PersistentVolumeSpec{}, + }, + sourcePVName: "pv-no-handle", + wantErrSubstr: "volumeID must not be empty for CBT", + }, + { + name: "return error when snapshot annotation is invalid", + vs: &snapshotv1api.VolumeSnapshot{ + ObjectMeta: metav1.ObjectMeta{ + Name: "vs-no-volume-handle", + Annotations: map[string]string{ + util.VSphereCNSChangeIDAnno: "change-id-1", + util.VSphereCNSSnapshotAnno: "volume-id-1:snapshot-id-1", + }, + }, + }, + vsc: &snapshotv1api.VolumeSnapshotContent{}, + pv: &corev1api.PersistentVolume{ + ObjectMeta: metav1.ObjectMeta{Name: "pv-1"}, + Spec: corev1api.PersistentVolumeSpec{ + PersistentVolumeSource: corev1api.PersistentVolumeSource{ + CSI: &corev1api.CSIPersistentVolumeSource{ + VolumeHandle: "csi-volume-handle-1", + }, + }, + }, + }, + sourcePVName: "pv-1", + wantErrSubstr: "volumeID must not be empty for CBT", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + var objs []runtime.Object + if tc.pv != nil { + objs = append(objs, tc.pv) + } + exposer := &csiSnapshotExposer{ + kubeClient: kubefake.NewSimpleClientset(objs...), + log: logrus.StandardLogger(), + } + + got, err := exposer.getCBTInfo(context.Background(), tc.vs, tc.vsc, tc.sourcePVName) + + if tc.wantErrSubstr != "" { + if err == nil { + t.Fatalf("expected error containing %q, got nil", tc.wantErrSubstr) + } + if !strings.Contains(err.Error(), tc.wantErrSubstr) { + t.Fatalf("expected error containing %q, got %q", tc.wantErrSubstr, err.Error()) + } + return + } + + 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) + } + }) + } +} diff --git a/pkg/uploader/provider/kopia.go b/pkg/uploader/provider/kopia.go index ba86c977c..682b2053e 100644 --- a/pkg/uploader/provider/kopia.go +++ b/pkg/uploader/provider/kopia.go @@ -120,7 +120,8 @@ func (kp *kopiaProvider) RunBackup( _ CBTParam, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, - updater uploader.ProgressUpdater) (string, bool, int64, int64, error) { + updater uploader.ProgressUpdater, +) (string, bool, int64, int64, error) { if updater == nil { return "", false, 0, 0, errors.New("Need to initial backup progress updater first") } diff --git a/pkg/util/third_party.go b/pkg/util/third_party.go index 400c7a898..81b964454 100644 --- a/pkg/util/third_party.go +++ b/pkg/util/third_party.go @@ -31,4 +31,6 @@ var ThirdPartyTolerations = []string{ const ( VSphereCNSFastCloneAnno = "csi.vsphere.volume/fast-provisioning" + VSphereCNSSnapshotAnno = "csi.vsphere.volume/snapshot" + VSphereCNSChangeIDAnno = "csi.vsphere.volume/change-id" )