repo init

Signed-off-by: Lyndon-Li <lyonghui@vmware.com>

pkg/repository/provider/provider.go

@@ -28,6 +28,7 @@ type RepoParam struct {
 	BackupRepo     *velerov1api.BackupRepository
 }
 
+// Provider defines the methods to manipulate a backup repository
 type Provider interface {
 	//InitRepo is to initialize a repository from a new storage place
 	InitRepo(ctx context.Context, param RepoParam) error

pkg/repository/provider/unified_repo.go

@@ -62,6 +62,8 @@ const (
 	repoOpDescFullMaintain  = "full maintenance"
 	repoOpDescQuickMaintain = "quick maintenance"
 	repoOpDescForget        = "forget"
+
+	repoConnectDesc = "unfied repo"
 )
 
 // NewUnifiedRepoProvider creates the service provider for Unified Repo
@@ -92,8 +94,14 @@ func (urp *unifiedRepoProvider) InitRepo(ctx context.Context, param RepoParam) e
 	repoOption, err := udmrepo.NewRepoOptions(
 		udmrepo.WithPassword(urp, param),
 		udmrepo.WithConfigFile(urp.workPath, string(param.BackupLocation.UID)),
+		udmrepo.WithGenOptions(
+			map[string]string{
+				udmrepo.GenOptionOwnerName:   udmrepo.GetRepoUser(),
+				udmrepo.GenOptionOwnerDomain: udmrepo.GetRepoDomain(),
+			},
+		),
 		udmrepo.WithStoreOptions(urp, param),
-		udmrepo.WithDescription(repoOpDescFullMaintain),
+		udmrepo.WithDescription(repoConnectDesc),
 	)
 
 	if err != nil {
@@ -121,8 +129,14 @@ func (urp *unifiedRepoProvider) ConnectToRepo(ctx context.Context, param RepoPar
 	repoOption, err := udmrepo.NewRepoOptions(
 		udmrepo.WithPassword(urp, param),
 		udmrepo.WithConfigFile(urp.workPath, string(param.BackupLocation.UID)),
+		udmrepo.WithGenOptions(
+			map[string]string{
+				udmrepo.GenOptionOwnerName:   udmrepo.GetRepoUser(),
+				udmrepo.GenOptionOwnerDomain: udmrepo.GetRepoDomain(),
+			},
+		),
 		udmrepo.WithStoreOptions(urp, param),
-		udmrepo.WithDescription(repoOpDescFullMaintain),
+		udmrepo.WithDescription(repoConnectDesc),
 	)
 
 	if err != nil {
@@ -150,8 +164,14 @@ func (urp *unifiedRepoProvider) PrepareRepo(ctx context.Context, param RepoParam
 	repoOption, err := udmrepo.NewRepoOptions(
 		udmrepo.WithPassword(urp, param),
 		udmrepo.WithConfigFile(urp.workPath, string(param.BackupLocation.UID)),
+		udmrepo.WithGenOptions(
+			map[string]string{
+				udmrepo.GenOptionOwnerName:   udmrepo.GetRepoUser(),
+				udmrepo.GenOptionOwnerDomain: udmrepo.GetRepoDomain(),
+			},
+		),
 		udmrepo.WithStoreOptions(urp, param),
-		udmrepo.WithDescription(repoOpDescFullMaintain),
+		udmrepo.WithDescription(repoConnectDesc),
 	)
 
 	if err != nil {
@@ -185,7 +205,11 @@ func (urp *unifiedRepoProvider) PruneRepo(ctx context.Context, param RepoParam)
 	repoOption, err := udmrepo.NewRepoOptions(
 		udmrepo.WithPassword(urp, param),
 		udmrepo.WithConfigFile(urp.workPath, string(param.BackupLocation.UID)),
-		udmrepo.WithGenOptions(map[string]string{udmrepo.GenOptionMaintainMode: udmrepo.GenOptionMaintainFull}),
+		udmrepo.WithGenOptions(
+			map[string]string{
+				udmrepo.GenOptionMaintainMode: udmrepo.GenOptionMaintainFull,
+			},
+		),
 		udmrepo.WithDescription(repoOpDescFullMaintain),
 	)
 
@@ -214,7 +238,11 @@ func (urp *unifiedRepoProvider) PruneRepoQuick(ctx context.Context, param RepoPa
 	repoOption, err := udmrepo.NewRepoOptions(
 		udmrepo.WithPassword(urp, param),
 		udmrepo.WithConfigFile(urp.workPath, string(param.BackupLocation.UID)),
-		udmrepo.WithGenOptions(map[string]string{udmrepo.GenOptionMaintainMode: udmrepo.GenOptionMaintainQuick}),
+		udmrepo.WithGenOptions(
+			map[string]string{
+				udmrepo.GenOptionMaintainMode: udmrepo.GenOptionMaintainQuick,
+			},
+		),
 		udmrepo.WithDescription(repoOpDescQuickMaintain),
 	)
 
@@ -280,7 +308,7 @@ func (urp *unifiedRepoProvider) Forget(ctx context.Context, snapshotID string, p
 func (urp *unifiedRepoProvider) GetPassword(param interface{}) (string, error) {
 	repoParam, ok := param.(RepoParam)
 	if !ok {
-		return "", errors.New("invalid parameter")
+		return "", errors.Errorf("invalid parameter, expect %T, actual %T", RepoParam{}, param)
 	}
 
 	repoPassword, err := getRepoPassword(urp.credentialGetter.FromSecret, repoParam)
@@ -294,7 +322,7 @@ func (urp *unifiedRepoProvider) GetPassword(param interface{}) (string, error) {
 func (urp *unifiedRepoProvider) GetStoreType(param interface{}) (string, error) {
 	repoParam, ok := param.(RepoParam)
 	if !ok {
-		return "", errors.New("invalid parameter")
+		return "", errors.Errorf("invalid parameter, expect %T, actual %T", RepoParam{}, param)
 	}
 
 	return getStorageType(repoParam.BackupLocation), nil
@@ -303,7 +331,7 @@ func (urp *unifiedRepoProvider) GetStoreType(param interface{}) (string, error)
 func (urp *unifiedRepoProvider) GetStoreOptions(param interface{}) (map[string]string, error) {
 	repoParam, ok := param.(RepoParam)
 	if !ok {
-		return map[string]string{}, errors.New("invalid parameter")
+		return map[string]string{}, errors.Errorf("invalid parameter, expect %T, actual %T", RepoParam{}, param)
 	}
 
 	storeVar, err := funcTable.getStorageVariables(repoParam.BackupLocation, repoParam.BackupRepo.Spec.VolumeNamespace)

pkg/repository/provider/unified_repo_test.go

@@ -503,7 +503,7 @@ func TestGetStoreOptions(t *testing.T) {
 			name:        "wrong param type",
 			repoParam:   struct{}{},
 			expected:    map[string]string{},
-			expectedErr: "invalid parameter",
+			expectedErr: "invalid parameter, expect provider.RepoParam, actual struct {}",
 		},
 		{
 			name: "get storage variable fail",

pkg/repository/udmrepo/kopialib/backend/azure.go

@@ -0,0 +1,60 @@
+/*
+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 backend
+
+import (
+	"context"
+
+	"github.com/kopia/kopia/repo/blob"
+	"github.com/kopia/kopia/repo/blob/azure"
+
+	"github.com/vmware-tanzu/velero/pkg/repository/udmrepo"
+)
+
+type AzureBackend struct {
+	options azure.Options
+}
+
+func (c *AzureBackend) Setup(ctx context.Context, flags map[string]string) error {
+	var err error
+	c.options.Container, err = mustHaveString(udmrepo.StoreOptionOssBucket, flags)
+	if err != nil {
+		return err
+	}
+
+	c.options.StorageAccount, err = mustHaveString(udmrepo.StoreOptionAzureStorageAccount, flags)
+	if err != nil {
+		return err
+	}
+
+	c.options.StorageKey, err = mustHaveString(udmrepo.StoreOptionAzureKey, flags)
+	if err != nil {
+		return err
+	}
+
+	c.options.Prefix = optionalHaveString(udmrepo.StoreOptionPrefix, flags)
+	c.options.SASToken = optionalHaveString(udmrepo.StoreOptionAzureToken, flags)
+	c.options.StorageDomain = optionalHaveString(udmrepo.StoreOptionAzureDomain, flags)
+
+	c.options.Limits = setupLimits(ctx, flags)
+
+	return nil
+}
+
+func (c *AzureBackend) Connect(ctx context.Context, isCreate bool) (blob.Storage, error) {
+	return azure.New(ctx, &c.options)
+}

pkg/repository/udmrepo/kopialib/backend/azure_test.go

@@ -0,0 +1,102 @@
+/*
+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 backend
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/vmware-tanzu/velero/pkg/repository/udmrepo"
+
+	"github.com/kopia/kopia/repo/blob/azure"
+	"github.com/kopia/kopia/repo/blob/throttling"
+)
+
+func TestAzureSetup(t *testing.T) {
+	testCases := []struct {
+		name        string
+		flags       map[string]string
+		expected    azure.Options
+		expectedErr string
+	}{
+		{
+			name:        "must have bucket name",
+			flags:       map[string]string{},
+			expectedErr: "key " + udmrepo.StoreOptionOssBucket + " not found",
+		},
+		{
+			name: "must have storage account",
+			flags: map[string]string{
+				udmrepo.StoreOptionOssBucket: "fake-bucket",
+			},
+			expected: azure.Options{
+				Container: "fake-bucket",
+			},
+			expectedErr: "key " + udmrepo.StoreOptionAzureStorageAccount + " not found",
+		},
+		{
+			name: "must have secret key",
+			flags: map[string]string{
+				udmrepo.StoreOptionOssBucket:           "fake-bucket",
+				udmrepo.StoreOptionAzureStorageAccount: "fake-account",
+			},
+			expected: azure.Options{
+				Container:      "fake-bucket",
+				StorageAccount: "fake-account",
+			},
+			expectedErr: "key " + udmrepo.StoreOptionAzureKey + " not found",
+		},
+		{
+			name: "with limits",
+			flags: map[string]string{
+				udmrepo.StoreOptionOssBucket:           "fake-bucket",
+				udmrepo.StoreOptionAzureStorageAccount: "fake-account",
+				udmrepo.StoreOptionAzureKey:            "fake-key",
+				udmrepo.ThrottleOptionReadOps:          "100",
+				udmrepo.ThrottleOptionUploadBytes:      "200",
+			},
+			expected: azure.Options{
+				Container:      "fake-bucket",
+				StorageAccount: "fake-account",
+				StorageKey:     "fake-key",
+				Limits: throttling.Limits{
+					ReadsPerSecond:       100,
+					UploadBytesPerSecond: 200,
+				},
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			azFlags := AzureBackend{}
+
+			err := azFlags.Setup(context.Background(), tc.flags)
+
+			require.Equal(t, tc.expected, azFlags.options)
+
+			if tc.expectedErr == "" {
+				assert.NoError(t, err)
+			} else {
+				assert.EqualError(t, err, tc.expectedErr)
+			}
+		})
+	}
+}

pkg/repository/udmrepo/kopialib/backend/backend.go

@@ -0,0 +1,33 @@
+/*
+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 backend
+
+import (
+	"context"
+
+	"github.com/kopia/kopia/repo/blob"
+)
+
+// Store defines the methods for Kopia to establish a connection to
+// the backend storage
+type Store interface {
+	// Setup setups the variables to a specific backend storage
+	Setup(ctx context.Context, flags map[string]string) error
+
+	// Connect connects to a specific backend storage with the storage variables
+	Connect(ctx context.Context, isCreate bool) (blob.Storage, error)
+}

pkg/repository/udmrepo/kopialib/backend/common.go

@@ -0,0 +1,83 @@
+/*
+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 backend
+
+import (
+	"context"
+	"time"
+
+	"github.com/kopia/kopia/repo"
+	"github.com/kopia/kopia/repo/blob"
+	"github.com/kopia/kopia/repo/blob/throttling"
+	"github.com/kopia/kopia/repo/content"
+	"github.com/kopia/kopia/repo/encryption"
+	"github.com/kopia/kopia/repo/hashing"
+	"github.com/kopia/kopia/repo/object"
+	"github.com/kopia/kopia/repo/splitter"
+
+	"github.com/vmware-tanzu/velero/pkg/repository/udmrepo"
+)
+
+const (
+	maxDataCacheMB         = 2000
+	maxMetadataCacheMB     = 2000
+	maxCacheDurationSecond = 30
+)
+
+func setupLimits(ctx context.Context, flags map[string]string) throttling.Limits {
+	return throttling.Limits{
+		DownloadBytesPerSecond: optionalHaveFloat64(ctx, udmrepo.ThrottleOptionDownloadBytes, flags),
+		ListsPerSecond:         optionalHaveFloat64(ctx, udmrepo.ThrottleOptionListOps, flags),
+		ReadsPerSecond:         optionalHaveFloat64(ctx, udmrepo.ThrottleOptionReadOps, flags),
+		UploadBytesPerSecond:   optionalHaveFloat64(ctx, udmrepo.ThrottleOptionUploadBytes, flags),
+		WritesPerSecond:        optionalHaveFloat64(ctx, udmrepo.ThrottleOptionWriteOps, flags),
+	}
+}
+
+// SetupNewRepositoryOptions setups the options when creating a new Kopia repository
+func SetupNewRepositoryOptions(ctx context.Context, flags map[string]string) repo.NewRepositoryOptions {
+	return repo.NewRepositoryOptions{
+		BlockFormat: content.FormattingOptions{
+			Hash:       optionalHaveStringWithDefault(udmrepo.StoreOptionGenHashAlgo, flags, hashing.DefaultAlgorithm),
+			Encryption: optionalHaveStringWithDefault(udmrepo.StoreOptionGenEncryptAlgo, flags, encryption.DefaultAlgorithm),
+		},
+
+		ObjectFormat: object.Format{
+			Splitter: optionalHaveStringWithDefault(udmrepo.StoreOptionGenSplitAlgo, flags, splitter.DefaultAlgorithm),
+		},
+
+		RetentionMode:   blob.RetentionMode(optionalHaveString(udmrepo.StoreOptionGenRetentionMode, flags)),
+		RetentionPeriod: optionalHaveDuration(ctx, udmrepo.StoreOptionGenRetentionPeriod, flags),
+	}
+}
+
+// SetupConnectOptions setups the options when connecting to an existing Kopia repository
+func SetupConnectOptions(ctx context.Context, repoOptions udmrepo.RepoOptions) repo.ConnectOptions {
+	return repo.ConnectOptions{
+		CachingOptions: content.CachingOptions{
+			MaxCacheSizeBytes:         maxDataCacheMB << 20,
+			MaxMetadataCacheSizeBytes: maxMetadataCacheMB << 20,
+			MaxListCacheDuration:      content.DurationSeconds(time.Duration(maxCacheDurationSecond) * time.Second),
+		},
+		ClientOptions: repo.ClientOptions{
+			Hostname:    optionalHaveString(udmrepo.GenOptionOwnerDomain, repoOptions.GeneralOptions),
+			Username:    optionalHaveString(udmrepo.GenOptionOwnerName, repoOptions.GeneralOptions),
+			ReadOnly:    optionalHaveBool(ctx, udmrepo.StoreOptionGenReadOnly, repoOptions.GeneralOptions),
+			Description: repoOptions.Description,
+		},
+	}
+}

pkg/repository/udmrepo/kopialib/backend/file_system.go

@@ -0,0 +1,62 @@
+/*
+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 backend
+
+import (
+	"context"
+	"path/filepath"
+
+	"github.com/kopia/kopia/repo/blob"
+	"github.com/kopia/kopia/repo/blob/filesystem"
+	"github.com/pkg/errors"
+
+	"github.com/vmware-tanzu/velero/pkg/repository/udmrepo"
+)
+
+type FsBackend struct {
+	options filesystem.Options
+}
+
+const (
+	defaultFileMode = 0o600
+	defaultDirMode  = 0o700
+)
+
+func (c *FsBackend) Setup(ctx context.Context, flags map[string]string) error {
+	path, err := mustHaveString(udmrepo.StoreOptionFsPath, flags)
+	if err != nil {
+		return err
+	}
+
+	prefix := optionalHaveString(udmrepo.StoreOptionPrefix, flags)
+
+	c.options.Path = filepath.Join(path, prefix)
+	c.options.FileMode = defaultFileMode
+	c.options.DirectoryMode = defaultDirMode
+
+	c.options.Limits = setupLimits(ctx, flags)
+
+	return nil
+}
+
+func (c *FsBackend) Connect(ctx context.Context, isCreate bool) (blob.Storage, error) {
+	if !filepath.IsAbs(c.options.Path) {
+		return nil, errors.Errorf("filesystem repository path is not absolute, path: %s", c.options.Path)
+	}
+
+	return filesystem.New(ctx, &c.options, isCreate)
+}

pkg/repository/udmrepo/kopialib/backend/gcs.go

@@ -0,0 +1,54 @@
+/*
+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 backend
+
+import (
+	"context"
+
+	"github.com/kopia/kopia/repo/blob"
+	"github.com/kopia/kopia/repo/blob/gcs"
+
+	"github.com/vmware-tanzu/velero/pkg/repository/udmrepo"
+)
+
+type GCSBackend struct {
+	options gcs.Options
+}
+
+func (c *GCSBackend) Setup(ctx context.Context, flags map[string]string) error {
+	var err error
+	c.options.BucketName, err = mustHaveString(udmrepo.StoreOptionOssBucket, flags)
+	if err != nil {
+		return err
+	}
+
+	c.options.ServiceAccountCredentialsFile, err = mustHaveString(udmrepo.StoreOptionCredentialFile, flags)
+	if err != nil {
+		return err
+	}
+
+	c.options.Prefix = optionalHaveString(udmrepo.StoreOptionPrefix, flags)
+	c.options.ReadOnly = optionalHaveBool(ctx, udmrepo.StoreOptionGcsReadonly, flags)
+
+	c.options.Limits = setupLimits(ctx, flags)
+
+	return nil
+}
+
+func (c *GCSBackend) Connect(ctx context.Context, isCreate bool) (blob.Storage, error) {
+	return gcs.New(ctx, &c.options)
+}

pkg/repository/udmrepo/kopialib/backend/gcs_test.go

@@ -0,0 +1,61 @@
+/*
+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 backend
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/vmware-tanzu/velero/pkg/repository/udmrepo"
+)
+
+func TestGcsSetup(t *testing.T) {
+	testCases := []struct {
+		name        string
+		flags       map[string]string
+		expectedErr string
+	}{
+		{
+			name:        "must have bucket name",
+			flags:       map[string]string{},
+			expectedErr: "key " + udmrepo.StoreOptionOssBucket + " not found",
+		},
+		{
+			name: "must have credential file",
+			flags: map[string]string{
+				udmrepo.StoreOptionOssBucket: "fake-bucket",
+			},
+			expectedErr: "key " + udmrepo.StoreOptionCredentialFile + " not found",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			gcsFlags := GCSBackend{}
+
+			err := gcsFlags.Setup(context.Background(), tc.flags)
+
+			if tc.expectedErr == "" {
+				assert.NoError(t, err)
+			} else {
+				assert.EqualError(t, err, tc.expectedErr)
+			}
+		})
+	}
+}

pkg/repository/udmrepo/kopialib/backend/mocks/Logger.go

@@ -0,0 +1,65 @@
+// Code generated by mockery v2.14.0. DO NOT EDIT.
+
+package mocks
+
+import mock "github.com/stretchr/testify/mock"
+
+// Logger is an autogenerated mock type for the Logger type
+type Logger struct {
+	mock.Mock
+}
+
+// Debugf provides a mock function with given fields: msg, args
+func (_m *Logger) Debugf(msg string, args ...interface{}) {
+	var _ca []interface{}
+	_ca = append(_ca, msg)
+	_ca = append(_ca, args...)
+	_m.Called(_ca...)
+}
+
+// Debugw provides a mock function with given fields: msg, keyValuePairs
+func (_m *Logger) Debugw(msg string, keyValuePairs ...interface{}) {
+	var _ca []interface{}
+	_ca = append(_ca, msg)
+	_ca = append(_ca, keyValuePairs...)
+	_m.Called(_ca...)
+}
+
+// Errorf provides a mock function with given fields: msg, args
+func (_m *Logger) Errorf(msg string, args ...interface{}) {
+	var _ca []interface{}
+	_ca = append(_ca, msg)
+	_ca = append(_ca, args...)
+	_m.Called(_ca...)
+}
+
+// Infof provides a mock function with given fields: msg, args
+func (_m *Logger) Infof(msg string, args ...interface{}) {
+	var _ca []interface{}
+	_ca = append(_ca, msg)
+	_ca = append(_ca, args...)
+	_m.Called(_ca...)
+}
+
+// Warnf provides a mock function with given fields: msg, args
+func (_m *Logger) Warnf(msg string, args ...interface{}) {
+	var _ca []interface{}
+	_ca = append(_ca, msg)
+	_ca = append(_ca, args...)
+	_m.Called(_ca...)
+}
+
+type mockConstructorTestingTNewLogger interface {
+	mock.TestingT
+	Cleanup(func())
+}
+
+// NewLogger creates a new instance of Logger. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+func NewLogger(t mockConstructorTestingTNewLogger) *Logger {
+	mock := &Logger{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

pkg/repository/udmrepo/kopialib/backend/mocks/Storage.go

@@ -0,0 +1,185 @@
+// Code generated by mockery v2.14.0. DO NOT EDIT.
+
+package mocks
+
+import (
+	context "context"
+
+	blob "github.com/kopia/kopia/repo/blob"
+
+	mock "github.com/stretchr/testify/mock"
+)
+
+// Storage is an autogenerated mock type for the Storage type
+type Storage struct {
+	mock.Mock
+}
+
+// Close provides a mock function with given fields: ctx
+func (_m *Storage) Close(ctx context.Context) error {
+	ret := _m.Called(ctx)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context) error); ok {
+		r0 = rf(ctx)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// ConnectionInfo provides a mock function with given fields:
+func (_m *Storage) ConnectionInfo() blob.ConnectionInfo {
+	ret := _m.Called()
+
+	var r0 blob.ConnectionInfo
+	if rf, ok := ret.Get(0).(func() blob.ConnectionInfo); ok {
+		r0 = rf()
+	} else {
+		r0 = ret.Get(0).(blob.ConnectionInfo)
+	}
+
+	return r0
+}
+
+// DeleteBlob provides a mock function with given fields: ctx, blobID
+func (_m *Storage) DeleteBlob(ctx context.Context, blobID blob.ID) error {
+	ret := _m.Called(ctx, blobID)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, blob.ID) error); ok {
+		r0 = rf(ctx, blobID)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// DisplayName provides a mock function with given fields:
+func (_m *Storage) DisplayName() string {
+	ret := _m.Called()
+
+	var r0 string
+	if rf, ok := ret.Get(0).(func() string); ok {
+		r0 = rf()
+	} else {
+		r0 = ret.Get(0).(string)
+	}
+
+	return r0
+}
+
+// FlushCaches provides a mock function with given fields: ctx
+func (_m *Storage) FlushCaches(ctx context.Context) error {
+	ret := _m.Called(ctx)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context) error); ok {
+		r0 = rf(ctx)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// GetBlob provides a mock function with given fields: ctx, blobID, offset, length, output
+func (_m *Storage) GetBlob(ctx context.Context, blobID blob.ID, offset int64, length int64, output blob.OutputBuffer) error {
+	ret := _m.Called(ctx, blobID, offset, length, output)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, blob.ID, int64, int64, blob.OutputBuffer) error); ok {
+		r0 = rf(ctx, blobID, offset, length, output)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// GetCapacity provides a mock function with given fields: ctx
+func (_m *Storage) GetCapacity(ctx context.Context) (blob.Capacity, error) {
+	ret := _m.Called(ctx)
+
+	var r0 blob.Capacity
+	if rf, ok := ret.Get(0).(func(context.Context) blob.Capacity); ok {
+		r0 = rf(ctx)
+	} else {
+		r0 = ret.Get(0).(blob.Capacity)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context) error); ok {
+		r1 = rf(ctx)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// GetMetadata provides a mock function with given fields: ctx, blobID
+func (_m *Storage) GetMetadata(ctx context.Context, blobID blob.ID) (blob.Metadata, error) {
+	ret := _m.Called(ctx, blobID)
+
+	var r0 blob.Metadata
+	if rf, ok := ret.Get(0).(func(context.Context, blob.ID) blob.Metadata); ok {
+		r0 = rf(ctx, blobID)
+	} else {
+		r0 = ret.Get(0).(blob.Metadata)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, blob.ID) error); ok {
+		r1 = rf(ctx, blobID)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// ListBlobs provides a mock function with given fields: ctx, blobIDPrefix, cb
+func (_m *Storage) ListBlobs(ctx context.Context, blobIDPrefix blob.ID, cb func(blob.Metadata) error) error {
+	ret := _m.Called(ctx, blobIDPrefix, cb)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, blob.ID, func(blob.Metadata) error) error); ok {
+		r0 = rf(ctx, blobIDPrefix, cb)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// PutBlob provides a mock function with given fields: ctx, blobID, data, opts
+func (_m *Storage) PutBlob(ctx context.Context, blobID blob.ID, data blob.Bytes, opts blob.PutOptions) error {
+	ret := _m.Called(ctx, blobID, data, opts)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, blob.ID, blob.Bytes, blob.PutOptions) error); ok {
+		r0 = rf(ctx, blobID, data, opts)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+type mockConstructorTestingTNewStorage interface {
+	mock.TestingT
+	Cleanup(func())
+}
+
+// NewStorage creates a new instance of Storage. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+func NewStorage(t mockConstructorTestingTNewStorage) *Storage {
+	mock := &Storage{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

pkg/repository/udmrepo/kopialib/backend/mocks/Store.go

@@ -0,0 +1,68 @@
+// Code generated by mockery v2.14.0. DO NOT EDIT.
+
+package mocks
+
+import (
+	context "context"
+
+	blob "github.com/kopia/kopia/repo/blob"
+
+	mock "github.com/stretchr/testify/mock"
+)
+
+// Store is an autogenerated mock type for the Store type
+type Store struct {
+	mock.Mock
+}
+
+// Connect provides a mock function with given fields: ctx, isCreate
+func (_m *Store) Connect(ctx context.Context, isCreate bool) (blob.Storage, error) {
+	ret := _m.Called(ctx, isCreate)
+
+	var r0 blob.Storage
+	if rf, ok := ret.Get(0).(func(context.Context, bool) blob.Storage); ok {
+		r0 = rf(ctx, isCreate)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(blob.Storage)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, bool) error); ok {
+		r1 = rf(ctx, isCreate)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// Setup provides a mock function with given fields: ctx, flags
+func (_m *Store) Setup(ctx context.Context, flags map[string]string) error {
+	ret := _m.Called(ctx, flags)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, map[string]string) error); ok {
+		r0 = rf(ctx, flags)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+type mockConstructorTestingTNewStore interface {
+	mock.TestingT
+	Cleanup(func())
+}
+
+// NewStore creates a new instance of Store. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+func NewStore(t mockConstructorTestingTNewStore) *Store {
+	mock := &Store{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

pkg/repository/udmrepo/kopialib/backend/s3.go

@@ -0,0 +1,63 @@
+/*
+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 backend
+
+import (
+	"context"
+
+	"github.com/kopia/kopia/repo/blob"
+	"github.com/kopia/kopia/repo/blob/s3"
+
+	"github.com/vmware-tanzu/velero/pkg/repository/udmrepo"
+)
+
+type S3Backend struct {
+	options s3.Options
+}
+
+func (c *S3Backend) Setup(ctx context.Context, flags map[string]string) error {
+	var err error
+	c.options.BucketName, err = mustHaveString(udmrepo.StoreOptionOssBucket, flags)
+	if err != nil {
+		return err
+	}
+
+	c.options.AccessKeyID, err = mustHaveString(udmrepo.StoreOptionS3KeyId, flags)
+	if err != nil {
+		return err
+	}
+
+	c.options.SecretAccessKey, err = mustHaveString(udmrepo.StoreOptionS3SecretKey, flags)
+	if err != nil {
+		return err
+	}
+
+	c.options.Endpoint = optionalHaveString(udmrepo.StoreOptionS3Endpoint, flags)
+	c.options.Region = optionalHaveString(udmrepo.StoreOptionOssRegion, flags)
+	c.options.Prefix = optionalHaveString(udmrepo.StoreOptionPrefix, flags)
+	c.options.DoNotUseTLS = optionalHaveBool(ctx, udmrepo.StoreOptionS3DisableTls, flags)
+	c.options.DoNotVerifyTLS = optionalHaveBool(ctx, udmrepo.StoreOptionS3DisableTlsVerify, flags)
+	c.options.SessionToken = optionalHaveString(udmrepo.StoreOptionS3Token, flags)
+
+	c.options.Limits = setupLimits(ctx, flags)
+
+	return nil
+}
+
+func (c *S3Backend) Connect(ctx context.Context, isCreate bool) (blob.Storage, error) {
+	return s3.New(ctx, &c.options)
+}

pkg/repository/udmrepo/kopialib/backend/s3_test.go

@@ -0,0 +1,69 @@
+/*
+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 backend
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/vmware-tanzu/velero/pkg/repository/udmrepo"
+)
+
+func TestS3Setup(t *testing.T) {
+	testCases := []struct {
+		name        string
+		flags       map[string]string
+		expectedErr string
+	}{
+		{
+			name:        "must have bucket name",
+			flags:       map[string]string{},
+			expectedErr: "key " + udmrepo.StoreOptionOssBucket + " not found",
+		},
+		{
+			name: "must have access key Id",
+			flags: map[string]string{
+				udmrepo.StoreOptionOssBucket: "fake-bucket",
+			},
+			expectedErr: "key " + udmrepo.StoreOptionS3KeyId + " not found",
+		},
+		{
+			name: "must have access key",
+			flags: map[string]string{
+				udmrepo.StoreOptionOssBucket: "fake-bucket",
+				udmrepo.StoreOptionS3KeyId:   "fake-key-id",
+			},
+			expectedErr: "key " + udmrepo.StoreOptionS3SecretKey + " not found",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			s3Flags := S3Backend{}
+
+			err := s3Flags.Setup(context.Background(), tc.flags)
+
+			if tc.expectedErr == "" {
+				assert.NoError(t, err)
+			} else {
+				assert.EqualError(t, err, tc.expectedErr)
+			}
+		})
+	}
+}

pkg/repository/udmrepo/kopialib/backend/utils.go

@@ -0,0 +1,89 @@
+/*
+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 backend
+
+import (
+	"context"
+	"strconv"
+	"time"
+
+	"github.com/kopia/kopia/repo/logging"
+	"github.com/pkg/errors"
+)
+
+func mustHaveString(key string, flags map[string]string) (string, error) {
+	if value, exist := flags[key]; exist {
+		return value, nil
+	} else {
+		return "", errors.New("key " + key + " not found")
+	}
+}
+
+func optionalHaveString(key string, flags map[string]string) string {
+	return optionalHaveStringWithDefault(key, flags, "")
+}
+
+func optionalHaveBool(ctx context.Context, key string, flags map[string]string) bool {
+	if value, exist := flags[key]; exist {
+		ret, err := strconv.ParseBool(value)
+		if err == nil {
+			return ret
+		}
+
+		backendLog()(ctx).Errorf("Ignore %s, value [%s] is invalid, err %v", key, value, err)
+	}
+
+	return false
+}
+
+func optionalHaveFloat64(ctx context.Context, key string, flags map[string]string) float64 {
+	if value, exist := flags[key]; exist {
+		ret, err := strconv.ParseFloat(value, 64)
+		if err == nil {
+			return ret
+		}
+
+		backendLog()(ctx).Errorf("Ignore %s, value [%s] is invalid, err %v", key, value, err)
+	}
+
+	return 0
+}
+
+func optionalHaveStringWithDefault(key string, flags map[string]string, defValue string) string {
+	if value, exist := flags[key]; exist {
+		return value
+	} else {
+		return defValue
+	}
+}
+
+func optionalHaveDuration(ctx context.Context, key string, flags map[string]string) time.Duration {
+	if value, exist := flags[key]; exist {
+		ret, err := time.ParseDuration(value)
+		if err == nil {
+			return ret
+		}
+
+		backendLog()(ctx).Errorf("Ignore %s, value [%s] is invalid, err %v", key, value, err)
+	}
+
+	return 0
+}
+
+func backendLog() func(ctx context.Context) logging.Logger {
+	return logging.Module("kopialib-bd")
+}

pkg/repository/udmrepo/kopialib/backend/utils_test.go

@@ -0,0 +1,87 @@
+/*
+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 backend
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/kopia/kopia/repo/logging"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+
+	storagemocks "github.com/vmware-tanzu/velero/pkg/repository/udmrepo/kopialib/backend/mocks"
+)
+
+func TestOptionalHaveBool(t *testing.T) {
+	var expectMsg string
+	testCases := []struct {
+		name          string
+		key           string
+		flags         map[string]string
+		logger        *storagemocks.Logger
+		retFuncErrorf func(mock.Arguments)
+		expectMsg     string
+		retValue      bool
+	}{
+		{
+			name:     "key not exist",
+			key:      "fake-key",
+			flags:    map[string]string{},
+			retValue: false,
+		},
+		{
+			name: "value valid",
+			key:  "fake-key",
+			flags: map[string]string{
+				"fake-key": "true",
+			},
+			retValue: true,
+		},
+		{
+			name: "value invalid",
+			key:  "fake-key",
+			flags: map[string]string{
+				"fake-key": "fake-value",
+			},
+			logger: new(storagemocks.Logger),
+			retFuncErrorf: func(args mock.Arguments) {
+				expectMsg = fmt.Sprintf(args[0].(string), args[1].(string), args[2].(string), args[3].(error))
+			},
+			expectMsg: "Ignore fake-key, value [fake-value] is invalid, err strconv.ParseBool: parsing \"fake-value\": invalid syntax",
+			retValue:  false,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			if tc.logger != nil {
+				tc.logger.On("Errorf", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Run(tc.retFuncErrorf)
+			}
+
+			ctx := logging.WithLogger(context.Background(), func(module string) logging.Logger {
+				return tc.logger
+			})
+
+			retValue := optionalHaveBool(ctx, tc.key, tc.flags)
+
+			require.Equal(t, retValue, tc.retValue)
+			require.Equal(t, tc.expectMsg, expectMsg)
+		})
+	}
+}

pkg/repository/udmrepo/kopialib/repo_init.go

@@ -0,0 +1,160 @@
+/*
+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 kopialib
+
+import (
+	"context"
+	"strings"
+
+	"github.com/kopia/kopia/repo"
+	"github.com/kopia/kopia/repo/blob"
+	"github.com/pkg/errors"
+
+	"github.com/vmware-tanzu/velero/pkg/repository/udmrepo"
+	"github.com/vmware-tanzu/velero/pkg/repository/udmrepo/kopialib/backend"
+)
+
+type kopiaBackendStore struct {
+	name        string
+	description string
+	store       backend.Store
+}
+
+// backendStores lists the supported backend storages at present
+var backendStores []kopiaBackendStore = []kopiaBackendStore{
+	{udmrepo.StorageTypeAzure, "an Azure blob storage", &backend.AzureBackend{}},
+	{udmrepo.StorageTypeFs, "a filesystem", &backend.FsBackend{}},
+	{udmrepo.StorageTypeGcs, "a Google Cloud Storage bucket", &backend.GCSBackend{}},
+	{udmrepo.StorageTypeS3, "an S3 bucket", &backend.S3Backend{}},
+}
+
+// CreateBackupRepo creates a Kopia repository and then connect to it.
+// The storage must be empty, otherwise, it will fail
+func CreateBackupRepo(ctx context.Context, repoOption udmrepo.RepoOptions) error {
+	if repoOption.ConfigFilePath == "" {
+		return errors.New("invalid config file path")
+	}
+
+	backendStore, err := setupBackendStore(ctx, repoOption.StorageType, repoOption.StorageOptions)
+	if err != nil {
+		return errors.Wrap(err, "error to setup backend storage")
+	}
+
+	st, err := backendStore.store.Connect(ctx, true)
+	if err != nil {
+		return errors.Wrap(err, "error to connect to storage")
+	}
+
+	err = createWithStorage(ctx, st, repoOption)
+	if err != nil {
+		return errors.Wrap(err, "error to create repo with storage")
+	}
+
+	err = connectWithStorage(ctx, st, repoOption)
+	if err != nil {
+		return errors.Wrap(err, "error to connect repo with storage")
+	}
+
+	return nil
+}
+
+// ConnectBackupRepo connects to an existing Kopia repository.
+// If the repository doesn't exist, it will fail
+func ConnectBackupRepo(ctx context.Context, repoOption udmrepo.RepoOptions) error {
+	if repoOption.ConfigFilePath == "" {
+		return errors.New("invalid config file path")
+	}
+
+	backendStore, err := setupBackendStore(ctx, repoOption.StorageType, repoOption.StorageOptions)
+	if err != nil {
+		return errors.Wrap(err, "error to setup backend storage")
+	}
+
+	st, err := backendStore.store.Connect(ctx, false)
+	if err != nil {
+		return errors.Wrap(err, "error to connect to storage")
+	}
+
+	err = connectWithStorage(ctx, st, repoOption)
+	if err != nil {
+		return errors.Wrap(err, "error to connect repo with storage")
+	}
+
+	return nil
+}
+
+func findBackendStore(storage string) *kopiaBackendStore {
+	for _, options := range backendStores {
+		if strings.EqualFold(options.name, storage) {
+			return &options
+		}
+	}
+
+	return nil
+}
+
+func setupBackendStore(ctx context.Context, storageType string, storageOptions map[string]string) (*kopiaBackendStore, error) {
+	backendStore := findBackendStore(storageType)
+	if backendStore == nil {
+		return nil, errors.New("error to find storage type")
+	}
+
+	err := backendStore.store.Setup(ctx, storageOptions)
+	if err != nil {
+		return nil, errors.Wrap(err, "error to setup storage")
+	}
+
+	return backendStore, nil
+}
+
+func createWithStorage(ctx context.Context, st blob.Storage, repoOption udmrepo.RepoOptions) error {
+	err := ensureEmpty(ctx, st)
+	if err != nil {
+		return errors.Wrap(err, "error to ensure repository storage empty")
+	}
+
+	options := backend.SetupNewRepositoryOptions(ctx, repoOption.GeneralOptions)
+
+	if err := repo.Initialize(ctx, st, &options, repoOption.RepoPassword); err != nil {
+		return errors.Wrap(err, "error to initialize repository")
+	}
+
+	return nil
+}
+
+func connectWithStorage(ctx context.Context, st blob.Storage, repoOption udmrepo.RepoOptions) error {
+	options := backend.SetupConnectOptions(ctx, repoOption)
+	if err := repo.Connect(ctx, repoOption.ConfigFilePath, st, repoOption.RepoPassword, &options); err != nil {
+		return errors.Wrap(err, "error to connect to repository")
+	}
+
+	return nil
+}
+
+func ensureEmpty(ctx context.Context, s blob.Storage) error {
+	hasDataError := errors.Errorf("has data")
+
+	err := s.ListBlobs(ctx, "", func(cb blob.Metadata) error {
+		return hasDataError
+	})
+
+	if errors.Is(err, hasDataError) {
+		return errors.New("found existing data in storage location")
+	}
+
+	return errors.Wrap(err, "error to list blobs")
+}

pkg/repository/udmrepo/kopialib/repo_init_test.go

@@ -0,0 +1,237 @@
+/*
+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 kopialib
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+
+	"github.com/vmware-tanzu/velero/pkg/repository/udmrepo"
+	storagemocks "github.com/vmware-tanzu/velero/pkg/repository/udmrepo/kopialib/backend/mocks"
+
+	"github.com/pkg/errors"
+)
+
+type comparableError struct {
+	message string
+}
+
+func (ce *comparableError) Error() string {
+	return ce.message
+}
+
+func (ce *comparableError) Is(err error) bool {
+	return err.Error() == ce.message
+}
+
+func TestCreateBackupRepo(t *testing.T) {
+	testCases := []struct {
+		name         string
+		backendStore *storagemocks.Store
+		repoOptions  udmrepo.RepoOptions
+		connectErr   error
+		setupError   error
+		returnStore  *storagemocks.Storage
+		storeListErr error
+		getBlobErr   error
+		listBlobErr  error
+		expectedErr  string
+	}{
+		{
+			name:        "invalid config file",
+			expectedErr: "invalid config file path",
+		},
+		{
+			name: "storage setup fail, invalid type",
+			repoOptions: udmrepo.RepoOptions{
+				ConfigFilePath: "fake-file",
+			},
+			expectedErr: "error to setup backend storage: error to find storage type",
+		},
+		{
+			name: "storage setup fail, backend store steup fail",
+			repoOptions: udmrepo.RepoOptions{
+				ConfigFilePath: "fake-file",
+				StorageType:    udmrepo.StorageTypeAzure,
+			},
+			backendStore: new(storagemocks.Store),
+			setupError:   errors.New("fake-setup-error"),
+			expectedErr:  "error to setup backend storage: error to setup storage: fake-setup-error",
+		},
+		{
+			name: "storage connect fail",
+			repoOptions: udmrepo.RepoOptions{
+				ConfigFilePath: "fake-file",
+				StorageType:    udmrepo.StorageTypeAzure,
+			},
+			backendStore: new(storagemocks.Store),
+			connectErr:   errors.New("fake-connect-error"),
+			expectedErr:  "error to connect to storage: fake-connect-error",
+		},
+		{
+			name: "create repository error, exist blobs",
+			repoOptions: udmrepo.RepoOptions{
+				ConfigFilePath: "fake-file",
+				StorageType:    udmrepo.StorageTypeAzure,
+			},
+			backendStore: new(storagemocks.Store),
+			returnStore:  new(storagemocks.Storage),
+			listBlobErr: &comparableError{
+				message: "has data",
+			},
+			expectedErr: "error to create repo with storage: error to ensure repository storage empty: found existing data in storage location",
+		},
+		{
+			name: "create repository error, error list blobs",
+			repoOptions: udmrepo.RepoOptions{
+				ConfigFilePath: "fake-file",
+				StorageType:    udmrepo.StorageTypeAzure,
+			},
+			backendStore: new(storagemocks.Store),
+			returnStore:  new(storagemocks.Storage),
+			listBlobErr:  errors.New("fake-list-blob-error"),
+			expectedErr:  "error to create repo with storage: error to ensure repository storage empty: error to list blobs: fake-list-blob-error",
+		},
+		{
+			name: "create repository error, initialize error",
+			repoOptions: udmrepo.RepoOptions{
+				ConfigFilePath: "fake-file",
+				StorageType:    udmrepo.StorageTypeAzure,
+			},
+			backendStore: new(storagemocks.Store),
+			returnStore:  new(storagemocks.Storage),
+			getBlobErr:   errors.New("fake-list-blob-error-01"),
+			expectedErr:  "error to create repo with storage: error to initialize repository: unexpected error when checking for format blob: fake-list-blob-error-01",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			backendStores = []kopiaBackendStore{
+				{udmrepo.StorageTypeAzure, "fake store", tc.backendStore},
+				{udmrepo.StorageTypeFs, "fake store", tc.backendStore},
+				{udmrepo.StorageTypeGcs, "fake store", tc.backendStore},
+				{udmrepo.StorageTypeS3, "fake store", tc.backendStore},
+			}
+
+			if tc.backendStore != nil {
+				tc.backendStore.On("Connect", mock.Anything, mock.Anything, mock.Anything).Return(tc.returnStore, tc.connectErr)
+				tc.backendStore.On("Setup", mock.Anything, mock.Anything, mock.Anything).Return(tc.setupError)
+			}
+
+			if tc.returnStore != nil {
+				tc.returnStore.On("ListBlobs", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(tc.listBlobErr)
+				tc.returnStore.On("GetBlob", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(tc.getBlobErr)
+			}
+
+			err := CreateBackupRepo(context.Background(), tc.repoOptions)
+
+			if tc.expectedErr == "" {
+				assert.NoError(t, err)
+			} else {
+				assert.EqualError(t, err, tc.expectedErr)
+			}
+		})
+	}
+}
+
+func TestConnectBackupRepo(t *testing.T) {
+	testCases := []struct {
+		name         string
+		backendStore *storagemocks.Store
+		repoOptions  udmrepo.RepoOptions
+		connectErr   error
+		setupError   error
+		returnStore  *storagemocks.Storage
+		getBlobErr   error
+		expectedErr  string
+	}{
+		{
+			name:        "invalid config file",
+			expectedErr: "invalid config file path",
+		},
+		{
+			name: "storage setup fail, invalid type",
+			repoOptions: udmrepo.RepoOptions{
+				ConfigFilePath: "fake-file",
+			},
+			expectedErr: "error to setup backend storage: error to find storage type",
+		},
+		{
+			name: "storage setup fail, backend store steup fail",
+			repoOptions: udmrepo.RepoOptions{
+				ConfigFilePath: "fake-file",
+				StorageType:    udmrepo.StorageTypeAzure,
+			},
+			backendStore: new(storagemocks.Store),
+			setupError:   errors.New("fake-setup-error"),
+			expectedErr:  "error to setup backend storage: error to setup storage: fake-setup-error",
+		},
+		{
+			name: "storage connect fail",
+			repoOptions: udmrepo.RepoOptions{
+				ConfigFilePath: "fake-file",
+				StorageType:    udmrepo.StorageTypeAzure,
+			},
+			backendStore: new(storagemocks.Store),
+			connectErr:   errors.New("fake-connect-error"),
+			expectedErr:  "error to connect to storage: fake-connect-error",
+		},
+		{
+			name: "connect repository error",
+			repoOptions: udmrepo.RepoOptions{
+				ConfigFilePath: "fake-file",
+				StorageType:    udmrepo.StorageTypeAzure,
+			},
+			backendStore: new(storagemocks.Store),
+			returnStore:  new(storagemocks.Storage),
+			getBlobErr:   errors.New("fake-get-blob-error"),
+			expectedErr:  "error to connect repo with storage: error to connect to repository: unable to read format blob: fake-get-blob-error",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			backendStores = []kopiaBackendStore{
+				{udmrepo.StorageTypeAzure, "fake store", tc.backendStore},
+				{udmrepo.StorageTypeFs, "fake store", tc.backendStore},
+				{udmrepo.StorageTypeGcs, "fake store", tc.backendStore},
+				{udmrepo.StorageTypeS3, "fake store", tc.backendStore},
+			}
+
+			if tc.backendStore != nil {
+				tc.backendStore.On("Connect", mock.Anything, mock.Anything, mock.Anything).Return(tc.returnStore, tc.connectErr)
+				tc.backendStore.On("Setup", mock.Anything, mock.Anything, mock.Anything).Return(tc.setupError)
+			}
+
+			if tc.returnStore != nil {
+				tc.returnStore.On("GetBlob", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(tc.getBlobErr)
+			}
+
+			err := ConnectBackupRepo(context.Background(), tc.repoOptions)
+
+			if tc.expectedErr == "" {
+				assert.NoError(t, err)
+			} else {
+				assert.EqualError(t, err, tc.expectedErr)
+			}
+		})
+	}
+}

pkg/repository/udmrepo/repo_options.go

@@ -32,6 +32,9 @@ const (
 	GenOptionMaintainFull  = "full"
 	GenOptionMaintainQuick = "quick"
 
+	GenOptionOwnerName   = "username"
+	GenOptionOwnerDomain = "domainname"
+
 	StoreOptionS3KeyId            = "accessKeyID"
 	StoreOptionS3Provider         = "providerName"
 	StoreOptionS3SecretKey        = "secretAccessKey"
@@ -56,6 +59,14 @@ const (
 	StoreOptionPrefix         = "prefix"
 	StoreOptionPrefixName     = "unified-repo"
 
+	StoreOptionGenHashAlgo    = "hashAlgo"
+	StoreOptionGenEncryptAlgo = "encryptAlgo"
+	StoreOptionGenSplitAlgo   = "splitAlgo"
+
+	StoreOptionGenRetentionMode   = "retentionMode"
+	StoreOptionGenRetentionPeriod = "retentionPeriod"
+	StoreOptionGenReadOnly        = "readOnly"
+
 	ThrottleOptionReadOps       = "readOPS"
 	ThrottleOptionWriteOps      = "writeOPS"
 	ThrottleOptionListOps       = "listOPS"
@@ -63,6 +74,11 @@ const (
 	ThrottleOptionDownloadBytes = "downloadBytes"
 )
 
+const (
+	defaultUsername = "default"
+	defaultDomain   = "default"
+)
+
 type RepoOptions struct {
 	// StorageType is a repository specific string to identify a backup storage, i.e., "s3", "filesystem"
 	StorageType string
@@ -80,17 +96,24 @@ type RepoOptions struct {
 	Description string
 }
 
+// PasswordGetter defines the method to get a repository password.
 type PasswordGetter interface {
 	GetPassword(param interface{}) (string, error)
 }
 
+// StoreOptionsGetter defines the methods to get the storage related options.
 type StoreOptionsGetter interface {
 	GetStoreType(param interface{}) (string, error)
 	GetStoreOptions(param interface{}) (map[string]string, error)
 }
 
+// NewRepoOptions creates a new RepoOptions for different purpose
 func NewRepoOptions(optionFuncs ...func(*RepoOptions) error) (*RepoOptions, error) {
-	options := &RepoOptions{}
+	options := &RepoOptions{
+		GeneralOptions: make(map[string]string),
+		StorageOptions: make(map[string]string),
+	}
+
 	for _, optionFunc := range optionFuncs {
 		err := optionFunc(options)
 		if err != nil {
@@ -101,6 +124,8 @@ func NewRepoOptions(optionFuncs ...func(*RepoOptions) error) (*RepoOptions, erro
 	return options, nil
 }
 
+// WithPassword sets the RepoPassword to RepoOptions, the password is acquired through
+// the provided interface
 func WithPassword(getter PasswordGetter, param interface{}) func(*RepoOptions) error {
 	return func(options *RepoOptions) error {
 		password, err := getter.GetPassword(param)
@@ -114,6 +139,7 @@ func WithPassword(getter PasswordGetter, param interface{}) func(*RepoOptions) e
 	}
 }
 
+// WithConfigFile sets the ConfigFilePath to RepoOptions
 func WithConfigFile(workPath string, repoID string) func(*RepoOptions) error {
 	return func(options *RepoOptions) error {
 		options.ConfigFilePath = getRepoConfigFile(workPath, repoID)
@@ -121,6 +147,7 @@ func WithConfigFile(workPath string, repoID string) func(*RepoOptions) error {
 	}
 }
 
+// WithGenOptions sets the GeneralOptions to RepoOptions
 func WithGenOptions(genOptions map[string]string) func(*RepoOptions) error {
 	return func(options *RepoOptions) error {
 		for k, v := range genOptions {
@@ -131,6 +158,8 @@ func WithGenOptions(genOptions map[string]string) func(*RepoOptions) error {
 	}
 }
 
+// WithStoreOptions sets the StorageOptions to RepoOptions, the store options are acquired through
+// the provided interface
 func WithStoreOptions(getter StoreOptionsGetter, param interface{}) func(*RepoOptions) error {
 	return func(options *RepoOptions) error {
 		storeType, err := getter.GetStoreType(param)
@@ -153,6 +182,7 @@ func WithStoreOptions(getter StoreOptionsGetter, param interface{}) func(*RepoOp
 	}
 }
 
+// WithDescription sets the Description to RepoOptions
 func WithDescription(desc string) func(*RepoOptions) error {
 	return func(options *RepoOptions) error {
 		options.Description = desc
@@ -160,6 +190,16 @@ func WithDescription(desc string) func(*RepoOptions) error {
 	}
 }
 
+// GetRepoUser returns the default username that is used to manipulate the Unified Repo
+func GetRepoUser() string {
+	return defaultUsername
+}
+
+// GetRepoDomain returns the default user domain that is used to manipulate the Unified Repo
+func GetRepoDomain() string {
+	return defaultDomain
+}
+
 func getRepoConfigFile(workPath string, repoID string) string {
 	if workPath == "" {
 		workPath = filepath.Join(os.Getenv("HOME"), "udmrepo")

pkg/repository/udmrepo/service/service.go

@@ -22,16 +22,8 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/repository/udmrepo"
 )
 
-const (
-	defaultUsername = "default"
-	defaultDomain   = "default"
-)
-
+// Create creates an instance of BackupRepoService
 func Create(logger logrus.FieldLogger) udmrepo.BackupRepoService {
 	///TODO: create from kopiaLib
 	return nil
 }
-
-func GetRepoUser() (username, domain string) {
-	return defaultUsername, defaultDomain
-}