Add compactor auto-forget from ring on unhealthy (#6563)

* Add metric name in limiter per-metric exceeded errors

Signed-off-by: Essam Eldaly <[email protected]>

* Remove unused labels from errors

Signed-off-by: Essam Eldaly <[email protected]>

* fmt

Signed-off-by: Essam Eldaly <[email protected]>

* Add  for ring to auto forget unhealthy compactors

Signed-off-by: Essam Eldaly <[email protected]>

* Changed LifecyclerDelegate implementation to more closely match BasicLifecyclerDelegate

Signed-off-by: Essam Eldaly <[email protected]>

* Changelog fix

Signed-off-by: Essam Eldaly <[email protected]>

* Changelog fix

Signed-off-by: Essam Eldaly <[email protected]>

* auto forget compactor default set to 2*HeartbeatTimeout

Signed-off-by: Essam Eldaly <[email protected]>

* added compactor auto forget test

Signed-off-by: Essam Eldaly <[email protected]>

* gofmt

Signed-off-by: Essam Eldaly <[email protected]>

* forget compactor test race condition fix

Signed-off-by: Essam Eldaly <[email protected]>

* gofmt

Signed-off-by: Essam Eldaly <[email protected]>

* check-doc

Signed-off-by: Essam Eldaly <[email protected]>

* removed sleep

Signed-off-by: Essam Eldaly <[email protected]>

* removed no error check on compactor stop in auto forget test

Signed-off-by: Essam Eldaly <[email protected]>

* fixed auto forget compactor test

Signed-off-by: Essam Eldaly <[email protected]>

* fixed auto forget compactor test

Signed-off-by: Essam Eldaly <[email protected]>

---------

Signed-off-by: Essam Eldaly <[email protected]>
Signed-off-by: Ben Ye <[email protected]>
Co-authored-by: Ben Ye <[email protected]>

Author: eeldaly

CHANGELOG.md

@@ -1,6 +1,7 @@
 # Changelog
 
 ## master / unreleased
+* [ENHANCEMENT] Add `compactor.auto-forget-delay` for compactor to auto forget compactors after X minutes without heartbeat. #6533
 
 * [FEATURE] Querier/Ruler: Add `query_partial_data` and `rules_partial_data` limits to allow queries/rules to be evaluated with data from a single zone, if other zones are not available. #6526
 * [ENHANCEMENT] StoreGateway: Emit more histogram buckets on the `cortex_querier_storegateway_refetches_per_query` metric. #6570

docs/blocks-storage/compactor.md

@@ -263,6 +263,11 @@ compactor:
     # CLI flag: -compactor.ring.heartbeat-timeout
     [heartbeat_timeout: <duration> | default = 1m]
 
+    # Time since last heartbeat before compactor will be removed from ring. 0 to
+    # disable
+    # CLI flag: -compactor.auto-forget-delay
+    [auto_forget_delay: <duration> | default = 2m]
+
     # Minimum time to wait for ring stability at startup. 0 to disable.
     # CLI flag: -compactor.ring.wait-stability-min-duration
     [wait_stability_min_duration: <duration> | default = 1m]

docs/configuration/config-file-reference.md

@@ -2328,6 +2328,11 @@ sharding_ring:
   # CLI flag: -compactor.ring.heartbeat-timeout
   [heartbeat_timeout: <duration> | default = 1m]
 
+  # Time since last heartbeat before compactor will be removed from ring. 0 to
+  # disable
+  # CLI flag: -compactor.auto-forget-delay
+  [auto_forget_delay: <duration> | default = 2m]
+
   # Minimum time to wait for ring stability at startup. 0 to disable.
   # CLI flag: -compactor.ring.wait-stability-min-duration
   [wait_stability_min_duration: <duration> | default = 1m]

pkg/compactor/compactor.go

@@ -656,7 +656,12 @@ func (c *Compactor) starting(ctx context.Context) error {
 	// Initialize the compactors ring if sharding is enabled.
 	if c.compactorCfg.ShardingEnabled {
 		lifecyclerCfg := c.compactorCfg.ShardingRing.ToLifecyclerConfig()
-		c.ringLifecycler, err = ring.NewLifecycler(lifecyclerCfg, ring.NewNoopFlushTransferer(), "compactor", ringKey, true, false, c.logger, prometheus.WrapRegistererWithPrefix("cortex_", c.registerer))
+		var delegate ring.LifecyclerDelegate
+		delegate = &ring.DefaultLifecyclerDelegate{}
+		if c.compactorCfg.ShardingRing.AutoForgetDelay > 0 {
+			delegate = ring.NewLifecyclerAutoForgetDelegate(c.compactorCfg.ShardingRing.AutoForgetDelay, delegate, c.logger)
+		}
+		c.ringLifecycler, err = ring.NewLifecyclerWithDelegate(lifecyclerCfg, ring.NewNoopFlushTransferer(), "compactor", ringKey, true, false, c.logger, prometheus.WrapRegistererWithPrefix("cortex_", c.registerer), delegate)
 		if err != nil {
 			return errors.Wrap(err, "unable to initialize compactor ring lifecycler")
 		}

pkg/compactor/compactor_ring.go

@@ -21,6 +21,7 @@ type RingConfig struct {
 	KVStore          kv.Config     `yaml:"kvstore"`
 	HeartbeatPeriod  time.Duration `yaml:"heartbeat_period"`
 	HeartbeatTimeout time.Duration `yaml:"heartbeat_timeout"`
+	AutoForgetDelay  time.Duration `yaml:"auto_forget_delay"`
 
 	// Wait ring stability.
 	WaitStabilityMinDuration time.Duration `yaml:"wait_stability_min_duration"`
@@ -54,6 +55,7 @@ func (cfg *RingConfig) RegisterFlags(f *flag.FlagSet) {
 	cfg.KVStore.RegisterFlagsWithPrefix("compactor.ring.", "collectors/", f)
 	f.DurationVar(&cfg.HeartbeatPeriod, "compactor.ring.heartbeat-period", 5*time.Second, "Period at which to heartbeat to the ring. 0 = disabled.")
 	f.DurationVar(&cfg.HeartbeatTimeout, "compactor.ring.heartbeat-timeout", time.Minute, "The heartbeat timeout after which compactors are considered unhealthy within the ring. 0 = never (timeout disabled).")
+	f.DurationVar(&cfg.AutoForgetDelay, "compactor.auto-forget-delay", 2*cfg.HeartbeatTimeout, "Time since last heartbeat before compactor will be removed from ring. 0 to disable")
 
 	// Wait stability flags.
 	f.DurationVar(&cfg.WaitStabilityMinDuration, "compactor.ring.wait-stability-min-duration", time.Minute, "Minimum time to wait for ring stability at startup. 0 to disable.")

pkg/compactor/compactor_test.go

@@ -2113,3 +2113,74 @@ func TestCompactor_FailedWithHaltError(t *testing.T) {
 		"cortex_compactor_compaction_error_total",
 	))
 }
+
+func TestCompactor_RingLifecyclerShouldAutoForgetUnhealthyInstances(t *testing.T) {
+	// Setup user IDs
+	userID := "user-0"
+	inmem := objstore.WithNoopInstr(objstore.NewInMemBucket())
+
+	id, err := ulid.New(ulid.Now(), rand.Reader)
+	require.NoError(t, err)
+	require.NoError(t, inmem.Upload(context.Background(), userID+"/"+id.String()+"/meta.json", strings.NewReader(mockBlockMetaJSON(id.String()))))
+
+	// Create a shared KV Store
+	kvstore, closer := consul.NewInMemoryClient(ring.GetCodec(), log.NewNopLogger(), nil)
+	t.Cleanup(func() { assert.NoError(t, closer.Close()) })
+
+	// Create two compactors
+	var compactors []*Compactor
+
+	for i := 0; i < 2; i++ {
+		// Setup config
+		cfg := prepareConfig()
+
+		cfg.ShardingEnabled = true
+		cfg.ShardingRing.InstanceID = fmt.Sprintf("compactor-%d", i)
+		cfg.ShardingRing.InstanceAddr = fmt.Sprintf("127.0.0.%d", i)
+		cfg.ShardingRing.WaitStabilityMinDuration = time.Second
+		cfg.ShardingRing.WaitStabilityMaxDuration = 5 * time.Second
+		cfg.ShardingRing.KVStore.Mock = kvstore
+		cfg.ShardingRing.HeartbeatPeriod = 200 * time.Millisecond
+		cfg.ShardingRing.UnregisterOnShutdown = false
+		cfg.ShardingRing.AutoForgetDelay = 400 * time.Millisecond
+
+		// Compactor will get its own temp dir for storing local files.
+		compactor, _, tsdbPlanner, _, _ := prepare(t, cfg, inmem, nil)
+		compactor.logger = log.NewNopLogger()
+
+		compactors = append(compactors, compactor)
+
+		// Mock the planner as if there's no compaction to do,
+		// in order to simplify tests (all in all, we just want to
+		// test our logic and not TSDB compactor which we expect to
+		// be already tested).
+		tsdbPlanner.On("Plan", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return([]*metadata.Meta{}, nil)
+	}
+
+	require.Equal(t, 2, len(compactors))
+	compactor := compactors[0]
+	compactor2 := compactors[1]
+
+	// Start compactors
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), compactor))
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), compactor2))
+
+	// Wait until a run has completed.
+	cortex_testutil.Poll(t, 20*time.Second, 1.0, func() interface{} {
+		return prom_testutil.ToFloat64(compactor2.CompactionRunsCompleted)
+	})
+
+	cortex_testutil.Poll(t, 5000*time.Millisecond, true, func() interface{} {
+		healthy, unhealthy, _ := compactor.ring.GetAllInstanceDescs(ring.Reporting)
+		return len(healthy) == 2 && len(unhealthy) == 0
+	})
+
+	// Make one compactor unhealthy in ring by stopping the
+	// compactor service while UnregisterOnShutdown is false
+	require.NoError(t, services.StopAndAwaitTerminated(context.Background(), compactor2))
+
+	cortex_testutil.Poll(t, 5000*time.Millisecond, true, func() interface{} {
+		healthy, unhealthy, _ := compactor.ring.GetAllInstanceDescs(ring.Reporting)
+		return len(healthy) == 1 && len(unhealthy) == 0
+	})
+}

pkg/ring/basic_lifecycler_delegates.go

@@ -141,14 +141,6 @@ func (d *AutoForgetDelegate) OnRingInstanceStopping(lifecycler *BasicLifecycler)
 }
 
 func (d *AutoForgetDelegate) OnRingInstanceHeartbeat(lifecycler *BasicLifecycler, ringDesc *Desc, instanceDesc *InstanceDesc) {
-	for id, instance := range ringDesc.Ingesters {
-		lastHeartbeat := time.Unix(instance.GetTimestamp(), 0)
-
-		if time.Since(lastHeartbeat) > d.forgetPeriod {
-			level.Warn(d.logger).Log("msg", "auto-forgetting instance from the ring because it is unhealthy for a long time", "instance", id, "last_heartbeat", lastHeartbeat.String(), "forget_period", d.forgetPeriod)
-			ringDesc.RemoveIngester(id)
-		}
-	}
-
+	AutoForgetFromRing(ringDesc, d.forgetPeriod, d.logger)
 	d.next.OnRingInstanceHeartbeat(lifecycler, ringDesc, instanceDesc)
 }

pkg/ring/lifecycler.go

@@ -27,6 +27,17 @@ var (
 	errInvalidTokensGeneratorStrategy = errors.New("invalid token generator strategy")
 )
 
+type LifecyclerDelegate interface {
+	// OnRingInstanceHeartbeat is called while the instance is updating its heartbeat
+	// in the ring.
+	OnRingInstanceHeartbeat(lifecycler *Lifecycler, ringDesc *Desc)
+}
+
+type DefaultLifecyclerDelegate struct{}
+
+func (d DefaultLifecyclerDelegate) OnRingInstanceHeartbeat(lifecycler *Lifecycler, ringDesc *Desc) {
+}
+
 // LifecyclerConfig is the config to build a Lifecycler.
 type LifecyclerConfig struct {
 	RingConfig Config `yaml:"ring"`
@@ -108,6 +119,7 @@ type Lifecycler struct {
 	cfg             LifecyclerConfig
 	flushTransferer FlushTransferer
 	KVStore         kv.Client
+	delegate        LifecyclerDelegate
 
 	actorChan    chan func()
 	autojoinChan chan struct{}
@@ -150,6 +162,22 @@ type Lifecycler struct {
 	tg TokenGenerator
 }
 
+func NewLifecyclerWithDelegate(
+	cfg LifecyclerConfig,
+	flushTransferer FlushTransferer,
+	ringName, ringKey string,
+	autoJoinOnStartup, flushOnShutdown bool,
+	logger log.Logger,
+	reg prometheus.Registerer,
+	delegate LifecyclerDelegate,
+) (*Lifecycler, error) {
+	l, err := NewLifecycler(cfg, flushTransferer, ringName, ringKey, autoJoinOnStartup, flushOnShutdown, logger, reg)
+	if l != nil {
+		l.delegate = delegate
+	}
+	return l, err
+}
+
 // NewLifecycler creates new Lifecycler. It must be started via StartAsync.
 func NewLifecycler(
 	cfg LifecyclerConfig,
@@ -209,6 +237,7 @@ func NewLifecycler(
 		lifecyclerMetrics:    NewLifecyclerMetrics(ringName, reg),
 		logger:               logger,
 		tg:                   tg,
+		delegate:             &DefaultLifecyclerDelegate{},
 	}
 
 	l.lifecyclerMetrics.tokensToOwn.Set(float64(cfg.NumTokens))
@@ -973,6 +1002,7 @@ func (i *Lifecycler) updateConsul(ctx context.Context) error {
 			instanceDesc.RegisteredTimestamp = i.getRegisteredAt().Unix()
 			ringDesc.Ingesters[i.ID] = instanceDesc
 		}
+		i.delegate.OnRingInstanceHeartbeat(i, ringDesc)
 
 		return ringDesc, true, nil
 	})

pkg/ring/lifecycler_delegates.go

@@ -0,0 +1,28 @@
+package ring
+
+import (
+	"time"
+
+	"github.com/go-kit/log"
+)
+
+// AutoForgetDelegate automatically remove an instance from the ring if the last
+// heartbeat is older than a configured period.
+type LifecyclerAutoForgetDelegate struct {
+	next         LifecyclerDelegate
+	logger       log.Logger
+	forgetPeriod time.Duration
+}
+
+func NewLifecyclerAutoForgetDelegate(forgetPeriod time.Duration, next LifecyclerDelegate, logger log.Logger) *LifecyclerAutoForgetDelegate {
+	return &LifecyclerAutoForgetDelegate{
+		next:         next,
+		logger:       logger,
+		forgetPeriod: forgetPeriod,
+	}
+}
+
+func (d *LifecyclerAutoForgetDelegate) OnRingInstanceHeartbeat(lifecycler *Lifecycler, ringDesc *Desc) {
+	AutoForgetFromRing(ringDesc, d.forgetPeriod, d.logger)
+	d.next.OnRingInstanceHeartbeat(lifecycler, ringDesc)
+}

pkg/ring/ring.go

@@ -1029,3 +1029,14 @@ func (op Operation) ShouldExtendReplicaSetOnState(s InstanceState) bool {
 
 // All states are healthy, no states extend replica set.
 var allStatesRingOperation = Operation(0x0000ffff)
+
+func AutoForgetFromRing(ringDesc *Desc, forgetPeriod time.Duration, logger log.Logger) {
+	for id, instance := range ringDesc.Ingesters {
+		lastHeartbeat := time.Unix(instance.GetTimestamp(), 0)
+
+		if time.Since(lastHeartbeat) > forgetPeriod {
+			level.Warn(logger).Log("msg", "auto-forgetting instance from the ring because it is unhealthy for a long time", "instance", id, "last_heartbeat", lastHeartbeat.String(), "forget_period", forgetPeriod)
+			ringDesc.RemoveIngester(id)
+		}
+	}
+}