receive/multitsdb: don't forget to close the store

Don't forget to close the store. Trying to reproduce prod problems but cannot trigger them :/ Signed-off-by: Giedrius Statkevičius <giedrius.statkevicius@vinted.com>
vinted · Aug 28, 2024 · 0cbe697 · 0cbe697
1 parent e0221e6
commit 0cbe697
Show file tree

Hide file tree

Showing 13 changed files with 142 additions and 19 deletions.
diff --git a/cmd/thanos/compact_test.go b/cmd/thanos/compact_test.go
@@ -0,0 +1,32 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"testing"
+
+	"github.com/efficientgo/core/testutil"
+	"github.com/go-kit/log"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/prometheus/storage"
+	"github.com/prometheus/prometheus/tsdb"
+	"github.com/thanos-io/thanos/pkg/compact/downsample"
+)
+
+func TestCompactorProdData(t *testing.T) {
+	levels, err := compactions.levels(4)
+	testutil.Ok(t, err)
+
+	comp, err := tsdb.NewLeveledCompactor(context.Background(), prometheus.NewRegistry(), log.NewLogfmtLogger(os.Stderr), levels, downsample.NewPool(), storage.NewCompactingChunkSeriesMerger(storage.ChainedSeriesMerge))
+	testutil.Ok(t, err)
+
+	out, err := comp.Compact(
+		"/tmp/otpt",
+		[]string{
+			"/home/giedriusstatkevicius/dev/compact_debug/01HV5FDABSVP6Z0F4RX2EQ0YVR",
+			"/home/giedriusstatkevicius/dev/compact_debug/01HV5GCTBA973S5T1W9Q81DEMR",
+		}, nil,
+	)
+	fmt.Println(out, err)
+}
diff --git a/cmd/thanos/receive.go b/cmd/thanos/receive.go
@@ -214,6 +214,7 @@ func runReceive(
 		bkt,
 		conf.allowOutOfOrderUpload,
 		hashFunc,
+		15*time.Second,
 	)
 	writer := receive.NewWriter(log.With(logger, "component", "receive-writer"), dbs, &receive.WriterOptions{
 		Intern:                   conf.writerInterning,

diff --git a/cmd/thanos/rule.go b/cmd/thanos/rule.go
@@ -736,7 +736,7 @@ func runRule(
 	}
 	infoOptions := []info.ServerOptionFunc{info.WithRulesInfoFunc()}
 	if tsdbDB != nil {
-		tsdbStore := store.NewTSDBStore(logger, tsdbDB, component.Rule, conf.lset)
+		tsdbStore := store.NewTSDBStore(logger, tsdbDB, component.Rule, conf.lset, 15*time.Second)
 		infoOptions = append(
 			infoOptions,
 			info.WithLabelSetFunc(func() []labelpb.ZLabelSet {

diff --git a/pkg/api/query/v1_test.go b/pkg/api/query/v1_test.go
@@ -195,7 +195,7 @@ func TestQueryEndpoints(t *testing.T) {
 		baseAPI: &baseAPI.BaseAPI{
 			Now: func() time.Time { return now },
 		},
-		queryableCreate:       query.NewQueryableCreator(nil, nil, newProxyStoreWithTSDBStore(db), 2, timeout),
+		queryableCreate:       query.NewQueryableCreator(nil, nil, newProxyStoreWithTSDBStore(t, db), 2, timeout),
 		engineFactory:         ef,
 		defaultEngine:         PromqlEnginePrometheus,
 		lookbackDeltaCreate:   func(m int64) time.Duration { return time.Duration(0) },
@@ -648,7 +648,7 @@ func TestQueryExplainEndpoints(t *testing.T) {
 		baseAPI: &baseAPI.BaseAPI{
 			Now: func() time.Time { return now },
 		},
-		queryableCreate:       query.NewQueryableCreator(nil, nil, newProxyStoreWithTSDBStore(db), 2, timeout),
+		queryableCreate:       query.NewQueryableCreator(nil, nil, newProxyStoreWithTSDBStore(t, db), 2, timeout),
 		engineFactory:         ef,
 		defaultEngine:         PromqlEnginePrometheus,
 		lookbackDeltaCreate:   func(m int64) time.Duration { return time.Duration(0) },
@@ -712,7 +712,7 @@ func TestQueryAnalyzeEndpoints(t *testing.T) {
 		baseAPI: &baseAPI.BaseAPI{
 			Now: func() time.Time { return now },
 		},
-		queryableCreate:       query.NewQueryableCreator(nil, nil, newProxyStoreWithTSDBStore(db), 2, timeout),
+		queryableCreate:       query.NewQueryableCreator(nil, nil, newProxyStoreWithTSDBStore(t, db), 2, timeout),
 		engineFactory:         ef,
 		defaultEngine:         PromqlEnginePrometheus,
 		lookbackDeltaCreate:   func(m int64) time.Duration { return time.Duration(0) },
@@ -777,10 +777,13 @@ func TestQueryAnalyzeEndpoints(t *testing.T) {
 	}
 }
 
-func newProxyStoreWithTSDBStore(db store.TSDBReader) *store.ProxyStore {
+func newProxyStoreWithTSDBStore(t *testing.T, db store.TSDBReader) *store.ProxyStore {
+	tsdbStore := store.NewTSDBStore(nil, db, component.Query, nil, 15*time.Second)
+	t.Cleanup(tsdbStore.Close)
+
 	c := &storetestutil.TestClient{
 		Name:        "1",
-		StoreClient: storepb.ServerAsClient(store.NewTSDBStore(nil, db, component.Query, nil), 0),
+		StoreClient: storepb.ServerAsClient(tsdbStore, 0),
 		MinTime:     math.MinInt64, MaxTime: math.MaxInt64,
 	}
 
@@ -886,7 +889,7 @@ func TestMetadataEndpoints(t *testing.T) {
 		baseAPI: &baseAPI.BaseAPI{
 			Now: func() time.Time { return now },
 		},
-		queryableCreate:     query.NewQueryableCreator(nil, nil, newProxyStoreWithTSDBStore(db), 2, timeout),
+		queryableCreate:     query.NewQueryableCreator(nil, nil, newProxyStoreWithTSDBStore(t, db), 2, timeout),
 		engineFactory:       ef,
 		defaultEngine:       PromqlEnginePrometheus,
 		lookbackDeltaCreate: func(m int64) time.Duration { return time.Duration(0) },
@@ -902,7 +905,7 @@ func TestMetadataEndpoints(t *testing.T) {
 		baseAPI: &baseAPI.BaseAPI{
 			Now: func() time.Time { return now },
 		},
-		queryableCreate:          query.NewQueryableCreator(nil, nil, newProxyStoreWithTSDBStore(db), 2, timeout),
+		queryableCreate:          query.NewQueryableCreator(nil, nil, newProxyStoreWithTSDBStore(t, db), 2, timeout),
 		engineFactory:            ef,
 		defaultEngine:            PromqlEnginePrometheus,
 		lookbackDeltaCreate:      func(m int64) time.Duration { return time.Duration(0) },

diff --git a/pkg/query/query_test.go b/pkg/query/query_test.go
@@ -48,10 +48,13 @@ func TestQuerier_Proxy(t *testing.T) {
 				m, err := storepb.PromMatchersToMatchers(st.matchers...)
 				testutil.Ok(t, err)
 
+				tsdbStore := store.NewTSDBStore(logger, st.storage.DB, component.Debug, nil, 15*time.Second)
+				t.Cleanup(tsdbStore.Close)
+
 				// TODO(bwplotka): Parse external labels.
 				clients = append(clients, &storetestutil.TestClient{
 					Name:        fmt.Sprintf("store number %v", i),
-					StoreClient: storepb.ServerAsClient(selectedStore(store.NewTSDBStore(logger, st.storage.DB, component.Debug, nil), m, st.mint, st.maxt), 0),
+					StoreClient: storepb.ServerAsClient(selectedStore(tsdbStore, m, st.mint, st.maxt), 0),
 					MinTime:     st.mint,
 					MaxTime:     st.maxt,
 				})

diff --git a/pkg/receive/handler_test.go b/pkg/receive/handler_test.go
@@ -1005,6 +1005,7 @@ func benchmarkHandlerMultiTSDBReceiveRemoteWrite(b testutil.TB) {
 		nil,
 		false,
 		metadata.NoneFunc,
+		15*time.Second,
 	)
 	defer func() { testutil.Ok(b, m.Close()) }()
 	handler.writer = NewWriter(logger, m, &WriterOptions{})

diff --git a/pkg/receive/multitsdb.go b/pkg/receive/multitsdb.go
@@ -67,6 +67,7 @@ type MultiTSDB struct {
 	allowOutOfOrderUpload bool
 	hashFunc              metadata.HashFunc
 	hashringConfigs       []HashringConfig
+	filterInterval        time.Duration
 }
 
 // NewMultiTSDB creates new MultiTSDB.
@@ -81,6 +82,7 @@ func NewMultiTSDB(
 	bucket objstore.Bucket,
 	allowOutOfOrderUpload bool,
 	hashFunc metadata.HashFunc,
+	filterInterval time.Duration,
 ) *MultiTSDB {
 	if l == nil {
 		l = log.NewNopLogger()
@@ -98,6 +100,7 @@ func NewMultiTSDB(
 		bucket:                bucket,
 		allowOutOfOrderUpload: allowOutOfOrderUpload,
 		hashFunc:              hashFunc,
+		filterInterval:        filterInterval,
 	}
 }
 
@@ -398,6 +401,7 @@ func (t *MultiTSDB) Close() error {
 			level.Error(t.logger).Log("msg", "closing TSDB failed; not ready", "tenant", id)
 			continue
 		}
+		tenant.storeTSDB.Close()
 		level.Info(t.logger).Log("msg", "closing TSDB", "tenant", id)
 		merr.Add(db.Close())
 	}
@@ -736,7 +740,7 @@ func (t *MultiTSDB) startTSDB(logger log.Logger, tenantID string, tenant *tenant
 			shipper.DefaultMetaFilename,
 		)
 	}
-	tenant.set(store.NewTSDBStore(logger, s, component.Receive, lset), s, ship, exemplars.NewTSDB(s, lset))
+	tenant.set(store.NewTSDBStore(logger, s, component.Receive, lset, t.filterInterval), s, ship, exemplars.NewTSDB(s, lset))
 	level.Info(logger).Log("msg", "TSDB is now ready")
 	return nil
 }

diff --git a/pkg/receive/multitsdb_test.go b/pkg/receive/multitsdb_test.go
@@ -18,6 +18,7 @@ import (
 	"github.com/efficientgo/core/testutil"
 	"github.com/go-kit/log"
 	"github.com/oklog/ulid"
+	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/prometheus/model/exemplar"
 	"github.com/prometheus/prometheus/model/labels"
@@ -57,6 +58,7 @@ func TestMultiTSDB(t *testing.T) {
 			nil,
 			false,
 			metadata.NoneFunc,
+			15*time.Second,
 		)
 		defer func() { testutil.Ok(t, m.Close()) }()
 
@@ -141,6 +143,7 @@ func TestMultiTSDB(t *testing.T) {
 			nil,
 			false,
 			metadata.NoneFunc,
+			15*time.Second,
 		)
 		defer func() { testutil.Ok(t, m.Close()) }()
 
@@ -184,6 +187,7 @@ func TestMultiTSDB(t *testing.T) {
 			nil,
 			false,
 			metadata.NoneFunc,
+			15*time.Second,
 		)
 		defer func() { testutil.Ok(t, m.Close()) }()
 
@@ -415,6 +419,65 @@ func checkExemplarsResponse(t *testing.T, expected, data []exemplarspb.ExemplarD
 	}
 }
 
+func TestMultiTSDBPruneWithMetricNameFilter(t *testing.T) {
+	dir := t.TempDir()
+
+	m := NewMultiTSDB(dir, log.NewLogfmtLogger(os.Stderr), prometheus.NewRegistry(),
+		&tsdb.Options{
+			MinBlockDuration:  (2 * time.Hour).Milliseconds(),
+			MaxBlockDuration:  (2 * time.Hour).Milliseconds(),
+			RetentionDuration: (6 * time.Hour).Milliseconds(),
+		},
+		labels.FromStrings("replica", "test"),
+		"tenant_id",
+		nil,
+		false,
+		metadata.NoneFunc,
+		1*time.Nanosecond,
+	)
+	defer func() { testutil.Ok(t, m.Close()) }()
+
+	wg := &sync.WaitGroup{}
+	ctx, cancel := context.WithCancel(context.Background())
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			default:
+				for i := 0; i < 100; i++ {
+					err := appendSample(m, "deleted-tenant", time.UnixMilli(int64(10+i)))
+					if errors.Is(err, tsdb.ErrNotReady) || errors.Is(err, ErrNotReady) {
+						continue
+					}
+					testutil.Ok(t, err)
+				}
+			}
+		}
+	}()
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			default:
+				testutil.Ok(t, m.Prune(ctx))
+			}
+		}
+	}()
+
+	time.Sleep(15 * time.Second)
+
+	cancel()
+	wg.Wait()
+}
+
 func TestMultiTSDBPrune(t *testing.T) {
 	tests := []struct {
 		name            string
@@ -451,6 +514,7 @@ func TestMultiTSDBPrune(t *testing.T) {
 				test.bucket,
 				false,
 				metadata.NoneFunc,
+				15*time.Second,
 			)
 			defer func() { testutil.Ok(t, m.Close()) }()
 
@@ -520,6 +584,7 @@ func TestMultiTSDBRecreatePrunedTenant(t *testing.T) {
 		objstore.NewInMemBucket(),
 		false,
 		metadata.NoneFunc,
+		15*time.Second,
 	)
 	defer func() { testutil.Ok(t, m.Close()) }()
 
@@ -581,6 +646,7 @@ func TestAlignedHeadFlush(t *testing.T) {
 				test.bucket,
 				false,
 				metadata.NoneFunc,
+				15*time.Second,
 			)
 			defer func() { testutil.Ok(t, m.Close()) }()
 
@@ -655,6 +721,7 @@ func TestMultiTSDBStats(t *testing.T) {
 				nil,
 				false,
 				metadata.NoneFunc,
+				15*time.Second,
 			)
 			defer func() { testutil.Ok(t, m.Close()) }()
 
@@ -684,6 +751,7 @@ func TestMultiTSDBWithNilStore(t *testing.T) {
 		nil,
 		false,
 		metadata.NoneFunc,
+		15*time.Second,
 	)
 	defer func() { testutil.Ok(t, m.Close()) }()
 
@@ -725,6 +793,7 @@ func TestProxyLabelValues(t *testing.T) {
 		nil,
 		false,
 		metadata.NoneFunc,
+		15*time.Second,
 	)
 	defer func() { testutil.Ok(t, m.Close()) }()
 
@@ -815,6 +884,7 @@ func BenchmarkMultiTSDB(b *testing.B) {
 		nil,
 		false,
 		metadata.NoneFunc,
+		15*time.Second,
 	)
 	defer func() { testutil.Ok(b, m.Close()) }()
 

diff --git a/pkg/receive/receive_test.go b/pkg/receive/receive_test.go
@@ -815,6 +815,7 @@ func initializeMultiTSDB(dir string) *MultiTSDB {
 		bucket,
 		false,
 		metadata.NoneFunc,
+		15*time.Second,
 	)
 
 	return m

diff --git a/pkg/receive/writer_test.go b/pkg/receive/writer_test.go
@@ -344,6 +344,7 @@ func TestWriter(t *testing.T) {
 				nil,
 				false,
 				metadata.NoneFunc,
+				15*time.Second,
 			)
 			t.Cleanup(func() { testutil.Ok(t, m.Close()) })
 
@@ -436,6 +437,7 @@ func benchmarkWriter(b *testing.B, labelsNum int, seriesNum int, generateHistogr
 		nil,
 		false,
 		metadata.NoneFunc,
+		15*time.Second,
 	)
 	b.Cleanup(func() { testutil.Ok(b, m.Close()) })
 

diff --git a/pkg/store/acceptance_test.go b/pkg/store/acceptance_test.go
@@ -983,7 +983,11 @@ func TestTSDBStore_Acceptance(t *testing.T) {
 		tt.Cleanup(func() { testutil.Ok(tt, db.Close()) })
 		appendFn(db.Appender(context.Background()))
 
-		return NewTSDBStore(nil, db, component.Rule, extLset)
+		st := NewTSDBStore(nil, db, component.Rule, extLset, 15*time.Second)
+
+		tt.Cleanup(st.Close)
+
+		return st
 	}
 
 	testStoreAPIsAcceptance(t, startStore)

diff --git a/pkg/store/tsdb.go b/pkg/store/tsdb.go
@@ -72,7 +72,7 @@ type ReadWriteTSDBStore struct {
 
 // NewTSDBStore creates a new TSDBStore.
 // NOTE: Given lset has to be sorted.
-func NewTSDBStore(logger log.Logger, db TSDBReader, component component.StoreAPI, extLset labels.Labels) *TSDBStore {
+func NewTSDBStore(logger log.Logger, db TSDBReader, component component.StoreAPI, extLset labels.Labels, updateInterval time.Duration) *TSDBStore {
 	if logger == nil {
 		logger = log.NewNopLogger()
 	}
@@ -91,7 +91,7 @@ func NewTSDBStore(logger log.Logger, db TSDBReader, component component.StoreAPI
 		MetricNameFilter: filter.NewCuckooFilterMetricNameFilter(1000000),
 	}
 
-	t := time.NewTicker(15 * time.Second)
+	t := time.NewTicker(updateInterval)
 	ctx, cancel := context.WithCancel(context.Background())
 	updateMetricNames := func() {
 		vals, err := st.LabelValues(context.Background(), &storepb.LabelValuesRequest{