raft

README

General instructions for testing:

• A test file is a single file with a collection of test cases.

• The test file should be named with a desired filename followed by _test.go, like yourfilename_test.go.

• Each test case is a function with its name beginning with the word Test followed by a word or a phrase that starts with a capital letter. For example TestRequestVote and TestRaft are valid test case names while Testrequestvote, TestrequestVote and Testraft are invalid test case names.

• The first and only parameter for every test case function is t *testing.T.

• For handling failure conditions, call t.Error() or t.Fail(). These are similar to fmt.Print(). Use fmt.Errorf() if you want formatting similar to fmt.Printf().

• t.Log() can be used to provide debug information.

• Check the test cases written in the file raft_test.go for reference.

• Please write proper documentation for the tests, describing what feature it tests and how it manages to do so. This can be written in a comment block before the corresponding test case function.

How to write new test cases:

• The functions in the file testing_utils.go, should contain the necessary functions needed for testing, for example, functions for setting up a Raft system, crashing a single node, destroying the whole system, counting number of leaders, etc.

• Always start a test by calling the start_test() function which should return a pointer to the testing struct of type testing_st. The parameters to start_test are the t *testing.T object and the number of nodes that should exist in the Raft system to be created.

• testing_st is a struct that needs to be created for every test case. This will hold the RaftNode structs and other data related to individual nodes so that a global view of the system is available.

• Perform operations on the testing system and check its status using the functions in testing_utils.go.

• End the test case by using end_test() and passing the testing system to it as a parameter.

Documentation

Index

• Variables
• func CheckErrorFatal(err error)
• type NodeMetadata
• type RaftNode
  • func InitializeNode(n_replica int32, rid int, keyvalue_addr string) *RaftNode
  • func Setup_raft_node(ctx context.Context, id int, n_replicas int, testing bool) *RaftNode
  • func (node *RaftNode) AppendEntries(ctx context.Context, in *protos.AppendEntriesMessage) (*protos.AppendEntriesResponse, error)
  • func (node *RaftNode) ApplyToStateMachine(ctx context.Context, testing bool)
  • func (node *RaftNode) ConnectToPeerReplicas(ctx context.Context, rep_addrs []string)
  • func (node *RaftNode) Connect_raft_node(ctx context.Context, id int, rep_addrs []string, testing bool)
  • func (node *RaftNode) DeleteHandler(w http.ResponseWriter, r *http.Request)
  • func (node *RaftNode) GetHandler(w http.ResponseWriter, r *http.Request)
  • func (node *RaftNode) GetLock(where string)
  • func (node *RaftNode) GetRLock(where string)
  • func (node *RaftNode) HeartBeats(ctx context.Context)
  • func (node *RaftNode) LeaderSendAE(parent_ctx context.Context, replica_id int32, upper_index int32, ...) (status bool)
  • func (node *RaftNode) LeaderSendAEs(msg_type string, msg *protos.AppendEntriesMessage, upper_index int32, ...)
  • func (node *RaftNode) ListenForShutdown(master_cancel context.CancelFunc)
  • func (node *RaftNode) PersistToStorage()
  • func (node *RaftNode) PostHandler(w http.ResponseWriter, r *http.Request)
  • func (node *RaftNode) PutHandler(w http.ResponseWriter, r *http.Request)
  • func (node *RaftNode) ReadCommand(key string) (string, error)
  • func (node *RaftNode) ReleaseLock(where string)
  • func (node *RaftNode) ReleaseRLock(where string)
  • func (node *RaftNode) RequestVote(ctx context.Context, in *protos.RequestVoteMessage) (*protos.RequestVoteResponse, error)
  • func (node *RaftNode) RestoreFromStorage(storage *Storage)
  • func (node *RaftNode) RunElectionTimer(parent_ctx context.Context)
  • func (node *RaftNode) StaleReadCheck(heartbeat_success chan bool)
  • func (node *RaftNode) StartElection(ctx context.Context)
  • func (node *RaftNode) StartGRPCServer(ctx context.Context, grpc_address string, listener *net.TCPListener, ...)
  • func (node *RaftNode) StartKVStore(ctx context.Context, addr string, num int, testing bool)
  • func (node *RaftNode) StartRaftServer(ctx context.Context, addr string, testing bool)
  • func (node *RaftNode) TestHandler(w http.ResponseWriter, r *http.Request)
  • func (node *RaftNode) ToCandidate(ctx context.Context)
  • func (node *RaftNode) ToFollower(ctx context.Context, term int32)
  • func (node *RaftNode) ToLeader(ctx context.Context)
  • func (node *RaftNode) WriteCommand(operation []string, client string) (bool, error)
• type RaftNodeState
• type Storage
  • func NewStorage() *Storage
  • func (stored *Storage) Get(key string, filename string) (interface{}, bool)
  • func (stored *Storage) HasData(filename string) bool
  • func (stored *Storage) ReadFile(filename string)
  • func (stored *Storage) Set(key string, value interface{})
  • func (stored *Storage) WriteFile(filename string)

Variables

var Blue = "\033[34m"

var Cyan = "\033[36m"

var Gray = "\033[37m"

var Green = "\033[32m"

var Purple = "\033[35m"

var Red = "\033[31m"

var Reset = "\033[0m"

var White = "\033[97m"

var Yellow = "\033[33m"

Functions

func CheckErrorFatal

func CheckErrorFatal(err error)

Types

type NodeMetadata

type NodeMetadata struct {
	Master_ctx    context.Context    // A context derived from the master context for graceful shutdown
	Master_cancel context.CancelFunc // The cancel function for the above master context
	// contains filtered or unexported fields
}

Store metadata related to the key value store and the raft node.

type RaftNode

type RaftNode struct {
	protos.UnimplementedConsensusServiceServer

	Meta *NodeMetadata
	// contains filtered or unexported fields
}

Main struct storing different aspects of the replica and it's state Refer to figure 2 of the raft paper

func InitializeNode

func InitializeNode(n_replica int32, rid int, keyvalue_addr string) *RaftNode

Initialize the RaftNode (and NodeMetadata) objects. Also restores persisted raft state, if any.

func Setup_raft_node

func Setup_raft_node(ctx context.Context, id int, n_replicas int, testing bool) *RaftNode

This function initializes the node and imports the persistent state information to the node.

func (*RaftNode) AppendEntries

func (node *RaftNode) AppendEntries(ctx context.Context, in *protos.AppendEntriesMessage) (*protos.AppendEntriesResponse, error)

Implements the functionality involved when a replica recieves an AppendEntries RPC, as the Raft paper describes it.

func (*RaftNode) ApplyToStateMachine

func (node *RaftNode) ApplyToStateMachine(ctx context.Context, testing bool)

Apply committed entries to our key-value store.

func (*RaftNode) ConnectToPeerReplicas

func (node *RaftNode) ConnectToPeerReplicas(ctx context.Context, rep_addrs []string)

Attempt to connect to the gRPC servers of all other replicas, and obtain the client stubs

func (*RaftNode) Connect_raft_node

func (node *RaftNode) Connect_raft_node(ctx context.Context, id int, rep_addrs []string, testing bool)

It connects the current node to the other nodes. This mechanism includes the initiation of their various services, like the KV store server, the gRPC server and the Raft server.

The `connect_chan` channel is used to signify the end of execution of this function for synchronization and error handling.

func (*RaftNode) DeleteHandler

func (node *RaftNode) DeleteHandler(w http.ResponseWriter, r *http.Request)

Handles DELETE requests

func (*RaftNode) GetHandler

func (node *RaftNode) GetHandler(w http.ResponseWriter, r *http.Request)

Handle GET requests

func (*RaftNode) GetLock

func (node *RaftNode) GetLock(where string)

func (*RaftNode) GetRLock

func (node *RaftNode) GetRLock(where string)

func (*RaftNode) HeartBeats

func (node *RaftNode) HeartBeats(ctx context.Context)

HeartBeats is a goroutine that periodically sends heartbeats as long as the replicas thinks it's a leader

func (*RaftNode) LeaderSendAE

func (node *RaftNode) LeaderSendAE(parent_ctx context.Context, replica_id int32, upper_index int32, client_obj protos.ConsensusServiceClient, msg *protos.AppendEntriesMessage) (status bool)

To send AppendEntry to single replica, and retry if needed (called by LeaderSendAEs defined below).

func (*RaftNode) LeaderSendAEs

func (node *RaftNode) LeaderSendAEs(msg_type string, msg *protos.AppendEntriesMessage, upper_index int32, successful_write chan bool)

Called when the replica wants to send AppendEntries to all other replicas.

func (*RaftNode) ListenForShutdown

func (node *RaftNode) ListenForShutdown(master_cancel context.CancelFunc)

Listen for termination signal and call master cancel. Wait for spawned goroutines to exit.

func (*RaftNode) PersistToStorage

func (node *RaftNode) PersistToStorage()

func (*RaftNode) PostHandler

func (node *RaftNode) PostHandler(w http.ResponseWriter, r *http.Request)

Handle POST requests

func (*RaftNode) PutHandler

func (node *RaftNode) PutHandler(w http.ResponseWriter, r *http.Request)

Handle PUT requests

func (*RaftNode) ReadCommand

func (node *RaftNode) ReadCommand(key string) (string, error)

ReadCommand is called when the client sends the replica a read request. Read operations do not need to be added to the log.

func (*RaftNode) ReleaseLock

func (node *RaftNode) ReleaseLock(where string)

func (*RaftNode) ReleaseRLock

func (node *RaftNode) ReleaseRLock(where string)

func (*RaftNode) RequestVote

func (node *RaftNode) RequestVote(ctx context.Context, in *protos.RequestVoteMessage) (*protos.RequestVoteResponse, error)

Implements the functionality involved when a replica recieves a RequestVote RPC. If the received message's term is greater than the replica's current term, transition to follower (if not already a follower) and update term. If in.Term < node.currentTerm, reject vote. If the candidate's log is not atleast as up-to-date as the replica's, reject vote.

func (*RaftNode) RestoreFromStorage

func (node *RaftNode) RestoreFromStorage(storage *Storage)

Restore persisted Raft state from non volatile memory.

func (*RaftNode) RunElectionTimer

func (node *RaftNode) RunElectionTimer(parent_ctx context.Context)

RunElectionTimer runs an election and initiates transition to candidate if a heartbeat/appendentries RPC is not received within the timeout duration.

func (*RaftNode) StaleReadCheck

func (node *RaftNode) StaleReadCheck(heartbeat_success chan bool)

StaleReadCheck sends dummy heartbeats to make sure that a new leader has not been elected.

func (*RaftNode) StartElection

func (node *RaftNode) StartElection(ctx context.Context)

StartElection is called when a node transitions to a candidate

func (*RaftNode) StartGRPCServer

func (node *RaftNode) StartGRPCServer(ctx context.Context, grpc_address string, listener *net.TCPListener, testing bool)

This function starts the gRPC server for the raft node and shuts it down when context is cancelled.

func (*RaftNode) StartKVStore

func (node *RaftNode) StartKVStore(ctx context.Context, addr string, num int, testing bool)

Start the local key-value store and the HTTP server it listens for requests on.

func (*RaftNode) StartRaftServer

func (node *RaftNode) StartRaftServer(ctx context.Context, addr string, testing bool)

HTTP server to listen for client requests

func (*RaftNode) TestHandler

func (node *RaftNode) TestHandler(w http.ResponseWriter, r *http.Request)

Clients can make a request to the /test endpoint to check if the server is up.

func (*RaftNode) ToCandidate

func (node *RaftNode) ToCandidate(ctx context.Context)

ToCandidate is called when election timer runs out without heartbeat from leader

func (*RaftNode) ToFollower

func (node *RaftNode) ToFollower(ctx context.Context, term int32)

Method to transition the replica to Follower state.

func (*RaftNode) ToLeader

func (node *RaftNode) ToLeader(ctx context.Context)

ToLeader is called when the candidate gets majority votes in election

func (*RaftNode) WriteCommand

func (node *RaftNode) WriteCommand(operation []string, client string) (bool, error)

WriteCommand is called by the HTTP handler functions when the client sends the replica a write (POST/PUT/DELETE) request. It parses the command and calls LeaderSendAEs for performing the write. If the write request is successfully persisted across a majority of replicas, it informs ApplyToStateMachine to perform the write operation on the key-value store.

type RaftNodeState

type RaftNodeState int32

const (
	Follower RaftNodeState = iota
	Candidate
	Leader
	Down
)

type Storage

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

func NewStorage

func NewStorage() *Storage

Initialise Storage object

func (*Storage) Get

func (stored *Storage) Get(key string, filename string) (interface{}, bool)

Get a particular value from the persisted data.

func (*Storage) HasData

func (stored *Storage) HasData(filename string) bool

Check if file where data is persisted has data or not.

func (*Storage) ReadFile

func (stored *Storage) ReadFile(filename string)

Read the file and decode the gob.

func (*Storage) Set

func (stored *Storage) Set(key string, value interface{})

Write a particular value to be persisted.

func (*Storage) WriteFile

func (stored *Storage) WriteFile(filename string)

Encode as gob and write to file for persistence.