Skip to content

tests: adopt synctest in some places #16321

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
npazosmendez merged 1 commit into from
Aug 11, 2026
Merged

tests: adopt synctest in some places #16321

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions pkg/distributor/ha_tracker_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import (
"os"
"strings"
"testing"
"testing/synctest"
"time"

"github.com/go-kit/log"
Expand Down Expand Up @@ -742,6 +743,10 @@ func TestHATrackerCheckReplicaUpdateTimeout(t *testing.T) {
}

func TestHATrackerCheckReplicaShouldFixZeroElectedAtTimestamp(t *testing.T) {
synctest.Test(t, testHATrackerCheckReplicaShouldFixZeroElectedAtTimestamp)
}

func testHATrackerCheckReplicaShouldFixZeroElectedAtTimestamp(t *testing.T) {
const (
replica = "r1"
cluster = "c1"
Expand Down
45 changes: 24 additions & 21 deletions pkg/frontend/querymiddleware/running_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ package querymiddleware
import (
"context"
"testing"
"testing/synctest"
"time"

"github.com/go-kit/log"
Expand Down Expand Up @@ -44,27 +45,29 @@ func TestAwaitQueryFrontendServiceRunning_ServiceIsNotReadyWaitDisabled(t *testi
}

func TestAwaitQueryFrontendServiceRunning_ServiceIsNotReadyInitially(t *testing.T) {
startChan := make(chan struct{})
start := func(context.Context) error {
<-startChan
return nil
}
run := func(ctx context.Context) error {
<-ctx.Done()
return nil
}

service := services.NewBasicService(start, run, nil)
require.NoError(t, service.StartAsync(context.Background()))
defer func() { require.NoError(t, services.StopAndAwaitTerminated(context.Background(), service)) }()

go func() {
time.Sleep(500 * time.Millisecond)
close(startChan)
}()

err := awaitQueryFrontendServiceRunning(context.Background(), service, time.Second, log.NewNopLogger())
require.NoError(t, err)
synctest.Test(t, func(t *testing.T) {
startChan := make(chan struct{})
start := func(context.Context) error {
<-startChan
return nil
}
run := func(ctx context.Context) error {
<-ctx.Done()
return nil
}

service := services.NewBasicService(start, run, nil)
require.NoError(t, service.StartAsync(context.Background()))
defer func() { require.NoError(t, services.StopAndAwaitTerminated(context.Background(), service)) }()

go func() {
time.Sleep(500 * time.Millisecond)
close(startChan)
}()

err := awaitQueryFrontendServiceRunning(context.Background(), service, time.Second, log.NewNopLogger())
require.NoError(t, err)
})
}

func TestAwaitQueryFrontendServiceRunning_ServiceIsNotReadyAfterTimeout(t *testing.T) {
Expand Down
87 changes: 45 additions & 42 deletions pkg/ingester/ingester_ring_lifecycler_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ import (
"fmt"
"strings"
"testing"
"testing/synctest"
"time"

"github.com/go-kit/log"
Expand Down Expand Up @@ -253,55 +254,57 @@ func TestLifecycler_CheckReady(t *testing.T) {

for _, implName := range []string{"classic Lifecycler", "tokenlessLifecycler"} {
t.Run(implName, func(t *testing.T) {
cfg := defaultIngesterTestConfig(t)
cfg.IngesterRing.MinReadyDuration = minReadyDuration
kvClient := cfg.IngesterRing.KVStore.Mock
testRing := createTestRing(t, cfg.IngesterRing)

var lifecycler ingesterLifecycler
switch implName {
case "classic Lifecycler":
lifecycler = createClassicLifecycler(t, cfg.IngesterRing, false, &noopFlushTransferer{}, nil)
case "tokenlessLifecycler":
cfg.IngesterRing.NumTokens = 0
lifecycler = createTokenlessLifecycler(t, cfg.IngesterRing, kvClient, false, nil, nil)
}

ctx := context.Background()

// Before starting: CheckReady should return an error.
require.Error(t, lifecycler.CheckReady(ctx))

// Start the lifecycler.
require.NoError(t, services.StartAndAwaitRunning(ctx, lifecycler))
t.Cleanup(func() {
require.NoError(t, services.StopAndAwaitTerminated(ctx, lifecycler))
})
synctest.Test(t, func(t *testing.T) {
cfg := defaultIngesterTestConfig(t)
cfg.IngesterRing.MinReadyDuration = minReadyDuration
kvClient := cfg.IngesterRing.KVStore.Mock
testRing := createTestRing(t, cfg.IngesterRing)

var lifecycler ingesterLifecycler
switch implName {
case "classic Lifecycler":
lifecycler = createClassicLifecycler(t, cfg.IngesterRing, false, &noopFlushTransferer{}, nil)
case "tokenlessLifecycler":
cfg.IngesterRing.NumTokens = 0
lifecycler = createTokenlessLifecycler(t, cfg.IngesterRing, kvClient, false, nil, nil)
}

ctx := context.Background()

// Before starting: CheckReady should return an error.
require.Error(t, lifecycler.CheckReady(ctx))

// Start the lifecycler.
require.NoError(t, services.StartAndAwaitRunning(ctx, lifecycler))
t.Cleanup(func() {
require.NoError(t, services.StopAndAwaitTerminated(ctx, lifecycler))
})

// Wait for instance to be ACTIVE in the ring.
require.NoError(t, ring.WaitInstanceState(ctx, testRing, cfg.IngesterRing.InstanceID, ring.ACTIVE))
// Wait for instance to be ACTIVE in the ring.
require.NoError(t, ring.WaitInstanceState(ctx, testRing, cfg.IngesterRing.InstanceID, ring.ACTIVE))

// Verify token count matches expected for this lifecycler type.
tokenCount, exists := getInstanceTokenCount(t, kvClient, IngesterRingKey, cfg.IngesterRing.InstanceID)
require.True(t, exists)
require.Equal(t, cfg.IngesterRing.NumTokens, tokenCount)
// Verify token count matches expected for this lifecycler type.
tokenCount, exists := getInstanceTokenCount(t, kvClient, IngesterRingKey, cfg.IngesterRing.InstanceID)
require.True(t, exists)
require.Equal(t, cfg.IngesterRing.NumTokens, tokenCount)

// Immediately after ACTIVE: CheckReady should return error because
// MinReadyDuration hasn't elapsed yet.
require.ErrorContains(t, lifecycler.CheckReady(ctx), "waiting for")
// Immediately after ACTIVE: CheckReady should return error because
// MinReadyDuration hasn't elapsed yet.
require.ErrorContains(t, lifecycler.CheckReady(ctx), "waiting for")

// Wait for MinReadyDuration to fully elapse.
time.Sleep(minReadyDuration)
// Wait for MinReadyDuration to fully elapse.
time.Sleep(minReadyDuration)

// After MinReadyDuration: CheckReady should return nil.
require.NoError(t, lifecycler.CheckReady(ctx))
// After MinReadyDuration: CheckReady should return nil.
require.NoError(t, lifecycler.CheckReady(ctx))

// Change instance to read-only. The ready state should be latched,
// so CheckReady should still return nil.
require.NoError(t, lifecycler.ChangeReadOnlyState(ctx, true))
// Change instance to read-only. The ready state should be latched,
// so CheckReady should still return nil.
require.NoError(t, lifecycler.ChangeReadOnlyState(ctx, true))

// CheckReady should still return nil (latched ready state).
require.NoError(t, lifecycler.CheckReady(ctx))
// CheckReady should still return nil (latched ready state).
require.NoError(t, lifecycler.CheckReady(ctx))
})
})
}
}
Expand Down
53 changes: 28 additions & 25 deletions pkg/ingester/lifecycle_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import (
"net/http"
"net/http/httptest"
"testing"
"testing/synctest"
"time"

"github.com/go-kit/log"
Expand Down Expand Up @@ -65,40 +66,42 @@ func defaultLimitsTestConfig() validation.Limits {

// TestIngesterRestart tests a restarting ingester doesn't keep adding more tokens.
func TestIngesterRestart(t *testing.T) {
config := defaultIngesterTestConfig(t)
limits := defaultLimitsTestConfig()
config.IngesterRing.UnregisterOnShutdown = false
synctest.Test(t, func(t *testing.T) {
config := defaultIngesterTestConfig(t)
limits := defaultLimitsTestConfig()
config.IngesterRing.UnregisterOnShutdown = false

{
ing, _, err := prepareIngesterWithBlocksStorageAndLimits(t, config, limits, nil, "", nil)
require.NoError(t, err)
{
ing, _, err := prepareIngesterWithBlocksStorageAndLimits(t, config, limits, nil, "", nil)
require.NoError(t, err)

require.NoError(t, services.StartAndAwaitRunning(context.Background(), ing))
require.NoError(t, services.StartAndAwaitRunning(context.Background(), ing))

time.Sleep(100 * time.Millisecond)
// Doesn't actually unregister due to UnregisterFromRing: false.
require.NoError(t, services.StopAndAwaitTerminated(context.Background(), ing))
}
time.Sleep(100 * time.Millisecond)
// Doesn't actually unregister due to UnregisterFromRing: false.
require.NoError(t, services.StopAndAwaitTerminated(context.Background(), ing))
}

test.Poll(t, 100*time.Millisecond, 1, func() interface{} {
return numTokens(config.IngesterRing.KVStore.Mock, "localhost", IngesterRingKey)
})
test.Poll(t, 100*time.Millisecond, 1, func() interface{} {
return numTokens(config.IngesterRing.KVStore.Mock, "localhost", IngesterRingKey)
})

{
ing, _, err := prepareIngesterWithBlocksStorageAndLimits(t, config, limits, nil, "", nil)
require.NoError(t, err)
{
ing, _, err := prepareIngesterWithBlocksStorageAndLimits(t, config, limits, nil, "", nil)
require.NoError(t, err)

require.NoError(t, services.StartAndAwaitRunning(context.Background(), ing))
require.NoError(t, services.StartAndAwaitRunning(context.Background(), ing))

time.Sleep(100 * time.Millisecond)
// Doesn't actually unregister due to UnregisterFromRing: false.
require.NoError(t, services.StopAndAwaitTerminated(context.Background(), ing))
}
time.Sleep(100 * time.Millisecond)
// Doesn't actually unregister due to UnregisterFromRing: false.
require.NoError(t, services.StopAndAwaitTerminated(context.Background(), ing))
}

time.Sleep(200 * time.Millisecond)
time.Sleep(200 * time.Millisecond)

test.Poll(t, 100*time.Millisecond, 1, func() interface{} {
return numTokens(config.IngesterRing.KVStore.Mock, "localhost", IngesterRingKey)
test.Poll(t, 100*time.Millisecond, 1, func() interface{} {
return numTokens(config.IngesterRing.KVStore.Mock, "localhost", IngesterRingKey)
})
})
}

Expand Down
Loading
Loading