Praefect should not emit Gitaly errors to Sentry

Stream interceptor made unexported to omit potential use
in production code.

Closes: https://gitlab.com/gitlab-org/gitaly/issues/2434

5 files changed, 92 insertions, 7 deletions

diff --git a/changelogs/unreleased/ps-sentry-gitaly-errors-duplication.yml b/changelogs/unreleased/ps-sentry-gitaly-errors-duplication.yml
new file mode 100644
index 000000000..3c4221a73
--- /dev/null
+++ b/changelogs/unreleased/ps-sentry-gitaly-errors-duplication.yml
@@ -0,0 +1,5 @@
+---
+title: Praefect should not emit Gitaly errors to Sentry
+merge_request: 1880
+author:
+type: fixed
diff --git a/internal/middleware/sentryhandler/sentryhandler.go b/internal/middleware/sentryhandler/sentryhandler.go
index 876937cdf..7766e14c5 100644
--- a/internal/middleware/sentryhandler/sentryhandler.go
+++ b/internal/middleware/sentryhandler/sentryhandler.go
@@ -15,6 +15,10 @@ import (
 	"google.golang.org/grpc/codes"
 )
 
+const (
+	skipSubmission = "sentry.skip"
+)
+
 var ignoredCodes = []codes.Code{
 	// OK means there was no error
 	codes.OK,
@@ -63,7 +67,7 @@ func methodToCulprit(methodName string) string {
 	return methodName
 }
 
-func logErrorToSentry(err error) (code codes.Code, bypass bool) {
+func logErrorToSentry(ctx context.Context, err error) (code codes.Code, bypass bool) {
 	code = helper.GrpcCode(err)
 
 	for _, ignoredCode := range ignoredCodes {
@@ -72,11 +76,16 @@ func logErrorToSentry(err error) (code codes.Code, bypass bool) {
 		}
 	}
 
+	tags := grpc_ctxtags.Extract(ctx)
+	if tags.Has(skipSubmission) {
+		return code, true
+	}
+
 	return code, false
 }
 
 func generateSentryEvent(ctx context.Context, method string, start time.Time, err error) *sentry.Event {
-	grpcErrorCode, bypass := logErrorToSentry(err)
+	grpcErrorCode, bypass := logErrorToSentry(ctx, err)
 	if bypass {
 		return nil
 	}
@@ -136,3 +145,9 @@ func newException(err error, stacktrace *sentry.Stacktrace) sentry.Exception {
 	}
 	return ex
 }
+
+// MarkToSkip propagate context with a special tag that signals to sentry handler that the error must not be reported.
+func MarkToSkip(ctx context.Context) {
+	tags := grpc_ctxtags.Extract(ctx)
+	tags.Set(skipSubmission, struct{}{})
+}
diff --git a/internal/middleware/sentryhandler/sentryhandler_test.go b/internal/middleware/sentryhandler/sentryhandler_test.go
index 653d72525..dcc78e9d4 100644
--- a/internal/middleware/sentryhandler/sentryhandler_test.go
+++ b/internal/middleware/sentryhandler/sentryhandler_test.go
@@ -6,6 +6,7 @@ import (
 	"testing"
 	"time"
 
+	grpc_ctxtags "github.com/grpc-ecosystem/go-grpc-middleware/tags"
 	"github.com/stretchr/testify/assert"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -14,6 +15,7 @@ import (
 func Test_generateSentryEvent(t *testing.T) {
 	tests := []struct {
 		name        string
+		ctx         context.Context
 		method      string
 		sinceStart  time.Duration
 		wantNil     bool
@@ -68,11 +70,30 @@ func Test_generateSentryEvent(t *testing.T) {
 			err:        status.Errorf(codes.FailedPrecondition, "Something failed"),
 			wantNil:    true,
 		},
+		{
+			name: "marked to skip",
+			ctx: func() context.Context {
+				var result context.Context
+				ctx := context.Background()
+				// this is the only way how we could populate context with `tags` assembler
+				grpc_ctxtags.UnaryServerInterceptor()(ctx, nil, nil, func(ctx context.Context, req interface{}) (interface{}, error) {
+					result = ctx
+					return nil, nil
+				})
+				MarkToSkip(result)
+				return result
+			}(),
+			wantNil: true,
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
+			ctx := context.Background()
+			if tt.ctx != nil {
+				ctx = tt.ctx
+			}
 			start := time.Now().Add(-tt.sinceStart)
-			event := generateSentryEvent(context.Background(), tt.method, start, tt.err)
+			event := generateSentryEvent(ctx, tt.method, start, tt.err)
 
 			if tt.wantNil {
 				assert.Nil(t, event)
diff --git a/internal/praefect/grpc-proxy/proxy/handler.go b/internal/praefect/grpc-proxy/proxy/handler.go
index 0b8c901f7..a47baa7c3 100644
--- a/internal/praefect/grpc-proxy/proxy/handler.go
+++ b/internal/praefect/grpc-proxy/proxy/handler.go
@@ -10,6 +10,7 @@ package proxy
 import (
 	"io"
 
+	"gitlab.com/gitlab-org/gitaly/internal/middleware/sentryhandler"
 	"golang.org/x/net/context"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
@@ -109,9 +110,14 @@ func (s *handler) handler(srv interface{}, serverStream grpc.ServerStream) error
 			// This happens when the clientStream has nothing else to offer (io.EOF), returned a gRPC error. In those two
 			// cases we may have received Trailers as part of the call. In case of other errors (stream closed) the trailers
 			// will be nil.
-			serverStream.SetTrailer(clientStream.Trailer())
+			trailer := clientStream.Trailer()
+			serverStream.SetTrailer(trailer)
 			// c2sErr will contain RPC error from client code. If not io.EOF return the RPC error as server stream error.
 			if c2sErr != io.EOF {
+				if trailer != nil {
+					// we must not propagate Gitaly errors into Sentry
+					sentryhandler.MarkToSkip(serverStream.Context())
+				}
 				return c2sErr
 			}
 			return nil
diff --git a/internal/praefect/grpc-proxy/proxy/handler_test.go b/internal/praefect/grpc-proxy/proxy/handler_test.go
index bbd89886d..676387a2d 100644
--- a/internal/praefect/grpc-proxy/proxy/handler_test.go
+++ b/internal/praefect/grpc-proxy/proxy/handler_test.go
@@ -12,14 +12,22 @@ import (
 	"io"
 	"log"
 	"net"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
 	"os"
 	"strings"
 	"testing"
 	"time"
 
+	"github.com/getsentry/sentry-go"
+	grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"
+	grpc_ctxtags "github.com/grpc-ecosystem/go-grpc-middleware/tags"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
+	"gitlab.com/gitlab-org/gitaly/internal/helper/fieldextractors"
+	"gitlab.com/gitlab-org/gitaly/internal/middleware/sentryhandler"
 	"gitlab.com/gitlab-org/gitaly/internal/praefect/grpc-proxy/proxy"
 	pb "gitlab.com/gitlab-org/gitaly/internal/praefect/grpc-proxy/testdata"
 	"golang.org/x/net/context"
@@ -62,7 +70,7 @@ func (s *assertingService) Ping(ctx context.Context, ping *pb.PingRequest) (*pb.
 }
 
 func (s *assertingService) PingError(ctx context.Context, ping *pb.PingRequest) (*pb.Empty, error) {
-	return nil, grpc.Errorf(codes.FailedPrecondition, "Userspace error.")
+	return nil, grpc.Errorf(codes.ResourceExhausted, "Userspace error.")
 }
 
 func (s *assertingService) PingList(ping *pb.PingRequest, stream pb.TestService_PingListServer) error {
@@ -141,10 +149,31 @@ func (s *ProxyHappySuite) TestPingCarriesServerHeadersAndTrailers() {
 }
 
 func (s *ProxyHappySuite) TestPingErrorPropagatesAppError() {
-	_, err := s.testClient.PingError(s.ctx(), &pb.PingRequest{Value: "foo"})
+	sentryTriggered := 0
+	sentrySrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		sentryTriggered++
+	}))
+	defer sentrySrv.Close()
+
+	// minimal required sentry client configuration
+	sentryURL, err := url.Parse(sentrySrv.URL)
+	require.NoError(s.T(), err)
+	sentryURL.User = url.UserPassword("stub", "stub")
+	sentryURL.Path = "/stub/1"
+
+	require.NoError(s.T(), sentry.Init(sentry.ClientOptions{
+		Dsn:       sentryURL.String(),
+		Transport: sentry.NewHTTPSyncTransport(),
+	}))
+
+	sentry.CaptureEvent(sentry.NewEvent())
+	require.Equal(s.T(), 1, sentryTriggered, "sentry configured incorrectly")
+
+	_, err = s.testClient.PingError(s.ctx(), &pb.PingRequest{Value: "foo"})
 	require.Error(s.T(), err, "PingError should never succeed")
-	assert.Equal(s.T(), codes.FailedPrecondition, grpc.Code(err))
+	assert.Equal(s.T(), codes.ResourceExhausted, grpc.Code(err))
 	assert.Equal(s.T(), "Userspace error.", grpc.ErrorDesc(err))
+	require.Equal(s.T(), 1, sentryTriggered, "sentry must not be triggered because errors from remote must be just propagated")
 }
 
 func (s *ProxyHappySuite) TestDirectorErrorIsPropagated() {
@@ -215,8 +244,17 @@ func (s *ProxyHappySuite) SetupSuite() {
 		// Explicitly copy the metadata, otherwise the tests will fail.
 		return proxy.NewStreamParameters(ctx, s.serverClientConn, nil, nil), nil
 	}
+
 	s.proxy = grpc.NewServer(
 		grpc.CustomCodec(proxy.Codec()),
+		grpc.StreamInterceptor(
+			grpc_middleware.ChainStreamServer(
+				// context tags usage is required by sentryhandler.StreamLogHandler
+				grpc_ctxtags.StreamServerInterceptor(grpc_ctxtags.WithFieldExtractorForInitialReq(fieldextractors.FieldExtractor)),
+				// sentry middleware to capture errors
+				sentryhandler.StreamLogHandler,
+			),
+		),
 		grpc.UnknownServiceHandler(proxy.TransparentHandler(director)),
 	)
 	// Ping handler is handled as an explicit registration and not as a TransparentHandler.