Sync object pools in ReplicateRepositoryjc-repl-object-pools

3 files changed, 146 insertions, 3 deletions

diff --git a/changelogs/unreleased/jc-repl-object-pools.yml b/changelogs/unreleased/jc-repl-object-pools.yml
new file mode 100644
index 000000000..7caa4d808
--- /dev/null
+++ b/changelogs/unreleased/jc-repl-object-pools.yml
@@ -0,0 +1,5 @@
+---
+title: Sync object pools in ReplicateRepository
+merge_request: 1705
+author:
+type: changed
diff --git a/internal/service/repository/replicate.go b/internal/service/repository/replicate.go
index 9cc5eae01..81e7196fc 100644
--- a/internal/service/repository/replicate.go
+++ b/internal/service/repository/replicate.go
@@ -8,11 +8,11 @@ import (
 	"os"
 	"path/filepath"
 
-	"gitlab.com/gitlab-org/gitaly/internal/safe"
-
 	"gitlab.com/gitlab-org/gitaly/client"
 	"gitlab.com/gitlab-org/gitaly/internal/config"
+	"gitlab.com/gitlab-org/gitaly/internal/git/objectpool"
 	"gitlab.com/gitlab-org/gitaly/internal/helper"
+	"gitlab.com/gitlab-org/gitaly/internal/safe"
 	"gitlab.com/gitlab-org/gitaly/proto/go/gitalypb"
 	"gitlab.com/gitlab-org/gitaly/streamio"
 	"golang.org/x/sync/errgroup"
@@ -35,6 +35,7 @@ func (s *server) ReplicateRepository(ctx context.Context, in *gitalypb.Replicate
 	for _, f := range []func(context.Context, *gitalypb.ReplicateRepositoryRequest) error{
 		syncRepository,
 		syncInfoAttributes,
+		s.syncObjectPool,
 	} {
 		f := f // rescoping f
 		g.Go(func() error { return f(outgoingCtx, in) })
@@ -132,6 +133,38 @@ func syncInfoAttributes(ctx context.Context, in *gitalypb.ReplicateRepositoryReq
 	return os.Rename(attributesPath, attributesPath)
 }
 
+func (s *server) syncObjectPool(ctx context.Context, in *gitalypb.ReplicateRepositoryRequest) error {
+	objectPoolClient, err := newObjectPoolClient(ctx, in.GetSource().GetStorageName())
+	if err != nil {
+		return err
+	}
+
+	resp, err := objectPoolClient.GetObjectPool(ctx, &gitalypb.GetObjectPoolRequest{
+		Repository: in.GetSource(),
+	})
+	if err != nil {
+		return err
+	}
+
+	if resp.GetObjectPool() == nil {
+		return nil
+	}
+
+	targetObjectPool, err := objectpool.NewObjectPool(
+		in.GetRepository().GetStorageName(),
+		resp.GetObjectPool().GetRepository().GetRelativePath(),
+	)
+	if err != nil {
+		return err
+	}
+
+	if targetObjectPool.Exists() && targetObjectPool.IsValid() {
+		return targetObjectPool.Link(ctx, in.GetRepository())
+	}
+
+	return targetObjectPool.Init(ctx)
+}
+
 // newRemoteClient creates a new RemoteClient that talks to the same gitaly server
 func newRemoteClient() (gitalypb.RemoteServiceClient, error) {
 	conn, err := client.Dial(fmt.Sprintf("unix:%s", config.GitalyInternalSocketPath()), nil)
@@ -151,3 +184,13 @@ func newRepoClient(ctx context.Context, storageName string) (gitalypb.Repository
 
 	return gitalypb.NewRepositoryServiceClient(conn), nil
 }
+
+// newObjectPoolClient creates a new RepositoryClient that talks to the gitaly of the source repository
+func newObjectPoolClient(ctx context.Context, storageName string) (gitalypb.ObjectPoolServiceClient, error) {
+	conn, err := helper.ClientConnection(ctx, storageName)
+	if err != nil {
+		return nil, err
+	}
+
+	return gitalypb.NewObjectPoolServiceClient(conn), nil
+}
diff --git a/internal/service/repository/replicate_test.go b/internal/service/repository/replicate_test.go
index b82c96ac3..189531dc4 100644
--- a/internal/service/repository/replicate_test.go
+++ b/internal/service/repository/replicate_test.go
@@ -8,6 +8,8 @@ import (
 	"path/filepath"
 	"testing"
 
+	"gitlab.com/gitlab-org/gitaly/internal/git/objectpool"
+
 	"github.com/stretchr/testify/require"
 	"gitlab.com/gitlab-org/gitaly/internal/config"
 	"gitlab.com/gitlab-org/gitaly/internal/helper"
@@ -25,9 +27,10 @@ func TestReplicateRepository(t *testing.T) {
 	replicaPath := filepath.Join(tmpPath, "replica")
 	require.NoError(t, os.MkdirAll(replicaPath, 0755))
 
+	storages := config.Config.Storages
 	defer func(storages []config.Storage) {
 		config.Config.Storages = storages
-	}(config.Config.Storages)
+	}(storages)
 
 	config.Config.Storages = []config.Storage{
 		config.Storage{
@@ -93,6 +96,98 @@ func TestReplicateRepository(t *testing.T) {
 	)
 }
 
+func TestReplicateRepositoryObjectPool(t *testing.T) {
+	tmpPath, cleanup := testhelper.TempDir(t, testhelper.GitlabTestStoragePath(), t.Name())
+	defer require.NoError(t, cleanup())
+
+	replicaPath := filepath.Join(tmpPath, "replica")
+	require.NoError(t, os.MkdirAll(replicaPath, 0755))
+
+	storages := config.Config.Storages
+	defer func(storages []config.Storage) {
+		config.Config.Storages = storages
+	}(storages)
+
+	config.Config.Storages = []config.Storage{
+		config.Storage{
+			Name: "default",
+			Path: testhelper.GitlabTestStoragePath(),
+		},
+		config.Storage{
+			Name: "replica",
+			Path: replicaPath,
+		},
+	}
+
+	server, serverSocketPath := runFullServer(t)
+	defer server.Stop()
+
+	testRepo, _, cleanupRepo := testhelper.NewTestRepo(t)
+	defer cleanupRepo()
+
+	config.Config.SocketPath = serverSocketPath
+
+	repoClient, conn := repository.NewRepositoryClient(t, serverSocketPath)
+	defer conn.Close()
+
+	// create object pool on the source
+	objectPoolPath := testhelper.NewTestObjectPoolName(t)
+	pool, err := objectpool.NewObjectPool(testRepo.GetStorageName(), objectPoolPath)
+	require.NoError(t, err)
+
+	poolCtx, cancel := testhelper.Context()
+
+	require.NoError(t, pool.Create(poolCtx, testRepo))
+	require.NoError(t, pool.Link(poolCtx, testRepo))
+	cancel()
+
+	ctx, cancel := testhelper.Context()
+	defer cancel()
+	md := testhelper.GitalyServersMetadata(t, serverSocketPath)
+	injectedCtx := metadata.NewOutgoingContext(ctx, md)
+
+	targetRepo := *testRepo
+	targetRepo.StorageName = "replica"
+
+	_, err = repoClient.ReplicateRepository(injectedCtx, &gitalypb.ReplicateRepositoryRequest{
+		Repository: &targetRepo,
+		Source:     testRepo,
+	})
+	require.NoError(t, err)
+
+	targetRepoPath, err := helper.GetRepoPath(&targetRepo)
+	require.NoError(t, err)
+
+	testhelper.MustRunCommand(t, nil, "git", "-C", targetRepoPath, "fsck")
+
+	// replicate object pool repository
+	targetObjectPoolRepo := *pool.ToProto().GetRepository()
+	targetObjectPoolRepo.StorageName = "replica"
+
+	ctx, cancel = testhelper.Context()
+	defer cancel()
+	injectedCtx = metadata.NewOutgoingContext(ctx, md)
+
+	_, err = repoClient.ReplicateRepository(injectedCtx, &gitalypb.ReplicateRepositoryRequest{
+		Repository: &targetObjectPoolRepo,
+		Source:     pool.ToProto().GetRepository(),
+	})
+	require.NoError(t, err)
+
+	ctx, cancel = testhelper.Context()
+	defer cancel()
+	injectedCtx = metadata.NewOutgoingContext(ctx, md)
+
+	_, err = repoClient.ReplicateRepository(injectedCtx, &gitalypb.ReplicateRepositoryRequest{
+		Repository: &targetRepo,
+		Source:     testRepo,
+	})
+	require.NoError(t, err)
+
+	testhelper.MustRunCommand(t, nil, "git", "-C", targetRepoPath, "gc")
+	require.True(t, getGitObjectDirSize(t, targetRepoPath) < 100, "expect a small object directory size")
+}
+
 func TestReplicateRepositoryInvalidArguments(t *testing.T) {
 	testCases := []struct {
 		description   string