Review feedback.

Signed-off-by: Tom Wilkie <[email protected]>

Author: tomwilkie

pkg/ingester/ingester.go

@@ -80,7 +80,6 @@ type Config struct {
 	LifecyclerConfig ring.LifecyclerConfig
 
 	// Config for transferring chunks.
-	SearchPendingFor   time.Duration
 	MaxTransferRetries int
 
 	// Config for chunk flushing.
@@ -102,8 +101,7 @@ type Config struct {
 func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	cfg.LifecyclerConfig.RegisterFlags(f)
 
-	f.DurationVar(&cfg.SearchPendingFor, "ingester.search-pending-for", 30*time.Second, "Time to spend searching for a pending ingester when shutting down.")
-	f.IntVar(&cfg.MaxTransferRetries, "ingester.max-transfer-retries", 5, "Number of times to try and transfer chunks before falling back to flushing.")
+	f.IntVar(&cfg.MaxTransferRetries, "ingester.max-transfer-retries", 10, "Number of times to try and transfer chunks before falling back to flushing.")
 	f.DurationVar(&cfg.FlushCheckPeriod, "ingester.flush-period", 1*time.Minute, "Period with which to attempt to flush chunks.")
 	f.DurationVar(&cfg.RetainPeriod, "ingester.retain-period", 5*time.Minute, "Period chunks will remain in memory after flushing.")
 	f.DurationVar(&cfg.FlushOpTimeout, "ingester.flush-op-timeout", 1*time.Minute, "Timeout for individual flush operations.")
@@ -114,6 +112,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.DurationVar(&cfg.RateUpdatePeriod, "ingester.rate-update-period", 15*time.Second, "Period with which to update the per-user ingestion rates.")
 
 	// DEPRECATED, no-op
+	f.Duration("ingester.search-pending-for", 30*time.Second, "DEPRECATED. Time to spend searching for a pending ingester when shutting down.")
 	f.Bool("ingester.reject-old-samples", false, "DEPRECATED. Reject old samples.")
 	f.Duration("ingester.reject-old-samples.max-age", 0, "DEPRECATED. Maximum accepted sample age before rejecting.")
 	f.Int("ingester.validation.max-length-label-name", 0, "DEPRECATED. Maximum length accepted for label names.")

pkg/ingester/lifecycle_test.go

@@ -96,7 +96,6 @@ func TestIngesterTransfer(t *testing.T) {
 	cfg1.LifecyclerConfig.Addr = "ingester1"
 	cfg1.LifecyclerConfig.ClaimOnRollout = true
 	cfg1.LifecyclerConfig.JoinAfter = 0 * time.Second
-	cfg1.SearchPendingFor = 1 * time.Second
 	ing1, err := New(cfg1, defaultClientTestConfig(), limits, nil)
 	require.NoError(t, err)
 
@@ -178,7 +177,6 @@ func TestIngesterBadTransfer(t *testing.T) {
 	cfg.LifecyclerConfig.Addr = "ingester1"
 	cfg.LifecyclerConfig.ClaimOnRollout = true
 	cfg.LifecyclerConfig.JoinAfter = 100 * time.Second
-	cfg.SearchPendingFor = 1 * time.Second
 	ing, err := New(cfg, defaultClientTestConfig(), limits, nil)
 	require.NoError(t, err)
 

pkg/ingester/transfer.go

@@ -20,10 +20,6 @@ import (
 	"github.com/weaveworks/common/user"
 )
 
-const (
-	pendingSearchIterations = 10
-)
-
 var (
 	sentChunks = prometheus.NewCounter(prometheus.CounterOpts{
 		Name: "cortex_ingester_sent_chunks",
@@ -190,7 +186,7 @@ func fromWireChunks(wireChunks []client.Chunk) ([]*desc, error) {
 func (i *Ingester) TransferOut(ctx context.Context) error {
 	backoff := util.NewBackoff(ctx, util.BackoffConfig{
 		MinBackoff: 100 * time.Millisecond,
-		MaxBackoff: 1 * time.Second,
+		MaxBackoff: 5 * time.Second,
 		MaxRetries: i.cfg.MaxTransferRetries,
 	})
 
@@ -279,44 +275,15 @@ func (i *Ingester) transferOut(ctx context.Context) error {
 
 // findTargetIngester finds an ingester in PENDING state.
 func (i *Ingester) findTargetIngester(ctx context.Context) (*ring.IngesterDesc, error) {
-	findIngester := func(ctx context.Context) (*ring.IngesterDesc, error) {
-		ringDesc, err := i.lifecycler.KVStore.Get(ctx, ring.ConsulKey)
-		if err != nil {
-			return nil, err
-		}
-
-		ingesters := ringDesc.(*ring.Desc).FindIngestersByState(ring.PENDING)
-		if len(ingesters) <= 0 {
-			return nil, fmt.Errorf("no pending ingesters")
-		}
-
-		return &ingesters[0], nil
+	ringDesc, err := i.lifecycler.KVStore.Get(ctx, ring.ConsulKey)
+	if err != nil {
+		return nil, err
 	}
 
-	deadline := time.NewTimer(i.cfg.SearchPendingFor)
-	defer deadline.Stop()
-
-	ticker := time.NewTicker(i.cfg.SearchPendingFor / pendingSearchIterations)
-	defer ticker.Stop()
-
-	for {
-		select {
-		case <-ticker.C:
-			ctx, cancel := context.WithTimeout(ctx, i.cfg.SearchPendingFor/pendingSearchIterations)
-			ingester, err := findIngester(ctx)
-			cancel()
-			if err != nil {
-				level.Warn(util.Logger).Log("msg", "Error looking for pending ingester", "err", err)
-				continue
-			}
-			return ingester, nil
-
-		case <-deadline.C:
-			level.Warn(util.Logger).Log("msg", "Could not find pending ingester before deadline")
-			return nil, fmt.Errorf("could not find pending ingester before deadline")
-
-		case <-ctx.Done():
-			return nil, ctx.Err()
-		}
+	ingesters := ringDesc.(*ring.Desc).FindIngestersByState(ring.PENDING)
+	if len(ingesters) <= 0 {
+		return nil, fmt.Errorf("no pending ingesters")
 	}
+
+	return &ingesters[0], nil
 }

pkg/ring/lifecycler.go

@@ -33,7 +33,7 @@ var (
 	shutdownDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
 		Name:    "cortex_shutdown_duration_seconds",
 		Help:    "Duration (in seconds) of cortex shutdown procedure (ie transfer or flush).",
-		Buckets: prometheus.DefBuckets,
+		Buckets: prometheus.ExponentialBuckets(10, 2, 8), // Biggest bucket is 10*2^(9-1) = 2560, or 42 mins.
 	}, []string{"op", "status"})
 )