Merge branch 'pks-git-stats-packfile-negotiation' into 'master'

Implement analysis for packfile negotiation

See merge request gitlab-org/gitaly!1936

5 files changed, 347 insertions, 84 deletions

diff --git a/internal/git/stats/packfile_negotiation.go b/internal/git/stats/packfile_negotiation.go
new file mode 100644
index 000000000..af1701ca3
--- /dev/null
+++ b/internal/git/stats/packfile_negotiation.go
@@ -0,0 +1,104 @@
+package stats
+
+import (
+	"errors"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"strings"
+
+	"gitlab.com/gitlab-org/gitaly/internal/git/pktline"
+	"gitlab.com/gitlab-org/gitaly/internal/helper/text"
+)
+
+type PackfileNegotiation struct {
+	// Total size of all pktlines' data
+	PayloadSize int64
+	// Total number of packets
+	Packets int
+	// Capabilities announced by the client
+	Caps []string
+	// Object IDs wanted by the client
+	Wants []string
+	// Object IDs the client has available
+	Haves []string
+	// Objects which the client has as shallow boundaries
+	Shallows []string
+	// Deepen-filter. One of "deepen <depth>", "deepen-since <timestamp>", "deepen-not <ref>".
+	Deepen string
+	// Filter-spec specified by the client.
+	Filter string
+}
+
+func ParsePackfileNegotiation(body io.Reader) (PackfileNegotiation, error) {
+	n := PackfileNegotiation{}
+	return n, n.Parse(body)
+}
+
+// Expected Format:
+// want <OID> <capabilities\n
+// [want <OID>...]
+// [shallow <OID>]
+// [deepen <depth>|deepen-since <timestamp>|deepen-not <ref>]
+// [filter <filter-spec>]
+// flush
+// have <OID>
+// flush|done
+func (n *PackfileNegotiation) Parse(body io.Reader) error {
+	defer io.Copy(ioutil.Discard, body)
+
+	scanner := pktline.NewScanner(body)
+
+	for ; scanner.Scan(); n.Packets++ {
+		pkt := scanner.Bytes()
+		data := text.ChompBytes(pktline.Data(pkt))
+		split := strings.Split(data, " ")
+		n.PayloadSize += int64(len(data))
+
+		switch split[0] {
+		case "want":
+			if len(split) < 2 {
+				return fmt.Errorf("invalid 'want' for packet %d: %v", n.Packets, data)
+			}
+			if len(split) > 2 && n.Caps != nil {
+				return fmt.Errorf("capabilities announced multiple times in packet %d: %v", n.Packets, data)
+			}
+
+			n.Wants = append(n.Wants, split[1])
+			if len(split) > 2 {
+				n.Caps = split[2:]
+			}
+		case "shallow":
+			if len(split) != 2 {
+				return fmt.Errorf("invalid 'shallow' for packet %d: %v", n.Packets, data)
+			}
+			n.Shallows = append(n.Shallows, split[1])
+		case "deepen", "deepen-since", "deepen-not":
+			if len(split) != 2 {
+				return fmt.Errorf("invalid 'deepen' for packet %d: %v", n.Packets, data)
+			}
+			n.Deepen = data
+		case "filter":
+			if len(split) != 2 {
+				return fmt.Errorf("invalid 'filter' for packet %d: %v", n.Packets, data)
+			}
+			n.Filter = split[1]
+		case "have":
+			if len(split) != 2 {
+				return fmt.Errorf("invalid 'have' for packet %d: %v", n.Packets, data)
+			}
+			n.Haves = append(n.Haves, split[1])
+		case "done":
+			break
+		}
+	}
+
+	if scanner.Err() != nil {
+		return scanner.Err()
+	}
+	if len(n.Wants) == 0 {
+		return errors.New("no 'want' sent by client")
+	}
+
+	return nil
+}
diff --git a/internal/git/stats/packfile_negotiation_test.go b/internal/git/stats/packfile_negotiation_test.go
new file mode 100644
index 000000000..903a00b08
--- /dev/null
+++ b/internal/git/stats/packfile_negotiation_test.go
@@ -0,0 +1,232 @@
+package stats
+
+import (
+	"bytes"
+	"io"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"gitlab.com/gitlab-org/gitaly/internal/git/pktline"
+)
+
+const (
+	oid1 = "78fb81a02b03f0013360292ec5106763af32c287"
+	oid2 = "0f6394307cd7d4909be96a0c818d8094a4cb0e5b"
+)
+
+func requireParses(t *testing.T, reader io.Reader, expected PackfileNegotiation) {
+	actual, err := ParsePackfileNegotiation(reader)
+	require.NoError(t, err)
+	actual.PayloadSize = 0
+	actual.Packets = 0
+
+	require.Equal(t, expected, actual)
+}
+
+func TestPackNegoWithInvalidPktline(t *testing.T) {
+	buf := new(bytes.Buffer)
+	pktline.WriteString(buf, "want "+oid1+" cap")
+	pktline.WriteFlush(buf)
+	// Write string with invalid length
+	buf.WriteString("0002xyz")
+	pktline.WriteString(buf, "done")
+
+	_, err := ParsePackfileNegotiation(buf)
+	require.Error(t, err, "invalid pktlines should be rejected")
+}
+
+func TestPackNegoWithSingleWant(t *testing.T) {
+	buf := new(bytes.Buffer)
+	pktline.WriteString(buf, "want "+oid1+" cap")
+	pktline.WriteFlush(buf)
+	pktline.WriteString(buf, "done")
+
+	requireParses(t, buf, PackfileNegotiation{
+		Wants: []string{oid1}, Caps: []string{"cap"},
+	})
+}
+
+func TestPackNegoWithMissingCaps(t *testing.T) {
+	buf := new(bytes.Buffer)
+	pktline.WriteString(buf, "want "+oid1)
+	pktline.WriteFlush(buf)
+	pktline.WriteString(buf, "done")
+
+	requireParses(t, buf, PackfileNegotiation{
+		Wants: []string{oid1},
+	})
+}
+
+func TestPackNegoWithMissingWant(t *testing.T) {
+	buf := new(bytes.Buffer)
+	pktline.WriteString(buf, "have "+oid2)
+	pktline.WriteString(buf, "done")
+
+	_, err := ParsePackfileNegotiation(buf)
+	require.Error(t, err, "packfile negotiation with missing 'want' is invalid")
+}
+
+func TestPackNegoWithHave(t *testing.T) {
+	buf := new(bytes.Buffer)
+	pktline.WriteString(buf, "want "+oid1+" cap")
+	pktline.WriteFlush(buf)
+	pktline.WriteString(buf, "have "+oid2)
+	pktline.WriteString(buf, "done")
+
+	requireParses(t, buf, PackfileNegotiation{
+		Wants: []string{oid1}, Haves: []string{oid2}, Caps: []string{"cap"},
+	})
+}
+
+func TestPackNegoWithMultipleHaveRoundds(t *testing.T) {
+	buf := new(bytes.Buffer)
+	pktline.WriteString(buf, "want "+oid1+" cap")
+	pktline.WriteFlush(buf)
+	pktline.WriteString(buf, "have "+oid2)
+	pktline.WriteFlush(buf)
+	pktline.WriteString(buf, "have "+oid1)
+	pktline.WriteFlush(buf)
+	pktline.WriteString(buf, "done")
+
+	requireParses(t, buf, PackfileNegotiation{
+		Wants: []string{oid1},
+		Haves: []string{oid2, oid1},
+		Caps:  []string{"cap"},
+	})
+}
+
+func TestPackNegoWithMultipleWants(t *testing.T) {
+	buf := new(bytes.Buffer)
+	pktline.WriteString(buf, "want "+oid1+" cap")
+	pktline.WriteString(buf, "want "+oid2)
+	pktline.WriteFlush(buf)
+	pktline.WriteString(buf, "done")
+
+	requireParses(t, buf, PackfileNegotiation{
+		Wants: []string{oid1, oid2}, Caps: []string{"cap"},
+	})
+}
+
+func TestPackNegoWithMultipleCapLines(t *testing.T) {
+	buf := new(bytes.Buffer)
+	pktline.WriteString(buf, "want "+oid1+" cap1")
+	pktline.WriteString(buf, "want "+oid2+" cap2")
+	pktline.WriteFlush(buf)
+	pktline.WriteString(buf, "done")
+
+	_, err := ParsePackfileNegotiation(buf)
+	require.Error(t, err, "multiple capability announcements should fail to parse")
+}
+
+func TestPackNegoWithDeepen(t *testing.T) {
+	buf := new(bytes.Buffer)
+	pktline.WriteString(buf, "want "+oid1+" cap")
+	pktline.WriteString(buf, "deepen 1")
+	pktline.WriteFlush(buf)
+	pktline.WriteString(buf, "done")
+
+	requireParses(t, buf, PackfileNegotiation{
+		Wants:  []string{oid1},
+		Caps:   []string{"cap"},
+		Deepen: "deepen 1",
+	})
+}
+
+func TestPackNegoWithMultipleDeepens(t *testing.T) {
+	buf := new(bytes.Buffer)
+	pktline.WriteString(buf, "want "+oid1+" cap")
+	pktline.WriteFlush(buf)
+	pktline.WriteString(buf, "deepen 1")
+	pktline.WriteString(buf, "deepen-not "+oid2)
+	pktline.WriteFlush(buf)
+	pktline.WriteString(buf, "done")
+
+	requireParses(t, buf, PackfileNegotiation{
+		Wants:  []string{oid1},
+		Caps:   []string{"cap"},
+		Deepen: "deepen-not " + oid2,
+	})
+}
+
+func TestPackNegoWithShallow(t *testing.T) {
+	buf := new(bytes.Buffer)
+	pktline.WriteString(buf, "want "+oid1+" cap")
+	pktline.WriteString(buf, "shallow "+oid1)
+	pktline.WriteFlush(buf)
+	pktline.WriteString(buf, "done")
+
+	requireParses(t, buf, PackfileNegotiation{
+		Wants:    []string{oid1},
+		Caps:     []string{"cap"},
+		Shallows: []string{oid1},
+	})
+}
+
+func TestPackNegoWithMultipleShallows(t *testing.T) {
+	buf := new(bytes.Buffer)
+	pktline.WriteString(buf, "want "+oid1+" cap")
+	pktline.WriteString(buf, "shallow "+oid1)
+	pktline.WriteString(buf, "shallow "+oid2)
+	pktline.WriteFlush(buf)
+	pktline.WriteString(buf, "done")
+
+	requireParses(t, buf, PackfileNegotiation{
+		Wants:    []string{oid1},
+		Caps:     []string{"cap"},
+		Shallows: []string{oid1, oid2},
+	})
+}
+
+func TestPackNegoWithFilter(t *testing.T) {
+	buf := new(bytes.Buffer)
+	pktline.WriteString(buf, "want "+oid1+" cap")
+	pktline.WriteString(buf, "filter blob:none")
+	pktline.WriteFlush(buf)
+	pktline.WriteString(buf, "done")
+
+	requireParses(t, buf, PackfileNegotiation{
+		Wants:  []string{oid1},
+		Caps:   []string{"cap"},
+		Filter: "blob:none",
+	})
+}
+
+func TestPackNegoWithMultipleFilters(t *testing.T) {
+	buf := new(bytes.Buffer)
+	pktline.WriteString(buf, "want "+oid1+" cap")
+	pktline.WriteString(buf, "filter blob:none")
+	pktline.WriteString(buf, "filter blob:limit=1m")
+	pktline.WriteFlush(buf)
+	pktline.WriteString(buf, "done")
+
+	requireParses(t, buf, PackfileNegotiation{
+		Wants:  []string{oid1},
+		Caps:   []string{"cap"},
+		Filter: "blob:limit=1m",
+	})
+}
+
+func TestPackNegoFullBlown(t *testing.T) {
+	buf := new(bytes.Buffer)
+	pktline.WriteString(buf, "want "+oid1+" cap1 cap2")
+	pktline.WriteString(buf, "want "+oid2)
+	pktline.WriteString(buf, "shallow "+oid2)
+	pktline.WriteString(buf, "shallow "+oid1)
+	pktline.WriteString(buf, "deepen 1")
+	pktline.WriteString(buf, "filter blob:none")
+	pktline.WriteFlush(buf)
+	pktline.WriteString(buf, "have "+oid2)
+	pktline.WriteFlush(buf)
+	pktline.WriteString(buf, "have "+oid1)
+	pktline.WriteFlush(buf)
+	pktline.WriteString(buf, "done")
+
+	requireParses(t, buf, PackfileNegotiation{
+		Wants:    []string{oid1, oid2},
+		Haves:    []string{oid2, oid1},
+		Caps:     []string{"cap1", "cap2"},
+		Shallows: []string{oid2, oid1},
+		Deepen:   "deepen 1",
+		Filter:   "blob:none",
+	})
+}
diff --git a/internal/service/smarthttp/scan_deepen.go b/internal/service/smarthttp/scan_deepen.go
deleted file mode 100644
index b6d997b57..000000000
--- a/internal/service/smarthttp/scan_deepen.go
+++ /dev/null
@@ -1,26 +0,0 @@
-package smarthttp
-
-import (
-	"bytes"
-	"io"
-	"io/ioutil"
-
-	"gitlab.com/gitlab-org/gitaly/internal/git/pktline"
-)
-
-func scanDeepen(body io.Reader) bool {
-	result := false
-
-	scanner := pktline.NewScanner(body)
-	for scanner.Scan() {
-		if bytes.HasPrefix(pktline.Data(scanner.Bytes()), []byte("deepen")) && scanner.Err() == nil {
-			result = true
-			break
-		}
-	}
-
-	// Because we are connected to another consumer via an io.Pipe and
-	// io.TeeReader we must consume all data.
-	io.Copy(ioutil.Discard, body)
-	return result
-}
diff --git a/internal/service/smarthttp/scan_deepen_test.go b/internal/service/smarthttp/scan_deepen_test.go
deleted file mode 100644
index 0a5f90ef3..000000000
--- a/internal/service/smarthttp/scan_deepen_test.go
+++ /dev/null
@@ -1,54 +0,0 @@
-package smarthttp
-
-import (
-	"bytes"
-	"fmt"
-	"strings"
-	"testing"
-)
-
-func TestSuccessfulScanDeepen(t *testing.T) {
-	examples := []struct {
-		input  string
-		output bool
-	}{
-		{"000dsomething000cdeepen 10000", true},
-		{"000dsomething0000000cdeepen 1", true}, // 0000 packet
-		{"000dsomething0001000cdeepen 1", true}, // 0001 packet
-		{"000dsomething0002000cdeepen 1", true}, // 0002 packet
-		{"000dsomething0003000cdeepen 1", true}, // 0003 packet
-		{"000dsomething0000000cdeepen 1" + strings.Repeat("garbage", 1000000), true},
-		{"ffff" + strings.Repeat("x", 65531) + "000cdeepen 1", true},
-		{"000dsomething0000", false},
-	}
-
-	for _, example := range examples {
-		t.Run(fmt.Sprintf("%.30s", example.input), func(t *testing.T) {
-			reader := bytes.NewReader([]byte(example.input))
-			hasDeepen := scanDeepen(reader)
-			if n := reader.Len(); n != 0 {
-				t.Fatalf("expected reader to be drained, found %d bytes left", n)
-			}
-
-			if hasDeepen != example.output {
-				t.Fatalf("expected %v, got %v", example.output, hasDeepen)
-			}
-		})
-	}
-}
-
-func TestFailedScanDeepen(t *testing.T) {
-	examples := []string{
-		"invalid data",
-		"deepen",
-		"000cdeepen",
-	}
-
-	for _, example := range examples {
-		t.Run(example, func(t *testing.T) {
-			if scanDeepen(bytes.NewReader([]byte(example))) {
-				t.Fatalf("scanDeepen %q: expected result to be false, got true", example)
-			}
-		})
-	}
-}
diff --git a/internal/service/smarthttp/upload_pack.go b/internal/service/smarthttp/upload_pack.go
index 7ed467281..e43add012 100644
--- a/internal/service/smarthttp/upload_pack.go
+++ b/internal/service/smarthttp/upload_pack.go
@@ -9,6 +9,7 @@ import (
 	"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"
 	"gitlab.com/gitlab-org/gitaly/internal/helper"
 	"gitlab.com/gitlab-org/gitaly/internal/metadata/featureflag"
 	"gitlab.com/gitlab-org/gitaly/internal/service/inspect"
@@ -54,9 +55,13 @@ func (s *server) PostUploadPack(stream gitalypb.SmartHTTPService_PostUploadPackS
 	pr, pw := io.Pipe()
 	defer pw.Close()
 	stdin := io.TeeReader(stdinReader, pw)
-	deepenCh := make(chan bool, 1)
+	statsCh := make(chan stats.PackfileNegotiation, 1)
 	go func() {
-		deepenCh <- scanDeepen(pr)
+		defer close(statsCh)
+		stats := stats.PackfileNegotiation{}
+		if err := stats.Parse(pr); err == nil {
+			statsCh <- stats
+		}
 	}()
 
 	var respBytes int64
@@ -100,8 +105,10 @@ func (s *server) PostUploadPack(stream gitalypb.SmartHTTPService_PostUploadPackS
 	}
 
 	if err := cmd.Wait(); err != nil {
-		pw.Close() // ensure scanDeepen returns
-		if _, ok := command.ExitStatus(err); ok && <-deepenCh {
+		pw.Close() // ensure PackfileNegotiation parser returns
+		stats := <-statsCh
+
+		if _, ok := command.ExitStatus(err); ok && stats.Deepen != "" {
 			// 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.