ssh: upload_pack: Expose deepen/filter/haves statistics via Prometheus

The smarthttp service already keeps track of how many clones use "have",
"deepen" and "filter" instructions. To get a better grasp of how these
features are used, let's introduce the same set of Premotheus counters
for our SSH upload-pack service.

4 files changed, 116 insertions, 9 deletions

diff --git a/internal/service/register.go b/internal/service/register.go
index 658771cab..eb5c9c2cd 100644
--- a/internal/service/register.go
+++ b/internal/service/register.go
@@ -54,6 +54,33 @@ var (
 			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() {
@@ -61,13 +88,15 @@ func init() {
 		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) {
-
 	gitalypb.RegisterBlobServiceServer(grpcServer, blob.NewServer(rubyServer))
 	gitalypb.RegisterCleanupServiceServer(grpcServer, cleanup.NewServer())
 	gitalypb.RegisterCommitServiceServer(grpcServer, commit.NewServer())
@@ -76,7 +105,11 @@ 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.RegisterSSHServiceServer(grpcServer, ssh.NewServer(
+		ssh.WithDeepensMetric(sshDeepensMetric),
+		ssh.WithHavesMetric(sshHavesMetric),
+		ssh.WithFiltersMetric(sshFiltersMetric),
+	))
 	gitalypb.RegisterSmartHTTPServiceServer(grpcServer, smarthttp.NewServer(
 		smarthttp.WithDeepensMetric(smarthttpDeepensMetric),
 		smarthttp.WithFiltersMetric(smarthttpFiltersMetric),
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)
 		})
 	}
 }