From 9bbadf346d857ba18cf660a61c7d296b1dd47946 Mon Sep 17 00:00:00 2001 From: Lyndon-Li Date: Wed, 22 Apr 2026 14:15:17 +0800 Subject: [PATCH 1/4] modify uploader provider interface for block data mover Signed-off-by: Lyndon-Li --- pkg/datapath/file_system.go | 2 +- pkg/datapath/file_system_test.go | 2 +- pkg/uploader/provider/kopia.go | 1 + pkg/uploader/provider/kopia_test.go | 2 +- pkg/uploader/provider/mocks/Provider.go | 396 +++++++++++++++++------- pkg/uploader/provider/provider.go | 7 + 6 files changed, 303 insertions(+), 107 deletions(-) diff --git a/pkg/datapath/file_system.go b/pkg/datapath/file_system.go index 61fac1e47..67db98fbe 100644 --- a/pkg/datapath/file_system.go +++ b/pkg/datapath/file_system.go @@ -183,7 +183,7 @@ func (fs *fileSystemBR) StartBackup(source AccessPoint, uploaderConfig map[strin }() snapshotID, emptySnapshot, totalBytes, incrementalBytes, err := fs.uploaderProv.RunBackup(fs.ctx, source.ByPath, backupParam.RealSource, backupParam.Tags, backupParam.ForceFull, - backupParam.ParentSnapshot, source.VolMode, uploaderConfig, fs) + backupParam.ParentSnapshot, provider.CBTParam{}, source.VolMode, uploaderConfig, fs) if err == provider.ErrorCanceled { fs.callbacks.OnCancelled(context.Background(), fs.namespace, fs.jobName) diff --git a/pkg/datapath/file_system_test.go b/pkg/datapath/file_system_test.go index 3887a82e3..c605311d2 100644 --- a/pkg/datapath/file_system_test.go +++ b/pkg/datapath/file_system_test.go @@ -96,7 +96,7 @@ func TestAsyncBackup(t *testing.T) { t.Run(test.name, func(t *testing.T) { fs := newFileSystemBR("job-1", "test", nil, "velero", Callbacks{}, velerotest.NewLogger()).(*fileSystemBR) mockProvider := providerMock.NewProvider(t) - mockProvider.On("RunBackup", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(test.result.Backup.SnapshotID, test.result.Backup.EmptySnapshot, test.result.Backup.TotalBytes, test.result.Backup.IncrementalBytes, test.err) + mockProvider.On("RunBackup", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(test.result.Backup.SnapshotID, test.result.Backup.EmptySnapshot, test.result.Backup.TotalBytes, test.result.Backup.IncrementalBytes, test.err) mockProvider.On("Close", mock.Anything).Return(nil) fs.uploaderProv = mockProvider fs.initialized = true diff --git a/pkg/uploader/provider/kopia.go b/pkg/uploader/provider/kopia.go index 16d8aeb52..51615fd2a 100644 --- a/pkg/uploader/provider/kopia.go +++ b/pkg/uploader/provider/kopia.go @@ -118,6 +118,7 @@ func (kp *kopiaProvider) RunBackup( tags map[string]string, forceFull bool, parentSnapshot string, + _ CBTParam, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, updater uploader.ProgressUpdater) (string, bool, int64, int64, error) { diff --git a/pkg/uploader/provider/kopia_test.go b/pkg/uploader/provider/kopia_test.go index 734bdb176..183a65d26 100644 --- a/pkg/uploader/provider/kopia_test.go +++ b/pkg/uploader/provider/kopia_test.go @@ -106,7 +106,7 @@ func TestRunBackup(t *testing.T) { tc.volMode = uploader.PersistentVolumeFilesystem } BackupFunc = tc.hookBackupFunc - _, _, _, _, err := kp.RunBackup(t.Context(), "var", "", nil, false, "", tc.volMode, map[string]string{}, &updater) + _, _, _, _, err := kp.RunBackup(t.Context(), "var", "", nil, false, "", CBTParam{}, tc.volMode, map[string]string{}, &updater) if tc.notError { assert.NoError(t, err) } else { diff --git a/pkg/uploader/provider/mocks/Provider.go b/pkg/uploader/provider/mocks/Provider.go index d4c3e83c3..71e60b84e 100644 --- a/pkg/uploader/provider/mocks/Provider.go +++ b/pkg/uploader/provider/mocks/Provider.go @@ -1,115 +1,17 @@ -// Code generated by mockery v2.53.5. DO NOT EDIT. +// Code generated by mockery; DO NOT EDIT. +// github.com/vektra/mockery +// template: testify package mocks import ( - context "context" + "context" mock "github.com/stretchr/testify/mock" - - uploader "github.com/vmware-tanzu/velero/pkg/uploader" + "github.com/vmware-tanzu/velero/pkg/uploader" + "github.com/vmware-tanzu/velero/pkg/uploader/provider" ) -// Provider is an autogenerated mock type for the Provider type -type Provider struct { - mock.Mock -} - -// Close provides a mock function with given fields: ctx -func (_m *Provider) Close(ctx context.Context) error { - ret := _m.Called(ctx) - - if len(ret) == 0 { - panic("no return value specified for Close") - } - - var r0 error - if rf, ok := ret.Get(0).(func(context.Context) error); ok { - r0 = rf(ctx) - } else { - r0 = ret.Error(0) - } - - return r0 -} - -// RunBackup provides a mock function with given fields: ctx, path, realSource, tags, forceFull, parentSnapshot, volMode, uploaderCfg, updater -func (_m *Provider) RunBackup(ctx context.Context, path string, realSource string, tags map[string]string, forceFull bool, parentSnapshot string, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, updater uploader.ProgressUpdater) (string, bool, int64, int64, error) { - ret := _m.Called(ctx, path, realSource, tags, forceFull, parentSnapshot, volMode, uploaderCfg, updater) - - if len(ret) == 0 { - panic("no return value specified for RunBackup") - } - - var r0 string - var r1 bool - var r2 int64 - var r3 int64 - var r4 error - if rf, ok := ret.Get(0).(func(context.Context, string, string, map[string]string, bool, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) (string, bool, int64, int64, error)); ok { - return rf(ctx, path, realSource, tags, forceFull, parentSnapshot, volMode, uploaderCfg, updater) - } - if rf, ok := ret.Get(0).(func(context.Context, string, string, map[string]string, bool, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) string); ok { - r0 = rf(ctx, path, realSource, tags, forceFull, parentSnapshot, volMode, uploaderCfg, updater) - } else { - r0 = ret.Get(0).(string) - } - - if rf, ok := ret.Get(1).(func(context.Context, string, string, map[string]string, bool, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) bool); ok { - r1 = rf(ctx, path, realSource, tags, forceFull, parentSnapshot, volMode, uploaderCfg, updater) - } else { - r1 = ret.Get(1).(bool) - } - - if rf, ok := ret.Get(2).(func(context.Context, string, string, map[string]string, bool, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) int64); ok { - r2 = rf(ctx, path, realSource, tags, forceFull, parentSnapshot, volMode, uploaderCfg, updater) - } else { - r2 = ret.Get(2).(int64) - } - - if rf, ok := ret.Get(3).(func(context.Context, string, string, map[string]string, bool, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) int64); ok { - r3 = rf(ctx, path, realSource, tags, forceFull, parentSnapshot, volMode, uploaderCfg, updater) - } else { - r3 = ret.Get(3).(int64) - } - - if rf, ok := ret.Get(4).(func(context.Context, string, string, map[string]string, bool, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) error); ok { - r4 = rf(ctx, path, realSource, tags, forceFull, parentSnapshot, volMode, uploaderCfg, updater) - } else { - r4 = ret.Error(4) - } - - return r0, r1, r2, r3, r4 -} - -// RunRestore provides a mock function with given fields: ctx, snapshotID, volumePath, volMode, uploaderConfig, updater -func (_m *Provider) RunRestore(ctx context.Context, snapshotID string, volumePath string, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater) (int64, error) { - ret := _m.Called(ctx, snapshotID, volumePath, volMode, uploaderConfig, updater) - - if len(ret) == 0 { - panic("no return value specified for RunRestore") - } - - var r0 int64 - var r1 error - if rf, ok := ret.Get(0).(func(context.Context, string, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) (int64, error)); ok { - return rf(ctx, snapshotID, volumePath, volMode, uploaderConfig, updater) - } - if rf, ok := ret.Get(0).(func(context.Context, string, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) int64); ok { - r0 = rf(ctx, snapshotID, volumePath, volMode, uploaderConfig, updater) - } else { - r0 = ret.Get(0).(int64) - } - - if rf, ok := ret.Get(1).(func(context.Context, string, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) error); ok { - r1 = rf(ctx, snapshotID, volumePath, volMode, uploaderConfig, updater) - } else { - r1 = ret.Error(1) - } - - return r0, r1 -} - // NewProvider creates a new instance of Provider. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations. // The first argument is typically a *testing.T value. func NewProvider(t interface { @@ -123,3 +25,289 @@ func NewProvider(t interface { return mock } + +// Provider is an autogenerated mock type for the Provider type +type Provider struct { + mock.Mock +} + +type Provider_Expecter struct { + mock *mock.Mock +} + +func (_m *Provider) EXPECT() *Provider_Expecter { + return &Provider_Expecter{mock: &_m.Mock} +} + +// Close provides a mock function for the type Provider +func (_mock *Provider) Close(ctx context.Context) error { + ret := _mock.Called(ctx) + + if len(ret) == 0 { + panic("no return value specified for Close") + } + + var r0 error + if returnFunc, ok := ret.Get(0).(func(context.Context) error); ok { + r0 = returnFunc(ctx) + } else { + r0 = ret.Error(0) + } + return r0 +} + +// Provider_Close_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Close' +type Provider_Close_Call struct { + *mock.Call +} + +// Close is a helper method to define mock.On call +// - ctx context.Context +func (_e *Provider_Expecter) Close(ctx interface{}) *Provider_Close_Call { + return &Provider_Close_Call{Call: _e.mock.On("Close", ctx)} +} + +func (_c *Provider_Close_Call) Run(run func(ctx context.Context)) *Provider_Close_Call { + _c.Call.Run(func(args mock.Arguments) { + var arg0 context.Context + if args[0] != nil { + arg0 = args[0].(context.Context) + } + run( + arg0, + ) + }) + return _c +} + +func (_c *Provider_Close_Call) Return(err error) *Provider_Close_Call { + _c.Call.Return(err) + return _c +} + +func (_c *Provider_Close_Call) RunAndReturn(run func(ctx context.Context) error) *Provider_Close_Call { + _c.Call.Return(run) + return _c +} + +// RunBackup provides a mock function for the type Provider +func (_mock *Provider) RunBackup(ctx context.Context, path string, realSource string, tags map[string]string, forceFull bool, parentSnapshot string, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, updater uploader.ProgressUpdater) (string, bool, int64, int64, error) { + ret := _mock.Called(ctx, path, realSource, tags, forceFull, parentSnapshot, cbtParam, volMode, uploaderCfg, updater) + + if len(ret) == 0 { + panic("no return value specified for RunBackup") + } + + var r0 string + var r1 bool + var r2 int64 + var r3 int64 + var r4 error + if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, map[string]string, bool, string, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) (string, bool, int64, int64, error)); ok { + return returnFunc(ctx, path, realSource, tags, forceFull, parentSnapshot, cbtParam, volMode, uploaderCfg, updater) + } + if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, map[string]string, bool, string, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) string); ok { + r0 = returnFunc(ctx, path, realSource, tags, forceFull, parentSnapshot, cbtParam, volMode, uploaderCfg, updater) + } else { + r0 = ret.Get(0).(string) + } + if returnFunc, ok := ret.Get(1).(func(context.Context, string, string, map[string]string, bool, string, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) bool); ok { + r1 = returnFunc(ctx, path, realSource, tags, forceFull, parentSnapshot, cbtParam, volMode, uploaderCfg, updater) + } else { + r1 = ret.Get(1).(bool) + } + if returnFunc, ok := ret.Get(2).(func(context.Context, string, string, map[string]string, bool, string, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) int64); ok { + r2 = returnFunc(ctx, path, realSource, tags, forceFull, parentSnapshot, cbtParam, volMode, uploaderCfg, updater) + } else { + r2 = ret.Get(2).(int64) + } + if returnFunc, ok := ret.Get(3).(func(context.Context, string, string, map[string]string, bool, string, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) int64); ok { + r3 = returnFunc(ctx, path, realSource, tags, forceFull, parentSnapshot, cbtParam, volMode, uploaderCfg, updater) + } else { + r3 = ret.Get(3).(int64) + } + if returnFunc, ok := ret.Get(4).(func(context.Context, string, string, map[string]string, bool, string, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) error); ok { + r4 = returnFunc(ctx, path, realSource, tags, forceFull, parentSnapshot, cbtParam, volMode, uploaderCfg, updater) + } else { + r4 = ret.Error(4) + } + return r0, r1, r2, r3, r4 +} + +// Provider_RunBackup_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'RunBackup' +type Provider_RunBackup_Call struct { + *mock.Call +} + +// RunBackup is a helper method to define mock.On call +// - ctx context.Context +// - path string +// - realSource string +// - tags map[string]string +// - forceFull bool +// - parentSnapshot string +// - cbtParam provider.CBTParam +// - volMode uploader.PersistentVolumeMode +// - uploaderCfg map[string]string +// - updater uploader.ProgressUpdater +func (_e *Provider_Expecter) RunBackup(ctx interface{}, path interface{}, realSource interface{}, tags interface{}, forceFull interface{}, parentSnapshot interface{}, cbtParam interface{}, volMode interface{}, uploaderCfg interface{}, updater interface{}) *Provider_RunBackup_Call { + return &Provider_RunBackup_Call{Call: _e.mock.On("RunBackup", ctx, path, realSource, tags, forceFull, parentSnapshot, cbtParam, volMode, uploaderCfg, updater)} +} + +func (_c *Provider_RunBackup_Call) Run(run func(ctx context.Context, path string, realSource string, tags map[string]string, forceFull bool, parentSnapshot string, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, updater uploader.ProgressUpdater)) *Provider_RunBackup_Call { + _c.Call.Run(func(args mock.Arguments) { + var arg0 context.Context + if args[0] != nil { + arg0 = args[0].(context.Context) + } + var arg1 string + if args[1] != nil { + arg1 = args[1].(string) + } + var arg2 string + if args[2] != nil { + arg2 = args[2].(string) + } + var arg3 map[string]string + if args[3] != nil { + arg3 = args[3].(map[string]string) + } + var arg4 bool + if args[4] != nil { + arg4 = args[4].(bool) + } + var arg5 string + if args[5] != nil { + arg5 = args[5].(string) + } + var arg6 provider.CBTParam + if args[6] != nil { + arg6 = args[6].(provider.CBTParam) + } + var arg7 uploader.PersistentVolumeMode + if args[7] != nil { + arg7 = args[7].(uploader.PersistentVolumeMode) + } + var arg8 map[string]string + if args[8] != nil { + arg8 = args[8].(map[string]string) + } + var arg9 uploader.ProgressUpdater + if args[9] != nil { + arg9 = args[9].(uploader.ProgressUpdater) + } + run( + arg0, + arg1, + arg2, + arg3, + arg4, + arg5, + arg6, + arg7, + arg8, + arg9, + ) + }) + return _c +} + +func (_c *Provider_RunBackup_Call) Return(s string, b bool, n int64, n1 int64, err error) *Provider_RunBackup_Call { + _c.Call.Return(s, b, n, n1, err) + return _c +} + +func (_c *Provider_RunBackup_Call) RunAndReturn(run func(ctx context.Context, path string, realSource string, tags map[string]string, forceFull bool, parentSnapshot string, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, updater uploader.ProgressUpdater) (string, bool, int64, int64, error)) *Provider_RunBackup_Call { + _c.Call.Return(run) + return _c +} + +// RunRestore provides a mock function for the type Provider +func (_mock *Provider) RunRestore(ctx context.Context, snapshotID string, volumePath string, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater) (int64, error) { + ret := _mock.Called(ctx, snapshotID, volumePath, volMode, uploaderConfig, updater) + + if len(ret) == 0 { + panic("no return value specified for RunRestore") + } + + var r0 int64 + var r1 error + if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) (int64, error)); ok { + return returnFunc(ctx, snapshotID, volumePath, volMode, uploaderConfig, updater) + } + if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) int64); ok { + r0 = returnFunc(ctx, snapshotID, volumePath, volMode, uploaderConfig, updater) + } else { + r0 = ret.Get(0).(int64) + } + if returnFunc, ok := ret.Get(1).(func(context.Context, string, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) error); ok { + r1 = returnFunc(ctx, snapshotID, volumePath, volMode, uploaderConfig, updater) + } else { + r1 = ret.Error(1) + } + return r0, r1 +} + +// Provider_RunRestore_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'RunRestore' +type Provider_RunRestore_Call struct { + *mock.Call +} + +// RunRestore is a helper method to define mock.On call +// - ctx context.Context +// - snapshotID string +// - volumePath string +// - volMode uploader.PersistentVolumeMode +// - uploaderConfig map[string]string +// - updater uploader.ProgressUpdater +func (_e *Provider_Expecter) RunRestore(ctx interface{}, snapshotID interface{}, volumePath interface{}, volMode interface{}, uploaderConfig interface{}, updater interface{}) *Provider_RunRestore_Call { + return &Provider_RunRestore_Call{Call: _e.mock.On("RunRestore", ctx, snapshotID, volumePath, volMode, uploaderConfig, updater)} +} + +func (_c *Provider_RunRestore_Call) Run(run func(ctx context.Context, snapshotID string, volumePath string, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater)) *Provider_RunRestore_Call { + _c.Call.Run(func(args mock.Arguments) { + var arg0 context.Context + if args[0] != nil { + arg0 = args[0].(context.Context) + } + var arg1 string + if args[1] != nil { + arg1 = args[1].(string) + } + var arg2 string + if args[2] != nil { + arg2 = args[2].(string) + } + var arg3 uploader.PersistentVolumeMode + if args[3] != nil { + arg3 = args[3].(uploader.PersistentVolumeMode) + } + var arg4 map[string]string + if args[4] != nil { + arg4 = args[4].(map[string]string) + } + var arg5 uploader.ProgressUpdater + if args[5] != nil { + arg5 = args[5].(uploader.ProgressUpdater) + } + run( + arg0, + arg1, + arg2, + arg3, + arg4, + arg5, + ) + }) + return _c +} + +func (_c *Provider_RunRestore_Call) Return(n int64, err error) *Provider_RunRestore_Call { + _c.Call.Return(n, err) + return _c +} + +func (_c *Provider_RunRestore_Call) RunAndReturn(run func(ctx context.Context, snapshotID string, volumePath string, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater) (int64, error)) *Provider_RunRestore_Call { + _c.Call.Return(run) + return _c +} diff --git a/pkg/uploader/provider/provider.go b/pkg/uploader/provider/provider.go index 95a34b1a0..766621261 100644 --- a/pkg/uploader/provider/provider.go +++ b/pkg/uploader/provider/provider.go @@ -29,6 +29,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/uploader" ) @@ -37,6 +38,11 @@ const backupProgressCheckInterval = 10 * time.Second var ErrorCanceled error = errors.New("uploader is canceled") +type CBTParam struct { + Source cbtservice.SourceInfo + Service cbtservice.Service +} + // Provider which is designed for one pod volume to do the backup or restore type Provider interface { // RunBackup which will do backup for one specific volume and return snapshotID, isSnapshotEmpty, error @@ -48,6 +54,7 @@ type Provider interface { tags map[string]string, forceFull bool, parentSnapshot string, + cbtParam CBTParam, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, updater uploader.ProgressUpdater) (string, bool, int64, int64, error) From 0be9fc7d091a84be0c56bae0743a9cadc6d64f86 Mon Sep 17 00:00:00 2001 From: Lyndon-Li Date: Mon, 11 May 2026 17:31:40 +0800 Subject: [PATCH 2/4] add block uploader provider Signed-off-by: Lyndon-Li --- pkg/uploader/provider/block.go | 116 +++++++++++++++++ pkg/uploader/provider/block_test.go | 187 ++++++++++++++++++++++++++++ pkg/uploader/types.go | 3 +- 3 files changed, 305 insertions(+), 1 deletion(-) create mode 100644 pkg/uploader/provider/block.go create mode 100644 pkg/uploader/provider/block_test.go diff --git a/pkg/uploader/provider/block.go b/pkg/uploader/provider/block.go new file mode 100644 index 000000000..c38387744 --- /dev/null +++ b/pkg/uploader/provider/block.go @@ -0,0 +1,116 @@ +/* +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 provider + +import ( + "context" + "strings" + + "github.com/pkg/errors" + "github.com/sirupsen/logrus" + "github.com/vmware-tanzu/velero/internal/credentials" + velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1" + repokeys "github.com/vmware-tanzu/velero/pkg/repository/keys" + "github.com/vmware-tanzu/velero/pkg/repository/udmrepo" + "github.com/vmware-tanzu/velero/pkg/uploader" +) + +type blockProvider struct { + requestorType string + bkRepo udmrepo.BackupRepo + credGetter *credentials.CredentialGetter + log logrus.FieldLogger + canceling int32 +} + +// NewBlockUploaderProvider initialized with open or create a repository +func NewBlockUploaderProvider( + requestorType string, + ctx context.Context, + credGetter *credentials.CredentialGetter, + backupRepo *velerov1api.BackupRepository, + log logrus.FieldLogger, +) (Provider, error) { + bp := &blockProvider{ + requestorType: requestorType, + log: log, + credGetter: credGetter, + } + + repoUID := string(backupRepo.GetUID()) + repoOpt, err := udmrepo.NewRepoOptions( + udmrepo.WithPassword(bp, ""), + udmrepo.WithConfigFile("", repoUID), + udmrepo.WithDescription("Initial velero block uploader provider"), + ) + if err != nil { + return nil, errors.Wrapf(err, "error to get repo options") + } + + repoSvc := BackupRepoServiceCreateFunc(backupRepo.Spec.RepositoryType, log) + log.WithField("repoUID", repoUID).Info("Opening backup repo") + + bp.bkRepo, err = repoSvc.Open(ctx, *repoOpt) + if err != nil { + return nil, errors.Wrapf(err, "Failed to find backup repository") + } + + return bp, nil +} + +func (bp *blockProvider) Close(ctx context.Context) error { + return bp.bkRepo.Close(ctx) +} + +func (bp *blockProvider) GetPassword(param any) (string, error) { + if bp.credGetter.FromSecret == nil { + return "", errors.New("invalid credentials interface") + } + rawPass, err := bp.credGetter.FromSecret.Get(repokeys.RepoKeySelector()) + if err != nil { + return "", errors.Wrap(err, "error to get password") + } + + return strings.TrimSpace(rawPass), nil +} + +// TODO: implement in the following PRs +func (bp *blockProvider) RunBackup( + ctx context.Context, + path string, + realSource string, + tags map[string]string, + forceFull bool, + parentSnapshot string, + cbtParam CBTParam, + volMode uploader.PersistentVolumeMode, + uploaderCfg map[string]string, + updater uploader.ProgressUpdater) (string, bool, int64, int64, error) { + + return "", false, 0, 0, nil +} + +// TODO: implement in the following PRs +func (bp *blockProvider) RunRestore( + ctx context.Context, + snapshotID string, + volumePath string, + volMode uploader.PersistentVolumeMode, + uploaderCfg map[string]string, + updater uploader.ProgressUpdater) (int64, error) { + return 0, nil +} diff --git a/pkg/uploader/provider/block_test.go b/pkg/uploader/provider/block_test.go new file mode 100644 index 000000000..3584e35f1 --- /dev/null +++ b/pkg/uploader/provider/block_test.go @@ -0,0 +1,187 @@ +/* +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 provider + +import ( + "testing" + + "github.com/pkg/errors" + "github.com/sirupsen/logrus" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + corev1api "k8s.io/api/core/v1" + + "github.com/vmware-tanzu/velero/internal/credentials" + "github.com/vmware-tanzu/velero/internal/credentials/mocks" + velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1" + "github.com/vmware-tanzu/velero/pkg/repository" + "github.com/vmware-tanzu/velero/pkg/repository/udmrepo" + udmrepomocks "github.com/vmware-tanzu/velero/pkg/repository/udmrepo/mocks" +) + +func TestNewBlockUploaderProvider(t *testing.T) { + requestorType := "testRequestor" + ctx := t.Context() + backupRepo := repository.NewBackupRepository(velerov1api.DefaultNamespace, repository.BackupRepositoryKey{VolumeNamespace: "fake-volume-ns-02", BackupLocation: "fake-bsl-02", RepositoryType: "fake-repository-type-02"}) + mockLog := logrus.New() + + testCases := []struct { + name string + mockCredGetter *mocks.SecretStore + mockBackupRepoService udmrepo.BackupRepoService + expectedError string + }{ + { + name: "Success", + mockCredGetter: func() *mocks.SecretStore { + mockCredGetter := &mocks.SecretStore{} + mockCredGetter.On("Get", mock.Anything).Return("test", nil) + return mockCredGetter + }(), + mockBackupRepoService: func() udmrepo.BackupRepoService { + backupRepoService := &udmrepomocks.BackupRepoService{} + var backupRepo udmrepo.BackupRepo + backupRepoService.On("Open", t.Context(), mock.Anything).Return(backupRepo, nil) + return backupRepoService + }(), + expectedError: "", + }, + { + name: "Error to get repo options", + mockCredGetter: func() *mocks.SecretStore { + mockCredGetter := &mocks.SecretStore{} + mockCredGetter.On("Get", mock.Anything).Return("test", errors.New("failed to get password")) + return mockCredGetter + }(), + mockBackupRepoService: func() udmrepo.BackupRepoService { + backupRepoService := &udmrepomocks.BackupRepoService{} + var backupRepo udmrepo.BackupRepo + backupRepoService.On("Open", t.Context(), mock.Anything).Return(backupRepo, nil) + return backupRepoService + }(), + expectedError: "error to get repo options", + }, + { + name: "Error open repository service", + mockCredGetter: func() *mocks.SecretStore { + mockCredGetter := &mocks.SecretStore{} + mockCredGetter.On("Get", mock.Anything).Return("test", nil) + return mockCredGetter + }(), + mockBackupRepoService: func() udmrepo.BackupRepoService { + backupRepoService := &udmrepomocks.BackupRepoService{} + var backupRepo udmrepo.BackupRepo + backupRepoService.On("Open", t.Context(), mock.Anything).Return(backupRepo, errors.New("failed to init repository")) + return backupRepoService + }(), + expectedError: "Failed to find backup repository", + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + credGetter := &credentials.CredentialGetter{FromSecret: tc.mockCredGetter} + BackupRepoServiceCreateFunc = func(string, logrus.FieldLogger) udmrepo.BackupRepoService { + return tc.mockBackupRepoService + } + _, err := NewBlockUploaderProvider(requestorType, ctx, credGetter, backupRepo, mockLog) + + if tc.expectedError != "" { + require.ErrorContains(t, err, tc.expectedError) + } else { + require.NoError(t, err) + } + + tc.mockCredGetter.AssertExpectations(t) + }) + } +} + +func TestBlockProviderClose(t *testing.T) { + mockBRepo := udmrepomocks.NewBackupRepo(t) + mockBRepo.On("Close", mock.Anything).Return(nil) + + bp := &blockProvider{ + bkRepo: mockBRepo, + } + + err := bp.Close(t.Context()) + assert.NoError(t, err) + mockBRepo.AssertExpectations(t) +} + +func TestBlockProviderGetPassword(t *testing.T) { + testCases := []struct { + name string + emptySecret bool + credGetterFunc func(*mocks.SecretStore, *corev1api.SecretKeySelector) + expectError bool + expectedPass string + }{ + { + name: "valid credentials interface", + credGetterFunc: func(ss *mocks.SecretStore, selector *corev1api.SecretKeySelector) { + ss.On("Get", selector).Return("test", nil) + }, + expectError: false, + expectedPass: "test", + }, + { + name: "empty from secret", + emptySecret: true, + expectError: true, + expectedPass: "", + }, + { + name: "ErrorGettingPassword", + credGetterFunc: func(ss *mocks.SecretStore, selector *corev1api.SecretKeySelector) { + ss.On("Get", selector).Return("", errors.New("error getting password")) + }, + expectError: true, + expectedPass: "", + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + credGetter := &credentials.CredentialGetter{} + mockCredGetter := &mocks.SecretStore{} + if !tc.emptySecret { + credGetter.FromSecret = mockCredGetter + } + repoKeySelector := &corev1api.SecretKeySelector{LocalObjectReference: corev1api.LocalObjectReference{Name: "velero-repo-credentials"}, Key: "repository-password"} + + if tc.credGetterFunc != nil { + tc.credGetterFunc(mockCredGetter, repoKeySelector) + } + + bp := &blockProvider{ + credGetter: credGetter, + } + + password, err := bp.GetPassword(nil) + if tc.expectError { + require.Error(t, err, "Expected an error") + } else { + require.NoError(t, err, "Expected no error") + } + + assert.Equal(t, tc.expectedPass, password, "Expected password to match") + }) + } +} diff --git a/pkg/uploader/types.go b/pkg/uploader/types.go index 52f8ca5bf..cc3321012 100644 --- a/pkg/uploader/types.go +++ b/pkg/uploader/types.go @@ -23,6 +23,7 @@ import ( const ( KopiaType = "kopia" + BlockType = "velero-block" SnapshotRequesterTag = "snapshot-requester" SnapshotUploaderTag = "snapshot-uploader" ) @@ -40,7 +41,7 @@ const ( // It will return an error if it's invalid. func ValidateUploaderType(t string) (string, error) { t = strings.TrimSpace(t) - if t != KopiaType { + if t != KopiaType && t != BlockType { return "", fmt.Errorf("invalid uploader type '%s', valid type: '%s'", t, KopiaType) } From 23463147297a5a12a9754a505ec07fdfb17f0dfb Mon Sep 17 00:00:00 2001 From: Lyndon-Li Date: Thu, 14 May 2026 16:17:51 +0800 Subject: [PATCH 3/4] call block uploader provider creation Signed-off-by: Lyndon-Li --- pkg/uploader/provider/provider.go | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/pkg/uploader/provider/provider.go b/pkg/uploader/provider/provider.go index 766621261..0c1caaffe 100644 --- a/pkg/uploader/provider/provider.go +++ b/pkg/uploader/provider/provider.go @@ -91,9 +91,13 @@ func NewUploaderProvider( if credGetter.FromFile == nil { return nil, errors.New("uninitialized FileStore credential is not supported") } - if uploaderType == uploader.KopiaType { + + switch uploaderType { + case uploader.KopiaType: return NewKopiaUploaderProvider(requesterType, ctx, credGetter, backupRepo, log) - } else { + case uploader.BlockType: + return NewBlockUploaderProvider(requesterType, ctx, credGetter, backupRepo, log) + default: return nil, errors.Errorf("unsupported uploader type %v", uploaderType) } } From 9f0194d8fe1a857ea639af2015d6456afb051015 Mon Sep 17 00:00:00 2001 From: Lyndon-Li Date: Thu, 21 May 2026 13:24:30 +0800 Subject: [PATCH 4/4] Uploader interface fo block data mover Signed-off-by: Lyndon-Li --- changelogs/unreleased/9807-Lyndon-Li | 1 + pkg/podvolume/backupper_test.go | 2 +- pkg/uploader/provider/block.go | 7 +++---- pkg/uploader/provider/block_test.go | 2 +- pkg/uploader/types.go | 2 +- pkg/uploader/types_test.go | 2 +- 6 files changed, 8 insertions(+), 8 deletions(-) create mode 100644 changelogs/unreleased/9807-Lyndon-Li diff --git a/changelogs/unreleased/9807-Lyndon-Li b/changelogs/unreleased/9807-Lyndon-Li new file mode 100644 index 000000000..58df214f7 --- /dev/null +++ b/changelogs/unreleased/9807-Lyndon-Li @@ -0,0 +1 @@ +Uploader interface for block data mover \ No newline at end of file diff --git a/pkg/podvolume/backupper_test.go b/pkg/podvolume/backupper_test.go index fba789664..e6042ede1 100644 --- a/pkg/podvolume/backupper_test.go +++ b/pkg/podvolume/backupper_test.go @@ -396,7 +396,7 @@ func TestBackupPodVolumes(t *testing.T) { }, uploaderType: "fake-uploader-type", errs: []string{ - "invalid uploader type 'fake-uploader-type', valid type: 'kopia'", + "invalid uploader type 'fake-uploader-type', valid types: 'kopia', 'velero-block'", }, }, { diff --git a/pkg/uploader/provider/block.go b/pkg/uploader/provider/block.go index c38387744..b1eed428a 100644 --- a/pkg/uploader/provider/block.go +++ b/pkg/uploader/provider/block.go @@ -22,6 +22,7 @@ import ( "github.com/pkg/errors" "github.com/sirupsen/logrus" + "github.com/vmware-tanzu/velero/internal/credentials" velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1" repokeys "github.com/vmware-tanzu/velero/pkg/repository/keys" @@ -34,7 +35,6 @@ type blockProvider struct { bkRepo udmrepo.BackupRepo credGetter *credentials.CredentialGetter log logrus.FieldLogger - canceling int32 } // NewBlockUploaderProvider initialized with open or create a repository @@ -100,8 +100,7 @@ func (bp *blockProvider) RunBackup( volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, updater uploader.ProgressUpdater) (string, bool, int64, int64, error) { - - return "", false, 0, 0, nil + return "", false, 0, 0, errors.New("block backup not implemented") } // TODO: implement in the following PRs @@ -112,5 +111,5 @@ func (bp *blockProvider) RunRestore( volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, updater uploader.ProgressUpdater) (int64, error) { - return 0, nil + return 0, errors.New("block restore not implemented") } diff --git a/pkg/uploader/provider/block_test.go b/pkg/uploader/provider/block_test.go index 3584e35f1..a48b0a862 100644 --- a/pkg/uploader/provider/block_test.go +++ b/pkg/uploader/provider/block_test.go @@ -121,7 +121,7 @@ func TestBlockProviderClose(t *testing.T) { } err := bp.Close(t.Context()) - assert.NoError(t, err) + require.NoError(t, err) mockBRepo.AssertExpectations(t) } diff --git a/pkg/uploader/types.go b/pkg/uploader/types.go index cc3321012..12ff1dc52 100644 --- a/pkg/uploader/types.go +++ b/pkg/uploader/types.go @@ -42,7 +42,7 @@ const ( func ValidateUploaderType(t string) (string, error) { t = strings.TrimSpace(t) if t != KopiaType && t != BlockType { - return "", fmt.Errorf("invalid uploader type '%s', valid type: '%s'", t, KopiaType) + return "", fmt.Errorf("invalid uploader type '%s', valid types: '%s', '%s'", t, KopiaType, BlockType) } return "", nil diff --git a/pkg/uploader/types_test.go b/pkg/uploader/types_test.go index dcdedcd62..0ae93d912 100644 --- a/pkg/uploader/types_test.go +++ b/pkg/uploader/types_test.go @@ -23,7 +23,7 @@ func TestValidateUploaderType(t *testing.T) { { "'anything_else' is invalid", "anything_else", - "invalid uploader type 'anything_else', valid type: 'kopia'", + "invalid uploader type 'anything_else', valid types: 'kopia', 'velero-block'", "", }, }