Function TestSampleDelivery

storage/remote/queue_manager_test.go:197–297

(t *testing.T)

}

func TestSampleDelivery(t *testing.T) {
	t.Parallel()
	n := 3

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

	// We need to set URL's so that metric creation doesn't panic.
	writeConfig := baseRemoteWriteConfig("http://test-storage.com")
	writeConfig.QueueConfig = queueConfig
	writeConfig.SendExemplars = true
	writeConfig.SendNativeHistograms = true

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

	for _, protoMsg := range []remoteapi.WriteMessageType{remoteapi.WriteV1MessageType, remoteapi.WriteV2MessageType} {
		for _, rc := range []recCase{
			{Series: n, SamplesPerSeries: n, HistogramsPerSeries: 0, FloatHistogramsPerSeries: 0, ExemplarsPerSeries: 0, Name: "samples only"},
			{Series: n, SamplesPerSeries: 0, HistogramsPerSeries: n, FloatHistogramsPerSeries: 0, ExemplarsPerSeries: 0, Name: "histograms only"},
			{Series: n, SamplesPerSeries: 0, HistogramsPerSeries: 0, FloatHistogramsPerSeries: n, ExemplarsPerSeries: 0, Name: "float histograms only"},
			{Series: n, SamplesPerSeries: 0, HistogramsPerSeries: 0, FloatHistogramsPerSeries: 0, ExemplarsPerSeries: n, Name: "exemplars only"},
			{Series: n, SamplesPerSeries: n, HistogramsPerSeries: n, FloatHistogramsPerSeries: n, ExemplarsPerSeries: n, Name: "all"},
		} {
			t.Run(fmt.Sprintf("proto=%s/case=%s", protoMsg, rc.Name), func(t *testing.T) {
				dir := t.TempDir()
				s := NewStorage(nil, nil, nil, dir, defaultFlushDeadline, nil, false)
				defer s.Close()

				rc.NoST = protoMsg == remoteapi.WriteV1MessageType // RW1 does not support ST.
				recs := testwal.GenerateRecords(rc)

				var (
					series          = recs.Series
					metadata        = recs.Metadata
					samples         = recs.Samples
					exemplars       = recs.Exemplars
					histograms      = recs.Histograms
					floatHistograms = recs.FloatHistograms
				)

				// Apply new config.
				queueConfig.Capacity = n
				queueConfig.MaxSamplesPerSend = n / 2
				conf.RemoteWriteConfigs[0].ProtobufMessage = protoMsg
				require.NoError(t, s.ApplyConfig(conf))
				hash, err := toHash(writeConfig)
				require.NoError(t, err)
				qm := s.rws.queues[hash]

				c := NewTestWriteClient(protoMsg)
				c.SetStoreWait(10 * time.Millisecond)
				qm.SetClient(c)