Function TestAutoForgetDelegate

pkg/ring/basic_lifecycler_delegates_test.go:236–309

(t *testing.T)

}

func TestAutoForgetDelegate(t *testing.T) {
	const forgetPeriod = time.Minute
	registeredAt := time.Now()

	tests := map[string]struct {
		setup             func(ringDesc *Desc)
		expectedInstances []string
	}{
		"no unhealthy instance in the ring": {
			setup: func(ringDesc *Desc) {
				ringDesc.AddIngester("instance-1", "1.1.1.1", "", nil, ACTIVE, registeredAt)
			},
			expectedInstances: []string{testInstanceID, "instance-1"},
		},
		"unhealthy instance in the ring that has NOTreached the forget period yet": {
			setup: func(ringDesc *Desc) {
				i := ringDesc.AddIngester("instance-1", "1.1.1.1", "", nil, ACTIVE, registeredAt)
				i.Timestamp = time.Now().Add(-forgetPeriod).Add(5 * time.Second).Unix()
				ringDesc.Ingesters["instance-1"] = i
			},
			expectedInstances: []string{testInstanceID, "instance-1"},
		},
		"unhealthy instance in the ring that has reached the forget period": {
			setup: func(ringDesc *Desc) {
				i := ringDesc.AddIngester("instance-1", "1.1.1.1", "", nil, ACTIVE, registeredAt)
				i.Timestamp = time.Now().Add(-forgetPeriod).Add(-5 * time.Second).Unix()
				ringDesc.Ingesters["instance-1"] = i
			},
			expectedInstances: []string{testInstanceID},
		},
	}

	for testName, testData := range tests {
		t.Run(testName, func(t *testing.T) {
			ctx := context.Background()
			cfg := prepareBasicLifecyclerConfig()
			cfg.HeartbeatPeriod = 100 * time.Millisecond

			testDelegate := &mockDelegate{}

			autoForgetDelegate := NewAutoForgetDelegate(forgetPeriod, testDelegate, log.NewNopLogger())
			lifecycler, store, err := prepareBasicLifecyclerWithDelegate(t, cfg, autoForgetDelegate)
			require.NoError(t, err)

			// Setup the initial state of the ring.
			require.NoError(t, store.CAS(ctx, testRingKey, func(in any) (out any, retry bool, err error) {
				ringDesc := NewDesc()
				testData.setup(ringDesc)
				return ringDesc, true, nil
			}))

			// Start the lifecycler.
			require.NoError(t, services.StartAndAwaitRunning(ctx, lifecycler))
			defer services.StopAndAwaitTerminated(ctx, lifecycler) //nolint:errcheck

			// Wait until an heartbeat has been sent.
			test.Poll(t, time.Second, true, func() any {
				return testutil.ToFloat64(lifecycler.metrics.heartbeats) > 0