store: lock around iterating over s.blocks

Hold a lock around s.blocks when iterating over it. I have experienced a case where a block had been added to a blockSet twice somehow and it being somehow removed from s.blocks is the only way it could happen. This is the only "bad" thing I've been able to spot. Remove the removeBlock() function as IMHO it is too shallow. Better to hold the lock consistently. Signed-off-by: Giedrius Statkevičius <[email protected]>
thanos-io · Feb 4, 2025 · ab9bd14 · ab9bd14
1 parent 57efc2a
commit ab9bd14
Show file tree

Hide file tree

Showing 6 changed files with 78 additions and 50 deletions.
diff --git a/pkg/extprom/http/instrument_client.go b/pkg/extprom/http/instrument_client.go
@@ -23,7 +23,7 @@ type ClientMetrics struct {
 
 // NewClientMetrics creates a new instance of ClientMetrics.
 // It will also register the metrics with the included register.
-// This ClientMetrics should be re-used for diff clients with the same purpose.
+// This ClientMetrics should be reused for diff clients with the same purpose.
 // e.g. 1 ClientMetrics should be used for all the clients that talk to Alertmanager.
 func NewClientMetrics(reg prometheus.Registerer) *ClientMetrics {
 	var m ClientMetrics

diff --git a/pkg/query/endpointset.go b/pkg/query/endpointset.go
@@ -366,7 +366,7 @@ func (e *EndpointSet) Update(ctx context.Context) {
 		if er.HasStoreAPI() && (er.ComponentType() == component.Sidecar || er.ComponentType() == component.Rule) &&
 			stats[component.Sidecar.String()][extLset]+stats[component.Rule.String()][extLset] > 0 {
 
-			level.Warn(e.logger).Log("msg", "found duplicate storeEndpoints producer (sidecar or ruler). This is not advices as it will malform data in in the same bucket",
+			level.Warn(e.logger).Log("msg", "found duplicate storeEndpoints producer (sidecar or ruler). This is not advised as it will malform data in in the same bucket",
 				"address", addr, "extLset", extLset, "duplicates", fmt.Sprintf("%v", stats[component.Sidecar.String()][extLset]+stats[component.Rule.String()][extLset]+1))
 		}
 		stats[er.ComponentType().String()][extLset]++

diff --git a/pkg/runutil/runutil.go b/pkg/runutil/runutil.go
@@ -45,7 +45,7 @@
 // The rununtil.Exhaust* family of functions provide the same functionality but
 // they take an io.ReadCloser and they exhaust the whole reader before closing
 // them. They are useful when trying to use http keep-alive connections because
-// for the same connection to be re-used the whole response body needs to be
+// for the same connection to be reused the whole response body needs to be
 // exhausted.
 package runutil
 

diff --git a/pkg/store/bucket.go b/pkg/store/bucket.go
@@ -742,6 +742,7 @@ func (s *BucketStore) SyncBlocks(ctx context.Context) error {
 					continue
 				}
 				if err := s.addBlock(ctx, meta); err != nil {
+					level.Warn(s.logger).Log("msg", "adding block failed", "err", err, "id", meta.ULID.String())
 					continue
 				}
 			}
@@ -766,17 +767,37 @@ func (s *BucketStore) SyncBlocks(ctx context.Context) error {
 		return metaFetchErr
 	}
 
+	var cleanupBlocks []*bucketBlock
+	s.mtx.RLock()
+	keys := make([]ulid.ULID, 0, len(s.blocks))
+	for k := range s.blocks {
+		keys = append(keys, k)
+	}
+	s.mtx.RUnlock()
+
 	// Drop all blocks that are no longer present in the bucket.
-	for id := range s.blocks {
+	for _, id := range keys {
 		if _, ok := metas[id]; ok {
 			continue
 		}
-		if err := s.removeBlock(id); err != nil {
-			level.Warn(s.logger).Log("msg", "drop of outdated block failed", "block", id, "err", err)
-			s.metrics.blockDropFailures.Inc()
+
+		s.mtx.Lock()
+		b := s.blocks[id]
+		if b == nil {
+			// NOTE(GiedriusS): this cannot really happen because SyncBlocks() is called in one thread only but just in case.
+			s.mtx.Unlock()
+			continue
 		}
-		level.Info(s.logger).Log("msg", "dropped outdated block", "block", id)
+		lset := labels.FromMap(b.meta.Thanos.Labels)
+		s.blockSets[lset.Hash()].remove(id)
+		delete(s.blocks, id)
+		s.mtx.Unlock()
+
+		s.metrics.blocksLoaded.Dec()
 		s.metrics.blockDrops.Inc()
+		cleanupBlocks = append(cleanupBlocks, b)
+
+		level.Info(s.logger).Log("msg", "dropped outdated block", "block", id)
 	}
 
 	// Sync advertise labels.
@@ -789,6 +810,25 @@ func (s *BucketStore) SyncBlocks(ctx context.Context) error {
 		return strings.Compare(s.advLabelSets[i].String(), s.advLabelSets[j].String()) < 0
 	})
 	s.mtx.Unlock()
+
+	go func() {
+		for _, b := range cleanupBlocks {
+			var errs prometheus.MultiError
+
+			errs.Append(b.Close())
+
+			if b.dir != "" {
+				errs.Append(os.RemoveAll(b.dir))
+			}
+
+			if len(errs) == 0 {
+				return
+			}
+
+			level.Warn(s.logger).Log("msg", "close of outdated block failed", "block", b.meta.ULID.String(), "err", errs.Error())
+			s.metrics.blockDropFailures.Inc()
+		}
+	}()
 	return nil
 }
 
@@ -921,32 +961,6 @@ func (s *BucketStore) addBlock(ctx context.Context, meta *metadata.Meta) (err er
 	return nil
 }
 
-func (s *BucketStore) removeBlock(id ulid.ULID) error {
-	s.mtx.Lock()
-	b, ok := s.blocks[id]
-	if ok {
-		lset := labels.FromMap(b.meta.Thanos.Labels)
-		s.blockSets[lset.Hash()].remove(id)
-		delete(s.blocks, id)
-	}
-	s.mtx.Unlock()
-
-	if !ok {
-		return nil
-	}
-
-	s.metrics.blocksLoaded.Dec()
-	if err := b.Close(); err != nil {
-		return errors.Wrap(err, "close block")
-	}
-
-	if b.dir == "" {
-		return nil
-	}
-
-	return os.RemoveAll(b.dir)
-}
-
 // TimeRange returns the minimum and maximum timestamp of data available in the store.
 func (s *BucketStore) TimeRange() (mint, maxt int64) {
 	s.mtx.RLock()

diff --git a/pkg/store/bucket_test.go b/pkg/store/bucket_test.go
@@ -1827,7 +1827,10 @@ func TestBucketSeries_OneBlock_InMemIndexCacheSegfault(t *testing.T) {
 		testutil.Equals(t, numSeries, len(srv.SeriesSet))
 	})
 	t.Run("remove second block. Cache stays. Ask for first again.", func(t *testing.T) {
-		testutil.Ok(t, store.removeBlock(b2.meta.ULID))
+		b := store.blocks[b2.meta.ULID]
+		lset := labels.FromMap(b.meta.Thanos.Labels)
+		store.blockSets[lset.Hash()].remove(b2.meta.ULID)
+		delete(store.blocks, b2.meta.ULID)
 
 		srv := newStoreSeriesServer(context.Background())
 		testutil.Ok(t, store.Series(&storepb.SeriesRequest{

diff --git a/test/e2e/store_gateway_test.go b/test/e2e/store_gateway_test.go
@@ -19,6 +19,7 @@ import (
 	"time"
 
 	"github.com/cortexproject/promqlsmith"
+	"github.com/efficientgo/core/backoff"
 	"github.com/efficientgo/core/testutil"
 	"github.com/efficientgo/e2e"
 	e2edb "github.com/efficientgo/e2e/db"
@@ -816,7 +817,13 @@ metafile_content_ttl: 0s`
 	// thanos_blocks_meta_synced: 1x loadedMeta 0x labelExcludedMeta 0x TooFreshMeta.
 	for _, st := range []*e2eobs.Observable{store1, store2, store3} {
 		t.Run(st.Name(), func(t *testing.T) {
-			testutil.Ok(t, st.WaitSumMetrics(e2emon.Equals(1), "thanos_blocks_meta_synced"))
+			testutil.Ok(t, st.WaitSumMetricsWithOptions(e2emon.Equals(1), []string{"thanos_blocks_meta_synced"}, e2emon.WaitMissingMetrics(), e2emon.WithWaitBackoff(
+				&backoff.Config{
+					Min:        1 * time.Second,
+					Max:        10 * time.Second,
+					MaxRetries: 30,
+				},
+			)))
 			testutil.Ok(t, st.WaitSumMetrics(e2emon.Equals(0), "thanos_blocks_meta_sync_failures_total"))
 
 			testutil.Ok(t, st.WaitSumMetrics(e2emon.Equals(1), "thanos_bucket_store_blocks_loaded"))
@@ -826,23 +833,27 @@ metafile_content_ttl: 0s`
 	}
 
 	t.Run("query with groupcache loading from object storage", func(t *testing.T) {
-		queryAndAssertSeries(t, ctx, q.Endpoint("http"), func() string { return testQuery },
-			time.Now, promclient.QueryOptions{
-				Deduplicate: false,
-			},
-			[]model.Metric{
-				{
-					"a":       "1",
-					"b":       "2",
-					"ext1":    "value1",
-					"replica": "1",
+		for i := 0; i < 3; i++ {
+			queryAndAssertSeries(t, ctx, q.Endpoint("http"), func() string { return testQuery },
+				time.Now, promclient.QueryOptions{
+					Deduplicate: false,
 				},
-			},
-		)
+				[]model.Metric{
+					{
+						"a":       "1",
+						"b":       "2",
+						"ext1":    "value1",
+						"replica": "1",
+					},
+				},
+			)
+		}
 
 		for _, st := range []*e2eobs.Observable{store1, store2, store3} {
-			testutil.Ok(t, st.WaitSumMetricsWithOptions(e2emon.Greater(0), []string{`thanos_cache_groupcache_loads_total`}))
-			testutil.Ok(t, st.WaitSumMetricsWithOptions(e2emon.Greater(0), []string{`thanos_store_bucket_cache_operation_hits_total`}, e2emon.WithLabelMatchers(matchers.MustNewMatcher(matchers.MatchEqual, "config", "chunks"))))
+			t.Run(st.Name(), func(t *testing.T) {
+				testutil.Ok(t, st.WaitSumMetricsWithOptions(e2emon.Greater(0), []string{`thanos_cache_groupcache_loads_total`}))
+				testutil.Ok(t, st.WaitSumMetricsWithOptions(e2emon.Greater(0), []string{`thanos_store_bucket_cache_operation_hits_total`}, e2emon.WithLabelMatchers(matchers.MustNewMatcher(matchers.MatchEqual, "config", "chunks"))))
+			})
 		}
 	})