Repository: cortexproject/cortex
Pull request #2467 (2020-04-15 13:35:07)

Update Thanos to v0.12.0.
**What this PR does**: This PR updated Thanos to just-released [Thanos v0.12.0](https://github.com/thanos-io/thanos/releases/tag/v0.12.0).

Note: `UserBucketClient` struct was split into `UserBucketReaderClient` and `UserBucketClient`. Reason is that new `ReaderWithExpectedErrs` method returns `BucketReader`, while `UserBucketClient` works with "full" `Bucket` only.

**Checklist**
- [x] Tests updated
- [ ] Documentation added
- [ ] `CHANGELOG.md` updated - the order of entries should be `[CHANGE]`, `[FEATURE]`, `[ENHANCEMENT]`, `[BUGFIX]`


pstibrany: Closes #2387 

Patch:
diff --git a/.circleci/config.yml b/.circleci/config.yml
index fd4cea9b7c..f340f1f5f4 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -163,6 +163,7 @@ jobs:
           export CORTEX_CHECKOUT_DIR="/home/circleci/.go_workspace/src/github.com/cortexproject/cortex"
           echo "Running integration tests with image: $CORTEX_IMAGE"
           go test -tags=requires_docker -timeout 600s -v -count=1 ./integration/...
+        no_output_timeout: 20m
 
   build:
     <<: *defaults
diff --git a/go.mod b/go.mod
index 027c77625f..81e9f80354 100644
--- a/go.mod
+++ b/go.mod
@@ -48,7 +48,7 @@ require (
 	github.com/segmentio/fasthash v0.0.0-20180216231524-a72b379d632e
 	github.com/spf13/afero v1.2.2
 	github.com/stretchr/testify v1.4.0
-	github.com/thanos-io/thanos v0.8.1-0.20200326105947-214ff4480e93
+	github.com/thanos-io/thanos v0.12.0
 	github.com/uber/jaeger-client-go v2.20.1+incompatible
 	github.com/weaveworks/common v0.0.0-20200310113808-2708ba4e60a4
 	go.etcd.io/bbolt v1.3.3
diff --git a/go.sum b/go.sum
index baef396093..0db23d30f4 100644
--- a/go.sum
+++ b/go.sum
@@ -777,8 +777,8 @@ github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UV
 github.com/stretchr/testify v1.4.0 h1:2E4SXV/wtOkTonXsotYi4li6zVWxYlZuYNCXe9XRJyk=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
 github.com/thanos-io/thanos v0.8.1-0.20200109203923-552ffa4c1a0d/go.mod h1:usT/TxtJQ7DzinTt+G9kinDQmRS5sxwu0unVKZ9vdcw=
-github.com/thanos-io/thanos v0.8.1-0.20200326105947-214ff4480e93 h1:6SlyEQmt5MivC6HxPGeYaB6pDeackPDzGTYHfjCdh04=
-github.com/thanos-io/thanos v0.8.1-0.20200326105947-214ff4480e93/go.mod h1:PeLHoE5XdPZss/3eLvuxDCFXnM6Sd2Kh+saQIRJVtBE=
+github.com/thanos-io/thanos v0.12.0 h1:Elo3ojC8H39an5D5GqxxUhM5+7FAToXEm8XX8pMZR88=
+github.com/thanos-io/thanos v0.12.0/go.mod h1:+nN9AzmfaIH2e2KJGyRxX0BoUGrRSyZmp+U8ToRxlDc=
 github.com/tidwall/pretty v0.0.0-20180105212114-65a9db5fad51/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
 github.com/tidwall/pretty v1.0.0 h1:HsD+QiTn7sK6flMKIvNmpqz1qrpP3Ps6jOKIKMooyg4=
 github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
diff --git a/pkg/compactor/compactor.go b/pkg/compactor/compactor.go
index 1b1b02e93a..8651d6e99b 100644
--- a/pkg/compactor/compactor.go
+++ b/pkg/compactor/compactor.go
@@ -348,6 +348,7 @@ func (c *Compactor) compactUser(ctx context.Context, userID string) error {
 			ignoreDeletionMarkFilter,
 			deduplicateBlocksFilter,
 		},
+		nil,
 	)
 	if err != nil {
 		return err
diff --git a/pkg/querier/blocks_scanner.go b/pkg/querier/blocks_scanner.go
index 3a99d6b098..58e9889ca6 100644
--- a/pkg/querier/blocks_scanner.go
+++ b/pkg/querier/blocks_scanner.go
@@ -285,6 +285,7 @@ func (d *BlocksScanner) createMetaFetcher(userID string) (block.MetadataFetcher,
 		filepath.Join(d.cfg.CacheDir, userID),
 		userReg,
 		filters,
+		nil,
 	)
 	if err != nil {
 		return nil, err
diff --git a/pkg/storage/tsdb/config.go b/pkg/storage/tsdb/config.go
index cbcd56f951..3bab9993b2 100644
--- a/pkg/storage/tsdb/config.go
+++ b/pkg/storage/tsdb/config.go
@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/alecthomas/units"
+	"github.com/thanos-io/thanos/pkg/store"
 
 	"github.com/cortexproject/cortex/pkg/storage/backend/azure"
 	"github.com/cortexproject/cortex/pkg/storage/backend/filesystem"
@@ -168,6 +169,13 @@ type BucketStoreConfig struct {
 	ConsistencyDelay         time.Duration    `yaml:"consistency_delay"`
 	IndexCache               IndexCacheConfig `yaml:"index_cache"`
 	IgnoreDeletionMarksDelay time.Duration    `yaml:"ignore_deletion_mark_delay"`
+
+	// Controls what is the ratio of postings offsets store will hold in memory.
+	// Larger value will keep less offsets, which will increase CPU cycles needed for query touching those postings.
+	// It's meant for setups that want low baseline memory pressure and where less traffic is expected.
+	// On the contrary, smaller value will increase baseline memory usage, but improve latency slightly.
+	// 1 will keep all in memory. Default value is the same as in Prometheus which gives a good balance.
+	PostingOffsetsInMemSampling int `yaml:"postings_offsets_in_mem_sampling" doc:"hidden"`
 }
 
 // RegisterFlags registers the BucketStore flags
@@ -187,6 +195,7 @@ func (cfg *BucketStoreConfig) RegisterFlags(f *flag.FlagSet) {
 	f.DurationVar(&cfg.IgnoreDeletionMarksDelay, "experimental.tsdb.bucket-store.ignore-deletion-marks-delay", time.Hour*6, "Duration after which the blocks marked for deletion will be filtered out while fetching blocks. "+
 		"The idea of ignore-deletion-marks-delay is to ignore blocks that are marked for deletion with some delay. This ensures store can still serve blocks that are meant to be deleted but do not have a replacement yet."+
 		"Default is 6h, half of the default value for -compactor.deletion-delay.")
+	f.IntVar(&cfg.PostingOffsetsInMemSampling, "experimental.tsdb.bucket-store.posting-offsets-in-mem-sampling", store.DefaultPostingOffsetInMemorySampling, "Controls what is the ratio of postings offsets that the store will hold in memory.")
 }
 
 // Validate the config.
diff --git a/pkg/storage/tsdb/user_bucket_client.go b/pkg/storage/tsdb/user_bucket_client.go
index bb1cb04b19..1705748f2e 100644
--- a/pkg/storage/tsdb/user_bucket_client.go
+++ b/pkg/storage/tsdb/user_bucket_client.go
@@ -9,21 +9,29 @@ import (
 	"github.com/thanos-io/thanos/pkg/objstore"
 )
 
+// UserBucketReaderClient is a wrapper around a objstore.BucketReader that reads from user-specific subfolder.
+type UserBucketReaderClient struct {
+	userID string
+	bucket objstore.BucketReader
+}
+
 // UserBucketClient is a wrapper around a objstore.Bucket that prepends writes with a userID
 type UserBucketClient struct {
-	userID string
+	UserBucketReaderClient
 	bucket objstore.Bucket
 }
 
-// NewUserBucketClient makes a new UserBucketClient.
 func NewUserBucketClient(userID string, bucket objstore.Bucket) *UserBucketClient {
 	return &UserBucketClient{
-		userID: userID,
+		UserBucketReaderClient: UserBucketReaderClient{
+			userID: userID,
+			bucket: bucket,
+		},
 		bucket: bucket,
 	}
 }
 
-func (b *UserBucketClient) fullName(name string) string {
+func (b *UserBucketReaderClient) fullName(name string) string {
 	return fmt.Sprintf("%s/%s", b.userID, name)
 }
 
@@ -45,7 +53,7 @@ func (b *UserBucketClient) Name() string { return b.bucket.Name() }
 
 // Iter calls f for each entry in the given directory (not recursive.). The argument to f is the full
 // object name including the prefix of the inspected directory.
-func (b *UserBucketClient) Iter(ctx context.Context, dir string, f func(string) error) error {
+func (b *UserBucketReaderClient) Iter(ctx context.Context, dir string, f func(string) error) error {
 	return b.bucket.Iter(ctx, b.fullName(dir), func(s string) error {
 		/*
 			Since all objects are prefixed with the userID we need to strip the userID
@@ -56,26 +64,57 @@ func (b *UserBucketClient) Iter(ctx context.Context, dir string, f func(string)
 }
 
 // Get returns a reader for the given object name.
-func (b *UserBucketClient) Get(ctx context.Context, name string) (io.ReadCloser, error) {
+func (b *UserBucketReaderClient) Get(ctx context.Context, name string) (io.ReadCloser, error) {
 	return b.bucket.Get(ctx, b.fullName(name))
 }
 
 // GetRange returns a new range reader for the given object name and range.
-func (b *UserBucketClient) GetRange(ctx context.Context, name string, off, length int64) (io.ReadCloser, error) {
+func (b *UserBucketReaderClient) GetRange(ctx context.Context, name string, off, length int64) (io.ReadCloser, error) {
 	return b.bucket.GetRange(ctx, b.fullName(name), off, length)
 }
 
 // Exists checks if the given object exists in the bucket.
-func (b *UserBucketClient) Exists(ctx context.Context, name string) (bool, error) {
+func (b *UserBucketReaderClient) Exists(ctx context.Context, name string) (bool, error) {
 	return b.bucket.Exists(ctx, b.fullName(name))
 }
 
 // IsObjNotFoundErr returns true if error means that object is not found. Relevant to Get operations.
-func (b *UserBucketClient) IsObjNotFoundErr(err error) bool {
+func (b *UserBucketReaderClient) IsObjNotFoundErr(err error) bool {
 	return b.bucket.IsObjNotFoundErr(err)
 }
 
 // ObjectSize returns the size of the specified object.
-func (b *UserBucketClient) ObjectSize(ctx context.Context, name string) (uint64, error) {
+func (b *UserBucketReaderClient) ObjectSize(ctx context.Context, name string) (uint64, error) {
 	return b.bucket.ObjectSize(ctx, b.fullName(name))
 }
+
+// ReaderWithExpectedErrs allows to specify a filter that marks certain errors as expected, so it will not increment
+// thanos_objstore_bucket_operation_failures_total metric.
+func (b *UserBucketReaderClient) ReaderWithExpectedErrs(fn objstore.IsOpFailureExpectedFunc) objstore.BucketReader {
+	if ib, ok := b.bucket.(objstore.InstrumentedBucketReader); ok {
+		return &UserBucketReaderClient{
+			userID: b.userID,
+			bucket: ib.ReaderWithExpectedErrs(fn),
+		}
+	}
+
+	return b
+}
+
+// WithExpectedErrs allows to specify a filter that marks certain errors as expected, so it will not increment
+// thanos_objstore_bucket_operation_failures_total metric.
+func (b *UserBucketClient) WithExpectedErrs(fn objstore.IsOpFailureExpectedFunc) objstore.Bucket {
+	if ib, ok := b.bucket.(objstore.InstrumentedBucket); ok {
+		nb := ib.WithExpectedErrs(fn)
+
+		return &UserBucketClient{
+			UserBucketReaderClient: UserBucketReaderClient{
+				userID: b.userID,
+				bucket: nb,
+			},
+			bucket: nb,
+		}
+	}
+
+	return b
+}
diff --git a/pkg/storegateway/bucket_stores.go b/pkg/storegateway/bucket_stores.go
index 60569d4713..3deaac6254 100644
--- a/pkg/storegateway/bucket_stores.go
+++ b/pkg/storegateway/bucket_stores.go
@@ -226,6 +226,7 @@ func (u *BucketStores) getOrCreateStore(userID string) (*store.BucketStore, erro
 			// TODO(pracucci) can this cause troubles with the upcoming blocks sharding in the store-gateway?
 			block.NewDeduplicateFilter(),
 		}...),
+		nil,
 	)
 	if err != nil {
 		return nil, err
@@ -248,6 +249,7 @@ func (u *BucketStores) getOrCreateStore(userID string) (*store.BucketStore, erro
 		false, // No need to enable backward compatibility with Thanos pre 0.8.0 queriers
 		u.cfg.BucketStore.BinaryIndexHeader,
 		u.cfg.BucketStore.IndexCache.PostingsCompression,
+		u.cfg.BucketStore.PostingOffsetsInMemSampling,
 	)
 	if err != nil {
 		return nil, err
diff --git a/pkg/storegateway/metadata_fetcher_metrics.go b/pkg/storegateway/metadata_fetcher_metrics.go
index 50e3a61a15..a8e8f6b40c 100644
--- a/pkg/storegateway/metadata_fetcher_metrics.go
+++ b/pkg/storegateway/metadata_fetcher_metrics.go
@@ -24,6 +24,7 @@ type MetadataFetcherMetrics struct {
 
 	// Ignored:
 	// blocks_meta_modified
+	// blocks_meta_base_syncs_total
 }
 
 func NewMetadataFetcherMetrics() *MetadataFetcherMetrics {
diff --git a/vendor/github.com/thanos-io/thanos/pkg/block/block.go b/vendor/github.com/thanos-io/thanos/pkg/block/block.go
index 40398cf3d9..2bc7f86aae 100644
--- a/vendor/github.com/thanos-io/thanos/pkg/block/block.go
+++ b/vendor/github.com/thanos-io/thanos/pkg/block/block.go
@@ -21,6 +21,7 @@ import (
 	"github.com/go-kit/kit/log/level"
 	"github.com/oklog/ulid"
 	"github.com/pkg/errors"
+	"github.com/prometheus/client_golang/prometheus"
 
 	"github.com/thanos-io/thanos/pkg/block/metadata"
 	"github.com/thanos-io/thanos/pkg/objstore"
@@ -129,14 +130,15 @@ func cleanUp(logger log.Logger, bkt objstore.Bucket, id ulid.ULID, err error) er
 }
 
 // MarkForDeletion creates a file which stores information about when the block was marked for deletion.
-func MarkForDeletion(ctx context.Context, logger log.Logger, bkt objstore.Bucket, id ulid.ULID) error {
+func MarkForDeletion(ctx context.Context, logger log.Logger, bkt objstore.Bucket, id ulid.ULID, markedForDeletion prometheus.Counter) error {
 	deletionMarkFile := path.Join(id.String(), metadata.DeletionMarkFilename)
 	deletionMarkExists, err := bkt.Exists(ctx, deletionMarkFile)
 	if err != nil {
 		return errors.Wrapf(err, "check exists %s in bucket", deletionMarkFile)
 	}
 	if deletionMarkExists {
-		return errors.Errorf("file %s already exists in bucket", deletionMarkFile)
+		level.Warn(logger).Log("msg", "requested to mark for deletion, but file already exists; this should not happen; investigate", "err", errors.Errorf("file %s already exists in bucket", deletionMarkFile))
+		return nil
 	}
 
 	deletionMark, err := json.Marshal(metadata.DeletionMark{
@@ -148,10 +150,10 @@ func MarkForDeletion(ctx context.Context, logger log.Logger, bkt objstore.Bucket
 		return errors.Wrap(err, "json encode deletion mark")
 	}
 
-	if err := bkt.Upload(ctx, deletionMarkFile, bytes.NewReader(deletionMark)); err != nil {
+	if err := bkt.Upload(ctx, deletionMarkFile, bytes.NewBuffer(deletionMark)); err != nil {
 		return errors.Wrapf(err, "upload file %s to bucket", deletionMarkFile)
 	}
-
+	markedForDeletion.Inc()
 	level.Info(logger).Log("msg", "block has been marked for deletion", "block", id)
 	return nil
 }
@@ -168,6 +170,7 @@ func Delete(ctx context.Context, logger log.Logger, bkt objstore.Bucket, id ulid
 	if err != nil {
 		return errors.Wrapf(err, "stat %s", metaFile)
 	}
+
 	if ok {
 		if err := bkt.Delete(ctx, metaFile); err != nil {
 			return errors.Wrapf(err, "delete %s", metaFile)
diff --git a/vendor/github.com/thanos-io/thanos/pkg/block/fetcher.go b/vendor/github.com/thanos-io/thanos/pkg/block/fetcher.go
index 80377041d9..7f21d4d801 100644
--- a/vendor/github.com/thanos-io/thanos/pkg/block/fetcher.go
+++ b/vendor/github.com/thanos-io/thanos/pkg/block/fetcher.go
@@ -16,6 +16,7 @@ import (
 
 	"github.com/go-kit/kit/log"
 	"github.com/go-kit/kit/log/level"
+	"github.com/golang/groupcache/singleflight"
 	"github.com/oklog/ulid"
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
@@ -30,6 +31,7 @@ import (
 	"github.com/thanos-io/thanos/pkg/model"
 	"github.com/thanos-io/thanos/pkg/objstore"
 	"github.com/thanos-io/thanos/pkg/runutil"
+	"golang.org/x/sync/errgroup"
 )
 
 type fetcherMetrics struct {
@@ -91,11 +93,13 @@ func newFetcherMetrics(reg prometheus.Registerer) *fetcherMetrics {
 		Help:      "Duration of the blocks metadata synchronization in seconds",
 		Buckets:   []float64{0.01, 1, 10, 100, 1000},
 	})
-	m.synced = extprom.NewTxGaugeVec(reg, prometheus.GaugeOpts{
-		Subsystem: fetcherSubSys,
-		Name:      "synced",
-		Help:      "Number of block metadata synced",
-	},
+	m.synced = extprom.NewTxGaugeVec(
+		reg,
+		prometheus.GaugeOpts{
+			Subsystem: fetcherSubSys,
+			Name:      "synced",
+			Help:      "Number of block metadata synced",
+		},
 		[]string{"state"},
 		[]string{corruptedMeta},
 		[]string{noMeta},
@@ -107,11 +111,13 @@ func newFetcherMetrics(reg prometheus.Registerer) *fetcherMetrics {
 		[]string{duplicateMeta},
 		[]string{markedForDeletionMeta},
 	)
-	m.modified = extprom.NewTxGaugeVec(reg, prometheus.GaugeOpts{
-		Subsystem: fetcherSubSys,
-		Name:      "modified",
-		Help:      "Number of blocks that their metadata modified",
-	},
+	m.modified = extprom.NewTxGaugeVec(
+		reg,
+		prometheus.GaugeOpts{
+			Subsystem: fetcherSubSys,
+			Name:      "modified",
+			Help:      "Number of blocks whose metadata changed",
+		},
 		[]string{"modified"},
 		[]string{replicaRemovedMeta},
 	)
@@ -120,6 +126,7 @@ func newFetcherMetrics(reg prometheus.Registerer) *fetcherMetrics {
 
 type MetadataFetcher interface {
 	Fetch(ctx context.Context) (metas map[ulid.ULID]*metadata.Meta, partial map[ulid.ULID]error, err error)
+	UpdateOnChange(func([]metadata.Meta, error))
 }
 
 type MetadataFilter interface {
@@ -130,25 +137,22 @@ type MetadataModifier interface {
 	Modify(ctx context.Context, metas map[ulid.ULID]*metadata.Meta, modified *extprom.TxGaugeVec, incompleteView bool) error
 }
 
-// MetaFetcher is a struct that synchronizes filtered metadata of all block in the object storage with the local state.
-// Not go-routine safe.
-type MetaFetcher struct {
+// BaseFetcher is a struct that synchronizes filtered metadata of all block in the object storage with the local state.
+// Go-routine safe.
+type BaseFetcher struct {
 	logger      log.Logger
 	concurrency int
-	bkt         objstore.BucketReader
+	bkt         objstore.InstrumentedBucketReader
 
 	// Optional local directory to cache meta.json files.
 	cacheDir string
-	metrics  *fetcherMetrics
-
-	filters   []MetadataFilter
-	modifiers []MetadataModifier
-
-	cached map[ulid.ULID]*metadata.Meta
+	cached   map[ulid.ULID]*metadata.Meta
+	syncs    prometheus.Counter
+	g        singleflight.Group
 }
 
-// NewMetaFetcher constructs MetaFetcher.
-func NewMetaFetcher(logger log.Logger, concurrency int, bkt objstore.BucketReader, dir string, r prometheus.Registerer, filters []MetadataFilter, modifiers ...MetadataModifier) (*MetaFetcher, error) {
+// NewBaseFetcher constructs BaseFetcher.
+func NewBaseFetcher(logger log.Logger, concurrency int, bkt objstore.InstrumentedBucketReader, dir string, reg prometheus.Registerer) (*BaseFetcher, error) {
 	if logger == nil {
 		logger = log.NewNopLogger()
 	}
@@ -161,18 +165,34 @@ func NewMetaFetcher(logger log.Logger, concurrency int, bkt objstore.BucketReade
 		}
 	}
 
-	return &MetaFetcher{
-		logger:      log.With(logger, "component", "block.MetaFetcher"),
+	return &BaseFetcher{
+		logger:      log.With(logger, "component", "block.BaseFetcher"),
 		concurrency: concurrency,
 		bkt:         bkt,
 		cacheDir:    cacheDir,
-		metrics:     newFetcherMetrics(r),
-		filters:     filters,
-		modifiers:   modifiers,
 		cached:      map[ulid.ULID]*metadata.Meta{},
+		syncs: promauto.With(reg).NewCounter(prometheus.CounterOpts{
+			Subsystem: fetcherSubSys,
+			Name:      "base_syncs_total",
+			Help:      "Total blocks metadata synchronization attempts by base Fetcher",
+		}),
 	}, nil
 }
 
+// NewMetaFetcher returns meta fetcher.
+func NewMetaFetcher(logger log.Logger, concurrency int, bkt objstore.InstrumentedBucketReader, dir string, reg prometheus.Registerer, filters []MetadataFilter, modifiers []MetadataModifier) (*MetaFetcher, error) {
+	b, err := NewBaseFetcher(logger, concurrency, bkt, dir, reg)
+	if err != nil {
+		return nil, err
+	}
+	return b.NewMetaFetcher(reg, filters, modifiers), nil
+}
+
+// NewMetaFetcher transforms BaseFetcher into actually usable *MetaFetcher.
+func (f *BaseFetcher) NewMetaFetcher(reg prometheus.Registerer, filters []MetadataFilter, modifiers []MetadataModifier, logTags ...interface{}) *MetaFetcher {
+	return &MetaFetcher{metrics: newFetcherMetrics(reg), wrapped: f, filters: filters, modifiers: modifiers, logger: log.With(f.logger, logTags...)}
+}
+
 var (
 	ErrorSyncMetaNotFound  = errors.New("meta.json not found")
 	ErrorSyncMetaCorrupted = errors.New("meta.json corrupted")
@@ -180,16 +200,16 @@ var (
 
 // loadMeta returns metadata from object storage or error.
 // It returns `ErrorSyncMetaNotFound` and `ErrorSyncMetaCorrupted` sentinel errors in those cases.
-func (s *MetaFetcher) loadMeta(ctx context.Context, id ulid.ULID) (*metadata.Meta, error) {
+func (f *BaseFetcher) loadMeta(ctx context.Context, id ulid.ULID) (*metadata.Meta, error) {
 	var (
 		metaFile       = path.Join(id.String(), MetaFilename)
-		cachedBlockDir = filepath.Join(s.cacheDir, id.String())
+		cachedBlockDir = filepath.Join(f.cacheDir, id.String())
 	)
 
 	// TODO(bwplotka): If that causes problems (obj store rate limits), add longer ttl to cached items.
 	// For 1y and 100 block sources this generates ~1.5-3k HEAD RPM. AWS handles 330k RPM per prefix.
 	// TODO(bwplotka): Consider filtering by consistency delay here (can't do until compactor healthyOverride work).
-	ok, err := s.bkt.Exists(ctx, metaFile)
+	ok, err := f.bkt.Exists(ctx, metaFile)
 	if err != nil {
 		return nil, errors.Wrapf(err, "meta.json file exists: %v", metaFile)
 	}
@@ -197,27 +217,27 @@ func (s *MetaFetcher) loadMeta(ctx context.Context, id ulid.ULID) (*metadata.Met
 		return nil, ErrorSyncMetaNotFound
 	}
 
-	if m, seen := s.cached[id]; seen {
+	if m, seen := f.cached[id]; seen {
 		return m, nil
 	}
 
 	// Best effort load from local dir.
-	if s.cacheDir != "" {
+	if f.cacheDir != "" {
 		m, err := metadata.Read(cachedBlockDir)
 		if err == nil {
 			return m, nil
 		}
 
 		if !errors.Is(err, os.ErrNotExist) {
-			level.Warn(s.logger).Log("msg", "best effort read of the local meta.json failed; removing cached block dir", "dir", cachedBlockDir, "err", err)
+			level.Warn(f.logger).Log("msg", "best effort read of the local meta.json failed; removing cached block dir", "dir", cachedBlockDir, "err", err)
 			if err := os.RemoveAll(cachedBlockDir); err != nil {
-				level.Warn(s.logger).Log("msg", "best effort remove of cached dir failed; ignoring", "dir", cachedBlockDir, "err", err)
+				level.Warn(f.logger).Log("msg", "best effort remove of cached dir failed; ignoring", "dir", cachedBlockDir, "err", err)
 			}
 		}
 	}
 
-	r, err := s.bkt.Get(ctx, metaFile)
-	if s.bkt.IsObjNotFoundErr(err) {
+	r, err := f.bkt.ReaderWithExpectedErrs(f.bkt.IsObjNotFoundErr).Get(ctx, metaFile)
+	if f.bkt.IsObjNotFoundErr(err) {
 		// Meta.json was deleted between bkt.Exists and here.
 		return nil, errors.Wrapf(ErrorSyncMetaNotFound, "%v", err)
 	}
@@ -225,7 +245,7 @@ func (s *MetaFetcher) loadMeta(ctx context.Context, id ulid.ULID) (*metadata.Met
 		return nil, errors.Wrapf(err, "get meta file: %v", metaFile)
 	}
 
-	defer runutil.CloseWithLogOnErr(s.logger, r, "close bkt meta get")
+	defer runutil.CloseWithLogOnErr(f.logger, r, "close bkt meta get")
 
 	metaContent, err := ioutil.ReadAll(r)
 	if err != nil {
@@ -242,166 +262,232 @@ func (s *MetaFetcher) loadMeta(ctx context.Context, id ulid.ULID) (*metadata.Met
 	}
 
 	// Best effort cache in local dir.
-	if s.cacheDir != "" {
+	if f.cacheDir != "" {
 		if err := os.MkdirAll(cachedBlockDir, os.ModePerm); err != nil {
-			level.Warn(s.logger).Log("msg", "best effort mkdir of the meta.json block dir failed; ignoring", "dir", cachedBlockDir, "err", err)
+			level.Warn(f.logger).Log("msg", "best effort mkdir of the meta.json block dir failed; ignoring", "dir", cachedBlockDir, "err", err)
 		}
 
-		if err := metadata.Write(s.logger, cachedBlockDir, m); err != nil {
-			level.Warn(s.logger).Log("msg", "best effort save of the meta.json to local dir failed; ignoring", "dir", cachedBlockDir, "err", err)
+		if err := metadata.Write(f.logger, cachedBlockDir, m); err != nil {
+			level.Warn(f.logger).Log("msg", "best effort save of the meta.json to local dir failed; ignoring", "dir", cachedBlockDir, "err", err)
 		}
 	}
 	return m, nil
 }
 
-// Fetch returns all block metas as well as partial blocks (blocks without or with corrupted meta file) from the bucket.
-// It's caller responsibility to not change the returned metadata files. Maps can be modified.
-//
-// Returned error indicates a failure in fetching metadata. Returned meta can be assumed as correct, with some blocks missing.
-func (s *MetaFetcher) Fetch(ctx context.Context) (metas map[ulid.ULID]*metadata.Meta, partial map[ulid.ULID]error, err error) {
-	start := time.Now()
-	defer func() {
-		s.metrics.syncDuration.Observe(time.Since(start).Seconds())
-		if err != nil {
-			s.metrics.syncFailures.Inc()
-		}
-	}()
-	s.metrics.syncs.Inc()
+type response struct {
+	metas   map[ulid.ULID]*metadata.Meta
+	partial map[ulid.ULID]error
+	// If metaErr > 0 it means incomplete view, so some metas, failed to be loaded.
+	metaErrs tsdberrors.MultiError
 
-	metas = make(map[ulid.ULID]*metadata.Meta)
-	partial = make(map[ulid.ULID]error)
+	noMetas        float64
+	corruptedMetas float64
+
+	incompleteView bool
+}
+
+func (f *BaseFetcher) fetchMetadata(ctx context.Context) (interface{}, error) {
+	f.syncs.Inc()
 
 	var (
-		wg  sync.WaitGroup
-		ch  = make(chan ulid.ULID, s.concurrency)
+		resp = response{
+			metas:   make(map[ulid.ULID]*metadata.Meta),
+			partial: make(map[ulid.ULID]error),
+		}
+		eg  errgroup.Group
+		ch  = make(chan ulid.ULID, f.concurrency)
 		mtx sync.Mutex
-
-		metaErrs tsdberrors.MultiError
 	)
-
-	s.metrics.resetTx()
-
-	for i := 0; i < s.concurrency; i++ {
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-
+	for i := 0; i < f.concurrency; i++ {
+		eg.Go(func() error {
 			for id := range ch {
-				meta, err := s.loadMeta(ctx, id)
+				meta, err := f.loadMeta(ctx, id)
 				if err == nil {
 					mtx.Lock()
-					metas[id] = meta
+					resp.metas[id] = meta
 					mtx.Unlock()
 					continue
 				}
 
 				switch errors.Cause(err) {
 				default:
-					s.metrics.synced.WithLabelValues(failedMeta).Inc()
 					mtx.Lock()
-					metaErrs.Add(err)
+					resp.metaErrs.Add(err)
 					mtx.Unlock()
 					continue
 				case ErrorSyncMetaNotFound:
-					s.metrics.synced.WithLabelValues(noMeta).Inc()
+					mtx.Lock()
+					resp.noMetas++
+					mtx.Unlock()
 				case ErrorSyncMetaCorrupted:
-					s.metrics.synced.WithLabelValues(corruptedMeta).Inc()
+					mtx.Lock()
+					resp.corruptedMetas++
+					mtx.Unlock()
 				}
 
 				mtx.Lock()
-				partial[id] = err
+				resp.partial[id] = err
 				mtx.Unlock()
 			}
-		}()
+			return nil
+		})
 	}
 
 	// Workers scheduled, distribute blocks.
-	err = s.bkt.Iter(ctx, "", func(name string) error {
-		id, ok := IsBlockDir(name)
-		if !ok {
-			return nil
-		}
+	eg.Go(func() error {
+		defer close(ch)
+		return f.bkt.Iter(ctx, "", func(name string) error {
+			id, ok := IsBlockDir(name)
+			if !ok {
+				return nil
+			}
 
-		select {
-		case <-ctx.Done():
-			return ctx.Err()
-		case ch <- id:
-		}
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			case ch <- id:
+			}
 
-		return nil
+			return nil
+		})
 	})
-	close(ch)
 
-	wg.Wait()
-	if err != nil {
-		return nil, nil, errors.Wrap(err, "MetaFetcher: iter bucket")
+	if err := eg.Wait(); err != nil {
+		return nil, errors.Wrap(err, "BaseFetcher: iter bucket")
 	}
 
-	incompleteView := len(metaErrs) > 0
+	if len(resp.metaErrs) > 0 {
+		return resp, nil
+	}
 
 	// Only for complete view of blocks update the cache.
-	if !incompleteView {
-		cached := make(map[ulid.ULID]*metadata.Meta, len(metas))
-		for id, m := range metas {
-			cached[id] = m
-		}
-		s.cached = cached
-
-		// Best effort cleanup of disk-cached metas.
-		if s.cacheDir != "" {
-			names, err := fileutil.ReadDir(s.cacheDir)
-			if err != nil {
-				level.Warn(s.logger).Log("msg", "best effort remove of not needed cached dirs failed; ignoring", "err", err)
-			} else {
-				for _, n := range names {
-					id, ok := IsBlockDir(n)
-					if !ok {
-						continue
-					}
-
-					if _, ok := metas[id]; ok {
-						continue
-					}
-
-					cachedBlockDir := filepath.Join(s.cacheDir, id.String())
-
-					// No such block loaded, remove the local dir.
-					if err := os.RemoveAll(cachedBlockDir); err != nil {
-						level.Warn(s.logger).Log("msg", "best effort remove of not needed cached dir failed; ignoring", "dir", cachedBlockDir, "err", err)
-					}
+	cached := make(map[ulid.ULID]*metadata.Meta, len(resp.metas))
+	for id, m := range resp.metas {
+		cached[id] = m
+	}
+	f.cached = cached
+
+	// Best effort cleanup of disk-cached metas.
+	if f.cacheDir != "" {
+		names, err := fileutil.ReadDir(f.cacheDir)
+		if err != nil {
+			level.Warn(f.logger).Log("msg", "best effort remove of not needed cached dirs failed; ignoring", "err", err)
+		} else {
+			for _, n := range names {
+				id, ok := IsBlockDir(n)
+				if !ok {
+					continue
+				}
+
+				if _, ok := resp.metas[id]; ok {
+					continue
+				}
+
+				cachedBlockDir := filepath.Join(f.cacheDir, id.String())
+
+				// No such block loaded, remove the local dir.
+				if err := os.RemoveAll(cachedBlockDir); err != nil {
+					level.Warn(f.logger).Log("msg", "best effort remove of not needed cached dir failed; ignoring", "dir", cachedBlockDir, "err", err)
 				}
 			}
 		}
 	}
+	return resp, nil
+}
+
+func (f *BaseFetcher) fetch(ctx context.Context, metrics *fetcherMetrics, filters []MetadataFilter, modifiers []MetadataModifier) (_ map[ulid.ULID]*metadata.Meta, _ map[ulid.ULID]error, err error) {
+	start := time.Now()
+	defer func() {
+		metrics.syncDuration.Observe(time.Since(start).Seconds())
+		if err != nil {
+			metrics.syncFailures.Inc()
+		}
+	}()
+	metrics.syncs.Inc()
+	metrics.resetTx()
+
+	// Run this in thread safe run group.
+	// TODO(bwplotka): Consider custom singleflight with ttl.
+	v, err := f.g.Do("", func() (i interface{}, err error) {
+		// NOTE: First go routine context will go through.
+		return f.fetchMetadata(ctx)
+	})
+	if err != nil {
+		return nil, nil, err
+	}
+	resp := v.(response)
+
+	// Copy as same response might be reused by different goroutines.
+	metas := make(map[ulid.ULID]*metadata.Meta, len(resp.metas))
+	for id, m := range resp.metas {
+		metas[id] = m
+	}
+
+	metrics.synced.WithLabelValues(failedMeta).Set(float64(len(resp.metaErrs)))
+	metrics.synced.WithLabelValues(noMeta).Set(resp.noMetas)
+	metrics.synced.WithLabelValues(corruptedMeta).Set(resp.corruptedMetas)
 
-	for _, f := range s.filters {
+	for _, filter := range filters {
 		// NOTE: filter can update synced metric accordingly to the reason of the exclude.
-		if err := f.Filter(ctx, metas, s.metrics.synced, incompleteView); err != nil {
+		if err := filter.Filter(ctx, metas, metrics.synced, resp.incompleteView); err != nil {
 			return nil, nil, errors.Wrap(err, "filter metas")
 		}
 	}
 
-	for _, m := range s.modifiers {
+	for _, m := range modifiers {
 		// NOTE: modifier can update modified metric accordingly to the reason of the modification.
-		if err := m.Modify(ctx, metas, s.metrics.modified, incompleteView); err != nil {
+		if err := m.Modify(ctx, metas, metrics.modified, resp.incompleteView); err != nil {
 			return nil, nil, errors.Wrap(err, "modify metas")
 		}
 	}
 
-	s.metrics.synced.WithLabelValues(loadedMeta).Set(float64(len(metas)))
-	s.metrics.submit()
+	metrics.synced.WithLabelValues(loadedMeta).Set(float64(len(metas)))
+	metrics.submit()
 
-	if incompleteView {
-		return metas, partial, errors.Wrap(metaErrs, "incomplete view")
+	if len(resp.metaErrs) > 0 {
+		return metas, resp.partial, errors.Wrap(resp.metaErrs, "incomplete view")
 	}
 
-	level.Debug(s.logger).Log("msg", "successfully fetched block metadata", "duration", time.Since(start).String(), "cached", len(s.cached), "returned", len(metas), "partial", len(partial))
-	return metas, partial, nil
+	level.Info(f.logger).Log("msg", "successfully synchronized block metadata", "duration", time.Since(start).String(), "cached", len(f.cached), "returned", len(metas), "partial", len(resp.partial))
+	return metas, resp.partial, nil
+}
+
+type MetaFetcher struct {
+	wrapped *BaseFetcher
+	metrics *fetcherMetrics
+
+	filters   []MetadataFilter
+	modifiers []MetadataModifier
+
+	listener func([]metadata.Meta, error)
+
+	logger log.Logger
+}
+
+// Fetch returns all block metas as well as partial blocks (blocks without or with corrupted meta file) from the bucket.
+// It's caller responsibility to not change the returned metadata files. Maps can be modified.
+//
+// Returned error indicates a failure in fetching metadata. Returned meta can be assumed as correct, with some blocks missing.
+func (f *MetaFetcher) Fetch(ctx context.Context) (metas map[ulid.ULID]*metadata.Meta, partial map[ulid.ULID]error, err error) {
+	metas, partial, err = f.wrapped.fetch(ctx, f.metrics, f.filters, f.modifiers)
+	if f.listener != nil {
+		blocks := make([]metadata.Meta, 0, len(metas))
+		for _, meta := range metas {
+			blocks = append(blocks, *meta)
+		}
+		f.listener(blocks, err)
+	}
+	return metas, partial, err
+}
+
+// UpdateOnChange allows to add listener that will be update on every change.
+func (f *MetaFetcher) UpdateOnChange(listener func([]metadata.Meta, error)) {
+	f.listener = listener
 }
 
 var _ MetadataFilter = &TimePartitionMetaFilter{}
 
-// TimePartitionMetaFilter is a MetaFetcher filter that filters out blocks that are outside of specified time range.
+// TimePartitionMetaFilter is a BaseFetcher filter that filters out blocks that are outside of specified time range.
 // Not go-routine safe.
 type TimePartitionMetaFilter struct {
 	minTime, maxTime model.TimeOrDurationValue
@@ -458,7 +544,9 @@ func (f *LabelShardedMetaFilter) Filter(_ context.Context, metas map[ulid.ULID]*
 	return nil
 }
 
-// DeduplicateFilter is a MetaFetcher filter that filters out older blocks that have exactly the same data.
+var _ MetadataFilter = &DeduplicateFilter{}
+
+// DeduplicateFilter is a BaseFetcher filter that filters out older blocks that have exactly the same data.
 // Not go-routine safe.
 type DeduplicateFilter struct {
 	duplicateIDs []ulid.ULID
@@ -488,7 +576,7 @@ func (f *DeduplicateFilter) Filter(_ context.Context, metas map[ulid.ULID]*metad
 				BlockMeta: tsdb.BlockMeta{
 					ULID: ulid.MustNew(uint64(0), nil),
 				},
-			}), metasByResolution[res], metas, res, synced)
+			}), metasByResolution[res], metas, synced)
 		}(res)
 	}
 
@@ -497,7 +585,7 @@ func (f *DeduplicateFilter) Filter(_ context.Context, metas map[ulid.ULID]*metad
 	return nil
 }
 
-func (f *DeduplicateFilter) filterForResolution(root *Node, metaSlice []*metadata.Meta, metas map[ulid.ULID]*metadata.Meta, res int64, synced *extprom.TxGaugeVec) {
+func (f *DeduplicateFilter) filterForResolution(root *Node, metaSlice []*metadata.Meta, metas map[ulid.ULID]*metadata.Meta, synced *extprom.TxGaugeVec) {
 	sort.Slice(metaSlice, func(i, j int) bool {
 		ilen := len(metaSlice[i].Compaction.Sources)
 		jlen := len(metaSlice[j].Compaction.Sources)
@@ -572,7 +660,9 @@ func contains(s1 []ulid.ULID, s2 []ulid.ULID) bool {
 	return true
 }
 
-// ReplicaLabelRemover is a MetaFetcher modifier modifies external labels of existing blocks, it removes given replica labels from the metadata of blocks that have it.
+var _ MetadataModifier = &ReplicaLabelRemover{}
+
+// ReplicaLabelRemover is a BaseFetcher modifier modifies external labels of existing blocks, it removes given replica labels from the metadata of blocks that have it.
 type ReplicaLabelRemover struct {
 	logger log.Logger
 
@@ -600,7 +690,7 @@ func (r *ReplicaLabelRemover) Modify(_ context.Context, metas map[ulid.ULID]*met
 	return nil
 }
 
-// ConsistencyDelayMetaFilter is a MetaFetcher filter that filters out blocks that are created before a specified consistency delay.
+// ConsistencyDelayMetaFilter is a BaseFetcher filter that filters out blocks that are created before a specified consistency delay.
 // Not go-routine safe.
 type ConsistencyDelayMetaFilter struct {
 	logger           log.Logger
@@ -652,12 +742,12 @@ func (f *ConsistencyDelayMetaFilter) Filter(_ context.Context, metas map[ulid.UL
 type IgnoreDeletionMarkFilter struct {
 	logger          log.Logger
 	delay           time.Duration
-	bkt             objstore.BucketReader
+	bkt             objstore.InstrumentedBucketReader
 	deletionMarkMap map[ulid.ULID]*metadata.DeletionMark
 }
 
 // NewIgnoreDeletionMarkFilter creates IgnoreDeletionMarkFilter.
-func NewIgnoreDeletionMarkFilter(logger log.Logger, bkt objstore.BucketReader, delay time.Duration) *IgnoreDeletionMarkFilter {
+func NewIgnoreDeletionMarkFilter(logger log.Logger, bkt objstore.InstrumentedBucketReader, delay time.Duration) *IgnoreDeletionMarkFilter {
 	return &IgnoreDeletionMarkFilter{
 		logger: logger,
 		bkt:    bkt,
diff --git a/vendor/github.com/thanos-io/thanos/pkg/block/indexheader/binary_reader.go b/vendor/github.com/thanos-io/thanos/pkg/block/indexheader/binary_reader.go
index 9005cd443b..a070d9275b 100644
--- a/vendor/github.com/thanos-io/thanos/pkg/block/indexheader/binary_reader.go
+++ b/vendor/github.com/thanos-io/thanos/pkg/block/indexheader/binary_reader.go
@@ -42,8 +42,6 @@ const (
 	// MagicIndex are 4 bytes at the head of an index-header file.
 	MagicIndex = 0xBAAAD792
 
-	symbolFactor = 32
-
 	postingLengthFieldSize = 4
 )
 
@@ -71,20 +69,21 @@ type BinaryTOC struct {
 }
 
 // WriteBinary build index-header file from the pieces of index in object storage.
-func WriteBinary(ctx context.Context, bkt objstore.BucketReader, id ulid.ULID, fn string) (err error) {
+func WriteBinary(ctx context.Context, bkt objstore.BucketReader, id ulid.ULID, filename string) (err error) {
 	ir, indexVersion, err := newChunkedIndexReader(ctx, bkt, id)
 	if err != nil {
 		return errors.Wrap(err, "new index reader")
 	}
+	tmpFilename := filename + ".tmp"
 
 	// Buffer for copying and encbuffers.
 	// This also will control the size of file writer buffer.
 	buf := make([]byte, 32*1024)
-	bw, err := newBinaryWriter(fn, buf)
+	bw, err := newBinaryWriter(tmpFilename, buf)
 	if err != nil {
 		return errors.Wrap(err, "new binary index header writer")
 	}
-	defer runutil.CloseWithErrCapture(&err, bw, "close binary writer for %s", fn)
+	defer runutil.CloseWithErrCapture(&err, bw, "close binary writer for %s", tmpFilename)
 
 	if err := bw.AddIndexMeta(indexVersion, ir.toc.PostingsTable); err != nil {
 		return errors.Wrap(err, "add index meta")
@@ -109,7 +108,17 @@ func WriteBinary(ctx context.Context, bkt objstore.BucketReader, id ulid.ULID, f
 	if err := bw.WriteTOC(); err != nil {
 		return errors.Wrap(err, "write index header TOC")
 	}
-	return nil
+
+	if err := bw.f.Flush(); err != nil {
+		return errors.Wrap(err, "flush")
+	}
+
+	if err := bw.f.f.Sync(); err != nil {
+		return errors.Wrap(err, "sync")
+	}
+
+	// Create index-header in atomic way, to avoid partial writes (e.g during restart or crash of store GW).
+	return os.Rename(tmpFilename, filename)
 }
 
 type chunkedIndexReader struct {
@@ -417,11 +426,12 @@ type BinaryReader struct {
 	c io.Closer
 
 	// Map of LabelName to a list of some LabelValues's position in the offset table.
-	// The first and last values for each name are always present.
+	// The first and last values for each name are always present, we keep only 1/postingOffsetsInMemSampling of the rest.
 	postings map[string]*postingValueOffsets
 	// For the v1 format, labelname -> labelvalue -> offset.
 	postingsV1 map[string]map[string]index.Range
 
+	// Symbols struct that keeps only 1/postingOffsetsInMemSampling in the memory, then looks up the rest via mmap.
 	symbols     *index.Symbols
 	nameSymbols map[uint32]string // Cache of the label name symbol lookups,
 	// as there are not many and they are half of all lookups.
@@ -431,12 +441,14 @@ type BinaryReader struct {
 	version             int
 	indexVersion        int
 	indexLastPostingEnd int64
+
+	postingOffsetsInMemSampling int
 }
 
 // NewBinaryReader loads or builds new index-header if not present on disk.
-func NewBinaryReader(ctx context.Context, logger log.Logger, bkt objstore.BucketReader, dir string, id ulid.ULID) (*BinaryReader, error) {
+func NewBinaryReader(ctx context.Context, logger log.Logger, bkt objstore.BucketReader, dir string, id ulid.ULID, postingOffsetsInMemSampling int) (*BinaryReader, error) {
 	binfn := filepath.Join(dir, id.String(), block.IndexHeaderFilename)
-	br, err := newFileBinaryReader(binfn)
+	br, err := newFileBinaryReader(binfn, postingOffsetsInMemSampling)
 	if err == nil {
 		return br, nil
 	}
@@ -449,11 +461,10 @@ func NewBinaryReader(ctx context.Context, logger log.Logger, bkt objstore.Bucket
 	}
 
 	level.Debug(logger).Log("msg", "built index-header file", "path", binfn, "elapsed", time.Since(start))
-
-	return newFileBinaryReader(binfn)
+	return newFileBinaryReader(binfn, postingOffsetsInMemSampling)
 }
 
-func newFileBinaryReader(path string) (bw *BinaryReader, err error) {
+func newFileBinaryReader(path string, postingOffsetsInMemSampling int) (bw *BinaryReader, err error) {
 	f, err := fileutil.OpenMmapFile(path)
 	if err != nil {
 		return nil, err
@@ -465,9 +476,10 @@ func newFileBinaryReader(path string) (bw *BinaryReader, err error) {
 	}()
 
 	r := &BinaryReader{
-		b:        realByteSlice(f.Bytes()),
-		c:        f,
-		postings: map[string]*postingValueOffsets{},
+		b:                           realByteSlice(f.Bytes()),
+		c:                           f,
+		postings:                    map[string]*postingValueOffsets{},
+		postingOffsetsInMemSampling: postingOffsetsInMemSampling,
 	}
 
 	// Verify header.
@@ -491,6 +503,7 @@ func newFileBinaryReader(path string) (bw *BinaryReader, err error) {
 		return nil, errors.Wrap(err, "read index header TOC")
 	}
 
+	// TODO(bwplotka): Consider contributing to Prometheus to allow specifying custom number for symbolsFactor.
 	r.symbols, err = index.NewSymbols(r.b, r.indexVersion, int(r.toc.Symbols))
 	if err != nil {
 		return nil, errors.Wrap(err, "read symbols")
@@ -540,11 +553,12 @@ func newFileBinaryReader(path string) (bw *BinaryReader, err error) {
 			}
 
 			if _, ok := r.postings[key[0]]; !ok {
-				// Next label name.
+				// Not seen before label name.
 				r.postings[key[0]] = &postingValueOffsets{}
 				if lastKey != nil {
-					if valueCount%symbolFactor != 0 {
-						// Always include last value for each label name.
+					// Always include last value for each label name, unless it was just added in previous iteration based
+					// on valueCount.
+					if (valueCount-1)%postingOffsetsInMemSampling != 0 {
 						r.postings[lastKey[0]].offsets = append(r.postings[lastKey[0]].offsets, postingOffset{value: lastKey[1], tableOff: lastTableOff})
 					}
 					r.postings[lastKey[0]].lastValOffset = int64(off - crc32.Size)
@@ -554,21 +568,24 @@ func newFileBinaryReader(path string) (bw *BinaryReader, err error) {
 			}
 
 			lastKey = key
-			if valueCount%symbolFactor == 0 {
+			lastTableOff = tableOff
+			valueCount++
+
+			if (valueCount-1)%postingOffsetsInMemSampling == 0 {
 				r.postings[key[0]].offsets = append(r.postings[key[0]].offsets, postingOffset{value: key[1], tableOff: tableOff})
-				return nil
 			}
 
-			lastTableOff = tableOff
-			valueCount++
 			return nil
 		}); err != nil {
 			return nil, errors.Wrap(err, "read postings table")
 		}
 		if lastKey != nil {
-			if valueCount%symbolFactor != 0 {
+			if (valueCount-1)%postingOffsetsInMemSampling != 0 {
+				// Always include last value for each label name if not included already based on valueCount.
 				r.postings[lastKey[0]].offsets = append(r.postings[lastKey[0]].offsets, postingOffset{value: lastKey[1], tableOff: lastTableOff})
 			}
+			// In any case lastValOffset is unknown as don't have next posting anymore. Guess from TOC table.
+			// In worst case we will overfetch a few bytes.
 			r.postings[lastKey[0]].lastValOffset = r.indexLastPostingEnd - crc32.Size
 		}
 		// Trim any extra space in the slices.
@@ -636,6 +653,18 @@ func (r BinaryReader) PostingsOffset(name string, value string) (index.Range, er
 	return rngs[0], nil
 }
 
+func skipNAndName(d *encoding.Decbuf, buf *int) {
+	if *buf == 0 {
+		// Keycount+LabelName are always the same number of bytes,
+		// and it's faster to skip than parse.
+		*buf = d.Len()
+		d.Uvarint()      // Keycount.
+		d.UvarintBytes() // Label name.
+		*buf -= d.Len()
+		return
+	}
+	d.Skip(*buf)
+}
 func (r BinaryReader) postingsOffset(name string, values ...string) ([]index.Range, error) {
 	rngs := make([]index.Range, 0, len(values))
 	if r.indexVersion == index.FormatV1 {
@@ -662,76 +691,107 @@ func (r BinaryReader) postingsOffset(name string, values ...string) ([]index.Ran
 		return nil, nil
 	}
 
-	skip := 0
+	buf := 0
 	valueIndex := 0
 	for valueIndex < len(values) && values[valueIndex] < e.offsets[0].value {
 		// Discard values before the start.
 		valueIndex++
 	}
 
-	var tmpRngs []index.Range // The start, end offsets in the postings table in the original index file.
+	var newSameRngs []index.Range // The start, end offsets in the postings table in the original index file.
 	for valueIndex < len(values) {
-		value := values[valueIndex]
+		wantedValue := values[valueIndex]
 
-		i := sort.Search(len(e.offsets), func(i int) bool { return e.offsets[i].value >= value })
+		i := sort.Search(len(e.offsets), func(i int) bool { return e.offsets[i].value >= wantedValue })
 		if i == len(e.offsets) {
 			// We're past the end.
 			break
 		}
-		if i > 0 && e.offsets[i].value != value {
+		if i > 0 && e.offsets[i].value != wantedValue {
 			// Need to look from previous entry.
 			i--
 		}
+
 		// Don't Crc32 the entire postings offset table, this is very slow
 		// so hope any issues were caught at startup.
 		d := encoding.NewDecbufAt(r.b, int(r.toc.PostingsOffsetTable), nil)
 		d.Skip(e.offsets[i].tableOff)
 
-		tmpRngs = tmpRngs[:0]
 		// Iterate on the offset table.
+		newSameRngs = newSameRngs[:0]
 		for d.Err() == nil {
-			if skip == 0 {
-				// These are always the same number of bytes,
-				// and it's faster to skip than parse.
-				skip = d.Len()
-				d.Uvarint()      // Keycount.
-				d.UvarintBytes() // Label name.
-				skip -= d.Len()
-			} else {
-				d.Skip(skip)
+			// Posting format entry is as follows:
+			// │ ┌────────────────────────────────────────┐ │
+			// │ │  n = 2 <1b>                            │ │
+			// │ ├──────────────────────┬─────────────────┤ │
+			// │ │ len(name) <uvarint>  │ name <bytes>    │ │
+			// │ ├──────────────────────┼─────────────────┤ │
+			// │ │ len(value) <uvarint> │ value <bytes>   │ │
+			// │ ├──────────────────────┴─────────────────┤ │
+			// │ │  offset <uvarint64>                    │ │
+			// │ └────────────────────────────────────────┘ │
+			// First, let's skip n and name.
+			skipNAndName(&d, &buf)
+			value := d.UvarintBytes() // Label value.
+			postingOffset := int64(d.Uvarint64())
+
+			if len(newSameRngs) > 0 {
+				// We added some ranges in previous iteration. Use next posting offset as end of all our new ranges.
+				for j := range newSameRngs {
+					newSameRngs[j].End = postingOffset - crc32.Size
+				}
+				rngs = append(rngs, newSameRngs...)
+				newSameRngs = newSameRngs[:0]
 			}
-			v := d.UvarintBytes()                 // Label value.
-			postingOffset := int64(d.Uvarint64()) // Offset.
-			for string(v) >= value {
-				if string(v) == value {
-					tmpRngs = append(tmpRngs, index.Range{Start: postingOffset + postingLengthFieldSize})
+
+			for string(value) >= wantedValue {
+				// If wantedValue is equals of greater than current value, loop over all given wanted values in the values until
+				// this is no longer true or there are no more values wanted.
+				// This ensures we cover case when someone asks for postingsOffset(name, value1, value1, value1).
+
+				// Record on the way if wanted value is equal to the current value.
+				if string(value) == wantedValue {
+					newSameRngs = append(newSameRngs, index.Range{Start: postingOffset + postingLengthFieldSize})
 				}
 				valueIndex++
 				if valueIndex == len(values) {
 					break
 				}
-				value = values[valueIndex]
+				wantedValue = values[valueIndex]
 			}
+
 			if i+1 == len(e.offsets) {
-				for i := range tmpRngs {
-					tmpRngs[i].End = e.lastValOffset
+				// No more offsets for this name.
+				// Break this loop and record lastOffset on the way for ranges we just added if any.
+				for j := range newSameRngs {
+					newSameRngs[j].End = e.lastValOffset
 				}
-				rngs = append(rngs, tmpRngs...)
-				// Need to go to a later postings offset entry, if there is one.
+				rngs = append(rngs, newSameRngs...)
 				break
 			}
 
-			if value >= e.offsets[i+1].value || valueIndex == len(values) {
-				d.Skip(skip)
-				d.UvarintBytes()                      // Label value.
-				postingOffset := int64(d.Uvarint64()) // Offset.
-				for j := range tmpRngs {
-					tmpRngs[j].End = postingOffset - crc32.Size
+			if valueIndex != len(values) && wantedValue <= e.offsets[i+1].value {
+				// wantedValue is smaller or same as the next offset we know about, let's iterate further to add those.
+				continue
+			}
+
+			// Nothing wanted or wantedValue is larger than next offset we know about.
+			// Let's exit and do binary search again / exit if nothing wanted.
+
+			if len(newSameRngs) > 0 {
+				// We added some ranges in this iteration. Use next posting offset as the end of our ranges.
+				// We know it exists as we never go further in this loop than e.offsets[i, i+1].
+
+				skipNAndName(&d, &buf)
+				d.UvarintBytes() // Label value.
+				postingOffset := int64(d.Uvarint64())
+
+				for j := range newSameRngs {
+					newSameRngs[j].End = postingOffset - crc32.Size
 				}
-				rngs = append(rngs, tmpRngs...)
-				// Need to go to a later postings offset entry, if there is one.
-				break
+				rngs = append(rngs, newSameRngs...)
 			}
+			break
 		}
 		if d.Err() != nil {
 			return nil, errors.Wrap(d.Err(), "get postings offset entry")
@@ -776,7 +836,7 @@ func (r BinaryReader) LabelValues(name string) ([]string, error) {
 	if len(e.offsets) == 0 {
 		return nil, nil
 	}
-	values := make([]string, 0, len(e.offsets)*symbolFactor)
+	values := make([]string, 0, len(e.offsets)*r.postingOffsetsInMemSampling)
 
 	d := encoding.NewDecbufAt(r.b, int(r.toc.PostingsOffsetTable), nil)
 	d.Skip(e.offsets[0].tableOff)
diff --git a/vendor/github.com/thanos-io/thanos/pkg/block/indexheader/json_reader.go b/vendor/github.com/thanos-io/thanos/pkg/block/indexheader/json_reader.go
index 5a3654f4d6..fe54534972 100644
--- a/vendor/github.com/thanos-io/thanos/pkg/block/indexheader/json_reader.go
+++ b/vendor/github.com/thanos-io/thanos/pkg/block/indexheader/json_reader.go
@@ -199,7 +199,7 @@ type JSONReader struct {
 }
 
 // NewJSONReader loads or builds new index-cache.json if not present on disk or object storage.
-func NewJSONReader(ctx context.Context, logger log.Logger, bkt objstore.BucketReader, dir string, id ulid.ULID) (*JSONReader, error) {
+func NewJSONReader(ctx context.Context, logger log.Logger, bkt objstore.InstrumentedBucketReader, dir string, id ulid.ULID) (*JSONReader, error) {
 	cachefn := filepath.Join(dir, id.String(), block.IndexCacheFilename)
 	jr, err := newFileJSONReader(logger, cachefn)
 	if err == nil {
@@ -216,7 +216,7 @@ func NewJSONReader(ctx context.Context, logger log.Logger, bkt objstore.BucketRe
 	}
 
 	// Try to download index cache file from object store.
-	if err = objstore.DownloadFile(ctx, logger, bkt, filepath.Join(id.String(), block.IndexCacheFilename), cachefn); err == nil {
+	if err = objstore.DownloadFile(ctx, logger, bkt.ReaderWithExpectedErrs(bkt.IsObjNotFoundErr), filepath.Join(id.String(), block.IndexCacheFilename), cachefn); err == nil {
 		return newFileJSONReader(logger, cachefn)
 	}
 
diff --git a/vendor/github.com/thanos-io/thanos/pkg/block/metadata/deletionmark.go b/vendor/github.com/thanos-io/thanos/pkg/block/metadata/deletionmark.go
index bda97e92c6..5f2a9f04ad 100644
--- a/vendor/github.com/thanos-io/thanos/pkg/block/metadata/deletionmark.go
+++ b/vendor/github.com/thanos-io/thanos/pkg/block/metadata/deletionmark.go
@@ -45,10 +45,10 @@ type DeletionMark struct {
 }
 
 // ReadDeletionMark reads the given deletion mark file from <dir>/deletion-mark.json in bucket.
-func ReadDeletionMark(ctx context.Context, bkt objstore.BucketReader, logger log.Logger, dir string) (*DeletionMark, error) {
+func ReadDeletionMark(ctx context.Context, bkt objstore.InstrumentedBucketReader, logger log.Logger, dir string) (*DeletionMark, error) {
 	deletionMarkFile := path.Join(dir, DeletionMarkFilename)
 
-	r, err := bkt.Get(ctx, deletionMarkFile)
+	r, err := bkt.ReaderWithExpectedErrs(bkt.IsObjNotFoundErr).Get(ctx, deletionMarkFile)
 	if err != nil {
 		if bkt.IsObjNotFoundErr(err) {
 			return nil, ErrorDeletionMarkNotFound
diff --git a/vendor/github.com/thanos-io/thanos/pkg/compact/blocks_cleaner.go b/vendor/github.com/thanos-io/thanos/pkg/compact/blocks_cleaner.go
index 820346a1bf..7381505118 100644
--- a/vendor/github.com/thanos-io/thanos/pkg/compact/blocks_cleaner.go
+++ b/vendor/github.com/thanos-io/thanos/pkg/compact/blocks_cleaner.go
@@ -37,7 +37,8 @@ func NewBlocksCleaner(logger log.Logger, bkt objstore.Bucket, ignoreDeletionMark
 	}
 }
 
-// DeleteMarkedBlocks uses ignoreDeletionMarkFilter to delete the blocks that are marked for deletion.
+// DeleteMarkedBlocks uses ignoreDeletionMarkFilter to gather the blocks that are marked for deletion and deletes those
+// if older than given deleteDelay.
 func (s *BlocksCleaner) DeleteMarkedBlocks(ctx context.Context) error {
 	level.Info(s.logger).Log("msg", "started cleaning of blocks marked for deletion")
 
diff --git a/vendor/github.com/thanos-io/thanos/pkg/compact/clean.go b/vendor/github.com/thanos-io/thanos/pkg/compact/clean.go
index 14fdfc4d7c..53cfb16908 100644
--- a/vendor/github.com/thanos-io/thanos/pkg/compact/clean.go
+++ b/vendor/github.com/thanos-io/thanos/pkg/compact/clean.go
@@ -21,12 +21,15 @@ const (
 	PartialUploadThresholdAge = 2 * 24 * time.Hour
 )
 
-func BestEffortCleanAbortedPartialUploads(ctx context.Context, logger log.Logger, fetcher block.MetadataFetcher, bkt objstore.Bucket, deleteAttempts prometheus.Counter, blocksMarkedForDeletion prometheus.Counter) {
+func BestEffortCleanAbortedPartialUploads(
+	ctx context.Context,
+	logger log.Logger,
+	partial map[ulid.ULID]error,
+	bkt objstore.Bucket,
+	deleteAttempts prometheus.Counter,
+	blocksMarkedForDeletion prometheus.Counter,
+) {
 	level.Info(logger).Log("msg", "started cleaning of aborted partial uploads")
-	_, partial, err := fetcher.Fetch(ctx)
-	if err != nil {
-		level.Warn(logger).Log("msg", "failed to fetch metadata for cleaning of aborted partial uploads; skipping", "err", err)
-	}
 
 	// Delete partial blocks that are older than partialUploadThresholdAge.
 	// TODO(bwplotka): This is can cause data loss if blocks are:
@@ -41,11 +44,11 @@ func BestEffortCleanAbortedPartialUploads(ctx context.Context, logger log.Logger
 		}
 
 		deleteAttempts.Inc()
-		if err := block.MarkForDeletion(ctx, logger, bkt, id); err != nil {
+		level.Info(logger).Log("msg", "found partially uploaded block; marking for deletion", "block", id)
+		if err := block.MarkForDeletion(ctx, logger, bkt, id, blocksMarkedForDeletion); err != nil {
 			level.Warn(logger).Log("msg", "failed to delete aborted partial upload; skipping", "block", id, "thresholdAge", PartialUploadThresholdAge, "err", err)
 			return
 		}
-		blocksMarkedForDeletion.Inc()
 		level.Info(logger).Log("msg", "deleted aborted partial upload", "block", id, "thresholdAge", PartialUploadThresholdAge)
 	}
 	level.Info(logger).Log("msg", "cleaning of aborted partial uploads done")
diff --git a/vendor/github.com/thanos-io/thanos/pkg/compact/compact.go b/vendor/github.com/thanos-io/thanos/pkg/compact/compact.go
index 6443ba5ae8..c0bb41b964 100644
--- a/vendor/github.com/thanos-io/thanos/pkg/compact/compact.go
+++ b/vendor/github.com/thanos-io/thanos/pkg/compact/compact.go
@@ -46,6 +46,7 @@ type Syncer struct {
 	fetcher                  block.MetadataFetcher
 	mtx                      sync.Mutex
 	blocks                   map[ulid.ULID]*metadata.Meta
+	partial                  map[ulid.ULID]error
 	blockSyncConcurrency     int
 	metrics                  *syncerMetrics
 	acceptMalformedIndex     bool
@@ -153,19 +154,36 @@ func UntilNextDownsampling(m *metadata.Meta) (time.Duration, error) {
 	}
 }
 
+// SyncMetas synchronises local state of block metas with what we have in the bucket.
 func (s *Syncer) SyncMetas(ctx context.Context) error {
 	s.mtx.Lock()
 	defer s.mtx.Unlock()
 
-	metas, _, err := s.fetcher.Fetch(ctx)
+	metas, partial, err := s.fetcher.Fetch(ctx)
 	if err != nil {
 		return retry(err)
 	}
 	s.blocks = metas
-
+	s.partial = partial
 	return nil
 }
 
+// Partial returns partial blocks since last sync.
+func (s *Syncer) Partial() map[ulid.ULID]error {
+	s.mtx.Lock()
+	defer s.mtx.Unlock()
+
+	return s.partial
+}
+
+// Metas returns loaded metadata blocks since last sync.
+func (s *Syncer) Metas() map[ulid.ULID]*metadata.Meta {
+	s.mtx.Lock()
+	defer s.mtx.Unlock()
+
+	return s.blocks
+}
+
 // GroupKey returns a unique identifier for the group the block belongs to. It considers
 // the downsampling resolution and the block's labels.
 func GroupKey(meta metadata.Thanos) string {
@@ -189,7 +207,7 @@ func (s *Syncer) Groups() (res []*Group, err error) {
 		if !ok {
 			lbls := labels.FromMap(m.Thanos.Labels)
 			g, err = newGroup(
-				log.With(s.logger, "compactionGroup", fmt.Sprintf("%d@%v", m.Thanos.Downsample.Resolution, lbls.String()), "compactionGroupKey", groupKey),
+				log.With(s.logger, "group", fmt.Sprintf("%d@%v", m.Thanos.Downsample.Resolution, lbls.String()), "groupKey", groupKey),
 				s.bkt,
 				lbls,
 				m.Thanos.Downsample.Resolution,
@@ -228,16 +246,18 @@ func (s *Syncer) GarbageCollect(ctx context.Context) error {
 
 	begin := time.Now()
 
-	duplicateIDs := s.duplicateBlocksFilter.DuplicateIDs()
+	// Ignore filter exists before deduplicate filter.
 	deletionMarkMap := s.ignoreDeletionMarkFilter.DeletionMarkBlocks()
+	duplicateIDs := s.duplicateBlocksFilter.DuplicateIDs()
 
 	// GarbageIDs contains the duplicateIDs, since these blocks can be replaced with other blocks.
 	// We also remove ids present in deletionMarkMap since these blocks are already marked for deletion.
 	garbageIDs := []ulid.ULID{}
 	for _, id := range duplicateIDs {
-		if _, exists := deletionMarkMap[id]; !exists {
-			garbageIDs = append(garbageIDs, id)
+		if _, exists := deletionMarkMap[id]; exists {
+			continue
 		}
+		garbageIDs = append(garbageIDs, id)
 	}
 
 	for _, id := range garbageIDs {
@@ -249,14 +269,12 @@ func (s *Syncer) GarbageCollect(ctx context.Context) error {
 		delCtx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
 
 		level.Info(s.logger).Log("msg", "marking outdated block for deletion", "block", id)
-
-		err := block.MarkForDeletion(delCtx, s.logger, s.bkt, id)
+		err := block.MarkForDeletion(delCtx, s.logger, s.bkt, id, s.metrics.blocksMarkedForDeletion)
 		cancel()
 		if err != nil {
 			s.metrics.garbageCollectionFailures.Inc()
-			return retry(errors.Wrapf(err, "delete block %s from bucket", id))
+			return retry(errors.Wrapf(err, "mark block %s for deletion", id))
 		}
-		s.metrics.blocksMarkedForDeletion.Inc()
 
 		// Immediately update our in-memory state so no further call to SyncMetas is needed
 		// after running garbage collection.
@@ -568,10 +586,9 @@ func RepairIssue347(ctx context.Context, logger log.Logger, bkt objstore.Bucket,
 	defer cancel()
 
 	// TODO(bplotka): Issue with this will introduce overlap that will halt compactor. Automate that (fix duplicate overlaps caused by this).
-	if err := block.MarkForDeletion(delCtx, logger, bkt, ie.id); err != nil {
+	if err := block.MarkForDeletion(delCtx, logger, bkt, ie.id, blocksMarkedForDeletion); err != nil {
 		return errors.Wrapf(err, "deleting old block %s failed. You need to delete this block manually", ie.id)
 	}
-	blocksMarkedForDeletion.Inc()
 	return nil
 }
 
@@ -613,6 +630,8 @@ func (cg *Group) compact(ctx context.Context, dir string, comp tsdb.Compactor) (
 		return false, ulid.ULID{}, nil
 	}
 
+	level.Info(cg.logger).Log("msg", "compaction available and planned; downloading blocks", "plan", fmt.Sprintf("%v", plan))
+
 	// Due to #183 we verify that none of the blocks in the plan have overlapping sources.
 	// This is one potential source of how we could end up with duplicated chunks.
 	uniqueSources := map[ulid.ULID]struct{}{}
@@ -670,8 +689,7 @@ func (cg *Group) compact(ctx context.Context, dir string, comp tsdb.Compactor) (
 				"block id %s, try running with --debug.accept-malformed-index", id)
 		}
 	}
-	level.Debug(cg.logger).Log("msg", "downloaded and verified blocks",
-		"blocks", fmt.Sprintf("%v", plan), "duration", time.Since(begin))
+	level.Info(cg.logger).Log("msg", "downloaded and verified blocks; compacting blocks", "plan", fmt.Sprintf("%v", plan), "duration", time.Since(begin))
 
 	begin = time.Now()
 
@@ -701,7 +719,7 @@ func (cg *Group) compact(ctx context.Context, dir string, comp tsdb.Compactor) (
 	if overlappingBlocks {
 		cg.verticalCompactions.Inc()
 	}
-	level.Info(cg.logger).Log("msg", "compacted blocks",
+	level.Info(cg.logger).Log("msg", "compacted blocks", "new", compID,
 		"blocks", fmt.Sprintf("%v", plan), "duration", time.Since(begin), "overlapping_blocks", overlappingBlocks)
 
 	bdir := filepath.Join(dir, compID.String())
@@ -772,10 +790,9 @@ func (cg *Group) deleteBlock(b string) error {
 	delCtx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
 	defer cancel()
 	level.Info(cg.logger).Log("msg", "marking compacted block for deletion", "old_block", id)
-	if err := block.MarkForDeletion(delCtx, cg.logger, cg.bkt, id); err != nil {
+	if err := block.MarkForDeletion(delCtx, cg.logger, cg.bkt, id, cg.blocksMarkedForDeletion); err != nil {
 		return errors.Wrapf(err, "delete block %s from bucket", id)
 	}
-	cg.blocksMarkedForDeletion.Inc()
 	return nil
 }
 
@@ -868,29 +885,26 @@ func (c *BucketCompactor) Compact(ctx context.Context) error {
 		}
 
 		level.Info(c.logger).Log("msg", "start sync of metas")
-
 		if err := c.sy.SyncMetas(ctx); err != nil {
 			return errors.Wrap(err, "sync")
 		}
 
 		level.Info(c.logger).Log("msg", "start of GC")
-
 		// Blocks that were compacted are garbage collected after each Compaction.
 		// However if compactor crashes we need to resolve those on startup.
 		if err := c.sy.GarbageCollect(ctx); err != nil {
 			return errors.Wrap(err, "garbage")
 		}
 
-		level.Info(c.logger).Log("msg", "start of compaction")
-
 		groups, err := c.sy.Groups()
 		if err != nil {
 			return errors.Wrap(err, "build compaction groups")
 		}
 
+		level.Info(c.logger).Log("msg", "start of compactions")
+
 		// Send all groups found during this pass to the compaction workers.
 		var groupErrs terrors.MultiError
-
 	groupLoop:
 		for _, g := range groups {
 			select {
diff --git a/vendor/github.com/thanos-io/thanos/pkg/compact/retention.go b/vendor/github.com/thanos-io/thanos/pkg/compact/retention.go
index eb07507d9d..8d1ba7d5fb 100644
--- a/vendor/github.com/thanos-io/thanos/pkg/compact/retention.go
+++ b/vendor/github.com/thanos-io/thanos/pkg/compact/retention.go
@@ -9,21 +9,25 @@ import (
 
 	"github.com/go-kit/kit/log"
 	"github.com/go-kit/kit/log/level"
+	"github.com/oklog/ulid"
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/thanos-io/thanos/pkg/block"
+	"github.com/thanos-io/thanos/pkg/block/metadata"
 	"github.com/thanos-io/thanos/pkg/objstore"
 )
 
 // ApplyRetentionPolicyByResolution removes blocks depending on the specified retentionByResolution based on blocks MaxTime.
 // A value of 0 disables the retention for its resolution.
-func ApplyRetentionPolicyByResolution(ctx context.Context, logger log.Logger, bkt objstore.Bucket, fetcher block.MetadataFetcher, retentionByResolution map[ResolutionLevel]time.Duration, blocksMarkedForDeletion prometheus.Counter) error {
+func ApplyRetentionPolicyByResolution(
+	ctx context.Context,
+	logger log.Logger,
+	bkt objstore.Bucket,
+	metas map[ulid.ULID]*metadata.Meta,
+	retentionByResolution map[ResolutionLevel]time.Duration,
+	blocksMarkedForDeletion prometheus.Counter,
+) error {
 	level.Info(logger).Log("msg", "start optional retention")
-	metas, _, err := fetcher.Fetch(ctx)
-	if err != nil {
-		return errors.Wrap(err, "fetch metas")
-	}
-
 	for id, m := range metas {
 		retentionDuration := retentionByResolution[ResolutionLevel(m.Thanos.Downsample.Resolution)]
 		if retentionDuration.Seconds() == 0 {
@@ -33,13 +37,11 @@ func ApplyRetentionPolicyByResolution(ctx context.Context, logger log.Logger, bk
 		maxTime := time.Unix(m.MaxTime/1000, 0)
 		if time.Now().After(maxTime.Add(retentionDuration)) {
 			level.Info(logger).Log("msg", "applying retention: marking block for deletion", "id", id, "maxTime", maxTime.String())
-			if err := block.MarkForDeletion(ctx, logger, bkt, id); err != nil {
+			if err := block.MarkForDeletion(ctx, logger, bkt, id, blocksMarkedForDeletion); err != nil {
 				return errors.Wrap(err, "delete block")
 			}
-			blocksMarkedForDeletion.Inc()
 		}
 	}
-
 	level.Info(logger).Log("msg", "optional retention apply done")
 	return nil
 }
diff --git a/vendor/github.com/thanos-io/thanos/pkg/discovery/dns/provider.go b/vendor/github.com/thanos-io/thanos/pkg/discovery/dns/provider.go
index 5df9e97320..0ef944574d 100644
--- a/vendor/github.com/thanos-io/thanos/pkg/discovery/dns/provider.go
+++ b/vendor/github.com/thanos-io/thanos/pkg/discovery/dns/provider.go
@@ -87,6 +87,23 @@ func (p *Provider) Clone() *Provider {
 	}
 }
 
+// IsDynamicNode returns if the specified StoreAPI addr uses
+// any kind of SD mechanism.
+func IsDynamicNode(addr string) bool {
+	qtype, _ := GetQTypeName(addr)
+	return qtype != ""
+}
+
+// GetQTypeName splits the provided addr into two parts: the QType (if any)
+// and the name.
+func GetQTypeName(addr string) (qtype string, name string) {
+	qtypeAndName := strings.SplitN(addr, "+", 2)
+	if len(qtypeAndName) != 2 {
+		return "", addr
+	}
+	return qtypeAndName[0], qtypeAndName[1]
+}
+
 // Resolve stores a list of provided addresses or their DNS records if requested.
 // Addresses prefixed with `dns+` or `dnssrv+` will be resolved through respective DNS lookup (A/AAAA or SRV).
 // defaultPort is used for non-SRV records when a port is not supplied.
@@ -100,14 +117,12 @@ func (p *Provider) Resolve(ctx context.Context, addrs []string) {
 	resolvedAddrs := map[string][]string{}
 	for _, addr := range addrs {
 		var resolved []string
-		qtypeAndName := strings.SplitN(addr, "+", 2)
-		if len(qtypeAndName) != 2 {
-			// No lookup specified. Add to results and continue to the next address.
-			resolvedAddrs[addr] = []string{addr}
-			p.resolverAddrs.WithLabelValues(addr).Set(1.0)
+		qtype, name := GetQTypeName(addr)
+		if qtype == "" {
+			resolvedAddrs[name] = []string{name}
+			p.resolverAddrs.WithLabelValues(name).Set(1.0)
 			continue
 		}
-		qtype, name := qtypeAndName[0], qtypeAndName[1]
 
 		resolved, err := p.resolver.Resolve(ctx, name, QType(qtype))
 		p.resolverLookupsCount.Inc()
diff --git a/vendor/github.com/thanos-io/thanos/pkg/objstore/objstore.go b/vendor/github.com/thanos-io/thanos/pkg/objstore/objstore.go
index ff63a9018a..c1bb44d079 100644
--- a/vendor/github.com/thanos-io/thanos/pkg/objstore/objstore.go
+++ b/vendor/github.com/thanos-io/thanos/pkg/objstore/objstore.go
@@ -4,14 +4,12 @@
 package objstore
 
 import (
+	"bytes"
 	"context"
-	"fmt"
 	"io"
-	"math/rand"
 	"os"
 	"path/filepath"
 	"strings"
-	"testing"
 	"time"
 
 	"github.com/go-kit/kit/log"
@@ -40,6 +38,20 @@ type Bucket interface {
 	Name() string
 }
 
+// InstrumentedBucket is a Bucket with optional instrumentation control on reader.
+type InstrumentedBucket interface {
+	Bucket
+
+	// WithExpectedErrs allows to specify a filter that marks certain errors as expected, so it will not increment
+	// thanos_objstore_bucket_operation_failures_total metric.
+	WithExpectedErrs(IsOpFailureExpectedFunc) Bucket
+
+	// ReaderWithExpectedErrs allows to specify a filter that marks certain errors as expected, so it will not increment
+	// thanos_objstore_bucket_operation_failures_total metric.
+	// TODO(bwplotka): Remove this when moved to Go 1.14 and replace with InstrumentedBucketReader.
+	ReaderWithExpectedErrs(IsOpFailureExpectedFunc) BucketReader
+}
+
 // BucketReader provides read access to an object storage bucket.
 type BucketReader interface {
 	// Iter calls f for each entry in the given directory (not recursive.). The argument to f is the full
@@ -53,7 +65,6 @@ type BucketReader interface {
 	GetRange(ctx context.Context, name string, off, length int64) (io.ReadCloser, error)
 
 	// Exists checks if the given object exists in the bucket.
-	// TODO(bplotka): Consider removing Exists in favor of helper that do Get & IsObjNotFoundErr (less code to maintain).
 	Exists(ctx context.Context, name string) (bool, error)
 
 	// IsObjNotFoundErr returns true if error means that object is not found. Relevant to Get operations.
@@ -63,6 +74,33 @@ type BucketReader interface {
 	ObjectSize(ctx context.Context, name string) (uint64, error)
 }
 
+// InstrumentedBucket is a BucketReader with optional instrumentation control.
+type InstrumentedBucketReader interface {
+	BucketReader
+
+	// ReaderWithExpectedErrs allows to specify a filter that marks certain errors as expected, so it will not increment
+	// thanos_objstore_bucket_operation_failures_total metric.
+	ReaderWithExpectedErrs(IsOpFailureExpectedFunc) BucketReader
+}
+
+// TryToGetSize tries to get upfront size from reader.
+// TODO(https://github.com/thanos-io/thanos/issues/678): Remove guessing length when minio provider will support multipart upload without this.
+func TryToGetSize(r io.Reader) (int64, error) {
+	switch f := r.(type) {
+	case *os.File:
+		fileInfo, err := f.Stat()
+		if err != nil {
+			return 0, errors.Wrap(err, "os.File.Stat()")
+		}
+		return fileInfo.Size(), nil
+	case *bytes.Buffer:
+		return int64(f.Len()), nil
+	case *strings.Reader:
+		return f.Size(), nil
+	}
+	return 0, errors.New("unsupported type of io.Reader")
+}
+
 // UploadDir uploads all files in srcdir to the bucket with into a top-level directory
 // named dstdir. It is a caller responsibility to clean partial upload in case of failure.
 func UploadDir(ctx context.Context, logger log.Logger, bkt Bucket, srcdir, dstdir string) error {
@@ -182,27 +220,32 @@ const (
 	deleteOp   = "delete"
 )
 
+// IsOpFailureExpectedFunc allows to mark certain errors as expected, so they will not increment thanos_objstore_bucket_operation_failures_total metric.
+type IsOpFailureExpectedFunc func(error) bool
+
+var _ InstrumentedBucket = &metricBucket{}
+
 // BucketWithMetrics takes a bucket and registers metrics with the given registry for
 // operations run against the bucket.
-func BucketWithMetrics(name string, b Bucket, reg prometheus.Registerer) Bucket {
+func BucketWithMetrics(name string, b Bucket, reg prometheus.Registerer) *metricBucket {
 	bkt := &metricBucket{
-		bkt: b,
-
+		bkt:                 b,
+		isOpFailureExpected: func(err error) bool { return false },
 		ops: promauto.With(reg).NewCounterVec(prometheus.CounterOpts{
 			Name:        "thanos_objstore_bucket_operations_total",
-			Help:        "Total number of operations against a bucket.",
+			Help:        "Total number of all attempted operations against a bucket.",
 			ConstLabels: prometheus.Labels{"bucket": name},
 		}, []string{"operation"}),
 
 		opsFailures: promauto.With(reg).NewCounterVec(prometheus.CounterOpts{
 			Name:        "thanos_objstore_bucket_operation_failures_total",
-			Help:        "Total number of operations against a bucket that failed.",
+			Help:        "Total number of operations against a bucket that failed, but were not expected to fail in certain way from caller perspective. Those errors have to be investigated.",
 			ConstLabels: prometheus.Labels{"bucket": name},
 		}, []string{"operation"}),
 
 		opsDuration: promauto.With(reg).NewHistogramVec(prometheus.HistogramOpts{
 			Name:        "thanos_objstore_bucket_operation_duration_seconds",
-			Help:        "Duration of operations against the bucket",
+			Help:        "Duration of successful operations against the bucket",
 			ConstLabels: prometheus.Labels{"bucket": name},
 			Buckets:     []float64{0.001, 0.01, 0.1, 0.3, 0.6, 1, 3, 6, 9, 20, 30, 60, 90, 120},
 		}, []string{"operation"}),
@@ -211,7 +254,15 @@ func BucketWithMetrics(name string, b Bucket, reg prometheus.Registerer) Bucket
 			Help: "Second timestamp of the last successful upload to the bucket.",
 		}, []string{"bucket"}),
 	}
-	for _, op := range []string{iterOp, sizeOp, getOp, getRangeOp, existsOp, uploadOp, deleteOp} {
+	for _, op := range []string{
+		iterOp,
+		sizeOp,
+		getOp,
+		getRangeOp,
+		existsOp,
+		uploadOp,
+		deleteOp,
+	} {
 		bkt.ops.WithLabelValues(op)
 		bkt.opsFailures.WithLabelValues(op)
 		bkt.opsDuration.WithLabelValues(op)
@@ -223,107 +274,142 @@ func BucketWithMetrics(name string, b Bucket, reg prometheus.Registerer) Bucket
 type metricBucket struct {
 	bkt Bucket
 
-	ops                      *prometheus.CounterVec
-	opsFailures              *prometheus.CounterVec
+	ops                 *prometheus.CounterVec
+	opsFailures         *prometheus.CounterVec
+	isOpFailureExpected IsOpFailureExpectedFunc
+
 	opsDuration              *prometheus.HistogramVec
 	lastSuccessfulUploadTime *prometheus.GaugeVec
 }
 
+func (b *metricBucket) WithExpectedErrs(fn IsOpFailureExpectedFunc) Bucket {
+	return &metricBucket{
+		bkt:                      b.bkt,
+		ops:                      b.ops,
+		opsFailures:              b.opsFailures,
+		isOpFailureExpected:      fn,
+		opsDuration:              b.opsDuration,
+		lastSuccessfulUploadTime: b.lastSuccessfulUploadTime,
+	}
+}
+
+func (b *metricBucket) ReaderWithExpectedErrs(fn IsOpFailureExpectedFunc) BucketReader {
+	return b.WithExpectedErrs(fn)
+}
+
 func (b *metricBucket) Iter(ctx context.Context, dir string, f func(name string) error) error {
+	const op = iterOp
+	b.ops.WithLabelValues(op).Inc()
+
 	err := b.bkt.Iter(ctx, dir, f)
-	if err != nil {
-		b.opsFailures.WithLabelValues(iterOp).Inc()
+	if err != nil && !b.isOpFailureExpected(err) {
+		b.opsFailures.WithLabelValues(op).Inc()
 	}
-	b.ops.WithLabelValues(iterOp).Inc()
-
 	return err
 }
 
-// ObjectSize returns the size of the specified object.
 func (b *metricBucket) ObjectSize(ctx context.Context, name string) (uint64, error) {
-	b.ops.WithLabelValues(sizeOp).Inc()
-	start := time.Now()
+	const op = sizeOp
+	b.ops.WithLabelValues(op).Inc()
 
+	start := time.Now()
 	rc, err := b.bkt.ObjectSize(ctx, name)
 	if err != nil {
-		b.opsFailures.WithLabelValues(sizeOp).Inc()
+		if !b.isOpFailureExpected(err) {
+			b.opsFailures.WithLabelValues(op).Inc()
+		}
 		return 0, err
 	}
-	b.opsDuration.WithLabelValues(sizeOp).Observe(time.Since(start).Seconds())
+	b.opsDuration.WithLabelValues(op).Observe(time.Since(start).Seconds())
 	return rc, nil
 }
 
 func (b *metricBucket) Get(ctx context.Context, name string) (io.ReadCloser, error) {
-	b.ops.WithLabelValues(getOp).Inc()
+	const op = getOp
+	b.ops.WithLabelValues(op).Inc()
 
 	rc, err := b.bkt.Get(ctx, name)
 	if err != nil {
-		b.opsFailures.WithLabelValues(getOp).Inc()
+		if !b.isOpFailureExpected(err) {
+			b.opsFailures.WithLabelValues(op).Inc()
+		}
 		return nil, err
 	}
 	return newTimingReadCloser(
 		rc,
-		getOp,
+		op,
 		b.opsDuration,
 		b.opsFailures,
+		b.isOpFailureExpected,
 	), nil
 }
 
 func (b *metricBucket) GetRange(ctx context.Context, name string, off, length int64) (io.ReadCloser, error) {
-	b.ops.WithLabelValues(getRangeOp).Inc()
+	const op = getRangeOp
+	b.ops.WithLabelValues(op).Inc()
 
 	rc, err := b.bkt.GetRange(ctx, name, off, length)
 	if err != nil {
-		b.opsFailures.WithLabelValues(getRangeOp).Inc()
+		if !b.isOpFailureExpected(err) {
+			b.opsFailures.WithLabelValues(op).Inc()
+		}
 		return nil, err
 	}
 	return newTimingReadCloser(
 		rc,
-		getRangeOp,
+		op,
 		b.opsDuration,
 		b.opsFailures,
+		b.isOpFailureExpected,
 	), nil
 }
 
 func (b *metricBucket) Exists(ctx context.Context, name string) (bool, error) {
-	start := time.Now()
+	const op = existsOp
+	b.ops.WithLabelValues(op).Inc()
 
+	start := time.Now()
 	ok, err := b.bkt.Exists(ctx, name)
 	if err != nil {
-		b.opsFailures.WithLabelValues(existsOp).Inc()
+		if !b.isOpFailureExpected(err) {
+			b.opsFailures.WithLabelValues(op).Inc()
+		}
+		return false, err
 	}
-	b.ops.WithLabelValues(existsOp).Inc()
-	b.opsDuration.WithLabelValues(existsOp).Observe(time.Since(start).Seconds())
-
-	return ok, err
+	b.opsDuration.WithLabelValues(op).Observe(time.Since(start).Seconds())
+	return ok, nil
 }
 
 func (b *metricBucket) Upload(ctx context.Context, name string, r io.Reader) error {
-	start := time.Now()
+	const op = uploadOp
+	b.ops.WithLabelValues(op).Inc()
 
-	err := b.bkt.Upload(ctx, name, r)
-	if err != nil {
-		b.opsFailures.WithLabelValues(uploadOp).Inc()
-	} else {
-		b.lastSuccessfulUploadTime.WithLabelValues(b.bkt.Name()).SetToCurrentTime()
+	start := time.Now()
+	if err := b.bkt.Upload(ctx, name, r); err != nil {
+		if !b.isOpFailureExpected(err) {
+			b.opsFailures.WithLabelValues(op).Inc()
+		}
+		return err
 	}
-	b.ops.WithLabelValues(uploadOp).Inc()
-	b.opsDuration.WithLabelValues(uploadOp).Observe(time.Since(start).Seconds())
-
-	return err
+	b.lastSuccessfulUploadTime.WithLabelValues(b.bkt.Name()).SetToCurrentTime()
+	b.opsDuration.WithLabelValues(op).Observe(time.Since(start).Seconds())
+	return nil
 }
 
 func (b *metricBucket) Delete(ctx context.Context, name string) error {
-	start := time.Now()
+	const op = deleteOp
+	b.ops.WithLabelValues(op).Inc()
 
-	err := b.bkt.Delete(ctx, name)
-	if err != nil {
-		b.opsFailures.WithLabelValues(deleteOp).Inc()
+	start := time.Now()
+	if err := b.bkt.Delete(ctx, name); err != nil {
+		if !b.isOpFailureExpected(err) {
+			b.opsFailures.WithLabelValues(op).Inc()
+		}
+		return err
 	}
-	b.ops.WithLabelValues(deleteOp).Inc()
-	b.opsDuration.WithLabelValues(deleteOp).Observe(time.Since(start).Seconds())
+	b.opsDuration.WithLabelValues(op).Observe(time.Since(start).Seconds())
 
-	return err
+	return nil
 }
 
 func (b *metricBucket) IsObjNotFoundErr(err error) bool {
@@ -341,53 +427,49 @@ func (b *metricBucket) Name() string {
 type timingReadCloser struct {
 	io.ReadCloser
 
-	ok       bool
-	start    time.Time
-	op       string
-	duration *prometheus.HistogramVec
-	failed   *prometheus.CounterVec
+	alreadyGotErr bool
+
+	start             time.Time
+	op                string
+	duration          *prometheus.HistogramVec
+	failed            *prometheus.CounterVec
+	isFailureExpected IsOpFailureExpectedFunc
 }
 
-func newTimingReadCloser(rc io.ReadCloser, op string, dur *prometheus.HistogramVec, failed *prometheus.CounterVec) *timingReadCloser {
+func newTimingReadCloser(rc io.ReadCloser, op string, dur *prometheus.HistogramVec, failed *prometheus.CounterVec, isFailureExpected IsOpFailureExpectedFunc) *timingReadCloser {
 	// Initialize the metrics with 0.
 	dur.WithLabelValues(op)
 	failed.WithLabelValues(op)
 	return &timingReadCloser{
-		ReadCloser: rc,
-		ok:         true,
-		start:      time.Now(),
-		op:         op,
-		duration:   dur,
-		failed:     failed,
+		ReadCloser:        rc,
+		start:             time.Now(),
+		op:                op,
+		duration:          dur,
+		failed:            failed,
+		isFailureExpected: isFailureExpected,
 	}
 }
 
 func (rc *timingReadCloser) Close() error {
 	err := rc.ReadCloser.Close()
-	rc.duration.WithLabelValues(rc.op).Observe(time.Since(rc.start).Seconds())
-	if rc.ok && err != nil {
+	if !rc.alreadyGotErr && err != nil {
 		rc.failed.WithLabelValues(rc.op).Inc()
-		rc.ok = false
+	}
+	if !rc.alreadyGotErr && err == nil {
+		rc.duration.WithLabelValues(rc.op).Observe(time.Since(rc.start).Seconds())
+		rc.alreadyGotErr = true
 	}
 	return err
 }
 
 func (rc *timingReadCloser) Read(b []byte) (n int, err error) {
 	n, err = rc.ReadCloser.Read(b)
-	if rc.ok && err != nil && err != io.EOF {
-		rc.failed.WithLabelValues(rc.op).Inc()
-		rc.ok = false
+	// Report metric just once.
+	if !rc.alreadyGotErr && err != nil && err != io.EOF {
+		if !rc.isFailureExpected(err) {
+			rc.failed.WithLabelValues(rc.op).Inc()
+		}
+		rc.alreadyGotErr = true
 	}
 	return n, err
 }
-
-func CreateTemporaryTestBucketName(t testing.TB) string {
-	src := rand.NewSource(time.Now().UnixNano())
-
-	// Bucket name need to conform: https://docs.aws.amazon.com/awscloudtrail/latest/userguide/cloudtrail-s3-bucket-naming-requirements.html.
-	name := strings.Replace(strings.Replace(fmt.Sprintf("test_%x_%s", src.Int63(), strings.ToLower(t.Name())), "_", "-", -1), "/", "-", -1)
-	if len(name) >= 63 {
-		name = name[:63]
-	}
-	return name
-}
diff --git a/vendor/github.com/thanos-io/thanos/pkg/objstore/s3/s3.go b/vendor/github.com/thanos-io/thanos/pkg/objstore/s3/s3.go
index d924eeb01f..59098125d0 100644
--- a/vendor/github.com/thanos-io/thanos/pkg/objstore/s3/s3.go
+++ b/vendor/github.com/thanos-io/thanos/pkg/objstore/s3/s3.go
@@ -287,7 +287,7 @@ func (b *Bucket) GetRange(ctx context.Context, name string, off, length int64) (
 }
 
 // Exists checks if the given object exists.
-func (b *Bucket) Exists(ctx context.Context, name string) (bool, error) {
+func (b *Bucket) Exists(_ context.Context, name string) (bool, error) {
 	_, err := b.client.StatObject(b.name, name, minio.StatObjectOptions{})
 	if err != nil {
 		if b.IsObjNotFoundErr(err) {
@@ -299,24 +299,14 @@ func (b *Bucket) Exists(ctx context.Context, name string) (bool, error) {
 	return true, nil
 }
 
-func (b *Bucket) guessFileSize(name string, r io.Reader) int64 {
-	if f, ok := r.(*os.File); ok {
-		fileInfo, err := f.Stat()
-		if err == nil {
-			return fileInfo.Size()
-		}
-		level.Warn(b.logger).Log("msg", "could not stat file for multipart upload", "name", name, "err", err)
-		return -1
-	}
-
-	level.Warn(b.logger).Log("msg", "could not guess file size for multipart upload", "name", name)
-	return -1
-}
-
 // Upload the contents of the reader as an object into the bucket.
 func (b *Bucket) Upload(ctx context.Context, name string, r io.Reader) error {
 	// TODO(https://github.com/thanos-io/thanos/issues/678): Remove guessing length when minio provider will support multipart upload without this.
-	size := b.guessFileSize(name, r)
+	size, err := objstore.TryToGetSize(r)
+	if err != nil {
+		level.Warn(b.logger).Log("msg", "could not guess file size for multipart upload; upload might be not optimized", "name", name, "err", err)
+		size = -1
+	}
 
 	// partSize cannot be larger than object size.
 	partSize := b.partSize
@@ -342,7 +332,7 @@ func (b *Bucket) Upload(ctx context.Context, name string, r io.Reader) error {
 }
 
 // ObjectSize returns the size of the specified object.
-func (b *Bucket) ObjectSize(ctx context.Context, name string) (uint64, error) {
+func (b *Bucket) ObjectSize(_ context.Context, name string) (uint64, error) {
 	objInfo, err := b.client.StatObject(b.name, name, minio.StatObjectOptions{})
 	if err != nil {
 		return 0, err
@@ -351,7 +341,7 @@ func (b *Bucket) ObjectSize(ctx context.Context, name string) (uint64, error) {
 }
 
 // Delete removes the object with the given name.
-func (b *Bucket) Delete(ctx context.Context, name string) error {
+func (b *Bucket) Delete(_ context.Context, name string) error {
 	return b.client.RemoveObject(b.name, name)
 }
 
diff --git a/vendor/github.com/thanos-io/thanos/pkg/store/bucket.go b/vendor/github.com/thanos-io/thanos/pkg/store/bucket.go
index 81dcb4ff1e..e0a1084312 100644
--- a/vendor/github.com/thanos-io/thanos/pkg/store/bucket.go
+++ b/vendor/github.com/thanos-io/thanos/pkg/store/bucket.go
@@ -59,10 +59,8 @@ const (
 	// because you barely get any improvements in compression when the number of samples is beyond this.
 	// Take a look at Figure 6 in this whitepaper http://www.vldb.org/pvldb/vol8/p1816-teller.pdf.
 	maxSamplesPerChunk = 120
-
-	maxChunkSize = 16000
-
-	maxSeriesSize = 64 * 1024
+	maxChunkSize       = 16000
+	maxSeriesSize      = 64 * 1024
 
 	// CompatibilityTypeLabelName is an artificial label that Store Gateway can optionally advertise. This is required for compatibility
 	// with pre v0.8.0 Querier. Previous Queriers was strict about duplicated external labels of all StoreAPIs that had any labels.
@@ -76,6 +74,11 @@ const (
 	// TODO(bwplotka): Remove it at some point.
 	CompatibilityTypeLabelName = "@thanos_compatibility_store_type"
 
+	// DefaultPostingOffsetInMemorySampling represents default value for --store.index-header-posting-offsets-in-mem-sampling.
+	// 32 value is chosen as it's a good balance for common setups. Sampling that is not too large (too many CPU cycles) and
+	// not too small (too much memory).
+	DefaultPostingOffsetInMemorySampling = 32
+
 	partitionerMaxGapSize = 512 * 1024
 )
 
@@ -221,7 +224,7 @@ type FilterConfig struct {
 type BucketStore struct {
 	logger     log.Logger
 	metrics    *bucketStoreMetrics
-	bkt        objstore.BucketReader
+	bkt        objstore.InstrumentedBucketReader
 	fetcher    block.MetadataFetcher
 	dir        string
 	indexCache storecache.IndexCache
@@ -252,7 +255,8 @@ type BucketStore struct {
 	// Reencode postings using diff+varint+snappy when storing to cache.
 	// This makes them smaller, but takes extra CPU and memory.
 	// When used with in-memory cache, memory usage should decrease overall, thanks to postings being smaller.
-	enablePostingsCompression bool
+	enablePostingsCompression   bool
+	postingOffsetsInMemSampling int
 }
 
 // NewBucketStore creates a new bucket backed store that implements the store API against
@@ -260,7 +264,7 @@ type BucketStore struct {
 func NewBucketStore(
 	logger log.Logger,
 	reg prometheus.Registerer,
-	bucket objstore.BucketReader,
+	bkt objstore.InstrumentedBucketReader,
 	fetcher block.MetadataFetcher,
 	dir string,
 	indexCache storecache.IndexCache,
@@ -273,6 +277,7 @@ func NewBucketStore(
 	enableCompatibilityLabel bool,
 	enableIndexHeader bool,
 	enablePostingsCompression bool,
+	postingOffsetsInMemSampling int,
 ) (*BucketStore, error) {
 	if logger == nil {
 		logger = log.NewNopLogger()
@@ -290,7 +295,7 @@ func NewBucketStore(
 	metrics := newBucketStoreMetrics(reg)
 	s := &BucketStore{
 		logger:               logger,
-		bkt:                  bucket,
+		bkt:                  bkt,
 		fetcher:              fetcher,
 		dir:                  dir,
 		indexCache:           indexCache,
@@ -304,11 +309,12 @@ func NewBucketStore(
 			maxConcurrent,
 			extprom.WrapRegistererWithPrefix("thanos_bucket_store_series_", reg),
 		),
-		samplesLimiter:            NewLimiter(maxSampleCount, metrics.queriesDropped),
-		partitioner:               gapBasedPartitioner{maxGapSize: partitionerMaxGapSize},
-		enableCompatibilityLabel:  enableCompatibilityLabel,
-		enableIndexHeader:         enableIndexHeader,
-		enablePostingsCompression: enablePostingsCompression,
+		samplesLimiter:              NewLimiter(maxSampleCount, metrics.queriesDropped),
+		partitioner:                 gapBasedPartitioner{maxGapSize: partitionerMaxGapSize},
+		enableCompatibilityLabel:    enableCompatibilityLabel,
+		enableIndexHeader:           enableIndexHeader,
+		enablePostingsCompression:   enablePostingsCompression,
+		postingOffsetsInMemSampling: postingOffsetsInMemSampling,
 	}
 	s.metrics = metrics
 
@@ -463,7 +469,7 @@ func (s *BucketStore) addBlock(ctx context.Context, meta *metadata.Meta) (err er
 
 	var indexHeaderReader indexheader.Reader
 	if s.enableIndexHeader {
-		indexHeaderReader, err = indexheader.NewBinaryReader(ctx, s.logger, s.bkt, s.dir, meta.ULID)
+		indexHeaderReader, err = indexheader.NewBinaryReader(ctx, s.logger, s.bkt, s.dir, meta.ULID, s.postingOffsetsInMemSampling)
 		if err != nil {
 			return errors.Wrap(err, "create index header reader")
 		}
diff --git a/vendor/github.com/thanos-io/thanos/pkg/store/postings_codec.go b/vendor/github.com/thanos-io/thanos/pkg/store/postings_codec.go
index 246d8bab81..7b1aaff477 100644
--- a/vendor/github.com/thanos-io/thanos/pkg/store/postings_codec.go
+++ b/vendor/github.com/thanos-io/thanos/pkg/store/postings_codec.go
@@ -80,7 +80,7 @@ func diffVarintSnappyDecode(input []byte) (index.Postings, error) {
 
 	raw, err := snappy.Decode(nil, input[len(codecHeaderSnappy):])
 	if err != nil {
-		return nil, errors.Errorf("snappy decode: %w", err)
+		return nil, errors.Wrap(err, "snappy decode")
 	}
 
 	return newDiffVarintPostings(raw), nil
diff --git a/vendor/modules.txt b/vendor/modules.txt
index c23bc172ac..67bbd03c81 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -196,6 +196,7 @@ github.com/golang-migrate/migrate/v4/source
 github.com/golang-migrate/migrate/v4/source/file
 # github.com/golang/groupcache v0.0.0-20191027212112-611e8accdfc9
 github.com/golang/groupcache/lru
+github.com/golang/groupcache/singleflight
 # github.com/golang/protobuf v1.3.2
 github.com/golang/protobuf/descriptor
 github.com/golang/protobuf/jsonpb
@@ -515,7 +516,7 @@ github.com/stretchr/objx
 github.com/stretchr/testify/assert
 github.com/stretchr/testify/mock
 github.com/stretchr/testify/require
-# github.com/thanos-io/thanos v0.8.1-0.20200326105947-214ff4480e93
+# github.com/thanos-io/thanos v0.12.0
 github.com/thanos-io/thanos/pkg/block
 github.com/thanos-io/thanos/pkg/block/indexheader
 github.com/thanos-io/thanos/pkg/block/metadata


Test patch:
diff --git a/pkg/compactor/compactor_test.go b/pkg/compactor/compactor_test.go
index 72ac973679..fac6405a6d 100644
--- a/pkg/compactor/compactor_test.go
+++ b/pkg/compactor/compactor_test.go
@@ -333,7 +333,7 @@ func TestCompactor_ShouldIterateOverUsersAndRunCompaction(t *testing.T) {
 		`level=info msg="starting compaction of user blocks" user=user-1`,
 		`level=info org_id=user-1 msg="start sync of metas"`,
 		`level=info org_id=user-1 msg="start of GC"`,
-		`level=info org_id=user-1 msg="start of compaction"`,
+		`level=info org_id=user-1 msg="start of compactions"`,
 		`level=info org_id=user-1 msg="compaction iterations done"`,
 		`level=info org_id=user-1 msg="started cleaning of blocks marked for deletion"`,
 		`level=info org_id=user-1 msg="cleaning of blocks marked for deletion done"`,
@@ -341,7 +341,7 @@ func TestCompactor_ShouldIterateOverUsersAndRunCompaction(t *testing.T) {
 		`level=info msg="starting compaction of user blocks" user=user-2`,
 		`level=info org_id=user-2 msg="start sync of metas"`,
 		`level=info org_id=user-2 msg="start of GC"`,
-		`level=info org_id=user-2 msg="start of compaction"`,
+		`level=info org_id=user-2 msg="start of compactions"`,
 		`level=info org_id=user-2 msg="compaction iterations done"`,
 		`level=info org_id=user-2 msg="started cleaning of blocks marked for deletion"`,
 		`level=info org_id=user-2 msg="cleaning of blocks marked for deletion done"`,
@@ -428,7 +428,7 @@ func TestCompactor_ShouldNotCompactBlocksMarkedForDeletion(t *testing.T) {
 		`level=info msg="starting compaction of user blocks" user=user-1`,
 		`level=info org_id=user-1 msg="start sync of metas"`,
 		`level=info org_id=user-1 msg="start of GC"`,
-		`level=info org_id=user-1 msg="start of compaction"`,
+		`level=info org_id=user-1 msg="start of compactions"`,
 		`level=info org_id=user-1 msg="compaction iterations done"`,
 		`level=info org_id=user-1 msg="started cleaning of blocks marked for deletion"`,
 		`level=debug org_id=user-1 msg="deleted file" file=01DTW0ZCPDDNV4BV83Q2SV4QAZ/meta.json bucket=mock`,
@@ -518,7 +518,7 @@ func TestCompactor_ShouldCompactAllUsersOnShardingEnabledButOnlyOneInstanceRunni
 		`level=info msg="starting compaction of user blocks" user=user-1`,
 		`level=info org_id=user-1 msg="start sync of metas"`,
 		`level=info org_id=user-1 msg="start of GC"`,
-		`level=info org_id=user-1 msg="start of compaction"`,
+		`level=info org_id=user-1 msg="start of compactions"`,
 		`level=info org_id=user-1 msg="compaction iterations done"`,
 		`level=info org_id=user-1 msg="started cleaning of blocks marked for deletion"`,
 		`level=info org_id=user-1 msg="cleaning of blocks marked for deletion done"`,
@@ -526,7 +526,7 @@ func TestCompactor_ShouldCompactAllUsersOnShardingEnabledButOnlyOneInstanceRunni
 		`level=info msg="starting compaction of user blocks" user=user-2`,
 		`level=info org_id=user-2 msg="start sync of metas"`,
 		`level=info org_id=user-2 msg="start of GC"`,
-		`level=info org_id=user-2 msg="start of compaction"`,
+		`level=info org_id=user-2 msg="start of compactions"`,
 		`level=info org_id=user-2 msg="compaction iterations done"`,
 		`level=info org_id=user-2 msg="started cleaning of blocks marked for deletion"`,
 		`level=info org_id=user-2 msg="cleaning of blocks marked for deletion done"`,
@@ -654,7 +654,7 @@ func removeMetaFetcherLogs(input []string) []string {
 	out := make([]string, 0, len(input))
 
 	for i := 0; i < len(input); i++ {
-		if !strings.Contains(input[i], "block.MetaFetcher") {
+		if !strings.Contains(input[i], "block.MetaFetcher") && !strings.Contains(input[i], "block.BaseFetcher") {
 			out = append(out, input[i])
 		}
 	}
diff --git a/vendor/github.com/thanos-io/thanos/pkg/objstore/testing.go b/vendor/github.com/thanos-io/thanos/pkg/objstore/testing.go
index 9e9a864b91..a33475db09 100644
--- a/vendor/github.com/thanos-io/thanos/pkg/objstore/testing.go
+++ b/vendor/github.com/thanos-io/thanos/pkg/objstore/testing.go
@@ -5,11 +5,29 @@ package objstore
 
 import (
 	"context"
+	"fmt"
+	"io/ioutil"
+	"math/rand"
+	"sort"
 	"strings"
 	"sync"
 	"testing"
+	"time"
+
+	"github.com/thanos-io/thanos/pkg/testutil"
 )
 
+func CreateTemporaryTestBucketName(t testing.TB) string {
+	src := rand.NewSource(time.Now().UnixNano())
+
+	// Bucket name need to conform: https://docs.aws.amazon.com/awscloudtrail/latest/userguide/cloudtrail-s3-bucket-naming-requirements.html.
+	name := strings.Replace(strings.Replace(fmt.Sprintf("test_%x_%s", src.Int63(), strings.ToLower(t.Name())), "_", "-", -1), "/", "-", -1)
+	if len(name) >= 63 {
+		name = name[:63]
+	}
+	return name
+}
+
 // EmptyBucket deletes all objects from bucket. This operation is required to properly delete bucket as a whole.
 // It is used for testing only.
 // TODO(bplotka): Add retries.
@@ -44,3 +62,158 @@ func EmptyBucket(t testing.TB, ctx context.Context, bkt Bucket) {
 	}
 	wg.Wait()
 }
+
+func WithNoopInstr(bkt Bucket) InstrumentedBucket {
+	return noopInstrumentedBucket{Bucket: bkt}
+}
+
+type noopInstrumentedBucket struct {
+	Bucket
+}
+
+func (b noopInstrumentedBucket) WithExpectedErrs(IsOpFailureExpectedFunc) Bucket {
+	return b
+}
+
+func (b noopInstrumentedBucket) ReaderWithExpectedErrs(IsOpFailureExpectedFunc) BucketReader {
+	return b
+}
+
+func AcceptanceTest(t *testing.T, bkt Bucket) {
+	ctx := context.Background()
+
+	_, err := bkt.Get(ctx, "")
+	testutil.NotOk(t, err)
+	testutil.Assert(t, !bkt.IsObjNotFoundErr(err), "expected user error got not found %s", err)
+
+	_, err = bkt.Get(ctx, "id1/obj_1.some")
+	testutil.NotOk(t, err)
+	testutil.Assert(t, bkt.IsObjNotFoundErr(err), "expected not found error got %s", err)
+
+	ok, err := bkt.Exists(ctx, "id1/obj_1.some")
+	testutil.Ok(t, err)
+	testutil.Assert(t, !ok, "expected not exits")
+
+	_, err = bkt.ObjectSize(ctx, "id1/obj_1.some")
+	testutil.NotOk(t, err)
+	testutil.Assert(t, bkt.IsObjNotFoundErr(err), "expected not found error but got %s", err)
+
+	// Upload first object.
+	testutil.Ok(t, bkt.Upload(ctx, "id1/obj_1.some", strings.NewReader("@test-data@")))
+
+	// Double check we can immediately read it.
+	rc1, err := bkt.Get(ctx, "id1/obj_1.some")
+	testutil.Ok(t, err)
+	defer func() { testutil.Ok(t, rc1.Close()) }()
+	content, err := ioutil.ReadAll(rc1)
+	testutil.Ok(t, err)
+	testutil.Equals(t, "@test-data@", string(content))
+
+	// Check if we can get the correct size.
+	sz, err := bkt.ObjectSize(ctx, "id1/obj_1.some")
+	testutil.Ok(t, err)
+	testutil.Assert(t, sz == 11, "expected size to be equal to 11")
+
+	rc2, err := bkt.GetRange(ctx, "id1/obj_1.some", 1, 3)
+	testutil.Ok(t, err)
+	defer func() { testutil.Ok(t, rc2.Close()) }()
+	content, err = ioutil.ReadAll(rc2)
+	testutil.Ok(t, err)
+	testutil.Equals(t, "tes", string(content))
+
+	// Unspecified range with offset.
+	rcUnspecifiedLen, err := bkt.GetRange(ctx, "id1/obj_1.some", 1, -1)
+	testutil.Ok(t, err)
+	defer func() { testutil.Ok(t, rcUnspecifiedLen.Close()) }()
+	content, err = ioutil.ReadAll(rcUnspecifiedLen)
+	testutil.Ok(t, err)
+	testutil.Equals(t, "test-data@", string(content))
+
+	// Out of band offset. Do not rely on outcome.
+	// NOTE: For various providers we have different outcome.
+	// * GCS is giving 416 status code
+	// * S3 errors immdiately with invalid range error.
+	// * inmem and filesystem are returning 0 bytes.
+	//rcOffset, err := bkt.GetRange(ctx, "id1/obj_1.some", 124141, 3)
+
+	// Out of band length. We expect to read file fully.
+	rcLength, err := bkt.GetRange(ctx, "id1/obj_1.some", 3, 9999)
+	testutil.Ok(t, err)
+	defer func() { testutil.Ok(t, rcLength.Close()) }()
+	content, err = ioutil.ReadAll(rcLength)
+	testutil.Ok(t, err)
+	testutil.Equals(t, "st-data@", string(content))
+
+	ok, err = bkt.Exists(ctx, "id1/obj_1.some")
+	testutil.Ok(t, err)
+	testutil.Assert(t, ok, "expected exits")
+
+	// Upload other objects.
+	testutil.Ok(t, bkt.Upload(ctx, "id1/obj_2.some", strings.NewReader("@test-data2@")))
+	// Upload should be idempotent.
+	testutil.Ok(t, bkt.Upload(ctx, "id1/obj_2.some", strings.NewReader("@test-data2@")))
+	testutil.Ok(t, bkt.Upload(ctx, "id1/obj_3.some", strings.NewReader("@test-data3@")))
+	testutil.Ok(t, bkt.Upload(ctx, "id2/obj_4.some", strings.NewReader("@test-data4@")))
+	testutil.Ok(t, bkt.Upload(ctx, "obj_5.some", strings.NewReader("@test-data5@")))
+
+	// Can we iter over items from top dir?
+	var seen []string
+	testutil.Ok(t, bkt.Iter(ctx, "", func(fn string) error {
+		seen = append(seen, fn)
+		return nil
+	}))
+	expected := []string{"obj_5.some", "id1/", "id2/"}
+	sort.Strings(expected)
+	sort.Strings(seen)
+	testutil.Equals(t, expected, seen)
+
+	// Can we iter over items from id1/ dir?
+	seen = []string{}
+	testutil.Ok(t, bkt.Iter(ctx, "id1/", func(fn string) error {
+		seen = append(seen, fn)
+		return nil
+	}))
+	testutil.Equals(t, []string{"id1/obj_1.some", "id1/obj_2.some", "id1/obj_3.some"}, seen)
+
+	// Can we iter over items from id1 dir?
+	seen = []string{}
+	testutil.Ok(t, bkt.Iter(ctx, "id1", func(fn string) error {
+		seen = append(seen, fn)
+		return nil
+	}))
+	testutil.Equals(t, []string{"id1/obj_1.some", "id1/obj_2.some", "id1/obj_3.some"}, seen)
+
+	// Can we iter over items from not existing dir?
+	testutil.Ok(t, bkt.Iter(ctx, "id0", func(fn string) error {
+		t.Error("Not expected to loop through not existing directory")
+		t.FailNow()
+
+		return nil
+	}))
+
+	testutil.Ok(t, bkt.Delete(ctx, "id1/obj_2.some"))
+
+	// Delete is expected to fail on non existing object.
+	// NOTE: Don't rely on this. S3 is not complying with this as GCS is.
+	// testutil.NotOk(t, bkt.Delete(ctx, "id1/obj_2.some"))
+
+	// Can we iter over items from id1/ dir and see obj2 being deleted?
+	seen = []string{}
+	testutil.Ok(t, bkt.Iter(ctx, "id1/", func(fn string) error {
+		seen = append(seen, fn)
+		return nil
+	}))
+	testutil.Equals(t, []string{"id1/obj_1.some", "id1/obj_3.some"}, seen)
+
+	testutil.Ok(t, bkt.Delete(ctx, "id2/obj_4.some"))
+
+	seen = []string{}
+	testutil.Ok(t, bkt.Iter(ctx, "", func(fn string) error {
+		seen = append(seen, fn)
+		return nil
+	}))
+	expected = []string{"obj_5.some", "id1/"}
+	sort.Strings(expected)
+	sort.Strings(seen)
+	testutil.Equals(t, expected, seen)
+}
diff --git a/vendor/github.com/thanos-io/thanos/pkg/testutil/testutil.go b/vendor/github.com/thanos-io/thanos/pkg/testutil/testutil.go
index e1ba298d7e..8b1e8e1d4f 100644
--- a/vendor/github.com/thanos-io/thanos/pkg/testutil/testutil.go
+++ b/vendor/github.com/thanos-io/thanos/pkg/testutil/testutil.go
@@ -18,6 +18,7 @@ import (
 
 // Assert fails the test if the condition is false.
 func Assert(tb testing.TB, condition bool, v ...interface{}) {
+	tb.Helper()
 	if condition {
 		return
 	}
@@ -27,11 +28,12 @@ func Assert(tb testing.TB, condition bool, v ...interface{}) {
 	if len(v) > 0 {
 		msg = fmt.Sprintf(v[0].(string), v[1:]...)
 	}
-	tb.Fatalf("\033[31m%s:%d: "+msg+"\033[39m\n\n", append([]interface{}{filepath.Base(file), line}, v...)...)
+	tb.Fatalf("\033[31m%s:%d: "+msg+"\033[39m\n\n", filepath.Base(file), line)
 }
 
 // Ok fails the test if an err is not nil.
 func Ok(tb testing.TB, err error, v ...interface{}) {
+	tb.Helper()
 	if err == nil {
 		return
 	}
@@ -46,6 +48,7 @@ func Ok(tb testing.TB, err error, v ...interface{}) {
 
 // NotOk fails the test if an err is nil.
 func NotOk(tb testing.TB, err error, v ...interface{}) {
+	tb.Helper()
 	if err != nil {
 		return
 	}
@@ -60,6 +63,7 @@ func NotOk(tb testing.TB, err error, v ...interface{}) {
 
 // Equals fails the test if exp is not equal to act.
 func Equals(tb testing.TB, exp, act interface{}, v ...interface{}) {
+	tb.Helper()
 	if reflect.DeepEqual(exp, act) {
 		return
 	}