memberlist

Documentation

Index

• type Client
  • func NewMemberlistClient(cfg Config, codec codec.Codec) (*Client, error)
  • func (m *Client) CAS(ctx context.Context, key string, ...) error
  • func (m *Client) Collect(ch chan<- prometheus.Metric)
  • func (m *Client) Describe(ch chan<- *prometheus.Desc)
  • func (m *Client) Get(ctx context.Context, key string) (interface{}, error)
  • func (m *Client) GetBroadcasts(overhead, limit int) [][]byte
  • func (m *Client) GetListeningPort() int
  • func (m *Client) JoinMembers(members []string) (int, error)
  • func (m *Client) LocalState(join bool) []byte
  • func (m *Client) MergeRemoteState(stateMsg []byte, join bool)
  • func (m *Client) NodeMeta(limit int) []byte
  • func (m *Client) NotifyMsg(msg []byte)
  • func (m *Client) Stop()
  • func (m *Client) WatchKey(ctx context.Context, key string, f func(interface{}) bool)
  • func (m *Client) WatchPrefix(ctx context.Context, prefix string, f func(string, interface{}) bool)
• type Config
  • func (cfg *Config) RegisterFlags(f *flag.FlagSet, prefix string)
• type Mergeable
• type TCPTransport
  • func NewTCPTransport(config TCPTransportConfig) (*TCPTransport, error)
  • func (t *TCPTransport) DialTimeout(addr string, timeout time.Duration) (net.Conn, error)
  • func (t *TCPTransport) FinalAdvertiseAddr(ip string, port int) (net.IP, int, error)
  • func (t *TCPTransport) GetAutoBindPort() int
  • func (t *TCPTransport) PacketCh() <-chan *memberlist.Packet
  • func (t *TCPTransport) Shutdown() error
  • func (t *TCPTransport) StreamCh() <-chan net.Conn
  • func (t *TCPTransport) WriteTo(b []byte, addr string) (time.Time, error)
• type TCPTransportConfig
  • func (cfg *TCPTransportConfig) RegisterFlags(f *flag.FlagSet, prefix string)

Types

type Client

type Client struct {
	// contains filtered or unexported fields
}

Client implements kv.Client interface by using memberlist gossiping library.

func NewMemberlistClient

func NewMemberlistClient(cfg Config, codec codec.Codec) (*Client, error)

NewMemberlistClient creates new Client instance. If cfg.JoinMembers is set, it will also try to connect to these members and join the cluster. If that fails and AbortIfJoinFails is true, error is returned and no client is created.

func (*Client) CAS

func (m *Client) CAS(ctx context.Context, key string, f func(in interface{}) (out interface{}, retry bool, err error)) error

CAS is part of kv.Client implementation.

CAS expects that value returned by 'f' function implements Mergeable interface. If it doesn't, CAS fails immediately.

This method combines Compare-And-Swap with Merge: it calls 'f' function to get a new state, and then merges this new state into current state, to find out what the change was. Resulting updated current state is then CAS-ed to KV store, and change is broadcast to cluster peers. Merge function is called with CAS flag on, so that it can detect removals. If Merge doesn't result in any change (returns nil), then operation fails and is retried again. After too many failed retries, this method returns error.

func (*Client) Collect

func (m *Client) Collect(ch chan<- prometheus.Metric)

Collect returns extra metrics via supplied channel

func (*Client) Describe

func (m *Client) Describe(ch chan<- *prometheus.Desc)

Describe returns prometheus descriptions via supplied channel

func (*Client) Get

func (m *Client) Get(ctx context.Context, key string) (interface{}, error)

Get returns current value associated with given key. No communication with other nodes in the cluster is done here. Part of kv.Client interface.

func (*Client) GetBroadcasts

func (m *Client) GetBroadcasts(overhead, limit int) [][]byte

GetBroadcasts is method from Memberlist Delegate interface It returns all pending broadcasts (within the size limit)

func (*Client) GetListeningPort

func (m *Client) GetListeningPort() int

GetListeningPort returns port used for listening for memberlist communication. Useful when BindPort is set to 0.

func (*Client) JoinMembers

func (m *Client) JoinMembers(members []string) (int, error)

JoinMembers joins the cluster with given members. See https://godoc.org/github.com/hashicorp/memberlist#Memberlist.Join

func (*Client) LocalState

func (m *Client) LocalState(join bool) []byte

LocalState is method from Memberlist Delegate interface

This is "pull" part of push/pull sync (either periodic, or when new node joins the cluster). Here we dump our entire state -- all keys and their values. There is no limit on message size here, as Memberlist uses 'stream' operations for transferring this state.

func (*Client) MergeRemoteState

func (m *Client) MergeRemoteState(stateMsg []byte, join bool)

MergeRemoteState is method from Memberlist Delegate interface

This is 'push' part of push/pull sync. We merge incoming KV store (all keys and values) with ours.

func (*Client) NodeMeta

func (m *Client) NodeMeta(limit int) []byte

NodeMeta is method from Memberlist Delegate interface

func (*Client) NotifyMsg

func (m *Client) NotifyMsg(msg []byte)

NotifyMsg is method from Memberlist Delegate interface Called when single message is received, i.e. what our broadcastNewValue has sent.

func (*Client) Stop

func (m *Client) Stop()

Stop tries to leave memberlist cluster and then shutdown memberlist client. We do this in order to send out last messages, typically that ingester has LEFT the ring.

func (*Client) WatchKey

func (m *Client) WatchKey(ctx context.Context, key string, f func(interface{}) bool)

WatchKey watches for value changes for given key. When value changes, 'f' function is called with the latest value. Notifications that arrive while 'f' is running are coalesced into one subsequent 'f' call.

Watching ends when 'f' returns false, context is done, or this client is shut down.

Part of kv.Client interface.

func (*Client) WatchPrefix

func (m *Client) WatchPrefix(ctx context.Context, prefix string, f func(string, interface{}) bool)

WatchPrefix watches for any change of values stored under keys with given prefix. When change occurs, function 'f' is called with key and current value. Each change of the key results in one notification. If there are too many pending notifications ('f' is slow), some notifications may be lost.

Watching ends when 'f' returns false, context is done, or this client is shut down.

Part of kv.Client interface.

type Config

type Config struct {
	// Memberlist options.
	NodeName         string        `yaml:"node_name"`
	StreamTimeout    time.Duration `yaml:"stream_timeout"`
	RetransmitMult   int           `yaml:"retransmit_factor"`
	PushPullInterval time.Duration `yaml:"pull_push_interval"`
	GossipInterval   time.Duration `yaml:"gossip_interval"`
	GossipNodes      int           `yaml:"gossip_nodes"`

	// List of members to join
	JoinMembers      flagext.StringSlice `yaml:"join_members"`
	AbortIfJoinFails bool                `yaml:"abort_if_cluster_join_fails"`

	// Remove LEFT ingesters from ring after this timeout.
	LeftIngestersTimeout time.Duration `yaml:"left_ingesters_timeout"`

	// Timeout used when leaving the memberlist cluster.
	LeaveTimeout time.Duration `yaml:"leave_timeout"`

	TCPTransport TCPTransportConfig `yaml:",inline"`

	// Where to put custom metrics. Metrics are not registered, if this is nil.
	MetricsRegisterer prometheus.Registerer `yaml:"-"`
	MetricsNamespace  string                `yaml:"-"`
}

Config for memberlist-based Client

func (*Config) RegisterFlags

func (cfg *Config) RegisterFlags(f *flag.FlagSet, prefix string)

RegisterFlags registers flags.

type Mergeable

type Mergeable interface {
	// Merge with other value in place. Returns change, that can be sent to other clients.
	// If merge doesn't result in any change, returns nil.
	// Error can be returned if merging with given 'other' value is not possible.
	//
	// In order for state merging to work correctly, Merge function must have some properties. When talking about the
	// result of the merge in the following text, we don't mean the return value ("change"), but the
	// end-state of receiver. That means Result of A.Merge(B) is end-state of A.
	//
	// Idempotency:
	// 		Result of applying the same state "B" to state "A" (A.Merge(B)) multiple times has the same effect as
	// 		applying it only once. Only first Merge will return non-empty change.
	//
	// Commutativity:
	//		A.Merge(B) has the same result as B.Merge(A) (result being the end state of A or B respectively)
	//
	// Associativity:
	//		calling B.Merge(C), and then A.Merge(B) has the same result as
	//		calling A.Merge(B) first, and then calling A.Merge(C),
	//		that is, A will end up in the same state in both cases.
	//
	// LocalCAS flag is used when doing Merge as part of local CAS operation on KV store. It can be used to detect
	// missing entries, and generate tombstones. (This breaks commutativity and associativity [!] so it can *only* be
	// used when doing CAS operation)
	Merge(other Mergeable, localCAS bool) (change Mergeable, error error)

	// Describes the content of this mergeable value. Used by memberlist client to decide if
	// one change-value can invalidate some other value, that was received previously.
	// Invalidation can happen only if output of MergeContent is a superset of some other MergeContent.
	MergeContent() []string

	// Remove tombstones older than given limit from this mergeable.
	// If limit is zero time, remove all tombstones. Memberlist client calls this method with zero limit each
	// time when client is accessing value from the store. It can be used to hide tombstones from the clients.
	RemoveTombstones(limit time.Time)
}

Mergeable is an interface that values used in gossiping KV Client must implement. It allows merging of different states, obtained via gossiping or CAS function.

type TCPTransport

type TCPTransport struct {
	// contains filtered or unexported fields
}

TCPTransport is a memberlist.Transport implementation that uses TCP for both packet and stream operations ("packet" and "stream" are terms used by memberlist). It uses a new TCP connections for each operation. There is no connection reuse.

func NewTCPTransport

func NewTCPTransport(config TCPTransportConfig) (*TCPTransport, error)

NewTCPTransport returns a new tcp-based transport with the given configuration. On success all the network listeners will be created and listening.

func (*TCPTransport) DialTimeout

func (t *TCPTransport) DialTimeout(addr string, timeout time.Duration) (net.Conn, error)

DialTimeout is used to create a connection that allows memberlist to perform two-way communication with a peer.

func (*TCPTransport) FinalAdvertiseAddr

func (t *TCPTransport) FinalAdvertiseAddr(ip string, port int) (net.IP, int, error)

FinalAdvertiseAddr is given the user's configured values (which might be empty) and returns the desired IP and port to advertise to the rest of the cluster. (Copied from memberlist' net_transport.go)

func (*TCPTransport) GetAutoBindPort

func (t *TCPTransport) GetAutoBindPort() int

GetAutoBindPort returns the bind port that was automatically given by the kernel, if a bind port of 0 was given.

func (*TCPTransport) PacketCh

func (t *TCPTransport) PacketCh() <-chan *memberlist.Packet

PacketCh returns a channel that can be read to receive incoming packets from other peers.

func (*TCPTransport) Shutdown

func (t *TCPTransport) Shutdown() error

Shutdown is called when memberlist is shutting down; this gives the transport a chance to clean up any listeners.

func (*TCPTransport) StreamCh

func (t *TCPTransport) StreamCh() <-chan net.Conn

StreamCh returns a channel that can be read to handle incoming stream connections from other peers.

func (*TCPTransport) WriteTo

func (t *TCPTransport) WriteTo(b []byte, addr string) (time.Time, error)

WriteTo is a packet-oriented interface that fires off the given payload to the given address.

type TCPTransportConfig

type TCPTransportConfig struct {
	// BindAddrs is a list of addresses to bind to.
	BindAddrs flagext.StringSlice `yaml:"bind_addr"`

	// BindPort is the port to listen on, for each address above.
	BindPort int `yaml:"bind_port"`

	// Timeout used when making connections to other nodes to send packet.
	// Zero = no timeout
	PacketDialTimeout time.Duration `yaml:"packet_dial_timeout"`

	// Timeout for writing packet data. Zero = no timeout.
	PacketWriteTimeout time.Duration `yaml:"packet_write_timeout"`

	// WriteTo is used to send "UDP" packets. Since we use TCP, we can detect more errors,
	// but memberlist doesn't seem to cope with that very well.
	ReportWriteToErrors bool `yaml:"-"`

	// Transport logs lot of messages at debug level, so it deserves an extra flag for turning it on
	TransportDebug bool `yaml:"-"`

	// Where to put custom metrics. nil = don't register.
	MetricsRegisterer prometheus.Registerer `yaml:"-"`
	MetricsNamespace  string                `yaml:"-"`
}

TCPTransportConfig is a configuration structure for creating new TCPTransport.

func (*TCPTransportConfig) RegisterFlags

func (cfg *TCPTransportConfig) RegisterFlags(f *flag.FlagSet, prefix string)

RegisterFlags registers flags.