Function TestIngester_inflightPushRequests

pkg/ingester/ingester_test.go:6821–6877

(t *testing.T)

}

func TestIngester_inflightPushRequests(t *testing.T) {
	limits := InstanceLimits{MaxInflightPushRequests: 1}

	// Create a mocked ingester
	cfg := defaultIngesterTestConfig(t)
	cfg.InstanceLimitsFn = func() *InstanceLimits { return &limits }
	cfg.LifecyclerConfig.JoinAfter = 0

	reg := prometheus.NewRegistry()
	i, err := prepareIngesterWithBlocksStorage(t, cfg, reg)
	require.NoError(t, err)
	require.NoError(t, services.StartAndAwaitRunning(context.Background(), i))
	defer services.StopAndAwaitTerminated(context.Background(), i) //nolint:errcheck

	// Wait until the ingester is ACTIVE
	test.Poll(t, 100*time.Millisecond, ring.ACTIVE, func() any {
		return i.lifecycler.GetState()
	})

	ctx := user.InjectOrgID(context.Background(), "test")

	startCh := make(chan struct{})

	g, ctx := errgroup.WithContext(ctx)
	g.Go(func() error {
		count := 150000
		req := generateSamplesForLabel(labels.FromStrings(labels.MetricName, fmt.Sprintf("real-%d", count)), count, 1)
		// Signal that we're going to do the real push now.
		close(startCh)

		_, err := i.Push(ctx, req)
		return err
	})

	g.Go(func() error {
		select {
		case <-ctx.Done():
		// failed to setup
		case <-startCh:
			// we can start the test.
		}

		time.Sleep(10 * time.Millisecond) // Give first goroutine a chance to start pushing...
		req := generateSamplesForLabel(labels.FromStrings(labels.MetricName, "testcase"), 1024, 1)

		_, err := i.Push(ctx, req)
		require.Equal(t, errTooManyInflightPushRequests, err)
		require.NoError(t, testutil.GatherAndCompare(reg, bytes.NewBufferString(`
		# HELP cortex_ingester_push_errors_total The total number of push errors per user.
		# TYPE cortex_ingester_push_errors_total counter
		cortex_ingester_push_errors_total{reason="tooManyInflightRequests",user="test"} 1
	`), "cortex_ingester_push_errors_total"))
		return nil
	})

	require.NoError(t, g.Wait())
}