diff --git a/CHANGELOG.md b/CHANGELOG.md index 9daa87d2f2c..2eaf5d44e3f 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/pkg/ring/lifecycler.go b/pkg/ring/lifecycler.go index 1db33a929f0..9ccfaffea27 100644 --- a/pkg/ring/lifecycler.go +++ b/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) diff --git a/pkg/ring/lifecycler_test.go b/pkg/ring/lifecycler_test.go index bc508370360..7f29a600707 100644 --- a/pkg/ring/lifecycler_test.go +++ b/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()) })