Shard ingest flush queue by username and metric name, to match the underlying sharding of the index. (cortexproject#271)

Author: tomwilkie

chunk/chunk_store.go

@@ -190,7 +190,7 @@ func (c *AWSStore) updateIndex(ctx context.Context, userID string, chunks []Chun
 func (c *AWSStore) calculateDynamoWrites(userID string, chunks []Chunk) (map[string][]*dynamodb.WriteRequest, error) {
 	writeReqs := map[string][]*dynamodb.WriteRequest{}
 	for _, chunk := range chunks {
-		metricName, err := extractMetricNameFromMetric(chunk.Metric)
+		metricName, err := util.ExtractMetricNameFromMetric(chunk.Metric)
 		if err != nil {
 			return nil, err
 		}
@@ -474,15 +474,6 @@ func (c *AWSStore) fetchChunkData(ctx context.Context, userID string, chunkSet [
 	return chunks, nil
 }
 
-func extractMetricNameFromMetric(m model.Metric) (model.LabelValue, error) {
-	for name, value := range m {
-		if name == model.MetricNameLabel {
-			return value, nil
-		}
-	}
-	return "", fmt.Errorf("no MetricNameLabel for chunk")
-}
-
 func extractMetricNameFromMatchers(matchers []*metric.LabelMatcher) (model.LabelValue, []*metric.LabelMatcher, error) {
 	for i, matcher := range matchers {
 		if matcher.Name != model.MetricNameLabel {

ingester/ingester.go

@@ -3,6 +3,7 @@ package ingester
 import (
 	"flag"
 	"fmt"
+	"hash/fnv"
 	"net/http"
 	"sync"
 	"time"
@@ -439,7 +440,9 @@ func (i *Ingester) sweepUsers(immediate bool) {
 	for id, state := range i.userStates.cp() {
 		for pair := range state.fpToSeries.iter() {
 			state.fpLocker.Lock(pair.fp)
-			i.sweepSeries(id, pair.fp, pair.series, immediate)
+			if err := i.sweepSeries(id, pair.fp, pair.series, immediate); err != nil {
+				log.Errorf("Error sweeping series: %v", err)
+			}
 			state.fpLocker.Unlock(pair.fp)
 		}
 	}
@@ -449,18 +452,33 @@ func (i *Ingester) sweepUsers(immediate bool) {
 //
 // NB we don't close the head chunk here, as the series could wait in the queue
 // for some time, and we want to encourage chunks to be as full as possible.
-func (i *Ingester) sweepSeries(userID string, fp model.Fingerprint, series *memorySeries, immediate bool) {
-	if len(series.chunkDescs) <= 0 {
-		return
+func (i *Ingester) sweepSeries(userID string, fp model.Fingerprint, series *memorySeries, immediate bool) error {
+	flush := i.shouldFlushSeries(series, immediate)
+	if !flush {
+		return nil
 	}
 
-	firstTime := series.firstTime()
-	flush := i.shouldFlushSeries(series, immediate)
+	return i.enqueueSeriesForFlushing(userID, fp, series, immediate)
+}
 
-	if flush {
-		flushQueueIndex := int(uint64(fp) % uint64(i.cfg.ConcurrentFlushes))
-		i.flushQueues[flushQueueIndex].Enqueue(&flushOp{firstTime, userID, fp, immediate})
+func (i *Ingester) enqueueSeriesForFlushing(userID string, fp model.Fingerprint, series *memorySeries, immediate bool) error {
+	metricName, err := util.ExtractMetricNameFromMetric(series.metric)
+	if err != nil {
+		return err
 	}
+
+	h := fnv.New32()
+	if _, err := h.Write([]byte(userID)); err != nil {
+		return err
+	}
+	if _, err := h.Write([]byte(metricName)); err != nil {
+		return err
+	}
+
+	flushQueueIndex := int(h.Sum32() % uint32(i.cfg.ConcurrentFlushes))
+	firstTime := series.firstTime()
+	i.flushQueues[flushQueueIndex].Enqueue(&flushOp{firstTime, userID, fp, immediate})
+	return nil
 }
 
 func (i *Ingester) shouldFlushSeries(series *memorySeries, immediate bool) bool {

util/matchers.go

@@ -1,6 +1,9 @@
 package util
 
 import (
+	"fmt"
+
+	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/storage/metric"
 )
 
@@ -15,3 +18,13 @@ func SplitFiltersAndMatchers(allMatchers []*metric.LabelMatcher) (filters, match
 	}
 	return
 }
+
+// ExtractMetricNameFromMetric extract the metric name from a model.Metric
+func ExtractMetricNameFromMetric(m model.Metric) (model.LabelValue, error) {
+	for name, value := range m {
+		if name == model.MetricNameLabel {
+			return value, nil
+		}
+	}
+	return "", fmt.Errorf("no MetricNameLabel for chunk")
+}