Distributor profiling (cortexproject#296)

    * Update weaveworks/common

    * Tag images as quay.io/weaveworks, to make dev/test easier

    Also, update circle.yaml for new image names

    * Refactor the distributor http code, add more tracing

    * Optimise Cortex Protobufs

    - Move Prometheus remote protos into cortex so we can own them
    - Compile with gogoprotobuf, to get generated marshall functions
    - Remove pointer from protos
    - Remove intermediate (prometheus) representation in distributor write path
    - Add vendored gogoproto code
    - Treat label names and values as custom byte arrays, to prevent a copy

Author: tomwilkie

Makefile

@@ -3,7 +3,7 @@
 
 # Boiler plate for bulding Docker containers.
 # All this must go at top of file I'm afraid.
-IMAGE_PREFIX := weaveworks/cortex-
+IMAGE_PREFIX := quay.io/weaveworks/cortex-
 IMAGE_TAG := $(shell ./tools/image-tag)
 UPTODATE := .uptodate
 
@@ -78,7 +78,7 @@ $(EXES): build/$(UPTODATE)
 	$(NETGO_CHECK)
 
 %.pb.go: build/$(UPTODATE)
-	protoc -I ./vendor:./$(@D) --go_out=plugins=grpc:./$(@D) ./$(patsubst %.pb.go,%.proto,$@)
+	protoc -I ./vendor:./$(@D) --gogoslick_out=plugins=grpc:./$(@D) ./$(patsubst %.pb.go,%.proto,$@)
 
 lint: build/$(UPTODATE)
 	./tools/lint -notestpackage -ignorespelling queriers -ignorespelling Queriers .

build/Dockerfile

@@ -12,6 +12,8 @@ RUN go get -tags netgo \
 		github.com/client9/misspell/cmd/misspell \
 		github.com/jteeuwen/go-bindata/go-bindata \
 		github.com/golang/protobuf/protoc-gen-go \
+		github.com/gogo/protobuf/protoc-gen-gogoslick \
+		github.com/gogo/protobuf/gogoproto \
 		gopkg.in/mvdan/sh.v1/cmd/shfmt && \
 	rm -rf /go/pkg /go/src
 RUN mkdir protoc && \

circle.yml

@@ -10,7 +10,7 @@ dependencies:
   override:
     - |
         cd build && \
-        ../tools/rebuild-image weaveworks/cortex-build . build.sh Dockerfile && \
+        ../tools/rebuild-image quay.io/weaveworks/cortex-build . build.sh Dockerfile && \
         touch .uptodate
 
 test:

cmd/ingester/main.go

@@ -49,7 +49,7 @@ func main() {
 
 	// Deferring a func to make ordering obvious
 	defer func() {
-		registration.ChangeState(ring.IngesterState_LEAVING)
+		registration.ChangeState(ring.LEAVING)
 		ingester.Stop()
 		registration.Unregister()
 		server.Stop()

cortex.proto

@@ -1,17 +1,24 @@
 syntax = "proto3";
 
-import "github.com/prometheus/prometheus/storage/remote/remote.proto";
-
 package cortex;
 
+import "github.com/gogo/protobuf/gogoproto/gogo.proto";
+
+option (gogoproto.marshaler_all) = true;
+option (gogoproto.unmarshaler_all) = true;
+
 service Ingester {
-  rpc Push(remote.WriteRequest) returns (WriteResponse) {};
+  rpc Push(WriteRequest) returns (WriteResponse) {};
   rpc Query(QueryRequest) returns (QueryResponse) {};
   rpc LabelValues(LabelValuesRequest) returns (LabelValuesResponse) {};
   rpc UserStats(UserStatsRequest) returns (UserStatsResponse) {};
   rpc MetricsForLabelMatchers(MetricsForLabelMatchersRequest) returns (MetricsForLabelMatchersResponse) {};
 }
 
+message WriteRequest {
+  repeated TimeSeries timeseries = 1 [(gogoproto.nullable) = false];
+}
+
 message WriteResponse {}
 
 message QueryRequest {
@@ -21,7 +28,7 @@ message QueryRequest {
 }
 
 message QueryResponse {
-  repeated remote.TimeSeries timeseries = 1;
+  repeated TimeSeries timeseries = 1 [(gogoproto.nullable) = false];
 }
 
 message LabelValuesRequest {
@@ -49,12 +56,29 @@ message MetricsForLabelMatchersResponse {
   repeated Metric metric = 1;
 }
 
+
+message TimeSeries {
+  repeated LabelPair labels = 1 [(gogoproto.nullable) = false];
+  // Sorted by time, oldest sample first.
+  repeated Sample samples   = 2 [(gogoproto.nullable) = false];
+}
+
+message LabelPair {
+  bytes name  = 1 [(gogoproto.customtype) = "github.com/weaveworks/cortex/util/wire.Bytes", (gogoproto.nullable) = false];
+  bytes value = 2 [(gogoproto.customtype) = "github.com/weaveworks/cortex/util/wire.Bytes", (gogoproto.nullable) = false];
+}
+
+message Sample {
+  double value       = 1;
+  int64 timestamp_ms = 2;
+}
+
 message LabelMatchers {
   repeated LabelMatcher matchers = 1;
 }
 
 message Metric {
-  repeated remote.LabelPair labels = 1;
+  repeated LabelPair labels = 1 [(gogoproto.nullable) = false];
 }
 
 enum MatchType {

distributor/distributor.go

@@ -20,7 +20,6 @@ import (
 	"github.com/prometheus/common/log"
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/storage/metric"
-	"github.com/prometheus/prometheus/storage/remote"
 
 	"github.com/weaveworks/common/instrument"
 	"github.com/weaveworks/common/middleware"
@@ -38,6 +37,7 @@ var (
 		"The current number of ingester clients.",
 		nil, nil,
 	)
+	labelNameBytes = []byte(model.MetricNameLabel)
 )
 
 // Distributor is a storage.SampleAppender and a cortex.Querier which
@@ -242,20 +242,25 @@ func (d *Distributor) getClientFor(ingester *ring.IngesterDesc) (cortex.Ingester
 	return client, nil
 }
 
-func tokenForMetric(userID string, metric model.Metric) uint32 {
-	name := metric[model.MetricNameLabel]
-	return tokenFor(userID, name)
+func tokenForLabels(userID string, labels []cortex.LabelPair) (uint32, error) {
+	for _, label := range labels {
+		if label.Name.Equal(labelNameBytes) {
+			return tokenFor(userID, label.Value), nil
+		}
+	}
+	return 0, fmt.Errorf("No metric name label")
 }
 
-func tokenFor(userID string, name model.LabelValue) uint32 {
+func tokenFor(userID string, name []byte) uint32 {
 	h := fnv.New32()
 	h.Write([]byte(userID))
-	h.Write([]byte(name))
+	h.Write(name)
 	return h.Sum32()
 }
 
 type sampleTracker struct {
-	sample      *model.Sample
+	labels      []cortex.LabelPair
+	sample      cortex.Sample
 	minSuccess  int
 	maxFailures int
 	succeeded   int32
@@ -270,13 +275,30 @@ type pushTracker struct {
 }
 
 // Push implements cortex.IngesterServer
-func (d *Distributor) Push(ctx context.Context, req *remote.WriteRequest) (*cortex.WriteResponse, error) {
+func (d *Distributor) Push(ctx context.Context, req *cortex.WriteRequest) (*cortex.WriteResponse, error) {
 	userID, err := user.GetID(ctx)
 	if err != nil {
 		return nil, err
 	}
 
-	samples := util.FromWriteRequest(req)
+	// First we flatten out the request into a list of samples.
+	// We use the heuristic of 1 sample per TS to size the array.
+	// We also work out the hash value at the same time.
+	samples := make([]sampleTracker, 0, len(req.Timeseries))
+	keys := make([]uint32, 0, len(req.Timeseries))
+	for _, ts := range req.Timeseries {
+		key, err := tokenForLabels(userID, ts.Labels)
+		if err != nil {
+			return nil, err
+		}
+		for _, s := range ts.Samples {
+			keys = append(keys, key)
+			samples = append(samples, sampleTracker{
+				labels: ts.Labels,
+				sample: s,
+			})
+		}
+	}
 	d.receivedSamples.Add(float64(len(samples)))
 
 	if len(samples) == 0 {
@@ -288,27 +310,24 @@ func (d *Distributor) Push(ctx context.Context, req *remote.WriteRequest) (*cort
 		return nil, errIngestionRateLimitExceeded
 	}
 
-	keys := make([]uint32, len(samples), len(samples))
-	for i, sample := range samples {
-		keys[i] = tokenForMetric(userID, sample.Metric)
-	}
-
-	ingesters, err := d.ring.BatchGet(keys, d.cfg.ReplicationFactor, ring.Write)
-	if err != nil {
+	var ingesters [][]*ring.IngesterDesc
+	if err := instrument.TimeRequestHistogram(ctx, "Distributor.Push[ring-lookup]", nil, func(ctx context.Context) error {
+		var err error
+		ingesters, err = d.ring.BatchGet(keys, d.cfg.ReplicationFactor, ring.Write)
+		if err != nil {
+			return err
+		}
+		return nil
+	}); err != nil {
 		return nil, err
 	}
 
-	sampleTrackers := make([]sampleTracker, len(samples), len(samples))
 	samplesByIngester := map[*ring.IngesterDesc][]*sampleTracker{}
 	for i := range samples {
 		// We need a response from a quorum of ingesters, which is n/2 + 1.
 		minSuccess := (len(ingesters[i]) / 2) + 1
-
-		sampleTrackers[i] = sampleTracker{
-			sample:      samples[i],
-			minSuccess:  minSuccess,
-			maxFailures: len(ingesters[i]) - minSuccess,
-		}
+		samples[i].minSuccess = minSuccess
+		samples[i].maxFailures = len(ingesters[i]) - minSuccess
 
 		// Skip those that have not heartbeated in a while. NB these are still
 		// included in the calculation of minSuccess, so if too many failed ingesters
@@ -322,14 +341,14 @@ func (d *Distributor) Push(ctx context.Context, req *remote.WriteRequest) (*cort
 
 		// This is just a shortcut - if there are not minSuccess available ingesters,
 		// after filtering out dead ones, don't even bother trying.
-		if len(liveIngesters) < sampleTrackers[i].minSuccess {
+		if len(liveIngesters) < minSuccess {
 			return nil, fmt.Errorf("wanted at least %d live ingesters to process write, had %d",
-				sampleTrackers[i].minSuccess, len(liveIngesters))
+				minSuccess, len(liveIngesters))
 		}
 
 		for _, liveIngester := range liveIngesters {
 			sampleForIngester := samplesByIngester[liveIngester]
-			samplesByIngester[liveIngester] = append(sampleForIngester, &sampleTrackers[i])
+			samplesByIngester[liveIngester] = append(sampleForIngester, &samples[i])
 		}
 	}
 
@@ -395,17 +414,24 @@ func (d *Distributor) sendSamples(ctx context.Context, ingester *ring.IngesterDe
 	}
 }
 
-func (d *Distributor) sendSamplesErr(ctx context.Context, ingester *ring.IngesterDesc, sampleTrackers []*sampleTracker) error {
+func (d *Distributor) sendSamplesErr(ctx context.Context, ingester *ring.IngesterDesc, samples []*sampleTracker) error {
 	client, err := d.getClientFor(ingester)
 	if err != nil {
 		return err
 	}
-	samples := make([]*model.Sample, len(sampleTrackers), len(sampleTrackers))
-	for i := range sampleTrackers {
-		samples[i] = sampleTrackers[i].sample
+
+	req := &cortex.WriteRequest{
+		Timeseries: make([]cortex.TimeSeries, 0, len(samples)),
 	}
+	for _, s := range samples {
+		req.Timeseries = append(req.Timeseries, cortex.TimeSeries{
+			Labels:  s.labels,
+			Samples: []cortex.Sample{s.sample},
+		})
+	}
+
 	err = instrument.TimeRequestHistogram(ctx, "Distributor.sendSamples", d.sendDuration, func(ctx context.Context) error {
-		_, err := client.Push(ctx, util.ToWriteRequest(samples))
+		_, err := client.Push(ctx, req)
 		return err
 	})
 	d.ingesterAppends.WithLabelValues(ingester.Addr).Inc()
@@ -434,7 +460,7 @@ func (d *Distributor) Query(ctx context.Context, from, to model.Time, matchers .
 			return err
 		}
 
-		ingesters, err := d.ring.Get(tokenFor(userID, metricName), d.cfg.ReplicationFactor, ring.Read)
+		ingesters, err := d.ring.Get(tokenFor(userID, []byte(metricName)), d.cfg.ReplicationFactor, ring.Read)
 		if err != nil {
 			return err
 		}

distributor/distributor_test.go

@@ -8,7 +8,6 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/storage/metric"
-	"github.com/prometheus/prometheus/storage/remote"
 	"github.com/stretchr/testify/assert"
 	"golang.org/x/net/context"
 	"google.golang.org/grpc"
@@ -44,7 +43,7 @@ type mockIngester struct {
 	happy bool
 }
 
-func (i mockIngester) Push(ctx context.Context, in *remote.WriteRequest, opts ...grpc.CallOption) (*cortex.WriteResponse, error) {
+func (i mockIngester) Push(ctx context.Context, in *cortex.WriteRequest, opts ...grpc.CallOption) (*cortex.WriteResponse, error) {
 	if !i.happy {
 		return nil, fmt.Errorf("Fail")
 	}
@@ -56,15 +55,15 @@ func (i mockIngester) Query(ctx context.Context, in *cortex.QueryRequest, opts .
 		return nil, fmt.Errorf("Fail")
 	}
 	return &cortex.QueryResponse{
-		Timeseries: []*remote.TimeSeries{
+		Timeseries: []cortex.TimeSeries{
 			{
-				Labels: []*remote.LabelPair{
+				Labels: []cortex.LabelPair{
 					{
-						Name:  "__name__",
-						Value: "foo",
+						Name:  []byte("__name__"),
+						Value: []byte("foo"),
 					},
 				},
-				Samples: []*remote.Sample{
+				Samples: []cortex.Sample{
 					{
 						Value:       0,
 						TimestampMs: 0,
@@ -169,16 +168,16 @@ func TestDistributorPush(t *testing.T) {
 			}
 			defer d.Stop()
 
-			request := &remote.WriteRequest{}
+			request := &cortex.WriteRequest{}
 			for i := 0; i < tc.samples; i++ {
-				ts := &remote.TimeSeries{
-					Labels: []*remote.LabelPair{
-						{"__name__", "foo"},
-						{"bar", "baz"},
-						{"sample", fmt.Sprintf("%d", i)},
+				ts := cortex.TimeSeries{
+					Labels: []cortex.LabelPair{
+						{[]byte("__name__"), []byte("foo")},
+						{[]byte("bar"), []byte("baz")},
+						{[]byte("sample"), []byte(fmt.Sprintf("%d", i))},
 					},
 				}
-				ts.Samples = []*remote.Sample{
+				ts.Samples = []cortex.Sample{
 					{
 						Value:       float64(i),
 						TimestampMs: int64(i),