gitaly-backup: Restore repositories as per backup.rake

Changelog: added

4 files changed, 235 insertions, 0 deletions

diff --git a/changelogs/unreleased/gitaly-backup-restore.yml b/changelogs/unreleased/gitaly-backup-restore.yml
new file mode 100644
index 000000000..ff2fcfb67
--- /dev/null
+++ b/changelogs/unreleased/gitaly-backup-restore.yml
@@ -0,0 +1,5 @@
+---
+title: 'gitlab-backup: Restore repositories as per backup.rake'
+merge_request: 3383
+author:
+type: added
diff --git a/internal/backup/backup.go b/internal/backup/backup.go
index ad04c8d0a..eee51d462 100644
--- a/internal/backup/backup.go
+++ b/internal/backup/backup.go
@@ -59,6 +59,35 @@ func (fs *Filesystem) BackupRepository(ctx context.Context, server storage.Serve
 	return nil
 }
 
+// RestoreRepository restores a repository from a backup on a local filesystem
+func (fs *Filesystem) RestoreRepository(ctx context.Context, server storage.ServerInfo, repo *gitalypb.Repository, alwaysCreate bool) error {
+	backupPath := strings.TrimSuffix(filepath.Join(fs.path, repo.RelativePath), ".git")
+	bundlePath := backupPath + ".bundle"
+	customHooksPath := filepath.Join(backupPath, "custom_hooks.tar")
+
+	if err := fs.removeRepository(ctx, server, repo); err != nil {
+		return fmt.Errorf("restore: %w", err)
+	}
+	if err := fs.restoreBundle(ctx, bundlePath, server, repo); err != nil {
+		// For compatibility with existing backups we need to always create the
+		// repository even if there's no bundle for project repositories
+		// (not wiki or snippet repositories).  Gitaly does not know which
+		// repository is which type so here we accept a parameter to tell us
+		// to employ this behaviour.
+		if alwaysCreate && errors.Is(err, ErrSkipped) {
+			if err := fs.createRepository(ctx, server, repo); err != nil {
+				return fmt.Errorf("restore: %w", err)
+			}
+		} else {
+			return fmt.Errorf("restore: %w", err)
+		}
+	}
+	if err := fs.restoreCustomHooks(ctx, customHooksPath, server, repo); err != nil {
+		return fmt.Errorf("restore: %w", err)
+	}
+	return nil
+}
+
 func (fs *Filesystem) isEmpty(ctx context.Context, server storage.ServerInfo, repo *gitalypb.Repository) (bool, error) {
 	repoClient, err := fs.newRepoClient(ctx, server)
 	if err != nil {
@@ -74,6 +103,28 @@ func (fs *Filesystem) isEmpty(ctx context.Context, server storage.ServerInfo, re
 	return !hasLocalBranches.GetValue(), nil
 }
 
+func (fs *Filesystem) removeRepository(ctx context.Context, server storage.ServerInfo, repo *gitalypb.Repository) error {
+	repoClient, err := fs.newRepoClient(ctx, server)
+	if err != nil {
+		return fmt.Errorf("remove repository: %w", err)
+	}
+	if _, err := repoClient.RemoveRepository(ctx, &gitalypb.RemoveRepositoryRequest{Repository: repo}); err != nil {
+		return fmt.Errorf("remove repository: %w", err)
+	}
+	return nil
+}
+
+func (fs *Filesystem) createRepository(ctx context.Context, server storage.ServerInfo, repo *gitalypb.Repository) error {
+	repoClient, err := fs.newRepoClient(ctx, server)
+	if err != nil {
+		return fmt.Errorf("create repository: %w", err)
+	}
+	if _, err := repoClient.CreateRepository(ctx, &gitalypb.CreateRepositoryRequest{Repository: repo}); err != nil {
+		return fmt.Errorf("create repository: %w", err)
+	}
+	return nil
+}
+
 func (fs *Filesystem) writeBundle(ctx context.Context, path string, server storage.ServerInfo, repo *gitalypb.Repository) error {
 	repoClient, err := fs.newRepoClient(ctx, server)
 	if err != nil {
@@ -90,6 +141,45 @@ func (fs *Filesystem) writeBundle(ctx context.Context, path string, server stora
 	return writeFile(path, bundle)
 }
 
+func (fs *Filesystem) restoreBundle(ctx context.Context, path string, server storage.ServerInfo, repo *gitalypb.Repository) error {
+	f, err := os.Open(path)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return fmt.Errorf("%w: bundle does not exist: %q", ErrSkipped, path)
+		}
+		return fmt.Errorf("restore bundle: %w", err)
+	}
+	defer f.Close()
+
+	repoClient, err := fs.newRepoClient(ctx, server)
+	if err != nil {
+		return fmt.Errorf("restore bundle: %q: %w", path, err)
+	}
+	stream, err := repoClient.CreateRepositoryFromBundle(ctx)
+	if err != nil {
+		return fmt.Errorf("restore bundle: %q: %w", path, err)
+	}
+	request := &gitalypb.CreateRepositoryFromBundleRequest{Repository: repo}
+	bundle := streamio.NewWriter(func(p []byte) error {
+		request.Data = p
+		if err := stream.Send(request); err != nil {
+			return err
+		}
+
+		// Only set `Repository` on the first `Send` of the stream
+		request = &gitalypb.CreateRepositoryFromBundleRequest{}
+
+		return nil
+	})
+	if _, err := io.Copy(bundle, f); err != nil {
+		return fmt.Errorf("restore bundle: %q: %w", path, err)
+	}
+	if _, err = stream.CloseAndRecv(); err != nil {
+		return fmt.Errorf("restore bundle: %q: %w", path, err)
+	}
+	return nil
+}
+
 func (fs *Filesystem) writeCustomHooks(ctx context.Context, path string, server storage.ServerInfo, repo *gitalypb.Repository) error {
 	repoClient, err := fs.newRepoClient(ctx, server)
 	if err != nil {
@@ -109,6 +199,46 @@ func (fs *Filesystem) writeCustomHooks(ctx context.Context, path string, server
 	return nil
 }
 
+func (fs *Filesystem) restoreCustomHooks(ctx context.Context, path string, server storage.ServerInfo, repo *gitalypb.Repository) error {
+	f, err := os.Open(path)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil
+		}
+		return fmt.Errorf("restore custom hooks: %w", err)
+	}
+	defer f.Close()
+
+	repoClient, err := fs.newRepoClient(ctx, server)
+	if err != nil {
+		return fmt.Errorf("restore custom hooks, %q: %w", path, err)
+	}
+	stream, err := repoClient.RestoreCustomHooks(ctx)
+	if err != nil {
+		return fmt.Errorf("restore custom hooks, %q: %w", path, err)
+	}
+
+	request := &gitalypb.RestoreCustomHooksRequest{Repository: repo}
+	bundle := streamio.NewWriter(func(p []byte) error {
+		request.Data = p
+		if err := stream.Send(request); err != nil {
+			return err
+		}
+
+		// Only set `Repository` on the first `Send` of the stream
+		request = &gitalypb.RestoreCustomHooksRequest{}
+
+		return nil
+	})
+	if _, err := io.Copy(bundle, f); err != nil {
+		return fmt.Errorf("restore custom hooks, %q: %w", path, err)
+	}
+	if _, err = stream.CloseAndRecv(); err != nil {
+		return fmt.Errorf("restore custom hooks, %q: %w", path, err)
+	}
+	return nil
+}
+
 func (fs *Filesystem) newRepoClient(ctx context.Context, server storage.ServerInfo) (gitalypb.RepositoryServiceClient, error) {
 	conn, err := fs.conns.Dial(ctx, server.Address, server.Token)
 	if err != nil {
diff --git a/internal/backup/backup_test.go b/internal/backup/backup_test.go
index ce85f235f..9ff37e371 100644
--- a/internal/backup/backup_test.go
+++ b/internal/backup/backup_test.go
@@ -1,6 +1,7 @@
 package backup
 
 import (
+	"errors"
 	"io/ioutil"
 	"os"
 	"path/filepath"
@@ -99,3 +100,88 @@ func TestFilesystem_BackupRepository(t *testing.T) {
 		})
 	}
 }
+
+func TestFilesystem_RestoreRepository(t *testing.T) {
+	cfg := testcfg.Build(t)
+	testhelper.ConfigureGitalyHooksBinary(cfg.BinDir)
+
+	gitalyAddr := testserver.RunGitalyServer(t, cfg, nil, setup.RegisterAll)
+
+	path := testhelper.TempDir(t)
+
+	existingRepo, existRepoPath, _ := gittest.CloneRepoAtStorage(t, cfg.Storages[0], "existing_repo")
+	existingRepoPath := filepath.Join(path, existingRepo.RelativePath)
+	existingRepoBundlePath := existingRepoPath + ".bundle"
+	existingRepoCustomHooksPath := filepath.Join(existingRepoPath, "custom_hooks.tar")
+	require.NoError(t, os.MkdirAll(existingRepoPath, os.ModePerm))
+
+	gittest.Exec(t, cfg, "-C", existRepoPath, "bundle", "create", existingRepoBundlePath, "--all")
+	testhelper.CopyFile(t, "../gitaly/service/repository/testdata/custom_hooks.tar", existingRepoCustomHooksPath)
+
+	newRepo := gittest.InitRepoDir(t, cfg.Storages[0].Path, "new_repo")
+	newRepoBundlePath := filepath.Join(path, newRepo.RelativePath+".bundle")
+	testhelper.CopyFile(t, existingRepoBundlePath, newRepoBundlePath)
+
+	missingBundleRepo := gittest.InitRepoDir(t, cfg.Storages[0].Path, "missing_bundle")
+	missingBundleRepoAlwaysCreate := gittest.InitRepoDir(t, cfg.Storages[0].Path, "missing_bundle_always_create")
+
+	for _, tc := range []struct {
+		desc          string
+		repo          *gitalypb.Repository
+		alwaysCreate  bool
+		expectedPaths []string
+		expectedErrAs error
+		expectVerify  bool
+	}{
+		{
+			desc:         "new repo, without hooks",
+			repo:         newRepo,
+			expectVerify: true,
+		},
+		{
+			desc: "existing repo, with hooks",
+			repo: existingRepo,
+			expectedPaths: []string{
+				"custom_hooks/pre-commit.sample",
+				"custom_hooks/prepare-commit-msg.sample",
+				"custom_hooks/pre-push.sample",
+			},
+			expectVerify: true,
+		},
+		{
+			desc:          "missing bundle",
+			repo:          missingBundleRepo,
+			expectedErrAs: ErrSkipped,
+		},
+		{
+			desc:         "missing bundle, always create",
+			repo:         missingBundleRepoAlwaysCreate,
+			alwaysCreate: true,
+		},
+	} {
+		t.Run(tc.desc, func(t *testing.T) {
+			repoPath := filepath.Join(cfg.Storages[0].Path, tc.repo.RelativePath)
+			bundlePath := filepath.Join(path, tc.repo.RelativePath+".bundle")
+
+			ctx, cancel := testhelper.Context()
+			defer cancel()
+
+			fsBackup := NewFilesystem(path)
+			err := fsBackup.RestoreRepository(ctx, storage.ServerInfo{Address: gitalyAddr, Token: cfg.Auth.Token}, tc.repo, tc.alwaysCreate)
+			if tc.expectedErrAs != nil {
+				require.True(t, errors.Is(err, tc.expectedErrAs), err.Error())
+			} else {
+				require.NoError(t, err)
+			}
+
+			if tc.expectVerify {
+				output := gittest.Exec(t, cfg, "-C", repoPath, "bundle", "verify", bundlePath)
+				require.Contains(t, string(output), "The bundle records a complete history")
+			}
+
+			for _, p := range tc.expectedPaths {
+				require.FileExists(t, filepath.Join(repoPath, p))
+			}
+		})
+	}
+}
diff --git a/internal/testhelper/testhelper.go b/internal/testhelper/testhelper.go
index 01b35d7dd..c5557c391 100644
--- a/internal/testhelper/testhelper.go
+++ b/internal/testhelper/testhelper.go
@@ -178,6 +178,20 @@ func MustClose(t testing.TB, closer io.Closer) {
 	require.NoError(t, closer.Close())
 }
 
+// CopyFile copies a file at the path src to a file at the path dst
+func CopyFile(t testing.TB, src, dst string) {
+	fsrc, err := os.Open(src)
+	require.NoError(t, err)
+	defer MustClose(t, fsrc)
+
+	fdst, err := os.Create(dst)
+	require.NoError(t, err)
+	defer MustClose(t, fdst)
+
+	_, err = io.Copy(fdst, fsrc)
+	require.NoError(t, err)
+}
+
 // GetTemporaryGitalySocketFileName will return a unique, useable socket file name
 func GetTemporaryGitalySocketFileName(t testing.TB) string {
 	require.NotEmpty(t, testDirectory, "you must call testhelper.Configure() before GetTemporaryGitalySocketFileName()")