store: lock around iterating over s.blocks

docs/getting-started.md

@@ -162,7 +162,7 @@ See up to date [jsonnet mixins](https://github.com/thanos-io/thanos/tree/main/mi
   * [HelloFresh blog posts part 1](https://engineering.hellofresh.com/monitoring-at-hellofresh-part-1-architecture-677b4bd6b728)
   * [HelloFresh blog posts part 2](https://engineering.hellofresh.com/monitoring-at-hellofresh-part-2-operating-the-monitoring-system-8175cd939c1d)
   * [Thanos deployment](https://www.metricfire.com/blog/ha-kubernetes-monitoring-using-prometheus-and-thanos)
-  * [Taboola user story](https://blog.taboola.com/monitoring-and-metering-scale/)
+  * [Taboola user story](https://www.taboola.com/engineering/monitoring-and-metering-scale/)
   * [Thanos via Prometheus Operator](https://kkc.github.io/2019/02/10/prometheus-operator-with-thanos/)
 
 * 2018:

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

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]++

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
 

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()

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{

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"))))
+			})
 		}
 	})