Avoid concurrency between heartbeat and autoJoin (#5438)

Signed-off-by: Daniel Deluiggi <[email protected]>

Author: danielblando

pkg/ring/lifecycler.go

@@ -315,12 +315,6 @@ func (i *Lifecycler) getTokens() Tokens {
 	return i.tokens
 }
 
-func (i *Lifecycler) getStateAndTokens() (InstanceState, Tokens) {
-	i.stateMtx.RLock()
-	defer i.stateMtx.RUnlock()
-	return i.state, i.tokens
-}
-
 func (i *Lifecycler) setTokens(tokens Tokens) {
 	i.lifecyclerMetrics.tokensOwned.Set(float64(len(tokens)))
 
@@ -440,7 +434,9 @@ func (i *Lifecycler) loop(ctx context.Context) error {
 	if uint64(i.cfg.HeartbeatPeriod) > 0 {
 		heartbeatTicker := time.NewTicker(i.cfg.HeartbeatPeriod)
 		heartbeatTicker.Stop()
-		time.AfterFunc(time.Duration(uint64(mathrand.Int63())%uint64(i.cfg.HeartbeatPeriod)), func() {
+		// We are jittering for at least half of the time and max the time of the heartbeat.
+		// If we jitter too soon, we can have problems of concurrency with autoJoin leaving the instance on ACTIVE without tokens
+		time.AfterFunc(time.Duration(uint64(i.cfg.HeartbeatPeriod/2)+uint64(mathrand.Int63())%uint64(i.cfg.HeartbeatPeriod/2)), func() {
 			i.heartbeat()
 			heartbeatTicker.Reset(i.cfg.HeartbeatPeriod)
 		})
@@ -806,7 +802,7 @@ func (i *Lifecycler) updateConsul(ctx context.Context) error {
 			ringDesc.AddIngester(i.ID, i.Addr, i.Zone, i.getTokens(), i.GetState(), i.getRegisteredAt())
 		} else {
 			instanceDesc.Timestamp = time.Now().Unix()
-			instanceDesc.State, instanceDesc.Tokens = i.getStateAndTokens()
+			instanceDesc.State = i.GetState()
 			instanceDesc.Addr = i.Addr
 			instanceDesc.Zone = i.Zone
 			instanceDesc.RegisteredTimestamp = i.getRegisteredAt().Unix()

pkg/ring/ring_test.go

@@ -2632,7 +2632,7 @@ func startLifecycler(t *testing.T, cfg Config, heartbeat time.Duration, lifecycl
 func TestShuffleShardWithCaching(t *testing.T) {
 	inmem, closer := consul.NewInMemoryClientWithConfig(GetCodec(), consul.Config{
 		MaxCasRetries: 20,
-		CasRetryDelay: 500 * time.Millisecond,
+		CasRetryDelay: 100 * time.Millisecond,
 	}, log.NewNopLogger(), nil)
 	t.Cleanup(func() { assert.NoError(t, closer.Close()) })
 
@@ -2661,8 +2661,6 @@ func TestShuffleShardWithCaching(t *testing.T) {
 		lcs = append(lcs, lc)
 	}
 
-	time.Sleep(5 * time.Second)
-
 	// Wait until all instances in the ring are ACTIVE.
 	test.Poll(t, 5*time.Second, numLifecyclers, func() interface{} {
 		active := 0