Function TestExpendedPostingsCacheIsolation

pkg/ingester/ingester_test.go:6255–6322

(t *testing.T)

}

func TestExpendedPostingsCacheIsolation(t *testing.T) {
	cfg := defaultIngesterTestConfig(t)
	cfg.BlocksStorageConfig.TSDB.BlockRanges = []time.Duration{2 * time.Hour}
	cfg.LifecyclerConfig.JoinAfter = 0
	cfg.BlocksStorageConfig.TSDB.PostingsCache = cortex_tsdb.TSDBPostingsCacheConfig{
		SeedSize: 3, // lets make sure all metric names collide
		Head: cortex_tsdb.PostingsCacheConfig{
			Enabled:  true,
			Ttl:      time.Hour,
			MaxBytes: 1024 * 1024 * 1024,
		},
		Blocks: cortex_tsdb.PostingsCacheConfig{
			Enabled:  true,
			Ttl:      time.Hour,
			MaxBytes: 1024 * 1024 * 1024,
		},
	}

	r := prometheus.NewRegistry()
	i, err := prepareIngesterWithBlocksStorage(t, cfg, r)
	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()
	})

	numberOfTenants := 100
	wg := sync.WaitGroup{}

	for k := range 10 {
		wg.Add(numberOfTenants)
		for j := range numberOfTenants {
			go func() {
				defer wg.Done()
				userId := fmt.Sprintf("user%v", j)
				ctx := user.InjectOrgID(context.Background(), userId)
				_, err := i.Push(ctx, cortexpb.ToWriteRequest(
					[]labels.Labels{labels.FromStrings(labels.MetricName, "foo", "userId", userId, "k", strconv.Itoa(k))}, []cortexpb.Sample{{Value: 2, TimestampMs: 4 * 60 * 60 * 1000}}, nil, nil, cortexpb.API))
				require.NoError(t, err)
			}()
		}
		wg.Wait()
	}

	wg.Add(numberOfTenants)
	for j := range numberOfTenants {
		go func() {
			defer wg.Done()
			userId := fmt.Sprintf("user%v", j)
			ctx := user.InjectOrgID(context.Background(), userId)
			s := &mockQueryStreamServer{ctx: ctx}

			err := i.QueryStream(&client.QueryRequest{
				StartTimestampMs: 0,
				EndTimestampMs:   math.MaxInt64,