Repository: cortexproject/cortex
Pull request #2465 (2020-04-15 12:29:31)

Enhance TestWAL to test more of WAL
We were not testing the WAL enough before as we checkpoint when shutting down the ingester. I have added `checkpointDuringShutdown` in config just for tests and using it to test more of the WAL.


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/CHANGELOG.md b/CHANGELOG.md
index e7fa95dee4..34a734390c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -23,12 +23,13 @@
 * [ENHANCEMENT] FIFO cache to support eviction based on memory usage. The `-<prefix>.fifocache.size` CLI flag has been renamed to `-<prefix>.fifocache.max-size-items` as well as its YAML config option `size` renamed to `max_size_items`. Added `-<prefix>.fifocache.max-size-bytes` CLI flag and YAML config option `max_size_bytes` to specify memory limit of the cache. #2319
 * [ENHANCEMENT] Single Binary: Added query-frontend to the single binary.  Single binary users will now benefit from various query-frontend features.  Primarily: sharding, parallelization, load shedding, additional caching (if configured), and query retries. #2437
 * [ENHANCEMENT] Allow 1w (where w denotes week) and 1y (where y denotes year) when setting `-store.cache-lookups-older-than` and `-store.max-look-back-period`. #2454
-* [ENHANCEMENT] Optimize index queries for matchers using "a|b|c"-type regex. #2446
+* [ENHANCEMENT] Optimize index queries for matchers using "a|b|c"-type regex. #2446 #2475
 * [BUGFIX] Fixes #2411, Ensure requests are properly routed to the prometheus api embedded in the query if `-server.path-prefix` is set. #2372
 * [BUGFIX] Experimental TSDB: fixed chunk data corruption when querying back series using the experimental blocks storage. #2400
 * [BUGFIX] Cassandra Storage: Fix endpoint TLS host verification. #2109
 * [BUGFIX] Experimental TSDB: fixed response status code from `422` to `500` when an error occurs while iterating chunks with the experimental blocks storage. #2402
 * [BUGFIX] Ring: Fixed a situation where upgrading from pre-1.0 cortex with a rolling strategy caused new 1.0 ingesters to lose their zone value in the ring until manually forced to re-register. #2404
+* [BUGFIX] Distributor: `/all_user_stats` now show API and Rule Ingest Rate correctly. #2457
 
 ## 1.0.0 / 2020-04-02
 
diff --git a/docs/configuration/arguments.md b/docs/configuration/arguments.md
index 4028a15c3e..3c00036129 100644
--- a/docs/configuration/arguments.md
+++ b/docs/configuration/arguments.md
@@ -126,6 +126,8 @@ The ingester query API was improved over time, but defaults to the old behaviour
 
    Set this flag to `true` for the new behaviour.
 
+   Important to note is that when setting this flag to `true`, it has to be set on both the distributor and the querier. If the flag is only set on the distributor and not on the querier, you will get incomplete query results because not all ingesters are queried.
+
    **Upgrade notes**: As this flag also makes all queries always read from all ingesters, the upgrade path is pretty trivial; just enable the flag. When you do enable it, you'll see a spike in the number of active series as the writes are "reshuffled" amongst the ingesters, but over the next stale period all the old series will be flushed, and you should end up with much better load balancing. With this flag enabled in the queriers, reads will always catch all the data from all ingesters.
 
 - `-distributor.extra-query-delay`
diff --git a/docs/production/caching.md b/docs/production/caching.md
index 0cfa4cced2..a823136649 100644
--- a/docs/production/caching.md
+++ b/docs/production/caching.md
@@ -163,7 +163,14 @@ These query results can potentially be very large, and as such the maximum value
 The cache is populated when there is a cache miss.
 Items stay in the cache indefinitely.
 
-The query results cache should be configured on the **query-frontend** using flags with `-frontend.cache` prefix.
+The query results cache should be configured on the **query-frontend** using flags with `-frontend` prefix:
+
+- `-frontend.memcached.*` flags to use Memcached backend
+- `-frontend.redis.*` flags to use Redis backend
+- `-frontend.fifocache.*` and `-frontend.cache.enable-fifocache` flags to use the per-process in-memory cache (not shared across multiple query-frontend instances)
+
+Please keep in mind to also enable `-querier.cache-results=true` and configure `-querier.split-queries-by-interval=24h` (`24h` is a good starting point).
+
 
 ### Index Write Cache
 
diff --git a/docs/production/ingesters-with-wal.md b/docs/production/ingesters-with-wal.md
index 43670cf460..8503d517c7 100644
--- a/docs/production/ingesters-with-wal.md
+++ b/docs/production/ingesters-with-wal.md
@@ -65,13 +65,12 @@ PS: Given you have to scale down 1 ingester at a time, you can pipeline the shut
 
 **Fallback option**
 
-There is a [flush mode ingester](https://github.com/cortexproject/cortex/pull/1747) in progress, and with recent discussions there will be a separate target called flusher in it's place.
+There is a `flusher` target that can be used to flush the data in the WAL. It's config can be found [here](../configuration/config-file-reference.md#flusher-config). As flusher depends on the chunk store and the http API components, you need to also set all the config related to them similar to ingesters (see [api,storage,chunk_store,limits,runtime_config](../configuration/config-file-reference.md#supported-contents-and-default-values-of-the-config-file) and [schema](../configuration/schema-config-reference.md)). Pro tip: Re-use the ingester config and set the `target` as `flusher` with additional flusher config, the irrelevant config will be ignored.
 
-You can run it as a kubernetes job which will
-* Attach to the volume of the scaled down ingester
-* Recover from the WAL
-* And flush all the chunks.
+You can run it as a Kubernetes job which will:
 
-This job is to be run for all the ingesters that you missed hitting the shutdown endpoint as a first option.
+1. Attach to the volume of the scaled down ingester.
+2. Recover from the WAL.
+3. And flush all the chunks.
 
-More info about the flusher target will be added once it's upstream.
\ No newline at end of file
+This job is to be run for all the PVCs linked to the ingesters that you missed hitting the shutdown endpoint as a first option.
diff --git a/go.mod b/go.mod
index 027c77625f..3971ea87d0 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.1-0.20200416112106-b391ca115ed8
 	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..e268bfc997 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.1-0.20200416112106-b391ca115ed8 h1:z7sOhoCEWnrQ2MIew3cJxsaxKT0AQu5pgXA8ZjdrYlk=
+github.com/thanos-io/thanos v0.12.1-0.20200416112106-b391ca115ed8/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/chunk/chunk_store.go b/pkg/chunk/chunk_store.go
index 02cc7da879..6e904f6d85 100644
--- a/pkg/chunk/chunk_store.go
+++ b/pkg/chunk/chunk_store.go
@@ -441,8 +441,8 @@ func (c *baseStore) lookupIdsByMetricNameMatcher(ctx context.Context, from, thro
 	} else if matcher.Type == labels.MatchEqual {
 		labelName = matcher.Name
 		queries, err = c.schema.GetReadQueriesForMetricLabelValue(from, through, userID, metricName, matcher.Name, matcher.Value)
-	} else if matcher.Type == labels.MatchRegexp && len(findSetMatches(matcher.Value)) > 0 {
-		set := findSetMatches(matcher.Value)
+	} else if matcher.Type == labels.MatchRegexp && len(FindSetMatches(matcher.Value)) > 0 {
+		set := FindSetMatches(matcher.Value)
 		for _, v := range set {
 			var qs []IndexQuery
 			qs, err = c.schema.GetReadQueriesForMetricLabelValue(from, through, userID, metricName, matcher.Name, v)
diff --git a/pkg/chunk/opts.go b/pkg/chunk/opts.go
index 10d07b012f..3384ecd7a4 100644
--- a/pkg/chunk/opts.go
+++ b/pkg/chunk/opts.go
@@ -19,9 +19,9 @@ func init() {
 	}
 }
 
+// FindSetMatches returns list of values that can be equality matched on.
 // copied from Prometheus querier.go, removed check for Prometheus wrapper.
-// Returns list of values that can regex matches.
-func findSetMatches(pattern string) []string {
+func FindSetMatches(pattern string) []string {
 	escaped := false
 	sets := []*strings.Builder{{}}
 	for i := 0; i < len(pattern); i++ {
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/distributor/distributor.go b/pkg/distributor/distributor.go
index 1ceb07e7c2..546dfa1578 100644
--- a/pkg/distributor/distributor.go
+++ b/pkg/distributor/distributor.go
@@ -536,7 +536,7 @@ func (d *Distributor) Push(ctx context.Context, req *client.WriteRequest) (*clie
 		if sp := opentracing.SpanFromContext(ctx); sp != nil {
 			localCtx = opentracing.ContextWithSpan(localCtx, sp)
 		}
-		return d.send(localCtx, ingester, timeseries, metadata)
+		return d.send(localCtx, ingester, timeseries, metadata, req.Source)
 	}, func() { client.ReuseSlice(req.Timeseries) })
 	if err != nil {
 		return nil, err
@@ -566,7 +566,7 @@ func sortLabelsIfNeeded(labels []client.LabelAdapter) {
 	})
 }
 
-func (d *Distributor) send(ctx context.Context, ingester ring.IngesterDesc, timeseries []client.PreallocTimeseries, metadata []*client.MetricMetadata) error {
+func (d *Distributor) send(ctx context.Context, ingester ring.IngesterDesc, timeseries []client.PreallocTimeseries, metadata []*client.MetricMetadata, source client.WriteRequest_SourceEnum) error {
 	h, err := d.ingesterPool.GetClientFor(ingester.Addr)
 	if err != nil {
 		return err
@@ -576,6 +576,7 @@ func (d *Distributor) send(ctx context.Context, ingester ring.IngesterDesc, time
 	req := client.WriteRequest{
 		Timeseries: timeseries,
 		Metadata:   metadata,
+		Source:     source,
 	}
 	_, err = c.Push(ctx, &req)
 
diff --git a/pkg/ingester/index/index.go b/pkg/ingester/index/index.go
index 7c1c46c570..001dd1a817 100644
--- a/pkg/ingester/index/index.go
+++ b/pkg/ingester/index/index.go
@@ -8,6 +8,7 @@ import (
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/pkg/labels"
 
+	"github.com/cortexproject/cortex/pkg/chunk"
 	"github.com/cortexproject/cortex/pkg/ingester/client"
 	"github.com/cortexproject/cortex/pkg/util"
 )
@@ -168,6 +169,13 @@ func (shard *indexShard) lookup(matchers []*labels.Matcher) []model.Fingerprint
 		if matcher.Type == labels.MatchEqual {
 			fps := values.fps[matcher.Value]
 			toIntersect = append(toIntersect, fps.fps...) // deliberate copy
+		} else if matcher.Type == labels.MatchRegexp && len(chunk.FindSetMatches(matcher.Value)) > 0 {
+			// The lookup is of the form `=~"a|b|c|d"`
+			set := chunk.FindSetMatches(matcher.Value)
+			for _, value := range set {
+				toIntersect = append(toIntersect, values.fps[value].fps...)
+			}
+			sort.Sort(toIntersect)
 		} else {
 			// accumulate the matching fingerprints (which are all distinct)
 			// then sort to maintain the invariant
diff --git a/pkg/ingester/wal.go b/pkg/ingester/wal.go
index 398a42388b..ba76c4f125 100644
--- a/pkg/ingester/wal.go
+++ b/pkg/ingester/wal.go
@@ -35,6 +35,8 @@ type WALConfig struct {
 	Recover            bool          `yaml:"recover_from_wal"`
 	Dir                string        `yaml:"wal_dir"`
 	CheckpointDuration time.Duration `yaml:"checkpoint_duration"`
+	// We always checkpoint during shutdown. This option exists for the tests.
+	checkpointDuringShutdown bool
 }
 
 // RegisterFlags adds the flags required to config this to the given FlagSet
@@ -44,6 +46,7 @@ func (cfg *WALConfig) RegisterFlags(f *flag.FlagSet) {
 	f.BoolVar(&cfg.WALEnabled, "ingester.wal-enabled", false, "Enable writing of ingested data into WAL.")
 	f.BoolVar(&cfg.CheckpointEnabled, "ingester.checkpoint-enabled", true, "Enable checkpointing of in-memory chunks. It should always be true when using normally. Set it to false iff you are doing some small tests as there is no mechanism to delete the old WAL yet if checkpoint is disabled.")
 	f.DurationVar(&cfg.CheckpointDuration, "ingester.checkpoint-duration", 30*time.Minute, "Interval at which checkpoints should be created.")
+	cfg.checkpointDuringShutdown = true
 }
 
 // WAL interface allows us to have a no-op WAL when the WAL is disabled.
@@ -172,9 +175,11 @@ func (w *walWrapper) run() {
 			level.Info(util.Logger).Log("msg", "checkpoint done", "time", elapsed.String())
 			w.checkpointDuration.Observe(elapsed.Seconds())
 		case <-w.quit:
-			level.Info(util.Logger).Log("msg", "creating checkpoint before shutdown")
-			if err := w.performCheckpoint(true); err != nil {
-				level.Error(util.Logger).Log("msg", "error checkpointing series during shutdown", "err", err)
+			if w.cfg.checkpointDuringShutdown {
+				level.Info(util.Logger).Log("msg", "creating checkpoint before shutdown")
+				if err := w.performCheckpoint(true); err != nil {
+					level.Error(util.Logger).Log("msg", "error checkpointing series during shutdown", "err", err)
+				}
 			}
 			return
 		}
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/ring/ring.go b/pkg/ring/ring.go
index 47a771de08..33751d1b9b 100644
--- a/pkg/ring/ring.go
+++ b/pkg/ring/ring.go
@@ -99,7 +99,6 @@ func (cfg *Config) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
 type Ring struct {
 	services.Service
 
-	name     string
 	key      string
 	cfg      Config
 	KVClient kv.Client
@@ -133,7 +132,6 @@ func NewWithStoreClientAndStrategy(cfg Config, name, key string, store kv.Client
 	}
 
 	r := &Ring{
-		name:     name,
 		key:      key,
 		cfg:      cfg,
 		KVClient: store,
@@ -142,27 +140,32 @@ func NewWithStoreClientAndStrategy(cfg Config, name, key string, store kv.Client
 		memberOwnershipDesc: prometheus.NewDesc(
 			"cortex_ring_member_ownership_percent",
 			"The percent ownership of the ring by member",
-			[]string{"member", "name"}, nil,
+			[]string{"member"},
+			map[string]string{"name": name},
 		),
 		numMembersDesc: prometheus.NewDesc(
 			"cortex_ring_members",
 			"Number of members in the ring",
-			[]string{"state", "name"}, nil,
+			[]string{"state"},
+			map[string]string{"name": name},
 		),
 		totalTokensDesc: prometheus.NewDesc(
 			"cortex_ring_tokens_total",
 			"Number of tokens in the ring",
-			[]string{"name"}, nil,
+			nil,
+			map[string]string{"name": name},
 		),
 		numTokensDesc: prometheus.NewDesc(
 			"cortex_ring_tokens_owned",
 			"The number of tokens in the ring owned by the member",
-			[]string{"member", "name"}, nil,
+			[]string{"member"},
+			map[string]string{"name": name},
 		),
 		oldestTimestampDesc: prometheus.NewDesc(
 			"cortex_ring_oldest_member_timestamp",
 			"Timestamp of the oldest member in the ring.",
-			[]string{"state", "name"}, nil,
+			[]string{"state"},
+			map[string]string{"name": name},
 		),
 	}
 
@@ -320,6 +323,7 @@ func (r *Ring) Describe(ch chan<- *prometheus.Desc) {
 	ch <- r.memberOwnershipDesc
 	ch <- r.numMembersDesc
 	ch <- r.totalTokensDesc
+	ch <- r.oldestTimestampDesc
 	ch <- r.numTokensDesc
 }
 
@@ -359,14 +363,12 @@ func (r *Ring) Collect(ch chan<- prometheus.Metric) {
 			prometheus.GaugeValue,
 			float64(totalOwned)/float64(math.MaxUint32),
 			id,
-			r.name,
 		)
 		ch <- prometheus.MustNewConstMetric(
 			r.numTokensDesc,
 			prometheus.GaugeValue,
 			float64(numTokens[id]),
 			id,
-			r.name,
 		)
 	}
 
@@ -396,7 +398,6 @@ func (r *Ring) Collect(ch chan<- prometheus.Metric) {
 			prometheus.GaugeValue,
 			float64(count),
 			state,
-			r.name,
 		)
 	}
 	for state, timestamp := range oldestTimestampByState {
@@ -405,7 +406,6 @@ func (r *Ring) Collect(ch chan<- prometheus.Metric) {
 			prometheus.GaugeValue,
 			float64(timestamp),
 			state,
-			r.name,
 		)
 	}
 
@@ -413,7 +413,6 @@ func (r *Ring) Collect(ch chan<- prometheus.Metric) {
 		r.totalTokensDesc,
 		prometheus.GaugeValue,
 		float64(len(r.ringTokens)),
-		r.name,
 	)
 }
 
@@ -464,7 +463,6 @@ func (r *Ring) Subring(key uint32, n int) (ReadRing, error) {
 	}
 
 	sub := &Ring{
-		name:     "subring",
 		cfg:      r.cfg,
 		strategy: r.strategy,
 		ringDesc: &Desc{
diff --git a/pkg/ruler/compat.go b/pkg/ruler/compat.go
index e62107218f..70494c4e67 100644
--- a/pkg/ruler/compat.go
+++ b/pkg/ruler/compat.go
@@ -39,6 +39,8 @@ func (a *appender) AddFast(l labels.Labels, ref uint64, t int64, v float64) erro
 }
 
 func (a *appender) Commit() error {
+	// Since a.pusher is distributor, client.ReuseSlice will be called in a.pusher.Push.
+	// We shouldn't call client.ReuseSlice here.
 	_, err := a.pusher.Push(user.InjectOrgID(context.Background(), a.userID), client.ToWriteRequest(a.labels, a.samples, nil, client.RULE))
 	a.labels = nil
 	a.samples = nil
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/gateway.go b/pkg/storegateway/gateway.go
index bb534bf4dc..9a33f85eca 100644
--- a/pkg/storegateway/gateway.go
+++ b/pkg/storegateway/gateway.go
@@ -113,17 +113,21 @@ func newStoreGateway(gatewayCfg Config, storageCfg cortex_tsdb.Config, bucketCli
 		delegate = ring.NewLeaveOnStoppingDelegate(delegate, logger)
 		delegate = ring.NewTokensPersistencyDelegate(gatewayCfg.ShardingRing.TokensFilePath, ring.JOINING, delegate, logger)
 
-		g.ringLifecycler, err = ring.NewBasicLifecycler(lifecyclerCfg, RingName, RingKey, ringStore, delegate, logger, reg)
+		g.ringLifecycler, err = ring.NewBasicLifecycler(lifecyclerCfg, RingNameForServer, RingKey, ringStore, delegate, logger, reg)
 		if err != nil {
 			return nil, errors.Wrap(err, "create ring lifecycler")
 		}
 
 		ringCfg := gatewayCfg.ShardingRing.ToRingConfig()
-		g.ring, err = ring.NewWithStoreClientAndStrategy(ringCfg, RingName, RingKey, ringStore, &BlocksReplicationStrategy{})
+		g.ring, err = ring.NewWithStoreClientAndStrategy(ringCfg, RingNameForServer, RingKey, ringStore, &BlocksReplicationStrategy{})
 		if err != nil {
 			return nil, errors.Wrap(err, "create ring client")
 		}
 
+		if reg != nil {
+			reg.MustRegister(g.ring)
+		}
+
 		// Filter blocks by the shard of this store-gateway instance if the
 		// sharding is enabled.
 		filters = append(filters, NewShardingMetadataFilter(g.ring, lifecyclerCfg.Addr, logger))
diff --git a/pkg/storegateway/gateway_ring.go b/pkg/storegateway/gateway_ring.go
index 1b144d4354..48e96d08ad 100644
--- a/pkg/storegateway/gateway_ring.go
+++ b/pkg/storegateway/gateway_ring.go
@@ -15,11 +15,15 @@ import (
 )
 
 const (
-	// StoreGatewayRingKey is the key under which we store the store gateways ring in the KVStore.
+	// RingKey is the key under which we store the store gateways ring in the KVStore.
 	RingKey = "store-gateway"
 
-	// StoreGatewayRingName is the name of the ring used by the store gateways.
-	RingName = "store-gateway"
+	// RingNameForServer is the name of the ring used by the store gateway server.
+	RingNameForServer = "store-gateway"
+
+	// RingNameForClient is the name of the ring used by the store gateway client (we need
+	// a different name to avoid clashing Prometheus metrics when running in single-binary).
+	RingNameForClient = "store-gateway-client"
 
 	// We use a safe default instead of exposing to config option to the user
 	// in order to simplify the config.
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/pkg/util/push/push.go b/pkg/util/push/push.go
index 3af2448e01..f8eeec5d51 100644
--- a/pkg/util/push/push.go
+++ b/pkg/util/push/push.go
@@ -17,7 +17,6 @@ func Handler(cfg distributor.Config, push func(context.Context, *client.WriteReq
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		compressionType := util.CompressionTypeFor(r.Header.Get("X-Prometheus-Remote-Write-Version"))
 		var req client.PreallocWriteRequest
-		req.Source = client.API
 		_, err := util.ParseProtoReader(r.Context(), r.Body, int(r.ContentLength), cfg.MaxRecvMsgSize, &req, compressionType)
 		logger := util.WithContext(r.Context(), util.Logger)
 		if err != nil {
@@ -25,6 +24,9 @@ func Handler(cfg distributor.Config, push func(context.Context, *client.WriteReq
 			http.Error(w, err.Error(), http.StatusBadRequest)
 			return
 		}
+		if req.Source == 0 {
+			req.Source = client.API
+		}
 
 		if _, err := push(r.Context(), &req.WriteRequest); err != nil {
 			resp, ok := httpgrpc.HTTPResponseFromError(err)
diff --git a/tools/website/website.go b/tools/website/website.go
index cb91f84b73..5c788b3ce7 100644
--- a/tools/website/website.go
+++ b/tools/website/website.go
@@ -39,7 +39,7 @@ func main() {
 }
 
 var (
-	ref    = regexp.MustCompile(`\[(.*)\](\(.*?\))`)
+	ref    = regexp.MustCompile(`\[([^\\\]]+?)\](\(.*?\))`)
 	link   = regexp.MustCompile(`(\(.*?\))`)
 	prRef  = regexp.MustCompile(`#(\d+)`)
 	images = regexp.MustCompile(`\..*images/(.*)\.(png|gif|jpeg|jpg|pdf)`)
diff --git a/vendor/github.com/golang/groupcache/singleflight/singleflight.go b/vendor/github.com/golang/groupcache/singleflight/singleflight.go
new file mode 100644
index 0000000000..ff2c2ee4f3
--- /dev/null
+++ b/vendor/github.com/golang/groupcache/singleflight/singleflight.go
@@ -0,0 +1,64 @@
+/*
+Copyright 2012 Google Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+     http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package singleflight provides a duplicate function call suppression
+// mechanism.
+package singleflight
+
+import "sync"
+
+// call is an in-flight or completed Do call
+type call struct {
+	wg  sync.WaitGroup
+	val interface{}
+	err error
+}
+
+// Group represents a class of work and forms a namespace in which
+// units of work can be executed with duplicate suppression.
+type Group struct {
+	mu sync.Mutex       // protects m
+	m  map[string]*call // lazily initialized
+}
+
+// Do executes and returns the results of the given function, making
+// sure that only one execution is in-flight for a given key at a
+// time. If a duplicate comes in, the duplicate caller waits for the
+// original to complete and receives the same results.
+func (g *Group) Do(key string, fn func() (interface{}, error)) (interface{}, error) {
+	g.mu.Lock()
+	if g.m == nil {
+		g.m = make(map[string]*call)
+	}
+	if c, ok := g.m[key]; ok {
+		g.mu.Unlock()
+		c.wg.Wait()
+		return c.val, c.err
+	}
+	c := new(call)
+	c.wg.Add(1)
+	g.m[key] = c
+	g.mu.Unlock()
+
+	c.val, c.err = fn()
+	c.wg.Done()
+
+	g.mu.Lock()
+	delete(g.m, key)
+	g.mu.Unlock()
+
+	return c.val, c.err
+}
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/cacheutil/memcached_client.go b/vendor/github.com/thanos-io/thanos/pkg/cacheutil/memcached_client.go
index 9c522bcc7c..29f124b6c3 100644
--- a/vendor/github.com/thanos-io/thanos/pkg/cacheutil/memcached_client.go
+++ b/vendor/github.com/thanos-io/thanos/pkg/cacheutil/memcached_client.go
@@ -217,18 +217,24 @@ func newMemcachedClient(
 		Help:        "Total number of operations against memcached.",
 		ConstLabels: prometheus.Labels{"name": name},
 	}, []string{"operation"})
+	c.operations.WithLabelValues(opGetMulti)
+	c.operations.WithLabelValues(opSet)
 
 	c.failures = promauto.With(reg).NewCounterVec(prometheus.CounterOpts{
 		Name:        "thanos_memcached_operation_failures_total",
 		Help:        "Total number of operations against memcached that failed.",
 		ConstLabels: prometheus.Labels{"name": name},
 	}, []string{"operation"})
+	c.failures.WithLabelValues(opGetMulti)
+	c.failures.WithLabelValues(opSet)
 
 	c.skipped = promauto.With(reg).NewCounterVec(prometheus.CounterOpts{
 		Name:        "thanos_memcached_operation_skipped_total",
 		Help:        "Total number of operations against memcached that have been skipped.",
 		ConstLabels: prometheus.Labels{"name": name},
 	}, []string{"operation", "reason"})
+	c.skipped.WithLabelValues(opGetMulti, reasonMaxItemSize)
+	c.skipped.WithLabelValues(opSet, reasonMaxItemSize)
 
 	c.duration = promauto.With(reg).NewHistogramVec(prometheus.HistogramOpts{
 		Name:        "thanos_memcached_operation_duration_seconds",
@@ -236,6 +242,8 @@ func newMemcachedClient(
 		ConstLabels: prometheus.Labels{"name": name},
 		Buckets:     []float64{0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.2, 0.5, 1},
 	}, []string{"operation"})
+	c.duration.WithLabelValues(opGetMulti)
+	c.duration.WithLabelValues(opSet)
 
 	// As soon as the client is created it must ensure that memcached server
 	// addresses are resolved, so we're going to trigger an initial addresses
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/inmem.go b/vendor/github.com/thanos-io/thanos/pkg/objstore/inmem.go
new file mode 100644
index 0000000000..0233f795f0
--- /dev/null
+++ b/vendor/github.com/thanos-io/thanos/pkg/objstore/inmem.go
@@ -0,0 +1,191 @@
+// Copyright (c) The Thanos Authors.
+// Licensed under the Apache License 2.0.
+
+package objstore
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"io/ioutil"
+	"sort"
+	"strings"
+	"sync"
+
+	"github.com/pkg/errors"
+)
+
+var errNotFound = errors.New("inmem: object not found")
+
+// InMemBucket implements the objstore.Bucket interfaces against local memory.
+// Methods from Bucket interface are thread-safe. Objects are assumed to be immutable.
+type InMemBucket struct {
+	mtx     sync.RWMutex
+	objects map[string][]byte
+}
+
+// NewInMemBucket returns a new in memory Bucket.
+// NOTE: Returned bucket is just a naive in memory bucket implementation. For test use cases only.
+func NewInMemBucket() *InMemBucket {
+	return &InMemBucket{objects: map[string][]byte{}}
+}
+
+// Objects returns internally stored objects.
+// NOTE: For assert purposes.
+func (b *InMemBucket) Objects() map[string][]byte {
+	return b.objects
+}
+
+// Iter calls f for each entry in the given directory. The argument to f is the full
+// object name including the prefix of the inspected directory.
+func (b *InMemBucket) Iter(_ context.Context, dir string, f func(string) error) error {
+	unique := map[string]struct{}{}
+
+	var dirPartsCount int
+	dirParts := strings.SplitAfter(dir, DirDelim)
+	for _, p := range dirParts {
+		if p == "" {
+			continue
+		}
+		dirPartsCount++
+	}
+
+	b.mtx.RLock()
+	for filename := range b.objects {
+		if !strings.HasPrefix(filename, dir) || dir == filename {
+			continue
+		}
+
+		parts := strings.SplitAfter(filename, DirDelim)
+		unique[strings.Join(parts[:dirPartsCount+1], "")] = struct{}{}
+	}
+	b.mtx.RUnlock()
+
+	var keys []string
+	for n := range unique {
+		keys = append(keys, n)
+	}
+	sort.Slice(keys, func(i, j int) bool {
+		if strings.HasSuffix(keys[i], DirDelim) && strings.HasSuffix(keys[j], DirDelim) {
+			return strings.Compare(keys[i], keys[j]) < 0
+		}
+		if strings.HasSuffix(keys[i], DirDelim) {
+			return false
+		}
+		if strings.HasSuffix(keys[j], DirDelim) {
+			return true
+		}
+
+		return strings.Compare(keys[i], keys[j]) < 0
+	})
+
+	for _, k := range keys {
+		if err := f(k); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// Get returns a reader for the given object name.
+func (b *InMemBucket) Get(_ context.Context, name string) (io.ReadCloser, error) {
+	if name == "" {
+		return nil, errors.New("inmem: object name is empty")
+	}
+
+	b.mtx.RLock()
+	file, ok := b.objects[name]
+	b.mtx.RUnlock()
+	if !ok {
+		return nil, errNotFound
+	}
+
+	return ioutil.NopCloser(bytes.NewReader(file)), nil
+}
+
+// GetRange returns a new range reader for the given object name and range.
+func (b *InMemBucket) GetRange(_ context.Context, name string, off, length int64) (io.ReadCloser, error) {
+	if name == "" {
+		return nil, errors.New("inmem: object name is empty")
+	}
+
+	b.mtx.RLock()
+	file, ok := b.objects[name]
+	b.mtx.RUnlock()
+	if !ok {
+		return nil, errNotFound
+	}
+
+	if int64(len(file)) < off {
+		return ioutil.NopCloser(bytes.NewReader(nil)), nil
+	}
+
+	if length == -1 {
+		return ioutil.NopCloser(bytes.NewReader(file[off:])), nil
+	}
+
+	if length <= 0 {
+		return ioutil.NopCloser(bytes.NewReader(nil)), errors.New("length cannot be smaller or equal 0")
+	}
+
+	if int64(len(file)) <= off+length {
+		// Just return maximum of what we have.
+		length = int64(len(file)) - off
+	}
+
+	return ioutil.NopCloser(bytes.NewReader(file[off : off+length])), nil
+}
+
+// Exists checks if the given directory exists in memory.
+func (b *InMemBucket) Exists(_ context.Context, name string) (bool, error) {
+	b.mtx.RLock()
+	defer b.mtx.RUnlock()
+	_, ok := b.objects[name]
+	return ok, nil
+}
+
+// ObjectSize returns the size of the specified object.
+func (b *InMemBucket) ObjectSize(_ context.Context, name string) (uint64, error) {
+	b.mtx.RLock()
+	file, ok := b.objects[name]
+	b.mtx.RUnlock()
+	if !ok {
+		return 0, errNotFound
+	}
+	return uint64(len(file)), nil
+}
+
+// Upload writes the file specified in src to into the memory.
+func (b *InMemBucket) Upload(_ context.Context, name string, r io.Reader) error {
+	b.mtx.Lock()
+	defer b.mtx.Unlock()
+	body, err := ioutil.ReadAll(r)
+	if err != nil {
+		return err
+	}
+	b.objects[name] = body
+	return nil
+}
+
+// Delete removes all data prefixed with the dir.
+func (b *InMemBucket) Delete(_ context.Context, name string) error {
+	b.mtx.Lock()
+	defer b.mtx.Unlock()
+	if _, ok := b.objects[name]; !ok {
+		return errNotFound
+	}
+	delete(b.objects, name)
+	return nil
+}
+
+// IsObjNotFoundErr returns true if error means that object is not found. Relevant to Get operations.
+func (b *InMemBucket) IsObjNotFoundErr(err error) bool {
+	return errors.Cause(err) == errNotFound
+}
+
+func (b *InMemBucket) Close() error { return nil }
+
+// Name returns the bucket name.
+func (b *InMemBucket) Name() string {
+	return "inmem"
+}
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..285bebaf7d 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
@@ -149,6 +149,7 @@ func NewBucketWithConfig(logger log.Logger, config Config, component string) (*B
 			DualStack: true,
 		}).DialContext,
 		MaxIdleConns:          100,
+		MaxIdleConnsPerHost:   100,
 		IdleConnTimeout:       time.Duration(config.HTTPConfig.IdleConnTimeout),
 		TLSHandshakeTimeout:   10 * time.Second,
 		ExpectContinueTimeout: 1 * time.Second,
@@ -287,7 +288,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 +300,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 +333,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 +342,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/github.com/thanos-io/thanos/pkg/store/proxy.go b/vendor/github.com/thanos-io/thanos/pkg/store/proxy.go
index 8440de2f96..c071848c05 100644
--- a/vendor/github.com/thanos-io/thanos/pkg/store/proxy.go
+++ b/vendor/github.com/thanos-io/thanos/pkg/store/proxy.go
@@ -434,7 +434,12 @@ func startStreamSeriesSet(
 				s.warnCh.send(storepb.NewWarnSeriesResponse(errors.New(w)))
 				continue
 			}
-			s.recvCh <- rr.r.GetSeries()
+			select {
+			case s.recvCh <- rr.r.GetSeries():
+			case <-ctx.Done():
+				s.handleErr(errors.Wrapf(ctx.Err(), "failed to receive any data from %s", s.name), done)
+				return
+			}
 		}
 	}()
 	return s
diff --git a/vendor/modules.txt b/vendor/modules.txt
index c23bc172ac..38880bae2b 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.1-0.20200416112106-b391ca115ed8
 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/chunk/opts_test.go b/pkg/chunk/opts_test.go
index 282049f669..01d05407e6 100644
--- a/pkg/chunk/opts_test.go
+++ b/pkg/chunk/opts_test.go
@@ -44,7 +44,7 @@ func TestFindSetMatches(t *testing.T) {
 	}
 
 	for _, c := range cases {
-		matches := findSetMatches(c.pattern)
+		matches := FindSetMatches(c.pattern)
 		require.Equal(t, c.exp, matches)
 	}
 }
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/pkg/ingester/index/index_test.go b/pkg/ingester/index/index_test.go
index 9b3351c55f..0ac06d5e09 100644
--- a/pkg/ingester/index/index_test.go
+++ b/pkg/ingester/index/index_test.go
@@ -1,6 +1,9 @@
 package index
 
 import (
+	"fmt"
+	"strconv"
+	"strings"
 	"testing"
 
 	"github.com/prometheus/common/model"
@@ -42,6 +45,18 @@ func TestIndex(t *testing.T) {
 		{mustParseMatcher(`{foo="bar", flip="flap"}`), []model.Fingerprint{2}},
 		{mustParseMatcher(`{foo="baz", flip="flop"}`), []model.Fingerprint{1}},
 		{mustParseMatcher(`{foo="baz", flip="flap"}`), []model.Fingerprint{0}},
+
+		{mustParseMatcher(`{fizz=~"b.*"}`), []model.Fingerprint{}},
+
+		{mustParseMatcher(`{foo=~"bar.*"}`), []model.Fingerprint{2, 3}},
+		{mustParseMatcher(`{foo=~"ba.*"}`), []model.Fingerprint{0, 1, 2, 3}},
+		{mustParseMatcher(`{flip=~"flop|flap"}`), []model.Fingerprint{0, 1, 2, 3}},
+		{mustParseMatcher(`{flip=~"flaps"}`), []model.Fingerprint{}},
+
+		{mustParseMatcher(`{foo=~"bar|bax", flip="flop"}`), []model.Fingerprint{3}},
+		{mustParseMatcher(`{foo=~"bar|baz", flip="flap"}`), []model.Fingerprint{0, 2}},
+		{mustParseMatcher(`{foo=~"baz.+", flip="flop"}`), []model.Fingerprint{}},
+		{mustParseMatcher(`{foo=~"baz", flip="flap"}`), []model.Fingerprint{0}},
 	} {
 		assert.Equal(t, tc.fps, index.Lookup(tc.matchers))
 	}
@@ -51,6 +66,67 @@ func TestIndex(t *testing.T) {
 	assert.Equal(t, []string{"flap", "flop"}, index.LabelValues("flip"))
 }
 
+func BenchmarkSetRegexLookup(b *testing.B) {
+	// Prepare the benchmark.
+	seriesLabels := []string{"a", "b", "c", "d", "e", "f", "g", "h", "i", "j"}
+	seriesPerLabel := 100000
+
+	idx := New()
+	for _, l := range seriesLabels {
+		for i := 0; i < seriesPerLabel; i++ {
+			lbls := labels.FromStrings("foo", l, "bar", strconv.Itoa(i))
+			idx.Add(client.FromLabelsToLabelAdapters(lbls), model.Fingerprint(lbls.Hash()))
+		}
+	}
+
+	selectionLabels := []string{}
+	for i := 0; i < 100; i++ {
+		selectionLabels = append(selectionLabels, strconv.Itoa(i))
+	}
+
+	tests := []struct {
+		name    string
+		matcher string
+	}{
+		{
+			name:    "select all",
+			matcher: fmt.Sprintf(`{bar=~"%s"}`, strings.Join(selectionLabels, "|")),
+		},
+		{
+			name:    "select two",
+			matcher: fmt.Sprintf(`{bar=~"%s"}`, strings.Join(selectionLabels[:2], "|")),
+		},
+		{
+			name:    "select half",
+			matcher: fmt.Sprintf(`{bar=~"%s"}`, strings.Join(selectionLabels[:len(selectionLabels)/2], "|")),
+		},
+		{
+			name:    "select none",
+			matcher: `{bar=~"bleep|bloop"}`,
+		},
+		{
+			name:    "equality matcher",
+			matcher: `{bar="1"}`,
+		},
+		{
+			name:    "regex (non-set) matcher",
+			matcher: `{bar=~"1.*"}`,
+		},
+	}
+
+	b.ResetTimer()
+
+	for _, tc := range tests {
+		b.Run(fmt.Sprintf("%s:%s", tc.name, tc.matcher), func(b *testing.B) {
+			matcher := mustParseMatcher(tc.matcher)
+			for n := 0; n < b.N; n++ {
+				idx.Lookup(matcher)
+			}
+		})
+	}
+
+}
+
 func mustParseMatcher(s string) []*labels.Matcher {
 	ms, err := promql.ParseMetricSelector(s)
 	if err != nil {
diff --git a/pkg/ingester/wal_test.go b/pkg/ingester/wal_test.go
index 2fc7693774..764db54580 100644
--- a/pkg/ingester/wal_test.go
+++ b/pkg/ingester/wal_test.go
@@ -27,22 +27,30 @@ func TestWAL(t *testing.T) {
 	cfg.WALConfig.CheckpointEnabled = true
 	cfg.WALConfig.Recover = true
 	cfg.WALConfig.Dir = dirname
-	cfg.WALConfig.CheckpointDuration = 100 * time.Millisecond
+	cfg.WALConfig.CheckpointDuration = 100 * time.Minute
+	cfg.WALConfig.checkpointDuringShutdown = true
 
 	numSeries := 100
 	numSamplesPerSeriesPerPush := 10
-	numRestarts := 3
+	numRestarts := 5
 
 	// Build an ingester, add some samples, then shut it down.
 	_, ing := newTestStore(t, cfg, defaultClientTestConfig(), defaultLimitsTestConfig(), nil)
 	userIDs, testData := pushTestSamples(t, ing, numSeries, numSamplesPerSeriesPerPush, 0)
+	// Checkpoint happens when stopping.
 	require.NoError(t, services.StopAndAwaitTerminated(context.Background(), ing))
 
 	for r := 0; r < numRestarts; r++ {
+		if r == 2 {
+			// From 3rd restart onwards, we are disabling checkpointing during shutdown
+			// to test both checkpoint+WAL replay.
+			cfg.WALConfig.checkpointDuringShutdown = false
+		}
 		if r == numRestarts-1 {
 			cfg.WALConfig.WALEnabled = false
 			cfg.WALConfig.CheckpointEnabled = false
 		}
+
 		// Start a new ingester and recover the WAL.
 		_, ing = newTestStore(t, cfg, defaultClientTestConfig(), defaultLimitsTestConfig(), nil)
 
diff --git a/pkg/ring/ring_test.go b/pkg/ring/ring_test.go
index 5a2d045890..01f5389ab1 100644
--- a/pkg/ring/ring_test.go
+++ b/pkg/ring/ring_test.go
@@ -47,7 +47,6 @@ func benchmarkBatch(b *testing.B, numIngester, numKeys int) {
 	cfg := Config{}
 	flagext.DefaultValues(&cfg)
 	r := Ring{
-		name:     "ingester",
 		cfg:      cfg,
 		ringDesc: desc,
 		strategy: &DefaultReplicationStrategy{},
@@ -89,7 +88,6 @@ func TestDoBatchZeroIngesters(t *testing.T) {
 	}
 	desc := NewDesc()
 	r := Ring{
-		name:     "ingester",
 		cfg:      Config{},
 		ringDesc: desc,
 		strategy: &DefaultReplicationStrategy{},
@@ -143,7 +141,6 @@ func TestSubring(t *testing.T) {
 
 	// Create a ring with the ingesters
 	ring := Ring{
-		name: "main ring",
 		cfg: Config{
 			HeartbeatTimeout: time.Hour,
 		},
@@ -198,7 +195,6 @@ func TestStableSubring(t *testing.T) {
 
 	// Create a ring with the ingesters
 	ring := Ring{
-		name: "main ring",
 		cfg: Config{
 			HeartbeatTimeout: time.Hour,
 		},
@@ -257,7 +253,6 @@ func TestZoneAwareIngesterAssignmentSucccess(t *testing.T) {
 
 	// Create a ring with the ingesters
 	ring := Ring{
-		name: "main ring",
 		cfg: Config{
 			HeartbeatTimeout:  time.Hour,
 			ReplicationFactor: 3,
@@ -322,7 +317,6 @@ func TestZoneAwareIngesterAssignmentFailure(t *testing.T) {
 
 	// Create a ring with the ingesters
 	ring := Ring{
-		name: "main ring",
 		cfg: Config{
 			HeartbeatTimeout:  time.Hour,
 			ReplicationFactor: 3,
diff --git a/pkg/util/push/push_test.go b/pkg/util/push/push_test.go
new file mode 100644
index 0000000000..ab87d4f5dc
--- /dev/null
+++ b/pkg/util/push/push_test.go
@@ -0,0 +1,95 @@
+package push
+
+import (
+	"bytes"
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/golang/snappy"
+	"github.com/prometheus/prometheus/prompb"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/cortexproject/cortex/pkg/distributor"
+	"github.com/cortexproject/cortex/pkg/ingester/client"
+)
+
+func TestHandler_remoteWrite(t *testing.T) {
+	req := createRequest(t, createPrometheusRemoteWriteProtobuf(t))
+	resp := httptest.NewRecorder()
+	handler := Handler(distributor.Config{MaxRecvMsgSize: 100000}, verifyWriteRequestHandler(t, client.API))
+	handler.ServeHTTP(resp, req)
+	assert.Equal(t, 200, resp.Code)
+}
+
+func TestHandler_cortexWriteRequest(t *testing.T) {
+	req := createRequest(t, createCortexWriteRequestProtobuf(t))
+	resp := httptest.NewRecorder()
+	handler := Handler(distributor.Config{MaxRecvMsgSize: 100000}, verifyWriteRequestHandler(t, client.RULE))
+	handler.ServeHTTP(resp, req)
+	assert.Equal(t, 200, resp.Code)
+}
+
+func verifyWriteRequestHandler(t *testing.T, expectSource client.WriteRequest_SourceEnum) func(ctx context.Context, request *client.WriteRequest) (response *client.WriteResponse, err error) {
+	t.Helper()
+	return func(ctx context.Context, request *client.WriteRequest) (response *client.WriteResponse, err error) {
+		assert.Len(t, request.Timeseries, 1)
+		assert.Equal(t, "__name__", request.Timeseries[0].Labels[0].Name)
+		assert.Equal(t, "foo", request.Timeseries[0].Labels[0].Value)
+		assert.Equal(t, expectSource, request.Source)
+		return &client.WriteResponse{}, nil
+	}
+}
+
+func createRequest(t *testing.T, protobuf []byte) *http.Request {
+	t.Helper()
+	inoutBytes := snappy.Encode(nil, protobuf)
+	req, err := http.NewRequest("POST", "http://localhost/", bytes.NewReader(inoutBytes))
+	require.NoError(t, err)
+	req.Header.Add("Content-Encoding", "snappy")
+	req.Header.Set("Content-Type", "application/x-protobuf")
+	req.Header.Set("X-Prometheus-Remote-Write-Version", "0.1.0")
+	return req
+}
+
+func createPrometheusRemoteWriteProtobuf(t *testing.T) []byte {
+	t.Helper()
+	input := prompb.WriteRequest{
+		Timeseries: []prompb.TimeSeries{
+			{
+				Labels: []prompb.Label{
+					{Name: "__name__", Value: "foo"},
+				},
+				Samples: []prompb.Sample{
+					{Value: 1, Timestamp: time.Date(2020, 4, 1, 0, 0, 0, 0, time.UTC).UnixNano()},
+				},
+			},
+		},
+	}
+	inoutBytes, err := input.Marshal()
+	require.NoError(t, err)
+	return inoutBytes
+}
+func createCortexWriteRequestProtobuf(t *testing.T) []byte {
+	t.Helper()
+	ts := client.PreallocTimeseries{
+		TimeSeries: &client.TimeSeries{
+			Labels: []client.LabelAdapter{
+				{Name: "__name__", Value: "foo"},
+			},
+			Samples: []client.Sample{
+				{Value: 1, TimestampMs: time.Date(2020, 4, 1, 0, 0, 0, 0, time.UTC).UnixNano()},
+			},
+		},
+	}
+	input := client.WriteRequest{
+		Timeseries: []client.PreallocTimeseries{ts},
+		Source:     client.RULE,
+	}
+	inoutBytes, err := input.Marshal()
+	require.NoError(t, err)
+	return inoutBytes
+}
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
 	}