Function TestAutoJoinWithSpreadMinimizingTokenGenerator

ring/lifecycler_test.go:1849–1940

(t *testing.T)

}

func TestAutoJoinWithSpreadMinimizingTokenGenerator(t *testing.T) {
	t.Parallel()

	canJoinDelay := 1 * time.Second
	canJoinTimeout := 5 * time.Second
	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

	r, err := New(ringConfig, "ingester", ringKey, log.NewNopLogger(), nil)
	require.NoError(t, err)
	require.NoError(t, services.StartAndAwaitRunning(context.Background(), r))
	defer services.StopAndAwaitTerminated(context.Background(), r) //nolint:errcheck

	var (
		firstInstanceRegistrationTimestamp  time.Time
		secondInstanceRegistrationTimestamp time.Time
		firstInstanceID                     = instanceName(0, 1)
		secondInstanceID                    = instanceName(1, 1)
		wg                                  sync.WaitGroup
	)
	wg.Add(1)
	go func() {
		var (
			firstInstanceRegistered  = atomic.NewBool(false)
			secondInstanceRegistered = atomic.NewBool(false)
		)
		test.Poll(t, 20*time.Second, true, func() interface{} {
			d, err := r.KVClient.Get(context.Background(), ringKey)
			require.NoError(t, err)

			desc, ok := d.(*Desc)
			if !ok {
				return false
			}

			firstInstanceDesc, ok := desc.Ingesters[firstInstanceID]
			if ok && firstInstanceDesc.State == ACTIVE && len(firstInstanceDesc.Tokens) != 0 {
				if firstInstanceRegistered.CompareAndSwap(false, true) {
					firstInstanceRegistrationTimestamp = time.Now()
				}
			}

			secondInstanceDesc, ok := desc.Ingesters[secondInstanceID]
			if ok && secondInstanceDesc.State == ACTIVE && len(secondInstanceDesc.Tokens) != 0 {
				if secondInstanceRegistered.CompareAndSwap(false, true) {
					secondInstanceRegistrationTimestamp = time.Now()
				}
			}

			return firstInstanceRegistered.Load() && secondInstanceRegistered.Load()
		})
		wg.Done()
	}()