From d95619fa33ac7e9503f59348b842987087c2c00b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nicol=C3=A1s=20Pazos?= Date: Thu, 6 Aug 2026 16:25:44 -0300 Subject: [PATCH] Speed up trivially convertible tests with testing/synctest Wrap wait-heavy tests in synctest.Test so their sleeps, tickers and retry backoffs run on fake time. Test-only change, no logic changes. --- pkg/distributor/ha_tracker_test.go | 5 + pkg/frontend/querymiddleware/running_test.go | 45 +++--- pkg/ingester/ingester_ring_lifecycler_test.go | 87 +++++----- pkg/ingester/lifecycle_test.go | 53 +++--- ...ing_algorithm_tree_queue_benchmark_test.go | 151 +++++++++--------- pkg/ruler/compat_test.go | 85 +++++----- pkg/ruler/remotequerier_test.go | 91 ++++++----- pkg/ruler/ruler_sync_queue_test.go | 91 ++++++----- pkg/ruler/ruler_test.go | 131 +++++++-------- pkg/storage/indexheader/reader_pool_test.go | 81 +++++----- pkg/storegateway/gateway_test.go | 33 ++-- 11 files changed, 445 insertions(+), 408 deletions(-) diff --git a/pkg/distributor/ha_tracker_test.go b/pkg/distributor/ha_tracker_test.go index dc05b24a42b..4fd578f108c 100644 --- a/pkg/distributor/ha_tracker_test.go +++ b/pkg/distributor/ha_tracker_test.go @@ -11,6 +11,7 @@ import ( "os" "strings" "testing" + "testing/synctest" "time" "github.com/go-kit/log" @@ -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" diff --git a/pkg/frontend/querymiddleware/running_test.go b/pkg/frontend/querymiddleware/running_test.go index 0d11f03e729..e98cc759ee3 100644 --- a/pkg/frontend/querymiddleware/running_test.go +++ b/pkg/frontend/querymiddleware/running_test.go @@ -5,6 +5,7 @@ package querymiddleware import ( "context" "testing" + "testing/synctest" "time" "github.com/go-kit/log" @@ -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) { diff --git a/pkg/ingester/ingester_ring_lifecycler_test.go b/pkg/ingester/ingester_ring_lifecycler_test.go index f532180c345..ad85450d0ed 100644 --- a/pkg/ingester/ingester_ring_lifecycler_test.go +++ b/pkg/ingester/ingester_ring_lifecycler_test.go @@ -7,6 +7,7 @@ import ( "fmt" "strings" "testing" + "testing/synctest" "time" "github.com/go-kit/log" @@ -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)) + }) }) } } diff --git a/pkg/ingester/lifecycle_test.go b/pkg/ingester/lifecycle_test.go index c8a8ec42fc4..36f9eff0afe 100644 --- a/pkg/ingester/lifecycle_test.go +++ b/pkg/ingester/lifecycle_test.go @@ -11,6 +11,7 @@ import ( "net/http" "net/http/httptest" "testing" + "testing/synctest" "time" "github.com/go-kit/log" @@ -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) + }) }) } diff --git a/pkg/queue/multi_queuing_algorithm_tree_queue_benchmark_test.go b/pkg/queue/multi_queuing_algorithm_tree_queue_benchmark_test.go index 5a8d9039fa6..f7d7eb2f05d 100644 --- a/pkg/queue/multi_queuing_algorithm_tree_queue_benchmark_test.go +++ b/pkg/queue/multi_queuing_algorithm_tree_queue_benchmark_test.go @@ -10,6 +10,7 @@ import ( "slices" "sync" "testing" + "testing/synctest" "time" "github.com/go-kit/log" @@ -414,81 +415,83 @@ func TestMultiDimensionalQueueAlgorithmSlowConsumerEffects(t *testing.T) { } t.Run(testCaseName, func(t *testing.T) { - queue, err := New( - log.NewNopLogger(), - maxOutStandingPerTenant, - consumerForgetDelay, - promauto.With(nil).NewGaugeVec(prometheus.GaugeOpts{}, []string{"user"}), - nil, - promauto.With(nil).NewCounterVec(prometheus.CounterOpts{}, []string{"user"}), - promauto.With(nil).NewHistogram(prometheus.HistogramOpts{}), - ) - require.NoError(t, err) - - // New constructor does not allow passing in a tree or tenantConsumerShards - // so we have to override here to use the same structures as the test case - queue.queueBroker.tenantConsumerAssignments = &tenantConsumerShards{ - consumerIDsSorted: make([]tree.ConsumerID, 0), - tenantsByID: make(map[string]*queueTenant), - queuingAlgorithm: scenario.tqa, - } - queue.queueBroker.tree = scenario.tree - - ctx := context.Background() - require.NoError(t, queue.starting(ctx)) - - t.Cleanup(func() { - // if the test has failed and the queue does not get cleared, - // we must send a shutdown signal for the remaining connected consumer - // or else StopAndAwaitTerminated will never complete. - assert.NoError(t, services.StopAndAwaitTerminated(ctx, queue)) + synctest.Test(t, func(t *testing.T) { + queue, err := New( + log.NewNopLogger(), + maxOutStandingPerTenant, + consumerForgetDelay, + promauto.With(nil).NewGaugeVec(prometheus.GaugeOpts{}, []string{"user"}), + nil, + promauto.With(nil).NewCounterVec(prometheus.CounterOpts{}, []string{"user"}), + promauto.With(nil).NewHistogram(prometheus.HistogramOpts{}), + ) + require.NoError(t, err) + + // New constructor does not allow passing in a tree or tenantConsumerShards + // so we have to override here to use the same structures as the test case + queue.queueBroker.tenantConsumerAssignments = &tenantConsumerShards{ + consumerIDsSorted: make([]tree.ConsumerID, 0), + tenantsByID: make(map[string]*queueTenant), + queuingAlgorithm: scenario.tqa, + } + queue.queueBroker.tree = scenario.tree + + ctx := context.Background() + require.NoError(t, queue.starting(ctx)) + + t.Cleanup(func() { + // if the test has failed and the queue does not get cleared, + // we must send a shutdown signal for the remaining connected consumer + // or else StopAndAwaitTerminated will never complete. + assert.NoError(t, services.StopAndAwaitTerminated(ctx, queue)) + }) + + // configure queue producers to enqueue requests with the query component + // randomly assigned according to the distribution defined in the test case + queueDimensionFunc := makeWeightedRandAdditionalQueueDimensionFunc( + weightedQueueDimensionTestCase.tenantQueueDimensionsWeights, + ) + producersChan, producersErrGroup := makeQueueProducerGroup( + queue, maxConsumersPerTenant, totalRequests, numProducers, numTenants, queueDimensionFunc, + ) + + // configure queue consumers with respective latencies for processing requests + // which were assigned the "normal" or "slow" query component + consumeFunc := makeQueueConsumeFuncWithSlowQueryComponent( + slowConsumerLatency, normalConsumerLatency, testCaseObservations, + ) + queueConsumerErrGroup, startConsumersChan := makeQueueConsumerGroup( + context.Background(), queue, totalRequests, numConsumers, numWorkersPerConsumer, consumeFunc, + ) + + // run queue consumers and producers and wait for completion + + // run producers to fill queue + close(producersChan) + err = producersErrGroup.Wait() + require.NoError(t, err) + + // run consumers to until queue is empty + close(startConsumersChan) + err = queueConsumerErrGroup.Wait() + require.NoError(t, err) + + report := testCaseObservations.Report() + t.Logf("%s: %s", testCaseName, report.QueryComponentReportString()) + t.Logf("%s: %s", testCaseName, report.TenantIDReportString()) + // collect results in order + testCaseNames = append(testCaseNames, testCaseName) + testCaseReports[testCaseName] = report + + require.NoError(t, queue.stop(nil)) + assert.NotEqual(t, "", tree.CurrentConsumer(scenario.tqa)) + // ensure everything was dequeued; we can pass a nil DequeueArgs because we don't + // want to update any state before doing this (i.e., we're dequeuing for _any_ consumer, + // just to make sure the tree is empty). + path, val := scenario.tree.Dequeue(nil) + assert.Nil(t, val) + assert.Equal(t, path, tree.QueuePath{}) }) - - // configure queue producers to enqueue requests with the query component - // randomly assigned according to the distribution defined in the test case - queueDimensionFunc := makeWeightedRandAdditionalQueueDimensionFunc( - weightedQueueDimensionTestCase.tenantQueueDimensionsWeights, - ) - producersChan, producersErrGroup := makeQueueProducerGroup( - queue, maxConsumersPerTenant, totalRequests, numProducers, numTenants, queueDimensionFunc, - ) - - // configure queue consumers with respective latencies for processing requests - // which were assigned the "normal" or "slow" query component - consumeFunc := makeQueueConsumeFuncWithSlowQueryComponent( - slowConsumerLatency, normalConsumerLatency, testCaseObservations, - ) - queueConsumerErrGroup, startConsumersChan := makeQueueConsumerGroup( - context.Background(), queue, totalRequests, numConsumers, numWorkersPerConsumer, consumeFunc, - ) - - // run queue consumers and producers and wait for completion - - // run producers to fill queue - close(producersChan) - err = producersErrGroup.Wait() - require.NoError(t, err) - - // run consumers to until queue is empty - close(startConsumersChan) - err = queueConsumerErrGroup.Wait() - require.NoError(t, err) - - report := testCaseObservations.Report() - t.Logf("%s: %s", testCaseName, report.QueryComponentReportString()) - t.Logf("%s: %s", testCaseName, report.TenantIDReportString()) - // collect results in order - testCaseNames = append(testCaseNames, testCaseName) - testCaseReports[testCaseName] = report - - require.NoError(t, queue.stop(nil)) - assert.NotEqual(t, "", tree.CurrentConsumer(scenario.tqa)) - // ensure everything was dequeued; we can pass a nil DequeueArgs because we don't - // want to update any state before doing this (i.e., we're dequeuing for _any_ consumer, - // just to make sure the tree is empty). - path, val := scenario.tree.Dequeue(nil) - assert.Nil(t, val) - assert.Equal(t, path, tree.QueuePath{}) }) } } diff --git a/pkg/ruler/compat_test.go b/pkg/ruler/compat_test.go index bb29640d683..db615bf3d86 100644 --- a/pkg/ruler/compat_test.go +++ b/pkg/ruler/compat_test.go @@ -15,6 +15,7 @@ import ( "net/http" "strings" "testing" + "testing/synctest" "time" "github.com/go-kit/log" @@ -414,48 +415,50 @@ func TestMetricsQueryFuncErrors(t *testing.T) { } func TestRecordAndReportRuleQueryMetrics(t *testing.T) { - queryTime := promauto.With(nil).NewCounterVec(prometheus.CounterOpts{}, []string{"user"}) - zeroFetchedSeriesCount := promauto.With(nil).NewCounterVec(prometheus.CounterOpts{}, []string{"user"}) + synctest.Test(t, func(t *testing.T) { + queryTime := promauto.With(nil).NewCounterVec(prometheus.CounterOpts{}, []string{"user"}) + zeroFetchedSeriesCount := promauto.With(nil).NewCounterVec(prometheus.CounterOpts{}, []string{"user"}) - mockFunc := func(context.Context, string, time.Time) (promql.Vector, error) { - time.Sleep(1 * time.Second) - return promql.Vector{}, nil - } - qf := RecordAndReportRuleQueryMetrics(mockFunc, queryTime.WithLabelValues("userID"), zeroFetchedSeriesCount.WithLabelValues("userID"), false, log.NewNopLogger()) - - // Ensure we start with counters at 0. - require.LessOrEqual(t, float64(0), testutil.ToFloat64(queryTime.WithLabelValues("userID"))) - require.Equal(t, float64(0), testutil.ToFloat64(zeroFetchedSeriesCount.WithLabelValues("userID"))) - - // Increment zeroFetchedSeriesCount for non-existent series. - _, _ = qf(context.Background(), "test", time.Now()) - require.LessOrEqual(t, float64(1), testutil.ToFloat64(queryTime.WithLabelValues("userID"))) - require.Equal(t, float64(1), testutil.ToFloat64(zeroFetchedSeriesCount.WithLabelValues("userID"))) - - // Increment zeroFetchedSeriesCount for another non-existent series. - _, _ = qf(context.Background(), "test2", time.Now()) - require.LessOrEqual(t, float64(2), testutil.ToFloat64(queryTime.WithLabelValues("userID"))) - require.Equal(t, float64(2), testutil.ToFloat64(zeroFetchedSeriesCount.WithLabelValues("userID"))) - - // Don't increment zeroFetchedSeriesCount for query without series selectors. - _, _ = qf(context.Background(), "vector(0.995)", time.Now()) - require.LessOrEqual(t, float64(3), testutil.ToFloat64(queryTime.WithLabelValues("userID"))) - require.Equal(t, float64(2), testutil.ToFloat64(zeroFetchedSeriesCount.WithLabelValues("userID"))) - - // Don't increment zeroFetchedSeriesCount for another query without series selectors. - _, _ = qf(context.Background(), "vector(2.4192e+15 / 1e+09)", time.Now()) - require.LessOrEqual(t, float64(4), testutil.ToFloat64(queryTime.WithLabelValues("userID"))) - require.Equal(t, float64(2), testutil.ToFloat64(zeroFetchedSeriesCount.WithLabelValues("userID"))) - - // Increment zeroFetchedSeriesCount for non-existent series even when combined with a non-series selector. - _, _ = qf(context.Background(), "test + vector(0.995)", time.Now()) - require.LessOrEqual(t, float64(5), testutil.ToFloat64(queryTime.WithLabelValues("userID"))) - require.Equal(t, float64(3), testutil.ToFloat64(zeroFetchedSeriesCount.WithLabelValues("userID"))) - - // Don't increment zeroFetchedSeriesCount for queries with errors. - _, _ = qf(context.Background(), "rate(test)", time.Now()) - require.LessOrEqual(t, float64(6), testutil.ToFloat64(queryTime.WithLabelValues("userID"))) - require.Equal(t, float64(3), testutil.ToFloat64(zeroFetchedSeriesCount.WithLabelValues("userID"))) + mockFunc := func(context.Context, string, time.Time) (promql.Vector, error) { + time.Sleep(1 * time.Second) + return promql.Vector{}, nil + } + qf := RecordAndReportRuleQueryMetrics(mockFunc, queryTime.WithLabelValues("userID"), zeroFetchedSeriesCount.WithLabelValues("userID"), false, log.NewNopLogger()) + + // Ensure we start with counters at 0. + require.LessOrEqual(t, float64(0), testutil.ToFloat64(queryTime.WithLabelValues("userID"))) + require.Equal(t, float64(0), testutil.ToFloat64(zeroFetchedSeriesCount.WithLabelValues("userID"))) + + // Increment zeroFetchedSeriesCount for non-existent series. + _, _ = qf(context.Background(), "test", time.Now()) + require.LessOrEqual(t, float64(1), testutil.ToFloat64(queryTime.WithLabelValues("userID"))) + require.Equal(t, float64(1), testutil.ToFloat64(zeroFetchedSeriesCount.WithLabelValues("userID"))) + + // Increment zeroFetchedSeriesCount for another non-existent series. + _, _ = qf(context.Background(), "test2", time.Now()) + require.LessOrEqual(t, float64(2), testutil.ToFloat64(queryTime.WithLabelValues("userID"))) + require.Equal(t, float64(2), testutil.ToFloat64(zeroFetchedSeriesCount.WithLabelValues("userID"))) + + // Don't increment zeroFetchedSeriesCount for query without series selectors. + _, _ = qf(context.Background(), "vector(0.995)", time.Now()) + require.LessOrEqual(t, float64(3), testutil.ToFloat64(queryTime.WithLabelValues("userID"))) + require.Equal(t, float64(2), testutil.ToFloat64(zeroFetchedSeriesCount.WithLabelValues("userID"))) + + // Don't increment zeroFetchedSeriesCount for another query without series selectors. + _, _ = qf(context.Background(), "vector(2.4192e+15 / 1e+09)", time.Now()) + require.LessOrEqual(t, float64(4), testutil.ToFloat64(queryTime.WithLabelValues("userID"))) + require.Equal(t, float64(2), testutil.ToFloat64(zeroFetchedSeriesCount.WithLabelValues("userID"))) + + // Increment zeroFetchedSeriesCount for non-existent series even when combined with a non-series selector. + _, _ = qf(context.Background(), "test + vector(0.995)", time.Now()) + require.LessOrEqual(t, float64(5), testutil.ToFloat64(queryTime.WithLabelValues("userID"))) + require.Equal(t, float64(3), testutil.ToFloat64(zeroFetchedSeriesCount.WithLabelValues("userID"))) + + // Don't increment zeroFetchedSeriesCount for queries with errors. + _, _ = qf(context.Background(), "rate(test)", time.Now()) + require.LessOrEqual(t, float64(6), testutil.ToFloat64(queryTime.WithLabelValues("userID"))) + require.Equal(t, float64(3), testutil.ToFloat64(zeroFetchedSeriesCount.WithLabelValues("userID"))) + }) } func TestRecordAndReportRuleQueryMetrics_Logging(t *testing.T) { diff --git a/pkg/ruler/remotequerier_test.go b/pkg/ruler/remotequerier_test.go index 54276e76030..871386d2a5c 100644 --- a/pkg/ruler/remotequerier_test.go +++ b/pkg/ruler/remotequerier_test.go @@ -9,6 +9,7 @@ import ( "net/http" "net/url" "testing" + "testing/synctest" "time" "github.com/go-kit/log" @@ -333,39 +334,41 @@ func TestRemoteQuerier_QueryRetryOnFailure(t *testing.T) { } for testName, testCase := range tests { t.Run(testName, func(t *testing.T) { - var count atomic.Int64 - - ctx, cancel := context.WithCancel(context.Background()) - mockClientFn := func(context.Context, *httpgrpc.HTTPRequest, ...grpc.CallOption) (*httpgrpc.HTTPResponse, error) { - count.Add(1) - if testCase.err != nil { - if grpcutil.IsCanceled(testCase.err) { - cancel() + synctest.Test(t, func(t *testing.T) { + var count atomic.Int64 + + ctx, cancel := context.WithCancel(context.Background()) + mockClientFn := func(context.Context, *httpgrpc.HTTPRequest, ...grpc.CallOption) (*httpgrpc.HTTPResponse, error) { + count.Add(1) + if testCase.err != nil { + if grpcutil.IsCanceled(testCase.err) { + cancel() + } + return nil, testCase.err } - return nil, testCase.err + return testCase.response, nil } - return testCase.response, nil - } - q := NewRemoteQuerier(newGrpcRoundTripper(mockHTTPGRPCClient(mockClientFn)), time.Minute, 1, formatJSON, prometheusGrpcURL, log.NewNopLogger()) - require.Equal(t, int64(0), count.Load()) - _, err := q.Query(ctx, "qs", time.Now()) - if testCase.err == nil { - if testCase.expectedError == nil { - require.NoError(t, err) + q := NewRemoteQuerier(newGrpcRoundTripper(mockHTTPGRPCClient(mockClientFn)), time.Minute, 1, formatJSON, prometheusGrpcURL, log.NewNopLogger()) + require.Equal(t, int64(0), count.Load()) + _, err := q.Query(ctx, "qs", time.Now()) + if testCase.err == nil { + if testCase.expectedError == nil { + require.NoError(t, err) + } else { + require.Error(t, err) + require.ErrorContains(t, err, testCase.expectedError.Error()) + } + require.Equal(t, int64(1), count.Load()) } else { require.Error(t, err) require.ErrorContains(t, err, testCase.expectedError.Error()) + if testCase.expectedRetries { + require.Greater(t, count.Load(), int64(1)) + } else { + require.Equal(t, int64(1), count.Load()) + } } - require.Equal(t, int64(1), count.Load()) - } else { - require.Error(t, err) - require.ErrorContains(t, err, testCase.expectedError.Error()) - if testCase.expectedRetries { - require.Greater(t, count.Load(), int64(1)) - } else { - require.Equal(t, int64(1), count.Load()) - } - } + }) }) } } @@ -874,25 +877,27 @@ func TestRemoteQuerier_StatusErrorResponses(t *testing.T) { } for testName, testCase := range testCases { t.Run(testName, func(t *testing.T) { - mockClientFn := func(context.Context, *httpgrpc.HTTPRequest, ...grpc.CallOption) (*httpgrpc.HTTPResponse, error) { - return testCase.resp, testCase.err - } - logger := newLoggerWithCounter() - q := NewRemoteQuerier(newGrpcRoundTripper(mockHTTPGRPCClient(mockClientFn)), time.Minute, 1, formatJSON, prometheusGrpcURL, logger) - - tm := time.Unix(1649092025, 515834) + synctest.Test(t, func(t *testing.T) { + mockClientFn := func(context.Context, *httpgrpc.HTTPRequest, ...grpc.CallOption) (*httpgrpc.HTTPResponse, error) { + return testCase.resp, testCase.err + } + logger := newLoggerWithCounter() + q := NewRemoteQuerier(newGrpcRoundTripper(mockHTTPGRPCClient(mockClientFn)), time.Minute, 1, formatJSON, prometheusGrpcURL, logger) - require.Equal(t, int64(0), logger.count()) - _, err := q.Query(context.Background(), "qs", tm) + tm := time.Unix(1649092025, 515834) - require.Error(t, err) - code := grpcutil.ErrorToStatusCode(err) - require.Equal(t, codes.Code(testCase.expectedCode), code) - if testCase.expectedLogs { - require.Greater(t, logger.count(), int64(0)) - } else { require.Equal(t, int64(0), logger.count()) - } + _, err := q.Query(context.Background(), "qs", tm) + + require.Error(t, err) + code := grpcutil.ErrorToStatusCode(err) + require.Equal(t, codes.Code(testCase.expectedCode), code) + if testCase.expectedLogs { + require.Greater(t, logger.count(), int64(0)) + } else { + require.Equal(t, int64(0), logger.count()) + } + }) }) } } diff --git a/pkg/ruler/ruler_sync_queue_test.go b/pkg/ruler/ruler_sync_queue_test.go index 7556c093ff3..023e4eb92c7 100644 --- a/pkg/ruler/ruler_sync_queue_test.go +++ b/pkg/ruler/ruler_sync_queue_test.go @@ -8,6 +8,7 @@ import ( "slices" "sync" "testing" + "testing/synctest" "time" "github.com/grafana/dskit/services" @@ -73,57 +74,59 @@ func TestRulerSyncQueue_EnqueueAndPoll(t *testing.T) { } func TestRulerSyncQueue_ShouldNotNotifyChannelMoreFrequentlyThanPollFrequency(t *testing.T) { - ctx := context.Background() + synctest.Test(t, func(t *testing.T) { + ctx := context.Background() - const pollFrequency = time.Second - q := newRulerSyncQueue(pollFrequency) - require.NoError(t, services.StartAndAwaitRunning(ctx, q)) - t.Cleanup(func() { - require.NoError(t, services.StopAndAwaitTerminated(ctx, q)) - }) + const pollFrequency = time.Second + q := newRulerSyncQueue(pollFrequency) + require.NoError(t, services.StartAndAwaitRunning(ctx, q)) + t.Cleanup(func() { + require.NoError(t, services.StopAndAwaitTerminated(ctx, q)) + }) - done := make(chan struct{}, 1) - wg := sync.WaitGroup{} + done := make(chan struct{}, 1) + wg := sync.WaitGroup{} + + // Start a worker which continuously enqueue. + wg.Add(1) + go func() { + defer wg.Done() + userID := 0 - // Start a worker which continuously enqueue. - wg.Add(1) - go func() { - defer wg.Done() - userID := 0 - - for { - select { - case <-time.After(pollFrequency / 10): - q.enqueue(fmt.Sprintf("user-%d", userID)) - userID++ - case <-done: - return + for { + select { + case <-time.After(pollFrequency / 10): + q.enqueue(fmt.Sprintf("user-%d", userID)) + userID++ + case <-done: + return + } } - } - }() - - // Start a worker which continuously poll from the queue. - numPolls := 0 - wg.Add(1) - go func() { - defer wg.Done() - - for { - select { - case <-q.poll(): - numPolls++ - case <-done: - return + }() + + // Start a worker which continuously poll from the queue. + numPolls := 0 + wg.Add(1) + go func() { + defer wg.Done() + + for { + select { + case <-q.poll(): + numPolls++ + case <-done: + return + } } - } - }() + }() - // Run for 5x poll frequency and then stop the workers - time.Sleep(5 * pollFrequency) - close(done) - wg.Wait() + // Run for 5x poll frequency and then stop the workers + time.Sleep(5 * pollFrequency) + close(done) + wg.Wait() - assert.LessOrEqual(t, numPolls, 6) + assert.LessOrEqual(t, numPolls, 6) + }) } func TestRulerSyncQueue_Concurrency(t *testing.T) { diff --git a/pkg/ruler/ruler_test.go b/pkg/ruler/ruler_test.go index 38a2178dc71..d20c3487320 100644 --- a/pkg/ruler/ruler_test.go +++ b/pkg/ruler/ruler_test.go @@ -19,6 +19,7 @@ import ( "strings" "sync" "testing" + "testing/synctest" "time" "github.com/go-kit/log" @@ -1768,86 +1769,88 @@ func TestRuler_NotifySyncRulesAsync_ShouldNotTriggerRulesSyncingOnAllRulersWhenD namespace = "test" ) - var ( - ctx = context.Background() - logger = log.NewNopLogger() - rulerAddrMap = map[string]*Ruler{} - ) + synctest.Test(t, func(t *testing.T) { + var ( + ctx = context.Background() + logger = log.NewNopLogger() + rulerAddrMap = map[string]*Ruler{} + ) - // Create a filesystem backed storage. - bucketCfg := bucket.Config{StorageBackendConfig: bucket.StorageBackendConfig{Backend: "filesystem", Filesystem: filesystem.Config{Directory: t.TempDir()}}} - bucketClient, err := bucket.NewClient(ctx, bucketCfg, "ruler-storage", logger, nil) - require.NoError(t, err) + // Create a filesystem backed storage. + bucketCfg := bucket.Config{StorageBackendConfig: bucket.StorageBackendConfig{Backend: "filesystem", Filesystem: filesystem.Config{Directory: t.TempDir()}}} + bucketClient, err := bucket.NewClient(ctx, bucketCfg, "ruler-storage", logger, nil) + require.NoError(t, err) - store := bucketclient.NewBucketRuleStore(bucketClient, nil, logger) + store := bucketclient.NewBucketRuleStore(bucketClient, nil, logger) - // Create an in-memory ring backend. - kvStore, cleanUp := consul.NewInMemoryClient(ring.GetCodec(), logger, nil) - t.Cleanup(func() { assert.NoError(t, cleanUp.Close()) }) + // Create an in-memory ring backend. + kvStore, cleanUp := consul.NewInMemoryClient(ring.GetCodec(), logger, nil) + t.Cleanup(func() { assert.NoError(t, cleanUp.Close()) }) - // Create rulers. The rulers are configured with a very long polling interval - // so that they will not trigger after the initial sync. Once the ruler has started, - // the initial sync already occurred. - rulers := make([]*Ruler, numRulers) - regs := make([]*prometheus.Registry, numRulers) + // Create rulers. The rulers are configured with a very long polling interval + // so that they will not trigger after the initial sync. Once the ruler has started, + // the initial sync already occurred. + rulers := make([]*Ruler, numRulers) + regs := make([]*prometheus.Registry, numRulers) - for i := 0; i < len(rulers); i++ { - rulerAddr := fmt.Sprintf("ruler-%d", i) + for i := 0; i < len(rulers); i++ { + rulerAddr := fmt.Sprintf("ruler-%d", i) - rulerCfg := defaultRulerConfig(t) - rulerCfg.PollInterval = time.Hour - rulerCfg.OutboundSyncQueuePollInterval = 100 * time.Millisecond - rulerCfg.InboundSyncQueuePollInterval = 100 * time.Millisecond - rulerCfg.Ring.NumTokens = 128 - rulerCfg.Ring.Common.InstanceID = rulerAddr - rulerCfg.Ring.Common.InstanceAddr = rulerAddr - rulerCfg.Ring.Common.KVStore = kv.Config{Mock: kvStore} + rulerCfg := defaultRulerConfig(t) + rulerCfg.PollInterval = time.Hour + rulerCfg.OutboundSyncQueuePollInterval = 100 * time.Millisecond + rulerCfg.InboundSyncQueuePollInterval = 100 * time.Millisecond + rulerCfg.Ring.NumTokens = 128 + rulerCfg.Ring.Common.InstanceID = rulerAddr + rulerCfg.Ring.Common.InstanceAddr = rulerAddr + rulerCfg.Ring.Common.KVStore = kv.Config{Mock: kvStore} - limits := validation.MockOverrides(func(defaults *validation.Limits, _ map[string]*validation.Limits) { - defaults.RulerSyncRulesOnChangesEnabled = false - }) + limits := validation.MockOverrides(func(defaults *validation.Limits, _ map[string]*validation.Limits) { + defaults.RulerSyncRulesOnChangesEnabled = false + }) - regs[i] = prometheus.NewPedanticRegistry() - rulers[i] = prepareRuler(t, rulerCfg, store, withLimits(limits), withRulerAddrMap(rulerAddrMap), withRulerAddrAutomaticMapping(), withStart(), withPrometheusRegisterer(regs[i])) - } + regs[i] = prometheus.NewPedanticRegistry() + rulers[i] = prepareRuler(t, rulerCfg, store, withLimits(limits), withRulerAddrMap(rulerAddrMap), withRulerAddrAutomaticMapping(), withStart(), withPrometheusRegisterer(regs[i])) + } - // Pre-condition check: each ruler should have synced the rules once (at startup). - for _, reg := range regs { - verifySyncRulesMetrics(t, reg, 1, 0) - } + // Pre-condition check: each ruler should have synced the rules once (at startup). + for _, reg := range regs { + verifySyncRulesMetrics(t, reg, 1, 0) + } - // Pre-condition check: each ruler should have an updated view over the ring. - for _, reg := range regs { - verifyRingMembersMetric(t, reg, 2) - } + // Pre-condition check: each ruler should have an updated view over the ring. + for _, reg := range regs { + verifyRingMembersMetric(t, reg, 2) + } - // Create some rule groups in the storage. - for i := 0; i < numRuleGroups; i++ { - groupID := fmt.Sprintf("group-%d", i) - record := fmt.Sprintf("count:metric_%d", i) - expr := fmt.Sprintf("count(metric_%d)", i) + // Create some rule groups in the storage. + for i := 0; i < numRuleGroups; i++ { + groupID := fmt.Sprintf("group-%d", i) + record := fmt.Sprintf("count:metric_%d", i) + expr := fmt.Sprintf("count(metric_%d)", i) - require.NoError(t, store.SetRuleGroup(ctx, userID, namespace, createRuleGroup(groupID, userID, createRecordingRule(record, expr)))) - } + require.NoError(t, store.SetRuleGroup(ctx, userID, namespace, createRuleGroup(groupID, userID, createRecordingRule(record, expr)))) + } - // Call NotifySyncRulesAsync() on 1 ruler. - rulers[0].NotifySyncRulesAsync("user-1") + // Call NotifySyncRulesAsync() on 1 ruler. + rulers[0].NotifySyncRulesAsync("user-1") - // Give rulers enough time to eventually re-sync based on config change, if it was enabled (but it's not). - // Unfortunately there's no better to way than waiting some time, since we're waiting for a condition to NOT happen. - time.Sleep(time.Second) + // Give rulers enough time to eventually re-sync based on config change, if it was enabled (but it's not). + // Unfortunately there's no better to way than waiting some time, since we're waiting for a condition to NOT happen. + time.Sleep(time.Second) - // Ensure no rules syncing has been triggered in any ruler. - for _, reg := range regs { - verifySyncRulesMetrics(t, reg, 1, 0) - } + // Ensure no rules syncing has been triggered in any ruler. + for _, reg := range regs { + verifySyncRulesMetrics(t, reg, 1, 0) + } - // GetRules() should return no configured rule groups, because no re-sync happened. - for _, ruler := range rulers { - list, _, err := ruler.GetRules(user.InjectOrgID(ctx, userID), RulesRequest{Filter: AnyRule}) - require.NoError(t, err) - require.Empty(t, list.Groups) - } + // GetRules() should return no configured rule groups, because no re-sync happened. + for _, ruler := range rulers { + list, _, err := ruler.GetRules(user.InjectOrgID(ctx, userID), RulesRequest{Filter: AnyRule}) + require.NoError(t, err) + require.Empty(t, list.Groups) + } + }) } // User shuffle shard token. diff --git a/pkg/storage/indexheader/reader_pool_test.go b/pkg/storage/indexheader/reader_pool_test.go index 11aae8f0033..7815530f08c 100644 --- a/pkg/storage/indexheader/reader_pool_test.go +++ b/pkg/storage/indexheader/reader_pool_test.go @@ -11,6 +11,7 @@ import ( "os" "path/filepath" "testing" + "testing/synctest" "time" "github.com/go-kit/log" @@ -83,45 +84,47 @@ func TestReaderPool_ShouldCloseIdleLazyReaders(t *testing.T) { ctx, tmpDir, bkt, blockID, metrics := prepareReaderPool(t) defer func() { require.NoError(t, os.RemoveAll(tmpDir)) }() - // Note that we are creating a ReaderPool that doesn't run a background cleanup task for idle - // Reader instances. We'll manually invoke the cleanup task when we need it as part of this test. - pool := newReaderPool(log.NewNopLogger(), Config{ - LazyLoadingEnabled: true, - LazyLoadingIdleTimeout: idleTimeout, - }, gate.NewNoop(), metrics) - - r, err := pool.NewBinaryReader(ctx, log.NewNopLogger(), bkt, tmpDir, blockID, 3, Config{}) - require.NoError(t, err) - - // Ensure it can read data. - labelNames, err := r.LabelNames(ctx) - require.NoError(t, err) - require.Equal(t, []string{"a"}, labelNames) - require.Equal(t, float64(1), promtestutil.ToFloat64(metrics.lazyReader.loadCount)) - require.Equal(t, float64(0), promtestutil.ToFloat64(metrics.lazyReader.unloadCount)) - - // Wait enough time before checking it. - time.Sleep(idleTimeout * 2) - require.NoError(t, pool.unloadIdleReaders(context.Background()), "closing idle readers shouldn't ever fail because it will abort periodically checking for idle readers") - - // We expect the reader has been closed, but not released from the pool. - require.True(t, pool.isTracking(r.(*LazyBinaryReader))) - require.Equal(t, float64(1), promtestutil.ToFloat64(metrics.lazyReader.loadCount)) - require.Equal(t, float64(1), promtestutil.ToFloat64(metrics.lazyReader.unloadCount)) - - // Ensure it can still read data (will be re-opened). - labelNames, err = r.LabelNames(ctx) - require.NoError(t, err) - require.Equal(t, []string{"a"}, labelNames) - require.True(t, pool.isTracking(r.(*LazyBinaryReader))) - require.Equal(t, float64(2), promtestutil.ToFloat64(metrics.lazyReader.loadCount)) - require.Equal(t, float64(1), promtestutil.ToFloat64(metrics.lazyReader.unloadCount)) - - // We expect an explicit call to Close() to close the reader and release it from the pool too. - require.NoError(t, r.Close()) - require.True(t, !pool.isTracking(r.(*LazyBinaryReader))) - require.Equal(t, float64(2), promtestutil.ToFloat64(metrics.lazyReader.loadCount)) - require.Equal(t, float64(2), promtestutil.ToFloat64(metrics.lazyReader.unloadCount)) + synctest.Test(t, func(t *testing.T) { + // Note that we are creating a ReaderPool that doesn't run a background cleanup task for idle + // Reader instances. We'll manually invoke the cleanup task when we need it as part of this test. + pool := newReaderPool(log.NewNopLogger(), Config{ + LazyLoadingEnabled: true, + LazyLoadingIdleTimeout: idleTimeout, + }, gate.NewNoop(), metrics) + + r, err := pool.NewBinaryReader(ctx, log.NewNopLogger(), bkt, tmpDir, blockID, 3, Config{}) + require.NoError(t, err) + + // Ensure it can read data. + labelNames, err := r.LabelNames(ctx) + require.NoError(t, err) + require.Equal(t, []string{"a"}, labelNames) + require.Equal(t, float64(1), promtestutil.ToFloat64(metrics.lazyReader.loadCount)) + require.Equal(t, float64(0), promtestutil.ToFloat64(metrics.lazyReader.unloadCount)) + + // Wait enough time before checking it. + time.Sleep(idleTimeout * 2) + require.NoError(t, pool.unloadIdleReaders(context.Background()), "closing idle readers shouldn't ever fail because it will abort periodically checking for idle readers") + + // We expect the reader has been closed, but not released from the pool. + require.True(t, pool.isTracking(r.(*LazyBinaryReader))) + require.Equal(t, float64(1), promtestutil.ToFloat64(metrics.lazyReader.loadCount)) + require.Equal(t, float64(1), promtestutil.ToFloat64(metrics.lazyReader.unloadCount)) + + // Ensure it can still read data (will be re-opened). + labelNames, err = r.LabelNames(ctx) + require.NoError(t, err) + require.Equal(t, []string{"a"}, labelNames) + require.True(t, pool.isTracking(r.(*LazyBinaryReader))) + require.Equal(t, float64(2), promtestutil.ToFloat64(metrics.lazyReader.loadCount)) + require.Equal(t, float64(1), promtestutil.ToFloat64(metrics.lazyReader.unloadCount)) + + // We expect an explicit call to Close() to close the reader and release it from the pool too. + require.NoError(t, r.Close()) + require.True(t, !pool.isTracking(r.(*LazyBinaryReader))) + require.Equal(t, float64(2), promtestutil.ToFloat64(metrics.lazyReader.loadCount)) + require.Equal(t, float64(2), promtestutil.ToFloat64(metrics.lazyReader.unloadCount)) + }) } func TestReaderPool_LoadedBlocks(t *testing.T) { diff --git a/pkg/storegateway/gateway_test.go b/pkg/storegateway/gateway_test.go index a3b88d30034..68f579fea26 100644 --- a/pkg/storegateway/gateway_test.go +++ b/pkg/storegateway/gateway_test.go @@ -20,6 +20,7 @@ import ( "strings" "sync" "testing" + "testing/synctest" "time" "github.com/go-kit/log" @@ -220,25 +221,27 @@ func TestStoreGateway_InitialSyncWithDefaultShardingEnabled(t *testing.T) { func TestStoreGateway_InitialSyncFailure(t *testing.T) { test.VerifyNoLeak(t) - ctx := context.Background() - gatewayCfg := mockGatewayConfig() - storageCfg := mockStorageConfig(t) - ringStore, closer := consul.NewInMemoryClient(ring.GetCodec(), log.NewNopLogger(), nil) - t.Cleanup(func() { assert.NoError(t, closer.Close()) }) + synctest.Test(t, func(t *testing.T) { + ctx := context.Background() + gatewayCfg := mockGatewayConfig() + storageCfg := mockStorageConfig(t) + ringStore, closer := consul.NewInMemoryClient(ring.GetCodec(), log.NewNopLogger(), nil) + t.Cleanup(func() { assert.NoError(t, closer.Close()) }) - bucketClient := &bucket.ErrorInjectedBucketClient{Injector: func(bucket.Operation, string) error { return assert.AnError }} + bucketClient := &bucket.ErrorInjectedBucketClient{Injector: func(bucket.Operation, string) error { return assert.AnError }} - g, err := newStoreGateway(gatewayCfg, storageCfg, bucketClient, ringStore, defaultLimitsOverrides(t), log.NewLogfmtLogger(os.Stdout), nil, nil) - require.NoError(t, err) + g, err := newStoreGateway(gatewayCfg, storageCfg, bucketClient, ringStore, defaultLimitsOverrides(t), log.NewLogfmtLogger(os.Stdout), nil, nil) + require.NoError(t, err) - require.NoError(t, g.StartAsync(ctx)) - err = g.AwaitRunning(ctx) - assert.Error(t, err) - assert.Equal(t, services.Failed, g.State()) + require.NoError(t, g.StartAsync(ctx)) + err = g.AwaitRunning(ctx) + assert.Error(t, err) + assert.Equal(t, services.Failed, g.State()) - // We expect a clean shutdown, including unregistering the instance from the ring. - assert.False(t, g.ringLifecycler.IsRegistered()) - _ = services.StopAndAwaitTerminated(ctx, g) // There will be an error since the initial sync failed + // We expect a clean shutdown, including unregistering the instance from the ring. + assert.False(t, g.ringLifecycler.IsRegistered()) + _ = services.StopAndAwaitTerminated(ctx, g) // There will be an error since the initial sync failed + }) } // TestStoreGateway_InitialSyncWithWaitRingTokensStability tests the store-gateway cold start case.