mirrors/velero
1
0
Fork 0
mirror of https://github.com/vmware-tanzu/velero.git synced 2026-08-17 20:56:08 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge pull request #9863 from blackpiglet/jxun/change-id

Browse Source 
Support change-id and volume-id in backup workflow.
This commit is contained in:
Xun Jiang/Bruce Jiang
2026-07-02 11:32:15 +08:00
committed by GitHub
parent 08385354db 0bc06323bf
commit 96ae6586d0
15 changed files with 393 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
changelogs/unreleased/9863-blackpiglet
+1
View File
@@ -0,0 +1 @@
Support change-id and volume-id in backup workflow.
pkg/backup/actions/csi/pvc_action.go
+4 -6
View File
@@ -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,
)
pkg/cbtservice/csi_service_impl.go
+2 -2
View File
@@ -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,
pkg/cbtservice/csi_service_impl_test.go
+1 -1
View File
@@ -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)
pkg/cmd/cli/datamover/backup.go
+28 -6
View File
@@ -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
}
pkg/controller/data_upload_controller.go
+7 -3
View File
@@ -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)
}
pkg/controller/data_upload_controller_test.go
+29 -2
View File
@@ -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 {
pkg/datamover/backup_micro_service.go
+11 -1
View File
@@ -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")
}
pkg/datamover/backup_micro_service_test.go
+10 -12
View File
@@ -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 {
pkg/datapath/data_path.go
+22 -2
View File
@@ -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)
pkg/exposer/csi_snapshot.go
+66
View File
@@ -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...)
pkg/exposer/csi_snapshot_priority_test.go
+2
View File
@@ -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
pkg/exposer/csi_snapshot_test.go
+206 -2
View File
@@ -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)
}
})
}
}
pkg/uploader/provider/kopia.go
+2 -1
View File
@@ -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")
}
pkg/util/third_party.go
+2
View File
@@ -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"
)