Implement kv.Client based on memberlist library. (#1727)

Signed-off-by: Peter Štibraný <[email protected]>

Author: pstibrany

CHANGELOG.md

@@ -9,6 +9,7 @@
   * `-ingester.max-global-series-per-user`
   * `-ingester.max-global-series-per-metric`
 * [FEATURE] Flush chunks with stale markers early with `ingester.max-stale-chunk-idle`. #1759
+* [FEATURE] EXPERIMENTAL: Added new KV Store backend based on memberlist library. Components can gossip about tokens and ingester states, instead of using Consul or Etcd. #1721
 * [ENHANCEMENT] Allocation improvements in adding samples to Chunk. #1706
 * [ENHANCEMENT] Consul client now follows recommended practices for blocking queries wrt returned Index value. #1708
 * [ENHANCEMENT] Consul client can optionally rate-limit itself during Watch (used e.g. by ring watchers) and WatchPrefix (used by HA feature) operations. Rate limiting is disabled by default. New flags added: `--consul.watch-rate-limit`, and `--consul.watch-burst-size`. #1708

docs/arguments.md

@@ -142,6 +142,39 @@ prefix these flags with `distributor.ha-tracker.`
 - `etcd.max-retries`
    The maximum number of retries to do for failed ops.
 
+#### memberlist (EXPERIMENTAL)
+
+Flags for configuring KV store based on memberlist library. This feature is experimental, please don't use it yet.
+
+- `memberlist.nodename`
+   Name of the node in memberlist cluster. Defaults to hostname.
+- `memberlist.retransmit-factor`
+   Multiplication factor used when sending out messages (factor * log(N+1)). If not set, default value is used.
+- `memberlist.join`
+   Other cluster members to join. Can be specified multiple times.
+- `memberlist.abort-if-join-fails`
+   If this node fails to join memberlist cluster, abort.
+- `memberlist.left-ingesters-timeout`
+   How long to keep LEFT ingesters in the ring. Note: this is only used for gossiping, LEFT ingesters are otherwise invisible.
+- `memberlist.leave-timeout`
+   Timeout for leaving memberlist cluster.
+- `memberlist.gossip-interval`
+   How often to gossip with other cluster members. Uses memberlist LAN defaults if 0.
+- `memberlist.gossip-nodes`
+   How many nodes to gossip with in each gossip interval. Uses memberlist LAN defaults if 0.
+- `memberlist.pullpush-interval`
+   How often to use pull/push sync. Uses memberlist LAN defaults if 0.
+- `memberlist.bind-addr`
+   IP address to listen on for gossip messages. Multiple addresses may be specified. Defaults to 0.0.0.0.
+- `memberlist.bind-port`
+   Port to listen on for gossip messages. Defaults to 7946.
+- `memberlist.packet-dial-timeout`
+   Timeout used when connecting to other nodes to send packet.
+- `memberlist.packet-write-timeout`
+   Timeout for writing 'packet' data.
+- `memberlist.transport-debug`
+   Log debug transport messages. Note: global log.level must be at debug level as well.
+
 ### HA Tracker
 
 HA tracking has two of it's own flags:

go.mod

@@ -38,6 +38,8 @@ require (
 	github.com/hailocab/go-hostpool v0.0.0-20160125115350-e80d13ce29ed // indirect
 	github.com/hashicorp/consul/api v1.1.0
 	github.com/hashicorp/go-cleanhttp v0.5.1
+	github.com/hashicorp/go-sockaddr v1.0.2
+	github.com/hashicorp/memberlist v0.1.4
 	github.com/jonboulle/clockwork v0.1.0
 	github.com/json-iterator/go v1.1.7
 	github.com/konsorten/go-windows-terminal-sequences v1.0.2 // indirect

pkg/ingester/transfer.go

@@ -76,6 +76,12 @@ func (i *Ingester) TransferChunks(stream client.Ingester_TransferChunksServer) e
 			if fromIngesterID == "" {
 				fromIngesterID = wireSeries.FromIngesterId
 				level.Info(util.Logger).Log("msg", "processing TransferChunks request", "from_ingester", fromIngesterID)
+
+				// Before transfer, make sure 'from' ingester is in correct state to call ClaimTokensFor later
+				err := i.checkFromIngesterIsInLeavingState(stream.Context(), fromIngesterID)
+				if err != nil {
+					return err
+				}
 			}
 			descs, err := fromWireChunks(wireSeries.Chunks)
 			if err != nil {
@@ -136,6 +142,37 @@ func (i *Ingester) TransferChunks(stream client.Ingester_TransferChunksServer) e
 	return nil
 }
 
+// Ring gossiping: check if "from" ingester is in LEAVING state. It should be, but we may not see that yet
+// when using gossip ring. If we cannot see ingester is the LEAVING state yet, we don't accept this
+// transfer, as claiming tokens would possibly end up with this ingester owning no tokens, due to conflict
+// resolution in ring merge function. Hopefully the leaving ingester will retry transfer again.
+func (i *Ingester) checkFromIngesterIsInLeavingState(ctx context.Context, fromIngesterID string) error {
+	v, err := i.lifecycler.KVStore.Get(ctx, ring.ConsulKey)
+	if err != nil {
+		return errors.Wrap(err, "TransferChunks: get ring")
+	}
+	if v == nil {
+		return fmt.Errorf("TransferChunks: ring not found when checking state of source ingester")
+	}
+	r, ok := v.(*ring.Desc)
+	if !ok || r == nil {
+		return fmt.Errorf("TransferChunks: ring not found, got %T", v)
+	}
+
+	if r.Ingesters == nil || r.Ingesters[fromIngesterID].State != ring.LEAVING {
+		return fmt.Errorf("TransferChunks: source ingester is not in a LEAVING state, found state=%v", r.Ingesters[fromIngesterID].State)
+	}
+
+	if r.Ingesters == nil || r.Ingesters[fromIngesterID].State != ring.LEAVING {
+		err = fmt.Errorf("source ingester is not in a LEAVING state, found state=%v", r.Ingesters[fromIngesterID].State)
+		util.Logger.Log("msg", "TransferChunks error", "err", err)
+		return err
+	}
+
+	// all fine
+	return nil
+}
+
 func (i *Ingester) transfer(ctx context.Context, xfer func() error) error {
 	// Enter JOINING state (only valid from PENDING)
 	if err := i.lifecycler.ChangeState(ctx, ring.JOINING); err != nil {
@@ -198,6 +235,12 @@ func (i *Ingester) TransferTSDB(stream client.Ingester_TransferTSDBServer) error
 			if fromIngesterID == "" {
 				fromIngesterID = f.FromIngesterId
 				level.Info(util.Logger).Log("msg", "processing TransferTSDB request", "from_ingester", fromIngesterID)
+
+				// Before transfer, make sure 'from' ingester is in correct state to call ClaimTokensFor later
+				err := i.checkFromIngesterIsInLeavingState(stream.Context(), fromIngesterID)
+				if err != nil {
+					return err
+				}
 			}
 			filesXfer++
 

pkg/ring/kv/client.go

@@ -6,9 +6,12 @@ import (
 	"fmt"
 	"sync"
 
+	"github.com/prometheus/client_golang/prometheus"
+
 	"github.com/cortexproject/cortex/pkg/ring/kv/codec"
 	"github.com/cortexproject/cortex/pkg/ring/kv/consul"
 	"github.com/cortexproject/cortex/pkg/ring/kv/etcd"
+	"github.com/cortexproject/cortex/pkg/ring/kv/memberlist"
 )
 
 // The NewInMemoryKVClient returned by NewClient() is a singleton, so
@@ -20,10 +23,11 @@ var inmemoryStore Client
 // Config is config for a KVStore currently used by ring and HA tracker,
 // where store can be consul or inmemory.
 type Config struct {
-	Store  string        `yaml:"store,omitempty"`
-	Consul consul.Config `yaml:"consul,omitempty"`
-	Etcd   etcd.Config   `yaml:"etcd,omitempty"`
-	Prefix string        `yaml:"prefix,omitempty"`
+	Store      string            `yaml:"store,omitempty"`
+	Consul     consul.Config     `yaml:"consul,omitempty"`
+	Etcd       etcd.Config       `yaml:"etcd,omitempty"`
+	Memberlist memberlist.Config `yaml:"memberlist,omitempty"`
+	Prefix     string            `yaml:"prefix,omitempty"`
 
 	Mock Client
 }
@@ -40,11 +44,12 @@ func (cfg *Config) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
 	// be easier to have everything under ring, so ring.consul.<flag-name>
 	cfg.Consul.RegisterFlags(f, prefix)
 	cfg.Etcd.RegisterFlagsWithPrefix(f, prefix)
+	cfg.Memberlist.RegisterFlags(f, prefix)
 	if prefix == "" {
 		prefix = "ring."
 	}
 	f.StringVar(&cfg.Prefix, prefix+"prefix", "collectors/", "The prefix for the keys in the store. Should end with a /.")
-	f.StringVar(&cfg.Store, prefix+"store", "consul", "Backend storage to use for the ring (consul, etcd, inmemory).")
+	f.StringVar(&cfg.Store, prefix+"store", "consul", "Backend storage to use for the ring (consul, etcd, inmemory, memberlist [experimental]).")
 }
 
 // Client is a high-level client for key-value stores (such as Etcd and
@@ -69,6 +74,9 @@ type Client interface {
 
 	// WatchPrefix calls f whenever any value stored under prefix changes.
 	WatchPrefix(ctx context.Context, prefix string, f func(string, interface{}) bool)
+
+	// If client needs to do some cleanup, it can do it here.
+	Stop()
 }
 
 // NewClient creates a new Client (consul, etcd or inmemory) based on the config,
@@ -96,6 +104,10 @@ func NewClient(cfg Config, codec codec.Codec) (Client, error) {
 		})
 		client = inmemoryStore
 
+	case "memberlist":
+		cfg.Memberlist.MetricsRegisterer = prometheus.DefaultRegisterer
+		client, err = memberlist.NewMemberlistClient(cfg.Memberlist, codec)
+
 	default:
 		return nil, fmt.Errorf("invalid KV store type: %s", cfg.Store)
 	}

pkg/ring/kv/consul/client.go

@@ -328,3 +328,8 @@ func (c *Client) createRateLimiter() *rate.Limiter {
 	}
 	return rate.NewLimiter(rate.Limit(c.cfg.WatchKeyRateLimit), burst)
 }
+
+// Stop does nothing in Consul client.
+func (c *Client) Stop() {
+	// nothing to do
+}

pkg/ring/kv/etcd/etcd.go

@@ -194,3 +194,8 @@ func (c *Client) Get(ctx context.Context, key string) (interface{}, error) {
 	}
 	return c.codec.Decode(resp.Kvs[0].Value)
 }
+
+// Stop does nothing in etcd client.
+func (c *Client) Stop() {
+	// nothing to do
+}

pkg/ring/kv/memberlist/broadcast.go

@@ -0,0 +1,64 @@
+package memberlist
+
+import (
+	"fmt"
+
+	"github.com/go-kit/kit/log/level"
+	"github.com/hashicorp/memberlist"
+
+	"github.com/cortexproject/cortex/pkg/util"
+)
+
+// ringBroadcast implements memberlist.Broadcast interface, which is used by memberlist.TransmitLimitedQueue.
+type ringBroadcast struct {
+	key      string
+	content  []string // Description of what is stored in this value. Used for invalidation.
+	version  uint     // local version of the value, generated by merging this change
+	msg      []byte   // encoded key and value
+	finished func(b ringBroadcast)
+}
+
+func (r ringBroadcast) Invalidates(old memberlist.Broadcast) bool {
+	if oldb, ok := old.(ringBroadcast); ok {
+		if r.key != oldb.key {
+			return false
+		}
+
+		// if 'content' (result of Mergeable.MergeContent) of this broadcast is a superset of content of old value,
+		// and this broadcast has resulted in a newer ring update, we can invalidate the old value
+
+		for _, oldName := range oldb.content {
+			found := false
+			for _, newName := range r.content {
+				if oldName == newName {
+					found = true
+					break
+				}
+			}
+
+			if !found {
+				return false
+			}
+		}
+
+		// only do this check if this ringBroadcast covers same ingesters as 'b'
+		// otherwise, we may be invalidating some older messages, which however covered different
+		// ingesters
+		if r.version >= oldb.version {
+			level.Debug(util.Logger).Log("msg", "Invalidating forwarded broadcast", "key", r.key, "version", r.version, "oldVersion", oldb.version, "content", fmt.Sprintf("%v", r.content), "oldContent", fmt.Sprintf("%v", oldb.content))
+			return true
+		}
+	}
+	return false
+}
+
+func (r ringBroadcast) Message() []byte {
+	return r.msg
+}
+
+func (r ringBroadcast) Finished() {
+	if r.finished != nil {
+		r.finished(r)
+	}
+	r.msg = nil
+}

pkg/ring/kv/memberlist/broadcast_test.go

@@ -0,0 +1,52 @@
+package memberlist
+
+import "testing"
+
+func TestInvalidates(t *testing.T) {
+	const key = "ring"
+
+	messages := map[string]ringBroadcast{
+		"b1": {key: key, content: []string{"A", "B", "C"}, version: 1},
+		"b2": {key: key, content: []string{"A", "B", "C"}, version: 2},
+		"b3": {key: key, content: []string{"A"}, version: 3},
+		"b4": {key: key, content: []string{"A", "B"}, version: 4},
+		"b5": {key: key, content: []string{"A", "B", "D"}, version: 5},
+		"b6": {key: key, content: []string{"A", "B", "C", "D"}, version: 6},
+	}
+
+	checkInvalidate(t, messages, "b2", "b1", true, false)
+	checkInvalidate(t, messages, "b3", "b1", false, false)
+	checkInvalidate(t, messages, "b3", "b2", false, false)
+	checkInvalidate(t, messages, "b4", "b1", false, false)
+	checkInvalidate(t, messages, "b4", "b2", false, false)
+	checkInvalidate(t, messages, "b4", "b3", true, false)
+	checkInvalidate(t, messages, "b5", "b1", false, false)
+	checkInvalidate(t, messages, "b5", "b2", false, false)
+	checkInvalidate(t, messages, "b5", "b3", true, false)
+	checkInvalidate(t, messages, "b5", "b4", true, false)
+	checkInvalidate(t, messages, "b6", "b1", true, false)
+	checkInvalidate(t, messages, "b6", "b2", true, false)
+	checkInvalidate(t, messages, "b6", "b3", true, false)
+	checkInvalidate(t, messages, "b6", "b4", true, false)
+	checkInvalidate(t, messages, "b6", "b5", true, false)
+}
+
+func checkInvalidate(t *testing.T, messages map[string]ringBroadcast, key1, key2 string, firstInvalidatesSecond, secondInvalidatesFirst bool) {
+	b1, ok := messages[key1]
+	if !ok {
+		t.Fatal("cannot find", key1)
+	}
+
+	b2, ok := messages[key2]
+	if !ok {
+		t.Fatal("cannot find", key2)
+	}
+
+	if b1.Invalidates(b2) != firstInvalidatesSecond {
+		t.Errorf("%s.Invalidates(%s) returned %t. %s={%v, %d}, %s={%v, %d}", key1, key2, !firstInvalidatesSecond, key1, b1.content, b1.version, key2, b2.content, b2.version)
+	}
+
+	if b2.Invalidates(b1) != secondInvalidatesFirst {
+		t.Errorf("%s.Invalidates(%s) returned %t. %s={%v, %d}, %s={%v, %d}", key2, key1, !secondInvalidatesFirst, key2, b2.content, b2.version, key1, b1.content, b1.version)
+	}
+}