gitaly-backup: Add -server-side option to restore sub-command

Changelog: added

2 files changed, 100 insertions, 11 deletions

diff --git a/cmd/gitaly-backup/restore.go b/cmd/gitaly-backup/restore.go
index 3f9426c6a..c81b3e1aa 100644
--- a/cmd/gitaly-backup/restore.go
+++ b/cmd/gitaly-backup/restore.go
@@ -33,6 +33,7 @@ type restoreSubcommand struct {
 	layout                string
 	removeAllRepositories []string
 	backupID              string
+	serverSide            bool
 }
 
 func (cmd *restoreSubcommand) Flags(fs *flag.FlagSet) {
@@ -45,23 +46,32 @@ func (cmd *restoreSubcommand) Flags(fs *flag.FlagSet) {
 		return nil
 	})
 	fs.StringVar(&cmd.backupID, "id", "", "ID of full backup to restore. If not specified, the latest backup is restored.")
+	fs.BoolVar(&cmd.serverSide, "server-side", false, "use server-side backups. Note: The feature is not ready for production use.")
 }
 
 func (cmd *restoreSubcommand) Run(ctx context.Context, stdin io.Reader, stdout io.Writer) error {
-	sink, err := backup.ResolveSink(ctx, cmd.backupPath)
-	if err != nil {
-		return fmt.Errorf("restore: resolve sink: %w", err)
-	}
-
-	locator, err := backup.ResolveLocator(cmd.layout, sink)
-	if err != nil {
-		return fmt.Errorf("restore: resolve locator: %w", err)
-	}
-
 	pool := client.NewPool(internalclient.UnaryInterceptor(), internalclient.StreamInterceptor())
 	defer pool.Close()
 
-	manager := backup.NewManager(sink, locator, pool)
+	var manager backup.Strategy
+	if cmd.serverSide {
+		if cmd.backupPath != "" {
+			return fmt.Errorf("restore: path cannot be used with server-side backups")
+		}
+
+		manager = backup.NewServerSideAdapter(pool)
+	} else {
+		sink, err := backup.ResolveSink(ctx, cmd.backupPath)
+		if err != nil {
+			return fmt.Errorf("restore: resolve sink: %w", err)
+		}
+		locator, err := backup.ResolveLocator(cmd.layout, sink)
+		if err != nil {
+			return fmt.Errorf("restore: resolve locator: %w", err)
+		}
+		manager = backup.NewManager(sink, locator, pool)
+	}
+
 	logger := log.StandardLogger()
 
 	for _, storageName := range cmd.removeAllRepositories {
diff --git a/cmd/gitaly-backup/restore_test.go b/cmd/gitaly-backup/restore_test.go
index 3492c6c15..440d2b91e 100644
--- a/cmd/gitaly-backup/restore_test.go
+++ b/cmd/gitaly-backup/restore_test.go
@@ -12,6 +12,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/require"
+	"gitlab.com/gitlab-org/gitaly/v16/internal/backup"
 	"gitlab.com/gitlab-org/gitaly/v16/internal/git/gittest"
 	"gitlab.com/gitlab-org/gitaly/v16/internal/gitaly/service/setup"
 	"gitlab.com/gitlab-org/gitaly/v16/internal/testhelper"
@@ -88,3 +89,81 @@ func TestRestoreSubcommand(t *testing.T) {
 		require.Contains(t, string(output), "The bundle records a complete history")
 	}
 }
+
+func TestRestoreSubcommand_serverSide(t *testing.T) {
+	t.Parallel()
+	ctx := testhelper.Context(t)
+
+	path := testhelper.TempDir(t)
+	backupSink, err := backup.ResolveSink(ctx, path)
+	require.NoError(t, err)
+
+	backupLocator, err := backup.ResolveLocator("pointer", backupSink)
+	require.NoError(t, err)
+
+	cfg := testcfg.Build(t)
+	testcfg.BuildGitalyHooks(t, cfg)
+
+	cfg.SocketPath = testserver.RunGitalyServer(t, cfg, setup.RegisterAll,
+		testserver.WithBackupSink(backupSink),
+		testserver.WithBackupLocator(backupLocator),
+	)
+
+	existingRepo, existRepoPath := gittest.CreateRepository(t, ctx, cfg, gittest.CreateRepositoryConfig{
+		Seed:         gittest.SeedGitLabTest,
+		RelativePath: "existing_repo",
+	})
+
+	existingRepoBundlePath := filepath.Join(path, existingRepo.RelativePath+".bundle")
+	gittest.Exec(t, cfg, "-C", existRepoPath, "bundle", "create", existingRepoBundlePath, "--all")
+
+	var repos []*gitalypb.Repository
+	for i := 0; i < 2; i++ {
+		repo := gittest.InitRepoDir(t, cfg.Storages[0].Path, fmt.Sprintf("repo-%d", i))
+		repoBundlePath := filepath.Join(path, repo.RelativePath+".bundle")
+		testhelper.CopyFile(t, existingRepoBundlePath, repoBundlePath)
+		repos = append(repos, repo)
+	}
+
+	var stdin bytes.Buffer
+
+	encoder := json.NewEncoder(&stdin)
+	for _, repo := range repos {
+		require.NoError(t, encoder.Encode(map[string]string{
+			"address":         cfg.SocketPath,
+			"token":           cfg.Auth.Token,
+			"storage_name":    repo.StorageName,
+			"relative_path":   repo.RelativePath,
+			"gl_project_path": repo.GlProjectPath,
+		}))
+	}
+
+	require.NoError(t, encoder.Encode(map[string]string{
+		"address":       "invalid",
+		"token":         "invalid",
+		"relative_path": "invalid",
+	}))
+
+	ctx = testhelper.MergeIncomingMetadata(ctx, testcfg.GitalyServersMetadataFromCfg(t, cfg))
+	cmd := restoreSubcommand{}
+
+	fs := flag.NewFlagSet("restore", flag.ContinueOnError)
+	cmd.Flags(fs)
+
+	require.DirExists(t, existRepoPath)
+
+	require.NoError(t, fs.Parse([]string{"-server-side", "-remove-all-repositories", existingRepo.StorageName}))
+	require.EqualError(t,
+		cmd.Run(ctx, &stdin, io.Discard),
+		"restore: pipeline: 1 failures encountered:\n - invalid: server-side restore: could not dial source: invalid connection string: \"invalid\"\n")
+
+	require.NoDirExists(t, existRepoPath)
+
+	for _, repo := range repos {
+		repoPath := filepath.Join(cfg.Storages[0].Path, gittest.GetReplicaPath(t, ctx, cfg, repo))
+		bundlePath := filepath.Join(path, repo.RelativePath+".bundle")
+
+		output := gittest.Exec(t, cfg, "-C", repoPath, "bundle", "verify", bundlePath)
+		require.Contains(t, string(output), "The bundle records a complete history")
+	}
+}