Merge branch 'pks-fetch-internal-remote-read-logs' into 'master'

Improve hooks error handling and fix flaky FetchInternalRemote

Closes #3866

See merge request gitlab-org/gitaly!4006

3 files changed, 128 insertions, 102 deletions

diff --git a/cmd/gitaly-hooks/hooks.go b/cmd/gitaly-hooks/hooks.go
index 58462ed02..14aae4d0a 100644
--- a/cmd/gitaly-hooks/hooks.go
+++ b/cmd/gitaly-hooks/hooks.go
@@ -5,7 +5,6 @@ import (
 	"errors"
 	"fmt"
 	"io"
-	"log"
 	"net"
 	"os"
 
@@ -28,52 +27,63 @@ import (
 	"google.golang.org/grpc"
 )
 
+type hookError struct {
+	returnCode int
+	err        error
+}
+
+func (e hookError) Error() string {
+	return fmt.Sprintf("hook returned error code %d", e.returnCode)
+}
+
 type hookCommand struct {
-	exec     func(context.Context, git.HooksPayload, gitalypb.HookServiceClient, []string) (int, error)
+	exec     func(context.Context, git.HooksPayload, gitalypb.HookServiceClient, []string) error
 	hookType git.Hook
 }
 
-var (
-	hooksBySubcommand = map[string]hookCommand{
-		"update": {
-			exec:     updateHook,
-			hookType: git.UpdateHook,
-		},
-		"pre-receive": {
-			exec:     preReceiveHook,
-			hookType: git.PreReceiveHook,
-		},
-		"post-receive": {
-			exec:     postReceiveHook,
-			hookType: git.PostReceiveHook,
-		},
-		"reference-transaction": {
-			exec:     referenceTransactionHook,
-			hookType: git.ReferenceTransactionHook,
-		},
-		"git": {
-			exec:     packObjectsHook,
-			hookType: git.PackObjectsHook,
-		},
-	}
-
-	logger *gitalylog.HookLogger
-)
+var hooksBySubcommand = map[string]hookCommand{
+	"update": {
+		exec:     updateHook,
+		hookType: git.UpdateHook,
+	},
+	"pre-receive": {
+		exec:     preReceiveHook,
+		hookType: git.PreReceiveHook,
+	},
+	"post-receive": {
+		exec:     postReceiveHook,
+		hookType: git.PostReceiveHook,
+	},
+	"reference-transaction": {
+		exec:     referenceTransactionHook,
+		hookType: git.ReferenceTransactionHook,
+	},
+	"git": {
+		exec:     packObjectsHook,
+		hookType: git.PackObjectsHook,
+	},
+}
 
 func main() {
-	logger = gitalylog.NewHookLogger()
+	logger := gitalylog.NewHookLogger()
+
+	if err := run(os.Args); err != nil {
+		var hookError hookError
+		if errors.As(err, &hookError) {
+			if hookError.err != nil {
+				logger.Fatalf("%s", err)
+			}
+			os.Exit(hookError.returnCode)
+		}
 
-	returnCode, err := run(os.Args, logger)
-	if err != nil {
 		logger.Fatalf("%s", err)
+		os.Exit(1)
 	}
-
-	os.Exit(returnCode)
 }
 
-func run(args []string, logger *gitalylog.HookLogger) (int, error) {
+func run(args []string) error {
 	if len(args) < 2 {
-		return 0, fmt.Errorf("requires hook name. args: %v", args)
+		return fmt.Errorf("requires hook name. args: %v", args)
 	}
 
 	subCmd := args[1]
@@ -81,16 +91,18 @@ func run(args []string, logger *gitalylog.HookLogger) (int, error) {
 	if subCmd == "check" {
 		logrus.SetLevel(logrus.ErrorLevel)
 		if len(args) != 3 {
-			log.Fatal(errors.New("no configuration file path provided invoke with: gitaly-hooks check <config_path>"))
+			fmt.Fprint(os.Stderr, "no configuration file path provided invoke with: gitaly-hooks check <config_path>")
+			os.Exit(1)
 		}
 
 		configPath := args[2]
 		fmt.Print("Checking GitLab API access: ")
 
-		info, err := check(logger.Logger(), configPath)
+		info, err := check(configPath)
 		if err != nil {
 			fmt.Print("FAIL\n")
-			log.Fatal(err)
+			fmt.Fprint(os.Stderr, err)
+			os.Exit(1)
 		}
 
 		fmt.Print("OK\n")
@@ -100,7 +112,7 @@ func run(args []string, logger *gitalylog.HookLogger) (int, error) {
 		fmt.Printf("Redis reachable for GitLab: %t\n", info.RedisReachable)
 		fmt.Println("OK")
 
-		return 0, nil
+		return nil
 	}
 
 	ctx, cancel := context.WithCancel(context.Background())
@@ -114,35 +126,34 @@ func run(args []string, logger *gitalylog.HookLogger) (int, error) {
 
 	payload, err := git.HooksPayloadFromEnv(os.Environ())
 	if err != nil {
-		return 0, fmt.Errorf("error when getting hooks payload: %v", err)
+		return fmt.Errorf("error when getting hooks payload: %v", err)
 	}
 
 	hookCommand, ok := hooksBySubcommand[subCmd]
 	if !ok {
-		return 0, fmt.Errorf("subcommand name invalid: %q", subCmd)
+		return fmt.Errorf("subcommand name invalid: %q", subCmd)
 	}
 
 	// If the hook wasn't requested, then we simply skip executing any
 	// logic.
 	if !payload.IsHookRequested(hookCommand.hookType) {
-		return 0, nil
+		return nil
 	}
 
 	conn, err := dialGitaly(payload)
 	if err != nil {
-		return 0, fmt.Errorf("error when connecting to gitaly: %v", err)
+		return fmt.Errorf("error when connecting to gitaly: %v", err)
 	}
 	defer conn.Close()
 
 	hookClient := gitalypb.NewHookServiceClient(conn)
 
 	ctx = featureflag.OutgoingWithRaw(ctx, payload.FeatureFlags)
-	returnCode, err := hookCommand.exec(ctx, payload, hookClient, args)
-	if err != nil {
-		return 0, err
+	if err := hookCommand.exec(ctx, payload, hookClient, args); err != nil {
+		return err
 	}
 
-	return returnCode, nil
+	return nil
 }
 
 func noopSender(c chan error) {}
@@ -179,12 +190,16 @@ func gitPushOptions() []string {
 func sendFunc(reqWriter io.Writer, stream grpc.ClientStream, stdin io.Reader) func(errC chan error) {
 	return func(errC chan error) {
 		_, errSend := io.Copy(reqWriter, stdin)
-		stream.CloseSend()
-		errC <- errSend
+		errClose := stream.CloseSend()
+		if errSend != nil {
+			errC <- errSend
+		} else {
+			errC <- errClose
+		}
 	}
 }
 
-func check(logger logrus.FieldLogger, configPath string) (*gitlab.CheckInfo, error) {
+func check(configPath string) (*gitlab.CheckInfo, error) {
 	cfgFile, err := os.Open(configPath)
 	if err != nil {
 		return nil, fmt.Errorf("failed to open config file: %w", err)
@@ -196,7 +211,7 @@ func check(logger logrus.FieldLogger, configPath string) (*gitlab.CheckInfo, err
 		return nil, err
 	}
 
-	gitlabAPI, err := gitlab.NewHTTPClient(logger, cfg.Gitlab, cfg.TLS, prometheus.Config{})
+	gitlabAPI, err := gitlab.NewHTTPClient(logrus.New(), cfg.Gitlab, cfg.TLS, prometheus.Config{})
 	if err != nil {
 		return nil, err
 	}
@@ -204,10 +219,10 @@ func check(logger logrus.FieldLogger, configPath string) (*gitlab.CheckInfo, err
 	return hook.NewManager(config.NewLocator(cfg), nil, gitlabAPI, cfg).Check(context.TODO())
 }
 
-func updateHook(ctx context.Context, payload git.HooksPayload, hookClient gitalypb.HookServiceClient, args []string) (int, error) {
+func updateHook(ctx context.Context, payload git.HooksPayload, hookClient gitalypb.HookServiceClient, args []string) error {
 	args = args[2:]
 	if len(args) != 3 {
-		return 1, errors.New("update hook expects exactly three arguments")
+		return fmt.Errorf("update hook expects exactly three arguments, got %q", args)
 	}
 	ref, oldValue, newValue := args[0], args[1], args[2]
 
@@ -221,23 +236,24 @@ func updateHook(ctx context.Context, payload git.HooksPayload, hookClient gitaly
 
 	updateHookStream, err := hookClient.UpdateHook(ctx, req)
 	if err != nil {
-		return 1, fmt.Errorf("error when starting command for update hook: %v", err)
+		return fmt.Errorf("error when starting command for update hook: %v", err)
 	}
 
-	var returnCode int32
-	if returnCode, err = stream.Handler(func() (stream.StdoutStderrResponse, error) {
+	if returnCode, err := stream.Handler(func() (stream.StdoutStderrResponse, error) {
 		return updateHookStream.Recv()
 	}, noopSender, os.Stdout, os.Stderr); err != nil {
-		return 1, fmt.Errorf("error when receiving data for update hook: %v", err)
+		return fmt.Errorf("error when receiving data for update hook: %v", err)
+	} else if returnCode != 0 {
+		return hookError{returnCode: int(returnCode)}
 	}
 
-	return int(returnCode), nil
+	return nil
 }
 
-func preReceiveHook(ctx context.Context, payload git.HooksPayload, hookClient gitalypb.HookServiceClient, args []string) (int, error) {
+func preReceiveHook(ctx context.Context, payload git.HooksPayload, hookClient gitalypb.HookServiceClient, args []string) error {
 	preReceiveHookStream, err := hookClient.PreReceiveHook(ctx)
 	if err != nil {
-		return 1, fmt.Errorf("error when getting preReceiveHookStream client for: %v", err)
+		return fmt.Errorf("error when getting preReceiveHookStream client for: %v", err)
 	}
 
 	if err := preReceiveHookStream.Send(&gitalypb.PreReceiveHookRequest{
@@ -245,27 +261,28 @@ func preReceiveHook(ctx context.Context, payload git.HooksPayload, hookClient gi
 		EnvironmentVariables: os.Environ(),
 		GitPushOptions:       gitPushOptions(),
 	}); err != nil {
-		return 1, fmt.Errorf("error when sending request for pre-receive hook: %v", err)
+		return fmt.Errorf("error when sending request for pre-receive hook: %v", err)
 	}
 
 	f := sendFunc(streamio.NewWriter(func(p []byte) error {
 		return preReceiveHookStream.Send(&gitalypb.PreReceiveHookRequest{Stdin: p})
 	}), preReceiveHookStream, os.Stdin)
 
-	var returnCode int32
-	if returnCode, err = stream.Handler(func() (stream.StdoutStderrResponse, error) {
+	if returnCode, err := stream.Handler(func() (stream.StdoutStderrResponse, error) {
 		return preReceiveHookStream.Recv()
 	}, f, os.Stdout, os.Stderr); err != nil {
-		return 1, fmt.Errorf("error when receiving data for pre-receive hook: %v", err)
+		return fmt.Errorf("error when receiving data for pre-receive hook: %v", err)
+	} else if returnCode != 0 {
+		return hookError{returnCode: int(returnCode)}
 	}
 
-	return int(returnCode), nil
+	return nil
 }
 
-func postReceiveHook(ctx context.Context, payload git.HooksPayload, hookClient gitalypb.HookServiceClient, args []string) (int, error) {
+func postReceiveHook(ctx context.Context, payload git.HooksPayload, hookClient gitalypb.HookServiceClient, args []string) error {
 	postReceiveHookStream, err := hookClient.PostReceiveHook(ctx)
 	if err != nil {
-		return 1, fmt.Errorf("error when getting stream client for post-receive hook: %v", err)
+		return fmt.Errorf("error when getting stream client for post-receive hook: %v", err)
 	}
 
 	if err := postReceiveHookStream.Send(&gitalypb.PostReceiveHookRequest{
@@ -273,26 +290,27 @@ func postReceiveHook(ctx context.Context, payload git.HooksPayload, hookClient g
 		EnvironmentVariables: os.Environ(),
 		GitPushOptions:       gitPushOptions(),
 	}); err != nil {
-		return 1, fmt.Errorf("error when sending request for post-receive hook: %v", err)
+		return fmt.Errorf("error when sending request for post-receive hook: %v", err)
 	}
 
 	f := sendFunc(streamio.NewWriter(func(p []byte) error {
 		return postReceiveHookStream.Send(&gitalypb.PostReceiveHookRequest{Stdin: p})
 	}), postReceiveHookStream, os.Stdin)
 
-	var returnCode int32
-	if returnCode, err = stream.Handler(func() (stream.StdoutStderrResponse, error) {
+	if returnCode, err := stream.Handler(func() (stream.StdoutStderrResponse, error) {
 		return postReceiveHookStream.Recv()
 	}, f, os.Stdout, os.Stderr); err != nil {
-		return 1, fmt.Errorf("error when receiving data for post-receive hook: %v", err)
+		return fmt.Errorf("error when receiving data for post-receive hook: %v", err)
+	} else if returnCode != 0 {
+		return hookError{returnCode: int(returnCode)}
 	}
 
-	return int(returnCode), nil
+	return nil
 }
 
-func referenceTransactionHook(ctx context.Context, payload git.HooksPayload, hookClient gitalypb.HookServiceClient, args []string) (int, error) {
+func referenceTransactionHook(ctx context.Context, payload git.HooksPayload, hookClient gitalypb.HookServiceClient, args []string) error {
 	if len(args) != 3 {
-		return 1, errors.New("reference-transaction hook is missing required arguments")
+		return fmt.Errorf("reference-transaction hook is missing required arguments, got %q", args)
 	}
 
 	var state gitalypb.ReferenceTransactionHookRequest_State
@@ -304,12 +322,12 @@ func referenceTransactionHook(ctx context.Context, payload git.HooksPayload, hoo
 	case "aborted":
 		state = gitalypb.ReferenceTransactionHookRequest_ABORTED
 	default:
-		return 1, fmt.Errorf("reference-transaction hook has invalid state: %q", args[2])
+		return fmt.Errorf("reference-transaction hook has invalid state: %q", args[2])
 	}
 
 	referenceTransactionHookStream, err := hookClient.ReferenceTransactionHook(ctx)
 	if err != nil {
-		return 1, fmt.Errorf("error when getting referenceTransactionHookStream client: %v", err)
+		return fmt.Errorf("error when getting referenceTransactionHookStream client: %v", err)
 	}
 
 	if err := referenceTransactionHookStream.Send(&gitalypb.ReferenceTransactionHookRequest{
@@ -317,33 +335,30 @@ func referenceTransactionHook(ctx context.Context, payload git.HooksPayload, hoo
 		EnvironmentVariables: os.Environ(),
 		State:                state,
 	}); err != nil {
-		return 1, fmt.Errorf("error when sending request for reference-transaction hook: %v", err)
+		return fmt.Errorf("error when sending request for reference-transaction hook: %v", err)
 	}
 
 	f := sendFunc(streamio.NewWriter(func(p []byte) error {
 		return referenceTransactionHookStream.Send(&gitalypb.ReferenceTransactionHookRequest{Stdin: p})
 	}), referenceTransactionHookStream, os.Stdin)
 
-	var returnCode int32
-	if returnCode, err = stream.Handler(func() (stream.StdoutStderrResponse, error) {
+	if returnCode, err := stream.Handler(func() (stream.StdoutStderrResponse, error) {
 		return referenceTransactionHookStream.Recv()
 	}, f, os.Stdout, os.Stderr); err != nil {
-		return 1, fmt.Errorf("error when receiving data for reference-transaction hook: %v", err)
+		return fmt.Errorf("error when receiving data for reference-transaction hook: %v", err)
+	} else if returnCode != 0 {
+		return hookError{returnCode: int(returnCode)}
 	}
 
-	return int(returnCode), nil
+	return nil
 }
 
-func packObjectsHook(ctx context.Context, payload git.HooksPayload, hookClient gitalypb.HookServiceClient, args []string) (int, error) {
+func packObjectsHook(ctx context.Context, payload git.HooksPayload, hookClient gitalypb.HookServiceClient, args []string) error {
 	if err := handlePackObjectsWithSidechannel(ctx, hookClient, payload.Repo, args[2:]); err != nil {
-		logger.Logger().WithFields(logrus.Fields{
-			"args": args,
-			"rpc":  "PackObjectsHookWithSidechannel",
-		}).WithError(err).Error("RPC failed")
-		return 1, nil
+		return hookError{returnCode: 1, err: fmt.Errorf("RPC failed: %w", err)}
 	}
 
-	return 0, nil
+	return nil
 }
 
 func handlePackObjectsWithSidechannel(ctx context.Context, hookClient gitalypb.HookServiceClient, repo *gitalypb.Repository, args []string) error {
diff --git a/cmd/gitaly-hooks/hooks_test.go b/cmd/gitaly-hooks/hooks_test.go
index 9772a949d..06ea2e041 100644
--- a/cmd/gitaly-hooks/hooks_test.go
+++ b/cmd/gitaly-hooks/hooks_test.go
@@ -136,8 +136,7 @@ func testHooksPrePostReceive(t *testing.T, cfg config.Cfg, repo *gitalypb.Reposi
 	gitlabUser, gitlabPassword := "gitlab_user-1234", "gitlabsecret9887"
 	httpProxy, httpsProxy, noProxy := "http://test.example.com:8080", "https://test.example.com:8080", "*"
 
-	defer func(old *gitalylog.HookLogger) { logger = old }(logger)
-	logger = gitalylog.NewHookLogger()
+	logger, _ := test.NewNullLogger()
 
 	c := gitlab.TestServerOptions{
 		User:                        gitlabUser,
@@ -161,7 +160,7 @@ func testHooksPrePostReceive(t *testing.T, cfg config.Cfg, repo *gitalypb.Reposi
 	cfg.Gitlab.HTTPSettings.User = gitlabUser
 	cfg.Gitlab.HTTPSettings.Password = gitlabPassword
 
-	gitlabClient, err := gitlab.NewHTTPClient(logger.Logger(), cfg.Gitlab, cfg.TLS, prometheus.Config{})
+	gitlabClient, err := gitlab.NewHTTPClient(logger, cfg.Gitlab, cfg.TLS, prometheus.Config{})
 	require.NoError(t, err)
 
 	runHookServiceWithGitlabClient(t, cfg, gitlabClient)
@@ -324,8 +323,7 @@ func TestHooksPostReceiveFailed(t *testing.T) {
 	glProtocol := "ssh"
 	changes := "oldhead newhead"
 
-	defer func(old *gitalylog.HookLogger) { logger = old }(logger)
-	logger = gitalylog.NewHookLogger()
+	logger, _ := test.NewNullLogger()
 
 	cfg, repo, repoPath := testcfg.BuildWithRepo(t, testcfg.WithBase(config.Cfg{Auth: auth.Config{Token: "abc123"}}))
 	testhelper.BuildGitalyHooks(t, cfg)
@@ -350,7 +348,7 @@ func TestHooksPostReceiveFailed(t *testing.T) {
 	cfg.Gitlab.URL = serverURL
 	cfg.Gitlab.SecretFile = gitlab.WriteShellSecretFile(t, cfg.GitlabShell.Dir, secretToken)
 
-	gitlabClient, err := gitlab.NewHTTPClient(logger.Logger(), cfg.Gitlab, cfg.TLS, prometheus.Config{})
+	gitlabClient, err := gitlab.NewHTTPClient(logger, cfg.Gitlab, cfg.TLS, prometheus.Config{})
 	require.NoError(t, err)
 
 	runHookServiceWithGitlabClient(t, cfg, gitlabClient)
@@ -437,8 +435,7 @@ func TestHooksNotAllowed(t *testing.T) {
 	glProtocol := "ssh"
 	changes := "oldhead newhead"
 
-	defer func(old *gitalylog.HookLogger) { logger = old }(logger)
-	logger = gitalylog.NewHookLogger()
+	logger, _ := test.NewNullLogger()
 
 	cfg, repo, repoPath := testcfg.BuildWithRepo(t, testcfg.WithBase(config.Cfg{Auth: auth.Config{Token: "abc123"}}))
 	testhelper.BuildGitalyHooks(t, cfg)
@@ -462,7 +459,7 @@ func TestHooksNotAllowed(t *testing.T) {
 
 	customHookOutputPath := gittest.WriteEnvToCustomHook(t, repoPath, "post-receive")
 
-	gitlabClient, err := gitlab.NewHTTPClient(logger.Logger(), cfg.Gitlab, cfg.TLS, prometheus.Config{})
+	gitlabClient, err := gitlab.NewHTTPClient(logger, cfg.Gitlab, cfg.TLS, prometheus.Config{})
 	require.NoError(t, err)
 
 	runHookServiceWithGitlabClient(t, cfg, gitlabClient)
diff --git a/internal/gitaly/service/remote/fetch_internal_remote_test.go b/internal/gitaly/service/remote/fetch_internal_remote_test.go
index 0a7dce19a..e9839647d 100644
--- a/internal/gitaly/service/remote/fetch_internal_remote_test.go
+++ b/internal/gitaly/service/remote/fetch_internal_remote_test.go
@@ -9,6 +9,7 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"gitlab.com/gitlab-org/gitaly/v14/client"
 	"gitlab.com/gitlab-org/gitaly/v14/internal/git/gittest"
@@ -113,10 +114,17 @@ func TestFetchInternalRemote_successful(t *testing.T) {
 			deps.GetGitCmdFactory(),
 			deps.GetPackObjectsCache(),
 		))
-	}, testserver.WithHookManager(gitalyhook.NewMockManager(t, nil, nil, nil, func(*testing.T, context.Context, gitalyhook.ReferenceTransactionState, []string, io.Reader) error {
-		referenceTransactionHookCalled++
-		return nil
-	})), testserver.WithDisablePraefect())
+	}, testserver.WithHookManager(gitalyhook.NewMockManager(t, nil, nil, nil,
+		func(t *testing.T, _ context.Context, _ gitalyhook.ReferenceTransactionState, _ []string, stdin io.Reader) error {
+			// We need to discard stdin or otherwise the sending Goroutine may return an
+			// EOF error and cause the test to fail.
+			_, err := io.Copy(io.Discard, stdin)
+			require.NoError(t, err)
+
+			referenceTransactionHookCalled++
+			return nil
+		}),
+	), testserver.WithDisablePraefect())
 
 	ctx, err := storage.InjectGitalyServers(ctx, remoteRepo.GetStorageName(), remoteAddr, remoteCfg.Auth.Token)
 	require.NoError(t, err)
@@ -127,7 +135,13 @@ func TestFetchInternalRemote_successful(t *testing.T) {
 	connsPool := client.NewPool()
 	defer connsPool.Close()
 
-	require.NoError(t, FetchInternalRemote(ctx, localCfg, connsPool, localRepo, remoteRepo))
+	// Use the `assert` package such that we can get information about why hooks have failed via
+	// the hook logs in case it did fail unexpectedly.
+	assert.NoError(t, FetchInternalRemote(ctx, localCfg, connsPool, localRepo, remoteRepo))
+
+	hookLogs := filepath.Join(localCfg.Logging.Dir, "gitaly_hooks.log")
+	require.FileExists(t, hookLogs)
+	require.Equal(t, "", string(testhelper.MustReadFile(t, hookLogs)))
 
 	require.Equal(t,
 		string(gittest.Exec(t, remoteCfg, "-C", remoteRepoPath, "show-ref", "--head")),