Merge pull request #9807 from Lyndon-Li/uploader-interface-fo-block-data-mover

Uploader interface for block data mover

changelogs/unreleased/9807-Lyndon-Li

@@ -0,0 +1 @@
+Uploader interface for block data mover

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)

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

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'",
 			},
 		},
 		{

pkg/uploader/provider/block.go

@@ -0,0 +1,115 @@
+/*
+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
+}
+
+// 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, errors.New("block backup not implemented")
+}
+
+// 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, errors.New("block restore not implemented")
+}

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())
+	require.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")
+		})
+	}
+}

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) {

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 {

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
+}

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)
@@ -84,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)
 	}
 }

pkg/uploader/types.go

@@ -23,6 +23,7 @@ import (
 
 const (
 	KopiaType            = "kopia"
+	BlockType            = "velero-block"
 	SnapshotRequesterTag = "snapshot-requester"
 	SnapshotUploaderTag  = "snapshot-uploader"
 )
@@ -40,8 +41,8 @@ const (
 // It will return an error if it's invalid.
 func ValidateUploaderType(t string) (string, error) {
 	t = strings.TrimSpace(t)
-	if t != KopiaType {
-		return "", fmt.Errorf("invalid uploader type '%s', valid type: '%s'", t, KopiaType)
+	if t != KopiaType && t != BlockType {
+		return "", fmt.Errorf("invalid uploader type '%s', valid types: '%s', '%s'", t, KopiaType, BlockType)
 	}
 
 	return "", nil

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'",
 			"",
 		},
 	}