gogossip

README

go-gossip

This is Go implementation of the Gossip protocol.

Gossip is a communication protocol that delivers messages in a distributed system.
The point is that each node transmits data while periodically exchanging metadata based on TCP/UDP without a broadcast master.
In general, each node periodically performs health checks of other nodes and communicates with them, but this library relies on an externally imported discovery layer.
The gossip protocol is divided into two main categories: Push and Pull. If implemented as a push, it becomes inefficient if a large number of peers are already infected. If implemented as Pull, it will propagate efficiently, but there is a point to be concerned about because the message that needs to be propagated to the peer needs to be managed.
In this project, by properly mixing push and pull methods, it is possible to efficiently propagate even when a large number of peers are already infected, and the goal is to reduce the difficulty of managing messages that need to be propagated to other peers.
It works almost identically to the existing Push-based gossip protocol. It selects a set number of random peers for a new message and send the message. The difference here is that the peer that receives the 'Push' message sends an ACK message to the sender.
If the target peer does not operate normally, or if the message has already received before, does not send ACK message.
The sender collects the number of ACKs to see if it has received a majority of the number of messages it has sent. If a 'Push' message is sent to 3 random peers, the 'Push' process will correctly end only when two or more 'ACK' are received.

What if I didn't get more than a majority?

Suppose you sent a 'Push' message to 3 random peers, but only received 1 'ACK'. The sender adds a certain value to the previously set 3 and sends a 'PullSync' to 5 peers for example. The message is sent with the id of the data the sender was trying to propagate. The peer receiving the 'PullSync' sends a 'Pull request' including the data ID to the sender of the message. Finally, the original sender peer sends a 'Pull response' containing the requested data.
If implemented as above, the inefficiency of the push-based gossip protocol, which requires sending and receiving messages between already infected peers, and the hassle of managing data to respond to pull requests can be reduced.

Take a look at the list of supported features below.

• Message propagation
• Secure transport

It's forcus on gossip through relying on discovery layer from outside.

Layer

<< LAYER IMAGE >>

Discovery layer

Discovery layer serves as the managed peer table. That could be static peer table, also could dynamic peer table(like DHT).
The required(MUST) method is Gossipiers. Gossipiers is used to select random peers to send gossip messages to.

type Discovery interface {
	Gossipiers() []string
}

(It means DHT or importers covers registration to Gossip protocol)
Gossipiers returns array of raw addresses. The raw addresses will validate when gossip layer. Even if rely on validation of externally imported methods, We need to double-check internally here(trust will make unexpected panic).
A consideration is whether Gossipiers return value actually needs a peer ID.
In generally, there is need each peer's meta datas(about memberlist), the unique id is required. However, since this library does not support health checks, the peer id is not required from a metadata required point of view.
In addition, if you receive and use a unique ID from outside, the dependency relationship becomes severe, so I think it is correct not to have an peer id.
When making a request, such as checking gossip node stat or something, we decide to use the raw address.

Gossip layer

Gossip layer serves core features that propagating gossip messages and relay data to application when needed.
For serve that, it's detect packet and handles them correctly to the packet types.
There is three tasks what this layer have to do.

1. The node should be able to be a gossip culprit. Provide surface interface for application programs to push gossip messages.
2. MUST have to respond like, PUSH_MESSAGE -> PUSH_ACK, PULL_SYN -> PULL_REQUEST, PULL_REQUEST -> PULL_RESPONSE.
3. Detects is the gossip message already exist in memory and relay the gossip messages to the application if necessary.

The gossip node needs two buffers: a buffer where the application program can receive gossip messages and a buffer to temporarily store it to propagate to other gossip nodes.
I decided to use the LRU cache for message temporary storage for propagation. Gossip messages that no longer propagate are likely not to be referenced by other nodes in the future. One thing to consider here is that the size of the cache should be much larger than the number of times a node can make PUSH requests. Otherwise, the cache will be replaced as soon as it starts propagate gossip messages.
We need to find a suitable config values.

Transport/Security layer

Security layer resolve the secure between peer to peer trnasmission. It should be possible to add multiple encryption algorithms. I'm just considering a method of encrypting and decrypting using a passphrase.

Packet

┏---------------------┓
| Label | Actual data |
┗---------------------┛

Label

┏----------------------------------------------------------┓
| Packet type| Encrypt algorithm | Actual data size (May?) | 
┗----------------------------------------------------------┛

Packet type (1 byte)

1: PushMessage
2: PullSync
3: PullRequest
4: PullResponse

Encrypt alogrithm (1 byte)

1: AES-256-CBC

Actual data size (4 byte); BigEndian ordered uint32
This is not necessary unless you add a specific flag (eg checksum) after the data.

PULL_REQUEST

PULL_RESPONSE

Documentation

Index

• Constants
• func AddLabelFromPacket(packet Packet, encType EncryptType) ([]byte, error)
• type Cipher
  • func NewCipher(kind EncryptType) Cipher
  • func (s *Cipher) Is(kind EncryptType) bool
• type CipherMethod
• type Config
• type Discovery
• type EncryptType
  • func RemoveLabelFromPacket(d []byte) ([]byte, byte, EncryptType, error)
  • func (e EncryptType) String() string
• type Gossiper
  • func NewGossiper(discv Discovery, transport Transport, cfg *Config) (*Gossiper, error)
  • func (g *Gossiper) MessagePipe() chan []byte
  • func (g *Gossiper) Push(buf []byte)
  • func (g *Gossiper) SelectRandomPeers(n int) []string
  • func (g *Gossiper) Start()
• type Packet
• type PullRequest
  • func (req *PullRequest) ID() uint32
  • func (req *PullRequest) Kind() byte
• type PullResponse
  • func (res *PullResponse) ID() uint32
  • func (res *PullResponse) Kind() byte
• type PullSync
  • func (p *PullSync) ID() uint32
  • func (p *PullSync) Kind() byte
• type PushAck
  • func (p *PushAck) ID() uint32
  • func (p *PushAck) Kind() byte
• type PushMessage
  • func (p *PushMessage) ID() uint32
  • func (p *PushMessage) Kind() byte
• type Transport

Constants

const (
	//
	GossipNumber = 3
	// Set to twice the predicted value of messages to occur per second
	MessageCacheSize = 512
	//
	MessagePipeSize = 4096
	//
	MaxPacketSize = 8 * 1024
	//
	AckTimeout = 300 * time.Millisecond
)

const (
	PushMessageType  = 0x01
	PushAckType      = 0x02
	PullSyncType     = 0x03
	PullRequestType  = 0x04
	PullResponseType = 0x05
)

const (
	TEMP_NONE_ENC   = 0x00
	AES256_CBC_TYPE = 0x01
)

Functions

func AddLabelFromPacket

func AddLabelFromPacket(packet Packet, encType EncryptType) ([]byte, error)

Types

type Cipher

type Cipher struct {
	CipherMethod
	// contains filtered or unexported fields
}

func NewCipher

func NewCipher(kind EncryptType) Cipher

func (*Cipher) Is

func (s *Cipher) Is(kind EncryptType) bool

type CipherMethod

type CipherMethod interface {
	Encrypt(string, []byte) ([]byte, error)
	Decrypt(string, []byte) ([]byte, error)
}

type Config

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

type Discovery

type Discovery interface {
	// The imported DHT object must have method Gossipiers implement thread
	// safety. It returns all of the peer that target of gossip protocol by
	// array of raw addresses. The raw address validate when before send gossip.
	//
	// It'll used 'SelectRandomPeers' in Gossiper.
	Gossipiers() []string
}

OR use name 'Registration'

type EncryptType

type EncryptType byte

func RemoveLabelFromPacket

func RemoveLabelFromPacket(d []byte) ([]byte, byte, EncryptType, error)

func (EncryptType) String

func (e EncryptType) String() string

type Gossiper

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

func NewGossiper

func NewGossiper(discv Discovery, transport Transport, cfg *Config) (*Gossiper, error)

func (*Gossiper) MessagePipe

func (g *Gossiper) MessagePipe() chan []byte

func (*Gossiper) Push

func (g *Gossiper) Push(buf []byte)

func (*Gossiper) SelectRandomPeers

func (g *Gossiper) SelectRandomPeers(n int) []string

func (*Gossiper) Start

func (g *Gossiper) Start()

type Packet

type Packet interface {
	ID() uint32
	Kind() byte
}

type PullRequest

type PullRequest struct {
	Target [8]byte
	// contains filtered or unexported fields
}

func (*PullRequest) ID

func (req *PullRequest) ID() uint32

func (*PullRequest) Kind

func (req *PullRequest) Kind() byte

type PullResponse

type PullResponse struct {
	Target [8]byte
	Data   []byte
	// contains filtered or unexported fields
}

func (*PullResponse) ID

func (res *PullResponse) ID() uint32

func (*PullResponse) Kind

func (res *PullResponse) Kind() byte

type PullSync

type PullSync struct {
	Target [8]byte
	// contains filtered or unexported fields
}

func (*PullSync) ID

func (p *PullSync) ID() uint32

func (*PullSync) Kind

func (p *PullSync) Kind() byte

type PushAck

type PushAck struct {
	Key [8]byte
	// contains filtered or unexported fields
}

func (*PushAck) ID

func (p *PushAck) ID() uint32

func (*PushAck) Kind

func (p *PushAck) Kind() byte

type PushMessage

type PushMessage struct {
	Key  [8]byte
	Data []byte
	// contains filtered or unexported fields
}

func (*PushMessage) ID

func (p *PushMessage) ID() uint32

func (*PushMessage) Kind

func (p *PushMessage) Kind() byte

type Transport

type Transport interface {
	ReadFromUDP(b []byte) (n int, addr *net.UDPAddr, err error)
	WriteToUDP(b []byte, addr *net.UDPAddr) (int, error)
}