swarm

Documentation

Overview

Package swarm implements the exchange of information between Skipper instances using a gossip protocol called SWIM. This aims at a solution that can work in a context of multiple readers and writers, with the guarantee of low latency, weakly consistent data, from which derives the decision to use such protocol. As an example the implementation of the filter clusterRatelimit uses the swarm data exchange to have a global state of current requests.

A swarm instance needs to find some of it's peers before joining the cluster. Current implementations to find peers are swarmKubernetes to find skipper instances running in a Kubernetes cluster and swarmFake for testing.

Background information:

The current skipper implementation uses hashicorp's memberlist, https://github.com/hashicorp/memberlist, which is an implementation of the swim protocol. You can find a detailed paper at http://www.cs.cornell.edu/~asdas/research/dsn02-SWIM.pdf.

Quote from a nice overview https://prakhar.me/articles/swim/

The SWIM or the Scalable Weakly-consistent Infection-style process
group Membership protocol is a protocol used for maintaining
membership amongst processes in a distributed system.

While starting, Skipper will find its swarm peers through the Kubernetes API server. It will do that using a label selector query to find Pods of the swarm.

Index

• Constants
• Variables
• func NewNodeInfoClient(o Options) (nodeInfoClient, func())
• func NewNodeInfoClientFake(o Options) *nodeInfoClientFake
• func NewNodeInfoClientKubernetes(o Options) *nodeInfoClientKubernetes
• type ClientKubernetes
  • func NewClientKubernetes(kubernetesInCluster bool, kubernetesURL string) (*ClientKubernetes, error)
  • func (c *ClientKubernetes) Get(s string) (*http.Response, error)
  • func (c *ClientKubernetes) Stop()
• type EntryPoint
• type KubernetesOptions
• type Message
• type NodeInfo
  • func NewFakeNodeInfo(name string, addr net.IP, port uint16) *NodeInfo
  • func NewStaticNodeInfo(name, addr string) (*NodeInfo, error)
  • func (ni NodeInfo) String() string
• type NodeState
• type Options
• type Self
• type StaticSwarm
  • func NewStaticSwarm(self *NodeInfo, all []*NodeInfo) *StaticSwarm
  • func (s *StaticSwarm) GetNodeInfo() ([]*NodeInfo, error)
  • func (s *StaticSwarm) Self() *NodeInfo
• type Swarm
  • func Join(o Options, self *NodeInfo, nodes []*NodeInfo, cleanupF func()) (*Swarm, error)
  • func NewSwarm(optr *Options) (*Swarm, error)
  • func Start(o Options) (*Swarm, error)
  • func (s *Swarm) Broadcast(m interface{}) error
  • func (s *Swarm) Leave()
  • func (s *Swarm) Local() *NodeInfo
  • func (s *Swarm) ShareValue(key string, value interface{}) error
  • func (s *Swarm) Values(key string) map[string]interface{}

Constants

const (
	// DefaultNamespace is the default namespace where swarm searches for peer information
	DefaultNamespace = "kube-system"
	// DefaultLabelSelectorKey is the default label key to select Pods for peer information
	DefaultLabelSelectorKey = "application"
	// DefaultLabelSelectorValue is the default label value to select Pods for peer information
	DefaultLabelSelectorValue = "skipper-ingress"
)

const (
	// DefaultMaxMessageBuffer is the default maximum size of the
	// exchange packets send out to peers.
	DefaultMaxMessageBuffer = 1 << 22
	// DefaultPort is used as default to connect to other
	// known swarm peers.
	DefaultPort = 9990
	// DefaultLeaveTimeout is the default timeout to wait for responses
	// for a leave message send by this instance to other peers.
	DefaultLeaveTimeout = time.Duration(5 * time.Second)
)

Variables

var (
	ErrUnknownSwarm = errors.New("unknown swarm type")
)

Functions

func NewNodeInfoClient

func NewNodeInfoClient(o Options) (nodeInfoClient, func())

func NewNodeInfoClientFake

func NewNodeInfoClientFake(o Options) *nodeInfoClientFake

func NewNodeInfoClientKubernetes

func NewNodeInfoClientKubernetes(o Options) *nodeInfoClientKubernetes

Types

type ClientKubernetes

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

ClientKubernetes is the client to access kubernetes resources to find the peers to join a swarm.

func NewClientKubernetes

func NewClientKubernetes(kubernetesInCluster bool, kubernetesURL string) (*ClientKubernetes, error)

NewClientKubernetes creates and initializes a Kubernetes client to find peers. A partial copy of the Kubernetes dataclient.

func (*ClientKubernetes) Get

func (c *ClientKubernetes) Get(s string) (*http.Response, error)

Get does the http GET call to kubernetes API to find the initial peers of a swarm.

func (*ClientKubernetes) Stop

func (c *ClientKubernetes) Stop()

type EntryPoint

type EntryPoint interface {
	Nodes() []*NodeInfo
}

EntryPoint knows its peers of nodes which contains itself

type KubernetesOptions

type KubernetesOptions struct {
	KubernetesInCluster  bool
	KubernetesAPIBaseURL string
	Namespace            string
	LabelSelectorKey     string
	LabelSelectorValue   string
}

KubernetesOptions are Kubernetes specific swarm options, that are needed to find peers.

type Message

type Message struct {
	Source string
	Value  interface{}
}

type NodeInfo

type NodeInfo struct {
	Name string
	Addr net.IP
	Port uint16
}

NodeInfo is a value object tat contains information about swarm cluster nodes, that is required to access member nodes.

func NewFakeNodeInfo

func NewFakeNodeInfo(name string, addr net.IP, port uint16) *NodeInfo

NewFakeNodeInfo used to create a FakeSwarm

func NewStaticNodeInfo

func NewStaticNodeInfo(name, addr string) (*NodeInfo, error)

func (NodeInfo) String

func (ni NodeInfo) String() string

String will only show initial peers when created this peer

type NodeState

type NodeState int

NodeState represents the current state of a cluster node known by this instance.

const (
	Initial NodeState = iota
	Connected
	Disconnected
)

type Options

type Options struct {

	// MaxMessageBuffer is the maximum size of the exchange
	// packets send out to peers.
	MaxMessageBuffer int

	// LeaveTimeout is the timeout to wait for responses for a
	// leave message send by this instance to other peers.
	LeaveTimeout time.Duration

	// SwarmPort port to listen for incoming swarm packets.
	SwarmPort uint16

	// KubernetesOptions are options required to find your peers in Kubernetes
	KubernetesOptions *KubernetesOptions

	StaticSwarm *StaticSwarm

	// FakeSwarm enable a test swarm
	FakeSwarm bool

	// FakeSwarmLocalNode is the node name of the local node
	// joining a fakeSwarm to have better log output
	FakeSwarmLocalNode string

	// Debug enables swarm debug logs and also enables memberlist logs
	Debug bool
	// contains filtered or unexported fields
}

Options configure swarm objects.

type Self

type Self interface {
	Node() *NodeInfo
}

Self can return itself as NodeInfo

type StaticSwarm

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

func NewStaticSwarm

func NewStaticSwarm(self *NodeInfo, all []*NodeInfo) *StaticSwarm

func (*StaticSwarm) GetNodeInfo

func (s *StaticSwarm) GetNodeInfo() ([]*NodeInfo, error)

func (*StaticSwarm) Self

func (s *StaticSwarm) Self() *NodeInfo

type Swarm

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

Swarm is the main type for exchanging low latency, weakly consistent information with other skipper peers.

func Join

func Join(o Options, self *NodeInfo, nodes []*NodeInfo, cleanupF func()) (*Swarm, error)

Join will join given Swarm peers and return an initialiazed Swarm object if successful.

func NewSwarm

func NewSwarm(optr *Options) (*Swarm, error)

NewSwarm creates a Swarm for given Options.

func Start

func Start(o Options) (*Swarm, error)

Start will find Swarm peers based on the chosen swarm type and join the Swarm.

func (*Swarm) Broadcast

func (s *Swarm) Broadcast(m interface{}) error

Broadcast sends a broadcast message with a value to all peers.

func (*Swarm) Leave

func (s *Swarm) Leave()

Leave sends a signal for the local node to leave the Swarm.

func (*Swarm) Local

func (s *Swarm) Local() *NodeInfo

Local is a getter to the local member of a swarm.

func (*Swarm) ShareValue

func (s *Swarm) ShareValue(key string, value interface{}) error

ShareValue sends a broadcast message with a sharedValue to all peers. It implements the ratelimit.Swarmer interface.

func (*Swarm) Values

func (s *Swarm) Values(key string) map[string]interface{}

Values sends a request and wait blocking for a response. It implements the ratelimit.Swarmer interface.