Repository: cortexproject/cortex (go)

Issue:
compactor went into tailspin
**Describe the bug**

Consul was down at the time compactor started, and it never recovered:

```
level=info ts=2021-03-04T20:23:01.350819279Z caller=main.go:188 msg="Starting Cortex" version="(version=1.6.0, branch=master, revision=56f794d)"
level=info ts=2021-03-04T20:23:01.35627101Z caller=module_service.go:59 msg=initialising module=server
level=info ts=2021-03-04T20:23:01.358032071Z caller=module_service.go:59 msg=initialising module=compactor
level=info ts=2021-03-04T20:23:01.351964058Z caller=server.go:229 http=[::]:80 grpc=[::]:9095 msg="server listening on addresses"
level=info ts=2021-03-04T20:23:01.356491008Z caller=module_service.go:59 msg=initialising module=memberlist-kv
level=info ts=2021-03-04T20:23:01.36340712Z caller=compactor.go:373 component=compactor msg="waiting until compactor is ACTIVE in the ring"
level=info ts=2021-03-04T20:23:01.366220882Z caller=lifecycler.go:527 msg="not loading tokens from file, tokens file path is empty"
level=error ts=2021-03-04T20:23:01.391200499Z caller=client.go:147 msg="error getting key" key=compactor err="Get \"http://consul.cortex.svc.cluster.local:8500/v1/kv/compactor\": dial tcp: lookup consul.cortex.svc.cluster.local on 10.0.0.10:53: no such host"
level=error ts=2021-03-04T20:23:01.394262836Z caller=client.go:147 msg="error getting key" key=compactor err="Get \"http://consul.cortex.svc.cluster.local:8500/v1/kv/compactor\": dial tcp: lookup consul.cortex.svc.cluster.local on 10.0.0.10:53: no such host"
level=error ts=2021-03-04T20:23:01.391467754Z caller=client.go:234 msg="error getting path" key=compactor err="Get \"http://consul.cortex.svc.cluster.local:8500/v1/kv/compactor?stale=&wait=10000ms\": dial tcp: lookup consul.cortex.svc.cluster.local on 10.0.0.10:53: no such host"
level=error ts=2021-03-04T20:23:01.399955867Z caller=client.go:147 msg="error getting key" key=compactor err="Get \"http://consul.cortex.svc.cluster.local:8500/v1/kv/compactor\": dial tcp: lookup consul.cortex.svc.cluster.local on 10.0.0.10:53: no such host"
level=error ts=2021-03-04T20:23:01.397642587Z caller=client.go:147 msg="error getting key" key=compactor err="Get \"http://consul.cortex.svc.cluster.local:8500/v1/kv/compactor\": dial tcp: lookup consul.cortex.svc.cluster.local on 10.0.0.10:53: no such host"
level=error ts=2021-03-04T20:23:01.402454173Z caller=client.go:147 msg="error getting key" key=compactor err="Get \"http://consul.cortex.svc.cluster.local:8500/v1/kv/compactor\": dial tcp: lookup consul.cortex.svc.cluster.local on 10.0.0.10:53: no such host"
level=error ts=2021-03-04T20:23:01.404896477Z caller=client.go:147 msg="error getting key" key=compactor err="Get \"http://consul.cortex.svc.cluster.local:8500/v1/kv/compactor\": dial tcp: lookup consul.cortex.svc.cluster.local on 10.0.0.10:53: no such host"
level=error ts=2021-03-04T20:23:01.407444107Z caller=client.go:147 msg="error getting key" key=compactor err="Get \"http://consul.cortex.svc.cluster.local:8500/v1/kv/compactor\": dial tcp: lookup consul.cortex.svc.cluster.local on 10.0.0.10:53: no such host"
level=error ts=2021-03-04T20:23:01.410901197Z caller=client.go:147 msg="error getting key" key=compactor err="Get \"http://consul.cortex.svc.cluster.local:8500/v1/kv/compactor\": dial tcp: lookup consul.cortex.svc.cluster.local on 10.0.0.10:53: no such host"
level=error ts=2021-03-04T20:23:01.413247689Z caller=client.go:147 msg="error getting key" key=compactor err="Get \"http://consul.cortex.svc.cluster.local:8500/v1/kv/compactor\": dial tcp: lookup consul.cortex.svc.cluster.local on 10.0.0.10:53: no such host"
level=error ts=2021-03-04T20:23:01.4157827Z caller=client.go:147 msg="error getting key" key=compactor err="Get \"http://consul.cortex.svc.cluster.local:8500/v1/kv/compactor\": dial tcp: lookup consul.cortex.svc.cluster.local on 10.0.0.10:53: no such host"
level=error ts=2021-03-04T20:23:02.848432818Z caller=client.go:234 msg="error getting path" key=compactor err="Get \"http://consul.cortex.svc.cluster.local:8500/v1/kv/compactor?stale=&wait=10000ms\": dial tcp: lookup consul.cortex.svc.cluster.local on 10.0.0.10:53: no such host"
level=error ts=2021-03-04T20:23:06.895736201Z caller=client.go:234 msg="error getting path" key=compactor err="Get \"http://consul.cortex.svc.cluster.local:8500/v1/kv/compactor?stale=&wait=10000ms\": dial tcp: lookup consul.cortex.svc.cluster.local on 10.0.0.10:53: no such host"
level=error ts=2021-03-04T20:23:14.397161217Z caller=client.go:234 msg="error getting path" key=compactor err="Get \"http://consul.cortex.svc.cluster.local:8500/v1/kv/compactor?stale=&wait=10000ms\": dial tcp: lookup consul.cortex.svc.cluster.local on 10.0.0.10:53: no such host"
level=error ts=2021-03-04T20:23:25.078769837Z caller=client.go:234 msg="error getting path" key=compactor err="Get \"http://consul.cortex.svc.cluster.local:8500/v1/kv/compactor?stale=&wait=10000ms\": dial tcp: lookup consul.cortex.svc.cluster.local on 10.0.0.10:53: no such host"
level=error ts=2021-03-04T20:23:49.041899802Z caller=client.go:234 msg="error getting path" key=compactor err="Get \"http://consul.cortex.svc.cluster.local:8500/v1/kv/compactor?stale=&wait=10000ms\": dial tcp: lookup consul.cortex.svc.cluster.local on 10.0.0.10:53: no such host"
level=error ts=2021-03-04T20:24:35.634427793Z caller=client.go:234 msg="error getting path" key=compactor err="Get \"http://consul.cortex.svc.cluster.local:8500/v1/kv/compactor?stale=&wait=10000ms\": dial tcp: lookup consul.cortex.svc.cluster.local on 10.0.0.10:53: no such host"
level=error ts=2021-03-04T20:25:35.127795679Z caller=client.go:234 msg="error getting path" key=compactor err="Get \"http://consul.cortex.svc.cluster.local:8500/v1/kv/compactor?stale=&wait=10000ms\": dial tcp: lookup consul.cortex.svc.cluster.local on 10.0.0.10:53: no such host"
level=error ts=2021-03-04T20:26:37.176005708Z caller=client.go:234 msg="error getting path" key=compactor err="Unexpected response code: 500"
level=error ts=2021-03-04T20:27:42.508689928Z caller=client.go:234 msg="error getting path" key=compactor err="Unexpected response code: 500"
level=error ts=2021-03-04T20:28:35.350979964Z caller=client.go:234 msg="error getting path" key=compactor err="Unexpected response code: 500"
level=error ts=2021-03-04T20:29:41.117751261Z caller=client.go:234 msg="error getting path" key=compactor err="Unexpected response code: 500"
level=error ts=2021-03-04T20:30:39.87687265Z caller=client.go:234 msg="error getting path" key=compactor err="Unexpected response code: 500"
level=info ts=2021-03-04T20:31:36.036454597Z caller=client.go:247 msg="value is nil" key=compactor index=22
level=info ts=2021-03-04T20:31:36.930199939Z caller=client.go:247 msg="value is nil" key=compactor index=24
level=info ts=2021-03-04T20:31:37.932395975Z caller=client.go:247 msg="value is nil" key=compactor index=25
level=info ts=2021-03-04T20:31:41.887641219Z caller=client.go:247 msg="value is nil" key=compactor index=28
level=info ts=2021-03-04T20:31:41.941646251Z caller=client.go:247 msg="value is nil" key=compactor index=29
level=info ts=2021-03-04T20:31:46.926493102Z caller=client.go:247 msg="value is nil" key=compactor index=31
level=info ts=2021-03-04T20:31:46.967013886Z caller=client.go:247 msg="value is nil" key=compactor index=32
...
level=info ts=2021-03-05T10:12:35.156472897Z caller=client.go:247 msg="value is nil" key=compactor index=78733
level=info ts=2021-03-05T10:12:35.455228914Z caller=client.go:247 msg="value is nil" key=compactor index=78734
level=info ts=2021-03-05T10:12:36.157313862Z caller=client.go:247 msg="value is nil" key=compactor index=78736
level=info ts=2021-03-05T10:12:37.157305834Z caller=client.go:247 msg="value is nil" key=compactor index=78738
```

We have compactor sharding turned on:
```
        - -compactor.ring.consul.hostname=consul.cortex.svc.cluster.local:8500
        - -compactor.ring.prefix=
        - -compactor.ring.store=consul
        - -compactor.sharding-enabled=true
```

**Expected behavior**
I think it should exit with error in this situation; crashlooping would make the fault more obvious to the operator, and after a few restarts it would have managed to talk to Consul in my case.



Pull request description:
Add timeout for waiting on compactor to become ACTIVE in the ring.
Signed-off-by: Albert <ac1214@users.noreply.github.com>

**What this PR does**:
Addresses #3914
**Which issue(s) this PR fixes**:
Fixes #3914

**Checklist**
- [x] Tests updated
- [x] Documentation added
- [x] `CHANGELOG.md` updated - the order of entries should be `[CHANGE]`, `[FEATURE]`, `[ENHANCEMENT]`, `[BUGFIX]`


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 00ff7a81c2..2f825f9bf2 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -3,6 +3,7 @@
 ## master / unreleased
 
 * [CHANGE] Querier / ruler: Change `-querier.max-fetched-chunks-per-query` configuration to limit to maximum number of chunks that can be fetched in a single query. The number of chunks fetched by ingesters AND long-term storare combined should not exceed the value configured on `-querier.max-fetched-chunks-per-query`. #4260
+* [ENHANCEMENT] Add timeout for waiting on compactor to become ACTIVE in the ring. #4262
 * [BUGFIX] HA Tracker: when cleaning up obsolete elected replicas from KV store, tracker didn't update number of cluster per user correctly. #4336
 
 ## 1.10.0-rc.0 / 2021-06-28
diff --git a/docs/blocks-storage/compactor.md b/docs/blocks-storage/compactor.md
index 213405822c..280545afab 100644
--- a/docs/blocks-storage/compactor.md
+++ b/docs/blocks-storage/compactor.md
@@ -230,4 +230,8 @@ compactor:
     # Name of network interface to read address from.
     # CLI flag: -compactor.ring.instance-interface-names
     [instance_interface_names: <list of string> | default = [eth0 en0]]
+
+    # Timeout for waiting on compactor to become ACTIVE in the ring.
+    # CLI flag: -compactor.ring.wait-active-instance-timeout
+    [wait_active_instance_timeout: <duration> | default = 10m]
 ```
diff --git a/docs/configuration/config-file-reference.md b/docs/configuration/config-file-reference.md
index 9b20d81bea..0267d01e88 100644
--- a/docs/configuration/config-file-reference.md
+++ b/docs/configuration/config-file-reference.md
@@ -5195,6 +5195,10 @@ sharding_ring:
   # Name of network interface to read address from.
   # CLI flag: -compactor.ring.instance-interface-names
   [instance_interface_names: <list of string> | default = [eth0 en0]]
+
+  # Timeout for waiting on compactor to become ACTIVE in the ring.
+  # CLI flag: -compactor.ring.wait-active-instance-timeout
+  [wait_active_instance_timeout: <duration> | default = 10m]
 ```
 
 ### `store_gateway_config`
diff --git a/pkg/compactor/compactor.go b/pkg/compactor/compactor.go
index c41f7a9ccc..0af2088bd5 100644
--- a/pkg/compactor/compactor.go
+++ b/pkg/compactor/compactor.go
@@ -393,7 +393,11 @@ func (c *Compactor) starting(ctx context.Context) error {
 		// users scanner depends on the ring (to check whether an user belongs
 		// to this shard or not).
 		level.Info(c.logger).Log("msg", "waiting until compactor is ACTIVE in the ring")
-		if err := ring.WaitInstanceState(ctx, c.ring, c.ringLifecycler.ID, ring.ACTIVE); err != nil {
+
+		ctxWithTimeout, cancel := context.WithTimeout(ctx, c.compactorCfg.ShardingRing.WaitActiveInstanceTimeout)
+		defer cancel()
+		if err := ring.WaitInstanceState(ctxWithTimeout, c.ring, c.ringLifecycler.ID, ring.ACTIVE); err != nil {
+			level.Error(c.logger).Log("msg", "compactor failed to become ACTIVE in the ring", "err", err)
 			return err
 		}
 		level.Info(c.logger).Log("msg", "compactor is ACTIVE in the ring")
diff --git a/pkg/compactor/compactor_ring.go b/pkg/compactor/compactor_ring.go
index 74c3a77ad3..5b55c8a871 100644
--- a/pkg/compactor/compactor_ring.go
+++ b/pkg/compactor/compactor_ring.go
@@ -34,6 +34,10 @@ type RingConfig struct {
 
 	// Injected internally
 	ListenPort int `yaml:"-"`
+
+	WaitActiveInstanceTimeout time.Duration `yaml:"wait_active_instance_timeout"`
+
+	ObservePeriod time.Duration `yaml:"-"`
 }
 
 // RegisterFlags adds the flags required to config this to the given FlagSet
@@ -59,6 +63,9 @@ func (cfg *RingConfig) RegisterFlags(f *flag.FlagSet) {
 	f.StringVar(&cfg.InstanceAddr, "compactor.ring.instance-addr", "", "IP address to advertise in the ring.")
 	f.IntVar(&cfg.InstancePort, "compactor.ring.instance-port", 0, "Port to advertise in the ring (defaults to server.grpc-listen-port).")
 	f.StringVar(&cfg.InstanceID, "compactor.ring.instance-id", hostname, "Instance ID to register in the ring.")
+
+	// Timeout durations
+	f.DurationVar(&cfg.WaitActiveInstanceTimeout, "compactor.ring.wait-active-instance-timeout", 10*time.Minute, "Timeout for waiting on compactor to become ACTIVE in the ring.")
 }
 
 // ToLifecyclerConfig returns a LifecyclerConfig based on the compactor
@@ -87,7 +94,7 @@ func (cfg *RingConfig) ToLifecyclerConfig() ring.LifecyclerConfig {
 	lc.InfNames = cfg.InstanceInterfaceNames
 	lc.UnregisterOnShutdown = true
 	lc.HeartbeatPeriod = cfg.HeartbeatPeriod
-	lc.ObservePeriod = 0
+	lc.ObservePeriod = cfg.ObservePeriod
 	lc.JoinAfter = 0
 	lc.MinReadyDuration = 0
 	lc.FinalSleep = 0
diff --git a/pkg/ring/ring.go b/pkg/ring/ring.go
index 539ec65e75..8d5bd0c925 100644
--- a/pkg/ring/ring.go
+++ b/pkg/ring/ring.go
@@ -71,6 +71,10 @@ type ReadRing interface {
 	// and size (number of instances).
 	ShuffleShard(identifier string, size int) ReadRing
 
+	// GetInstanceState returns the current state of an instance or an error if the
+	// instance does not exist in the ring.
+	GetInstanceState(instanceID string) (InstanceState, error)
+
 	// ShuffleShardWithLookback is like ShuffleShard() but the returned subring includes
 	// all instances that have been part of the identifier's shard since "now - lookbackPeriod".
 	ShuffleShardWithLookback(identifier string, size int, lookbackPeriod time.Duration, now time.Time) ReadRing
diff --git a/pkg/ring/util.go b/pkg/ring/util.go
index ac5c27388c..06e053dd26 100644
--- a/pkg/ring/util.go
+++ b/pkg/ring/util.go
@@ -69,7 +69,7 @@ func GetInstancePort(configPort, listenPort int) int {
 
 // WaitInstanceState waits until the input instanceID is registered within the
 // ring matching the provided state. A timeout should be provided within the context.
-func WaitInstanceState(ctx context.Context, r *Ring, instanceID string, state InstanceState) error {
+func WaitInstanceState(ctx context.Context, r ReadRing, instanceID string, state InstanceState) error {
 	backoff := util.NewBackoff(ctx, util.BackoffConfig{
 		MinBackoff: 100 * time.Millisecond,
 		MaxBackoff: time.Second,


Test patch:
diff --git a/pkg/compactor/compactor_test.go b/pkg/compactor/compactor_test.go
index 9c14cb0924..da1a83cead 100644
--- a/pkg/compactor/compactor_test.go
+++ b/pkg/compactor/compactor_test.go
@@ -1092,6 +1092,9 @@ func prepareConfig() Config {
 	compactorCfg.ShardingRing.WaitStabilityMinDuration = 0
 	compactorCfg.ShardingRing.WaitStabilityMaxDuration = 0
 
+	// Set lower timeout for waiting on compactor to become ACTIVE in the ring for unit tests
+	compactorCfg.ShardingRing.WaitActiveInstanceTimeout = 5 * time.Second
+
 	return compactorCfg
 }
 
@@ -1279,3 +1282,33 @@ func TestCompactor_DeleteLocalSyncFiles(t *testing.T) {
 	require.NotEqual(t, numUsers, c1Users)
 	require.Equal(t, numUsers, c1Users+c2Users)
 }
+
+func TestCompactor_ShouldFailCompactionOnTimeout(t *testing.T) {
+	t.Parallel()
+
+	// Mock the bucket
+	bucketClient := &bucket.ClientMock{}
+	bucketClient.MockIter("", []string{}, nil)
+
+	cfg := prepareConfig()
+	cfg.ShardingEnabled = true
+	cfg.ShardingRing.InstanceID = "compactor-1"
+	cfg.ShardingRing.InstanceAddr = "1.2.3.4"
+	cfg.ShardingRing.KVStore.Mock = consul.NewInMemoryClient(ring.GetCodec())
+
+	// Set ObservePeriod to longer than the timeout period to mock a timeout while waiting on ring to become ACTIVE
+	cfg.ShardingRing.ObservePeriod = time.Second * 10
+
+	c, _, _, logs, _ := prepare(t, cfg, bucketClient)
+
+	// Try to start the compactor with a bad consul kv-store. The
+	err := services.StartAndAwaitRunning(context.Background(), c)
+
+	// Assert that the compactor timed out
+	assert.Equal(t, context.DeadlineExceeded, err)
+
+	assert.ElementsMatch(t, []string{
+		`level=info component=compactor msg="waiting until compactor is ACTIVE in the ring"`,
+		`level=error component=compactor msg="compactor failed to become ACTIVE in the ring" err="context deadline exceeded"`,
+	}, removeIgnoredLogs(strings.Split(strings.TrimSpace(logs.String()), "\n")))
+}
diff --git a/pkg/ring/util_test.go b/pkg/ring/util_test.go
index c3a3e037e9..9a4a69c690 100644
--- a/pkg/ring/util_test.go
+++ b/pkg/ring/util_test.go
@@ -6,10 +6,65 @@ import (
 	"testing"
 	"time"
 
+	"github.com/pkg/errors"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 )
 
+type RingMock struct {
+	mock.Mock
+}
+
+func (r *RingMock) Collect(ch chan<- prometheus.Metric) {}
+
+func (r *RingMock) Describe(ch chan<- *prometheus.Desc) {}
+
+func (r *RingMock) Get(key uint32, op Operation, bufDescs []InstanceDesc, bufHosts, bufZones []string) (ReplicationSet, error) {
+	args := r.Called(key, op, bufDescs, bufHosts, bufZones)
+	return args.Get(0).(ReplicationSet), args.Error(1)
+}
+
+func (r *RingMock) GetAllHealthy(op Operation) (ReplicationSet, error) {
+	args := r.Called(op)
+	return args.Get(0).(ReplicationSet), args.Error(1)
+}
+
+func (r *RingMock) GetReplicationSetForOperation(op Operation) (ReplicationSet, error) {
+	args := r.Called(op)
+	return args.Get(0).(ReplicationSet), args.Error(1)
+}
+
+func (r *RingMock) ReplicationFactor() int {
+	return 0
+}
+
+func (r *RingMock) InstancesCount() int {
+	return 0
+}
+
+func (r *RingMock) ShuffleShard(identifier string, size int) ReadRing {
+	args := r.Called(identifier, size)
+	return args.Get(0).(ReadRing)
+}
+
+func (r *RingMock) GetInstanceState(instanceID string) (InstanceState, error) {
+	args := r.Called(instanceID)
+	return args.Get(0).(InstanceState), args.Error(1)
+}
+
+func (r *RingMock) ShuffleShardWithLookback(identifier string, size int, lookbackPeriod time.Duration, now time.Time) ReadRing {
+	args := r.Called(identifier, size, lookbackPeriod, now)
+	return args.Get(0).(ReadRing)
+}
+
+func (r *RingMock) HasInstance(instanceID string) bool {
+	return true
+}
+
+func (r *RingMock) CleanupShuffleShardCache(identifier string) {}
+
 func TestGenerateTokens(t *testing.T) {
 	tokens := GenerateTokens(1000000, nil)
 
@@ -184,3 +239,63 @@ func TestWaitRingStabilityShouldReturnErrorIfMaxWaitingIsReached(t *testing.T) {
 
 	assert.InDelta(t, maxWaiting, elapsedTime, float64(2*time.Second))
 }
+
+func TestWaitInstanceStateTimeout(t *testing.T) {
+	t.Parallel()
+
+	const (
+		instanceID      = "test"
+		timeoutDuration = time.Second
+	)
+
+	ctx, cancel := context.WithTimeout(context.Background(), timeoutDuration)
+	defer cancel()
+
+	ring := &RingMock{}
+	ring.On("GetInstanceState", mock.Anything, mock.Anything).Return(ACTIVE, nil)
+
+	err := WaitInstanceState(ctx, ring, instanceID, PENDING)
+
+	assert.Equal(t, context.DeadlineExceeded, err)
+	ring.AssertCalled(t, "GetInstanceState", instanceID)
+}
+
+func TestWaitInstanceStateTimeoutOnError(t *testing.T) {
+	t.Parallel()
+
+	const (
+		instanceID      = "test"
+		timeoutDuration = time.Second
+	)
+
+	ctx, cancel := context.WithTimeout(context.Background(), timeoutDuration)
+	defer cancel()
+
+	ring := &RingMock{}
+	ring.On("GetInstanceState", mock.Anything, mock.Anything).Return(PENDING, errors.New("instance not found in the ring"))
+
+	err := WaitInstanceState(ctx, ring, instanceID, ACTIVE)
+
+	assert.Equal(t, context.DeadlineExceeded, err)
+	ring.AssertCalled(t, "GetInstanceState", instanceID)
+}
+
+func TestWaitInstanceStateExitsAfterActualStateEqualsState(t *testing.T) {
+	t.Parallel()
+
+	const (
+		instanceID      = "test"
+		timeoutDuration = time.Second
+	)
+
+	ctx, cancel := context.WithTimeout(context.Background(), timeoutDuration)
+	defer cancel()
+
+	ring := &RingMock{}
+	ring.On("GetInstanceState", mock.Anything, mock.Anything).Return(ACTIVE, nil)
+
+	err := WaitInstanceState(ctx, ring, instanceID, ACTIVE)
+
+	assert.Nil(t, err)
+	ring.AssertNumberOfCalls(t, "GetInstanceState", 1)
+}