easyraft

README

An easy to use customizable library to make your Go application Distributed, Highly available, Fault Tolerant etc... using Hashicorp's Raft library which implements the Raft Consensus Algorithm.

Features

• Configure and start a fully functional Raft node by writing ~10 lines of code
• Automatic Node discovery (nodes are discovering each other using Discovery method)
  1. Built-in discovery methods:
     1. Static Discovery (having a fixed list of nodes addresses)
     2. mDNS Discovery for local network node discovery
     3. Kubernetes discovery
• Cloud Native because of kubernetes discovery and easy to load balance features
• Automatic forward to leader - you can contact any node to perform operations, everything will be forwarded to the actual leader node
• Node monitoring/removal - the nodes are monitoring each other and if there are some failures then the offline nodes get removed automatically from cluster
• Simplified state machine - there is an already implemented generic state machine which handles the basic operations and routes requests to State Machine Services (see Examples)
• All layers are customizable - you can select or implement your own State Machine Service, Message Serializer and Discovery Method
• gRPC transport layer - the internal communications are done through gRPC based communication, if needed you can add your own services

Note: snapshots are not supported at the moment, will be handled at later point Note: at the moment the communication between nodes are insecure, I recommend to not expose that port

Get Started

You can create a simple EasyRaft Node with local mDNS discovery, an in-memory Map service and MsgPack as serializer(this is the only one built-in at the moment)

import (
"github.com/ksrichard/easyraft"
"github.com/ksrichard/easyraft/discovery"
"github.com/ksrichard/easyraft/fsm"
"github.com/ksrichard/easyraft/serializer"
)

func main() {
    raftPort := 5000
    discoveryPort := 5001
    dataDir := "s1"
    node, err := easyraft.NewNode(
        raftPort,
        discoveryPort,
        dataDir,
        []fsm.FSMService{fsm.NewInMemoryMapService()},
        serializer.NewMsgPackSerializer(),
        discovery.NewMDNSDiscovery(),
        false,
    )
    
    if err != nil {
        panic(err)
    }
    stoppedCh, err := node.Start()
    if err != nil {
        panic(err)
    }
    defer node.Stop()
}

Examples

Examples can be found in the examples directory

Build

To regenerate gRPC code and install dependencies simply run make install

TODO

• Add more examples
• Test coverage
• Secure communication between nodes (SSL/TLS)
• Backup/Restore backup handling
• Allow configuration option to pass any custom raft.FSM

Documentation

Index

• func ApplyOnLeader(node *Node, payload []byte) (interface{}, error)
• func GetPeerDetails(address string) (*grpc.GetDetailsResponse, error)
• type ClientGrpcServices
  • func NewClientGrpcService(node *Node) *ClientGrpcServices
  • func (s *ClientGrpcServices) ApplyLog(ctx context.Context, request *rgrpc.ApplyRequest) (*rgrpc.ApplyResponse, error)
  • func (s *ClientGrpcServices) GetDetails(context.Context, *rgrpc.GetDetailsRequest) (*rgrpc.GetDetailsResponse, error)
• type Node
  • func NewNode(raftPort, discoveryPort int, dataDir string, services []fsm.FSMService, ...) (*Node, error)
  • func (n *Node) NotifyJoin(node *memberlist.Node)
  • func (n *Node) NotifyLeave(node *memberlist.Node)
  • func (n *Node) NotifyUpdate(_ *memberlist.Node)
  • func (n *Node) RaftApply(request interface{}, timeout time.Duration) (interface{}, error)
  • func (n *Node) Start() (chan interface{}, error)
  • func (n *Node) Stop()

Functions

func ApplyOnLeader

func ApplyOnLeader(node *Node, payload []byte) (interface{}, error)

func GetPeerDetails

func GetPeerDetails(address string) (*grpc.GetDetailsResponse, error)

Types

type ClientGrpcServices

type ClientGrpcServices struct {
	Node *Node
	rgrpc.UnimplementedRaftServer
}

func NewClientGrpcService

func NewClientGrpcService(node *Node) *ClientGrpcServices

func (*ClientGrpcServices) ApplyLog

func (s *ClientGrpcServices) ApplyLog(ctx context.Context, request *rgrpc.ApplyRequest) (*rgrpc.ApplyResponse, error)

func (*ClientGrpcServices) GetDetails

func (s *ClientGrpcServices) GetDetails(context.Context, *rgrpc.GetDetailsRequest) (*rgrpc.GetDetailsResponse, error)

type Node

type Node struct {
	ID            string
	RaftPort      int
	DiscoveryPort int

	Raft             *raft.Raft
	GrpcServer       *ggrpc.Server
	DiscoveryMethod  discovery.DiscoveryMethod
	TransportManager *transport.Manager
	Serializer       serializer.Serializer
	// contains filtered or unexported fields
}

func NewNode

func NewNode(raftPort, discoveryPort int, dataDir string, services []fsm.FSMService, serializer serializer.Serializer, discoveryMethod discovery.DiscoveryMethod, snapshotEnabled bool) (*Node, error)

NewNode returns an EasyRaft node

func (*Node) NotifyJoin

func (n *Node) NotifyJoin(node *memberlist.Node)

NotifyJoin triggered when a new Node has been joined to the cluster (discovery only) and capable of joining the Node to the raft cluster

func (*Node) NotifyLeave

func (n *Node) NotifyLeave(node *memberlist.Node)

NotifyLeave triggered when a Node becomes unavailable after a period of time it will remove the unavailable Node from the Raft cluster

func (*Node) NotifyUpdate

func (n *Node) NotifyUpdate(_ *memberlist.Node)

func (*Node) RaftApply

func (n *Node) RaftApply(request interface{}, timeout time.Duration) (interface{}, error)

RaftApply is used to apply any new logs to the raft cluster this method does automatic forwarding to Leader Node

func (*Node) Start

func (n *Node) Start() (chan interface{}, error)

Start starts the Node and returns a channel that indicates, that the node has been stopped properly

func (*Node) Stop

func (n *Node) Stop()

Stop stops the node and notifies on stopped channel returned in Start