raftgrpc

README

Raft gRPC Transport

A gRPC-based Transport implementation for HashiCorp's Raft consensus library.

This library allows you to use gRPC for communication between Raft nodes, leveraging HTTP/2 performance, streaming capabilities, and the rich ecosystem of gRPC middleware (auth, logging, tracing, etc.).

Features

• Standard gRPC: Uses Protocol Buffers for efficient serialization (via buf).
• Streamed Snapshots: Supports streaming large snapshots over gRPC.
• Connection Management: Lazily establishes and maintains persistent connections to peers.
• Drop-in Replacement: Implements the standard raft.Transport interface.

Installation

go get github.com/dhiaayachi/raft-grpc-transport

Usage

Here is a basic example of how to configure and use the gRPC transport with Raft.

package main

import (
	"fmt"
	"io"
	"net"
	"os"
	"time"

	raftgrpc "github.com/dhiaayachi/raft-grpc-transport"
	"github.com/hashicorp/raft"
	"google.golang.org/grpc"
)

func main() {
	// 1. Setup network listener
	addr := "127.0.0.1:50051"
	lis, err := net.Listen("tcp", addr)
	if err != nil {
		panic(fmt.Sprintf("failed to listen: %v", err))
	}

	// 2. Create gRPC server
	s := grpc.NewServer()

	// 3. Create the Raft transport and register it with the server
	transport, err := raftgrpc.NewGrpcTransport(lis.Addr().String(), s)
	if err != nil {
		panic(fmt.Sprintf("failed to create transport: %v", err))
	}
	defer transport.Close()

	// 4. Start the server (in a goroutine so it doesn't block)
	go func() {
		if err := s.Serve(lis); err != nil {
			panic(fmt.Sprintf("server error: %v", err))
		}
	}()

	// 5. Setup Raft configuration
	config := raft.DefaultConfig()
	config.LocalID = raft.ServerID("node-1")

	// 6. Setup other Raft dependencies (stores, snapshots, fsm)
	// For example purposes, using in-memory stores
	logStore := raft.NewInmemStore()
	stableStore := raft.NewInmemStore()
	snapStore := raft.NewInmemSnapshotStore()
	fsm := &MyFSM{} // Your state machine implementation

	// 7. Create the Raft node
	r, err := raft.NewRaft(config, fsm, logStore, stableStore, snapStore, transport)
	if err != nil {
		panic(fmt.Sprintf("failed to create raft node: %v", err))
	}

	// 8. Bootstrap the cluster (if this is the first node)
	// ONLY do this for the very first node in a new cluster
	configuration := raft.Configuration{
		Servers: []raft.Server{
			{
				ID:      config.LocalID,
				Address: transport.LocalAddr(),
			},
		},
	}
	r.BootstrapCluster(configuration)

	fmt.Println("Raft node started successfully!")
	
	// Keep the main goroutine running
	select {}
}

// Minimal FSM implementation
type MyFSM struct{}
func (f *MyFSM) Apply(l *raft.Log) interface{} { return nil }
func (f *MyFSM) Snapshot() (raft.FSMSnapshot, error) { return &MySnapshot{}, nil }
func (f *MyFSM) Restore(rc io.ReadCloser) error { return nil }

type MySnapshot struct{}
func (s *MySnapshot) Persist(sink raft.SnapshotSink) error { return sink.Close() }
func (s *MySnapshot) Release() {}

Development

Prerequisites

• Go 1.22+
• buf (for protobuf generation)

Generating Code

If you make changes to the .proto files, regenerate the Go code:

buf generate

Running Tests

Migration Transport

The MigrationTransport allows you to migrate from one transport (e.g., TCP) to another (e.g., gRPC) with zero downtime. It wraps two transports (Primary and Secondary) and attempts to use the Primary first. If the Primary fails with a specific error (controlled by a predicate function), it falls back to the Secondary.

Incoming RPCs from both transports are handled indiscriminately.

Usage

// 1. Create your two transports
oldTransport := ... // e.g., raft.NewTCPTransport(...)
newTransport, _ := raftgrpc.NewGrpcTransport(...)

// 2. Define a fallback predicate
// e.g. Fallback if the connection is refused or unavailable
shouldFallback := func(err error) bool {
    // Check error type/content
    return true
}

// 3. Create the migration transport
transport := raftgrpc.NewMigrationTransport(newTransport, oldTransport, shouldFallback)

// 4. Use 'transport' when creating your Raft node

Benchmarks

This library includes benchmarks comparing raft-grpc-transport against the default raft.NetworkTransport (TCP + MsgPack).

Results Summary

Running Benchmarks

You can run the benchmarks and generate a comparison graph locally:

go test -benchtime=10000x -count=10 -tags benchmark -v -bench=. -benchmem -timeout 20m -args -plot

This will generate a benchmark_comparison.svg file with visual results.

Documentation

Index

• type GrpcTransport
  • func NewGrpcTransport(localAddr string, server *grpc.Server, opts ...TransportOption) (*GrpcTransport, error)
  • func (t *GrpcTransport) AppendEntries(id raft.ServerID, target raft.ServerAddress, args *raft.AppendEntriesRequest, ...) error
  • func (t *GrpcTransport) AppendEntriesPipeline(id raft.ServerID, target raft.ServerAddress) (raft.AppendPipeline, error)
  • func (t *GrpcTransport) Close() error
  • func (t *GrpcTransport) Consumer() <-chan raft.RPC
  • func (t *GrpcTransport) DecodePeer(buf []byte) raft.ServerAddress
  • func (t *GrpcTransport) EncodePeer(id raft.ServerID, addr raft.ServerAddress) []byte
  • func (t *GrpcTransport) InstallSnapshot(id raft.ServerID, target raft.ServerAddress, args *raft.InstallSnapshotRequest, ...) error
  • func (t *GrpcTransport) LocalAddr() raft.ServerAddress
  • func (t *GrpcTransport) RequestPreVote(id raft.ServerID, target raft.ServerAddress, args *raft.RequestPreVoteRequest, ...) error
  • func (t *GrpcTransport) RequestVote(id raft.ServerID, target raft.ServerAddress, args *raft.RequestVoteRequest, ...) error
  • func (t *GrpcTransport) SetHeartbeatHandler(cb func(rpc raft.RPC))
  • func (t *GrpcTransport) TimeoutNow(id raft.ServerID, target raft.ServerAddress, args *raft.TimeoutNowRequest, ...) error
• type MigrationTransport
  • func NewMigrationTransport(primary, secondary raft.Transport, shouldFallback func(error) bool) *MigrationTransport
  • func (t *MigrationTransport) AppendEntries(id raft.ServerID, target raft.ServerAddress, args *raft.AppendEntriesRequest, ...) error
  • func (t *MigrationTransport) AppendEntriesPipeline(id raft.ServerID, target raft.ServerAddress) (raft.AppendPipeline, error)
  • func (t *MigrationTransport) Close() error
  • func (t *MigrationTransport) Consumer() <-chan raft.RPC
  • func (t *MigrationTransport) DecodePeer(buf []byte) raft.ServerAddress
  • func (t *MigrationTransport) EncodePeer(id raft.ServerID, addr raft.ServerAddress) []byte
  • func (t *MigrationTransport) InstallSnapshot(id raft.ServerID, target raft.ServerAddress, args *raft.InstallSnapshotRequest, ...) error
  • func (t *MigrationTransport) LocalAddr() raft.ServerAddress
  • func (t *MigrationTransport) RequestVote(id raft.ServerID, target raft.ServerAddress, args *raft.RequestVoteRequest, ...) error
  • func (t *MigrationTransport) SetHeartbeatHandler(cb func(rpc raft.RPC))
  • func (t *MigrationTransport) TimeoutNow(id raft.ServerID, target raft.ServerAddress, args *raft.TimeoutNowRequest, ...) error
• type TransportOption
  • func WithTimeout(d time.Duration) TransportOption

Types

type GrpcTransport

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

GrpcTransport implements raft.Transport using gRPC.

func NewGrpcTransport

func NewGrpcTransport(localAddr string, server *grpc.Server, opts ...TransportOption) (*GrpcTransport, error)

NewGrpcTransport creates a new GrpcTransport and registers it with the given gRPC server. The user is responsible for starting and serving the gRPC server.

func (*GrpcTransport) AppendEntries

func (t *GrpcTransport) AppendEntries(id raft.ServerID, target raft.ServerAddress, args *raft.AppendEntriesRequest, resp *raft.AppendEntriesResponse) error

AppendEntries sends an AppendEntries RPC to the given target.

func (*GrpcTransport) AppendEntriesPipeline

func (t *GrpcTransport) AppendEntriesPipeline(id raft.ServerID, target raft.ServerAddress) (raft.AppendPipeline, error)

AppendEntriesPipeline returns an interface that can be used to pipeline AppendEntries requests.

func (*GrpcTransport) Close

func (t *GrpcTransport) Close() error

Close closes the transport connections to peers. The gRPC server and listener are NOT closed.

func (*GrpcTransport) Consumer

func (t *GrpcTransport) Consumer() <-chan raft.RPC

Consumer returns a channel that can be used to consume upcoming RPCs.

func (*GrpcTransport) DecodePeer

func (t *GrpcTransport) DecodePeer(buf []byte) raft.ServerAddress

DecodePeer is used to deserialize a peer's address.

func (*GrpcTransport) EncodePeer

func (t *GrpcTransport) EncodePeer(id raft.ServerID, addr raft.ServerAddress) []byte

EncodePeer is used to serialize a peer's address.

func (*GrpcTransport) InstallSnapshot

func (t *GrpcTransport) InstallSnapshot(id raft.ServerID, target raft.ServerAddress, args *raft.InstallSnapshotRequest, resp *raft.InstallSnapshotResponse, data io.Reader) error

InstallSnapshot sends an InstallSnapshot RPC to the given target.

func (*GrpcTransport) LocalAddr

func (t *GrpcTransport) LocalAddr() raft.ServerAddress

LocalAddr returns the local address of the transport.

func (*GrpcTransport) RequestPreVote

func (t *GrpcTransport) RequestPreVote(id raft.ServerID, target raft.ServerAddress, args *raft.RequestPreVoteRequest, resp *raft.RequestPreVoteResponse) error

RequestPreVote sends a RequestPreVote RPC to the given target.

func (*GrpcTransport) RequestVote

func (t *GrpcTransport) RequestVote(id raft.ServerID, target raft.ServerAddress, args *raft.RequestVoteRequest, resp *raft.RequestVoteResponse) error

RequestVote sends a RequestVote RPC to the given target.

func (*GrpcTransport) SetHeartbeatHandler

func (t *GrpcTransport) SetHeartbeatHandler(cb func(rpc raft.RPC))

SetHeartbeatHandler is used to setup a heartbeat handler as a fast-pass. This is optional.

func (*GrpcTransport) TimeoutNow

func (t *GrpcTransport) TimeoutNow(id raft.ServerID, target raft.ServerAddress, args *raft.TimeoutNowRequest, resp *raft.TimeoutNowResponse) error

TimeoutNow is used to start a leadership transfer to the target node.

type MigrationTransport

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

MigrationTransport wraps two transports (Primary and Secondary) and allows failing over to the Secondary transport if the Primary fails with a specific error. It also merges incoming RPCs from both transports.

func NewMigrationTransport

func NewMigrationTransport(primary, secondary raft.Transport, shouldFallback func(error) bool) *MigrationTransport

NewMigrationTransport creates a new MigrationTransport.

func (*MigrationTransport) AppendEntries

func (t *MigrationTransport) AppendEntries(id raft.ServerID, target raft.ServerAddress, args *raft.AppendEntriesRequest, resp *raft.AppendEntriesResponse) error

func (*MigrationTransport) AppendEntriesPipeline

func (t *MigrationTransport) AppendEntriesPipeline(id raft.ServerID, target raft.ServerAddress) (raft.AppendPipeline, error)

func (*MigrationTransport) Close

func (t *MigrationTransport) Close() error

func (*MigrationTransport) Consumer

func (t *MigrationTransport) Consumer() <-chan raft.RPC

Consumer returns a channel that can be used to consume upcoming RPCs.

func (*MigrationTransport) DecodePeer

func (t *MigrationTransport) DecodePeer(buf []byte) raft.ServerAddress

func (*MigrationTransport) EncodePeer

func (t *MigrationTransport) EncodePeer(id raft.ServerID, addr raft.ServerAddress) []byte

func (*MigrationTransport) InstallSnapshot

func (t *MigrationTransport) InstallSnapshot(id raft.ServerID, target raft.ServerAddress, args *raft.InstallSnapshotRequest, resp *raft.InstallSnapshotResponse, data io.Reader) error

func (*MigrationTransport) LocalAddr

func (t *MigrationTransport) LocalAddr() raft.ServerAddress

LocalAddr returns the local address of the transport. It defaults to the Primary's address.

func (*MigrationTransport) RequestVote

func (t *MigrationTransport) RequestVote(id raft.ServerID, target raft.ServerAddress, args *raft.RequestVoteRequest, resp *raft.RequestVoteResponse) error

func (*MigrationTransport) SetHeartbeatHandler

func (t *MigrationTransport) SetHeartbeatHandler(cb func(rpc raft.RPC))

func (*MigrationTransport) TimeoutNow

func (t *MigrationTransport) TimeoutNow(id raft.ServerID, target raft.ServerAddress, args *raft.TimeoutNowRequest, resp *raft.TimeoutNowResponse) error

type TransportOption

type TransportOption func(*GrpcTransport)

TransportOption is a functional option for GrpcTransport.

func WithTimeout

func WithTimeout(d time.Duration) TransportOption

WithTimeout sets the timeout for RPCs.