Function TestChunkReader_ConcurrentReads

tsdb/db_test.go:3237–3274

TestChunkReader_ConcurrentReads checks that the chunk result can be read concurrently. Regression test for https://github.com/prometheus/prometheus/pull/6514.

(t *testing.T)

// TestChunkReader_ConcurrentReads checks that the chunk result can be read concurrently.
// Regression test for https://github.com/prometheus/prometheus/pull/6514.
func TestChunkReader_ConcurrentReads(t *testing.T) {
	t.Parallel()
	chks := []chunks.Meta{
		assureChunkFromSamples(t, []chunks.Sample{sample{0, 1, 1, nil, nil}}),
		assureChunkFromSamples(t, []chunks.Sample{sample{0, 1, 2, nil, nil}}),
		assureChunkFromSamples(t, []chunks.Sample{sample{0, 1, 3, nil, nil}}),
		assureChunkFromSamples(t, []chunks.Sample{sample{0, 1, 4, nil, nil}}),
		assureChunkFromSamples(t, []chunks.Sample{sample{0, 1, 5, nil, nil}}),
	}

	tempDir := t.TempDir()

	chunkw, err := chunks.NewWriter(tempDir)
	require.NoError(t, err)

	require.NoError(t, chunkw.WriteChunks(chks...))
	require.NoError(t, chunkw.Close())

	r, err := chunks.NewDirReader(tempDir, nil)
	require.NoError(t, err)

	var wg sync.WaitGroup
	for _, chk := range chks {
		for range 100 {
			wg.Add(1)
			go func(chunk chunks.Meta) {
				defer wg.Done()

				chkAct, iterable, err := r.ChunkOrIterable(chunk)
				require.NoError(t, err)
				require.Nil(t, iterable)
				require.Equal(t, chunk.Chunk.Bytes(), chkAct.Bytes())
			}(chk)
		}
		wg.Wait()
	}
	require.NoError(t, r.Close())
}

// TestCompactHead ensures that the head compaction
// creates a block that is ready for loading and