git: stats: Implement analysis for packfile negotiation

Similar to the reference discovery code, this commit adds a new parser
for the packfile negotiation protocol. It analyses a client stream and
records various information sent by the client like the haves,
capabilities and total number of pktlines sent. This will at a later
stage be used to analyze incoming packfile negotiations and fill in
Prometheus metrics to increase visibility of what features are used by
clients.

2 files changed, 336 insertions, 0 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",
+	})
+}