Port fetch repository as mirror

7 files changed, 250 insertions, 81 deletions

diff --git a/internal/gitalyssh/gitalyssh.go b/internal/gitalyssh/gitalyssh.go
new file mode 100644
index 000000000..718802ae2
--- /dev/null
+++ b/internal/gitalyssh/gitalyssh.go
@@ -0,0 +1,68 @@
+package gitalyssh
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+
+	"github.com/golang/protobuf/jsonpb"
+	"github.com/golang/protobuf/proto"
+	"gitlab.com/gitlab-org/gitaly/internal/config"
+	"gitlab.com/gitlab-org/gitaly/internal/helper"
+	gitaly_x509 "gitlab.com/gitlab-org/gitaly/internal/x509"
+	"gitlab.com/gitlab-org/gitaly/proto/go/gitalypb"
+	"gitlab.com/gitlab-org/labkit/tracing"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+var envInjector = tracing.NewEnvInjector()
+
+func UploadPackEnv(ctx context.Context, req *gitalypb.SSHUploadPackRequest) ([]string, error) {
+	env, err := commandEnv(ctx, req.Repository.StorageName, "upload-pack", req)
+	if err != nil {
+		return nil, err
+	}
+	return envInjector(ctx, env), nil
+}
+
+func commandEnv(ctx context.Context, storageName, command string, message proto.Message) ([]string, error) {
+	var pbMarshaler jsonpb.Marshaler
+	payload, err := pbMarshaler.MarshalToString(message)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "commandEnv: marshalling payload failed: %v", err)
+	}
+
+	serversInfo, err := helper.ExtractGitalyServers(ctx)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "commandEnv: extracting Gitaly servers: %v", err)
+	}
+
+	storageInfo, ok := serversInfo[storageName]
+	if !ok {
+		return nil, status.Errorf(codes.InvalidArgument, "commandEnv: no storage info for %s", storageName)
+	}
+
+	address := storageInfo["address"]
+	if address == "" {
+		return nil, status.Errorf(codes.InvalidArgument, "commandEnv: empty gitaly address")
+	}
+
+	token := storageInfo["token"]
+
+	return []string{
+		fmt.Sprintf("GITALY_PAYLOAD=%s", payload),
+		fmt.Sprintf("GIT_SSH_COMMAND=%s %s", gitalySSHPath(), command),
+		fmt.Sprintf("GITALY_ADDRESS=%s", address),
+		fmt.Sprintf("GITALY_TOKEN=%s", token),
+		// Pass through the SSL_CERT_* variables that indicate which
+		// system certs to trust
+		fmt.Sprintf("%s=%s", gitaly_x509.SSLCertDir, os.Getenv(gitaly_x509.SSLCertDir)),
+		fmt.Sprintf("%s=%s", gitaly_x509.SSLCertFile, os.Getenv(gitaly_x509.SSLCertFile)),
+	}, nil
+}
+
+func gitalySSHPath() string {
+	return filepath.Join(config.Config.BinDir, "gitaly-ssh")
+}
diff --git a/internal/gitalyssh/gitalyssh_test.go b/internal/gitalyssh/gitalyssh_test.go
new file mode 100644
index 000000000..3d4199c8b
--- /dev/null
+++ b/internal/gitalyssh/gitalyssh_test.go
@@ -0,0 +1,43 @@
+package gitalyssh
+
+import (
+	"context"
+	"encoding/base64"
+	"fmt"
+	"path/filepath"
+	"testing"
+
+	"github.com/golang/protobuf/jsonpb"
+	"github.com/stretchr/testify/require"
+	"gitlab.com/gitlab-org/gitaly/internal/config"
+	"gitlab.com/gitlab-org/gitaly/internal/testhelper"
+	"gitlab.com/gitlab-org/gitaly/proto/go/gitalypb"
+	"google.golang.org/grpc/metadata"
+)
+
+func TestUploadPackEnv(t *testing.T) {
+	testRepo, _, cleanupFn := testhelper.NewTestRepo(t)
+	defer cleanupFn()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	md := metadata.Pairs("gitaly-servers", base64.StdEncoding.EncodeToString([]byte(`{"default":{"address":"unix:///tmp/sock","token":"hunter1"}}`)))
+	ctx = metadata.NewIncomingContext(ctx, md)
+
+	req := gitalypb.SSHUploadPackRequest{
+		Repository: testRepo,
+	}
+
+	var pbMarshaler jsonpb.Marshaler
+	expectedPayload, err := pbMarshaler.MarshalToString(&req)
+	require.NoError(t, err)
+
+	env, err := UploadPackEnv(ctx, &req)
+
+	require.NoError(t, err)
+	require.Subset(t, env, []string{
+		fmt.Sprintf("GIT_SSH_COMMAND=%s upload-pack", filepath.Join(config.Config.BinDir, "gitaly-ssh")),
+		fmt.Sprintf("GITALY_PAYLOAD=%s", expectedPayload),
+	})
+}
diff --git a/internal/gitalyssh/testhelper_test.go b/internal/gitalyssh/testhelper_test.go
new file mode 100644
index 000000000..233792790
--- /dev/null
+++ b/internal/gitalyssh/testhelper_test.go
@@ -0,0 +1,13 @@
+package gitalyssh
+
+import (
+	"os"
+	"testing"
+
+	"gitlab.com/gitlab-org/gitaly/internal/testhelper"
+)
+
+func TestMain(m *testing.M) {
+	testhelper.Configure()
+	os.Exit(m.Run())
+}
diff --git a/internal/metadata/featureflag/feature_flags.go b/internal/metadata/featureflag/feature_flags.go
index af3beb7c8..f3db74067 100644
--- a/internal/metadata/featureflag/feature_flags.go
+++ b/internal/metadata/featureflag/feature_flags.go
@@ -8,6 +8,8 @@ const (
 	LinguistFileCountStats = "linguist_file_count_stats"
 	// HooksRPC will invoke update, pre receive, and post receive hooks by using RPCs
 	HooksRPC = "hooks_rpc"
+	// GoFetchInternalRemote enables a go implementation of FetchInternalRemote
+	GoFetchInternalRemote = "go_fetch_internal_remote"
 )
 
 const (
diff --git a/internal/service/remote/fetch_internal_remote.go b/internal/service/remote/fetch_internal_remote.go
index c8eaed507..47ba315d8 100644
--- a/internal/service/remote/fetch_internal_remote.go
+++ b/internal/service/remote/fetch_internal_remote.go
@@ -4,18 +4,63 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/grpc-ecosystem/go-grpc-middleware/logging/logrus/ctxlogrus"
+	"gitlab.com/gitlab-org/gitaly/internal/git"
+	"gitlab.com/gitlab-org/gitaly/internal/gitalyssh"
+	"gitlab.com/gitlab-org/gitaly/internal/helper"
+	"gitlab.com/gitlab-org/gitaly/internal/metadata/featureflag"
 	"gitlab.com/gitlab-org/gitaly/internal/rubyserver"
 	"gitlab.com/gitlab-org/gitaly/proto/go/gitalypb"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 )
 
+const (
+	gitalyInternalURL = "ssh://gitaly/internal.git"
+	mirrorRefSpec     = "+refs/*:refs/*"
+)
+
 // FetchInternalRemote fetches another Gitaly repository set as a remote
 func (s *server) FetchInternalRemote(ctx context.Context, req *gitalypb.FetchInternalRemoteRequest) (*gitalypb.FetchInternalRemoteResponse, error) {
 	if err := validateFetchInternalRemoteRequest(req); err != nil {
 		return nil, status.Errorf(codes.InvalidArgument, "FetchInternalRemote: %v", err)
 	}
 
+	if featureflag.IsDisabled(ctx, featureflag.GoFetchInternalRemote) {
+		return s.rubyFetchInternalRemote(ctx, req)
+	}
+
+	env, err := gitalyssh.UploadPackEnv(ctx, &gitalypb.SSHUploadPackRequest{Repository: req.RemoteRepository})
+	if err != nil {
+		return nil, err
+	}
+
+	repoPath, err := helper.GetRepoPath(req.Repository)
+	if err != nil {
+		return nil, err
+	}
+
+	cmd, err := git.SafeBareCmd(ctx, nil, nil, nil, env,
+		[]git.Option{git.ValueFlag{"--git-dir", repoPath}},
+		git.SubCmd{
+			Name:  "fetch",
+			Flags: []git.Option{git.Flag{"--prune"}},
+			Args:  []string{gitalyInternalURL, mirrorRefSpec},
+		},
+	)
+	if err != nil {
+		return nil, err
+	}
+	if err := cmd.Wait(); err != nil {
+		// Design quirk: if the fetch fails, this RPC returns Result: false, but no error.
+		ctxlogrus.Extract(ctx).WithError(err).Warn("git fetch failed")
+		return &gitalypb.FetchInternalRemoteResponse{Result: false}, nil
+	}
+
+	return &gitalypb.FetchInternalRemoteResponse{Result: true}, nil
+}
+
+func (s *server) rubyFetchInternalRemote(ctx context.Context, req *gitalypb.FetchInternalRemoteRequest) (*gitalypb.FetchInternalRemoteResponse, error) {
 	client, err := s.ruby.RemoteServiceClient(ctx)
 	if err != nil {
 		return nil, err
diff --git a/internal/service/remote/fetch_internal_remote_test.go b/internal/service/remote/fetch_internal_remote_test.go
index 09a89b06b..ea7ea78b0 100644
--- a/internal/service/remote/fetch_internal_remote_test.go
+++ b/internal/service/remote/fetch_internal_remote_test.go
@@ -7,6 +7,7 @@ import (
 
 	"github.com/stretchr/testify/require"
 	"gitlab.com/gitlab-org/gitaly/internal/config"
+	"gitlab.com/gitlab-org/gitaly/internal/metadata/featureflag"
 	serverPkg "gitlab.com/gitlab-org/gitaly/internal/server"
 	"gitlab.com/gitlab-org/gitaly/internal/service/remote"
 	"gitlab.com/gitlab-org/gitaly/internal/testhelper"
@@ -26,27 +27,45 @@ func TestSuccessfulFetchInternalRemote(t *testing.T) {
 	remoteRepo, remoteRepoPath, remoteCleanupFn := testhelper.NewTestRepo(t)
 	defer remoteCleanupFn()
 
-	repo, repoPath, cleanupFn := testhelper.InitBareRepo(t)
-	defer cleanupFn()
-
-	ctxOuter, cancel := testhelper.Context()
-	defer cancel()
-
-	md := testhelper.GitalyServersMetadata(t, serverSocketPath)
-	ctx := metadata.NewOutgoingContext(ctxOuter, md)
-
-	request := &gitalypb.FetchInternalRemoteRequest{
-		Repository:       repo,
-		RemoteRepository: remoteRepo,
+	for _, tc := range []struct {
+		CaseName     string
+		FeatureFlags []string
+	}{
+		{
+			CaseName: "ruby",
+		},
+		{
+			CaseName:     "go",
+			FeatureFlags: []string{featureflag.GoFetchInternalRemote},
+		},
+	} {
+		t.Run(tc.CaseName, func(t *testing.T) {
+			repo, repoPath, cleanupFn := testhelper.InitBareRepo(t)
+			defer cleanupFn()
+
+			ctx, cancel := testhelper.Context()
+			defer cancel()
+
+			md := testhelper.GitalyServersMetadata(t, serverSocketPath)
+			ctx = metadata.NewOutgoingContext(ctx, md)
+			for _, feature := range tc.FeatureFlags {
+				ctx = featureflag.OutgoingCtxWithFeatureFlag(ctx, feature)
+			}
+
+			request := &gitalypb.FetchInternalRemoteRequest{
+				Repository:       repo,
+				RemoteRepository: remoteRepo,
+			}
+
+			c, err := client.FetchInternalRemote(ctx, request)
+			require.NoError(t, err)
+			require.True(t, c.GetResult())
+
+			remoteRefs := testhelper.GetRepositoryRefs(t, remoteRepoPath)
+			refs := testhelper.GetRepositoryRefs(t, repoPath)
+			require.Equal(t, remoteRefs, refs)
+		})
 	}
-
-	c, err := client.FetchInternalRemote(ctx, request)
-	require.NoError(t, err)
-	require.True(t, c.GetResult())
-
-	remoteRefs := testhelper.GetRepositoryRefs(t, remoteRepoPath)
-	refs := testhelper.GetRepositoryRefs(t, repoPath)
-	require.Equal(t, remoteRefs, refs)
 }
 
 func TestFailedFetchInternalRemote(t *testing.T) {
@@ -56,26 +75,44 @@ func TestFailedFetchInternalRemote(t *testing.T) {
 	client, conn := remote.NewRemoteClient(t, serverSocketPath)
 	defer conn.Close()
 
-	repo, _, cleanupFn := testhelper.InitBareRepo(t)
-	defer cleanupFn()
-
-	ctxOuter, cancel := testhelper.Context()
-	defer cancel()
-
-	md := testhelper.GitalyServersMetadata(t, serverSocketPath)
-	ctx := metadata.NewOutgoingContext(ctxOuter, md)
-
-	// Non-existing remote repo
-	remoteRepo := &gitalypb.Repository{StorageName: "default", RelativePath: "fake.git"}
-
-	request := &gitalypb.FetchInternalRemoteRequest{
-		Repository:       repo,
-		RemoteRepository: remoteRepo,
+	for _, tc := range []struct {
+		CaseName     string
+		FeatureFlags []string
+	}{
+		{
+			CaseName: "ruby",
+		},
+		{
+			CaseName:     "go",
+			FeatureFlags: []string{featureflag.GoFetchInternalRemote},
+		},
+	} {
+		t.Run(tc.CaseName, func(t *testing.T) {
+			repo, _, cleanupFn := testhelper.InitBareRepo(t)
+			defer cleanupFn()
+
+			ctx, cancel := testhelper.Context()
+			defer cancel()
+
+			md := testhelper.GitalyServersMetadata(t, serverSocketPath)
+			ctx = metadata.NewOutgoingContext(ctx, md)
+			for _, feature := range tc.FeatureFlags {
+				ctx = featureflag.OutgoingCtxWithFeatureFlag(ctx, feature)
+			}
+
+			// Non-existing remote repo
+			remoteRepo := &gitalypb.Repository{StorageName: "default", RelativePath: "fake.git"}
+
+			request := &gitalypb.FetchInternalRemoteRequest{
+				Repository:       repo,
+				RemoteRepository: remoteRepo,
+			}
+
+			c, err := client.FetchInternalRemote(ctx, request)
+			require.NoError(t, err, "FetchInternalRemote is not supposed to return an error when 'git fetch' fails")
+			require.False(t, c.GetResult())
+		})
 	}
-
-	c, err := client.FetchInternalRemote(ctx, request)
-	require.NoError(t, err)
-	require.False(t, c.GetResult())
 }
 
 func TestFailedFetchInternalRemoteDueToValidations(t *testing.T) {
diff --git a/internal/service/repository/fork.go b/internal/service/repository/fork.go
index 765e261ba..c379e835a 100644
--- a/internal/service/repository/fork.go
+++ b/internal/service/repository/fork.go
@@ -4,23 +4,17 @@ import (
 	"context"
 	"fmt"
 	"os"
-	"path"
 
-	"github.com/golang/protobuf/jsonpb"
-	"gitlab.com/gitlab-org/gitaly/internal/config"
 	"gitlab.com/gitlab-org/gitaly/internal/git"
+	"gitlab.com/gitlab-org/gitaly/internal/gitalyssh"
 	"gitlab.com/gitlab-org/gitaly/internal/helper"
-	gitaly_x509 "gitlab.com/gitlab-org/gitaly/internal/x509"
 	"gitlab.com/gitlab-org/gitaly/proto/go/gitalypb"
-	"gitlab.com/gitlab-org/labkit/tracing"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 )
 
 const gitalyInternalURL = "ssh://gitaly/internal.git"
 
-var envInjector = tracing.NewEnvInjector()
-
 func (s *server) CreateFork(ctx context.Context, req *gitalypb.CreateForkRequest) (*gitalypb.CreateForkResponse, error) {
 	targetRepository := req.Repository
 	sourceRepository := req.SourceRepository
@@ -45,43 +39,10 @@ func (s *server) CreateFork(ctx context.Context, req *gitalypb.CreateForkRequest
 		return nil, status.Errorf(codes.Internal, "CreateFork: create dest dir: %v", err)
 	}
 
-	gitalyServersInfo, err := helper.ExtractGitalyServers(ctx)
+	env, err := gitalyssh.UploadPackEnv(ctx, &gitalypb.SSHUploadPackRequest{Repository: sourceRepository})
 	if err != nil {
-		return nil, status.Errorf(codes.Internal, "CreateFork: extracting Gitaly servers: %v", err)
-	}
-
-	sourceRepositoryStorageInfo, ok := gitalyServersInfo[sourceRepository.StorageName]
-	if !ok {
-		return nil, status.Errorf(codes.InvalidArgument, "CreateFork: no storage info for %s", sourceRepository.StorageName)
-	}
-
-	sourceRepositoryGitalyAddress := sourceRepositoryStorageInfo["address"]
-	if sourceRepositoryGitalyAddress == "" {
-		return nil, status.Errorf(codes.InvalidArgument, "CreateFork: empty gitaly address")
-	}
-
-	sourceRepositoryGitalyToken := sourceRepositoryStorageInfo["token"]
-
-	cloneReq := &gitalypb.SSHUploadPackRequest{Repository: sourceRepository}
-	pbMarshaler := &jsonpb.Marshaler{}
-	payload, err := pbMarshaler.MarshalToString(cloneReq)
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "CreateFork: marshalling payload failed: %v", err)
-	}
-
-	gitalySSHPath := path.Join(config.Config.BinDir, "gitaly-ssh")
-
-	env := []string{
-		fmt.Sprintf("GITALY_ADDRESS=%s", sourceRepositoryGitalyAddress),
-		fmt.Sprintf("GITALY_PAYLOAD=%s", payload),
-		fmt.Sprintf("GITALY_TOKEN=%s", sourceRepositoryGitalyToken),
-		fmt.Sprintf("GIT_SSH_COMMAND=%s upload-pack", gitalySSHPath),
-		// Pass through the SSL_CERT_* variables that indicate which
-		// system certs to trust
-		fmt.Sprintf("%s=%s", gitaly_x509.SSLCertDir, os.Getenv(gitaly_x509.SSLCertDir)),
-		fmt.Sprintf("%s=%s", gitaly_x509.SSLCertFile, os.Getenv(gitaly_x509.SSLCertFile)),
+		return nil, err
 	}
-	env = envInjector(ctx, env)
 
 	cmd, err := git.SafeBareCmd(ctx, nil, nil, nil, env, nil, git.SubCmd{
 		Name:  "clone",