receive/rule: Fixed Segfault issue; Added tests & benchmarks for TSDB…

…Store, 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.

Fix: We deletage closer now to ensure multitsdb operate on valid data.

Signed-off-by: Bartlomiej Plotka <[email protected]>

# Conflicts:
#	pkg/store/tsdb_test.go
#	pkg/testutil/testutil.go

pkg/store/bucket_test.go

@@ -1175,7 +1175,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))
 	)
 
@@ -1210,7 +1210,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,
@@ -1533,17 +1533,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())
 
@@ -1607,7 +1612,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.go

@@ -6,13 +6,16 @@ package store
 import (
 	"context"
 	"fmt"
+	"io"
 	"sync"
 
 	"github.com/go-kit/kit/log"
 	grpc_opentracing "github.com/grpc-ecosystem/go-grpc-middleware/tracing/opentracing"
 	"github.com/opentracing/opentracing-go"
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
+	tsdb_errors "github.com/prometheus/prometheus/tsdb/errors"
+	"github.com/thanos-io/thanos/pkg/runutil"
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
@@ -96,6 +99,8 @@ type tenantSeriesSetServer struct {
 
 	err    error
 	tenant string
+
+	closers []io.Closer
 }
 
 // TODO(bwplotka): Remove tenant awareness; keep it simple with single functionality.
@@ -156,6 +161,18 @@ func (s *tenantSeriesSetServer) Send(r *storepb.SeriesResponse) error {
 	}
 }
 
+func (s *tenantSeriesSetServer) Delegate(closer io.Closer) {
+	s.closers = append(s.closers, closer)
+}
+
+func (s *tenantSeriesSetServer) Close() error {
+	var merr tsdb_errors.MultiError
+	for _, c := range s.closers {
+		merr.Add(c.Close())
+	}
+	return merr.Err()
+}
+
 func (s *tenantSeriesSetServer) Next() (ok bool) {
 	s.cur, ok = <-s.recv
 	return ok
@@ -188,6 +205,7 @@ func (s *MultiTSDBStore) Series(r *storepb.SeriesRequest, srv storepb.Store_Seri
 	// Each might be quite large (multi chunk long series given by sidecar).
 	respSender, respCh := newCancelableRespChannel(gctx, 10)
 
+	var closers []io.Closer
 	g.Go(func() error {
 		// This go routine is responsible for calling store's Series concurrently. Merged results
 		// are passed to respCh and sent concurrently to client (if buffer of 10 have room).
@@ -216,6 +234,8 @@ func (s *MultiTSDBStore) Series(r *storepb.SeriesRequest, srv storepb.Store_Seri
 				defer wg.Done()
 				ss.Series(store, r)
 			}()
+
+			closers = append(closers, ss)
 			seriesSet = append(seriesSet, ss)
 		}
 
@@ -237,7 +257,12 @@ func (s *MultiTSDBStore) Series(r *storepb.SeriesRequest, srv storepb.Store_Seri
 		}
 		return nil
 	})
-	return g.Wait()
+	err := g.Wait()
+	for _, c := range closers {
+		runutil.CloseWithLogOnErr(s.logger, c, "close tenant series request")
+	}
+	return err
+
 }
 
 // LabelNames returns all known label names.

pkg/store/multitsdb_test.go

@@ -91,29 +91,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)
@@ -128,7 +123,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 {
@@ -140,7 +135,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{