Function TestPushRace

pkg/ingester/ingester_test.go:666–752

(t *testing.T)

}

func TestPushRace(t *testing.T) {
	cfg := defaultIngesterTestConfig(t)
	l := defaultLimitsTestConfig()
	l.EnableNativeHistograms = true
	cfg.LabelsStringInterningEnabled = true
	cfg.LifecyclerConfig.JoinAfter = 0

	l.LimitsPerLabelSet = []validation.LimitsPerLabelSet{
		{
			LabelSet: labels.FromMap(map[string]string{
				labels.MetricName: "foo",
			}),
			Limits: validation.LimitsPerLabelSetEntry{
				MaxSeries: 10e10,
			},
		},
		{
			// Default partition.
			LabelSet: labels.EmptyLabels(),
			Limits: validation.LimitsPerLabelSetEntry{
				MaxSeries: 10e10,
			},
		},
	}

	dir := t.TempDir()
	blocksDir := filepath.Join(dir, "blocks")
	require.NoError(t, os.Mkdir(blocksDir, os.ModePerm))

	ing, err := prepareIngesterWithBlocksStorageAndLimits(t, cfg, l, nil, blocksDir, prometheus.NewRegistry())
	require.NoError(t, err)
	defer services.StopAndAwaitTerminated(context.Background(), ing) //nolint:errcheck
	require.NoError(t, services.StartAndAwaitRunning(context.Background(), ing))
	// Wait until it's ACTIVE
	test.Poll(t, time.Second, ring.ACTIVE, func() any {
		return ing.lifecycler.GetState()
	})

	ctx := user.InjectOrgID(context.Background(), userID)
	sample1 := cortexpb.Sample{
		TimestampMs: 0,
		Value:       1,
	}

	concurrentRequest := 100
	numberOfSeries := 100
	wg := sync.WaitGroup{}
	wg.Add(numberOfSeries * concurrentRequest)
	for k := range numberOfSeries {
		for range concurrentRequest {
			go func() {
				defer wg.Done()
				_, err := ing.Push(ctx, cortexpb.ToWriteRequest([]labels.Labels{labels.FromStrings(labels.MetricName, "foo", "userId", userID, "k", strconv.Itoa(k))}, []cortexpb.Sample{sample1}, nil, nil, cortexpb.API))
				require.NoError(t, err)

				// Go to default partition.
				_, err = ing.Push(ctx, cortexpb.ToWriteRequest([]labels.Labels{labels.FromStrings(labels.MetricName, "bar", "userId", userID, "k", strconv.Itoa(k))}, []cortexpb.Sample{sample1}, nil, nil, cortexpb.API))
				require.NoError(t, err)