Merge pull request #119 from weaveworks/remove-put-parallelisation

Remove chunk store parallelisation; I want this controlled by the ingester

Author: tomwilkie

chunk/chunk_store.go

@@ -33,7 +33,6 @@ const (
 
 	// See http://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Limits.html.
 	dynamoMaxBatchSize = 25
-	maxConcurrentPuts  = 100
 )
 
 var (
@@ -129,7 +128,6 @@ func NewAWSStore(cfg StoreConfig) (*AWSStore, error) {
 		chunkCache: cfg.ChunkCache,
 		tableName:  tableName,
 		bucketName: bucketName,
-		putLimiter: NewSemaphore(maxConcurrentPuts),
 	}, nil
 }
 
@@ -155,7 +153,6 @@ type AWSStore struct {
 	chunkCache *Cache
 	tableName  string
 	bucketName string
-	putLimiter Semaphore
 }
 
 type dynamodbClient interface {
@@ -268,10 +265,8 @@ func (c *AWSStore) Put(ctx context.Context, chunks []Chunk) error {
 func (c *AWSStore) putChunks(userID string, chunks []Chunk) error {
 	incomingErrors := make(chan error)
 	for _, chunk := range chunks {
-		c.putLimiter.Acquire()
 		go func(chunk Chunk) {
 			incomingErrors <- c.putChunk(userID, &chunk)
-			c.putLimiter.Release()
 		}(chunk)
 	}
 
@@ -314,25 +309,12 @@ func (c *AWSStore) updateIndex(userID string, chunks []Chunk) error {
 		return err
 	}
 
-	batches := c.batchRequests(writeReqs)
-
-	// Request all the batches in parallel.
-	incomingErrors := make(chan error)
-	for _, batch := range batches {
-		c.putLimiter.Acquire()
-		go func(batch []*dynamodb.WriteRequest) {
-			incomingErrors <- c.batchWriteDynamo(batch)
-			c.putLimiter.Release()
-		}(batch)
-	}
-	var lastErr error
-	for range batches {
-		err = <-incomingErrors
-		if err != nil {
-			lastErr = err
+	for _, batch := range c.batchRequests(writeReqs) {
+		if err := c.batchWriteDynamo(batch); err != nil {
+			return err
 		}
 	}
-	return lastErr
+	return nil
 }
 
 // calculateDynamoWrites creates a set of batched WriteRequests to dynamo for all

chunk/chunk_store_test.go

@@ -49,7 +49,6 @@ func TestChunkStore(t *testing.T) {
 		chunkCache: nil,
 		tableName:  "tablename",
 		bucketName: "bucketname",
-		putLimiter: NoopSemaphore,
 	}
 	store.CreateTables()
 

ingester/ingester.go

@@ -6,6 +6,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/aws/aws-sdk-go/aws/awserr"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/common/log"
 	"github.com/prometheus/common/model"
@@ -25,6 +26,12 @@ const (
 	// Reasons to discard samples.
 	outOfOrderTimestamp = "timestamp_out_of_order"
 	duplicateSample     = "multiple_values_for_timestamp"
+
+	// Backoff for flush
+	minBackoff = 100 * time.Millisecond
+	maxBackoff = 1 * time.Second
+
+	provisionedThroughputExceededException = "ProvisionedThroughputExceededException"
 )
 
 var (
@@ -525,6 +532,8 @@ func (i *Ingester) flushSeries(u *userState, fp model.Fingerprint, series *memor
 }
 
 func (i *Ingester) flushLoop(j int) {
+	backoff := minBackoff
+
 	defer func() {
 		log.Info("Ingester.flushLoop() exited")
 		i.done.Done()
@@ -570,12 +579,23 @@ func (i *Ingester) flushLoop(j int) {
 		}
 
 		// flush the chunks without locking the series
-		if err := i.flushChunks(ctx, op.fp, series.metric, chunks); err != nil {
+		err := i.flushChunks(ctx, op.fp, series.metric, chunks)
+		if err != nil {
+			if awsErr, ok := err.(awserr.Error); ok && awsErr.Code() == provisionedThroughputExceededException {
+				time.Sleep(backoff)
+				backoff = backoff * 2
+				if backoff > maxBackoff {
+					backoff = maxBackoff
+				}
+			}
+
 			log.Errorf("Failed to flush chunks: %v", err)
 			i.chunkStoreFailures.Add(float64(len(chunks)))
 			continue
 		}
 
+		backoff = minBackoff
+
 		// now remove the chunks
 		userState.fpLocker.Lock(op.fp)
 		series.chunkDescs = series.chunkDescs[len(chunks):]