Automatically forget unhealthy Rulers (#2587)

* add ability to autoforget ruler instances

Signed-off-by: Jacob Lisi <[email protected]>

* update unit tests to handle autoforgetting ruler ring

Signed-off-by: Jacob Lisi <[email protected]>

* update changelog

Signed-off-by: Jacob Lisi <[email protected]>

* update per PR comments

Signed-off-by: Jacob Lisi <[email protected]>

Author: jtlisi

CHANGELOG.md

@@ -34,6 +34,7 @@
 * [FEATURE] Experimental: Added support for `/api/v1/metadata` Prometheus-based endpoint. #2549
 * [FEATURE] Add ability to limit concurrent queries to Cassandra with `-cassandra.query-concurrency` flag. #2562
 * [FEATURE] TLS config options added for GRPC clients in Querier (Query-frontend client & Ingester client), Ruler, Store Gateway, as well as HTTP client in Config store client. #2502
+* [ENHANCEMENT] Ruler: Automatically remove unhealthy rulers from the ring. #2587
 * [ENHANCEMENT] Experimental TSDB: sample ingestion errors are now reported via existing `cortex_discarded_samples_total` metric. #2370
 * [ENHANCEMENT] Failures on samples at distributors and ingesters return the first validation error as opposed to the last. #2383
 * [ENHANCEMENT] Experimental TSDB: Added `cortex_querier_blocks_meta_synced`, which reflects current state of synced blocks over all tenants. #2392

pkg/ruler/lifecycle.go

@@ -1,16 +1,28 @@
 package ruler
 
 import (
-	"context"
+	"github.com/cortexproject/cortex/pkg/ring"
 )
 
-// TransferOut is a noop for the ruler
-func (r *Ruler) TransferOut(ctx context.Context) error {
-	return nil
+func (r *Ruler) OnRingInstanceRegister(_ *ring.BasicLifecycler, ringDesc ring.Desc, instanceExists bool, instanceID string, instanceDesc ring.IngesterDesc) (ring.IngesterState, ring.Tokens) {
+	// When we initialize the ruler instance in the ring we want to start from
+	// a clean situation, so whatever is the state we set it ACTIVE, while we keep existing
+	// tokens (if any).
+	var tokens []uint32
+	if instanceExists {
+		tokens = instanceDesc.GetTokens()
+	}
+
+	_, takenTokens := ringDesc.TokensFor(instanceID)
+	newTokens := ring.GenerateTokens(r.cfg.Ring.NumTokens-len(tokens), takenTokens)
+
+	// Tokens sorting will be enforced by the parent caller.
+	tokens = append(tokens, newTokens...)
+
+	return ring.ACTIVE, tokens
 }
 
-// Flush triggers a flush of all the work items currently
-// scheduled by the ruler, currently every ruler will
-// query a backend rule store for it's rules so no
-// flush is required.
-func (r *Ruler) Flush() {}
+func (r *Ruler) OnRingInstanceTokens(_ *ring.BasicLifecycler, _ ring.Tokens) {}
+func (r *Ruler) OnRingInstanceStopping(_ *ring.BasicLifecycler)              {}
+func (r *Ruler) OnRingInstanceHeartbeat(_ *ring.BasicLifecycler, _ *ring.Desc, _ *ring.IngesterDesc) {
+}

pkg/ruler/lifecycle_test.go

@@ -2,73 +2,104 @@ package ruler
 
 import (
 	"context"
+	"sort"
 	"testing"
 	"time"
 
-	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/cortexproject/cortex/pkg/ring"
+	"github.com/cortexproject/cortex/pkg/ring/kv/consul"
 	"github.com/cortexproject/cortex/pkg/ring/testutils"
 	"github.com/cortexproject/cortex/pkg/util/services"
 	"github.com/cortexproject/cortex/pkg/util/test"
 )
 
 // TestRulerShutdown tests shutting down ruler unregisters correctly
 func TestRulerShutdown(t *testing.T) {
+	ctx := context.Background()
+
 	config, cleanup := defaultRulerConfig(newMockRuleStore(mockRules))
-	config.EnableSharding = true
-	config.Ring.SkipUnregister = false
 	defer cleanup()
 
-	r, rcleanup := newTestRuler(t, config)
+	r, rcleanup := newRuler(t, config)
 	defer rcleanup()
 
+	r.cfg.EnableSharding = true
+	ringStore := consul.NewInMemoryClient(ring.GetCodec())
+
+	err := enableSharding(r, ringStore)
+	require.NoError(t, err)
+
+	require.NoError(t, services.StartAndAwaitRunning(ctx, r))
+	defer services.StopAndAwaitTerminated(ctx, r) //nolint:errcheck
+
 	// Wait until the tokens are registered in the ring
 	test.Poll(t, 100*time.Millisecond, config.Ring.NumTokens, func() interface{} {
-		return testutils.NumTokens(config.Ring.KVStore.Mock, "localhost", ring.RulerRingKey)
+		return testutils.NumTokens(ringStore, "localhost", ring.RulerRingKey)
 	})
 
+	require.Equal(t, ring.ACTIVE, r.lifecycler.GetState())
+
 	require.NoError(t, services.StopAndAwaitTerminated(context.Background(), r))
 
 	// Wait until the tokens are unregistered from the ring
 	test.Poll(t, 100*time.Millisecond, 0, func() interface{} {
-		return testutils.NumTokens(config.Ring.KVStore.Mock, "localhost", ring.RulerRingKey)
+		return testutils.NumTokens(ringStore, "localhost", ring.RulerRingKey)
 	})
 }
 
-// TestRulerRestart tests a restarting ruler doesn't keep adding more tokens.
-func TestRulerRestart(t *testing.T) {
+func TestRuler_RingLifecyclerShouldAutoForgetUnhealthyInstances(t *testing.T) {
+	const unhealthyInstanceID = "unhealthy-id"
+	const heartbeatTimeout = time.Minute
+
+	ctx := context.Background()
 	config, cleanup := defaultRulerConfig(newMockRuleStore(mockRules))
-	config.Ring.SkipUnregister = true
-	config.EnableSharding = true
 	defer cleanup()
-
-	r, rcleanup := newTestRuler(t, config)
+	r, rcleanup := newRuler(t, config)
 	defer rcleanup()
+	r.cfg.EnableSharding = true
+	r.cfg.Ring.HeartbeatPeriod = 100 * time.Millisecond
+	r.cfg.Ring.HeartbeatTimeout = heartbeatTimeout
 
-	// Wait until the tokens are registered in the ring
-	test.Poll(t, 100*time.Millisecond, config.Ring.NumTokens, func() interface{} {
-		return testutils.NumTokens(config.Ring.KVStore.Mock, "localhost", ring.RulerRingKey)
-	})
+	ringStore := consul.NewInMemoryClient(ring.GetCodec())
 
-	// Stop the ruler. Doesn't actually unregister due to skipUnregister: true
-	r.StopAsync()
-	require.NoError(t, r.AwaitTerminated(context.Background()))
+	err := enableSharding(r, ringStore)
+	require.NoError(t, err)
 
-	// We expect the tokens are preserved in the ring.
-	assert.Equal(t, config.Ring.NumTokens, testutils.NumTokens(config.Ring.KVStore.Mock, "localhost", ring.RulerRingKey))
+	require.NoError(t, services.StartAndAwaitRunning(ctx, r))
+	defer services.StopAndAwaitTerminated(ctx, r) //nolint:errcheck
 
-	// Create a new ruler which is expected to pick up tokens from the ring.
-	r, rcleanup = newTestRuler(t, config)
-	defer rcleanup()
-	defer services.StopAndAwaitTerminated(context.Background(), r) //nolint:errcheck
+	// Add an unhealthy instance to the ring.
+	require.NoError(t, ringStore.CAS(ctx, ring.RulerRingKey, func(in interface{}) (interface{}, bool, error) {
+		ringDesc := ring.GetOrCreateRingDesc(in)
+
+		instance := ringDesc.AddIngester(unhealthyInstanceID, "1.1.1.1", "", generateSortedTokens(config.Ring.NumTokens), ring.ACTIVE)
+		instance.Timestamp = time.Now().Add(-(ringAutoForgetUnhealthyPeriods + 1) * heartbeatTimeout).Unix()
+		ringDesc.Ingesters[unhealthyInstanceID] = instance
+
+		return ringDesc, true, nil
+	}))
+
+	// Ensure the unhealthy instance is removed from the ring.
+	test.Poll(t, time.Second*5, false, func() interface{} {
+		d, err := ringStore.Get(ctx, ring.RulerRingKey)
+		if err != nil {
+			return err
+		}
+
+		_, ok := ring.GetOrCreateRingDesc(d).Ingesters[unhealthyInstanceID]
+		return ok
+	})
+}
+
+func generateSortedTokens(numTokens int) ring.Tokens {
+	tokens := ring.GenerateTokens(numTokens, nil)
 
-	// Wait until the ruler is ACTIVE in the ring.
-	test.Poll(t, 100*time.Millisecond, ring.ACTIVE, func() interface{} {
-		return r.lifecycler.GetState()
+	// Ensure generated tokens are sorted.
+	sort.Slice(tokens, func(i, j int) bool {
+		return tokens[i] < tokens[j]
 	})
 
-	// We expect no new tokens have been added to the ring.
-	assert.Equal(t, config.Ring.NumTokens, testutils.NumTokens(config.Ring.KVStore.Mock, "localhost", ring.RulerRingKey))
+	return ring.Tokens(tokens)
 }

pkg/ruler/ruler.go

@@ -30,6 +30,7 @@ import (
 
 	"github.com/cortexproject/cortex/pkg/ingester/client"
 	"github.com/cortexproject/cortex/pkg/ring"
+	"github.com/cortexproject/cortex/pkg/ring/kv"
 	"github.com/cortexproject/cortex/pkg/ruler/rules"
 	store "github.com/cortexproject/cortex/pkg/ruler/rules"
 	"github.com/cortexproject/cortex/pkg/util"
@@ -147,7 +148,7 @@ type Ruler struct {
 	alertURL    *url.URL
 	notifierCfg *config.Config
 
-	lifecycler  *ring.Lifecycler
+	lifecycler  *ring.BasicLifecycler
 	ring        *ring.Ring
 	subservices *services.Manager
 
@@ -191,26 +192,50 @@ func NewRuler(cfg Config, engine *promql.Engine, queryable promStorage.Queryable
 		logger:       logger,
 	}
 
+	if cfg.EnableSharding {
+		ringStore, err := kv.NewClient(cfg.Ring.KVStore, ring.GetCodec())
+		if err != nil {
+			return nil, errors.Wrap(err, "create KV store client")
+		}
+
+		if err = enableSharding(ruler, ringStore); err != nil {
+			return nil, errors.Wrap(err, "setup ruler sharding ring")
+		}
+	}
+
 	ruler.Service = services.NewBasicService(ruler.starting, ruler.run, ruler.stopping)
 	return ruler, nil
 }
 
+func enableSharding(r *Ruler, ringStore kv.Client) error {
+	lifecyclerCfg, err := r.cfg.Ring.ToLifecyclerConfig()
+	if err != nil {
+		return errors.Wrap(err, "failed to initialize ruler's lifecycler config")
+	}
+
+	// Define lifecycler delegates in reverse order (last to be called defined first because they're
+	// chained via "next delegate").
+	delegate := ring.BasicLifecyclerDelegate(r)
+	delegate = ring.NewLeaveOnStoppingDelegate(delegate, r.logger)
+	delegate = ring.NewAutoForgetDelegate(r.cfg.Ring.HeartbeatTimeout*ringAutoForgetUnhealthyPeriods, delegate, r.logger)
+
+	r.lifecycler, err = ring.NewBasicLifecycler(lifecyclerCfg, ring.RulerRingKey, ring.RulerRingKey, ringStore, delegate, r.logger, r.registry)
+	if err != nil {
+		return errors.Wrap(err, "failed to initialize ruler's lifecycler")
+	}
+
+	r.ring, err = ring.NewWithStoreClientAndStrategy(r.cfg.Ring.ToRingConfig(), ring.RulerRingKey, ring.RulerRingKey, ringStore, &ring.DefaultReplicationStrategy{})
+	if err != nil {
+		return errors.Wrap(err, "failed to initialize ruler's ring")
+	}
+
+	return nil
+}
+
 func (r *Ruler) starting(ctx context.Context) error {
-	// If sharding is enabled, create/join a ring to distribute tokens to
-	// the ruler
+	// If sharding is enabled, start the ruler ring subservices
 	if r.cfg.EnableSharding {
-		lifecyclerCfg := r.cfg.Ring.ToLifecyclerConfig()
 		var err error
-		r.lifecycler, err = ring.NewLifecycler(lifecyclerCfg, r, "ruler", ring.RulerRingKey, true)
-		if err != nil {
-			return errors.Wrap(err, "failed to initialize ruler's lifecycler")
-		}
-
-		r.ring, err = ring.New(lifecyclerCfg.RingConfig, "ruler", ring.RulerRingKey)
-		if err != nil {
-			return errors.Wrap(err, "failed to initialize ruler's ring")
-		}
-
 		r.subservices, err = services.NewManager(r.lifecycler, r.ring)
 		if err == nil {
 			err = services.StartManagerAndAwaitHealthy(ctx, r.subservices)
@@ -331,11 +356,14 @@ func (r *Ruler) ownsRule(hash uint32) (bool, error) {
 		ringCheckErrors.Inc()
 		return false, err
 	}
-	if rlrs.Ingesters[0].Addr == r.lifecycler.Addr {
-		level.Debug(r.logger).Log("msg", "rule group owned", "owner_addr", rlrs.Ingesters[0].Addr, "addr", r.lifecycler.Addr)
+
+	localAddr := r.lifecycler.GetInstanceAddr()
+
+	if rlrs.Ingesters[0].Addr == localAddr {
+		level.Debug(r.logger).Log("msg", "rule group owned", "owner_addr", rlrs.Ingesters[0].Addr, "addr", localAddr)
 		return true, nil
 	}
-	level.Debug(r.logger).Log("msg", "rule group not owned, address does not match", "owner_addr", rlrs.Ingesters[0].Addr, "addr", r.lifecycler.Addr)
+	level.Debug(r.logger).Log("msg", "rule group not owned, address does not match", "owner_addr", rlrs.Ingesters[0].Addr, "addr", localAddr)
 	return false, nil
 }
 

pkg/ruler/ruler_ring.go

@@ -2,6 +2,7 @@ package ruler
 
 import (
 	"flag"
+	"fmt"
 	"os"
 	"time"
 
@@ -13,6 +14,13 @@ import (
 	"github.com/cortexproject/cortex/pkg/util/flagext"
 )
 
+const (
+	// If a ruler is unable to heartbeat the ring, its better to quickly remove it and resume
+	// the evaluation of all rules since the worst case scenario is that some rulers will
+	// receive duplicate/out-of-order sample errors.
+	ringAutoForgetUnhealthyPeriods = 2
+)
+
 // RingConfig masks the ring lifecycler config which contains
 // many options not really required by the rulers ring. This config
 // is used to strip down the config to the minimum, and avoid confusion
@@ -60,34 +68,30 @@ func (cfg *RingConfig) RegisterFlags(f *flag.FlagSet) {
 
 // ToLifecyclerConfig returns a LifecyclerConfig based on the ruler
 // ring config.
-func (cfg *RingConfig) ToLifecyclerConfig() ring.LifecyclerConfig {
-	// We have to make sure that the ring.LifecyclerConfig and ring.Config
-	// defaults are preserved
-	lc := ring.LifecyclerConfig{}
-	rc := ring.Config{}
+func (cfg *RingConfig) ToLifecyclerConfig() (ring.BasicLifecyclerConfig, error) {
+	instanceAddr, err := ring.GetInstanceAddr(cfg.InstanceAddr, cfg.InstanceInterfaceNames)
+	if err != nil {
+		return ring.BasicLifecyclerConfig{}, err
+	}
 
-	flagext.DefaultValues(&lc)
+	instancePort := ring.GetInstancePort(cfg.InstancePort, cfg.ListenPort)
+
+	return ring.BasicLifecyclerConfig{
+		ID:                  cfg.InstanceID,
+		Addr:                fmt.Sprintf("%s:%d", instanceAddr, instancePort),
+		HeartbeatPeriod:     cfg.HeartbeatPeriod,
+		TokensObservePeriod: 0,
+		NumTokens:           cfg.NumTokens,
+	}, nil
+}
+
+func (cfg *RingConfig) ToRingConfig() ring.Config {
+	rc := ring.Config{}
 	flagext.DefaultValues(&rc)
 
-	// Configure ring
 	rc.KVStore = cfg.KVStore
 	rc.HeartbeatTimeout = cfg.HeartbeatTimeout
 	rc.ReplicationFactor = 1
 
-	// Configure lifecycler
-	lc.RingConfig = rc
-	lc.ListenPort = cfg.ListenPort
-	lc.Addr = cfg.InstanceAddr
-	lc.Port = cfg.InstancePort
-	lc.ID = cfg.InstanceID
-	lc.InfNames = cfg.InstanceInterfaceNames
-	lc.SkipUnregister = cfg.SkipUnregister
-	lc.HeartbeatPeriod = cfg.HeartbeatPeriod
-	lc.NumTokens = cfg.NumTokens
-	lc.ObservePeriod = 0
-	lc.JoinAfter = 0
-	lc.MinReadyDuration = 0
-	lc.FinalSleep = 0
-
-	return lc
+	return rc
 }

pkg/ruler/ruler_test.go

@@ -55,7 +55,7 @@ func defaultRulerConfig(store rules.RuleStore) (Config, func()) {
 	return cfg, cleanup
 }
 
-func newTestRuler(t *testing.T, cfg Config) (*Ruler, func()) {
+func newRuler(t *testing.T, cfg Config) (*Ruler, func()) {
 	dir, err := ioutil.TempDir("", t.Name())
 	testutil.Ok(t, err)
 	cleanup := func() {
@@ -82,6 +82,12 @@ func newTestRuler(t *testing.T, cfg Config) (*Ruler, func()) {
 	l = level.NewFilter(l, level.AllowInfo())
 	ruler, err := NewRuler(cfg, engine, noopQueryable, pusher, prometheus.NewRegistry(), l)
 	require.NoError(t, err)
+
+	return ruler, cleanup
+}
+
+func newTestRuler(t *testing.T, cfg Config) (*Ruler, func()) {
+	ruler, cleanup := newRuler(t, cfg)
 	require.NoError(t, services.StartAndAwaitRunning(context.Background(), ruler))
 
 	// Ensure all rules are loaded before usage