Function TestSampleBufferConcurrently

output/helpers_test.go:54–116

(t *testing.T)

}

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

	seed := time.Now().UnixNano()
	r := rand.New(rand.NewSource(seed))
	t.Logf("Random source seeded with %d\n", seed)

	registry := metrics.NewRegistry()
	metric, err := registry.NewMetric("my_metric", metrics.Gauge)
	require.NoError(t, err)

	producersCount := 50 + r.Intn(50)
	sampleCount := 10 + r.Intn(10)
	sleepModifier := 10 + r.Intn(10)
	buffer := SampleBuffer{}

	wg := make(chan struct{})
	fillBuffer := func() {
		for i := range sampleCount {
			buffer.AddMetricSamples([]metrics.SampleContainer{metrics.Sample{
				TimeSeries: metrics.TimeSeries{
					Metric: metric,
					Tags:   registry.RootTagSet().WithTagsFromMap(map[string]string{"tag1": "val1"}),
				},
				Time:  time.Unix(1562324644, 0),
				Value: float64(i),
			}})
			time.Sleep(time.Duration(i*sleepModifier) * time.Microsecond)
		}
		wg <- struct{}{}
	}
	for range producersCount {
		go fillBuffer()
	}

	timer := time.NewTicker(5 * time.Millisecond)
	timeout := time.After(5 * time.Second)
	defer timer.Stop()
	readSamples := make([]metrics.SampleContainer, 0, sampleCount*producersCount)
	finishedProducers := 0
loop:
	for {
		select {
		case <-timer.C:
			readSamples = append(readSamples, buffer.GetBufferedSamples()...)
		case <-wg:
			finishedProducers++
			if finishedProducers == producersCount {
				readSamples = append(readSamples, buffer.GetBufferedSamples()...)
				break loop
			}
		case <-timeout:
			t.Fatalf("test timed out")
		}
	}
	assert.Equal(t, sampleCount*producersCount, len(readSamples))
	for _, s := range readSamples {
		require.NotNil(t, s)