Merge branch 'restore_concurrency' into 'master'

Refactor backup pipeline to use commands

See merge request gitlab-org/gitaly!3796

2 files changed, 18 insertions, 15 deletions

diff --git a/cmd/gitaly-backup/create.go b/cmd/gitaly-backup/create.go
index 12f087838..9dae3d4a5 100644
--- a/cmd/gitaly-backup/create.go
+++ b/cmd/gitaly-backup/create.go
@@ -48,10 +48,10 @@ func (cmd *createSubcommand) Run(ctx context.Context, stdin io.Reader, stdout io
 
 	manager := backup.NewManager(sink, locator)
 
-	var pipeline backup.CreatePipeline
-	pipeline = backup.NewPipeline(log.StandardLogger(), manager)
+	var pipeline backup.Pipeline
+	pipeline = backup.NewLoggingPipeline(log.StandardLogger())
 	if cmd.parallel > 0 || cmd.parallelStorage > 0 {
-		pipeline = backup.NewParallelCreatePipeline(pipeline, cmd.parallel, cmd.parallelStorage)
+		pipeline = backup.NewParallelPipeline(pipeline, cmd.parallel, cmd.parallelStorage)
 	}
 
 	decoder := json.NewDecoder(stdin)
@@ -67,10 +67,7 @@ func (cmd *createSubcommand) Run(ctx context.Context, stdin io.Reader, stdout io
 			RelativePath:  sr.RelativePath,
 			GlProjectPath: sr.GlProjectPath,
 		}
-		pipeline.Create(ctx, &backup.CreateRequest{
-			Server:     sr.ServerInfo,
-			Repository: &repo,
-		})
+		pipeline.Handle(ctx, backup.NewCreateCommand(manager, sr.ServerInfo, &repo))
 	}
 
 	if err := pipeline.Done(); err != nil {
diff --git a/cmd/gitaly-backup/restore.go b/cmd/gitaly-backup/restore.go
index d5304cbd7..695e98cfb 100644
--- a/cmd/gitaly-backup/restore.go
+++ b/cmd/gitaly-backup/restore.go
@@ -7,6 +7,7 @@ import (
 	"flag"
 	"fmt"
 	"io"
+	"runtime"
 
 	log "github.com/sirupsen/logrus"
 	"gitlab.com/gitlab-org/gitaly/v14/internal/backup"
@@ -23,12 +24,16 @@ type restoreRequest struct {
 }
 
 type restoreSubcommand struct {
-	backupPath string
-	locator    string
+	backupPath      string
+	parallel        int
+	parallelStorage int
+	locator         string
 }
 
 func (cmd *restoreSubcommand) Flags(fs *flag.FlagSet) {
 	fs.StringVar(&cmd.backupPath, "path", "", "repository backup path")
+	fs.IntVar(&cmd.parallel, "parallel", runtime.NumCPU(), "maximum number of parallel restores")
+	fs.IntVar(&cmd.parallelStorage, "parallel-storage", 2, "maximum number of parallel restores per storage. Note: actual parallelism when combined with `-parallel` depends on the order the repositories are received.")
 	fs.StringVar(&cmd.locator, "locator", "legacy", "determines how backup files are located. One of legacy, pointer. Note: The feature is not ready for production use.")
 }
 
@@ -44,7 +49,12 @@ func (cmd *restoreSubcommand) Run(ctx context.Context, stdin io.Reader, stdout i
 	}
 
 	manager := backup.NewManager(sink, locator)
-	pipeline := backup.NewPipeline(log.StandardLogger(), manager)
+
+	var pipeline backup.Pipeline
+	pipeline = backup.NewLoggingPipeline(log.StandardLogger())
+	if cmd.parallel > 0 || cmd.parallelStorage > 0 {
+		pipeline = backup.NewParallelPipeline(pipeline, cmd.parallel, cmd.parallelStorage)
+	}
 
 	decoder := json.NewDecoder(stdin)
 	for {
@@ -60,11 +70,7 @@ func (cmd *restoreSubcommand) Run(ctx context.Context, stdin io.Reader, stdout i
 			RelativePath:  req.RelativePath,
 			GlProjectPath: req.GlProjectPath,
 		}
-		pipeline.Restore(ctx, &backup.RestoreRequest{
-			Server:       req.ServerInfo,
-			Repository:   &repo,
-			AlwaysCreate: req.AlwaysCreate,
-		})
+		pipeline.Handle(ctx, backup.NewRestoreCommand(manager, req.ServerInfo, &repo, req.AlwaysCreate))
 	}
 
 	if err := pipeline.Done(); err != nil {