receive/rule: Fixed Segfault issue; Added tests & benchmarks for TSDBStore, fixed multitsdb benchmarks.

Fixed #3013

Also:
* Fixed other quite big issue with reusing chunk slice.
* Fixed framing - previously it was wrongly sending single-chunk frames, taking
huge amount of time.

Before upgrading go.mod with latest TSDB:

 TestTSDBStore_SeriesChunkBytesCopied/flush_WAL_and_access_results: tsdb_test.go:487: tsdb_test.go:487:

         unexpected error: invoked function panicked or caused segmentation fault: runtime error: invalid memory address or nil pointer dereference

After all ok.

Signed-off-by: Bartlomiej Plotka <bwplotka@gmail.com>

Author: bwplotka

pkg/store/bucket_test.go

@@ -1178,7 +1178,7 @@ func benchBucketSeries(t testutil.TB, samplesPerSeries, totalSeries int, request
 	var (
 		logger = log.NewNopLogger()
 		blocks []*bucketBlock
-		series []storepb.Series
+		series []*storepb.Series
 		random = rand.New(rand.NewSource(120))
 	)
 
@@ -1213,7 +1213,7 @@ func benchBucketSeries(t testutil.TB, samplesPerSeries, totalSeries int, request
 	// This allows to pick time range that will correspond to number of series picked 1:1.
 	for bi := 0; bi < numOfBlocks; bi++ {
 		head, bSeries := storetestutil.CreateHeadWithSeries(t, bi, storetestutil.HeadGenOptions{
-			Dir:              tmpDir,
+			TSDBDir:          filepath.Join(tmpDir, fmt.Sprintf("%d", bi)),
 			SamplesPerSeries: samplesPerSeriesPerBlock,
 			Series:           seriesPerBlock,
 			PrependLabels:    extLset,
@@ -1536,17 +1536,22 @@ func TestSeries_RequestAndResponseHints(t *testing.T) {
 	}
 
 	// Create TSDB blocks.
-	opts := storetestutil.HeadGenOptions{
-		Dir:              tmpDir,
+	head, seriesSet1 := storetestutil.CreateHeadWithSeries(t, 0, storetestutil.HeadGenOptions{
+		TSDBDir:          filepath.Join(tmpDir, "0"),
 		SamplesPerSeries: 1,
 		Series:           2,
 		PrependLabels:    extLset,
 		Random:           random,
-	}
-	head, seriesSet1 := storetestutil.CreateHeadWithSeries(t, 0, opts)
+	})
 	block1 := createBlockFromHead(t, bktDir, head)
 	testutil.Ok(t, head.Close())
-	head2, seriesSet2 := storetestutil.CreateHeadWithSeries(t, 1, opts)
+	head2, seriesSet2 := storetestutil.CreateHeadWithSeries(t, 1, storetestutil.HeadGenOptions{
+		TSDBDir:          filepath.Join(tmpDir, "1"),
+		SamplesPerSeries: 1,
+		Series:           2,
+		PrependLabels:    extLset,
+		Random:           random,
+	})
 	block2 := createBlockFromHead(t, bktDir, head2)
 	testutil.Ok(t, head2.Close())
 
@@ -1610,7 +1615,7 @@ func TestSeries_RequestAndResponseHints(t *testing.T) {
 					{Type: storepb.LabelMatcher_EQ, Name: "foo", Value: "bar"},
 				},
 			},
-			ExpectedSeries: append(append([]storepb.Series{}, seriesSet1...), seriesSet2...),
+			ExpectedSeries: append(append([]*storepb.Series{}, seriesSet1...), seriesSet2...),
 			ExpectedHints: []hintspb.SeriesResponseHints{
 				{
 					QueriedBlocks: []hintspb.Block{

pkg/store/multitsdb_test.go

@@ -92,29 +92,24 @@ func benchMultiTSDBSeries(t testutil.TB, totalSamples, totalSeries int, flushToB
 		}
 	}()
 	for j := range dbs {
+		tsdbDir := filepath.Join(tmpDir, fmt.Sprintf("%d", j))
+
 		head, created := storetestutil.CreateHeadWithSeries(t, j, storetestutil.HeadGenOptions{
-			Dir:              tmpDir,
+			TSDBDir:          tsdbDir,
 			SamplesPerSeries: samplesPerSeriesPerTSDB,
 			Series:           seriesPerTSDB,
-			WithWAL:          true,
+			WithWAL:          !flushToBlocks,
 			Random:           random,
 			SkipChunks:       t.IsBenchmark(),
 		})
-		testutil.Ok(t, head.Close())
-
-		tsdbDir := filepath.Join(tmpDir, fmt.Sprintf("%d", j))
-
 		for i := 0; i < len(created); i++ {
-			resps[j] = append(resps[j], storepb.NewSeriesResponse(&created[i]))
+			resps[j] = append(resps[j], storepb.NewSeriesResponse(created[i]))
 		}
 
 		if flushToBlocks {
-			db, err := tsdb.OpenDBReadOnly(tsdbDir, logger)
-			testutil.Ok(t, err)
-
-			testutil.Ok(t, db.FlushWAL(tmpDir))
-			testutil.Ok(t, db.Close())
+			_ = createBlockFromHead(t, tsdbDir, head)
 		}
+		testutil.Ok(t, head.Close())
 
 		db, err := tsdb.OpenDBReadOnly(tsdbDir, logger)
 		testutil.Ok(t, err)
@@ -129,7 +124,7 @@ func benchMultiTSDBSeries(t testutil.TB, totalSamples, totalSeries int, flushToB
 
 	store := NewMultiTSDBStore(logger, nil, component.Receive, func() map[string]storepb.StoreServer { return tsdbs })
 
-	var expected []storepb.Series
+	var expected []*storepb.Series
 	lastLabels := storepb.Series{}
 	for _, resp := range resps {
 		for _, r := range resp {
@@ -141,7 +136,7 @@ func benchMultiTSDBSeries(t testutil.TB, totalSamples, totalSeries int, flushToB
 				continue
 			}
 			lastLabels = x
-			expected = append(expected, *r.GetSeries())
+			expected = append(expected, r.GetSeries())
 		}
 	}
 

pkg/store/proxy_test.go

@@ -11,6 +11,7 @@ import (
 	"math"
 	"math/rand"
 	"os"
+	"path/filepath"
 	"sort"
 	"testing"
 	"time"
@@ -1616,17 +1617,16 @@ func benchProxySeries(t testutil.TB, totalSamples, totalSeries int) {
 		var resps []*storepb.SeriesResponse
 
 		head, created := storetestutil.CreateHeadWithSeries(t, j, storetestutil.HeadGenOptions{
-			Dir:              tmpDir,
+			TSDBDir:          filepath.Join(tmpDir, fmt.Sprintf("%d", j)),
 			SamplesPerSeries: samplesPerSeriesPerClient,
 			Series:           seriesPerClient,
-			MaxFrameBytes:    storetestutil.RemoteReadFrameLimit,
 			Random:           random,
 			SkipChunks:       t.IsBenchmark(),
 		})
 		testutil.Ok(t, head.Close())
 
 		for i := 0; i < len(created); i++ {
-			resps = append(resps, storepb.NewSeriesResponse(&created[i]))
+			resps = append(resps, storepb.NewSeriesResponse(created[i]))
 		}
 
 		clients[j] = &testClient{
@@ -1647,23 +1647,22 @@ func benchProxySeries(t testutil.TB, totalSamples, totalSeries int) {
 	}
 
 	var allResps []*storepb.SeriesResponse
-	var expected []storepb.Series
+	var expected []*storepb.Series
 	lastLabels := storepb.Series{}
 	for _, c := range clients {
 		m := c.(*testClient).StoreClient.(*mockedStoreAPI)
 
+		// NOTE: Proxy will merge all series with same labels without any frame limit (https://github.com/thanos-io/thanos/issues/2332).
 		for _, r := range m.RespSeries {
 			allResps = append(allResps, r)
 
-			// Proxy will merge all series with same labels without limit (https://github.com/thanos-io/thanos/issues/2332).
-			// Let's do this here as well.
 			x := storepb.Series{Labels: r.GetSeries().Labels}
 			if x.String() == lastLabels.String() {
 				expected[len(expected)-1].Chunks = append(expected[len(expected)-1].Chunks, r.GetSeries().Chunks...)
 				continue
 			}
 			lastLabels = x
-			expected = append(expected, *r.GetSeries())
+			expected = append(expected, r.GetSeries())
 		}
 
 	}
@@ -1700,7 +1699,7 @@ func benchProxySeries(t testutil.TB, totalSamples, totalSeries int) {
 	// In this we expect exactly the same response as input.
 	expected = expected[:0]
 	for _, r := range allResps {
-		expected = append(expected, *r.GetSeries())
+		expected = append(expected, r.GetSeries())
 	}
 	storetestutil.TestServerSeries(t, store,
 		&storetestutil.SeriesCase{

pkg/store/storepb/testutil/series.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"math"
 	"math/rand"
+	"os"
 	"path/filepath"
 	"runtime"
 	"sort"
@@ -39,13 +40,12 @@ func allPostings(t testing.TB, ix tsdb.IndexReader) index.Postings {
 const RemoteReadFrameLimit = 1048576
 
 type HeadGenOptions struct {
-	Dir                      string
+	TSDBDir                  string
 	SamplesPerSeries, Series int
 
-	MaxFrameBytes int // No limit by default.
 	WithWAL       bool
 	PrependLabels labels.Labels
-	SkipChunks    bool
+	SkipChunks    bool // Skips chunks in returned slice (not in generated head!).
 
 	Random *rand.Rand
 }
@@ -54,22 +54,23 @@ type HeadGenOptions struct {
 // Returned series list has "ext1"="1" prepended. Each series looks as follows:
 // {foo=bar,i=000001aaaaaaaaaabbbbbbbbbbccccccccccdddddddddd} <random value> where number indicate sample number from 0.
 // Returned series are frame in same way as remote read would frame them.
-func CreateHeadWithSeries(t testing.TB, j int, opts HeadGenOptions) (*tsdb.Head, []storepb.Series) {
+func CreateHeadWithSeries(t testing.TB, j int, opts HeadGenOptions) (*tsdb.Head, []*storepb.Series) {
 	if opts.SamplesPerSeries < 1 || opts.Series < 1 {
 		t.Fatal("samples and series has to be 1 or more")
 	}
 
-	tsdbDir := filepath.Join(opts.Dir, fmt.Sprintf("%d", j))
-	fmt.Printf("Creating %d %d-sample series in %s\n", opts.Series, opts.SamplesPerSeries, tsdbDir)
+	fmt.Printf("Creating %d %d-sample series in %s\n", opts.Series, opts.SamplesPerSeries, opts.TSDBDir)
 
 	var w *wal.WAL
 	var err error
 	if opts.WithWAL {
-		w, err = wal.New(nil, nil, filepath.Join(tsdbDir, "wal"), true)
+		w, err = wal.New(nil, nil, filepath.Join(opts.TSDBDir, "wal"), true)
 		testutil.Ok(t, err)
+	} else {
+		testutil.Ok(t, os.MkdirAll(filepath.Join(opts.TSDBDir, "wal"), os.ModePerm))
 	}
 
-	h, err := tsdb.NewHead(nil, nil, w, tsdb.DefaultBlockDuration, tsdbDir, nil, tsdb.DefaultStripeSize, nil)
+	h, err := tsdb.NewHead(nil, nil, w, tsdb.DefaultBlockDuration, opts.TSDBDir, nil, tsdb.DefaultStripeSize, nil)
 	testutil.Ok(t, err)
 
 	app := h.Appender(context.Background())
@@ -96,31 +97,20 @@ func CreateHeadWithSeries(t testing.TB, j int, opts HeadGenOptions) (*tsdb.Head,
 	var (
 		lset       labels.Labels
 		chunkMetas []chunks.Meta
-		expected   = make([]storepb.Series, 0, opts.Series)
-		sBytes     int
+		expected   = make([]*storepb.Series, 0, opts.Series)
 	)
 
 	all := allPostings(t, ir)
 	for all.Next() {
 		testutil.Ok(t, ir.Series(all.At(), &lset, &chunkMetas))
-		i := 0
 		sLset := storepb.PromLabelsToLabels(lset)
-		expected = append(expected, storepb.Series{Labels: append(storepb.PromLabelsToLabels(opts.PrependLabels), sLset...)})
+		expected = append(expected, &storepb.Series{Labels: append(storepb.PromLabelsToLabels(opts.PrependLabels), sLset...)})
 
 		if opts.SkipChunks {
 			continue
 		}
 
-		lBytes := 0
-		for _, l := range sLset {
-			lBytes += l.Size()
-		}
-		sBytes = lBytes
-
-		for {
-			c := chunkMetas[i]
-			i++
-
+		for _, c := range chunkMetas {
 			chEnc, err := chks.Chunk(c.Ref)
 			testutil.Ok(t, err)
 
@@ -129,22 +119,11 @@ func CreateHeadWithSeries(t testing.TB, j int, opts HeadGenOptions) (*tsdb.Head,
 				c.MaxTime = c.MinTime + int64(chEnc.NumSamples()) - 1
 			}
 
-			sBytes += len(chEnc.Bytes())
-
 			expected[len(expected)-1].Chunks = append(expected[len(expected)-1].Chunks, storepb.AggrChunk{
 				MinTime: c.MinTime,
 				MaxTime: c.MaxTime,
 				Raw:     &storepb.Chunk{Type: storepb.Chunk_XOR, Data: chEnc.Bytes()},
 			})
-			if i >= len(chunkMetas) {
-				break
-			}
-
-			// Compose many frames as remote read (so sidecar StoreAPI) would do if requested by  maxFrameBytes.
-			if opts.MaxFrameBytes > 0 && sBytes >= opts.MaxFrameBytes {
-				expected = append(expected, storepb.Series{Labels: sLset})
-				sBytes = lBytes
-			}
 		}
 	}
 	testutil.Ok(t, all.Err())
@@ -158,7 +137,7 @@ type SeriesServer struct {
 
 	ctx context.Context
 
-	SeriesSet []storepb.Series
+	SeriesSet []*storepb.Series
 	Warnings  []string
 	HintsSet  []*types.Any
 
@@ -178,7 +157,7 @@ func (s *SeriesServer) Send(r *storepb.SeriesResponse) error {
 	}
 
 	if r.GetSeries() != nil {
-		s.SeriesSet = append(s.SeriesSet, *r.GetSeries())
+		s.SeriesSet = append(s.SeriesSet, r.GetSeries())
 		return nil
 	}
 
@@ -227,7 +206,7 @@ type SeriesCase struct {
 	Req  *storepb.SeriesRequest
 
 	// Exact expectations are checked only for tests. For benchmarks only length is assured.
-	ExpectedSeries   []storepb.Series
+	ExpectedSeries   []*storepb.Series
 	ExpectedWarnings []string
 	ExpectedHints    []hintspb.SeriesResponseHints
 }
@@ -241,7 +220,7 @@ func TestServerSeries(t testutil.TB, store storepb.StoreServer, cases ...*Series
 				srv := NewSeriesServer(context.Background())
 				testutil.Ok(t, store.Series(c.Req, srv))
 				testutil.Equals(t, len(c.ExpectedWarnings), len(srv.Warnings), "%v", srv.Warnings)
-				testutil.Equals(t, len(c.ExpectedSeries), len(srv.SeriesSet))
+				//testutil.Equals(t, len(c.ExpectedSeries), len(srv.SeriesSet))
 				testutil.Equals(t, len(c.ExpectedHints), len(srv.HintsSet))
 
 				if !t.IsBenchmark() {
@@ -252,10 +231,18 @@ func TestServerSeries(t testutil.TB, store storepb.StoreServer, cases ...*Series
 						})
 					}
 
-					testutil.Equals(t, c.ExpectedSeries[0].Chunks[0], srv.SeriesSet[0].Chunks[0])
-
-					// This might give unreadable output for millions of series on fail..
-					testutil.Equals(t, c.ExpectedSeries, srv.SeriesSet)
+					// Huge responses can produce unreadable diffs - make it more human readable.
+					if len(c.ExpectedSeries) > 4 {
+						for j := range c.ExpectedSeries {
+							testutil.Equals(t, c.ExpectedSeries[j].Labels, srv.SeriesSet[j].Labels, "%v series chunks mismatch", j)
+							if len(c.ExpectedSeries[j].Chunks) > 20 {
+								testutil.Equals(t, len(c.ExpectedSeries[j].Chunks), len(srv.SeriesSet[j].Chunks), "%v series chunks number mismatch", j)
+							}
+							testutil.Equals(t, c.ExpectedSeries[j].Chunks, srv.SeriesSet[j].Chunks, "%v series chunks mismatch", j)
+						}
+					} else {
+						testutil.Equals(t, c.ExpectedSeries, srv.SeriesSet)
+					}
 
 					var actualHints []hintspb.SeriesResponseHints
 					for _, anyHints := range srv.HintsSet {