Merge branch 'ps-filter-shas-with-signatures' into 'master'

Filter collection of SHAs which has signatures and return those SHAs

Closes #2157

See merge request gitlab-org/gitaly!1672

5 files changed, 135 insertions, 36 deletions

diff --git a/changelogs/unreleased/ps-filter-shas-with-signatures.yml b/changelogs/unreleased/ps-filter-shas-with-signatures.yml
new file mode 100644
index 000000000..8d5d790a4
--- /dev/null
+++ b/changelogs/unreleased/ps-filter-shas-with-signatures.yml
@@ -0,0 +1,5 @@
+---
+title: 'Filter collection of SHAs which has signatures and return those SHAs: go implementation'
+merge_request: 1672
+author:
+type: performance
diff --git a/internal/metadata/featureflag/feature_flags.go b/internal/metadata/featureflag/feature_flags.go
index ab7e01744..7620a3e3f 100644
--- a/internal/metadata/featureflag/feature_flags.go
+++ b/internal/metadata/featureflag/feature_flags.go
@@ -8,6 +8,8 @@ const (
 	GetAllLFSPointersGo = "get_all_lfs_pointers_go"
 	// GetTagMessagesGo will cause the GetTagMessages RPC to use the go implementation when set
 	GetTagMessagesGo = "get_tag_messages_go"
+	// FilterShasWithSignaturesGo will cause the FilterShasWithSignatures RPC to use the go implementation when set
+	FilterShasWithSignaturesGo = "filter_shas_with_signatures_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/commit/filter_shas_with_signatures.go b/internal/service/commit/filter_shas_with_signatures.go
index ac9e4f6a1..3863bb649 100644
--- a/internal/service/commit/filter_shas_with_signatures.go
+++ b/internal/service/commit/filter_shas_with_signatures.go
@@ -1,24 +1,116 @@
 package commit
 
 import (
+	"errors"
+	"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"
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/status"
 )
 
+var filterShasWithSignaturesRequests = prometheus.NewCounterVec(
+	prometheus.CounterOpts{
+		Name: "gitaly_filter_shas_with_signatures_total",
+		Help: "Counter of go vs ruby implementation of FilterShasWithSignatures",
+	},
+	[]string{"implementation"},
+)
+
+func init() {
+	prometheus.MustRegister(filterShasWithSignaturesRequests)
+}
+
 func (s *server) FilterShasWithSignatures(bidi gitalypb.CommitService_FilterShasWithSignaturesServer) error {
 	firstRequest, err := bidi.Recv()
 	if err != nil {
 		return err
 	}
 
-	if err = verifyFirstFilterShasWithSignaturesRequest(firstRequest); err != nil {
+	if err = validateFirstFilterShasWithSignaturesRequest(firstRequest); err != nil {
+		return helper.ErrInvalidArgument(err)
+	}
+
+	if err := s.filterShasWithSignatures(bidi, firstRequest); err != nil {
+		return helper.ErrInternal(err)
+	}
+	return nil
+}
+
+func validateFirstFilterShasWithSignaturesRequest(in *gitalypb.FilterShasWithSignaturesRequest) error {
+	if in.Repository == nil {
+		return errors.New("no repository given")
+	}
+	return nil
+}
+
+func (s *server) filterShasWithSignatures(bidi gitalypb.CommitService_FilterShasWithSignaturesServer, firstRequest *gitalypb.FilterShasWithSignaturesRequest) error {
+	if featureflag.IsEnabled(bidi.Context(), featureflag.FilterShasWithSignaturesGo) {
+		filterShasWithSignaturesRequests.WithLabelValues("go").Inc()
+		return streamShasWithSignatures(bidi, firstRequest)
+	}
+
+	filterShasWithSignaturesRequests.WithLabelValues("ruby").Inc()
+	return filterShasWithSignaturesRuby(s.ruby, bidi, firstRequest)
+}
+
+func streamShasWithSignatures(bidi gitalypb.CommitService_FilterShasWithSignaturesServer, firstRequest *gitalypb.FilterShasWithSignaturesRequest) error {
+	c, err := catfile.New(bidi.Context(), firstRequest.GetRepository())
+	if err != nil {
 		return err
 	}
 
+	var request = firstRequest
+	for {
+		shas, err := filterCommitShasWithSignatures(c, request.GetShas())
+		if err != nil {
+			return err
+		}
+
+		if err := bidi.Send(&gitalypb.FilterShasWithSignaturesResponse{Shas: shas}); err != nil {
+			return err
+		}
+
+		request, err = bidi.Recv()
+		if err == io.EOF {
+			return nil
+		}
+
+		if err != nil {
+			return err
+		}
+	}
+}
+
+func filterCommitShasWithSignatures(c *catfile.Batch, shas [][]byte) ([][]byte, error) {
+	var foundShas [][]byte
+	for _, sha := range shas {
+		commit, err := log.GetCommitCatfile(c, string(sha))
+		if catfile.IsNotFound(err) {
+			continue
+		}
+
+		if err != nil {
+			return nil, err
+		}
+
+		if commit.SignatureType == gitalypb.SignatureType_NONE {
+			continue
+		}
+
+		foundShas = append(foundShas, sha)
+	}
+
+	return foundShas, nil
+}
+
+func filterShasWithSignaturesRuby(ruby *rubyserver.Server, bidi gitalypb.CommitService_FilterShasWithSignaturesServer, firstRequest *gitalypb.FilterShasWithSignaturesRequest) error {
 	ctx := bidi.Context()
-	client, err := s.ruby.CommitServiceClient(ctx)
+	client, err := ruby.CommitServiceClient(ctx)
 	if err != nil {
 		return err
 	}
@@ -57,10 +149,3 @@ func (s *server) FilterShasWithSignatures(bidi gitalypb.CommitService_FilterShas
 		},
 	)
 }
-
-func verifyFirstFilterShasWithSignaturesRequest(in *gitalypb.FilterShasWithSignaturesRequest) error {
-	if in.Repository == nil {
-		return status.Errorf(codes.InvalidArgument, "no repository given")
-	}
-	return nil
-}
diff --git a/internal/service/commit/filter_shas_with_signatures_test.go b/internal/service/commit/filter_shas_with_signatures_test.go
index 0bd5e9822..1db1feba7 100644
--- a/internal/service/commit/filter_shas_with_signatures_test.go
+++ b/internal/service/commit/filter_shas_with_signatures_test.go
@@ -1,10 +1,12 @@
 package commit
 
 import (
+	"context"
 	"io"
 	"testing"
 
 	"github.com/stretchr/testify/require"
+	"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"
@@ -23,11 +25,13 @@ func TestFilterShasWithSignaturesSuccessful(t *testing.T) {
 	testRepo, _, cleanupFn := testhelper.NewTestRepo(t)
 	defer cleanupFn()
 
-	testCases := []struct {
+	type testCase struct {
 		desc string
 		in   [][]byte
 		out  [][]byte
-	}{
+	}
+
+	testCases := []testCase{
 		{
 			desc: "3 shas, none signed",
 			in:   [][]byte{[]byte("6907208d755b60ebeacb2e9dfea74c92c3449a1f"), []byte("c347ca2e140aa667b968e51ed0ffe055501fe4f4"), []byte("d59c60028b053793cecfb4022de34602e1a9218e")},
@@ -50,17 +54,28 @@ func TestFilterShasWithSignaturesSuccessful(t *testing.T) {
 		},
 	}
 
-	for _, testCase := range testCases {
-		t.Run(testCase.desc, func(t *testing.T) {
-			stream, err := client.FilterShasWithSignatures(ctx)
-			require.NoError(t, err)
-			require.NoError(t, stream.Send(&gitalypb.FilterShasWithSignaturesRequest{Repository: testRepo, Shas: testCase.in}))
-			require.NoError(t, stream.CloseSend())
-			recvOut, err := recvFSWS(stream)
-			require.NoError(t, err)
-			require.Equal(t, testCase.out, recvOut)
-		})
+	check := func(t *testing.T, ctx context.Context, testCases []testCase) {
+		for _, tc := range testCases {
+			t.Run(tc.desc, func(t *testing.T) {
+				stream, err := client.FilterShasWithSignatures(ctx)
+				require.NoError(t, err)
+				require.NoError(t, stream.Send(&gitalypb.FilterShasWithSignaturesRequest{Repository: testRepo, Shas: tc.in}))
+				require.NoError(t, stream.CloseSend())
+				recvOut, err := recvFSWS(stream)
+				require.NoError(t, err)
+				require.Equal(t, tc.out, recvOut)
+			})
+		}
 	}
+
+	t.Run("enabled_feature_FilterShasWithSignaturesGo", func(t *testing.T) {
+		featureCtx := featureflag.ContextWithFeatureFlag(ctx, featureflag.FilterShasWithSignaturesGo)
+		check(t, featureCtx, testCases)
+	})
+
+	t.Run("disabled_feature_FilterShasWithSignaturesGo", func(t *testing.T) {
+		check(t, ctx, testCases)
+	})
 }
 
 func TestFilterShasWithSignaturesValidationError(t *testing.T) {
diff --git a/internal/service/ref/tag_messages_test.go b/internal/service/ref/tag_messages_test.go
index 1ab168b22..d5b38af1f 100644
--- a/internal/service/ref/tag_messages_test.go
+++ b/internal/service/ref/tag_messages_test.go
@@ -58,21 +58,13 @@ func TestSuccessfulGetTagMessagesRequest(t *testing.T) {
 			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)
+		t.Run(title, func(t *testing.T) {
+			c, err := client.GetTagMessages(ctxModifier(ctx), request)
+			require.NoError(t, err)
 
-				fetchedMessages := readAllMessagesFromClient(t, c)
+			fetchedMessages := readAllMessagesFromClient(t, c)
 
-				require.Equal(t, expectedMessages, fetchedMessages)
-			})
+			require.Equal(t, expectedMessages, fetchedMessages)
 		})
 	}
 }