Function TestInstance_multiTenancy

modules/generator/storage/instance_test.go:108–182

Verify multiple instances function next to each other, don't trample over each other and are isolated.

(t *testing.T)

// Verify multiple instances function next to each other, don't trample over each other and are isolated.
func TestInstance_multiTenancy(t *testing.T) {
	t.Parallel()
	var err error
	logger := log.NewLogfmtLogger(log.NewSyncWriter(os.Stdout))

	mockServer := newMockPrometheusRemoteWriterServer(logger)
	defer mockServer.close()

	var cfg Config
	cfg.RegisterFlagsAndApplyDefaults("", nil)
	cfg.Path = t.TempDir()
	cfg.RemoteWrite = mockServer.remoteWriteConfig()

	var instances []Storage

	for i := 0; i < 3; i++ {
		instance, err := New(&cfg, &mockOverrides{}, strconv.Itoa(i), &noopRegisterer{}, logger)
		assert.NoError(t, err)
		instances = append(instances, instance)
	}

	sendCtx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// Append some data every second
	go poll(sendCtx, time.Second, func() {
		for i, instance := range instances {
			appender := instance.Appender(context.Background())

			lbls := labels.FromMap(map[string]string{"__name__": "my-metric"})
			_, err := appender.Append(0, lbls, time.Now().UnixMilli(), float64(i))
			assert.NoError(t, err)

			if sendCtx.Err() != nil {
				return
			}

			err = appender.Commit()
			assert.NoError(t, err)
		}
	})

	// Wait until every tenant received at least one request
	test.Poll(t, 45*time.Second, true, func() interface{} {
		mockServer.mtx.Lock()
		defer mockServer.mtx.Unlock()

		for i := range instances {
			if mockServer.acceptedRequests[strconv.Itoa(i)] == 0 {
				return false
			}
		}
		return true
	})

	cancel()
	for _, instance := range instances {
		// Shutdown the instance - remote write should flush pending data