[Raft] Initialization and configuration of consensus

pkg/raftstore/consensus/consensus.go

@@ -6,8 +6,10 @@ import (
 	"fmt"
 	"net/http"
 	"net/url"
+	"time"
 
 	"github.com/interuss/dss/pkg/logging"
+	params "github.com/interuss/dss/pkg/raftstore/params"
 	"github.com/interuss/stacktrace"
 	"go.etcd.io/etcd/client/pkg/v3/types"
 	"go.etcd.io/etcd/server/v3/etcdserver/api/rafthttp"
@@ -17,43 +19,97 @@ import (
 	"go.uber.org/zap"
 )
 
-const (
-	defaultClusterID uint64 = 1
-)
-
 type Consensus struct {
 	logger *zap.Logger
 
-	node raft.Node
+	nodeID uint64
+	node   raft.Node
 
 	transport *rafthttp.Transport
 	server    *http.Server
 
 	storage *storage
-	errorC  chan error
+
+	confState     raftpb.ConfState
+	snapshotIndex uint64
+	appliedIndex  uint64
 }
 
-func NewConsensus(ctx context.Context, logger *zap.Logger, nodeID uint64, peers map[uint64]*url.URL, dataDir string, snapshotCatchupEntries uint64) (*Consensus, error) {
-	storage, _, err := newStorage(ctx, logger.With(zap.String("component", "storage")), dataDir, nodeID, snapshotCatchupEntries)
+func NewConsensus(ctx context.Context, logger *zap.Logger, peers map[uint64]*url.URL, connectParams params.ConnectParameters) (*Consensus, error) {
+	storage, old, err := newStorage(ctx, logger.With(zap.String("component", "storage")), connectParams.DataDir, connectParams.NodeID, connectParams.SnapshotCatchupEntries)
 	if err != nil {
 		return nil, stacktrace.Propagate(err, "failed to initialize storage")
 	}
 
+	nodeUrl, ok := peers[connectParams.NodeID]
+	if !ok {
+		return nil, stacktrace.NewError("node ID %d not found in peers map", connectParams.NodeID)
+	}
+
+	var node raft.Node
+	config := connectParams.RaftConfig(storage)
+	if old {
+		logger.Info("restarting raft node", zap.String("address", nodeUrl.String()))
+		node = raft.RestartNode(config)
+	} else {
+		logger.Info("starting new raft node", zap.String("address", nodeUrl.String()))
+		node = raft.StartNode(config, peersList(peers))
+	}
+
 	consensus := &Consensus{
 		logger: logging.WithValuesFromContext(ctx, logger),
 
+		nodeID: connectParams.NodeID,
+		node:   node,
+
 		storage: storage,
-		errorC:  make(chan error, 1),
 	}
 
-	err = consensus.initTransport(ctx, nodeID, defaultClusterID, peers)
+	err = consensus.initTransport(ctx, connectParams.NodeID, connectParams.ClusterID, peers)
 	if err != nil {
 		return nil, stacktrace.Propagate(err, "failed to initialize transport")
 	}
 
+	snap, err := consensus.storage.Snapshot()
+	if err != nil {
+		return nil, stacktrace.Propagate(err, "failed to get snapshot from storage")
+	}
+
+	consensus.confState = snap.Metadata.ConfState
+	consensus.snapshotIndex = snap.Metadata.Index
+	consensus.appliedIndex = snap.Metadata.Index
+
+	go func() {
+		err := consensus.handleReady(connectParams.TickInterval)
+		if err != nil {
+			consensus.logger.Error("handleReady exited with error, shutting down consensus", zap.Error(err))
+		}
+
+		shutdownCtx, cancel := context.WithTimeout(context.Background(), 2*connectParams.ElectionInterval())
+		defer cancel()
+		if shutdownErr := consensus.server.Shutdown(shutdownCtx); shutdownErr != nil {
+			consensus.logger.Error("failed to shutdown http server", zap.Error(shutdownErr))
+		} else {
+			consensus.logger.Info("http server shutdown complete")
+		}
+
+		consensus.transport.Stop()
+		consensus.logger.Info("transport stopped")
+		consensus.node.Stop()
+		consensus.logger.Info("raft node stopped")
+	}()
+
 	return consensus, nil
 }
 
+func peersList(peers map[uint64]*url.URL) []raft.Peer {
+	result := make([]raft.Peer, 0, len(peers))
+	for id := range peers {
+		result = append(result, raft.Peer{ID: id})
+	}
+	return result
+}
+
 func (c *Consensus) initTransport(ctx context.Context, nodeID uint64, clusterID uint64, peers map[uint64]*url.URL) error {
 	nodeIDStr := fmt.Sprintf("%d", nodeID)
 
@@ -64,7 +120,7 @@ func (c *Consensus) initTransport(ctx context.Context, nodeID uint64, clusterID
 		Raft:        c,
 		ServerStats: v2stats.NewServerStats(nodeIDStr, nodeIDStr),
 		LeaderStats: v2stats.NewLeaderStats(c.logger, nodeIDStr),
-		ErrorC:      c.errorC,
+		ErrorC:      make(chan error),
 	}
 
 	err := transport.Start()
@@ -95,14 +151,108 @@ func (c *Consensus) initTransport(ctx context.Context, nodeID uint64, clusterID
 		err := c.server.ListenAndServe()
 		if err != nil && !errors.Is(err, http.ErrServerClosed) {
 			c.logger.Error("http server error", zap.Error(err))
-			c.errorC <- err
+			c.transport.ErrorC <- err
 		}
 	}()
 
 	c.transport = transport
 	return nil
 }
 
+// handleReady processes the Ready channel of the Raft node and applies committed entries to the state machine
+func (c *Consensus) handleReady(tickInterval time.Duration) error {
+	ticker := time.NewTicker(tickInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			c.node.Tick()
+		case err := <-c.transport.ErrorC:
+			return stacktrace.Propagate(err, "transport error")
+		case rd, ok := <-c.node.Ready():
+			if !ok {
+				return stacktrace.NewError("could not read from Ready(), shutting down handler")
+			}
+
+			err := c.storage.handleReceivedState(rd.Snapshot, rd.HardState, rd.Entries)
+			if err != nil {
+				return stacktrace.Propagate(err, "failed to handle received snapshot")
+			}
+
+			if !raft.IsEmptySnap(rd.Snapshot) {
+				if rd.Snapshot.Metadata.Index <= c.appliedIndex {
+					return stacktrace.NewError("snapshot index %d shall be greater than current applied index %d", rd.Snapshot.Metadata.Index, c.appliedIndex)
+				}
+
+				err = c.dispatchSnapshot(rd.Snapshot.Data)
+				if err != nil {
+					return stacktrace.Propagate(err, "failed to dispatch snapshot")
+				}
+
+				c.confState = rd.Snapshot.Metadata.ConfState
+				c.snapshotIndex = rd.Snapshot.Metadata.Index
+				c.appliedIndex = rd.Snapshot.Metadata.Index
+			}
+
+			c.updateSnapshotConfState(rd.Messages)
+			c.transport.Send(rd.Messages)
+
+			entries, err := c.entriesToApply(rd.CommittedEntries)
+			if err != nil {
+				return stacktrace.Propagate(err, "failed to get entries to apply")
+			}
+
+			err = c.publishEntries(entries)
+			if err != nil {
+				return stacktrace.Propagate(err, "failed to publish entries")
+			}
+
+			c.node.Advance()
+		}
+	}
+}
+
+// TODO implement
+func (c *Consensus) publishEntries(_ []raftpb.Entry) error {
+	return nil
+}
+
+// TODO implement
+func (c *Consensus) dispatchSnapshot(_ []byte) error {
+	return nil
+}
+
+func (c *Consensus) entriesToApply(entries []raftpb.Entry) ([]raftpb.Entry, error) {
+	if len(entries) == 0 {
+		return entries, nil
+	}
+
+	result := make([]raftpb.Entry, 0)
+
+	firstIdx := entries[0].Index
+	if firstIdx > c.appliedIndex+1 {
+		return nil, stacktrace.NewError("unexpected gap: first committed entry index %d > applied index %d + 1", firstIdx, c.appliedIndex)
+	}
+
+	// Skip entries that have already been applied.
+	if skip := c.appliedIndex + 1 - firstIdx; skip < uint64(len(entries)) {
+		result = entries[skip:]
+	}
+
+	return result, nil
+}
+
+// updateSnapshotConfState updates the ConfState in the snapshot
+// of messages that contain one as it could be outdated.
+func (c *Consensus) updateSnapshotConfState(msgs []raftpb.Message) {
+	for i := range msgs {
+		if msgs[i].Type == raftpb.MsgSnap {
+			msgs[i].Snapshot.Metadata.ConfState = c.confState
+		}
+	}
+}
+
 // Process implements the rafthttp.Raft interface.
 func (c *Consensus) Process(ctx context.Context, m raftpb.Message) error {
 	return c.node.Step(ctx, m)

pkg/raftstore/params/params.go

@@ -5,22 +5,37 @@ import (
 	"net/url"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/interuss/stacktrace"
+	"go.etcd.io/raft/v3"
 )
 
 const (
 	defaultDataDir = "raft_data"
 
 	// the default Raft related parameters are the same as the default values used by etcd for the moment.
 	// TODO - review and adjust these parameters as needed based on testing and performance tuning.
-	defaultSnapshotCatchupEntries = 10000
+
+	defaultSnapshotCatchupEntries  = 5000
+	defaultSnapshotIntervalEntries = 10000
+	defaultTickInterval            = 100 * time.Millisecond
+
+	// follower waits 10 x defaultTickInterval without a heartbeat before starting an election
+	defaultElectionTick = 10
+	// leader sends a heartbeat every tick, must be < defaultElectionTick
+	defaultHeartbeatTick = 1
+
+	defaultMaxSizePerMsg   = 1024 * 1024
+	defaultMaxInflightMsgs = 4096 / 8
 )
 
 type (
 	// ConnectParameters bundles up parameters used for connecting nodes in a raftstore cluster.
 	ConnectParameters struct {
-		ID    uint64
+		// unique node identifier within the cluster, 0 is invalid
+		NodeID uint64
+		// comma-separated "nodeID=peerURL" pairs defining all cluster members including this node
 		Peers string
 
 		// DataDir is the directory where the node persists its Raft state (WAL segments and snapshots).
@@ -29,10 +44,26 @@ type (
 		// across restarts unless the node is being permanently shut down.
 		// If the directory is lost, the node will recover by receiving a snapshot from the leader.
 		DataDir string
+		// discriminates this cluster from others sharing the same network, must be identical on all nodes
+		ClusterID uint64
 
-		// SnapshotCatchupEntries is the number of entries for a slow follower to catch-up after compacting.
+		// number of entries for a slow follower to catch-up after compacting.
 		// This gives the follower a buffer of entries while avoiding the need to send a full snapshot.
 		SnapshotCatchupEntries uint64
+		// number of entries applied before triggering a snapshot
+		SnapshotIntervalEntries uint64
+		// base time unit for Raft's logical clock, scales both election and heartbeat timers
+		TickInterval time.Duration
+
+		// ticks without a heartbeat before a follower promotes to candidate, effective timeout = ElectionTick × TickInterval
+		ElectionTick int
+		// ticks between leader heartbeats, must be < ElectionTick
+		HeartbeatTick int
+
+		// max byte size of a message sent to a peer
+		MaxSizePerMsg uint64
+		// max number of in-flight messages during optimistic replication phase
+		MaxInflightMsgs int
 	}
 )
 
@@ -70,16 +101,44 @@ func (c ConnectParameters) PeerMap() (map[uint64]*url.URL, error) {
 	return peers, nil
 }
 
+func (c ConnectParameters) ElectionInterval() time.Duration {
+	return time.Duration(c.ElectionTick) * c.TickInterval
+}
+
+func (c ConnectParameters) RaftConfig(storage raft.Storage) *raft.Config {
+	return &raft.Config{
+		ID:              c.NodeID,
+		ElectionTick:    c.ElectionTick,
+		HeartbeatTick:   c.HeartbeatTick,
+		MaxSizePerMsg:   c.MaxSizePerMsg,
+		MaxInflightMsgs: c.MaxInflightMsgs,
+		Storage:         storage,
+	}
+}
+
 var (
 	connectParameters ConnectParameters
 )
 
 func init() {
-	flag.Uint64Var(&connectParameters.ID, "raft_node_id", 0, "raft node ID for this instance (must be non-zero and unique within the cluster)")
-	flag.StringVar(&connectParameters.Peers, "raft_peers", "", `comma-separated "nodeID=peerURL" pairs for all cluster members, including the current node, e.g. "1=http://node1:9021,2=http://node2:9021,3=http://node3:9021"`)
-	flag.StringVar(&connectParameters.DataDir, "raft_datadir", defaultDataDir, "directory for raft data (WAL segments and snapshots), required for restarts. These should not be deleted while the node is running or across restarts unless the node is being permanently shut down.")
-
-	flag.Uint64Var(&connectParameters.SnapshotCatchupEntries, "raft_snapshot_catchup_entries", defaultSnapshotCatchupEntries, "number of entries for a slow follower to catch-up after compacting")
+	flag.Uint64Var(&connectParameters.NodeID, "raft_node_id", 0, "Raft node ID for this instance (must be non-zero and unique within the cluster).")
+	flag.Uint64Var(&connectParameters.ClusterID, "raft_cluster_id", 1, "ID of the cluster, used to isolate different Raft clusters running in the same network (must be the same for all nodes in the cluster).")
+	flag.StringVar(&connectParameters.Peers, "raft_peers", "", `Comma-separated "nodeID=peerURL" pairs for all cluster members, including the current node, e.g. "1=http://node1:9021,2=http://node2:9021,3=http://node3:9021"`)
+	flag.StringVar(&connectParameters.DataDir, "raft_datadir", defaultDataDir, "Directory for raft data (WAL segments and snapshots), required for restarts. These should not be deleted while the node is running or across restarts unless the node is being permanently shut down.")
+
+	flag.Uint64Var(&connectParameters.SnapshotCatchupEntries, "raft_snapshot_catchup_entries", defaultSnapshotCatchupEntries,
+		"Log entries retained after compaction so a slow follower can catch up via replication rather than a full snapshot. Higher values tolerate slower followers but increase disk usage.")
+	flag.Uint64Var(&connectParameters.SnapshotIntervalEntries, "raft_snapshot_interval_entries", defaultSnapshotIntervalEntries,
+		"Applied Raft log entries to accumulate before triggering a snapshot. Lower values reduce recovery time but increase I/O frequency.")
+	flag.DurationVar(&connectParameters.TickInterval, "raft_tick_interval", defaultTickInterval,
+		"Base time unit for Raft's logical clock. Election timeout = raft_election_tick x this value; heartbeat interval = raft_heartbeat_tick x this value. Smaller values improve responsiveness but increase network traffic.")
+
+	flag.IntVar(&connectParameters.ElectionTick, "raft_election_tick", defaultElectionTick,
+		"Ticks a follower waits without a leader heartbeat before starting an election. Effective timeout = raft_election_tick x raft_tick_interval. Must be greater than raft_heartbeat_tick. Higher values tolerate slow leaders but delay failover.")
+	flag.IntVar(&connectParameters.HeartbeatTick, "raft_heartbeat_tick", defaultHeartbeatTick,
+		"Ticks between leader heartbeats. Effective interval = raft_heartbeat_tick x raft_tick_interval. Must be less than raft_election_tick. Lower values detect follower loss faster but increase network traffic.")
+	flag.Uint64Var(&connectParameters.MaxSizePerMsg, "raft_max_size_per_msg", defaultMaxSizePerMsg, "Maximum bytes in a single Raft message sent to a peer. Smaller values lower the recovery cost but increase the number of messages sent, affecting throughput during replication.")
+	flag.IntVar(&connectParameters.MaxInflightMsgs, "raft_max_inflight_msgs", defaultMaxInflightMsgs, "Maximum number of in-flight Raft messages during optimistic replication phase. This should be set to avoid overflowing the transport layer sending buffer.")
 }
 
 // GetConnectParameters returns a ConnectParameters instance that gets populated from well-known CLI flags.

pkg/raftstore/store.go

@@ -31,7 +31,7 @@ func Init[R any](ctx context.Context, logger *zap.Logger, newRepo func() R) (*St
 			return
 		}
 
-		sharedConsensus, sharedConsensusErr = consensus.NewConsensus(ctx, logger, params.ID, peers, params.DataDir, params.SnapshotCatchupEntries)
+		sharedConsensus, sharedConsensusErr = consensus.NewConsensus(ctx, logger, peers, params)
 		if sharedConsensusErr != nil {
 			sharedConsensusErr = stacktrace.Propagate(sharedConsensusErr, "failed to initialize consensus")
 		}