Merge branch 'pks-upload-pack-prometheus' into 'master'

Expose Prometheus metrics for shallow, partial and initial clones

See merge request gitlab-org/gitaly!1914

10 files changed, 266 insertions, 34 deletions

diff --git a/internal/prometheus/metrics/metrics.go b/internal/prometheus/metrics/metrics.go
index 9ca576de6..62b72d374 100644
--- a/internal/prometheus/metrics/metrics.go
+++ b/internal/prometheus/metrics/metrics.go
@@ -4,6 +4,12 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 )
 
+// Counter is a subset of a prometheus Counter
+type Counter interface {
+	Inc()
+	Add(float64)
+}
+
 // Gauge is a subset of a prometheus Gauge
 type Gauge interface {
 	Inc()
diff --git a/internal/service/register.go b/internal/service/register.go
index 992ebee45..eb5c9c2cd 100644
--- a/internal/service/register.go
+++ b/internal/service/register.go
@@ -1,6 +1,7 @@
 package service
 
 import (
+	"github.com/prometheus/client_golang/prometheus"
 	"gitlab.com/gitlab-org/gitaly/internal/config"
 	"gitlab.com/gitlab-org/gitaly/internal/rubyserver"
 	"gitlab.com/gitlab-org/gitaly/internal/service/blob"
@@ -26,6 +27,73 @@ import (
 	healthpb "google.golang.org/grpc/health/grpc_health_v1"
 )
 
+var (
+	smarthttpDeepensMetric = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Namespace: "gitaly",
+			Subsystem: "smarthttp",
+			Name:      "deepen_requests_total",
+			Help:      "Number of git-upload-pack requests processed that contained a 'deepen' message",
+		},
+	)
+
+	smarthttpFiltersMetric = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Namespace: "gitaly",
+			Subsystem: "smarthttp",
+			Name:      "filter_requests_total",
+			Help:      "Number of git-upload-pack requests processed that contained a 'filter' message",
+		},
+	)
+
+	smarthttpHavesMetric = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Namespace: "gitaly",
+			Subsystem: "smarthttp",
+			Name:      "haves_requests_total",
+			Help:      "Number of git-upload-pack requests processed that contained a 'have' message",
+		},
+	)
+
+	sshDeepensMetric = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Namespace: "gitaly",
+			Subsystem: "ssh",
+			Name:      "deepen_requests_total",
+			Help:      "Number of git-upload-pack requests processed that contained a 'deepen' message",
+		},
+	)
+
+	sshFiltersMetric = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Namespace: "gitaly",
+			Subsystem: "ssh",
+			Name:      "filter_requests_total",
+			Help:      "Number of git-upload-pack requests processed that contained a 'filter' message",
+		},
+	)
+
+	sshHavesMetric = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Namespace: "gitaly",
+			Subsystem: "ssh",
+			Name:      "haves_requests_total",
+			Help:      "Number of git-upload-pack requests processed that contained a 'have' message",
+		},
+	)
+)
+
+func init() {
+	prometheus.MustRegister(
+		smarthttpDeepensMetric,
+		smarthttpFiltersMetric,
+		smarthttpHavesMetric,
+		sshDeepensMetric,
+		sshFiltersMetric,
+		sshHavesMetric,
+	)
+}
+
 // RegisterAll will register all the known grpc services with
 // the specified grpc service instance
 func RegisterAll(grpcServer *grpc.Server, rubyServer *rubyserver.Server) {
@@ -37,8 +105,16 @@ func RegisterAll(grpcServer *grpc.Server, rubyServer *rubyserver.Server) {
 	gitalypb.RegisterOperationServiceServer(grpcServer, operations.NewServer(rubyServer))
 	gitalypb.RegisterRefServiceServer(grpcServer, ref.NewServer())
 	gitalypb.RegisterRepositoryServiceServer(grpcServer, repository.NewServer(rubyServer, config.GitalyInternalSocketPath()))
-	gitalypb.RegisterSSHServiceServer(grpcServer, ssh.NewServer())
-	gitalypb.RegisterSmartHTTPServiceServer(grpcServer, smarthttp.NewServer())
+	gitalypb.RegisterSSHServiceServer(grpcServer, ssh.NewServer(
+		ssh.WithDeepensMetric(sshDeepensMetric),
+		ssh.WithHavesMetric(sshHavesMetric),
+		ssh.WithFiltersMetric(sshFiltersMetric),
+	))
+	gitalypb.RegisterSmartHTTPServiceServer(grpcServer, smarthttp.NewServer(
+		smarthttp.WithDeepensMetric(smarthttpDeepensMetric),
+		smarthttp.WithFiltersMetric(smarthttpFiltersMetric),
+		smarthttp.WithHavesMetric(smarthttpHavesMetric),
+	))
 	gitalypb.RegisterWikiServiceServer(grpcServer, wiki.NewServer(rubyServer))
 	gitalypb.RegisterConflictsServiceServer(grpcServer, conflicts.NewServer(rubyServer))
 	gitalypb.RegisterRemoteServiceServer(grpcServer, remote.NewServer(rubyServer))
diff --git a/internal/service/smarthttp/server.go b/internal/service/smarthttp/server.go
index 5de729c22..b00dbb043 100644
--- a/internal/service/smarthttp/server.go
+++ b/internal/service/smarthttp/server.go
@@ -1,12 +1,50 @@
 package smarthttp
 
-import "gitlab.com/gitlab-org/gitaly/proto/go/gitalypb"
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"gitlab.com/gitlab-org/gitaly/internal/prometheus/metrics"
+	"gitlab.com/gitlab-org/gitaly/proto/go/gitalypb"
+)
 
 type server struct {
+	deepensMetric metrics.Counter
+	filtersMetric metrics.Counter
+	havesMetric   metrics.Counter
 	gitalypb.UnimplementedSmartHTTPServiceServer
 }
 
 // NewServer creates a new instance of a grpc SmartHTTPServer
-func NewServer() gitalypb.SmartHTTPServiceServer {
-	return &server{}
+func NewServer(serverOpts ...ServerOpt) gitalypb.SmartHTTPServiceServer {
+	s := &server{
+		deepensMetric: prometheus.NewCounter(prometheus.CounterOpts{}),
+		filtersMetric: prometheus.NewCounter(prometheus.CounterOpts{}),
+		havesMetric:   prometheus.NewCounter(prometheus.CounterOpts{}),
+	}
+
+	for _, serverOpt := range serverOpts {
+		serverOpt(s)
+	}
+
+	return s
+}
+
+// ServerOpt is a self referential option for server
+type ServerOpt func(s *server)
+
+func WithDeepensMetric(c metrics.Counter) ServerOpt {
+	return func(s *server) {
+		s.deepensMetric = c
+	}
+}
+
+func WithFiltersMetric(c metrics.Counter) ServerOpt {
+	return func(s *server) {
+		s.filtersMetric = c
+	}
+}
+
+func WithHavesMetric(c metrics.Counter) ServerOpt {
+	return func(s *server) {
+		s.havesMetric = c
+	}
 }
diff --git a/internal/service/smarthttp/testhelper_test.go b/internal/service/smarthttp/testhelper_test.go
index aef5803ce..587c7bc72 100644
--- a/internal/service/smarthttp/testhelper_test.go
+++ b/internal/service/smarthttp/testhelper_test.go
@@ -28,10 +28,10 @@ func testMain(m *testing.M) int {
 	return m.Run()
 }
 
-func runSmartHTTPServer(t *testing.T) (string, func()) {
+func runSmartHTTPServer(t *testing.T, serverOpts ...ServerOpt) (string, func()) {
 	srv := testhelper.NewServer(t, nil, nil)
 
-	gitalypb.RegisterSmartHTTPServiceServer(srv.GrpcServer(), NewServer())
+	gitalypb.RegisterSmartHTTPServiceServer(srv.GrpcServer(), NewServer(serverOpts...))
 	reflection.Register(srv.GrpcServer())
 
 	require.NoError(t, srv.Start())
diff --git a/internal/service/smarthttp/upload_pack.go b/internal/service/smarthttp/upload_pack.go
index e43add012..db18176e4 100644
--- a/internal/service/smarthttp/upload_pack.go
+++ b/internal/service/smarthttp/upload_pack.go
@@ -6,7 +6,6 @@ import (
 	"io"
 
 	grpc_logrus "github.com/grpc-ecosystem/go-grpc-middleware/logging/logrus"
-	"github.com/prometheus/client_golang/prometheus"
 	"gitlab.com/gitlab-org/gitaly/internal/command"
 	"gitlab.com/gitlab-org/gitaly/internal/git"
 	"gitlab.com/gitlab-org/gitaly/internal/git/stats"
@@ -19,19 +18,6 @@ import (
 	"google.golang.org/grpc/status"
 )
 
-var (
-	deepenCount = prometheus.NewCounter(
-		prometheus.CounterOpts{
-			Name: "gitaly_smarthttp_deepen_requests_total",
-			Help: "Number of git-upload-pack requests processed that contained a 'deepen' message",
-		},
-	)
-)
-
-func init() {
-	prometheus.MustRegister(deepenCount)
-}
-
 func (s *server) PostUploadPack(stream gitalypb.SmartHTTPService_PostUploadPackServer) error {
 	ctx := stream.Context()
 
@@ -58,10 +44,24 @@ func (s *server) PostUploadPack(stream gitalypb.SmartHTTPService_PostUploadPackS
 	statsCh := make(chan stats.PackfileNegotiation, 1)
 	go func() {
 		defer close(statsCh)
-		stats := stats.PackfileNegotiation{}
-		if err := stats.Parse(pr); err == nil {
-			statsCh <- stats
+
+		stats, err := stats.ParsePackfileNegotiation(pr)
+		if err != nil {
+			grpc_logrus.Extract(stream.Context()).WithError(err).Debug("failed parsing packfile negotiation")
+			return
 		}
+
+		if stats.Deepen != "" {
+			s.deepensMetric.Inc()
+		}
+		if stats.Filter != "" {
+			s.filtersMetric.Inc()
+		}
+		if len(stats.Haves) > 0 {
+			s.havesMetric.Inc()
+		}
+
+		statsCh <- stats
 	}()
 
 	var respBytes int64
@@ -112,7 +112,6 @@ func (s *server) PostUploadPack(stream gitalypb.SmartHTTPService_PostUploadPackS
 			// We have seen a 'deepen' message in the request. It is expected that
 			// git-upload-pack has a non-zero exit status: don't treat this as an
 			// error.
-			deepenCount.Inc()
 			return nil
 		}
 
diff --git a/internal/service/smarthttp/upload_pack_test.go b/internal/service/smarthttp/upload_pack_test.go
index aa1282e67..57a0f04f5 100644
--- a/internal/service/smarthttp/upload_pack_test.go
+++ b/internal/service/smarthttp/upload_pack_test.go
@@ -19,6 +19,7 @@ import (
 	"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/internal/testhelper/promtest"
 	"gitlab.com/gitlab-org/gitaly/proto/go/gitalypb"
 	"gitlab.com/gitlab-org/gitaly/streamio"
 	"google.golang.org/grpc/codes"
@@ -29,7 +30,11 @@ const (
 )
 
 func TestSuccessfulUploadPackRequest(t *testing.T) {
-	serverSocketPath, stop := runSmartHTTPServer(t)
+	haves := &promtest.MockCounter{}
+
+	serverSocketPath, stop := runSmartHTTPServer(
+		t, WithHavesMetric(haves),
+	)
 	defer stop()
 
 	ctx, cancel := testhelper.Context()
@@ -92,6 +97,8 @@ func TestSuccessfulUploadPackRequest(t *testing.T) {
 
 	// The fact that this command succeeds means that we got the commit correctly, no further checks should be needed.
 	testhelper.MustRunCommand(t, nil, "git", "-C", localRepoPath, "show", string(newHead))
+
+	require.Equal(t, 1.0, haves.Value())
 }
 
 func TestUploadPackRequestWithGitConfigOptions(t *testing.T) {
@@ -324,7 +331,11 @@ func extractPackDataFromResponse(t *testing.T, buf *bytes.Buffer) ([]byte, int,
 }
 
 func TestUploadPackRequestForPartialCloneSuccess(t *testing.T) {
-	serverSocketPath, stop := runSmartHTTPServer(t)
+	filter := &promtest.MockCounter{}
+
+	serverSocketPath, stop := runSmartHTTPServer(
+		t, WithFiltersMetric(filter),
+	)
 	defer stop()
 
 	testRepo := testhelper.TestRepository()
@@ -418,4 +429,6 @@ func TestUploadPackRequestForPartialCloneSuccess(t *testing.T) {
 
 	_, err = makePostUploadPackRequest(ctx, t, serverSocketPath, req, &requestBuffer)
 	require.NoError(t, err)
+
+	require.Equal(t, 2.0, filter.Value())
 }
diff --git a/internal/service/ssh/server.go b/internal/service/ssh/server.go
index 6933b76b6..808877f1f 100644
--- a/internal/service/ssh/server.go
+++ b/internal/service/ssh/server.go
@@ -3,6 +3,8 @@ package ssh
 import (
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
+	"gitlab.com/gitlab-org/gitaly/internal/prometheus/metrics"
 	"gitlab.com/gitlab-org/gitaly/proto/go/gitalypb"
 )
 
@@ -14,6 +16,9 @@ var (
 type server struct {
 	uploadPackRequestTimeout    time.Duration
 	uploadArchiveRequestTimeout time.Duration
+	havesMetric                 metrics.Counter
+	deepensMetric               metrics.Counter
+	filtersMetric               metrics.Counter
 	gitalypb.UnimplementedSSHServiceServer
 }
 
@@ -22,6 +27,9 @@ func NewServer(serverOpts ...ServerOpt) gitalypb.SSHServiceServer {
 	s := &server{
 		uploadPackRequestTimeout:    defaultUploadPackRequestTimeout,
 		uploadArchiveRequestTimeout: defaultUploadArchiveRequestTimeout,
+		deepensMetric:               prometheus.NewCounter(prometheus.CounterOpts{}),
+		filtersMetric:               prometheus.NewCounter(prometheus.CounterOpts{}),
+		havesMetric:                 prometheus.NewCounter(prometheus.CounterOpts{}),
 	}
 
 	for _, serverOpt := range serverOpts {
@@ -47,3 +55,21 @@ func WithArchiveRequestTimeout(d time.Duration) ServerOpt {
 		s.uploadArchiveRequestTimeout = d
 	}
 }
+
+func WithHavesMetric(c metrics.Counter) ServerOpt {
+	return func(s *server) {
+		s.havesMetric = c
+	}
+}
+
+func WithDeepensMetric(c metrics.Counter) ServerOpt {
+	return func(s *server) {
+		s.deepensMetric = c
+	}
+}
+
+func WithFiltersMetric(c metrics.Counter) ServerOpt {
+	return func(s *server) {
+		s.filtersMetric = c
+	}
+}
diff --git a/internal/service/ssh/upload_pack.go b/internal/service/ssh/upload_pack.go
index 94161fead..854eff492 100644
--- a/internal/service/ssh/upload_pack.go
+++ b/internal/service/ssh/upload_pack.go
@@ -3,12 +3,15 @@ package ssh
 import (
 	"context"
 	"fmt"
+	"io"
+	"sync"
 
 	grpc_logrus "github.com/grpc-ecosystem/go-grpc-middleware/logging/logrus"
 	log "github.com/sirupsen/logrus"
 	"gitlab.com/gitlab-org/gitaly/internal/command"
 	"gitlab.com/gitlab-org/gitaly/internal/git"
 	"gitlab.com/gitlab-org/gitaly/internal/git/pktline"
+	"gitlab.com/gitlab-org/gitaly/internal/git/stats"
 	"gitlab.com/gitlab-org/gitaly/internal/helper"
 	"gitlab.com/gitlab-org/gitaly/internal/metadata/featureflag"
 	"gitlab.com/gitlab-org/gitaly/internal/service/inspect"
@@ -83,6 +86,34 @@ func (s *server) sshUploadPack(stream gitalypb.SSHService_SSHUploadPackServer, r
 		globalOpts = append(globalOpts, git.ValueFlag{"-c", o})
 	}
 
+	pr, pw := io.Pipe()
+	defer pw.Close()
+	stdin = io.TeeReader(stdin, pw)
+	wg := sync.WaitGroup{}
+
+	wg.Add(1)
+	go func() {
+		defer func() {
+			wg.Done()
+			pr.Close()
+		}()
+
+		stats, err := stats.ParsePackfileNegotiation(pr)
+		if err != nil {
+			grpc_logrus.Extract(stream.Context()).WithError(err).Debug("failed parsing packfile negotiation")
+			return
+		}
+		if stats.Deepen != "" {
+			s.deepensMetric.Inc()
+		}
+		if stats.Filter != "" {
+			s.filtersMetric.Inc()
+		}
+		if len(stats.Haves) > 0 {
+			s.havesMetric.Inc()
+		}
+	}()
+
 	cmd, monitor, err := monitorStdinCommand(ctx, stdin, stdout, stderr, env, globalOpts, git.SubCmd{
 		Name: "upload-pack",
 		Args: []string{repoPath},
@@ -108,6 +139,9 @@ func (s *server) sshUploadPack(stream gitalypb.SSHService_SSHUploadPackServer, r
 		return fmt.Errorf("cmd wait: %v", err)
 	}
 
+	pw.Close()
+	wg.Wait()
+
 	return nil
 }
 
diff --git a/internal/service/ssh/upload_pack_test.go b/internal/service/ssh/upload_pack_test.go
index 04a53b357..447b9c8df 100644
--- a/internal/service/ssh/upload_pack_test.go
+++ b/internal/service/ssh/upload_pack_test.go
@@ -17,6 +17,7 @@ import (
 	"gitlab.com/gitlab-org/gitaly/internal/helper/text"
 	"gitlab.com/gitlab-org/gitaly/internal/metadata/featureflag"
 	"gitlab.com/gitlab-org/gitaly/internal/testhelper"
+	"gitlab.com/gitlab-org/gitaly/internal/testhelper/promtest"
 	"gitlab.com/gitlab-org/gitaly/proto/go/gitalypb"
 	"google.golang.org/grpc/codes"
 )
@@ -185,22 +186,31 @@ func TestFailedUploadPackRequestDueToValidationError(t *testing.T) {
 }
 
 func TestUploadPackCloneSuccess(t *testing.T) {
-	serverSocketPath, stop := runSSHServer(t)
+	deepen := &promtest.MockCounter{}
+	filter := &promtest.MockCounter{}
+	haves := &promtest.MockCounter{}
+
+	serverSocketPath, stop := runSSHServer(
+		t, WithDeepensMetric(deepen), WithFiltersMetric(filter), WithHavesMetric(haves),
+	)
 	defer stop()
 
 	localRepoPath := path.Join(testRepoRoot, "gitlab-test-upload-pack-local")
 
 	tests := []struct {
-		cmd  *exec.Cmd
-		desc string
+		cmd    *exec.Cmd
+		desc   string
+		deepen float64
 	}{
 		{
-			cmd:  exec.Command("git", "clone", "git@localhost:test/test.git", localRepoPath),
-			desc: "full clone",
+			cmd:    exec.Command("git", "clone", "git@localhost:test/test.git", localRepoPath),
+			desc:   "full clone",
+			deepen: 0,
 		},
 		{
-			cmd:  exec.Command("git", "clone", "--depth", "1", "git@localhost:test/test.git", localRepoPath),
-			desc: "shallow clone",
+			cmd:    exec.Command("git", "clone", "--depth", "1", "git@localhost:test/test.git", localRepoPath),
+			desc:   "shallow clone",
+			deepen: 1,
 		},
 	}
 
@@ -213,6 +223,10 @@ func TestUploadPackCloneSuccess(t *testing.T) {
 			}
 			lHead, rHead, _, _ := cmd.test(t, localRepoPath)
 			require.Equal(t, lHead, rHead, "local and remote head not equal")
+
+			require.Equal(t, deepen.Value(), tc.deepen)
+			require.Equal(t, filter.Value(), 0.0)
+			require.Equal(t, haves.Value(), 0.0)
 		})
 	}
 }
diff --git a/internal/testhelper/promtest/counter.go b/internal/testhelper/promtest/counter.go
new file mode 100644
index 000000000..665023436
--- /dev/null
+++ b/internal/testhelper/promtest/counter.go
@@ -0,0 +1,26 @@
+package promtest
+
+import (
+	"sync"
+)
+
+type MockCounter struct {
+	m     sync.RWMutex
+	value float64
+}
+
+func (m *MockCounter) Value() float64 {
+	m.m.RLock()
+	defer m.m.RUnlock()
+	return m.value
+}
+
+func (m *MockCounter) Inc() {
+	m.Add(1)
+}
+
+func (m *MockCounter) Add(v float64) {
+	m.m.Lock()
+	defer m.m.Unlock()
+	m.value += v
+}