Repository: cortexproject/cortex
Pull request #2491 (2020-04-21 06:52:52)

Change LifecyclerConfig.ListenPort from pointer to int
**What this PR does**:
Looking at the current state of the code (including Loki) I can't see a good reason why `LifecyclerConfig.ListenPort` should be a pointer instead of `int`. In this PR I'm proposing to switch it to `int` to cleanup the config.

**Which issue(s) this PR fixes**:
N/A

**Checklist**
- [ ] Tests updated
- [ ] Documentation added
- [ ] `CHANGELOG.md` updated - the order of entries should be `[CHANGE]`, `[FEATURE]`, `[ENHANCEMENT]`, `[BUGFIX]`



Patch:
diff --git a/Makefile b/Makefile
index a54cd87630..0e5597cb58 100644
--- a/Makefile
+++ b/Makefile
@@ -138,7 +138,7 @@ lint:
 	golangci-lint run
 
 	# Ensure no blacklisted package is imported.
-	faillint -paths "github.com/bmizerany/assert=github.com/stretchr/testify/assert" ./pkg/... ./cmd/... ./tools/... ./integration/...
+	faillint -paths "github.com/bmizerany/assert=github.com/stretchr/testify/assert,golang.org/x/net/context=context" ./pkg/... ./cmd/... ./tools/... ./integration/...
 
 	# Validate Kubernetes spec files. Requires:
 	# https://kubeval.instrumenta.dev
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/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


Test patch:
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