mirrors/velero
1
0
Fork 0
mirror of https://github.com/vmware-tanzu/velero.git synced 2026-07-19 14:32:36 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge branch 'main' into block-uploader-backup-implementation

Browse Source
This commit is contained in:
Lyndon-Li
2026-07-09 11:39:30 +08:00
parent 8d23c7e813 4934a92ee2
commit ce1c1803b2
24 changed files with 542 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/dependabot.yml
+14
View File
@@ -15,6 +15,20 @@ updates:
schedule:
interval: "weekly"
labels:
- "Dependencies"
- "go"
- "kind/changelog-not-required"
ignore:
- dependency-name: "*"
update-types: ["version-update:semver-major", "version-update:semver-minor", "version-update:semver-patch"]
# Dependencies listed in pkg/apis/go.mod
- package-ecosystem: "gomod"
directory: "/pkg/apis" # Location of package manifests
schedule:
interval: "weekly"
labels:
- "Dependencies"
- "go"
- "kind/changelog-not-required"
ignore:
- dependency-name: "*"
.github/workflows/e2e-test-kind.yaml
+1 -1
View File
@@ -187,7 +187,7 @@ jobs:
timeout-minutes: 30
- name: Upload debug bundle
if: ${{ failure() }}
uses: actions/upload-artifact@v5
uses: actions/upload-artifact@v7
with:
name: DebugBundle-k8s-${{ matrix.k8s }}-job-${{ strategy.job-index }}
path: /home/runner/work/velero/velero/test/e2e/debug-bundle*
changelogs/unreleased/9863-blackpiglet
+1
View File
@@ -0,0 +1 @@
Support change-id and volume-id in backup workflow.
changelogs/unreleased/9955-sseago
+1
View File
@@ -0,0 +1 @@
Optimize VSC handle readiness polling for VSS backups
pkg/apis/go.mod
+2 -2
View File
@@ -16,8 +16,8 @@ require (
github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect
github.com/x448/float16 v0.8.4 // indirect
go.yaml.in/yaml/v2 v2.4.3 // indirect
golang.org/x/net v0.49.0 // indirect
golang.org/x/text v0.33.0 // indirect
golang.org/x/net v0.55.0 // indirect
golang.org/x/text v0.37.0 // indirect
gopkg.in/inf.v0 v0.9.1 // indirect
k8s.io/klog/v2 v2.140.0 // indirect
k8s.io/kube-openapi v0.0.0-20260317180543-43fb72c5454a // indirect
pkg/apis/go.sum
+4 -4
View File
@@ -37,10 +37,10 @@ github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=
github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg=
go.yaml.in/yaml/v2 v2.4.3 h1:6gvOSjQoTB3vt1l+CU+tSyi/HOjfOjRLJ4YwYZGwRO0=
go.yaml.in/yaml/v2 v2.4.3/go.mod h1:zSxWcmIDjOzPXpjlTTbAsKokqkDNAVtZO0WOMiT90s8=
golang.org/x/net v0.49.0 h1:eeHFmOGUTtaaPSGNmjBKpbng9MulQsJURQUAfUwY++o=
golang.org/x/net v0.49.0/go.mod h1:/ysNB2EvaqvesRkuLAyjI1ycPZlQHM3q01F02UY/MV8=
golang.org/x/text v0.33.0 h1:B3njUFyqtHDUI5jMn1YIr5B0IE2U0qck04r6d4KPAxE=
golang.org/x/text v0.33.0/go.mod h1:LuMebE6+rBincTi9+xWTY8TztLzKHc/9C1uBCG27+q8=
golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8=
golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww=
golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc=
golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
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/cmd/cli/install/install.go
+14 -10
View File
@@ -81,6 +81,7 @@ type Options struct {
DefaultVolumesToFsBackup bool
UploaderType string
DefaultSnapshotMoveData bool
CSISnapshotEarlyFrequentPolling bool
DisableInformerCache bool
ScheduleSkipImmediately bool
PodResources kubeutil.PodResources
@@ -141,6 +142,7 @@ func (o *Options) BindFlags(flags *pflag.FlagSet) {
flags.BoolVar(&o.DefaultVolumesToFsBackup, "default-volumes-to-fs-backup", o.DefaultVolumesToFsBackup, "Bool flag to configure Velero server to use pod volume file system backup by default for all volumes on all backups. Optional.")
flags.StringVar(&o.UploaderType, "uploader-type", o.UploaderType, fmt.Sprintf("The type of uploader to transfer the data of pod volumes, supported value: '%s'", uploader.KopiaType))
flags.BoolVar(&o.DefaultSnapshotMoveData, "default-snapshot-move-data", o.DefaultSnapshotMoveData, "Bool flag to configure Velero server to move data by default for all snapshots supporting data movement. Optional.")
flags.BoolVar(&o.CSISnapshotEarlyFrequentPolling, "csi-snapshot-early-frequent-polling", o.CSISnapshotEarlyFrequentPolling, "Bool flag to configure Velero server to use early frequent polling by default for all CSI snapshots. Optional.")
flags.BoolVar(&o.DisableInformerCache, "disable-informer-cache", o.DisableInformerCache, "Disable informer cache for Get calls on restore. With this enabled, it will speed up restore in cases where there are backup resources which already exist in the cluster, but for very large clusters this will increase velero memory usage. Default is false (don't disable). Optional.")
flags.BoolVar(&o.ScheduleSkipImmediately, "schedule-skip-immediately", o.ScheduleSkipImmediately, "Skip the first scheduled backup immediately after creating a schedule. Default is false (don't skip).")
flags.BoolVar(&o.NodeAgentDisableHostPath, "node-agent-disable-host-path", o.NodeAgentDisableHostPath, "Don't mount the pod volume host path to node-agent. Optional. Pod volume host path mount is required by fs-backup but could be disabled for other backup methods.")
@@ -238,16 +240,17 @@ func NewInstallOptions() *Options {
NodeAgentPodCPULimit: install.DefaultNodeAgentPodCPULimit,
NodeAgentPodMemLimit: install.DefaultNodeAgentPodMemLimit,
// Default to creating a VSL unless we're told otherwise
UseVolumeSnapshots: true,
NoDefaultBackupLocation: false,
CRDsOnly: false,
DefaultVolumesToFsBackup: false,
UploaderType: uploader.KopiaType,
DefaultSnapshotMoveData: false,
DisableInformerCache: false,
ScheduleSkipImmediately: false,
kubeletRootDir: install.DefaultKubeletRootDir,
NodeAgentDisableHostPath: false,
UseVolumeSnapshots: true,
NoDefaultBackupLocation: false,
CRDsOnly: false,
DefaultVolumesToFsBackup: false,
UploaderType: uploader.KopiaType,
DefaultSnapshotMoveData: false,
CSISnapshotEarlyFrequentPolling: false,
DisableInformerCache: false,
ScheduleSkipImmediately: false,
kubeletRootDir: install.DefaultKubeletRootDir,
NodeAgentDisableHostPath: false,
}
}
@@ -324,6 +327,7 @@ func (o *Options) AsVeleroOptions() (*install.VeleroOptions, error) {
DefaultVolumesToFsBackup: o.DefaultVolumesToFsBackup,
UploaderType: o.UploaderType,
DefaultSnapshotMoveData: o.DefaultSnapshotMoveData,
CSISnapshotEarlyFrequentPolling: o.CSISnapshotEarlyFrequentPolling,
DisableInformerCache: o.DisableInformerCache,
ScheduleSkipImmediately: o.ScheduleSkipImmediately,
PodResources: o.PodResources,
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/install/deployment.go
+16
View File
@@ -50,6 +50,7 @@ type podTemplateConfig struct {
serviceAccountName string
uploaderType string
defaultSnapshotMoveData bool
csiSnapshotEarlyFrequentPolling bool
privilegedNodeAgent bool
disableInformerCache bool
scheduleSkipImmediately bool
@@ -166,6 +167,12 @@ func WithDefaultSnapshotMoveData(b bool) podTemplateOption {
}
}
func WithCSISnapshotEarlyFrequentPolling(b bool) podTemplateOption {
return func(c *podTemplateConfig) {
c.csiSnapshotEarlyFrequentPolling = b
}
}
func WithDisableInformerCache(b bool) podTemplateOption {
return func(c *podTemplateConfig) {
c.disableInformerCache = b
@@ -489,6 +496,15 @@ func Deployment(namespace string, opts ...podTemplateOption) *appsv1api.Deployme
}...)
}
if c.csiSnapshotEarlyFrequentPolling {
deployment.Spec.Template.Spec.Containers[0].Env = append(deployment.Spec.Template.Spec.Containers[0].Env, []corev1api.EnvVar{
{
Name: "CSI_SNAPSHOT_EARLY_FREQUENT_POLLING",
Value: "true",
},
}...)
}
deployment.Spec.Template.Spec.Containers[0].Env = append(deployment.Spec.Template.Spec.Containers[0].Env, c.envVars...)
if len(c.plugins) > 0 {
pkg/install/resources.go
+5
View File
@@ -263,6 +263,7 @@ type VeleroOptions struct {
DefaultVolumesToFsBackup bool
UploaderType string
DefaultSnapshotMoveData bool
CSISnapshotEarlyFrequentPolling bool
DisableInformerCache bool
ScheduleSkipImmediately bool
PodResources kube.PodResources
@@ -390,6 +391,10 @@ func AllResources(o *VeleroOptions) *unstructured.UnstructuredList {
deployOpts = append(deployOpts, WithDefaultSnapshotMoveData(true))
}
if o.CSISnapshotEarlyFrequentPolling {
deployOpts = append(deployOpts, WithCSISnapshotEarlyFrequentPolling(true))
}
if o.DisableInformerCache {
deployOpts = append(deployOpts, WithDisableInformerCache(true))
}
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/csi/volume_snapshot.go
+92 -60
View File
@@ -20,6 +20,8 @@ import (
"context"
"encoding/json"
"fmt"
"os"
"strconv"
"strings"
"time"
@@ -598,72 +600,102 @@ func WaitUntilVSCHandleIsReady(
log logrus.FieldLogger,
csiSnapshotTimeout time.Duration,
) (*snapshotv1api.VolumeSnapshotContent, error) {
// We'll wait 10m for the VSC to be reconciled polling
// every 5s unless backup's csiSnapshotTimeout is set
interval := 5 * time.Second
// We'll wait for the VSC to be reconciled, trying a fast poll interval first
// before falling back to a slower poll interval for the full csiSnapshotTimeout.
vsc := new(snapshotv1api.VolumeSnapshotContent)
var interval time.Duration
err := wait.PollUntilContextTimeout(
pollFunc := func(ctx context.Context) (bool, error) {
vs := new(snapshotv1api.VolumeSnapshot)
if err := crClient.Get(
ctx,
crclient.ObjectKeyFromObject(volSnap),
vs,
); err != nil {
return false,
errors.Wrapf(
err,
"failed to get volumesnapshot %s/%s",
volSnap.Namespace, volSnap.Name,
)
}
if vs.Status == nil || vs.Status.BoundVolumeSnapshotContentName == nil {
log.Infof("Waiting for CSI driver to reconcile volumesnapshot %s/%s. Retrying in %ds",
volSnap.Namespace, volSnap.Name, interval/time.Second)
return false, nil
}
if err := crClient.Get(
ctx,
crclient.ObjectKey{
Name: *vs.Status.BoundVolumeSnapshotContentName,
},
vsc,
); err != nil {
return false,
errors.Wrapf(
err,
"failed to get VolumeSnapshotContent %s for VolumeSnapshot %s/%s",
*vs.Status.BoundVolumeSnapshotContentName, vs.Namespace, vs.Name,
)
}
// we need to wait for the VolumeSnapshotContent
// to have a snapshot handle because during restore,
// we'll use that snapshot handle as the source for
// the VolumeSnapshotContent so it's statically
// bound to the existing snapshot.
if vsc.Status == nil ||
vsc.Status.SnapshotHandle == nil {
log.Infof(
"Waiting for VolumeSnapshotContents %s to have snapshot handle. Retrying in %ds",
vsc.Name, interval/time.Second)
if vsc.Status != nil &&
vsc.Status.Error != nil {
log.Warnf("VolumeSnapshotContent %s has error: %v",
vsc.Name, *vsc.Status.Error.Message)
}
return false, nil
}
return true, nil
}
var err error
frequentPolling, err := strconv.ParseBool(os.Getenv("CSI_SNAPSHOT_EARLY_FREQUENT_POLLING"))
if err == nil && frequentPolling {
// The short interval for the first ten seconds is due to the fact that
// Microsoft VSS backups have a hard-coded unfreeze call after 10 seconds,
// so we need to minimize waiting time during the first 10 seconds.
// First poll with a short interval and timeout.
interval = 1 * time.Second
timeout := 10 * time.Second
err = wait.PollUntilContextTimeout(
context.Background(),
interval,
timeout,
true,
pollFunc,
)
if err == nil {
return vsc, nil
}
if !wait.Interrupted(err) {
return nil, err
}
}
// If the first poll timed out, poll with a longer interval and the full timeout.
interval = 5 * time.Second
err = wait.PollUntilContextTimeout(
context.Background(),
interval,
csiSnapshotTimeout,
true,
func(ctx context.Context) (bool, error) {
vs := new(snapshotv1api.VolumeSnapshot)
if err := crClient.Get(
ctx,
crclient.ObjectKeyFromObject(volSnap),
vs,
); err != nil {
return false,
errors.Wrapf(
err,
"failed to get volumesnapshot %s/%s",
volSnap.Namespace, volSnap.Name,
)
}
if vs.Status == nil || vs.Status.BoundVolumeSnapshotContentName == nil {
log.Infof("Waiting for CSI driver to reconcile volumesnapshot %s/%s. Retrying in %ds",
volSnap.Namespace, volSnap.Name, interval/time.Second)
return false, nil
}
if err := crClient.Get(
ctx,
crclient.ObjectKey{
Name: *vs.Status.BoundVolumeSnapshotContentName,
},
vsc,
); err != nil {
return false,
errors.Wrapf(
err,
"failed to get VolumeSnapshotContent %s for VolumeSnapshot %s/%s",
*vs.Status.BoundVolumeSnapshotContentName, vs.Namespace, vs.Name,
)
}
// we need to wait for the VolumeSnapshotContent
// to have a snapshot handle because during restore,
// we'll use that snapshot handle as the source for
// the VolumeSnapshotContent so it's statically
// bound to the existing snapshot.
if vsc.Status == nil ||
vsc.Status.SnapshotHandle == nil {
log.Infof(
"Waiting for VolumeSnapshotContents %s to have snapshot handle. Retrying in %ds",
vsc.Name, interval/time.Second)
if vsc.Status != nil &&
vsc.Status.Error != nil {
log.Warnf("VolumeSnapshotContent %s has error: %v",
vsc.Name, *vsc.Status.Error.Message)
}
return false, nil
}
return true, nil
},
pollFunc,
)
if err != nil {
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"
)