Function TestInstance_remoteWriteHeaders

modules/generator/storage/instance_test.go:198–269

(t *testing.T)

}

func TestInstance_remoteWriteHeaders(t *testing.T) {
	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()

	headers := map[string]string{user.OrgIDHeaderName: "my-other-tenant"}

	instance, err := New(&cfg, &mockOverrides{headers, histograms.HistogramMethodClassic}, "test-tenant", &noopRegisterer{}, logger)
	require.NoError(t, err)

	// Refuse requests - the WAL should buffer data until requests succeed
	mockServer.refuseRequests.Store(true)

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

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

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

		_, err = appender.AppendExemplar(ref, lbls, exemplar.Exemplar{
			Labels: labels.FromMap(map[string]string{"traceID": "123"}),
			Value:  1.2,
		})
		assert.NoError(t, err)

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

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

	// Wait until remote.Storage has tried at least once to send data
	test.Poll(t, 30*time.Second, true, func() interface{} {
		mockServer.mtx.Lock()
		defer mockServer.mtx.Unlock()

		return mockServer.refusedRequests > 0
	})

	// Allow requests
	mockServer.refuseRequests.Store(false)

	// Shutdown the instance - even though previous requests failed, remote.Storage should flush pending data
	cancel()