Function TestWALMetadataDelivery

storage/remote/queue_manager_test.go:348–389

(t *testing.T)

}

func TestWALMetadataDelivery(t *testing.T) {
	dir := t.TempDir()
	s := NewStorage(nil, nil, nil, dir, defaultFlushDeadline, nil, false)
	defer s.Close()

	cfg := config.DefaultQueueConfig
	cfg.BatchSendDeadline = model.Duration(100 * time.Millisecond)
	cfg.MaxShards = 1

	writeConfig := baseRemoteWriteConfig("http://test-storage.com")
	writeConfig.QueueConfig = cfg
	writeConfig.ProtobufMessage = remoteapi.WriteV2MessageType

	conf := &config.Config{
		GlobalConfig: config.DefaultGlobalConfig,
		RemoteWriteConfigs: []*config.RemoteWriteConfig{
			writeConfig,
		},
	}

	n := 3
	recs := testwal.GenerateRecords(recCase{Series: n, SamplesPerSeries: n})

	require.NoError(t, s.ApplyConfig(conf))
	hash, err := toHash(writeConfig)
	require.NoError(t, err)
	qm := s.rws.queues[hash]

	c := NewTestWriteClient(remoteapi.WriteV2MessageType)
	qm.SetClient(c)

	qm.StoreSeries(recs.Series, 0)
	qm.StoreMetadata(recs.Metadata)

	require.Len(t, qm.seriesLabels, n)
	require.Len(t, qm.seriesMetadata, n)

	c.expectSamples(recs.Samples, recs.Series)
	c.expectMetadataForBatch(recs.Metadata, recs.Series, recs.Samples, nil, nil, nil)
	qm.Append(recs.Samples)
	c.waitForExpectedData(t, 30*time.Second)
}

func TestSampleDeliveryTimeout(t *testing.T) {
	t.Parallel()