refactor: Pass parallelism flag to snapshot restore command (#2959)

Co-authored-by: Vivek Singh <vivek.singh@veeam.com>
Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>

helm/kanister-operator/templates/deployment.yaml

@@ -40,7 +40,9 @@ spec:
         - name: LOG_LEVEL
           value: {{ .Values.controller.logLevel }}
         - name: DATA_STORE_PARALLEL_UPLOAD
-          value: {{ .Values.controller.parallelism | quote }}
+          value: {{ .Values.dataStore.parallelism.upload | quote }}
+        - name: DATA_STORE_PARALLEL_DOWNLOAD
+          value: {{ .Values.dataStore.parallelism.download | quote }}
         - name: KANISTER_METRICS_ENABLED
           value: {{ .Values.controller.metrics.enabled | quote }}
 {{- if .Values.resources }}

helm/kanister-operator/values.yaml

@@ -31,12 +31,15 @@ controller:
   # false : CRDs would be created by helm
   # true : CRDs would be created by kanister controller
   updateCRDs: true
-  parallelism: 8
   metrics:
     # metrics.enabled specified if the kanister-prometheus framework has been enabled
     # false : kanister-prometheus framework has been disabled
     # true: kanister-prometheus framework has been enabled
     enabled: false
+dataStore:
+  parallelism:
+    upload: 8
+    download: 8
 bpValidatingWebhook:
   enabled: true
   # `tls` field is used to specify TLS information for both blueprint and repositoryserver validating webhook server 

pkg/function/restore_data_using_kopia_server.go

@@ -33,6 +33,7 @@ import (
 	"github.com/kanisterio/kanister/pkg/kube"
 	"github.com/kanisterio/kanister/pkg/param"
 	"github.com/kanisterio/kanister/pkg/progress"
+	"github.com/kanisterio/kanister/pkg/utils"
 )
 
 const (
@@ -286,6 +287,7 @@ func restoreDataFromServerPodFunc(
 				TargetPath:             restorePath,
 				SparseRestore:          sparseRestore,
 				IgnorePermissionErrors: true,
+				Parallelism:            utils.GetEnvAsIntOrDefault(kankopia.DataStoreParallelDownloadName, kankopia.DefaultDataStoreParallelDownload),
 			})
 
 		stdout.Reset()

pkg/kopia/command/restore.go

@@ -14,11 +14,14 @@
 
 package command
 
+import "strconv"
+
 type RestoreCommandArgs struct {
 	*CommandArgs
 	RootID                 string
 	TargetPath             string
 	IgnorePermissionErrors bool
+	Parallelism            int
 }
 
 // Restore returns the kopia command for restoring root of a snapshot with given root ID
@@ -31,5 +34,10 @@ func Restore(cmdArgs RestoreCommandArgs) []string {
 		args = args.AppendLoggable(noIgnorePermissionsError)
 	}
 
+	if cmdArgs.Parallelism > 0 {
+		parallelismStr := strconv.Itoa(cmdArgs.Parallelism)
+		args = args.AppendLoggableKV(parallelFlag, parallelismStr)
+	}
+
 	return stringSliceCommand(args)
 }

pkg/kopia/command/restore_test.go

@@ -29,6 +29,22 @@ func (kRestore *KopiaRestoreTestSuite) TestRestoreCommands(c *C) {
 		f           func() []string
 		expectedLog string
 	}{
+		{
+			f: func() []string {
+				args := RestoreCommandArgs{
+					CommandArgs: &CommandArgs{
+						RepoPassword:   "encr-key",
+						ConfigFilePath: "path/kopia.config",
+						LogDirectory:   "cache/log",
+					},
+					RootID:      "snapshot-id",
+					TargetPath:  "target/path",
+					Parallelism: 8,
+				}
+				return Restore(args)
+			},
+			expectedLog: "kopia --log-level=error --config-file=path/kopia.config --log-dir=cache/log --password=encr-key restore snapshot-id target/path --no-ignore-permission-errors --parallel=8",
+		},
 		{
 			f: func() []string {
 				args := RestoreCommandArgs{
@@ -55,10 +71,11 @@ func (kRestore *KopiaRestoreTestSuite) TestRestoreCommands(c *C) {
 					RootID:                 "snapshot-id",
 					TargetPath:             "target/path",
 					IgnorePermissionErrors: true,
+					Parallelism:            32,
 				}
 				return Restore(args)
 			},
-			expectedLog: "kopia --log-level=error --config-file=path/kopia.config --log-dir=cache/log --password=encr-key restore snapshot-id target/path --ignore-permission-errors",
+			expectedLog: "kopia --log-level=error --config-file=path/kopia.config --log-dir=cache/log --password=encr-key restore snapshot-id target/path --ignore-permission-errors --parallel=32",
 		},
 	} {
 		cmd := strings.Join(tc.f(), " ")

pkg/kopia/command/snapshot.go

@@ -66,12 +66,17 @@ type SnapshotRestoreCommandArgs struct {
 	TargetPath             string
 	SparseRestore          bool
 	IgnorePermissionErrors bool
+	Parallelism            int
 }
 
 // SnapshotRestore returns kopia command restoring snapshots with given snap ID
 func SnapshotRestore(cmdArgs SnapshotRestoreCommandArgs) []string {
 	args := commonArgs(cmdArgs.CommandArgs)
 	args = args.AppendLoggable(snapshotSubCommand, restoreSubCommand, cmdArgs.SnapID, cmdArgs.TargetPath)
+	if cmdArgs.Parallelism > 0 {
+		parallelismStr := strconv.Itoa(cmdArgs.Parallelism)
+		args = args.AppendLoggableKV(parallelFlag, parallelismStr)
+	}
 	if cmdArgs.IgnorePermissionErrors {
 		args = args.AppendLoggable(ignorePermissionsError)
 	} else {

pkg/kopia/command/snapshot_test.go

@@ -79,21 +79,23 @@ func (kSnapshot *KopiaSnapshotTestSuite) TestSnapshotCommands(c *C) {
 					TargetPath:             "target/path",
 					SparseRestore:          false,
 					IgnorePermissionErrors: false,
+					Parallelism:            8,
 				}
 				return SnapshotRestore(args)
 			},
-			expectedLog: "kopia --log-level=error --config-file=path/kopia.config --log-dir=cache/log --password=encr-key snapshot restore snapshot-id target/path --no-ignore-permission-errors",
+			expectedLog: "kopia --log-level=error --config-file=path/kopia.config --log-dir=cache/log --password=encr-key snapshot restore snapshot-id target/path --parallel=8 --no-ignore-permission-errors",
 		},
 		{
 			f: func() []string {
 				args := SnapshotRestoreCommandArgs{
 					CommandArgs: commandArgs,
 					SnapID:      "snapshot-id",
 					TargetPath:  "target/path",
+					Parallelism: 16,
 				}
 				return SnapshotRestore(args)
 			},
-			expectedLog: "kopia --log-level=error --config-file=path/kopia.config --log-dir=cache/log --password=encr-key snapshot restore snapshot-id target/path --no-ignore-permission-errors",
+			expectedLog: "kopia --log-level=error --config-file=path/kopia.config --log-dir=cache/log --password=encr-key snapshot restore snapshot-id target/path --parallel=16 --no-ignore-permission-errors",
 		},
 		{
 			f: func() []string {

pkg/kopia/const.go

@@ -26,6 +26,11 @@ const (
 	DataStoreParallelUploadName = "DATA_STORE_PARALLEL_UPLOAD"
 	// DefaultDataStoreParallelUpload is the Default Value of Parallelism
 	DefaultDataStoreParallelUpload = 8
+	// DataStoreParallelDownloadName is the Environmental Variable set in Kanister
+	// for Parallelism to be used by Kopia for restore action
+	DataStoreParallelDownloadName = "DATA_STORE_PARALLEL_DOWNLOAD"
+	// DefaultDataStoreParallelDownload is the Default Value of Parallelism
+	DefaultDataStoreParallelDownload = 8
 )
 
 const (