Merge pull request #8093 from Lyndon-Li/backkup-repo-config

Backup repo config

changelogs/unreleased/8093-Lyndon-Li

@@ -0,0 +1 @@
+Fix issue #7620, add backup repository configuration implementation and support cacheLimit configuration for Kopia repo

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:

design/backup-repo-config.md

@@ -86,18 +86,6 @@ For any reason, if the configMap doesn't effect, nothing is specified to the bac
 The BackupRepository configMap supports backup repository type specific configurations, even though users can only specify one configMap.  
 So in the configMap struct, multiple entries are supported, indexed by the backup repository type. During the backup repository creation, the configMap is searched by the repository type.  
 
-Below are the struct for the configMap:  
-``` golang
-type RepoConfig struct {
-	CacheLimitMB int `json:"cacheLimitMB,omitempty"`
-	EnableCompression int `json:"enableCompression,omitempty"`
-}
-
-type RepoConfigs struct {
-	Configs map[string]RepoConfig `json:"configs"`
-}
-```
-
 ### Configurations
 
 With the above mechanisms, any kind of configuration could be added. Here list the configurations defined at present:  
@@ -106,22 +94,6 @@ With the above mechanisms, any kind of configuration could be added. Here list t
 
 ### Sample
 Below is an example of the BackupRepository configMap with the configurations:     
-json format:  
-```json
-{
-  "configs": {
-    "repo-type-1": {
-      "cacheLimitMB": 2048,
-      "enableCompression": true    
-    },
-    "repo-type-2": {
-      "cacheLimitMB": 1024,
-      "enableCompression": false    
-    }
-  }
-}
-```  
-yaml format:  
 ```yaml
 apiVersion: v1
 kind: ConfigMap
@@ -129,25 +101,20 @@ metadata:
   name: <config-name>
   namespace: velero
 data:
-  configs: |
+  <repository-type-1>: |
     {
-      "repo-type-1": {
-        "cacheLimitMB": 2048,
-        "enableCompression": true    
-      },
-      "repo-type-2": {
-        "cacheLimitMB": 1024,
-        "enableCompression": false    
-      }
+      "cacheLimitMB": 2048,
+      "enableCompression": true    
+    }
+  <repository-type-2>: |
+    {
+      "cacheLimitMB": 1,
+      "enableCompression": false    
     }        
 ```
 
 To create the configMap, users need to save something like the above sample to a file and then run below commands:  
 ```
-kubectl create cm <config-name> -n velero --from-file=<json file name>
-```  
-Or
-```
 kubectl apply -f <yaml file name>
 ```  
 

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
@@ -878,7 +881,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).Warn("Failed to get repo config, repo config is ignored")
+	} 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, mockOn string, arg interface{}, ret
 		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":       "",
+				"cacheLimitMB": "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    = 5000
 	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 cache 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/common_test.go

@@ -111,9 +111,11 @@ func TestSetupNewRepositoryOptions(t *testing.T) {
 
 func TestSetupConnectOptions(t *testing.T) {
 	defaultCacheOption := content.CachingOptions{
-		ContentCacheSizeBytes:  2000 << 20,
-		MetadataCacheSizeBytes: 2000 << 20,
-		MaxListCacheDuration:   content.DurationSeconds(time.Duration(30) * time.Second),
+		ContentCacheSizeBytes:       3200 << 20,
+		MetadataCacheSizeBytes:      800 << 20,
+		ContentCacheSizeLimitBytes:  4000 << 20,
+		MetadataCacheSizeLimitBytes: 1000 << 20,
+		MaxListCacheDuration:        content.DurationSeconds(time.Duration(30) * time.Second),
 	}
 
 	testCases := []struct {

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/kopialib/backend/utils_test.go

@@ -90,3 +90,68 @@ func TestOptionalHaveBool(t *testing.T) {
 		})
 	}
 }
+
+func TestOptionalHaveIntWithDefault(t *testing.T) {
+	var expectMsg string
+	testCases := []struct {
+		name         string
+		key          string
+		flags        map[string]string
+		defaultValue int64
+		logger       *storagemocks.Core
+		retFuncCheck func(mock.Arguments)
+		expectMsg    string
+		retValue     int64
+	}{
+		{
+			name:         "key not exist",
+			key:          "fake-key",
+			flags:        map[string]string{},
+			defaultValue: 2000,
+			retValue:     2000,
+		},
+		{
+			name: "value valid",
+			key:  "fake-key",
+			flags: map[string]string{
+				"fake-key": "1000",
+			},
+			retValue: 1000,
+		},
+		{
+			name: "value invalid",
+			key:  "fake-key",
+			flags: map[string]string{
+				"fake-key": "fake-value",
+			},
+			logger: new(storagemocks.Core),
+			retFuncCheck: func(args mock.Arguments) {
+				ent := args[0].(zapcore.Entry)
+				if ent.Level == zapcore.ErrorLevel {
+					expectMsg = ent.Message
+				}
+			},
+			expectMsg:    "Ignore fake-key, value [fake-value] is invalid, err strconv.ParseInt: parsing \"fake-value\": invalid syntax",
+			defaultValue: 2000,
+			retValue:     2000,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			if tc.logger != nil {
+				tc.logger.On("Enabled", mock.Anything).Return(true)
+				tc.logger.On("Check", mock.Anything, mock.Anything).Run(tc.retFuncCheck).Return(&zapcore.CheckedEntry{})
+			}
+
+			ctx := logging.WithLogger(context.Background(), func(module string) logging.Logger {
+				return zap.New(tc.logger).Sugar()
+			})
+
+			retValue := optionalHaveIntWithDefault(ctx, tc.key, tc.flags, tc.defaultValue)
+
+			require.Equal(t, retValue, tc.retValue)
+			require.Equal(t, tc.expectMsg, expectMsg)
+		})
+	}
+}

pkg/repository/udmrepo/repo_options.go

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