address some review feedback

Signed-off-by: Robert Fratto <[email protected]>

Author: rfratto

docs/configuration/config-file-reference.md

@@ -545,18 +545,9 @@ token_checker:
 # CLI flag: -ingester.spread-flushes
 [spreadflushes: <boolean> | default = false]
 
-# Check that newly created streams fall within expected token ranges
-# CLI flag: -ingester.check-token-on-create
-[check_token_on_create: <boolean> | default = false]
-
-# Check that existing streams appended to fall within expected token ranges
-# CLI flag: -ingester.check-token-on-append
-[check_token_on_append: <boolean> | default = false]
-
-# Check that streams transferred in using the transfer mechanism fall within
-# expected token ranges
-# CLI flag: -ingester.check-token-on-transfer
-[check_token_on_transfer: <boolean> | default = false]
+# Check tokens for streams that are created or appended to.
+# CLI flag: -ingester.check-tokens
+[check_tokens: <boolean> | default = false]
 
 # Period with which to update the per-user ingestion rates.
 # CLI flag: -ingester.rate-update-period

pkg/ingester/client/cortex.pb.go

@@ -29,9 +29,9 @@ service Ingester {
   // TransferTSDB transfers all files of a tsdb to a joining ingester
   rpc TransferTSDB(stream TimeSeriesFile) returns (TransferTSDBResponse) {};
 
-  // TransferChunksSubset is invoked by a leaving ingester (client), streaming
+  // AcceptChunksSubset is invoked by a leaving ingester (client), streaming
   // a subset of its chunks directly to an existing ingester.
-  rpc TransferChunksSubset(stream TimeSeriesChunk) returns (TransferChunksResponse) {};
+  rpc AcceptChunksSubset(stream TimeSeriesChunk) returns (TransferChunksResponse) {};
 
   // GetChunksSubset is invoked by a joining ingester (client). A subset of the
   // serving ingester's chunks will be sent as a stream.

pkg/ingester/client/cortex.proto

@@ -10,23 +10,20 @@ import (
 	"github.com/go-kit/kit/log/level"
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/weaveworks/common/user"
 	"golang.org/x/net/context"
 )
 
 var (
-	blockedRanges = prometheus.NewGauge(prometheus.GaugeOpts{
+	blockedRanges = promauto.NewGauge(prometheus.GaugeOpts{
 		Name: "cortex_ingester_blocked_ranges",
 		Help: "The current number of ranges that will not accept writes by this ingester.",
 	})
 )
 
-func init() {
-	prometheus.MustRegister(blockedRanges)
-}
-
-// TransferChunksSubset accepts chunks from a client and moves them into the local Ingester.
-func (i *Ingester) TransferChunksSubset(stream client.Ingester_TransferChunksSubsetServer) error {
+// AcceptChunksSubset accepts chunks from a client and moves them into the local Ingester.
+func (i *Ingester) AcceptChunksSubset(stream client.Ingester_AcceptChunksSubsetServer) error {
 	i.userStatesMtx.Lock()
 	defer i.userStatesMtx.Unlock()
 
@@ -90,8 +87,8 @@ func (i *Ingester) GetChunksSubset(req *client.GetChunksRequest, stream client.I
 // sitting around forever if a joining ingester crashes, as writes will continue
 // to go to us and get rejected for as long as the blocked range exists.
 func (i *Ingester) BlockRanges(ranges []ring.TokenRange) {
-	i.blockedTokenMtx.Lock()
-	defer i.blockedTokenMtx.Unlock()
+	i.blockedRangesMtx.Lock()
+	defer i.blockedRangesMtx.Unlock()
 
 	for _, rg := range ranges {
 		if exist := i.blockedRanges[rg]; exist {
@@ -112,8 +109,8 @@ func (i *Ingester) BlockRanges(ranges []ring.TokenRange) {
 
 // UnblockRanges manually removes blocks for the provided ranges.
 func (i *Ingester) UnblockRanges(ctx context.Context, in *client.UnblockRangesRequest) (*client.UnblockRangesResponse, error) {
-	i.blockedTokenMtx.Lock()
-	defer i.blockedTokenMtx.Unlock()
+	i.blockedRangesMtx.Lock()
+	defer i.blockedRangesMtx.Unlock()
 
 	for _, rg := range in.Ranges {
 		if exist := i.blockedRanges[rg]; !exist {
@@ -146,8 +143,8 @@ func (i *Ingester) SendChunkRanges(ctx context.Context, ranges []ring.TokenRange
 	}
 	defer c.Close()
 
-	ctx = user.InjectOrgID(ctx, fakeOrgID)
-	stream, err := c.TransferChunksSubset(ctx)
+	ctx = user.InjectOrgID(ctx, noOrgID)
+	stream, err := c.AcceptChunksSubset(ctx)
 	if err != nil {
 		return errors.Wrap(err, "SendChunks")
 	}
@@ -186,7 +183,7 @@ func (i *Ingester) RequestChunkRanges(ctx context.Context, ranges []ring.TokenRa
 	}
 	defer c.Close()
 
-	ctx = user.InjectOrgID(ctx, fakeOrgID)
+	ctx = user.InjectOrgID(ctx, noOrgID)
 	stream, err := c.GetChunksSubset(ctx, &client.GetChunksRequest{
 		Ranges:         ranges,
 		Move:           move,
@@ -220,7 +217,7 @@ func (i *Ingester) RequestComplete(ctx context.Context, ranges []ring.TokenRange
 	}
 	defer c.Close()
 
-	ctx = user.InjectOrgID(ctx, fakeOrgID)
+	ctx = user.InjectOrgID(ctx, noOrgID)
 	_, err = c.UnblockRanges(ctx, &client.UnblockRangesRequest{Ranges: ranges})
 	if err != nil {
 		level.Error(util.Logger).Log("msg", "could not clean up target after transfer", "err", err)

pkg/ingester/incremental_transfer.go

@@ -63,9 +63,7 @@ type Config struct {
 	SpreadFlushes     bool
 
 	// Config for checking tokens.
-	CheckOnCreate   bool `yaml:"check_token_on_create,omitempty"`
-	CheckOnAppend   bool `yaml:"check_token_on_append,omitempty"`
-	CheckOnTransfer bool `yaml:"check_token_on_transfer,omitempty"`
+	CheckTokens bool `yaml:"check_tokens,omitempty"`
 
 	RateUpdatePeriod time.Duration
 
@@ -98,9 +96,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.DurationVar(&cfg.ChunkAgeJitter, "ingester.chunk-age-jitter", 20*time.Minute, "Range of time to subtract from MaxChunkAge to spread out flushes")
 	f.BoolVar(&cfg.SpreadFlushes, "ingester.spread-flushes", false, "If true, spread series flushes across the whole period of MaxChunkAge")
 	f.IntVar(&cfg.ConcurrentFlushes, "ingester.concurrent-flushes", 50, "Number of concurrent goroutines flushing to dynamodb.")
-	f.BoolVar(&cfg.CheckOnCreate, "ingester.check-token-on-create", false, "Check that newly created streams fall within expected token ranges")
-	f.BoolVar(&cfg.CheckOnAppend, "ingester.check-token-on-append", false, "Check that existing streams appended to fall within expected token ranges")
-	f.BoolVar(&cfg.CheckOnTransfer, "ingester.check-token-on-transfer", false, "Check that streams transferred in using the transfer mechanism fall within expected token ranges")
+	f.BoolVar(&cfg.CheckTokens, "ingester.check-tokens", false, "Check tokens for streams that are created or appended to.")
 	f.DurationVar(&cfg.RateUpdatePeriod, "ingester.rate-update-period", 15*time.Second, "Period with which to update the per-user ingestion rates.")
 }
 
@@ -134,8 +130,8 @@ type Ingester struct {
 	wal WAL
 
 	// Stops specific appends
-	blockedTokenMtx sync.RWMutex
-	blockedRanges   map[ring.TokenRange]bool
+	blockedRangesMtx sync.RWMutex
+	blockedRanges    map[ring.TokenRange]bool
 
 	// Hook for injecting behaviour from tests.
 	preFlushUserSeries func()
@@ -299,21 +295,22 @@ func (i *Ingester) StopIncomingRequests() {
 		i.userStatesMtx.Lock()
 		defer i.userStatesMtx.Unlock()
 		i.stopped = true
-		return
 	}
 
-	// When we are incrementally transferring tokens, we want to wait
-	// for there to be no blocked ranges on our local ingester.
-	for {
-		i.blockedTokenMtx.RLock()
-		numBlocked := len(i.blockedRanges)
-		i.blockedTokenMtx.RUnlock()
+	if i.cfg.LifecyclerConfig.LeaveIncrementalTransfer {
+		// When we are incrementally transferring tokens, we want to wait
+		// for there to be no blocked ranges on our local ingester.
+		for {
+			i.blockedRangesMtx.RLock()
+			numBlocked := len(i.blockedRanges)
+			i.blockedRangesMtx.RUnlock()
 
-		if numBlocked == 0 {
-			return
-		}
+			if numBlocked == 0 {
+				break
+			}
 
-		time.Sleep(time.Millisecond * 250)
+			time.Sleep(time.Millisecond * 250)
+		}
 	}
 }
 
@@ -379,8 +376,8 @@ func (i *Ingester) Push(ctx context.Context, req *client.WriteRequest) (*client.
 
 // isTokenBlocked checks to see if a token is in a blocked range.
 func (i *Ingester) isTokenBlocked(token uint32) error {
-	i.blockedTokenMtx.RLock()
-	defer i.blockedTokenMtx.RUnlock()
+	i.blockedRangesMtx.RLock()
+	defer i.blockedRangesMtx.RUnlock()
 
 	for rg := range i.blockedRanges {
 		if rg.Contains(token) {
@@ -429,12 +426,12 @@ func (i *Ingester) append(ctx context.Context, userID string, token uint32, labe
 		return err
 	}
 
-	if sstate == seriesCreated && i.cfg.CheckOnCreate {
+	if sstate == seriesCreated && i.cfg.CheckTokens {
 		if ok := i.tokenChecker.TokenExpected(token); !ok {
 			level.Debug(util.Logger).Log("msg", "unexpected stream created in ingester", "token", token)
 			i.metrics.unexpectedSeriesTotal.WithLabelValues("create").Inc()
 		}
-	} else if i.cfg.CheckOnAppend {
+	} else if i.cfg.CheckTokens {
 		if ok := i.tokenChecker.TokenExpected(token); !ok {
 			level.Debug(util.Logger).Log("msg", "unexpected stream appended in ingester", "token", token)
 			i.metrics.unexpectedSeriesTotal.WithLabelValues("append").Inc()

pkg/ingester/ingester.go

@@ -26,7 +26,7 @@ import (
 )
 
 var (
-	fakeOrgID = "-1"
+	noOrgID = "-1"
 
 	sentChunks = prometheus.NewCounter(prometheus.CounterOpts{
 		Name: "cortex_ingester_sent_chunks",
@@ -123,7 +123,7 @@ func (i *Ingester) acceptChunksFromStream(opts acceptChunksOptions) (fromIngeste
 		}
 		prevNumChunks := len(series.chunkDescs)
 
-		if i.cfg.CheckOnTransfer && !i.tokenChecker.TokenExpected(wireSeries.Token) {
+		if i.cfg.CheckTokens && !i.tokenChecker.TokenExpected(wireSeries.Token) {
 			level.Debug(util.Logger).Log("msg", "unexpected stream transferred to ingester", "token", wireSeries.Token)
 			i.metrics.unexpectedSeriesTotal.WithLabelValues("transfer").Inc()
 		}
@@ -577,7 +577,7 @@ func (i *Ingester) transferOut(ctx context.Context) error {
 	}
 	defer c.Close()
 
-	ctx = user.InjectOrgID(ctx, fakeOrgID)
+	ctx = user.InjectOrgID(ctx, noOrgID)
 	stream, err := c.TransferChunks(ctx)
 	if err != nil {
 		return errors.Wrap(err, "TransferChunks")
@@ -672,7 +672,7 @@ func (i *Ingester) v2TransferOut(ctx context.Context) error {
 	}
 	defer c.Close()
 
-	ctx = user.InjectOrgID(ctx, fakeOrgID)
+	ctx = user.InjectOrgID(ctx, noOrgID)
 	stream, err := c.TransferTSDB(ctx)
 	if err != nil {
 		return errors.Wrap(err, "TransferTSDB() has failed")

pkg/ingester/transfer.go

@@ -630,7 +630,7 @@ func (s *testSendChunksServer) SendAndClose(resp *client.TransferChunksResponse)
 	return nil
 }
 
-func (c *testIngesterClient) TransferChunksSubset(ctx context.Context, opts ...grpc.CallOption) (client.Ingester_TransferChunksSubsetClient, error) {
+func (c *testIngesterClient) AcceptChunksSubset(ctx context.Context, opts ...grpc.CallOption) (client.Ingester_AcceptChunksSubsetClient, error) {
 	ch := make(chan *client.TimeSeriesChunk)
 	resp := make(chan *client.TransferChunksResponse)
 
@@ -645,7 +645,7 @@ func (c *testIngesterClient) TransferChunksSubset(ctx context.Context, opts ...g
 	}
 
 	go func() {
-		_ = c.i.TransferChunksSubset(&srv)
+		_ = c.i.AcceptChunksSubset(&srv)
 	}()
 
 	return &cli, nil

pkg/ingester/transfer_test.go

@@ -471,7 +471,7 @@ func (i *Lifecycler) loop() {
 		}
 
 		if err := i.autoJoin(context.Background(), JOINING); err != nil {
-			level.Error(util.Logger).Log("msg", "failed to pick tokens in consul", "err", err)
+			level.Error(util.Logger).Log("msg", "failed to pick tokens in KV store", "err", err)
 			os.Exit(1)
 		}
 

pkg/ring/lifecycler.go

@@ -154,15 +154,26 @@ func (d *Desc) TokensFor(id string) (tokens, other []uint32) {
 func (i *IngesterDesc) IsHealthyState(op Operation) bool {
 	healthy := false
 
-	switch op {
-	case Write:
-		healthy = (i.State == ACTIVE)
+	if i.Incremental {
+		// Incremental transferring ingester: everything is healthy except
+		// writing or reading to a PENDING ingester.
+		switch op {
+		case Write:
+			healthy = (i.State != PENDING)
+		default:
+			healthy = true
+		}
+	} else {
+		switch op {
+		case Write:
+			healthy = (i.State == ACTIVE)
 
-	case Read:
-		healthy = (i.State == ACTIVE) || (i.State == LEAVING) || (i.State == PENDING)
+		case Read:
+			healthy = (i.State == ACTIVE) || (i.State == LEAVING) || (i.State == PENDING)
 
-	case Reporting:
-		healthy = true
+		case Reporting:
+			healthy = true
+		}
 	}
 
 	return healthy

pkg/ring/model.go

@@ -215,6 +215,22 @@ func (r *Ring) Get(key uint32, op Operation, buf []IngesterDesc) (ReplicationSet
 		// is increased for the key. Dead ingesters will be filtered later by
 		// replication_strategy.go. Filtering later means that we can calculate
 		// a healthiness quorum.
+		if !ingester.Incremental {
+			if op == Write && ingester.State != ACTIVE {
+				n++
+			} else if op == Read && (ingester.State != ACTIVE && ingester.State != LEAVING) {
+				n++
+			}
+		} else {
+			// If the ingester is incrementally transferring tokens, its current
+			// state is just informational and is used to inform users what phase
+			// the transfer is in. Incremental transfers only disallow writing in
+			// the PENDING state.
+			if op == Write && ingester.State == PENDING {
+				n++
+			}
+		}
+
 		if !ingester.IsHealthyState(op) {
 			n++
 		}