issue 7620: backup repo config

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

config/crd/v1/bases/velero.io_backuprepositories.yaml

@@ -54,6 +54,13 @@ spec:
                 description: MaintenanceFrequency is how often maintenance should
                   be run.
                 type: string
+              repositoryConfig:
+                additionalProperties:
+                  type: string
+                description: RepositoryConfig is for repository-specific configuration
+                  fields.
+                nullable: true
+                type: object
               repositoryType:
                 description: RepositoryType indicates the type of the backend repository
                 enum:

pkg/apis/velero/v1/backup_repository_types.go

@@ -41,6 +41,11 @@ type BackupRepositorySpec struct {
 
 	// MaintenanceFrequency is how often maintenance should be run.
 	MaintenanceFrequency metav1.Duration `json:"maintenanceFrequency"`
+
+	// RepositoryConfig is for repository-specific configuration fields.
+	// +optional
+	// +nullable
+	RepositoryConfig map[string]string `json:"repositoryConfig,omitempty"`
 }
 
 // BackupRepositoryPhase represents the lifecycle phase of a BackupRepository.

pkg/apis/velero/v1/zz_generated.deepcopy.go

@@ -111,7 +111,7 @@ func (in *BackupRepository) DeepCopyInto(out *BackupRepository) {
 	*out = *in
 	out.TypeMeta = in.TypeMeta
 	in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
-	out.Spec = in.Spec
+	in.Spec.DeepCopyInto(&out.Spec)
 	in.Status.DeepCopyInto(&out.Status)
 }
 
@@ -169,6 +169,13 @@ func (in *BackupRepositoryList) DeepCopyObject() runtime.Object {
 func (in *BackupRepositorySpec) DeepCopyInto(out *BackupRepositorySpec) {
 	*out = *in
 	out.MaintenanceFrequency = in.MaintenanceFrequency
+	if in.RepositoryConfig != nil {
+		in, out := &in.RepositoryConfig, &out.RepositoryConfig
+		*out = make(map[string]string, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val
+		}
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new BackupRepositorySpec.

pkg/cmd/server/server.go

@@ -140,6 +140,7 @@ type serverConfig struct {
 	disableInformerCache                                                    bool
 	scheduleSkipImmediately                                                 bool
 	maintenanceCfg                                                          repository.MaintenanceConfig
+	backukpRepoConfig                                                       string
 }
 
 func NewCommand(f client.Factory) *cobra.Command {
@@ -253,6 +254,8 @@ func NewCommand(f client.Factory) *cobra.Command {
 	command.Flags().StringVar(&config.maintenanceCfg.CPULimit, "maintenance-job-cpu-limit", config.maintenanceCfg.CPULimit, "CPU limit for maintenance job. Default is no limit.")
 	command.Flags().StringVar(&config.maintenanceCfg.MemLimit, "maintenance-job-mem-limit", config.maintenanceCfg.MemLimit, "Memory limit for maintenance job. Default is no limit.")
 
+	command.Flags().StringVar(&config.backukpRepoConfig, "backup-repository-config", config.backukpRepoConfig, "The name of configMap containing backup repository configurations.")
+
 	// maintenance job log setting inherited from velero server
 	config.maintenanceCfg.FormatFlag = config.formatFlag
 	config.maintenanceCfg.LogLevelFlag = logLevelFlag
@@ -876,7 +879,7 @@ func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string
 	}
 
 	if _, ok := enabledRuntimeControllers[controller.BackupRepo]; ok {
-		if err := controller.NewBackupRepoReconciler(s.namespace, s.logger, s.mgr.GetClient(), s.config.repoMaintenanceFrequency, s.repoManager).SetupWithManager(s.mgr); err != nil {
+		if err := controller.NewBackupRepoReconciler(s.namespace, s.logger, s.mgr.GetClient(), s.config.repoMaintenanceFrequency, s.config.backukpRepoConfig, s.repoManager).SetupWithManager(s.mgr); err != nil {
 			s.logger.Fatal(err, "unable to create controller", "controller", controller.BackupRepo)
 		}
 	}

pkg/controller/backup_repository_controller.go

@@ -19,6 +19,8 @@ package controller
 import (
 	"bytes"
 	"context"
+	"encoding/json"
+	"fmt"
 	"reflect"
 	"time"
 
@@ -38,6 +40,8 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/repository"
 	repoconfig "github.com/vmware-tanzu/velero/pkg/repository/config"
 	"github.com/vmware-tanzu/velero/pkg/util/kube"
+
+	corev1api "k8s.io/api/core/v1"
 )
 
 const (
@@ -51,17 +55,19 @@ type BackupRepoReconciler struct {
 	logger               logrus.FieldLogger
 	clock                clocks.WithTickerAndDelayedExecution
 	maintenanceFrequency time.Duration
+	backukpRepoConfig    string
 	repositoryManager    repository.Manager
 }
 
 func NewBackupRepoReconciler(namespace string, logger logrus.FieldLogger, client client.Client,
-	maintenanceFrequency time.Duration, repositoryManager repository.Manager) *BackupRepoReconciler {
+	maintenanceFrequency time.Duration, backukpRepoConfig string, repositoryManager repository.Manager) *BackupRepoReconciler {
 	c := &BackupRepoReconciler{
 		client,
 		namespace,
 		logger,
 		clocks.RealClock{},
 		maintenanceFrequency,
+		backukpRepoConfig,
 		repositoryManager,
 	}
 
@@ -223,7 +229,7 @@ func (r *BackupRepoReconciler) getIdentiferByBSL(ctx context.Context, req *veler
 }
 
 func (r *BackupRepoReconciler) initializeRepo(ctx context.Context, req *velerov1api.BackupRepository, log logrus.FieldLogger) error {
-	log.Info("Initializing backup repository")
+	log.WithField("repoConfig", r.backukpRepoConfig).Info("Initializing backup repository")
 
 	// confirm the repo's BackupStorageLocation is valid
 	repoIdentifier, err := r.getIdentiferByBSL(ctx, req)
@@ -238,6 +244,13 @@ func (r *BackupRepoReconciler) initializeRepo(ctx context.Context, req *velerov1
 		})
 	}
 
+	config, err := getBackupRepositoryConfig(ctx, r, r.backukpRepoConfig, r.namespace, req.Name, req.Spec.RepositoryType, log)
+	if err != nil {
+		log.WithError(err).Warnf("Failed to get repo config from %s for repo %s, repo config is ignored", r.backukpRepoConfig, req.Name)
+	} else if config != nil {
+		log.Infof("Init repo with config %v", config)
+	}
+
 	// defaulting - if the patch fails, return an error so the item is returned to the queue
 	if err := r.patchBackupRepository(ctx, req, func(rr *velerov1api.BackupRepository) {
 		rr.Spec.ResticIdentifier = repoIdentifier
@@ -245,6 +258,8 @@ func (r *BackupRepoReconciler) initializeRepo(ctx context.Context, req *velerov1
 		if rr.Spec.MaintenanceFrequency.Duration <= 0 {
 			rr.Spec.MaintenanceFrequency = metav1.Duration{Duration: r.getRepositoryMaintenanceFrequency(req)}
 		}
+
+		rr.Spec.RepositoryConfig = config
 	}); err != nil {
 		return err
 	}
@@ -366,3 +381,35 @@ func (r *BackupRepoReconciler) patchBackupRepository(ctx context.Context, req *v
 	}
 	return nil
 }
+
+func getBackupRepositoryConfig(ctx context.Context, ctrlClient client.Client, configName, namespace, repoName, repoType string, log logrus.FieldLogger) (map[string]string, error) {
+	if configName == "" {
+		return nil, nil
+	}
+
+	loc := &corev1api.ConfigMap{}
+	if err := ctrlClient.Get(ctx, client.ObjectKey{
+		Namespace: namespace,
+		Name:      configName,
+	}, loc); err != nil {
+		return nil, errors.Wrapf(err, "error getting configMap %s", configName)
+	}
+
+	jsonData, found := loc.Data[repoType]
+	if !found {
+		log.Info("No data for repo type %s in config map %s", repoType, configName)
+		return nil, nil
+	}
+
+	var unmarshalled map[string]interface{}
+	if err := json.Unmarshal([]byte(jsonData), &unmarshalled); err != nil {
+		return nil, errors.Wrapf(err, "error unmarshalling config data from %s for repo %s, repo type %s", configName, repoName, repoType)
+	}
+
+	result := map[string]string{}
+	for k, v := range unmarshalled {
+		result[k] = fmt.Sprintf("%v", v)
+	}
+
+	return result, nil
+}

pkg/controller/backup_repository_controller_test.go

@@ -21,7 +21,9 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
+	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	ctrl "sigs.k8s.io/controller-runtime"
 
@@ -29,6 +31,8 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/repository"
 	repomokes "github.com/vmware-tanzu/velero/pkg/repository/mocks"
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
+
+	clientFake "sigs.k8s.io/controller-runtime/pkg/client/fake"
 )
 
 const testMaintenanceFrequency = 10 * time.Minute
@@ -43,6 +47,7 @@ func mockBackupRepoReconciler(t *testing.T, rr *velerov1api.BackupRepository, mo
 		velerotest.NewLogger(),
 		velerotest.NewFakeControllerRuntimeClient(t),
 		testMaintenanceFrequency,
+		"fake-repo-config",
 		mgr,
 	)
 }
@@ -243,6 +248,7 @@ func TestGetRepositoryMaintenanceFrequency(t *testing.T) {
 				velerotest.NewLogger(),
 				velerotest.NewFakeControllerRuntimeClient(t),
 				test.userDefinedFreq,
+				"",
 				&mgr,
 			)
 
@@ -370,10 +376,112 @@ func TestNeedInvalidBackupRepo(t *testing.T) {
 				velerov1api.DefaultNamespace,
 				velerotest.NewLogger(),
 				velerotest.NewFakeControllerRuntimeClient(t),
-				time.Duration(0), nil)
+				time.Duration(0), "", nil)
 
 			need := reconciler.needInvalidBackupRepo(test.oldBSL, test.newBSL)
 			assert.Equal(t, test.expect, need)
 		})
 	}
 }
+
+func TestGetBackupRepositoryConfig(t *testing.T) {
+	configWithNoData := &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "config-1",
+			Namespace: velerov1api.DefaultNamespace,
+		},
+	}
+
+	configWithWrongData := &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "config-1",
+			Namespace: velerov1api.DefaultNamespace,
+		},
+		Data: map[string]string{
+			"fake-repo-type": "",
+		},
+	}
+
+	configWithData := &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "config-1",
+			Namespace: velerov1api.DefaultNamespace,
+		},
+		Data: map[string]string{
+			"fake-repo-type":   "{\"cacheLimitMB\": 1000, \"enableCompression\": true}",
+			"fake-repo-type-1": "{\"cacheLimitMB\": 1, \"enableCompression\": false}",
+		},
+	}
+
+	tests := []struct {
+		name           string
+		congiName      string
+		repoName       string
+		repoType       string
+		kubeClientObj  []runtime.Object
+		expectedErr    string
+		expectedResult map[string]string
+	}{
+		{
+			name: "empty configName",
+		},
+		{
+			name:        "get error",
+			congiName:   "config-1",
+			expectedErr: "error getting configMap config-1: configmaps \"config-1\" not found",
+		},
+		{
+			name:      "no config for repo",
+			congiName: "config-1",
+			repoName:  "fake-repo",
+			repoType:  "fake-repo-type",
+			kubeClientObj: []runtime.Object{
+				configWithNoData,
+			},
+		},
+		{
+			name:      "unmarshall error",
+			congiName: "config-1",
+			repoName:  "fake-repo",
+			repoType:  "fake-repo-type",
+			kubeClientObj: []runtime.Object{
+				configWithWrongData,
+			},
+			expectedErr: "error unmarshalling config data from config-1 for repo fake-repo, repo type fake-repo-type: unexpected end of JSON input",
+		},
+		{
+			name:      "succeed",
+			congiName: "config-1",
+			repoName:  "fake-repo",
+			repoType:  "fake-repo-type",
+			kubeClientObj: []runtime.Object{
+				configWithData,
+			},
+			expectedResult: map[string]string{
+				"cacheLimitMB":      "1000",
+				"enableCompression": "true",
+			},
+		},
+	}
+
+	scheme := runtime.NewScheme()
+	corev1.AddToScheme(scheme)
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			fakeClientBuilder := clientFake.NewClientBuilder()
+			fakeClientBuilder = fakeClientBuilder.WithScheme(scheme)
+
+			fakeClient := fakeClientBuilder.WithRuntimeObjects(test.kubeClientObj...).Build()
+
+			result, err := getBackupRepositoryConfig(context.Background(), fakeClient, test.congiName, velerov1api.DefaultNamespace, test.repoName, test.repoType, velerotest.NewLogger())
+
+			if test.expectedErr != "" {
+				assert.EqualError(t, err, test.expectedErr)
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, test.expectedResult, result)
+			}
+		})
+	}
+}

pkg/repository/provider/unified_repo.go

@@ -53,7 +53,7 @@ var getGCPCredentials = repoconfig.GetGCPCredentials
 var getS3BucketRegion = repoconfig.GetAWSBucketRegion
 
 type localFuncTable struct {
-	getStorageVariables   func(*velerov1api.BackupStorageLocation, string, string) (map[string]string, error)
+	getStorageVariables   func(*velerov1api.BackupStorageLocation, string, string, map[string]string) (map[string]string, error)
 	getStorageCredentials func(*velerov1api.BackupStorageLocation, credentials.FileStore) (map[string]string, error)
 }
 
@@ -397,7 +397,7 @@ func (urp *unifiedRepoProvider) GetStoreOptions(param interface{}) (map[string]s
 		return map[string]string{}, errors.Errorf("invalid parameter, expect %T, actual %T", RepoParam{}, param)
 	}
 
-	storeVar, err := funcTable.getStorageVariables(repoParam.BackupLocation, urp.repoBackend, repoParam.BackupRepo.Spec.VolumeNamespace)
+	storeVar, err := funcTable.getStorageVariables(repoParam.BackupLocation, urp.repoBackend, repoParam.BackupRepo.Spec.VolumeNamespace, repoParam.BackupRepo.Spec.RepositoryConfig)
 	if err != nil {
 		return map[string]string{}, errors.Wrap(err, "error to get storage variables")
 	}
@@ -498,7 +498,7 @@ func getStorageCredentials(backupLocation *velerov1api.BackupStorageLocation, cr
 	return result, nil
 }
 
-func getStorageVariables(backupLocation *velerov1api.BackupStorageLocation, repoBackend string, repoName string) (map[string]string, error) {
+func getStorageVariables(backupLocation *velerov1api.BackupStorageLocation, repoBackend string, repoName string, backupRepoConfig map[string]string) (map[string]string, error) {
 	result := make(map[string]string)
 
 	backendType := repoconfig.GetBackendType(backupLocation.Spec.Provider, backupLocation.Spec.Config)
@@ -568,6 +568,12 @@ func getStorageVariables(backupLocation *velerov1api.BackupStorageLocation, repo
 	result[udmrepo.StoreOptionOssRegion] = strings.Trim(region, "/")
 	result[udmrepo.StoreOptionFsPath] = config["fspath"]
 
+	if backupRepoConfig != nil {
+		if v, found := backupRepoConfig[udmrepo.StoreOptionCacheLimit]; found {
+			result[udmrepo.StoreOptionCacheLimit] = v
+		}
+	}
+
 	return result, nil
 }
 

pkg/repository/provider/unified_repo_test.go

@@ -221,6 +221,7 @@ func TestGetStorageVariables(t *testing.T) {
 		credFileStore     *credmock.FileStore
 		repoName          string
 		repoBackend       string
+		repoConfig        map[string]string
 		getS3BucketRegion func(string) (string, error)
 		expected          map[string]string
 		expectedErr       string
@@ -435,13 +436,36 @@ func TestGetStorageVariables(t *testing.T) {
 				"region": "",
 			},
 		},
+		{
+			name: "fs with repo config",
+			backupLocation: velerov1api.BackupStorageLocation{
+				Spec: velerov1api.BackupStorageLocationSpec{
+					Provider: "velero.io/fs",
+					Config: map[string]string{
+						"fspath": "fake-path",
+						"prefix": "fake-prefix",
+					},
+				},
+			},
+			repoBackend: "fake-repo-type",
+			repoConfig: map[string]string{
+				udmrepo.StoreOptionCacheLimit: "1000",
+			},
+			expected: map[string]string{
+				"fspath":     "fake-path",
+				"bucket":     "",
+				"prefix":     "fake-prefix/fake-repo-type/",
+				"region":     "",
+				"cacheLimit": "1000",
+			},
+		},
 	}
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
 			getS3BucketRegion = tc.getS3BucketRegion
 
-			actual, err := getStorageVariables(&tc.backupLocation, tc.repoBackend, tc.repoName)
+			actual, err := getStorageVariables(&tc.backupLocation, tc.repoBackend, tc.repoName, tc.repoConfig)
 
 			require.Equal(t, tc.expected, actual)
 
@@ -530,7 +554,7 @@ func TestGetStoreOptions(t *testing.T) {
 				BackupRepo:     &velerov1api.BackupRepository{},
 			},
 			funcTable: localFuncTable{
-				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string) (map[string]string, error) {
+				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string, map[string]string) (map[string]string, error) {
 					return map[string]string{}, errors.New("fake-error-2")
 				},
 			},
@@ -544,7 +568,7 @@ func TestGetStoreOptions(t *testing.T) {
 				BackupRepo:     &velerov1api.BackupRepository{},
 			},
 			funcTable: localFuncTable{
-				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string) (map[string]string, error) {
+				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string, map[string]string) (map[string]string, error) {
 					return map[string]string{}, nil
 				},
 				getStorageCredentials: func(*velerov1api.BackupStorageLocation, velerocredentials.FileStore) (map[string]string, error) {
@@ -604,7 +628,7 @@ func TestPrepareRepo(t *testing.T) {
 			repoService:     new(reposervicenmocks.BackupRepoService),
 			credStoreReturn: "fake-password",
 			funcTable: localFuncTable{
-				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string) (map[string]string, error) {
+				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string, map[string]string) (map[string]string, error) {
 					return map[string]string{}, errors.New("fake-store-option-error")
 				},
 			},
@@ -615,7 +639,7 @@ func TestPrepareRepo(t *testing.T) {
 			getter:          new(credmock.SecretStore),
 			credStoreReturn: "fake-password",
 			funcTable: localFuncTable{
-				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string) (map[string]string, error) {
+				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string, map[string]string) (map[string]string, error) {
 					return map[string]string{}, nil
 				},
 				getStorageCredentials: func(*velerov1api.BackupStorageLocation, velerocredentials.FileStore) (map[string]string, error) {
@@ -635,7 +659,7 @@ func TestPrepareRepo(t *testing.T) {
 			getter:          new(credmock.SecretStore),
 			credStoreReturn: "fake-password",
 			funcTable: localFuncTable{
-				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string) (map[string]string, error) {
+				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string, map[string]string) (map[string]string, error) {
 					return map[string]string{}, nil
 				},
 				getStorageCredentials: func(*velerov1api.BackupStorageLocation, velerocredentials.FileStore) (map[string]string, error) {
@@ -656,7 +680,7 @@ func TestPrepareRepo(t *testing.T) {
 			getter:          new(credmock.SecretStore),
 			credStoreReturn: "fake-password",
 			funcTable: localFuncTable{
-				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string) (map[string]string, error) {
+				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string, map[string]string) (map[string]string, error) {
 					return map[string]string{}, nil
 				},
 				getStorageCredentials: func(*velerov1api.BackupStorageLocation, velerocredentials.FileStore) (map[string]string, error) {
@@ -733,7 +757,7 @@ func TestForget(t *testing.T) {
 			getter:          new(credmock.SecretStore),
 			credStoreReturn: "fake-password",
 			funcTable: localFuncTable{
-				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string) (map[string]string, error) {
+				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string, map[string]string) (map[string]string, error) {
 					return map[string]string{}, nil
 				},
 				getStorageCredentials: func(*velerov1api.BackupStorageLocation, velerocredentials.FileStore) (map[string]string, error) {
@@ -757,7 +781,7 @@ func TestForget(t *testing.T) {
 			getter:          new(credmock.SecretStore),
 			credStoreReturn: "fake-password",
 			funcTable: localFuncTable{
-				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string) (map[string]string, error) {
+				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string, map[string]string) (map[string]string, error) {
 					return map[string]string{}, nil
 				},
 				getStorageCredentials: func(*velerov1api.BackupStorageLocation, velerocredentials.FileStore) (map[string]string, error) {
@@ -785,7 +809,7 @@ func TestForget(t *testing.T) {
 			getter:          new(credmock.SecretStore),
 			credStoreReturn: "fake-password",
 			funcTable: localFuncTable{
-				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string) (map[string]string, error) {
+				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string, map[string]string) (map[string]string, error) {
 					return map[string]string{}, nil
 				},
 				getStorageCredentials: func(*velerov1api.BackupStorageLocation, velerocredentials.FileStore) (map[string]string, error) {
@@ -883,7 +907,7 @@ func TestBatchForget(t *testing.T) {
 			getter:          new(credmock.SecretStore),
 			credStoreReturn: "fake-password",
 			funcTable: localFuncTable{
-				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string) (map[string]string, error) {
+				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string, map[string]string) (map[string]string, error) {
 					return map[string]string{}, nil
 				},
 				getStorageCredentials: func(*velerov1api.BackupStorageLocation, velerocredentials.FileStore) (map[string]string, error) {
@@ -907,7 +931,7 @@ func TestBatchForget(t *testing.T) {
 			getter:          new(credmock.SecretStore),
 			credStoreReturn: "fake-password",
 			funcTable: localFuncTable{
-				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string) (map[string]string, error) {
+				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string, map[string]string) (map[string]string, error) {
 					return map[string]string{}, nil
 				},
 				getStorageCredentials: func(*velerov1api.BackupStorageLocation, velerocredentials.FileStore) (map[string]string, error) {
@@ -936,7 +960,7 @@ func TestBatchForget(t *testing.T) {
 			getter:          new(credmock.SecretStore),
 			credStoreReturn: "fake-password",
 			funcTable: localFuncTable{
-				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string) (map[string]string, error) {
+				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string, map[string]string) (map[string]string, error) {
 					return map[string]string{}, nil
 				},
 				getStorageCredentials: func(*velerov1api.BackupStorageLocation, velerocredentials.FileStore) (map[string]string, error) {
@@ -1032,7 +1056,7 @@ func TestInitRepo(t *testing.T) {
 			getter:          new(credmock.SecretStore),
 			credStoreReturn: "fake-password",
 			funcTable: localFuncTable{
-				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string) (map[string]string, error) {
+				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string, map[string]string) (map[string]string, error) {
 					return map[string]string{}, nil
 				},
 				getStorageCredentials: func(*velerov1api.BackupStorageLocation, velerocredentials.FileStore) (map[string]string, error) {
@@ -1050,7 +1074,7 @@ func TestInitRepo(t *testing.T) {
 			getter:          new(credmock.SecretStore),
 			credStoreReturn: "fake-password",
 			funcTable: localFuncTable{
-				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string) (map[string]string, error) {
+				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string, map[string]string) (map[string]string, error) {
 					return map[string]string{}, nil
 				},
 				getStorageCredentials: func(*velerov1api.BackupStorageLocation, velerocredentials.FileStore) (map[string]string, error) {
@@ -1120,7 +1144,7 @@ func TestConnectToRepo(t *testing.T) {
 			getter:          new(credmock.SecretStore),
 			credStoreReturn: "fake-password",
 			funcTable: localFuncTable{
-				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string) (map[string]string, error) {
+				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string, map[string]string) (map[string]string, error) {
 					return map[string]string{}, nil
 				},
 				getStorageCredentials: func(*velerov1api.BackupStorageLocation, velerocredentials.FileStore) (map[string]string, error) {
@@ -1138,7 +1162,7 @@ func TestConnectToRepo(t *testing.T) {
 			getter:          new(credmock.SecretStore),
 			credStoreReturn: "fake-password",
 			funcTable: localFuncTable{
-				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string) (map[string]string, error) {
+				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string, map[string]string) (map[string]string, error) {
 					return map[string]string{}, nil
 				},
 				getStorageCredentials: func(*velerov1api.BackupStorageLocation, velerocredentials.FileStore) (map[string]string, error) {
@@ -1212,7 +1236,7 @@ func TestBoostRepoConnect(t *testing.T) {
 			getter:          new(credmock.SecretStore),
 			credStoreReturn: "fake-password",
 			funcTable: localFuncTable{
-				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string) (map[string]string, error) {
+				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string, map[string]string) (map[string]string, error) {
 					return map[string]string{}, nil
 				},
 				getStorageCredentials: func(*velerov1api.BackupStorageLocation, velerocredentials.FileStore) (map[string]string, error) {
@@ -1239,7 +1263,7 @@ func TestBoostRepoConnect(t *testing.T) {
 			getter:          new(credmock.SecretStore),
 			credStoreReturn: "fake-password",
 			funcTable: localFuncTable{
-				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string) (map[string]string, error) {
+				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string, map[string]string) (map[string]string, error) {
 					return map[string]string{}, nil
 				},
 				getStorageCredentials: func(*velerov1api.BackupStorageLocation, velerocredentials.FileStore) (map[string]string, error) {
@@ -1265,7 +1289,7 @@ func TestBoostRepoConnect(t *testing.T) {
 			getter:          new(credmock.SecretStore),
 			credStoreReturn: "fake-password",
 			funcTable: localFuncTable{
-				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string) (map[string]string, error) {
+				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string, map[string]string) (map[string]string, error) {
 					return map[string]string{}, nil
 				},
 				getStorageCredentials: func(*velerov1api.BackupStorageLocation, velerocredentials.FileStore) (map[string]string, error) {
@@ -1352,7 +1376,7 @@ func TestPruneRepo(t *testing.T) {
 			getter:          new(credmock.SecretStore),
 			credStoreReturn: "fake-password",
 			funcTable: localFuncTable{
-				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string) (map[string]string, error) {
+				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string, map[string]string) (map[string]string, error) {
 					return map[string]string{}, nil
 				},
 				getStorageCredentials: func(*velerov1api.BackupStorageLocation, velerocredentials.FileStore) (map[string]string, error) {
@@ -1370,7 +1394,7 @@ func TestPruneRepo(t *testing.T) {
 			getter:          new(credmock.SecretStore),
 			credStoreReturn: "fake-password",
 			funcTable: localFuncTable{
-				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string) (map[string]string, error) {
+				getStorageVariables: func(*velerov1api.BackupStorageLocation, string, string, map[string]string) (map[string]string, error) {
 					return map[string]string{}, nil
 				},
 				getStorageCredentials: func(*velerov1api.BackupStorageLocation, velerocredentials.FileStore) (map[string]string, error) {

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

@@ -33,8 +33,7 @@ import (
 )
 
 const (
-	maxDataCacheMB         = 2000
-	maxMetadataCacheMB     = 2000
+	defaultCacheLimitMB    = 2000
 	maxCacheDurationSecond = 30
 )
 
@@ -67,11 +66,21 @@ func SetupNewRepositoryOptions(ctx context.Context, flags map[string]string) rep
 
 // SetupConnectOptions setups the options when connecting to an existing Kopia repository
 func SetupConnectOptions(ctx context.Context, repoOptions udmrepo.RepoOptions) repo.ConnectOptions {
+	cacheLimit := optionalHaveIntWithDefault(ctx, udmrepo.StoreOptionCacheLimit, repoOptions.StorageOptions, defaultCacheLimitMB) << 20
+
+	// 80% for data cache and 20% for metadata cahce and align to KB
+	dataCacheLimit := (cacheLimit / 5 * 4) >> 10
+	metadataCacheLimit := (cacheLimit / 5) >> 10
+
 	return repo.ConnectOptions{
 		CachingOptions: content.CachingOptions{
-			ContentCacheSizeBytes:  maxDataCacheMB << 20,
-			MetadataCacheSizeBytes: maxMetadataCacheMB << 20,
-			MaxListCacheDuration:   content.DurationSeconds(time.Duration(maxCacheDurationSecond) * time.Second),
+			// softLimit 80%
+			ContentCacheSizeBytes:  (dataCacheLimit / 5 * 4) << 10,
+			MetadataCacheSizeBytes: (metadataCacheLimit / 5 * 4) << 10,
+			// hardLimit 100%
+			ContentCacheSizeLimitBytes:  dataCacheLimit << 10,
+			MetadataCacheSizeLimitBytes: metadataCacheLimit << 10,
+			MaxListCacheDuration:        content.DurationSeconds(time.Duration(maxCacheDurationSecond) * time.Second),
 		},
 		ClientOptions: repo.ClientOptions{
 			Hostname:    optionalHaveString(udmrepo.GenOptionOwnerDomain, repoOptions.GeneralOptions),

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

@@ -98,6 +98,21 @@ func optionalHaveBase64(ctx context.Context, key string, flags map[string]string
 	return nil
 }
 
+func optionalHaveIntWithDefault(ctx context.Context, key string, flags map[string]string, defValue int64) int64 {
+	if value, exist := flags[key]; exist {
+		if value != "" {
+			ret, err := strconv.ParseInt(value, 10, 64)
+			if err == nil {
+				return ret
+			}
+
+			backendLog()(ctx).Errorf("Ignore %s, value [%s] is invalid, err %v", key, value, err)
+		}
+	}
+
+	return defValue
+}
+
 func backendLog() func(ctx context.Context) logging.Logger {
 	return logging.Module("kopialib-bd")
 }

pkg/repository/udmrepo/repo_options.go

@@ -63,6 +63,8 @@ const (
 	StoreOptionGenRetentionPeriod = "retentionPeriod"
 	StoreOptionGenReadOnly        = "readOnly"
 
+	StoreOptionCacheLimit = "cacheLimitMB"
+
 	ThrottleOptionReadOps       = "readOPS"
 	ThrottleOptionWriteOps      = "writeOPS"
 	ThrottleOptionListOps       = "listOPS"