Fix data race in ingester (#2327)

pkg/ingester/flush.go

@@ -3,6 +3,7 @@ package ingester
 import (
 	"fmt"
 	"net/http"
+	"sync"
 	"time"
 
 	"golang.org/x/net/context"
@@ -221,7 +222,7 @@ func (i *Ingester) flushUserSeries(userID string, fp model.Fingerprint, immediat
 	ctx := user.InjectOrgID(context.Background(), userID)
 	ctx, cancel := context.WithTimeout(ctx, i.cfg.FlushOpTimeout)
 	defer cancel()
-	err := i.flushChunks(ctx, fp, labels, chunks)
+	err := i.flushChunks(ctx, fp, labels, chunks, &instance.streamsMtx)
 	if err != nil {
 		return err
 	}
@@ -306,7 +307,7 @@ func (i *Ingester) removeFlushedChunks(instance *instance, stream *stream) {
 	}
 }
 
-func (i *Ingester) flushChunks(ctx context.Context, fp model.Fingerprint, labelPairs labels.Labels, cs []*chunkDesc) error {
+func (i *Ingester) flushChunks(ctx context.Context, fp model.Fingerprint, labelPairs labels.Labels, cs []*chunkDesc, streamsMtx *sync.RWMutex) error {
 	userID, err := user.ExtractOrgID(ctx)
 	if err != nil {
 		return err
@@ -327,7 +328,10 @@ func (i *Ingester) flushChunks(ctx context.Context, fp model.Fingerprint, labelP
 		)
 
 		start := time.Now()
-		if err := c.Encode(); err != nil {
+		streamsMtx.Lock()
+		err := c.Encode()
+		streamsMtx.Unlock()
+		if err != nil {
 			return err
 		}
 		chunkEncodeTime.Observe(time.Since(start).Seconds())

pkg/ingester/transfer_test.go

@@ -88,6 +88,7 @@ func TestTransferOut(t *testing.T) {
 		lines := []string{}
 
 		// Get all the lines back and make sure the blocks transferred successfully
+		ing2.instances["test"].streamsMtx.RLock()
 		for _, stream := range ing2.instances["test"].streams {
 			it, err := stream.Iterator(
 				context.TODO(),
@@ -105,7 +106,7 @@ func TestTransferOut(t *testing.T) {
 				lines = append(lines, entry.Line)
 			}
 		}
-
+		ing2.instances["test"].streamsMtx.RUnlock()
 		sort.Strings(lines)
 
 		assert.Equal(