Repository: cortexproject/cortex
Pull request #2494 (2020-04-21 09:51:00)

return directories in respose from List calls to object stores
<!--  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**:
Adds list of directories(synthetic directories in object stores like S3, GCS and Azure) in the response from List calls to `chunk.ObjectStore`.
This would be helpful in custom `TableClient` that I need to implement in Loki for BoltDB Shipper.
It would be used in `TableClient.ListTables` call for listing all the directories that are there in the store.

**Checklist**
- [ ] Tests updated
- [ ] Documentation added
- [ ] `CHANGELOG.md` updated - the order of entries should be `[CHANGE]`, `[FEATURE]`, `[ENHANCEMENT]`, `[BUGFIX]`


gouthamve: Can't the client generate the list of directories from the full object names? I'm asking because I think this will deviate from the Thanos Object Store implementation which we're looking to implement. I might be wrong though.

Just curious why we need the directories and if we can't get them by looping through the returned objects.
sandeepsukhani: > Can't the client generate the list of directories from the full object names?
It is not a recursive list operation so I guess we can't. It just lists the immediate children so I am not sure how client will generate the list of directories.
If I am not wrong Thanos ObjectStore has iterator pattern and it includes both the objects and directories in that.

Patch:
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/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/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/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/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/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