Function TestDropOldTimeSeries

storage/remote/queue_manager_test.go:1915–1963

(t *testing.T)

}

func TestDropOldTimeSeries(t *testing.T) {
	t.Parallel()
	// Test both v1 and v2 remote write protocols
	for _, protoMsg := range []remoteapi.WriteMessageType{remoteapi.WriteV1MessageType, remoteapi.WriteV2MessageType} {
		t.Run(fmt.Sprint(protoMsg), func(t *testing.T) {
			size := 10
			nSeries := 6
			nSamples := config.DefaultQueueConfig.Capacity * size
			noST := protoMsg == remoteapi.WriteV1MessageType // RW1
			pastRecs := testwal.GenerateRecords(recCase{
				NoST:             noST,
				Series:           nSeries,
				SamplesPerSeries: (nSamples / nSeries) / 2, // Half data is past.
				TsFn: func(_, j int) int64 {
					past := timestamp.FromTime(time.Now().Add(-5 * time.Minute))
					return past + int64(j)
				},
			})
			newRecs := testwal.GenerateRecords(recCase{
				NoST:             noST,
				Series:           nSeries,
				SamplesPerSeries: (nSamples / nSeries) / 2, // Half data is past.
				TsFn: func(_, j int) int64 {
					return time.Now().UnixMilli() + int64(j)
				},
			})

			series := pastRecs.Series // Series is the same for both old and new.
			newSamples := newRecs.Samples
			samples := append(pastRecs.Samples, newRecs.Samples...)

			c := NewTestWriteClient(protoMsg)
			c.expectSamples(newSamples, series)

			cfg := config.DefaultQueueConfig
			mcfg := config.DefaultMetadataConfig
			cfg.MaxShards = 1
			cfg.SampleAgeLimit = model.Duration(60 * time.Second)
			m := newTestQueueManager(t, cfg, mcfg, defaultFlushDeadline, c, protoMsg)
			m.StoreSeries(series, 0)

			m.Start()
			defer m.Stop()

			m.Append(samples)
			c.waitForExpectedData(t, 30*time.Second)
		})
	}
}

func TestIsSampleOld(t *testing.T) {
	currentTime := time.Now()