Merge branch 'ps-get-tag-messages-go' into 'master'

Port GetTagMessages from Gitaly-Ruby to Gitaly

Closes #2123

See merge request gitlab-org/gitaly!1618

6 files changed, 154 insertions, 30 deletions

diff --git a/internal/git/log/tag.go b/internal/git/log/tag.go
index 52c340673..9750aca57 100644
--- a/internal/git/log/tag.go
+++ b/internal/git/log/tag.go
@@ -19,21 +19,23 @@ const (
 )
 
 // GetTagCatfile looks up a commit by tagID using an existing *catfile.Batch instance.
+// When 'trim' is 'true', the tag message will be trimmed to fit in a gRPC message.
+// When 'trimRightNewLine' is 'true', the tag message will be trimmed to remove all '\n' characters from right.
 // note: we pass in the tagName because the tag name from refs/tags may be different
 // than the name found in the actual tag object. We want to use the tagName found in refs/tags
-func GetTagCatfile(c *catfile.Batch, tagID, tagName string) (*gitalypb.Tag, error) {
+func GetTagCatfile(c *catfile.Batch, tagID, tagName string, trimLen, trimRightNewLine bool) (*gitalypb.Tag, error) {
 	r, err := c.Tag(tagID)
 	if err != nil {
 		return nil, err
 	}
 
-	header, body, err := splitRawTag(r)
+	header, body, err := splitRawTag(r, trimRightNewLine)
 	if err != nil {
 		return nil, err
 	}
 
 	// the tagID is the oid of the tag object
-	tag, err := buildAnnotatedTag(c, tagID, tagName, header, body)
+	tag, err := buildAnnotatedTag(c, tagID, tagName, header, body, trimLen, trimRightNewLine)
 	if err != nil {
 		return nil, err
 	}
@@ -48,7 +50,7 @@ type tagHeader struct {
 	tagger  string
 }
 
-func splitRawTag(r io.Reader) (*tagHeader, []byte, error) {
+func splitRawTag(r io.Reader, trimRightNewLine bool) (*tagHeader, []byte, error) {
 	raw, err := ioutil.ReadAll(r)
 	if err != nil {
 		return nil, nil, err
@@ -57,10 +59,13 @@ func splitRawTag(r io.Reader) (*tagHeader, []byte, error) {
 	var body []byte
 	split := bytes.SplitN(raw, []byte("\n\n"), 2)
 	if len(split) == 2 {
-		// Remove trailing newline, if any, to preserve existing behavior the old GitLab tag finding code.
-		// See https://gitlab.com/gitlab-org/gitaly/blob/5e94dc966ac1900c11794b107a77496552591f9b/ruby/lib/gitlab/git/repository.rb#L211.
-		// Maybe this belongs in the FindAllTags handler, or even on the gitlab-ce client side, instead of here?
-		body = bytes.TrimRight(split[1], "\n")
+		body = split[1]
+		if trimRightNewLine {
+			// Remove trailing newline, if any, to preserve existing behavior the old GitLab tag finding code.
+			// See https://gitlab.com/gitlab-org/gitaly/blob/5e94dc966ac1900c11794b107a77496552591f9b/ruby/lib/gitlab/git/repository.rb#L211.
+			// Maybe this belongs in the FindAllTags handler, or even on the gitlab-ce client side, instead of here?
+			body = bytes.TrimRight(body, "\n")
+		}
 	}
 
 	var header tagHeader
@@ -87,7 +92,7 @@ func splitRawTag(r io.Reader) (*tagHeader, []byte, error) {
 	return &header, body, nil
 }
 
-func buildAnnotatedTag(b *catfile.Batch, tagID, name string, header *tagHeader, body []byte) (*gitalypb.Tag, error) {
+func buildAnnotatedTag(b *catfile.Batch, tagID, name string, header *tagHeader, body []byte, trimLen, trimRightNewLine bool) (*gitalypb.Tag, error) {
 	tag := &gitalypb.Tag{
 		Id:          tagID,
 		Name:        []byte(name),
@@ -95,7 +100,7 @@ func buildAnnotatedTag(b *catfile.Batch, tagID, name string, header *tagHeader,
 		Message:     body,
 	}
 
-	if max := helper.MaxCommitOrTagMessageSize; len(body) > max {
+	if max := helper.MaxCommitOrTagMessageSize; trimLen && len(body) > max {
 		tag.Message = tag.Message[:max]
 	}
 
@@ -138,7 +143,7 @@ func dereferenceTag(b *catfile.Batch, Oid string) (*gitalypb.GitCommit, error) {
 				return nil, err
 			}
 
-			header, _, err := splitRawTag(r)
+			header, _, err := splitRawTag(r, true)
 			if err != nil {
 				return nil, err
 			}
diff --git a/internal/git/log/tag_test.go b/internal/git/log/tag_test.go
index daa8540b4..92750d693 100644
--- a/internal/git/log/tag_test.go
+++ b/internal/git/log/tag_test.go
@@ -25,16 +25,25 @@ func TestGetTag(t *testing.T) {
 		tagName string
 		rev     string
 		message string
+		trim    bool
 	}{
 		{
+			tagName: fmt.Sprintf("%s-v1.0.2", t.Name()),
+			rev:     "master^^^^",
+			message: strings.Repeat("a", helper.MaxCommitOrTagMessageSize+1),
+			trim:    false,
+		},
+		{
 			tagName: fmt.Sprintf("%s-v1.0.0", t.Name()),
 			rev:     "master^^^",
 			message: "Prod Release v1.0.0",
+			trim:    true,
 		},
 		{
 			tagName: fmt.Sprintf("%s-v1.0.1", t.Name()),
 			rev:     "master^^",
 			message: strings.Repeat("a", helper.MaxCommitOrTagMessageSize+1),
+			trim:    true,
 		},
 	}
 
@@ -44,9 +53,9 @@ func TestGetTag(t *testing.T) {
 		t.Run(testCase.tagName, func(t *testing.T) {
 			tagID := testhelper.CreateTag(t, testRepoPath, testCase.tagName, testCase.rev, &testhelper.CreateTagOpts{Message: testCase.message})
 
-			tag, err := GetTagCatfile(c, string(tagID), testCase.tagName)
+			tag, err := GetTagCatfile(c, tagID, testCase.tagName, testCase.trim, true)
 			require.NoError(t, err)
-			if len(testCase.message) >= helper.MaxCommitOrTagMessageSize {
+			if testCase.trim && len(testCase.message) >= helper.MaxCommitOrTagMessageSize {
 				testCase.message = testCase.message[:helper.MaxCommitOrTagMessageSize]
 			}
 
@@ -62,6 +71,7 @@ func TestSplitRawTag(t *testing.T) {
 		tagContent  string
 		header      tagHeader
 		body        []byte
+		trimNewLine bool
 	}{
 		{
 			description: "tag without a message",
@@ -107,9 +117,21 @@ func TestSplitRawTag(t *testing.T) {
 			},
 			body: []byte("Hello world\n\nThis is a message"),
 		},
+		{
+			description: "tag with message with empty line and right side new line trimming",
+			tagContent:  "object c92faf3e0a557270141be67f206d7cdb99bfc3a2\ntype commit\ntag v2.6.16.28\ntagger Adrian Bunk <bunk@stusta.de> 1156539089 +0200\n\nHello world\n\nThis is a message\n\n",
+			header: tagHeader{
+				oid:     "c92faf3e0a557270141be67f206d7cdb99bfc3a2",
+				tagType: "commit",
+				tag:     "v2.6.16.28",
+				tagger:  "Adrian Bunk <bunk@stusta.de> 1156539089 +0200",
+			},
+			body:        []byte("Hello world\n\nThis is a message"),
+			trimNewLine: true,
+		},
 	}
 	for _, tc := range testCases {
-		header, body, err := splitRawTag(bytes.NewReader([]byte(tc.tagContent)))
+		header, body, err := splitRawTag(bytes.NewReader([]byte(tc.tagContent)), tc.trimNewLine)
 		assert.Equal(t, tc.header, *header)
 		assert.Equal(t, tc.body, body)
 		assert.NoError(t, err)
diff --git a/internal/metadata/featureflag/feature_flags.go b/internal/metadata/featureflag/feature_flags.go
index 7855b6333..ab7e01744 100644
--- a/internal/metadata/featureflag/feature_flags.go
+++ b/internal/metadata/featureflag/feature_flags.go
@@ -6,7 +6,8 @@ const (
 	UploadPackFilter = "upload_pack_filter"
 	// GetAllLFSPointersGo will cause the GetAllLFSPointers RPC to use the go implementation when set
 	GetAllLFSPointersGo = "get_all_lfs_pointers_go"
-
+	// GetTagMessagesGo will cause the GetTagMessages RPC to use the go implementation when set
+	GetTagMessagesGo = "get_tag_messages_go"
 	// LinguistFileCountStats will invoke an additional git-linguist command to get the number of files per language
 	LinguistFileCountStats = "linguist_file_count_stats"
 )
diff --git a/internal/service/ref/refs.go b/internal/service/ref/refs.go
index b153cb09f..f7de0d344 100644
--- a/internal/service/ref/refs.go
+++ b/internal/service/ref/refs.go
@@ -376,7 +376,7 @@ func parseTagLine(c *catfile.Batch, tagLine string) (*gitalypb.Tag, error) {
 	switch refType {
 	// annotated tag
 	case "tag":
-		tag, err := gitlog.GetTagCatfile(c, tagID, refName)
+		tag, err := gitlog.GetTagCatfile(c, tagID, refName, true, true)
 		if err != nil {
 			return nil, fmt.Errorf("getting annotated tag: %v", err)
 		}
diff --git a/internal/service/ref/tag_messages.go b/internal/service/ref/tag_messages.go
index 71acc16e6..88bf9b27d 100644
--- a/internal/service/ref/tag_messages.go
+++ b/internal/service/ref/tag_messages.go
@@ -1,22 +1,97 @@
 package ref
 
 import (
+	"bytes"
 	"fmt"
+	"io"
 
+	"github.com/prometheus/client_golang/prometheus"
+	"gitlab.com/gitlab-org/gitaly/internal/git/catfile"
+	"gitlab.com/gitlab-org/gitaly/internal/git/log"
+	"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"
+	"gitlab.com/gitlab-org/gitaly/streamio"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 )
 
+var getTagMessagesRequests = prometheus.NewCounterVec(
+	prometheus.CounterOpts{
+		Name: "gitaly_get_tag_messages_total",
+		Help: "Counter of go vs ruby implementation of GetTagMessages",
+	},
+	[]string{"implementation"},
+)
+
+func init() {
+	prometheus.MustRegister(getTagMessagesRequests)
+}
+
 func (s *server) GetTagMessages(request *gitalypb.GetTagMessagesRequest, stream gitalypb.RefService_GetTagMessagesServer) error {
 	if err := validateGetTagMessagesRequest(request); err != nil {
 		return status.Errorf(codes.InvalidArgument, "GetTagMessages: %v", err)
 	}
+	if err := s.getAndStreamTagMessages(request, stream); err != nil {
+		return helper.ErrInternal(err)
+	}
+
+	return nil
+}
+
+func validateGetTagMessagesRequest(request *gitalypb.GetTagMessagesRequest) error {
+	if request.GetRepository() == nil {
+		return fmt.Errorf("empty Repository")
+	}
+
+	return nil
+}
+
+func (s *server) getAndStreamTagMessages(request *gitalypb.GetTagMessagesRequest, stream gitalypb.RefService_GetTagMessagesServer) error {
+	if featureflag.IsEnabled(stream.Context(), featureflag.GetTagMessagesGo) {
+		getTagMessagesRequests.WithLabelValues("go").Inc()
+		return getAndStreamTagMessagesGo(request, stream)
+	}
+
+	getTagMessagesRequests.WithLabelValues("ruby").Inc()
+	return getAndStreamTagMessagesRuby(s.ruby, request, stream)
+}
+
+func getAndStreamTagMessagesGo(request *gitalypb.GetTagMessagesRequest, stream gitalypb.RefService_GetTagMessagesServer) error {
+	ctx := stream.Context()
+
+	c, err := catfile.New(ctx, request.GetRepository())
+	if err != nil {
+		return err
+	}
+
+	for _, tagID := range request.GetTagIds() {
+		tag, err := log.GetTagCatfile(c, tagID, "", false, false)
+		if err != nil {
+			return fmt.Errorf("failed to get tag: %v", err)
+		}
 
+		if err := stream.Send(&gitalypb.GetTagMessagesResponse{TagId: tagID}); err != nil {
+			return err
+		}
+		sw := streamio.NewWriter(func(p []byte) error {
+			return stream.Send(&gitalypb.GetTagMessagesResponse{Message: p})
+		})
+
+		msgReader := bytes.NewReader(tag.Message)
+
+		if _, err = io.Copy(sw, msgReader); err != nil {
+			return fmt.Errorf("failed to send response: %v", err)
+		}
+	}
+	return nil
+}
+
+func getAndStreamTagMessagesRuby(ruby *rubyserver.Server, request *gitalypb.GetTagMessagesRequest, stream gitalypb.RefService_GetTagMessagesServer) error {
 	ctx := stream.Context()
 
-	client, err := s.ruby.RefServiceClient(ctx)
+	client, err := ruby.RefServiceClient(ctx)
 	if err != nil {
 		return err
 	}
@@ -41,11 +116,3 @@ func (s *server) GetTagMessages(request *gitalypb.GetTagMessagesRequest, stream
 		return stream.Send(resp)
 	})
 }
-
-func validateGetTagMessagesRequest(request *gitalypb.GetTagMessagesRequest) error {
-	if request.GetRepository() == nil {
-		return fmt.Errorf("empty Repository")
-	}
-
-	return nil
-}
diff --git a/internal/service/ref/tag_messages_test.go b/internal/service/ref/tag_messages_test.go
index 9514bd38a..1ab168b22 100644
--- a/internal/service/ref/tag_messages_test.go
+++ b/internal/service/ref/tag_messages_test.go
@@ -1,15 +1,18 @@
 package ref
 
 import (
+	"context"
 	"io"
 	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/require"
 	"gitlab.com/gitlab-org/gitaly/internal/helper"
+	"gitlab.com/gitlab-org/gitaly/internal/metadata/featureflag"
 	"gitlab.com/gitlab-org/gitaly/internal/testhelper"
 	"gitlab.com/gitlab-org/gitaly/proto/go/gitalypb"
 	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/metadata"
 )
 
 func TestSuccessfulGetTagMessagesRequest(t *testing.T) {
@@ -36,9 +39,6 @@ func TestSuccessfulGetTagMessagesRequest(t *testing.T) {
 		TagIds:     []string{tag1ID, tag2ID},
 	}
 
-	c, err := client.GetTagMessages(ctx, request)
-	require.NoError(t, err)
-
 	expectedMessages := []*gitalypb.GetTagMessagesResponse{
 		{
 			TagId:   tag1ID,
@@ -49,9 +49,32 @@ func TestSuccessfulGetTagMessagesRequest(t *testing.T) {
 			Message: []byte(message2 + "\n"),
 		},
 	}
-	fetchedMessages := readAllMessagesFromClient(t, c)
 
-	require.Equal(t, expectedMessages, fetchedMessages)
+	for title, ctxModifier := range map[string]func(context.Context) context.Context{
+		"enabled_feature_GetTagMessagesGo": func(ctx context.Context) context.Context {
+			return enableGetTagMessagesFeatureFlag(ctx)
+		},
+		"disabled_feature_GetTagMessagesGo": func(ctx context.Context) context.Context {
+			return ctx
+		},
+	} {
+		title := title
+		ctxModifier := ctxModifier
+
+		// all sub-tests are read-only
+		t.Run("parallel", func(t *testing.T) {
+			t.Run(title, func(t *testing.T) {
+				t.Parallel()
+
+				c, err := client.GetTagMessages(ctxModifier(ctx), request)
+				require.NoError(t, err)
+
+				fetchedMessages := readAllMessagesFromClient(t, c)
+
+				require.Equal(t, expectedMessages, fetchedMessages)
+			})
+		})
+	}
 }
 
 func TestFailedGetTagMessagesRequest(t *testing.T) {
@@ -116,3 +139,9 @@ func readAllMessagesFromClient(t *testing.T, c gitalypb.RefService_GetTagMessage
 
 	return
 }
+
+func enableGetTagMessagesFeatureFlag(ctx context.Context) context.Context {
+	return metadata.NewOutgoingContext(ctx, metadata.New(map[string]string{
+		featureflag.HeaderKey(featureflag.GetTagMessagesGo): "true",
+	}))
+}