Function TestSampleDeliveryTimeout

storage/remote/queue_manager_test.go:391–419

(t *testing.T)

}

func TestSampleDeliveryTimeout(t *testing.T) {
	t.Parallel()
	for _, protoMsg := range []remoteapi.WriteMessageType{remoteapi.WriteV1MessageType, remoteapi.WriteV2MessageType} {
		t.Run(fmt.Sprint(protoMsg), func(t *testing.T) {
			recs := testwal.GenerateRecords(recCase{
				NoST:   protoMsg == remoteapi.WriteV1MessageType, // RW1 does not support ST.
				Series: 10, SamplesPerSeries: 10,
			})
			cfg := testDefaultQueueConfig()
			mcfg := config.DefaultMetadataConfig
			cfg.MaxShards = 1

			c := NewTestWriteClient(protoMsg)
			m := newTestQueueManager(t, cfg, mcfg, defaultFlushDeadline, c, protoMsg)
			m.StoreSeries(recs.Series, 0)
			m.Start()
			defer m.Stop()

			// Send the samples twice, waiting for the samples in the meantime.
			c.expectSamples(recs.Samples, recs.Series)
			m.Append(recs.Samples)
			c.waitForExpectedData(t, 30*time.Second)

			c.expectSamples(recs.Samples, recs.Series)
			m.Append(recs.Samples)
			c.waitForExpectedData(t, 30*time.Second)
		})
	}
}

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