gossip

Documentation

Overview

Package gossip manages cluster membership, anti-entropy and failure detection for the local node.

At the gossip layer, a nodes state is represented as string key-value pairs which will be gossiped to the other nodes in the cluster. Therefore each node will have an eventually consistent view of the other nodes state.

Index

• type Config
  • func (c *Config) RegisterFlags(fs *pflag.FlagSet, prefix string)
  • func (c *Config) Validate() error
• type Entry
• type Gossip
  • func New(nodeID string, config *Config, streamLn net.Listener, packetLn net.PacketConn, ...) *Gossip
  • func (g *Gossip) Close() error
  • func (g *Gossip) DeleteLocal(key string)
  • func (g *Gossip) Join(addrs []string) ([]string, error)
  • func (g *Gossip) Leave() error
  • func (g *Gossip) LocalNode() *NodeState
  • func (g *Gossip) Metrics() *Metrics
  • func (g *Gossip) Node(id string) (*NodeState, bool)
  • func (g *Gossip) Nodes() []NodeMetadata
  • func (g *Gossip) UpsertLocal(key, value string)
• type Metrics
  • func (m *Metrics) Register(reg *prometheus.Registry)
• type NodeMetadata
• type NodeState
• type Watcher

Types

type Config

type Config struct {
	// BindAddr is the address to bind to listen for gossip traffic.
	BindAddr string `json:"bind_addr" yaml:"bind_addr"`

	// AdvertiseAddr is the address to advertise to other nodes.
	AdvertiseAddr string `json:"advertise_addr" yaml:"advertise_addr"`

	// Interval is the rate to initiate a gossip round.
	Interval time.Duration `json:"interval" yaml:"interval"`

	// MaxPacketSize is the maximum size of any packet sent.
	MaxPacketSize int `json:"max_packet_size" yaml:"max_packet_size"`
}

func (*Config) RegisterFlags

func (c *Config) RegisterFlags(fs *pflag.FlagSet, prefix string)

func (*Config) Validate

func (c *Config) Validate() error

type Entry

type Entry struct {
	Key     string `json:"key" codec:"key"`
	Value   string `json:"value" codec:"value"`
	Version uint64 `json:"version" codec:"version"`

	// Internal indicates whether this is an internal entry.
	Internal bool `json:"internal" codec:"internal"`
	// Deleted indicates whether this entry represents a deleted key.
	Deleted bool `json:"deleted" codec:"deleted"`
}

Entry represents a versioned key-value pair state.

type Gossip

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

func New

func New(
	nodeID string,
	config *Config,
	streamLn net.Listener,
	packetLn net.PacketConn,
	watcher Watcher,
	logger log.Logger,
) *Gossip

func (*Gossip) Close

func (g *Gossip) Close() error

Close stops gossiping and closes all listeners.

To leave gracefully, first call Leave, otherwise other nodes in the cluster will detect this nodes as failed rather than as having left.

func (*Gossip) DeleteLocal

func (g *Gossip) DeleteLocal(key string)

DeleteLocal deletes the local state entry with the given key.

func (*Gossip) Join

func (g *Gossip) Join(addrs []string) ([]string, error)

Join attempts to join an existing cluster by syncronising with the nodes at the given addresses.

The addresses may contain either IP addresses or domain names. When a domain name is used, the domain is resolved and each resolved IP address is attempted. Will also periodically re-resolve the joined domains and attempt to gossip with any unknown nodes. If the port is omitted the default bind port is used.

Returns the IDs of joined nodes. Or if addresses were provided by no nodes could be joined an error is returned. Note if a domain was provided that only resolved to the current node then Join will return nil.

func (*Gossip) Leave

func (g *Gossip) Leave() error

Leave gracefully leaves the cluster.

This block while it attempts to notify upto 3 nodes in the cluster that the node is leaving to ensure the status update is propagated.

After the node has left it's state should not be updated again.

Returns an error if no nodes could be notified.

func (*Gossip) LocalNode

func (g *Gossip) LocalNode() *NodeState

LocalNode returns the state of the local node.

func (*Gossip) Metrics

func (g *Gossip) Metrics() *Metrics

func (*Gossip) Node

func (g *Gossip) Node(id string) (*NodeState, bool)

Node returns the known state for the node with the given ID.

func (*Gossip) Nodes

func (g *Gossip) Nodes() []NodeMetadata

Nodes returns the known metadata of each node in the cluster.

func (*Gossip) UpsertLocal

func (g *Gossip) UpsertLocal(key, value string)

UpsertLocal updates the local node state entry with the given key.

type Metrics

type Metrics struct {
	// ConnectionsInbound is the total number of incoming stream
	// connections.
	ConnectionsInbound prometheus.Counter

	// StreamBytesInbound is the total number of read bytes via a stream
	// connection.
	StreamBytesInbound prometheus.Counter

	// PacketBytesInbound is the total number of read bytes via a packet
	// connection.
	PacketBytesInbound prometheus.Counter

	// ConnectionsOutbound is the total number of outgoing stream
	// connections.
	ConnectionsOutbound prometheus.Counter

	// StreamBytesOutbound is the total number of written bytes via a stream
	// connection.
	StreamBytesOutbound prometheus.Counter

	// PacketBytesOutbound is the total number of written bytes via a packet
	// connection.
	PacketBytesOutbound prometheus.Counter

	// DigestEntriesInbound is the total number of incoming digest entries.
	DigestEntriesInbound prometheus.Counter

	// DeltaEntriesInbound is the total number of incoming delta entries.
	DeltaEntriesInbound prometheus.Counter

	// DigestEntriesOutbound is the total number of outgoing digest entries.
	DigestEntriesOutbound prometheus.Counter

	// DeltaEntriesOutbound is the total number of outgoing delta entries.
	DeltaEntriesOutbound prometheus.Counter

	// Entries is the number of entries labelled by node_id, deleted and
	// internal.
	Entries *prometheus.GaugeVec
}

func (*Metrics) Register

func (m *Metrics) Register(reg *prometheus.Registry)

type NodeMetadata

type NodeMetadata struct {
	// ID is a unique identifier for the node.
	ID string `json:"id"`

	// Addr is the gossip address of the node.
	Addr string `json:"addr"`

	// Version is the latest known version of the node.
	Version uint64 `json:"version"`

	// Left indicates whether the node has left the cluster.
	Left bool `json:"left"`

	// Unreachable indicates whether the node is considered unreachable.
	Unreachable bool `json:"unreachable"`

	// Expiry contains the time the node state will expire. This is only set
	// if the node is considered left or unreachable until the expiry.
	Expiry time.Time `json:"expiry"`
}

NodeMetadata contains the known metadata about the node.

type NodeState

type NodeState struct {
	NodeMetadata

	Entries []Entry
}

NodeState contains the known state for the node.

type Watcher

type Watcher interface {
	// OnJoin notifies that a new node joined the cluster.
	OnJoin(nodeID string)

	// OnLeave notifies that a node gracefully left the cluster.
	OnLeave(nodeID string)

	// OnReachable notifies that a node that was considered unreachable has
	// recovered.
	OnReachable(nodeID string)

	// OnUnreachable notifies that a node is considered unreachable.
	OnUnreachable(nodeID string)

	// OnUpsertKey notifies that a nodes state key has been updated.
	OnUpsertKey(nodeID, key, value string)

	// OnDeleteKey notifies that a nodes state key has been deleted.
	OnDeleteKey(nodeID, key string)

	// OnExpired notifies that a nodes state has expired and been removed.
	OnExpired(nodeID string)
}

Watcher is used to receive notifications when the known remote node state changes.

The implementations of Watcher must not block. Watcher is also called with the state mutex held so should not call back to Gossip.