Function TestShutdown

storage/remote/queue_manager_test.go:445–482

(t *testing.T)

}

func TestShutdown(t *testing.T) {
	t.Parallel()
	for _, protoMsg := range []remoteapi.WriteMessageType{remoteapi.WriteV1MessageType, remoteapi.WriteV2MessageType} {
		t.Run(fmt.Sprint(protoMsg), func(t *testing.T) {
			synctest.Test(t, func(t *testing.T) {
				deadline := 15 * time.Second
				c := NewTestBlockedWriteClient()

				cfg := config.DefaultQueueConfig
				mcfg := config.DefaultMetadataConfig

				m := newTestQueueManager(t, cfg, mcfg, deadline, c, protoMsg)
				// Send 2x batch size, so we know it will need at least two sends.
				n := 2 * config.DefaultQueueConfig.MaxSamplesPerSend
				recs := testwal.GenerateRecords(recCase{
					NoST:   protoMsg == remoteapi.WriteV1MessageType, // RW1 does not support ST.
					Series: n / 1000, SamplesPerSeries: 1000,
				})
				m.StoreSeries(recs.Series, 0)
				m.Start()

				// Append blocks to guarantee delivery, so we do it in the background.
				go func() {
					m.Append(recs.Samples)
				}()
				synctest.Wait()

				// Test to ensure that Stop doesn't block.
				start := time.Now()
				m.Stop()
				// The samples will never be delivered, so duration should
				// be at least equal to deadline, otherwise the flush deadline
				// was not respected.
				require.Equal(t, time.Since(start), deadline)
			})
		})
	}
}

func TestSeriesReset(t *testing.T) {
	for _, protoMsg := range []remoteapi.WriteMessageType{remoteapi.WriteV1MessageType, remoteapi.WriteV2MessageType} {