Ring: don't publish instance state before tokens are generated

CHANGELOG.md

@@ -15,6 +15,7 @@
 * [BUGFIX] Store Gateway: Fix the Parquet bucket store ignoring query sharding, which made every shard return all series. #7892
 * [BUGFIX] Querier: Fix the querier crashing instead of reporting the error when a blocks store set subservice fails. `blocksStoreReplicationSet` never initialised its `FailureWatcher`, so the failure callback dereferenced a nil pointer on the goroutine the service manager runs listeners on, and the resulting panic's stack trace buried the error logged for the actual cause. #7896
 * [BUGFIX] Compactor: Delete block deletion markers left in the global `markers/` location of a tenant after the block they refer to was removed. They were skipped at every bucket index update and kept the tenant location in the bucket forever. #7890
+* [BUGFIX] Ring: Fix an instance becoming permanently unable to acquire tokens. #7901
 
 ## 1.22.0 in progress
 * [CHANGE] Ruler: Remove the deprecated `-ruler.evaluation-delay-duration` flag and its `ruler_evaluation_delay_duration` per-tenant limit. Use `-ruler.query-offset` / `ruler_query_offset`, which no longer takes the higher of the two values. Cortex decodes the runtime config strictly, so a leftover `ruler_evaluation_delay_duration` override makes the runtime config fail to load: Cortex **exits at startup** (`module failed`, `module=runtime-config`), and on an already-running process every reload fails, pinning the last good overrides and dropping `cortex_runtime_config_last_reload_successful` to 0. Run `grep -r ruler_evaluation_delay_duration` over your runtime configs before upgrading. #7792

pkg/ring/lifecycler.go

@@ -798,6 +798,23 @@ func (i *Lifecycler) initRing(ctx context.Context) (bool, error) {
 			return ringDesc, true, nil
 		}
 
+		// An instance registered in the ring without tokens owns no part of the ring, so
+		// it can neither receive writes nor pass its readiness check. Unless it is
+		// PENDING it will also never run autoJoin to acquire tokens (see loop()), so it
+		// stays unusable for as long as the process lives. Reset it to PENDING in order
+		// to start the lifecycle from the beginning.
+		if len(instanceDesc.Tokens) == 0 && instanceDesc.State != PENDING {
+			level.Warn(i.logger).Log("msg", "instance found in ring without tokens, setting to PENDING",
+				"ring", i.RingName, "state", instanceDesc.State)
+			if instanceDesc.State == READONLY {
+				i.setPreviousState(READONLY)
+			}
+			instanceDesc.State = PENDING
+			instanceDesc.Timestamp = time.Now().Unix()
+			ringDesc.Ingesters[i.ID] = instanceDesc
+			return ringDesc, true, nil
+		}
+
 		// If the ingester failed to clean its ring entry up in can leave its state in LEAVING
 		// OR unregister_on_shutdown=false
 		// if autoJoinOnStartup, move it into previous state based on token file (default: ACTIVE)
@@ -964,8 +981,6 @@ func (i *Lifecycler) autoJoin(ctx context.Context, targetState InstanceState, al
 			ringDesc = in.(*Desc)
 		}
 
-		i.setState(targetState)
-
 		// At this point, we should not have any tokens, and we should be in PENDING state.
 		// Need to make sure we didn't change the num of tokens configured
 		myTokens, _ := ringDesc.TokensFor(i.ID)
@@ -976,6 +991,7 @@ func (i *Lifecycler) autoJoin(ctx context.Context, targetState InstanceState, al
 
 		if needTokens == 0 && myTokens.Equals(i.getTokens()) {
 			// Tokens have been verified. No need to change them.
+			i.setState(targetState)
 			state := i.GetState()
 			ringDesc.AddIngester(i.ID, i.Addr, i.Zone, i.getTokens(), state, i.getRegisteredAt())
 			level.Info(i.logger).Log("msg", "auto joined with existing tokens", "ring", i.RingName, "state", state)
@@ -984,6 +1000,11 @@ func (i *Lifecycler) autoJoin(ctx context.Context, targetState InstanceState, al
 
 		newTokens := i.tg.GenerateTokens(ringDesc, i.ID, i.Zone, needTokens, false)
 		if len(newTokens) != needTokens {
+			// Leave the instance in its current state: it still owns no tokens. Moving it
+			// to targetState here would let a concurrent heartbeat persist a tokenless
+			// instance in targetState, and on the next startup initRing adopts that state
+			// from the ring. Since autoJoin only runs while the instance is PENDING, the
+			// instance would then never be able to acquire tokens again.
 			level.Warn(i.logger).Log("msg", "retrying generate tokens")
 			return ringDesc, true, errors.New("could not generate tokens")
 		}
@@ -992,6 +1013,7 @@ func (i *Lifecycler) autoJoin(ctx context.Context, targetState InstanceState, al
 		sort.Sort(myTokens)
 		i.setTokens(myTokens)
 
+		i.setState(targetState)
 		state := i.GetState()
 		ringDesc.AddIngester(i.ID, i.Addr, i.Zone, i.getTokens(), state, i.getRegisteredAt())
 		level.Info(i.logger).Log("msg", "auto joined with new tokens", "ring", i.RingName, "state", state)

pkg/ring/lifecycler_test.go

@@ -126,6 +126,107 @@ func TestLifecycler_RenewTokens(t *testing.T) {
 	require.Equal(t, 51, diff)
 }
 
+// tokenRefusingGenerator never returns tokens, mimicking MinimizeSpreadTokenGenerator
+// declining to generate for every instance but the earliest registered one when several
+// instances in the same zone hold no tokens. autoJoin() always calls it with force=false.
+type tokenRefusingGenerator struct{}
+
+func (g *tokenRefusingGenerator) GenerateTokens(*Desc, string, string, int, bool) []uint32 {
+	return []uint32{}
+}
+
+func TestLifecycler_AutoJoinShouldNotChangeStateWhenTokenGenerationFails(t *testing.T) {
+	ringStore, closer := consul.NewInMemoryClient(GetCodec(), log.NewNopLogger(), nil)
+	t.Cleanup(func() { assert.NoError(t, closer.Close()) })
+
+	var ringConfig Config
+	flagext.DefaultValues(&ringConfig)
+	ringConfig.KVStore.Mock = ringStore
+
+	lifecyclerConfig := testLifecyclerConfig(ringConfig, "ing1")
+	l, err := NewLifecycler(lifecyclerConfig, &nopFlushTransferer{}, "ingester", ringKey, false, true, log.NewNopLogger(), nil)
+	require.NoError(t, err)
+	l.tg = &tokenRefusingGenerator{}
+
+	// autoJoin() must fail, because no tokens could be generated.
+	require.Error(t, l.autoJoin(context.Background(), ACTIVE, false))
+
+	// The instance must be left in PENDING. Were it moved to ACTIVE here, a concurrent
+	// heartbeat would persist a tokenless instance in ACTIVE and, on the next startup,
+	// initRing() would adopt that state from the ring. autoJoin() only runs while the
+	// instance is PENDING, so the instance would never acquire tokens again.
+	assert.Equal(t, PENDING, l.GetState())
+	assert.Empty(t, l.getTokens())
+}
+
+func TestLifecycler_InitRingShouldResetInstanceWithoutTokensToPending(t *testing.T) {
+	tests := map[string]struct {
+		ringState         InstanceState
+		ringTokens        []uint32
+		autoJoinOnStartup bool
+
+		// expectedState is the lifecycler state, which is what gates autoJoin().
+		expectedState  InstanceState
+		expectedTokens int
+		// expectedRingState is the state persisted back to the ring by initRing().
+		expectedRingState InstanceState
+	}{
+		// An instance owning no tokens cannot acquire any unless it is PENDING, whichever
+		// state the ring entry holds, so all of these must be reset.
+		"ACTIVE without tokens is reset to PENDING": {
+			ringState: ACTIVE, ringTokens: []uint32{},
+			expectedState: PENDING, expectedTokens: 0, expectedRingState: PENDING,
+		},
+		"READONLY without tokens is reset to PENDING": {
+			ringState: READONLY, ringTokens: []uint32{},
+			expectedState: PENDING, expectedTokens: 0, expectedRingState: PENDING,
+		},
+		"LEAVING without tokens is reset to PENDING": {
+			ringState: LEAVING, ringTokens: []uint32{},
+			expectedState: PENDING, expectedTokens: 0, expectedRingState: PENDING,
+		},
+		// An instance that kept its tokens must still follow the existing LEAVING
+		// recovery path and rejoin with the tokens it already owns. That path updates the
+		// lifecycler state and lets the next heartbeat persist it, so the ring entry is
+		// deliberately still LEAVING at this point.
+		"LEAVING with tokens keeps following the existing recovery path": {
+			ringState: LEAVING, ringTokens: []uint32{1, 2, 3}, autoJoinOnStartup: true,
+			expectedState: ACTIVE, expectedTokens: 3, expectedRingState: LEAVING,
+		},
+	}
+
+	for name, testData := range tests {
+		t.Run(name, func(t *testing.T) {
+			ringStore, closer := consul.NewInMemoryClient(GetCodec(), log.NewNopLogger(), nil)
+			t.Cleanup(func() { assert.NoError(t, closer.Close()) })
+
+			require.NoError(t, ringStore.CAS(context.Background(), ringKey, func(any) (any, bool, error) {
+				desc := NewDesc()
+				desc.AddIngester("ing1", "0.0.0.0:1", "zone1", testData.ringTokens, testData.ringState, time.Now())
+				return desc, true, nil
+			}))
+
+			var ringConfig Config
+			flagext.DefaultValues(&ringConfig)
+			ringConfig.KVStore.Mock = ringStore
+
+			lifecyclerConfig := testLifecyclerConfig(ringConfig, "ing1")
+			l, err := NewLifecycler(lifecyclerConfig, &nopFlushTransferer{}, "ingester", ringKey, testData.autoJoinOnStartup, true, log.NewNopLogger(), nil)
+			require.NoError(t, err)
+
+			_, err = l.initRing(context.Background())
+			require.NoError(t, err)
+
+			assert.Equal(t, testData.expectedState, l.GetState())
+			assert.Len(t, l.getTokens(), testData.expectedTokens)
+
+			desc, err := ringStore.Get(context.Background(), ringKey)
+			require.NoError(t, err)
+			assert.Equal(t, testData.expectedRingState, desc.(*Desc).Ingesters["ing1"].State)
+		})
+	}
+}
+
 func TestLifecycler_DefferedJoin(t *testing.T) {
 	ringStore, closer := consul.NewInMemoryClient(GetCodec(), log.NewNopLogger(), nil)
 	t.Cleanup(func() { assert.NoError(t, closer.Close()) })