Function TestRestartIngester_NoUnregister_LongHeartbeat

ring/lifecycler_test.go:1434–1490

(t *testing.T)

func (f *noopFlushTransferer) TransferOut(_ context.Context) error { return nil }

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

	ringStore, closer := consul.NewInMemoryClient(GetCodec(), log.NewNopLogger(), nil)
	t.Cleanup(func() { assert.NoError(t, closer.Close()) })

	origTokens := initTokenGenerator(t).GenerateTokens(100, nil)

	const id = "test"
	registeredAt := time.Now().Add(-1 * time.Hour)

	err := ringStore.CAS(context.Background(), ringKey, func(in interface{}) (out interface{}, retry bool, err error) {
		// Create ring with LEAVING entry with some tokens
		r := GetOrCreateRingDesc(in)
		r.AddIngester(id, "3.3.3.3:333", "old", origTokens, LEAVING, registeredAt, false, time.Time{}, nil)
		return r, true, err
	})
	require.NoError(t, err)

	var lifecyclerConfig LifecyclerConfig
	flagext.DefaultValues(&lifecyclerConfig)
	lifecyclerConfig.Addr = "1.1.1.1"
	lifecyclerConfig.Port = 111
	lifecyclerConfig.Zone = "new"
	lifecyclerConfig.RingConfig.KVStore.Mock = ringStore
	lifecyclerConfig.NumTokens = len(origTokens)
	lifecyclerConfig.ID = id
	lifecyclerConfig.HeartbeatPeriod = 5 * time.Minute // Long hearbeat period.
	lifecyclerConfig.MinReadyDuration = 0              // Disable waiting extra time for Ready
	lifecyclerConfig.JoinAfter = 1 * time.Minute       // Use long value to make sure that we don't use "join" code path.

	l, err := NewLifecycler(lifecyclerConfig, &noopFlushTransferer{}, "test", ringKey, false, log.NewNopLogger(), nil)
	require.NoError(t, err)

	require.NoError(t, services.StartAndAwaitRunning(context.Background(), l))
	defer services.StopAndAwaitTerminated(context.Background(), l) //nolint:errcheck

	test.Poll(t, 1*time.Second, nil, func() interface{} {
		return l.CheckReady(context.Background())
	})

	// Lifecycler should be in ACTIVE state, using tokens from the ring.
	require.Equal(t, ACTIVE, l.GetState())
	require.Equal(t, origTokens, l.getTokens())
	require.Equal(t, registeredAt.Truncate(time.Second), l.getRegisteredAt())

	// check that ring entry has updated address and state
	desc, err := ringStore.Get(context.Background(), ringKey)
	require.NoError(t, err)

	r := GetOrCreateRingDesc(desc)
	require.Equal(t, lifecyclerConfig.ID, r.Ingesters[id].Id)
	require.Equal(t, ACTIVE, r.Ingesters[id].State)
	require.Equal(t, "1.1.1.1:111", r.Ingesters[id].Addr)
	require.Equal(t, "new", r.Ingesters[id].Zone)
	require.Equal(t, registeredAt.Unix(), r.Ingesters[id].RegisteredTimestamp)
}