Function TestDisableReshardOnRetry

storage/remote/queue_manager_test.go:766–826

TestDisableReshardOnRetry asserts that resharding should be disabled when a recoverable error is returned from remote_write.

(t *testing.T)

// TestDisableReshardOnRetry asserts that resharding should be disabled when a
// recoverable error is returned from remote_write.
func TestDisableReshardOnRetry(t *testing.T) {
	t.Parallel()
	onStoredContext, onStoreCalled := context.WithCancel(context.Background())
	defer onStoreCalled()

	var (
		recs = testwal.GenerateRecords(recCase{Series: 100, SamplesPerSeries: 100})

		cfg        = config.DefaultQueueConfig
		mcfg       = config.DefaultMetadataConfig
		retryAfter = time.Second

		metrics = newQueueManagerMetrics(nil, "", "")

		client = &MockWriteClient{
			StoreFunc: func(context.Context, []byte, int) (WriteResponseStats, error) {
				onStoreCalled()

				return WriteResponseStats{}, RecoverableError{
					error:      errors.New("fake error"),
					retryAfter: model.Duration(retryAfter),
				}
			},
			NameFunc:     func() string { return "mock" },
			EndpointFunc: func() string { return "http://fake:9090/api/v1/write" },
		}
	)

	m := NewQueueManager(metrics, nil, nil, nil, "", newEWMARate(ewmaWeight, shardUpdateDuration), cfg, mcfg, labels.EmptyLabels(), nil, client, 0, newPool(), newHighestTimestampMetric(), nil, false, false, false, remoteapi.WriteV1MessageType, nil)
	m.StoreSeries(recs.Series, 0)

	// Attempt to samples while the manager is running. We immediately stop the
	// manager after the recoverable error is generated to prevent the manager
	// from resharding itself.
	m.Start()
	{
		m.Append(recs.Samples)

		select {
		case <-onStoredContext.Done():
		case <-time.After(time.Minute):
			require.FailNow(t, "timed out waiting for client to be sent metrics")
		}
	}
	m.Stop()

	require.Eventually(t, func() bool {
		// Force m.lastSendTimestamp to be current so the last send timestamp isn't
		// the reason resharding is disabled.
		m.lastSendTimestamp.Store(time.Now().Unix())
		return m.shouldReshard(m.numShards+1) == false
	}, time.Minute, 10*time.Millisecond, "shouldReshard was never disabled")

	// After 2x retryAfter, resharding should be enabled again.
	require.Eventually(t, func() bool {
		// Force m.lastSendTimestamp to be current so the last send timestamp isn't
		// the reason resharding is disabled.
		m.lastSendTimestamp.Store(time.Now().Unix())