Repository: cortexproject/cortex
Pull request #2485 (2020-04-19 18:22:28)

Add a counter to track chunks which were de-duplicated at store time.
Signed-off-by: Ed Welch <edward.welch@grafana.com>

<!--  Thanks for sending a pull request!  Before submitting:

1. Read our CONTRIBUTING.md guide
2. Rebase your PR if it gets out of sync with master
-->

**What this PR does**:

At the moment it's a bit tricky to understand how effectively chunks are being de-duplicated at the store with the cache based de-dupe code.

This PR adds a counter which explicitly increments when a chunk was not stored because it was already in the cache.  

This can be combined with `cortex_ingester_flush_reasons` to get a ratio of chunks flushed vs deduped to better tell how well the de-duplication is working.

**Which issue(s) this PR fixes**:
Fixes #<issue number>

**Checklist**
- [ ] Tests updated
- [ ] Documentation added
- [x] `CHANGELOG.md` updated - the order of entries should be `[CHANGE]`, `[FEATURE]`, `[ENHANCEMENT]`, `[BUGFIX]`


pracucci: I think we could assume this fixes #2353.
bboreham: Is this different to the dedupe cache hit rate?
slim-bean: > Is this different to the dedupe cache hit rate?

If I followed all this correctly, the cache used for checking de-dupes is the chunks cache which is also used when `FetchChunks` is called, so the hit rate on this cache is affected by queries (if you have a lot of queries for really old data they would drive up the miss rate on this cache)

The index `writeDedupeCache` is only checked after the chunks cache is checked, so if the chunks cache returned a result, the writeDedupeCache is never checked and therefore the stats for it won't reflect successful de-dupes
bboreham: OK, is this different to the dedupe cache hit rate exposed by ingesters?
(ingesters never call FetchChunks)
slim-bean: > OK, is this different to the dedupe cache hit rate exposed by ingesters?
(ingesters never call FetchChunks)

Interesting I hadn't though about the runtime separation.

The problem here would be anyone running a singlebinary, which is possible to do with caching and distributed stores like dynamo/s3.  I know this probably wouldn't be advised but we see this a lot with Loki Helm users.

Also a recent change in Loki will allow ingesters to query the store directly (part of how the persisting BoltDB to the object store implementation works), which would result in ingesters calling FetchChunks and impacting using cache stats for this purpose.

For these reasons I still see value in a separate metric for this.

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index a971aff8ff..4c05e2ace7 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -35,6 +35,7 @@
 * [ENHANCEMENT] Added per tenant metrics for queries and chunks and bytes read from chunk store: #2463
   * `cortex_chunk_store_fetched_chunks_total` and `cortex_chunk_store_fetched_chunk_bytes_total`
   * `cortex_query_frontend_queries_total` (per tenant queries counted by the frontend)
+* [ENHANCEMENT] Add de-duplicated chunks counter `cortex_chunk_store_deduped_chunks_total` which counts every chunk not sent to the store because it was already sent by another replica. #2485
 * [ENHANCEMENT] query-frontend now also logs the POST data of long queries. #2481
 * [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
@@ -132,6 +133,7 @@ This is the first major release of Cortex. We made a lot of **breaking changes**
   * `-flusher.flush-op-timeout` is duration after which a flush should timeout.
 * [FEATURE] Ingesters can now have an optional availability zone set, to ensure metric replication is distributed across zones. This is set via the `-ingester.availability-zone` flag or the `availability_zone` field in the config file. #2317
 * [ENHANCEMENT] Better re-use of connections to DynamoDB and S3. #2268
+* [ENHANCEMENT] Reduce number of goroutines used while executing a single index query. #2280
 * [ENHANCEMENT] Experimental TSDB: Add support for local `filesystem` backend. #2245
 * [ENHANCEMENT] Experimental TSDB: Added memcached support for the TSDB index cache. #2290
 * [ENHANCEMENT] Experimental TSDB: Removed gRPC server to communicate between querier and BucketStore. #2324
diff --git a/docs/configuration/single-process-config.yaml b/docs/configuration/single-process-config.yaml
index 03c5680690..0f7ebabfc7 100644
--- a/docs/configuration/single-process-config.yaml
+++ b/docs/configuration/single-process-config.yaml
@@ -33,6 +33,11 @@ ingester:
   spread_flushes: true
   chunk_age_jitter: 0
 
+  walconfig:
+    wal_enabled: true
+    recover_from_wal: true
+    wal_dir: /tmp/cortex/wal
+
   lifecycler:
     # The address to advertise for this ingester.  Will be autodiscovered by
     # looking up address on eth0 or en0; can be specified if this fails.
@@ -43,6 +48,7 @@ ingester:
     min_ready_duration: 0s
     final_sleep: 0s
     num_tokens: 512
+    tokens_file_path: /tmp/cortex/wal/tokens
 
     # Use an in memory ring store, so we don't need to launch a Consul.
     ring:
diff --git a/pkg/chunk/aws/dynamodb_storage_client.go b/pkg/chunk/aws/dynamodb_storage_client.go
index d4395038c4..a39a0ef422 100644
--- a/pkg/chunk/aws/dynamodb_storage_client.go
+++ b/pkg/chunk/aws/dynamodb_storage_client.go
@@ -287,7 +287,7 @@ func (a dynamoDBStorageClient) QueryPages(ctx context.Context, queries []chunk.I
 	return chunk_util.DoParallelQueries(ctx, a.query, queries, callback)
 }
 
-func (a dynamoDBStorageClient) query(ctx context.Context, query chunk.IndexQuery, callback func(result chunk.ReadBatch) (shouldContinue bool)) error {
+func (a dynamoDBStorageClient) query(ctx context.Context, query chunk.IndexQuery, callback chunk_util.Callback) error {
 	input := &dynamodb.QueryInput{
 		TableName: aws.String(query.TableName),
 		KeyConditions: map[string]*dynamodb.Condition{
@@ -341,7 +341,7 @@ func (a dynamoDBStorageClient) query(ctx context.Context, query chunk.IndexQuery
 			return err
 		}
 
-		if !callback(response) {
+		if !callback(query, response) {
 			if err != nil {
 				return fmt.Errorf("QueryPages error: table=%v, err=%v", *input.TableName, page.Error())
 			}
diff --git a/pkg/chunk/aws/s3_storage_client.go b/pkg/chunk/aws/s3_storage_client.go
index f9fc0c5401..9744d3bf0d 100644
--- a/pkg/chunk/aws/s3_storage_client.go
+++ b/pkg/chunk/aws/s3_storage_client.go
@@ -168,9 +168,10 @@ func (a *S3ObjectClient) PutObject(ctx context.Context, objectKey string, object
 	})
 }
 
-// List only objects from the store non-recursively
-func (a *S3ObjectClient) List(ctx context.Context, prefix string) ([]chunk.StorageObject, error) {
+// List objects and common-prefixes i.e synthetic directories from the store non-recursively
+func (a *S3ObjectClient) List(ctx context.Context, prefix string) ([]chunk.StorageObject, []chunk.StorageCommonPrefix, error) {
 	var storageObjects []chunk.StorageObject
+	var commonPrefixes []chunk.StorageCommonPrefix
 
 	for i := range a.bucketNames {
 		err := instrument.CollectedRequest(ctx, "S3.List", s3RequestDuration, instrument.ErrorCode, func(ctx context.Context) error {
@@ -193,6 +194,10 @@ func (a *S3ObjectClient) List(ctx context.Context, prefix string) ([]chunk.Stora
 					})
 				}
 
+				for _, commonPrefix := range output.CommonPrefixes {
+					commonPrefixes = append(commonPrefixes, chunk.StorageCommonPrefix(commonPrefix.String()))
+				}
+
 				if !*output.IsTruncated {
 					// No more results to fetch
 					break
@@ -205,9 +210,9 @@ func (a *S3ObjectClient) List(ctx context.Context, prefix string) ([]chunk.Stora
 		})
 
 		if err != nil {
-			return nil, err
+			return nil, nil, err
 		}
 	}
 
-	return storageObjects, nil
+	return storageObjects, commonPrefixes, nil
 }
diff --git a/pkg/chunk/azure/blob_storage_client.go b/pkg/chunk/azure/blob_storage_client.go
index e38ca50e57..fd856302ed 100644
--- a/pkg/chunk/azure/blob_storage_client.go
+++ b/pkg/chunk/azure/blob_storage_client.go
@@ -167,18 +167,19 @@ func (b *BlobStorage) newPipeline() (pipeline.Pipeline, error) {
 	}), nil
 }
 
-// List only objects from the store non-recursively
-func (b *BlobStorage) List(ctx context.Context, prefix string) ([]chunk.StorageObject, error) {
+// List objects and common-prefixes i.e synthetic directories from the store non-recursively
+func (b *BlobStorage) List(ctx context.Context, prefix string) ([]chunk.StorageObject, []chunk.StorageCommonPrefix, error) {
 	var storageObjects []chunk.StorageObject
+	var commonPrefixes []chunk.StorageCommonPrefix
 
 	for marker := (azblob.Marker{}); marker.NotDone(); {
 		if ctx.Err() != nil {
-			return nil, ctx.Err()
+			return nil, nil, ctx.Err()
 		}
 
 		listBlob, err := b.containerURL.ListBlobsHierarchySegment(ctx, marker, b.delimiter, azblob.ListBlobsSegmentOptions{Prefix: prefix})
 		if err != nil {
-			return nil, err
+			return nil, nil, err
 		}
 
 		marker = listBlob.NextMarker
@@ -190,9 +191,14 @@ func (b *BlobStorage) List(ctx context.Context, prefix string) ([]chunk.StorageO
 				ModifiedAt: blobInfo.Properties.LastModified,
 			})
 		}
+
+		// Process the BlobPrefixes so called commonPrefixes or synthetic directories in the listed synthetic directory
+		for _, blobPrefix := range listBlob.Segment.BlobPrefixes {
+			commonPrefixes = append(commonPrefixes, chunk.StorageCommonPrefix(blobPrefix.Name))
+		}
 	}
 
-	return storageObjects, nil
+	return storageObjects, commonPrefixes, nil
 }
 
 func (b *BlobStorage) DeleteObject(ctx context.Context, chunkID string) error {
diff --git a/pkg/chunk/cassandra/storage_client.go b/pkg/chunk/cassandra/storage_client.go
index ce2b662cd9..34480c595b 100644
--- a/pkg/chunk/cassandra/storage_client.go
+++ b/pkg/chunk/cassandra/storage_client.go
@@ -276,7 +276,7 @@ func (s *StorageClient) QueryPages(ctx context.Context, queries []chunk.IndexQue
 	return util.DoParallelQueries(ctx, s.query, queries, callback)
 }
 
-func (s *StorageClient) query(ctx context.Context, query chunk.IndexQuery, callback func(result chunk.ReadBatch) (shouldContinue bool)) error {
+func (s *StorageClient) query(ctx context.Context, query chunk.IndexQuery, callback util.Callback) error {
 	var q *gocql.Query
 
 	switch {
@@ -313,7 +313,7 @@ func (s *StorageClient) query(ctx context.Context, query chunk.IndexQuery, callb
 		if err := scanner.Scan(&b.rangeValue, &b.value); err != nil {
 			return errors.WithStack(err)
 		}
-		if !callback(b) {
+		if !callback(query, b) {
 			return nil
 		}
 	}
diff --git a/pkg/chunk/gcp/bigtable_index_client.go b/pkg/chunk/gcp/bigtable_index_client.go
index 9e2628825e..f5822bdc5a 100644
--- a/pkg/chunk/gcp/bigtable_index_client.go
+++ b/pkg/chunk/gcp/bigtable_index_client.go
@@ -317,7 +317,7 @@ func (s *storageClientV1) QueryPages(ctx context.Context, queries []chunk.IndexQ
 	return chunk_util.DoParallelQueries(ctx, s.query, queries, callback)
 }
 
-func (s *storageClientV1) query(ctx context.Context, query chunk.IndexQuery, callback func(result chunk.ReadBatch) (shouldContinue bool)) error {
+func (s *storageClientV1) query(ctx context.Context, query chunk.IndexQuery, callback chunk_util.Callback) error {
 	const null = string('\xff')
 
 	sp, ctx := ot.StartSpanFromContext(ctx, "QueryPages", ot.Tag{Key: "tableName", Value: query.TableName}, ot.Tag{Key: "hashValue", Value: query.HashValue})
@@ -346,7 +346,7 @@ func (s *storageClientV1) query(ctx context.Context, query chunk.IndexQuery, cal
 
 	err := table.ReadRows(ctx, rowRange, func(r bigtable.Row) bool {
 		if query.ValueEqual == nil || bytes.Equal(r[columnFamily][0].Value, query.ValueEqual) {
-			return callback(&rowBatch{
+			return callback(query, &rowBatch{
 				row: r,
 			})
 		}
diff --git a/pkg/chunk/gcp/gcs_object_client.go b/pkg/chunk/gcp/gcs_object_client.go
index 49c497c736..2029e9e11b 100644
--- a/pkg/chunk/gcp/gcs_object_client.go
+++ b/pkg/chunk/gcp/gcs_object_client.go
@@ -107,14 +107,15 @@ func (s *GCSObjectClient) PutObject(ctx context.Context, objectKey string, objec
 	return nil
 }
 
-// List only objects from the store non-recursively
-func (s *GCSObjectClient) List(ctx context.Context, prefix string) ([]chunk.StorageObject, error) {
+// List objects and common-prefixes i.e synthetic directories from the store non-recursively
+func (s *GCSObjectClient) List(ctx context.Context, prefix string) ([]chunk.StorageObject, []chunk.StorageCommonPrefix, error) {
 	var storageObjects []chunk.StorageObject
+	var commonPrefixes []chunk.StorageCommonPrefix
 
 	iter := s.bucket.Objects(ctx, &storage.Query{Prefix: prefix, Delimiter: s.delimiter})
 	for {
 		if ctx.Err() != nil {
-			return nil, ctx.Err()
+			return nil, nil, ctx.Err()
 		}
 
 		attr, err := iter.Next()
@@ -122,12 +123,12 @@ func (s *GCSObjectClient) List(ctx context.Context, prefix string) ([]chunk.Stor
 			if err == iterator.Done {
 				break
 			}
-			return nil, err
+			return nil, nil, err
 		}
 
 		// When doing query with Delimiter, Prefix is the only field set for entries which represent synthetic "directory entries".
-		// We do not want to consider those entries since we are doing only non-recursive listing of objects for now.
 		if attr.Name == "" {
+			commonPrefixes = append(commonPrefixes, chunk.StorageCommonPrefix(attr.Prefix))
 			continue
 		}
 
@@ -137,7 +138,7 @@ func (s *GCSObjectClient) List(ctx context.Context, prefix string) ([]chunk.Stor
 		})
 	}
 
-	return storageObjects, nil
+	return storageObjects, commonPrefixes, nil
 }
 
 // DeleteObject deletes the specified object key from the configured GCS bucket. If the
diff --git a/pkg/chunk/inmemory_storage_client.go b/pkg/chunk/inmemory_storage_client.go
index 80c17678b9..5a3aed3b6d 100644
--- a/pkg/chunk/inmemory_storage_client.go
+++ b/pkg/chunk/inmemory_storage_client.go
@@ -374,7 +374,7 @@ func (m *MockStorage) DeleteObject(ctx context.Context, objectKey string) error
 	return nil
 }
 
-func (m *MockStorage) List(ctx context.Context, prefix string) ([]StorageObject, error) {
+func (m *MockStorage) List(ctx context.Context, prefix string) ([]StorageObject, []StorageCommonPrefix, error) {
 	m.mtx.RLock()
 	defer m.mtx.RUnlock()
 
@@ -384,7 +384,7 @@ func (m *MockStorage) List(ctx context.Context, prefix string) ([]StorageObject,
 		storageObjects = append(storageObjects, StorageObject{Key: key})
 	}
 
-	return storageObjects, nil
+	return storageObjects, []StorageCommonPrefix{}, nil
 }
 
 type mockWriteBatch struct {
diff --git a/pkg/chunk/local/boltdb_index_client.go b/pkg/chunk/local/boltdb_index_client.go
index b9aadd3382..8dad54ee5e 100644
--- a/pkg/chunk/local/boltdb_index_client.go
+++ b/pkg/chunk/local/boltdb_index_client.go
@@ -228,7 +228,7 @@ func (b *BoltIndexClient) QueryPages(ctx context.Context, queries []chunk.IndexQ
 	return chunk_util.DoParallelQueries(ctx, b.query, queries, callback)
 }
 
-func (b *BoltIndexClient) query(ctx context.Context, query chunk.IndexQuery, callback func(chunk.ReadBatch) (shouldContinue bool)) error {
+func (b *BoltIndexClient) query(ctx context.Context, query chunk.IndexQuery, callback chunk_util.Callback) error {
 	db, err := b.GetDB(query.TableName, DBOperationRead)
 	if err != nil {
 		if err == ErrUnexistentBoltDB {
@@ -241,7 +241,7 @@ func (b *BoltIndexClient) query(ctx context.Context, query chunk.IndexQuery, cal
 	return b.QueryDB(ctx, db, query, callback)
 }
 
-func (b *BoltIndexClient) QueryDB(ctx context.Context, db *bbolt.DB, query chunk.IndexQuery, callback func(chunk.ReadBatch) (shouldContinue bool)) error {
+func (b *BoltIndexClient) QueryDB(ctx context.Context, db *bbolt.DB, query chunk.IndexQuery, callback func(chunk.IndexQuery, chunk.ReadBatch) (shouldContinue bool)) error {
 	var start []byte
 	if len(query.RangeValuePrefix) > 0 {
 		start = []byte(query.HashValue + separator + string(query.RangeValuePrefix))
@@ -276,7 +276,7 @@ func (b *BoltIndexClient) QueryDB(ctx context.Context, db *bbolt.DB, query chunk
 
 			batch.rangeValue = k[len(rowPrefix):]
 			batch.value = v
-			if !callback(&batch) {
+			if !callback(query, &batch) {
 				break
 			}
 		}
diff --git a/pkg/chunk/local/fs_object_client.go b/pkg/chunk/local/fs_object_client.go
index 40b081c780..1cb066532a 100644
--- a/pkg/chunk/local/fs_object_client.go
+++ b/pkg/chunk/local/fs_object_client.go
@@ -90,35 +90,40 @@ func (f *FSObjectClient) PutObject(ctx context.Context, objectKey string, object
 	return fl.Close()
 }
 
-// List only objects from the store non-recursively
-func (f *FSObjectClient) List(ctx context.Context, prefix string) ([]chunk.StorageObject, error) {
+// List objects and common-prefixes i.e directories from the store non-recursively
+func (f *FSObjectClient) List(ctx context.Context, prefix string) ([]chunk.StorageObject, []chunk.StorageCommonPrefix, error) {
 	var storageObjects []chunk.StorageObject
+	var commonPrefixes []chunk.StorageCommonPrefix
+
 	folderPath := filepath.Join(f.cfg.Directory, prefix)
 
 	_, err := os.Stat(folderPath)
 	if err != nil {
 		if os.IsNotExist(err) {
-			return storageObjects, nil
+			return storageObjects, commonPrefixes, nil
 		}
-		return nil, err
+		return nil, nil, err
 	}
 
 	filesInfo, err := ioutil.ReadDir(folderPath)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
 	for _, fileInfo := range filesInfo {
+		nameWithPrefix := filepath.Join(prefix, fileInfo.Name())
+
 		if fileInfo.IsDir() {
+			commonPrefixes = append(commonPrefixes, chunk.StorageCommonPrefix(nameWithPrefix+chunk.DirDelim))
 			continue
 		}
 		storageObjects = append(storageObjects, chunk.StorageObject{
-			Key:        filepath.Join(prefix, fileInfo.Name()),
+			Key:        nameWithPrefix,
 			ModifiedAt: fileInfo.ModTime(),
 		})
 	}
 
-	return storageObjects, nil
+	return storageObjects, commonPrefixes, nil
 }
 
 func (f *FSObjectClient) DeleteObject(ctx context.Context, objectKey string) error {
diff --git a/pkg/chunk/series_store.go b/pkg/chunk/series_store.go
index 3c94143c07..1495d6579f 100644
--- a/pkg/chunk/series_store.go
+++ b/pkg/chunk/series_store.go
@@ -61,6 +61,11 @@ var (
 		// For 100k series for 7 week, could be 1.2m - 10*(8^(7-1)) = 2.6m.
 		Buckets: prometheus.ExponentialBuckets(10, 8, 7),
 	})
+	dedupedChunksTotal = promauto.NewCounter(prometheus.CounterOpts{
+		Namespace: "cortex",
+		Name:      "chunk_store_deduped_chunks_total",
+		Help:      "Count of chunks which were not stored because they have already been stored by another replica.",
+	})
 )
 
 // seriesStore implements Store
@@ -414,6 +419,7 @@ func (c *seriesStore) PutOne(ctx context.Context, from, through model.Time, chun
 	// If this chunk is in cache it must already be in the database so we don't need to write it again
 	found, _, _ := c.cache.Fetch(ctx, []string{chunk.ExternalKey()})
 	if len(found) > 0 {
+		dedupedChunksTotal.Inc()
 		return nil
 	}
 
diff --git a/pkg/chunk/storage_client.go b/pkg/chunk/storage_client.go
index 85c4babddc..6797e1493f 100644
--- a/pkg/chunk/storage_client.go
+++ b/pkg/chunk/storage_client.go
@@ -65,7 +65,7 @@ type ReadBatchIterator interface {
 type ObjectClient interface {
 	PutObject(ctx context.Context, objectKey string, object io.ReadSeeker) error
 	GetObject(ctx context.Context, objectKey string) (io.ReadCloser, error)
-	List(ctx context.Context, prefix string) ([]StorageObject, error)
+	List(ctx context.Context, prefix string) ([]StorageObject, []StorageCommonPrefix, error)
 	DeleteObject(ctx context.Context, objectKey string) error
 	Stop()
 }
@@ -75,3 +75,7 @@ type StorageObject struct {
 	Key        string
 	ModifiedAt time.Time
 }
+
+// StorageCommonPrefix represents a common prefix aka a synthetic directory in Object Store.
+// It is guaranteed to always end with DirDelim
+type StorageCommonPrefix string
diff --git a/pkg/chunk/util/util.go b/pkg/chunk/util/util.go
index 17ce783095..64ae96a969 100644
--- a/pkg/chunk/util/util.go
+++ b/pkg/chunk/util/util.go
@@ -12,11 +12,11 @@ import (
 	"github.com/cortexproject/cortex/pkg/util"
 )
 
+// Callback from an IndexQuery.
+type Callback func(chunk.IndexQuery, chunk.ReadBatch) bool
+
 // DoSingleQuery is the interface for indexes that don't support batching yet.
-type DoSingleQuery func(
-	ctx context.Context, query chunk.IndexQuery,
-	callback func(chunk.ReadBatch) bool,
-) error
+type DoSingleQuery func(context.Context, chunk.IndexQuery, Callback) error
 
 // QueryParallelism is the maximum number of subqueries run in
 // parallel per higher-level query
@@ -26,8 +26,12 @@ var QueryParallelism = 100
 // and indexes that don't yet support batching.
 func DoParallelQueries(
 	ctx context.Context, doSingleQuery DoSingleQuery, queries []chunk.IndexQuery,
-	callback func(chunk.IndexQuery, chunk.ReadBatch) bool,
+	callback Callback,
 ) error {
+	if len(queries) == 1 {
+		return doSingleQuery(ctx, queries[0], callback)
+	}
+
 	queue := make(chan chunk.IndexQuery)
 	incomingErrors := make(chan error)
 	n := util.Min(len(queries), QueryParallelism)
@@ -41,9 +45,7 @@ func DoParallelQueries(
 				if !ok {
 					return
 				}
-				incomingErrors <- doSingleQuery(ctx, query, func(r chunk.ReadBatch) bool {
-					return callback(query, r)
-				})
+				incomingErrors <- doSingleQuery(ctx, query, callback)
 			}
 		}()
 	}
@@ -67,9 +69,6 @@ func DoParallelQueries(
 	return lastErr
 }
 
-// Callback from an IndexQuery.
-type Callback func(chunk.IndexQuery, chunk.ReadBatch) bool
-
 type filteringBatch struct {
 	query chunk.IndexQuery
 	chunk.ReadBatch
diff --git a/pkg/compactor/compactor_ring.go b/pkg/compactor/compactor_ring.go
index 6caa4bd501..4db5f27478 100644
--- a/pkg/compactor/compactor_ring.go
+++ b/pkg/compactor/compactor_ring.go
@@ -71,7 +71,7 @@ func (cfg *RingConfig) ToLifecyclerConfig() ring.LifecyclerConfig {
 
 	// Configure lifecycler
 	lc.RingConfig = rc
-	lc.ListenPort = &cfg.ListenPort
+	lc.ListenPort = cfg.ListenPort
 	lc.Addr = cfg.InstanceAddr
 	lc.Port = cfg.InstancePort
 	lc.ID = cfg.InstanceID
diff --git a/pkg/cortex/modules.go b/pkg/cortex/modules.go
index 02489c2cd6..3975f84756 100644
--- a/pkg/cortex/modules.go
+++ b/pkg/cortex/modules.go
@@ -254,7 +254,7 @@ func (t *Cortex) initStoreQueryable(cfg *Config) (services.Service, error) {
 func (t *Cortex) initIngester(cfg *Config) (serv services.Service, err error) {
 	cfg.Ingester.LifecyclerConfig.RingConfig.KVStore.Multi.ConfigProvider = multiClientRuntimeConfigChannel(t.runtimeConfig)
 	cfg.Ingester.LifecyclerConfig.RingConfig.KVStore.MemberlistKV = t.memberlistKV.GetMemberlistKV
-	cfg.Ingester.LifecyclerConfig.ListenPort = &cfg.Server.GRPCListenPort
+	cfg.Ingester.LifecyclerConfig.ListenPort = cfg.Server.GRPCListenPort
 	cfg.Ingester.TSDBEnabled = cfg.Storage.Engine == storage.StorageEngineTSDB
 	cfg.Ingester.TSDBConfig = cfg.TSDB
 	cfg.Ingester.ShardByAllLabels = cfg.Distributor.ShardByAllLabels
diff --git a/pkg/distributor/distributor_ring.go b/pkg/distributor/distributor_ring.go
index 3251cdc7c4..2f3fba14b2 100644
--- a/pkg/distributor/distributor_ring.go
+++ b/pkg/distributor/distributor_ring.go
@@ -71,7 +71,7 @@ func (cfg *RingConfig) ToLifecyclerConfig() ring.LifecyclerConfig {
 
 	// Configure lifecycler
 	lc.RingConfig = rc
-	lc.ListenPort = &cfg.ListenPort
+	lc.ListenPort = cfg.ListenPort
 	lc.Addr = cfg.InstanceAddr
 	lc.Port = cfg.InstancePort
 	lc.ID = cfg.InstanceID
diff --git a/pkg/distributor/ingester_client_pool.go b/pkg/distributor/ingester_client_pool.go
index e8b6e03a40..cc0299bc02 100644
--- a/pkg/distributor/ingester_client_pool.go
+++ b/pkg/distributor/ingester_client_pool.go
@@ -38,5 +38,5 @@ func NewPool(cfg PoolConfig, ring ring.ReadRing, factory ring_client.PoolFactory
 		HealthCheckTimeout: cfg.RemoteTimeout,
 	}
 
-	return ring_client.NewPool("ingester", poolCfg, ring, factory, clients, logger)
+	return ring_client.NewPool("ingester", poolCfg, ring_client.NewRingServiceDiscovery(ring), factory, clients, logger)
 }
diff --git a/pkg/ingester/ingester.go b/pkg/ingester/ingester.go
index 6fa1e0fa7f..43eb996c3b 100644
--- a/pkg/ingester/ingester.go
+++ b/pkg/ingester/ingester.go
@@ -5,6 +5,7 @@ import (
 	"flag"
 	"fmt"
 	"net/http"
+	"os"
 	"sync"
 	"time"
 
@@ -157,6 +158,12 @@ func New(cfg Config, clientConfig client.Config, limits *validation.Overrides, c
 		}
 	}
 
+	if cfg.WALConfig.WALEnabled || cfg.WALConfig.Recover {
+		if err := os.MkdirAll(cfg.WALConfig.Dir, os.ModePerm); err != nil {
+			return nil, err
+		}
+	}
+
 	i := &Ingester{
 		cfg:          cfg,
 		clientConfig: clientConfig,
diff --git a/pkg/querier/blocks_store_replicated_set.go b/pkg/querier/blocks_store_replicated_set.go
index 992a28ac4f..ecaa0a4350 100644
--- a/pkg/querier/blocks_store_replicated_set.go
+++ b/pkg/querier/blocks_store_replicated_set.go
@@ -31,7 +31,7 @@ type blocksStoreReplicationSet struct {
 func newBlocksStoreReplicationSet(storesRing *ring.Ring, logger log.Logger, reg prometheus.Registerer) (*blocksStoreReplicationSet, error) {
 	s := &blocksStoreReplicationSet{
 		storesRing:  storesRing,
-		clientsPool: newStoreGatewayClientPool(storesRing, logger, reg),
+		clientsPool: newStoreGatewayClientPool(client.NewRingServiceDiscovery(storesRing), logger, reg),
 	}
 
 	var err error
diff --git a/pkg/querier/store_gateway_client.go b/pkg/querier/store_gateway_client.go
index bfaff7c779..98d0ef0648 100644
--- a/pkg/querier/store_gateway_client.go
+++ b/pkg/querier/store_gateway_client.go
@@ -10,7 +10,6 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/health/grpc_health_v1"
 
-	"github.com/cortexproject/cortex/pkg/ring"
 	"github.com/cortexproject/cortex/pkg/ring/client"
 	"github.com/cortexproject/cortex/pkg/storegateway/storegatewaypb"
 	"github.com/cortexproject/cortex/pkg/util/grpcclient"
@@ -59,7 +58,7 @@ func (c *storeGatewayClient) String() string {
 	return c.conn.Target()
 }
 
-func newStoreGatewayClientPool(r ring.ReadRing, logger log.Logger, reg prometheus.Registerer) *client.Pool {
+func newStoreGatewayClientPool(discovery client.PoolServiceDiscovery, logger log.Logger, reg prometheus.Registerer) *client.Pool {
 	// We prefer sane defaults instead of exposing further config options.
 	clientCfg := grpcclient.Config{
 		MaxRecvMsgSize:      100 << 20,
@@ -83,5 +82,5 @@ func newStoreGatewayClientPool(r ring.ReadRing, logger log.Logger, reg prometheu
 		ConstLabels: map[string]string{"client": "querier"},
 	})
 
-	return client.NewPool("store-gateway", poolCfg, r, newStoreGatewayClientFactory(clientCfg, reg), clientsCount, logger)
+	return client.NewPool("store-gateway", poolCfg, discovery, newStoreGatewayClientFactory(clientCfg, reg), clientsCount, logger)
 }
diff --git a/pkg/ring/client/pool.go b/pkg/ring/client/pool.go
index 3b90d6d087..41bc8728ab 100644
--- a/pkg/ring/client/pool.go
+++ b/pkg/ring/client/pool.go
@@ -13,7 +13,6 @@ import (
 	"github.com/weaveworks/common/user"
 	"google.golang.org/grpc/health/grpc_health_v1"
 
-	"github.com/cortexproject/cortex/pkg/ring"
 	"github.com/cortexproject/cortex/pkg/util"
 	"github.com/cortexproject/cortex/pkg/util/services"
 )
@@ -28,6 +27,12 @@ type PoolClient interface {
 // PoolFactory defines the signature for a client factory.
 type PoolFactory func(addr string) (PoolClient, error)
 
+// PoolServiceDiscovery defines the signature of a function returning the list
+// of known service endpoints. This function is used to remove stale clients from
+// the pool (a stale client is a client connected to a service endpoint no more
+// active).
+type PoolServiceDiscovery func() ([]string, error)
+
 // PoolConfig is config for creating a Pool.
 type PoolConfig struct {
 	CheckInterval      time.Duration
@@ -40,7 +45,7 @@ type Pool struct {
 	services.Service
 
 	cfg        PoolConfig
-	ring       ring.ReadRing
+	discovery  PoolServiceDiscovery
 	factory    PoolFactory
 	logger     log.Logger
 	clientName string
@@ -52,10 +57,10 @@ type Pool struct {
 }
 
 // NewPool creates a new Pool.
-func NewPool(clientName string, cfg PoolConfig, ring ring.ReadRing, factory PoolFactory, clientsMetric prometheus.Gauge, logger log.Logger) *Pool {
+func NewPool(clientName string, cfg PoolConfig, discovery PoolServiceDiscovery, factory PoolFactory, clientsMetric prometheus.Gauge, logger log.Logger) *Pool {
 	p := &Pool{
 		cfg:           cfg,
-		ring:          ring,
+		discovery:     discovery,
 		factory:       factory,
 		logger:        logger,
 		clientName:    clientName,
@@ -127,7 +132,7 @@ func (p *Pool) RemoveClientFor(addr string) {
 	}
 }
 
-// RegisteredAddresses returns all the addresses that a client is cached for
+// RegisteredAddresses returns all the service addresses for which there's an active client.
 func (p *Pool) RegisteredAddresses() []string {
 	result := []string{}
 	p.RLock()
@@ -146,24 +151,19 @@ func (p *Pool) Count() int {
 }
 
 func (p *Pool) removeStaleClients() {
-	// Only if this pool is based on the ring.
-	if p.ring == nil {
+	// Only if service discovery has been configured.
+	if p.discovery == nil {
 		return
 	}
 
-	clients := map[string]struct{}{}
-	replicationSet, err := p.ring.GetAll()
+	serviceAddrs, err := p.discovery()
 	if err != nil {
 		level.Error(util.Logger).Log("msg", "error removing stale clients", "err", err)
 		return
 	}
 
-	for _, ing := range replicationSet.Ingesters {
-		clients[ing.Addr] = struct{}{}
-	}
-
 	for _, addr := range p.RegisteredAddresses() {
-		if _, ok := clients[addr]; ok {
+		if util.StringsContain(serviceAddrs, addr) {
 			continue
 		}
 		level.Info(util.Logger).Log("msg", "removing stale client", "addr", addr)
diff --git a/pkg/ring/client/ring_service_discovery.go b/pkg/ring/client/ring_service_discovery.go
new file mode 100644
index 0000000000..4f02233d92
--- /dev/null
+++ b/pkg/ring/client/ring_service_discovery.go
@@ -0,0 +1,20 @@
+package client
+
+import (
+	"github.com/cortexproject/cortex/pkg/ring"
+)
+
+func NewRingServiceDiscovery(r ring.ReadRing) PoolServiceDiscovery {
+	return func() ([]string, error) {
+		replicationSet, err := r.GetAll()
+		if err != nil {
+			return nil, err
+		}
+
+		var addrs []string
+		for _, instance := range replicationSet.Ingesters {
+			addrs = append(addrs, instance.Addr)
+		}
+		return addrs, nil
+	}
+}
diff --git a/pkg/ring/lifecycler.go b/pkg/ring/lifecycler.go
index 1aa0f7aadf..4a09480bd4 100644
--- a/pkg/ring/lifecycler.go
+++ b/pkg/ring/lifecycler.go
@@ -46,7 +46,6 @@ type LifecyclerConfig struct {
 	RingConfig Config `yaml:"ring"`
 
 	// Config for the ingester lifecycle control
-	ListenPort       *int          `yaml:"-"`
 	NumTokens        int           `yaml:"num_tokens"`
 	HeartbeatPeriod  time.Duration `yaml:"heartbeat_period"`
 	ObservePeriod    time.Duration `yaml:"observe_period"`
@@ -62,6 +61,9 @@ type LifecyclerConfig struct {
 	Port           int    `doc:"hidden"`
 	ID             string `doc:"hidden"`
 	SkipUnregister bool   `yaml:"-"`
+
+	// Injected internally
+	ListenPort int `yaml:"-"`
 }
 
 // RegisterFlags adds the flags required to config this to the given FlagSet
@@ -143,7 +145,7 @@ func NewLifecycler(cfg LifecyclerConfig, flushTransferer FlushTransferer, ringNa
 	if err != nil {
 		return nil, err
 	}
-	port := GetInstancePort(cfg.Port, *cfg.ListenPort)
+	port := GetInstancePort(cfg.Port, cfg.ListenPort)
 	codec := GetCodec()
 	store, err := kv.NewClient(cfg.RingConfig.KVStore, codec)
 	if err != nil {
diff --git a/pkg/ruler/ruler_ring.go b/pkg/ruler/ruler_ring.go
index fb45cb2c8b..566474da12 100644
--- a/pkg/ruler/ruler_ring.go
+++ b/pkg/ruler/ruler_ring.go
@@ -76,7 +76,7 @@ func (cfg *RingConfig) ToLifecyclerConfig() ring.LifecyclerConfig {
 
 	// Configure lifecycler
 	lc.RingConfig = rc
-	lc.ListenPort = &cfg.ListenPort
+	lc.ListenPort = cfg.ListenPort
 	lc.Addr = cfg.InstanceAddr
 	lc.Port = cfg.InstancePort
 	lc.ID = cfg.InstanceID
diff --git a/pkg/ruler/rules/objectclient/rule_store.go b/pkg/ruler/rules/objectclient/rule_store.go
index f03a3d5dbf..b76074efde 100644
--- a/pkg/ruler/rules/objectclient/rule_store.go
+++ b/pkg/ruler/rules/objectclient/rule_store.go
@@ -64,7 +64,7 @@ func (o *RuleStore) getRuleGroup(ctx context.Context, objectKey string) (*rules.
 
 // ListAllRuleGroups returns all the active rule groups
 func (o *RuleStore) ListAllRuleGroups(ctx context.Context) (map[string]rules.RuleGroupList, error) {
-	ruleGroupObjects, err := o.client.List(ctx, generateRuleObjectKey("", "", ""))
+	ruleGroupObjects, _, err := o.client.List(ctx, generateRuleObjectKey("", "", ""))
 	if err != nil {
 		return nil, err
 	}
@@ -93,7 +93,7 @@ func (o *RuleStore) ListAllRuleGroups(ctx context.Context) (map[string]rules.Rul
 
 // ListRuleGroups returns all the active rule groups for a user
 func (o *RuleStore) ListRuleGroups(ctx context.Context, userID, namespace string) (rules.RuleGroupList, error) {
-	ruleGroupObjects, err := o.client.List(ctx, generateRuleObjectKey(userID, namespace, ""))
+	ruleGroupObjects, _, err := o.client.List(ctx, generateRuleObjectKey(userID, namespace, ""))
 	if err != nil {
 		return nil, err
 	}


Test patch:
diff --git a/pkg/chunk/local/fs_object_client_test.go b/pkg/chunk/local/fs_object_client_test.go
index 141e61dc5a..76709b098a 100644
--- a/pkg/chunk/local/fs_object_client_test.go
+++ b/pkg/chunk/local/fs_object_client_test.go
@@ -1,6 +1,7 @@
 package local
 
 import (
+	"bytes"
 	"context"
 	"io/ioutil"
 	"os"
@@ -10,7 +11,7 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestFsObjectClient_DeleteChunksBefore(t *testing.T) {
+func TestFSObjectClient_DeleteChunksBefore(t *testing.T) {
 	deleteFilesOlderThan := 10 * time.Minute
 
 	fsChunksDir, err := ioutil.TempDir(os.TempDir(), "fs-chunks")
@@ -56,3 +57,61 @@ func TestFsObjectClient_DeleteChunksBefore(t *testing.T) {
 	files, _ = ioutil.ReadDir(".")
 	require.Equal(t, 1, len(files), "Number of files should be 1 after enforcing retention")
 }
+
+func TestFSObjectClient_List(t *testing.T) {
+	fsObjectsDir, err := ioutil.TempDir(os.TempDir(), "fs-objects")
+	require.NoError(t, err)
+
+	bucketClient, err := NewFSObjectClient(FSConfig{
+		Directory: fsObjectsDir,
+	})
+	require.NoError(t, err)
+
+	defer func() {
+		require.NoError(t, os.RemoveAll(fsObjectsDir))
+	}()
+
+	foldersWithFiles := make(map[string][]string)
+	foldersWithFiles["folder1/"] = []string{"file1", "file2"}
+	foldersWithFiles["folder2/"] = []string{"file3", "file4", "file5"}
+
+	for folder, files := range foldersWithFiles {
+		for _, filename := range files {
+			err := bucketClient.PutObject(context.Background(), folder+filename, bytes.NewReader([]byte(filename)))
+			require.NoError(t, err)
+		}
+	}
+
+	files := []string{"outer-file1", "outer-file2"}
+
+	for _, fl := range files {
+		err := bucketClient.PutObject(context.Background(), fl, bytes.NewReader([]byte(fl)))
+		require.NoError(t, err)
+	}
+
+	storageObjects, commonPrefixes, err := bucketClient.List(context.Background(), "")
+	require.NoError(t, err)
+
+	require.Len(t, storageObjects, len(files))
+	for i := range storageObjects {
+		require.Equal(t, storageObjects[i].Key, files[i])
+	}
+
+	require.Len(t, commonPrefixes, len(foldersWithFiles))
+	for _, commonPrefix := range commonPrefixes {
+		_, ok := foldersWithFiles[string(commonPrefix)]
+		require.True(t, ok)
+	}
+
+	for folder, files := range foldersWithFiles {
+		storageObjects, commonPrefixes, err := bucketClient.List(context.Background(), folder)
+		require.NoError(t, err)
+
+		require.Len(t, storageObjects, len(files))
+		for i := range storageObjects {
+			require.Equal(t, storageObjects[i].Key, folder+files[i])
+		}
+
+		require.Len(t, commonPrefixes, 0)
+	}
+}
diff --git a/pkg/chunk/purger/purger_test.go b/pkg/chunk/purger/purger_test.go
index fbdf2eef6a..e46f2755f2 100644
--- a/pkg/chunk/purger/purger_test.go
+++ b/pkg/chunk/purger/purger_test.go
@@ -189,7 +189,7 @@ func TestDataPurger_BuildPlan(t *testing.T) {
 				require.NoError(t, err)
 				planPath := fmt.Sprintf("%s:%s/", userID, deleteRequest.RequestID)
 
-				plans, err := storageClient.List(context.Background(), planPath)
+				plans, _, err := storageClient.List(context.Background(), planPath)
 				require.NoError(t, err)
 				require.Equal(t, tc.expectedNumberOfPlans, len(plans))
 
diff --git a/pkg/compactor/compactor_ring_test.go b/pkg/compactor/compactor_ring_test.go
index fd3299854a..d25c996f90 100644
--- a/pkg/compactor/compactor_ring_test.go
+++ b/pkg/compactor/compactor_ring_test.go
@@ -18,7 +18,7 @@ func TestRingConfig_DefaultConfigToLifecyclerConfig(t *testing.T) {
 	// The default config of the compactor ring must be the exact same
 	// of the default lifecycler config, except few options which are
 	// intentionally overridden
-	expected.ListenPort = &cfg.ListenPort
+	expected.ListenPort = cfg.ListenPort
 	expected.RingConfig.ReplicationFactor = 1
 	expected.NumTokens = 512
 	expected.MinReadyDuration = 0
@@ -49,7 +49,7 @@ func TestRingConfig_CustomConfigToLifecyclerConfig(t *testing.T) {
 	expected.InfNames = cfg.InstanceInterfaceNames
 	expected.Port = cfg.InstancePort
 	expected.Addr = cfg.InstanceAddr
-	expected.ListenPort = &cfg.ListenPort
+	expected.ListenPort = cfg.ListenPort
 
 	// Hardcoded config
 	expected.RingConfig.ReplicationFactor = 1
diff --git a/pkg/distributor/distributor_ring_test.go b/pkg/distributor/distributor_ring_test.go
index c33fc7a818..24811aae90 100644
--- a/pkg/distributor/distributor_ring_test.go
+++ b/pkg/distributor/distributor_ring_test.go
@@ -20,7 +20,7 @@ func TestRingConfig_DefaultConfigToLifecyclerConfig(t *testing.T) {
 	// The default config of the distributor ring must be the exact same
 	// of the default lifecycler config, except few options which are
 	// intentionally overridden
-	expected.ListenPort = &cfg.ListenPort
+	expected.ListenPort = cfg.ListenPort
 	expected.RingConfig.ReplicationFactor = 1
 	expected.NumTokens = 1
 	expected.MinReadyDuration = 0
@@ -53,7 +53,7 @@ func TestRingConfig_CustomConfigToLifecyclerConfig(t *testing.T) {
 	expected.InfNames = cfg.InstanceInterfaceNames
 	expected.Port = cfg.InstancePort
 	expected.Addr = cfg.InstanceAddr
-	expected.ListenPort = &cfg.ListenPort
+	expected.ListenPort = cfg.ListenPort
 
 	// Hardcoded config
 	expected.RingConfig.ReplicationFactor = 1
diff --git a/pkg/ingester/ingester_test.go b/pkg/ingester/ingester_test.go
index f8c5528a22..9621687870 100644
--- a/pkg/ingester/ingester_test.go
+++ b/pkg/ingester/ingester_test.go
@@ -159,7 +159,7 @@ func runTestQueryTimes(ctx context.Context, t *testing.T, ing *Ingester, ty labe
 	return res, req, nil
 }
 
-func pushTestSamples(t *testing.T, ing *Ingester, numSeries, samplesPerSeries, offset int) ([]string, map[string]model.Matrix) {
+func pushTestSamples(t testing.TB, ing *Ingester, numSeries, samplesPerSeries, offset int) ([]string, map[string]model.Matrix) {
 	userIDs := []string{"1", "2", "3"}
 
 	// Create test samples.
diff --git a/pkg/ingester/lifecycle_test.go b/pkg/ingester/lifecycle_test.go
index fe0048d740..d55bded7ff 100644
--- a/pkg/ingester/lifecycle_test.go
+++ b/pkg/ingester/lifecycle_test.go
@@ -43,7 +43,7 @@ func defaultIngesterTestConfig() Config {
 	cfg.ConcurrentFlushes = 1
 	cfg.LifecyclerConfig.RingConfig.KVStore.Mock = consul
 	cfg.LifecyclerConfig.NumTokens = 1
-	cfg.LifecyclerConfig.ListenPort = func(i int) *int { return &i }(0)
+	cfg.LifecyclerConfig.ListenPort = 0
 	cfg.LifecyclerConfig.Addr = "localhost"
 	cfg.LifecyclerConfig.ID = "localhost"
 	cfg.LifecyclerConfig.FinalSleep = 0
diff --git a/pkg/ingester/wal_test.go b/pkg/ingester/wal_test.go
index 764db54580..edd9358491 100644
--- a/pkg/ingester/wal_test.go
+++ b/pkg/ingester/wal_test.go
@@ -183,3 +183,47 @@ func TestCheckpointRepair(t *testing.T) {
 	}
 
 }
+
+func BenchmarkWALReplay(b *testing.B) {
+	dirname, err := ioutil.TempDir("", "cortex-wal")
+	require.NoError(b, err)
+	defer func() {
+		require.NoError(b, os.RemoveAll(dirname))
+	}()
+
+	cfg := defaultIngesterTestConfig()
+	cfg.WALConfig.WALEnabled = true
+	cfg.WALConfig.CheckpointEnabled = true
+	cfg.WALConfig.Recover = true
+	cfg.WALConfig.Dir = dirname
+	cfg.WALConfig.CheckpointDuration = 100 * time.Minute
+	cfg.WALConfig.checkpointDuringShutdown = false
+
+	numSeries := 10
+	numSamplesPerSeriesPerPush := 2
+	numPushes := 100000
+
+	_, ing := newTestStore(b, cfg, defaultClientTestConfig(), defaultLimitsTestConfig(), nil)
+
+	// Add samples for the checkpoint.
+	for r := 0; r < numPushes; r++ {
+		_, _ = pushTestSamples(b, ing, numSeries, numSamplesPerSeriesPerPush, r*numSamplesPerSeriesPerPush)
+	}
+	w, ok := ing.wal.(*walWrapper)
+	require.True(b, ok)
+	require.NoError(b, w.performCheckpoint(true))
+
+	// Add samples for the additional WAL not in checkpoint.
+	for r := 0; r < numPushes; r++ {
+		_, _ = pushTestSamples(b, ing, numSeries, numSamplesPerSeriesPerPush, (numPushes+r)*numSamplesPerSeriesPerPush)
+	}
+
+	require.NoError(b, services.StopAndAwaitTerminated(context.Background(), ing))
+
+	var ing2 *Ingester
+	b.Run("wal replay", func(b *testing.B) {
+		// Replay will happen here.
+		_, ing2 = newTestStore(b, cfg, defaultClientTestConfig(), defaultLimitsTestConfig(), nil)
+	})
+	require.NoError(b, services.StopAndAwaitTerminated(context.Background(), ing2))
+}
diff --git a/pkg/ring/client/pool_test.go b/pkg/ring/client/pool_test.go
index 1757aff4ef..1c813999dc 100644
--- a/pkg/ring/client/pool_test.go
+++ b/pkg/ring/client/pool_test.go
@@ -13,7 +13,6 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/health/grpc_health_v1"
 
-	"github.com/cortexproject/cortex/pkg/ring"
 	"github.com/cortexproject/cortex/pkg/util/services"
 )
 
@@ -37,14 +36,6 @@ func (i mockClient) Watch(ctx context.Context, in *grpc_health_v1.HealthCheckReq
 	return nil, status.Error(codes.Unimplemented, "Watching is not supported")
 }
 
-type mockReadRing struct {
-	ring.ReadRing
-}
-
-func (mockReadRing) GetAll() (ring.ReplicationSet, error) {
-	return ring.ReplicationSet{}, nil
-}
-
 func TestHealthCheck(t *testing.T) {
 	tcs := []struct {
 		client   mockClient
@@ -81,7 +72,7 @@ func TestPoolCache(t *testing.T) {
 		CheckInterval:      10 * time.Second,
 	}
 
-	pool := NewPool("test", cfg, mockReadRing{}, factory, nil, log.NewNopLogger())
+	pool := NewPool("test", cfg, nil, factory, nil, log.NewNopLogger())
 	require.NoError(t, services.StartAndAwaitRunning(context.Background(), pool))
 	defer services.StopAndAwaitTerminated(context.Background(), pool) //nolint:errcheck
 
diff --git a/pkg/ring/client/ring_service_discovery_test.go b/pkg/ring/client/ring_service_discovery_test.go
new file mode 100644
index 0000000000..86b9969183
--- /dev/null
+++ b/pkg/ring/client/ring_service_discovery_test.go
@@ -0,0 +1,63 @@
+package client
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/cortexproject/cortex/pkg/ring"
+)
+
+func TestNewRingServiceDiscovery(t *testing.T) {
+	tests := map[string]struct {
+		ringReplicationSet ring.ReplicationSet
+		ringErr            error
+		expectedAddrs      []string
+		expectedErr        error
+	}{
+		"discovery failure": {
+			ringErr:     errors.New("mocked error"),
+			expectedErr: errors.New("mocked error"),
+		},
+		"empty replication set": {
+			ringReplicationSet: ring.ReplicationSet{
+				Ingesters: []ring.IngesterDesc{},
+			},
+			expectedAddrs: nil,
+		},
+		"replication containing some endpoints": {
+			ringReplicationSet: ring.ReplicationSet{
+				Ingesters: []ring.IngesterDesc{
+					{Addr: "1.1.1.1"},
+					{Addr: "2.2.2.2"},
+				},
+			},
+			expectedAddrs: []string{"1.1.1.1", "2.2.2.2"},
+		},
+	}
+
+	for testName, testData := range tests {
+		t.Run(testName, func(t *testing.T) {
+			r := &mockReadRing{}
+			r.mockedReplicationSet = testData.ringReplicationSet
+			r.mockedErr = testData.ringErr
+
+			d := NewRingServiceDiscovery(r)
+			addrs, err := d()
+			assert.Equal(t, testData.expectedErr, err)
+			assert.Equal(t, testData.expectedAddrs, addrs)
+		})
+	}
+}
+
+type mockReadRing struct {
+	ring.ReadRing
+
+	mockedReplicationSet ring.ReplicationSet
+	mockedErr            error
+}
+
+func (m *mockReadRing) GetAll() (ring.ReplicationSet, error) {
+	return m.mockedReplicationSet, m.mockedErr
+}
diff --git a/pkg/ring/lifecycler_test.go b/pkg/ring/lifecycler_test.go
index cb689b1e5b..643629cb62 100644
--- a/pkg/ring/lifecycler_test.go
+++ b/pkg/ring/lifecycler_test.go
@@ -34,7 +34,7 @@ func testLifecyclerConfig(ringConfig Config, id string) LifecyclerConfig {
 	flagext.DefaultValues(&lifecyclerConfig)
 	lifecyclerConfig.Addr = "0.0.0.0"
 	lifecyclerConfig.Port = 1
-	lifecyclerConfig.ListenPort = func(i int) *int { return &i }(0)
+	lifecyclerConfig.ListenPort = 0
 	lifecyclerConfig.RingConfig = ringConfig
 	lifecyclerConfig.NumTokens = 1
 	lifecyclerConfig.ID = id