cherry pick store gateway fix to release 0.30 (thanos-io#6089)

* Fix: Failure to close BlockSeriesClient cause store-gateway deadlock (thanos-io#6086) * Fix: Failure to close BlockSeriesClient cause store-gateway deadlock Signed-off-by: Alan Protasio <alanprot@gmail.com> * Adding tests Signed-off-by: Alan Protasio <alanprot@gmail.com> * reverting the change on get series Signed-off-by: Alan Protasio <alanprot@gmail.com> * fix lint Signed-off-by: Alan Protasio <alanprot@gmail.com> --------- Signed-off-by: Alan Protasio <alanprot@gmail.com> * update changelog Signed-off-by: Ben Ye <benye@amazon.com> --------- Signed-off-by: Alan Protasio <alanprot@gmail.com> Signed-off-by: Ben Ye <benye@amazon.com> Co-authored-by: Alan Protasio <alanprot@gmail.com>
ngraham20 · Mar 16, 2023 · 6cfab68 · 6cfab68
1 parent 8084848
commit 6cfab68
Show file tree

Hide file tree

Showing 3 changed files with 29 additions and 0 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -24,6 +24,7 @@ We use *breaking :warning:* to mark changes that are not backward compatible (re
 ### Fixed
 
 - [#6066](https://github.com/thanos-io/thanos/pull/6066) Tracing: fixed panic because of nil sampler
+- [#6086](https://github.com/thanos-io/thanos/pull/6086) Store Gateway: Fix store-gateway deadlock due to not close BlockSeriesClient
 
 ## [v0.30.1](https://github.com/thanos-io/thanos/tree/release-0.30) - 4.01.2023
 

diff --git a/pkg/store/bucket.go b/pkg/store/bucket.go
@@ -1208,6 +1208,7 @@ func (s *BucketStore) Series(req *storepb.SeriesRequest, srv storepb.Store_Serie
 			}
 
 			shardMatcher := req.ShardInfo.Matcher(&s.buffers)
+
 			blockClient := newBlockSeriesClient(
 				srv.Context(),
 				s.logger,
@@ -1220,9 +1221,11 @@ func (s *BucketStore) Series(req *storepb.SeriesRequest, srv storepb.Store_Serie
 				s.seriesBatchSize,
 				s.metrics.chunkFetchDuration,
 			)
+
 			defer blockClient.Close()
 
 			g.Go(func() error {
+
 				span, _ := tracing.StartSpan(gctx, "bucket_store_block_series", tracing.Tags{
 					"block.id":         blk.meta.ULID,
 					"block.mint":       blk.meta.MinTime,
@@ -1464,6 +1467,7 @@ func (s *BucketStore) LabelNames(ctx context.Context, req *storepb.LabelNamesReq
 					SkipChunks: true,
 				}
 				blockClient := newBlockSeriesClient(newCtx, s.logger, b, seriesReq, nil, bytesLimiter, nil, true, SeriesBatchSize, s.metrics.chunkFetchDuration)
+				defer blockClient.Close()
 
 				if err := blockClient.ExpandPostings(
 					reqSeriesMatchersNoExtLabels,
@@ -1638,6 +1642,7 @@ func (s *BucketStore) LabelValues(ctx context.Context, req *storepb.LabelValuesR
 					SkipChunks: true,
 				}
 				blockClient := newBlockSeriesClient(newCtx, s.logger, b, seriesReq, nil, bytesLimiter, nil, true, SeriesBatchSize, s.metrics.chunkFetchDuration)
+				defer blockClient.Close()
 
 				if err := blockClient.ExpandPostings(
 					reqSeriesMatchersNoExtLabels,

diff --git a/pkg/store/bucket_e2e_test.go b/pkg/store/bucket_e2e_test.go
@@ -9,6 +9,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -765,6 +766,10 @@ func TestBucketStore_LabelNames_e2e(t *testing.T) {
 		} {
 			t.Run(name, func(t *testing.T) {
 				vals, err := s.store.LabelNames(ctx, tc.req)
+				for _, b := range s.store.blocks {
+					waitTimeout(t, &b.pendingReaders, 5*time.Second)
+				}
+
 				testutil.Ok(t, err)
 
 				testutil.Equals(t, tc.expected, vals.Names)
@@ -868,6 +873,10 @@ func TestBucketStore_LabelValues_e2e(t *testing.T) {
 		} {
 			t.Run(name, func(t *testing.T) {
 				vals, err := s.store.LabelValues(ctx, tc.req)
+				for _, b := range s.store.blocks {
+					waitTimeout(t, &b.pendingReaders, 5*time.Second)
+				}
+
 				testutil.Ok(t, err)
 
 				testutil.Equals(t, tc.expected, emptyToNil(vals.Values))
@@ -882,3 +891,17 @@ func emptyToNil(values []string) []string {
 	}
 	return values
 }
+
+func waitTimeout(t *testing.T, wg *sync.WaitGroup, timeout time.Duration) {
+	c := make(chan struct{})
+	go func() {
+		defer close(c)
+		wg.Wait()
+	}()
+	select {
+	case <-c:
+		return
+	case <-time.After(timeout):
+		t.Fatalf("timeout waiting wg for %v", timeout)
+	}
+}