[ingesters] Send heartbeat during wall replay (#4847)

* WIP

Signed-off-by: Alan Protasio <approtas@amazon.com>

* changelog

Signed-off-by: Alan Protasio <approtas@amazon.com>

* joing should not block

Signed-off-by: Alan Protasio <approtas@amazon.com>

* commments

Signed-off-by: Alan Protasio <approtas@amazon.com>

Signed-off-by: Alan Protasio <approtas@amazon.com>

Author: alanprot

CHANGELOG.md

@@ -43,6 +43,7 @@
 * [ENHANCEMENT] Querier/Ruler: Retry store-gateway in case of unexpected failure, instead of failing the query. #4532 #4839
 * [ENHANCEMENT] Ring: DoBatch prioritize 4xx errors when failing. #4783
 * [ENHANCEMENT] Cortex now built with Go 1.18. #4829
+* [ENHANCEMENT] Ingester: Prevent ingesters to become unhealthy during wall replay. #4847
 * [FEATURE] Compactor: Added `-compactor.block-files-concurrency` allowing to configure number of go routines for download/upload block files during compaction. #4784
 * [FEATURE] Compactor: Added -compactor.blocks-fetch-concurrency` allowing to configure number of go routines for blocks during compaction. #4787
 * [FEATURE] Compactor: Added configurations for Azure MSI in blocks-storage, ruler-storage and alertmanager-storage. #4818

pkg/compactor/compactor.go

@@ -476,7 +476,7 @@ func (c *Compactor) starting(ctx context.Context) error {
 	// Initialize the compactors ring if sharding is enabled.
 	if c.compactorCfg.ShardingEnabled {
 		lifecyclerCfg := c.compactorCfg.ShardingRing.ToLifecyclerConfig()
-		c.ringLifecycler, err = ring.NewLifecycler(lifecyclerCfg, ring.NewNoopFlushTransferer(), "compactor", ringKey, false, c.logger, prometheus.WrapRegistererWithPrefix("cortex_", c.registerer))
+		c.ringLifecycler, err = ring.NewLifecycler(lifecyclerCfg, ring.NewNoopFlushTransferer(), "compactor", ringKey, true, false, c.logger, prometheus.WrapRegistererWithPrefix("cortex_", c.registerer))
 		if err != nil {
 			return errors.Wrap(err, "unable to initialize compactor ring lifecycler")
 		}

pkg/distributor/distributor.go

@@ -216,7 +216,7 @@ func New(cfg Config, clientConfig ingester_client.Config, limits *validation.Ove
 	if !canJoinDistributorsRing {
 		ingestionRateStrategy = newInfiniteIngestionRateStrategy()
 	} else if limits.IngestionRateStrategy() == validation.GlobalIngestionRateStrategy {
-		distributorsLifeCycler, err = ring.NewLifecycler(cfg.DistributorRing.ToLifecyclerConfig(), nil, "distributor", ringKey, true, log, prometheus.WrapRegistererWithPrefix("cortex_", reg))
+		distributorsLifeCycler, err = ring.NewLifecycler(cfg.DistributorRing.ToLifecyclerConfig(), nil, "distributor", ringKey, true, true, log, prometheus.WrapRegistererWithPrefix("cortex_", reg))
 		if err != nil {
 			return nil, err
 		}

pkg/ingester/ingester.go

@@ -638,7 +638,7 @@ func New(cfg Config, limits *validation.Overrides, registerer prometheus.Registe
 		}, i.getOldestUnshippedBlockMetric)
 	}
 
-	i.lifecycler, err = ring.NewLifecycler(cfg.LifecyclerConfig, i, "ingester", RingKey, cfg.BlocksStorageConfig.TSDB.FlushBlocksOnShutdown, logger, prometheus.WrapRegistererWithPrefix("cortex_", registerer))
+	i.lifecycler, err = ring.NewLifecycler(cfg.LifecyclerConfig, i, "ingester", RingKey, false, cfg.BlocksStorageConfig.TSDB.FlushBlocksOnShutdown, logger, prometheus.WrapRegistererWithPrefix("cortex_", registerer))
 	if err != nil {
 		return nil, err
 	}
@@ -706,13 +706,6 @@ func (i *Ingester) startingV2ForFlusher(ctx context.Context) error {
 }
 
 func (i *Ingester) starting(ctx context.Context) error {
-	if err := i.openExistingTSDB(ctx); err != nil {
-		// Try to rollback and close opened TSDBs before halting the ingester.
-		i.closeAllTSDB()
-
-		return errors.Wrap(err, "opening existing TSDBs")
-	}
-
 	// Important: we want to keep lifecycler running until we ask it to stop, so we need to give it independent context
 	if err := i.lifecycler.StartAsync(context.Background()); err != nil {
 		return errors.Wrap(err, "failed to start lifecycler")
@@ -721,6 +714,15 @@ func (i *Ingester) starting(ctx context.Context) error {
 		return errors.Wrap(err, "failed to start lifecycler")
 	}
 
+	if err := i.openExistingTSDB(ctx); err != nil {
+		// Try to rollback and close opened TSDBs before halting the ingester.
+		i.closeAllTSDB()
+
+		return errors.Wrap(err, "opening existing TSDBs")
+	}
+
+	i.lifecycler.Join()
+
 	// let's start the rest of subservices via manager
 	servs := []services.Service(nil)
 

pkg/ring/lifecycler.go

@@ -93,7 +93,8 @@ type Lifecycler struct {
 	flushTransferer FlushTransferer
 	KVStore         kv.Client
 
-	actorChan chan func()
+	actorChan    chan func()
+	autojoinChan chan struct{}
 
 	// These values are initialised at startup, and never change
 	ID       string
@@ -106,6 +107,9 @@ type Lifecycler struct {
 	flushOnShutdown      *atomic.Bool
 	unregisterOnShutdown *atomic.Bool
 
+	// Whether to auto join on ring on startup. If set to false, Join should be called.
+	autoJoinOnStartup bool
+
 	// We need to remember the ingester state, tokens and registered timestamp just in case the KV store
 	// goes away and comes back empty. The state changes during lifecycle of instance.
 	stateMtx     sync.RWMutex
@@ -128,7 +132,14 @@ type Lifecycler struct {
 }
 
 // NewLifecycler creates new Lifecycler. It must be started via StartAsync.
-func NewLifecycler(cfg LifecyclerConfig, flushTransferer FlushTransferer, ringName, ringKey string, flushOnShutdown bool, logger log.Logger, reg prometheus.Registerer) (*Lifecycler, error) {
+func NewLifecycler(
+	cfg LifecyclerConfig,
+	flushTransferer FlushTransferer,
+	ringName, ringKey string,
+	autoJoinOnStartup, flushOnShutdown bool,
+	logger log.Logger,
+	reg prometheus.Registerer,
+) (*Lifecycler, error) {
 	addr, err := GetInstanceAddr(cfg.Addr, cfg.InfNames, logger)
 	if err != nil {
 		return nil, err
@@ -165,10 +176,12 @@ func NewLifecycler(cfg LifecyclerConfig, flushTransferer FlushTransferer, ringNa
 		ID:                   cfg.ID,
 		RingName:             ringName,
 		RingKey:              ringKey,
+		autoJoinOnStartup:    autoJoinOnStartup,
 		flushOnShutdown:      atomic.NewBool(flushOnShutdown),
 		unregisterOnShutdown: atomic.NewBool(cfg.UnregisterOnShutdown),
 		Zone:                 zone,
 		actorChan:            make(chan func()),
+		autojoinChan:         make(chan struct{}, 1),
 		state:                PENDING,
 		lifecyclerMetrics:    NewLifecyclerMetrics(ringName, reg),
 		logger:               logger,
@@ -391,23 +404,43 @@ func (i *Lifecycler) ZonesCount() int {
 	return i.zonesCount
 }
 
+// Join trigger the instance to join the ring, if autoJoinOnStartup is set to false.
+func (i *Lifecycler) Join() {
+	select {
+	case i.autojoinChan <- struct{}{}:
+	default:
+		level.Warn(i.logger).Log("msg", "join was called more than one time", "ring", i.RingName)
+	}
+}
+
 func (i *Lifecycler) loop(ctx context.Context) error {
+	joined := false
 	// First, see if we exist in the cluster, update our state to match if we do,
 	// and add ourselves (without tokens) if we don't.
 	if err := i.initRing(context.Background()); err != nil {
 		return perrors.Wrapf(err, "failed to join the ring %s", i.RingName)
 	}
 
 	// We do various period tasks
-	autoJoinAfter := time.After(i.cfg.JoinAfter)
+	var autoJoinAfter <-chan time.Time
 	var observeChan <-chan time.Time
 
+	if i.autoJoinOnStartup {
+		autoJoinAfter = time.After(i.cfg.JoinAfter)
+	}
+
 	heartbeatTickerStop, heartbeatTickerChan := newDisableableTicker(i.cfg.HeartbeatPeriod)
 	defer heartbeatTickerStop()
 
 	for {
 		select {
+		case <-i.autojoinChan:
+			autoJoinAfter = time.After(i.cfg.JoinAfter)
 		case <-autoJoinAfter:
+			if joined {
+				continue
+			}
+			joined = true
 			level.Debug(i.logger).Log("msg", "JoinAfter expired", "ring", i.RingName)
 			// Will only fire once, after auto join timeout.  If we haven't entered "JOINING" state,
 			// then pick some tokens and enter ACTIVE state.
@@ -556,7 +589,7 @@ func (i *Lifecycler) initRing(ctx context.Context) error {
 			// We use the tokens from the file only if it does not exist in the ring yet.
 			if len(tokensFromFile) > 0 {
 				level.Info(i.logger).Log("msg", "adding tokens from file", "num_tokens", len(tokensFromFile))
-				if len(tokensFromFile) >= i.cfg.NumTokens {
+				if len(tokensFromFile) >= i.cfg.NumTokens && i.autoJoinOnStartup {
 					i.setState(ACTIVE)
 				}
 				ringDesc.AddIngester(i.ID, i.Addr, i.Zone, tokensFromFile, i.GetState(), registeredAt)
@@ -587,9 +620,14 @@ func (i *Lifecycler) initRing(ctx context.Context) error {
 
 		// If the ingester failed to clean its ring entry up in can leave its state in LEAVING
 		// OR unregister_on_shutdown=false
-		// Move it into ACTIVE to ensure the ingester joins the ring.
+		// if autoJoinOnStartup, move it into ACTIVE to ensure the ingester joins the ring.
+		// else set to PENDING
 		if instanceDesc.State == LEAVING && len(instanceDesc.Tokens) == i.cfg.NumTokens {
-			instanceDesc.State = ACTIVE
+			if i.autoJoinOnStartup {
+				instanceDesc.State = ACTIVE
+			} else {
+				instanceDesc.State = PENDING
+			}
 		}
 
 		// We exist in the ring, so assume the ring is right and copy out tokens & state out of there.
@@ -699,9 +737,6 @@ func (i *Lifecycler) autoJoin(ctx context.Context, targetState InstanceState) er
 
 		// At this point, we should not have any tokens, and we should be in PENDING state.
 		myTokens, takenTokens := ringDesc.TokensFor(i.ID)
-		if len(myTokens) > 0 {
-			level.Error(i.logger).Log("msg", "tokens already exist for this instance - wasn't expecting any!", "num_tokens", len(myTokens), "ring", i.RingName)
-		}
 
 		newTokens := GenerateTokens(i.cfg.NumTokens-len(myTokens), takenTokens)
 		i.setState(targetState)