Function TestBasicLifecycler_TokensObservePeriod

ring/basic_lifecycler_test.go:547–586

(t *testing.T)

}

func TestBasicLifecycler_TokensObservePeriod(t *testing.T) {
	ctx := context.Background()
	cfg := prepareBasicLifecyclerConfig()
	cfg.NumTokens = 5
	cfg.TokensObservePeriod = time.Second

	lifecycler, delegate, store, err := prepareBasicLifecycler(t, cfg)
	require.NoError(t, err)
	defer services.StopAndAwaitTerminated(ctx, lifecycler) //nolint:errcheck

	delegate.onRegister = func(_ *BasicLifecycler, _ Desc, _ bool, _ string, _ InstanceDesc) (InstanceState, Tokens) {
		return ACTIVE, Tokens{1, 2, 3, 4, 5}
	}

	require.NoError(t, lifecycler.StartAsync(ctx))

	// While the lifecycler is starting we poll the ring. As soon as the instance
	// is registered, we remove some tokens to simulate how gossip memberlist
	// reconciliation works in case of clashing tokens.
	test.Poll(t, time.Second, true, func() interface{} {
		// Ensure the instance has been registered in the ring.
		desc, ok := getInstanceFromStore(t, store, testInstanceID)
		if !ok {
			return false
		}

		// Remove some tokens.
		return store.CAS(ctx, testRingKey, func(in interface{}) (out interface{}, retry bool, err error) {
			ringDesc := GetOrCreateRingDesc(in)
			ringDesc.AddIngester(testInstanceID, desc.Addr, desc.Zone, Tokens{4, 5}, desc.State, time.Now(), false, time.Time{}, nil)
			return ringDesc, true, nil
		}) == nil
	})

	require.NoError(t, lifecycler.AwaitRunning(ctx))
	assert.Subset(t, lifecycler.GetTokens(), Tokens{4, 5})
	assert.NotContains(t, lifecycler.GetTokens(), uint32(1))
	assert.NotContains(t, lifecycler.GetTokens(), uint32(2))
	assert.NotContains(t, lifecycler.GetTokens(), uint32(3))
}

func TestBasicLifecycler_updateInstance_ShouldAddInstanceToTheRingIfDoesNotExistEvenIfNotChanged(t *testing.T) {
	for _, readOnly := range []bool{false, true} {