diff --git a/.gitignore b/.gitignore index b20797585..6e8f25a97 100644 --- a/.gitignore +++ b/.gitignore @@ -137,4 +137,7 @@ go .terraform* # raft -raft_data/ \ No newline at end of file +raft_data/ +# local certificates +build/test-certs/raft-certs/*.crt +build/test-certs/raft-certs/*.key diff --git a/build/test-certs/raft-certs/gen-cert.sh b/build/test-certs/raft-certs/gen-cert.sh new file mode 100755 index 000000000..cca5fba49 --- /dev/null +++ b/build/test-certs/raft-certs/gen-cert.sh @@ -0,0 +1,60 @@ +#!/bin/bash + +# Default values +NODE_COUNT=3 + +# Help function +show_help() { + echo "Usage: ./gen-cert.sh [options]" + echo "" + echo "Options:" + echo " -n, --nodes Number of nodes to generate (default: 3)" + echo " -c, --clean Delete all .key, .crt, .csr, and .srl files" + echo " -h, --help Show this help message" + exit 0 +} + +# Cleanup function +clean_certs() { + echo "Cleaning up all certificates and keys..." + rm -f ./*.key ./*.crt ./*.csr ./*.srl + echo "Done." + exit 0 +} + +# Parse arguments +while [[ "$#" -gt 0 ]]; do + case $1 in + -n|--nodes) NODE_COUNT="$2"; shift ;; + -c|--clean) clean_certs ;; + -h|--help) show_help ;; + *) echo "Unknown parameter: $1"; show_help ;; + esac + shift +done + +echo "Creating CA and certificates for $NODE_COUNT nodes..." + +# 1. Generate the Root CA +openssl req -x509 -newkey rsa:4096 -keyout ca.key -out ca.crt -days 3650 -sha256 -nodes -subj "/CN=root" + +# 2. Loop to create node certificates +for ((i=1; i<=NODE_COUNT; i++)) +do + echo "Generating node$i..." + + # Create Request + openssl req -out "node$i.csr" -newkey rsa:2048 -keyout "node$i.key" -nodes \ + -subj "/CN=node$i" \ + -addext "subjectAltName=DNS:node$i,DNS:localhost,IP:127.0.0.1" + + # Sign Certificate + openssl x509 -req -in "node$i.csr" -CA ca.crt -CAkey ca.key \ + -out "node$i.crt" -days 365 -sha256 \ + -set_serial "$i" \ + -extfile <(echo "subjectAltName=DNS:node$i,DNS:localhost,IP:127.0.0.1") +done + +# 3. Final Cleanup of temporary request files +rm -f ./*.csr +echo "Success! Generated $NODE_COUNT node certificates." \ No newline at end of file diff --git a/build/test-certs/raft-certs/start_cluster.sh b/build/test-certs/raft-certs/start_cluster.sh new file mode 100755 index 000000000..10ffd813b --- /dev/null +++ b/build/test-certs/raft-certs/start_cluster.sh @@ -0,0 +1,27 @@ +#!/bin/bash + +ROOT_DIR="$(git rev-parse --show-toplevel)" + +# Ensure the logs directory exists +mkdir -p logs + +PEERS="1=https://127.0.0.1:9021,2=https://127.0.0.1:9022,3=https://127.0.0.1:9023" + +echo "=== Starting 3-node raft cluster ===" + +for NODE_ID in 1 2 3; do + ADDR=":808$NODE_ID" + OUT="logs/node$NODE_ID.log" + + echo "Starting node $NODE_ID on port $ADDR..." + go run "$ROOT_DIR"/cmds/core-service \ + --store_type=raft \ + --raft_node_id=$NODE_ID \ + --addr=$ADDR \ + --raft_peers=$PEERS \ + --raft_tls_ca=$ROOT_DIR/build/test-certs/raft-certs/ca.crt \ + --raft_tls_crt=$ROOT_DIR/build/test-certs/raft-certs/node${NODE_ID}.crt \ + --raft_tls_key=$ROOT_DIR/build/test-certs/raft-certs/node${NODE_ID}.key \ + --accepted_jwt_audiences=dss \ + --public_key_files="$ROOT_DIR"/build/test-certs/auth2.pem > "$OUT" 2>&1 & +done \ No newline at end of file diff --git a/pkg/raftstore/consensus/consensus.go b/pkg/raftstore/consensus/consensus.go index 2541b9141..b1f112b75 100644 --- a/pkg/raftstore/consensus/consensus.go +++ b/pkg/raftstore/consensus/consensus.go @@ -2,59 +2,166 @@ package consensus import ( "context" + "crypto/tls" + "encoding/json" "errors" "fmt" "net/http" "net/url" + "sync" + "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/transport" "go.etcd.io/etcd/client/pkg/v3/types" "go.etcd.io/etcd/server/v3/etcdserver/api/rafthttp" v2stats "go.etcd.io/etcd/server/v3/etcdserver/api/v2stats" + "go.etcd.io/raft/v3" "go.etcd.io/raft/v3/raftpb" "go.uber.org/zap" + "golang.org/x/sync/errgroup" ) -const ( - defaultClusterID uint64 = 1 -) +const serverShutdownTimeout = 5 * time.Second 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 + storage *storage + commitChs map[string]chan EntryCommit + + tracker *proposalsTracker + + 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), - storage: storage, - errorC: make(chan error, 1), + nodeID: connectParams.NodeID, + node: node, + + storage: storage, + commitChs: make(map[string]chan EntryCommit), + tracker: newProposalsTracker(), } - 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, connectParams.SnapshotIntervalEntries) + if err != nil { + consensus.logger.Error("handleReady exited with error, shutting down consensus", zap.Error(err)) + } + + shutdownCtx, cancel := context.WithTimeout(context.Background(), serverShutdownTimeout) + 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 } +// ProposeValue blocks until the proposal is committed and applied / dropped or until ctx is cancelled. +func (c *Consensus) ProposeValue(ctx context.Context, proposal Proposal) (any, error) { + buf, err := json.Marshal(proposal) + if err != nil { + return nil, stacktrace.Propagate(err, "failed to marshal proposal") + } + + applied := c.tracker.track(proposal.ID) + + err = c.node.Propose(ctx, buf) + if err != nil { + c.tracker.untrack(proposal.ID, ProposalResult{Error: err}) + return nil, stacktrace.Propagate(err, "failed to propose value to Raft") + } + + select { + case res := <-applied: + return res.Result, res.Error + + case <-ctx.Done(): + c.tracker.untrack(proposal.ID, ProposalResult{Error: ctx.Err()}) + return nil, ctx.Err() + } +} + +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 { + connectParams := params.GetConnectParameters() + + if connectParams.Insecure { + c.logger.Warn("Running in insecure mode: HTTPS is disabled or certificates are not verified. Use only for testing or trusted environments.") + } + + tlsInfo := transport.TLSInfo{ + TrustedCAFile: connectParams.CAFile, + CertFile: connectParams.CertFile, + KeyFile: connectParams.KeyFile, + InsecureSkipVerify: connectParams.Insecure, + } + nodeIDStr := fmt.Sprintf("%d", nodeID) transport := &rafthttp.Transport{ @@ -64,11 +171,14 @@ 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() - if err != nil { + if !(connectParams.Insecure && connectParams.CAFile == "" && connectParams.CertFile == "" && connectParams.KeyFile == "") { + transport.TLSInfo = tlsInfo + } + + if err := transport.Start(); err != nil { return stacktrace.Propagate(err, "failed to start transport") } @@ -78,7 +188,6 @@ func (c *Consensus) initTransport(ctx context.Context, nodeID uint64, clusterID listeningAddr = ":" + peerURL.Port() continue } - transport.AddPeer(types.ID(peerID), []string{peerURL.String()}) } @@ -86,23 +195,276 @@ func (c *Consensus) initTransport(ctx context.Context, nodeID uint64, clusterID return stacktrace.NewError("node ID %d not found in peers map", nodeID) } - c.server = &http.Server{ - Addr: listeningAddr, - Handler: transport.Handler(), + var err error + c.server, err = c.createServer(listeningAddr, transport.Handler(), tlsInfo, connectParams.Insecure) + if err != nil { + return err } - go func() { - 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 = transport + return nil +} + +func (c *Consensus) createServer(addr string, handler http.Handler, tlsInfo transport.TLSInfo, insecure bool) (*http.Server, error) { + server := &http.Server{ + Addr: addr, + Handler: handler, + } + + if insecure { + if tlsInfo.CertFile == "" || tlsInfo.KeyFile == "" { + // Certificates provided: no HTTPS + go c.startServer(server, false, "", "") + } else { + // Certificates provided: HTTPS with no verification + cfg, err := tlsInfo.ServerConfig() + if err != nil { + return nil, stacktrace.NewError("failed to create TLS config") + } + cfg.ClientAuth = tls.NoClientCert + server.TLSConfig = cfg + go c.startServer(server, true, tlsInfo.CertFile, tlsInfo.KeyFile) } - }() + } else { + // Secure mode: HTTPS with full verification + cfg, err := tlsInfo.ServerConfig() + if err != nil { + return nil, stacktrace.NewError("failed to create TLS config") + } + cfg.ClientAuth = tls.RequireAndVerifyClientCert + server.TLSConfig = cfg + go c.startServer(server, true, tlsInfo.CertFile, tlsInfo.KeyFile) + } - c.transport = transport + return server, nil +} + +func (c *Consensus) startServer(server *http.Server, useTLS bool, certFile, keyFile string) { + var err error + if useTLS { + err = server.ListenAndServeTLS(certFile, keyFile) + } else { + err = server.ListenAndServe() + } + if err != nil && !errors.Is(err, http.ErrServerClosed) { + c.logger.Error("http server error", zap.Error(err)) + c.transport.ErrorC <- err + } +} + +// handleReady processes the Ready channel of the Raft node and applies committed entries to the state machine +func (c *Consensus) handleReady(tickInterval time.Duration, snapshotInterval uint64) 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, snapshotInterval) + if err != nil { + return stacktrace.Propagate(err, "failed to publish entries") + } + + c.node.Advance() + } + } +} + +func (c *Consensus) publishEntries(entries []raftpb.Entry, snapshotInterval uint64) error { + if len(entries) == 0 { + return nil + } + + c.logger.Info("publishing entries", zap.Int("numEntries", len(entries)), zap.Uint64("firstIndex", entries[0].Index), zap.Uint64("lastIndex", entries[len(entries)-1].Index)) + + var triggerSnapshot bool + var err error + var wg sync.WaitGroup + for _, entry := range entries { + switch entry.Type { + case raftpb.EntryNormal: + err := c.processNormalEntry(entry.Data, &wg) + if err != nil { + return stacktrace.Propagate(err, "failed to process normal entry") + } + case raftpb.EntryConfChange: + err := c.processConfigChangeEntry(entry.Data) + if err != nil { + return stacktrace.Propagate(err, "failed to process config change entry") + } + case raftpb.EntryConfChangeV2: + triggerSnapshot, err = c.processConfigChangeV2Entry(entry.Data) + if err != nil { + return stacktrace.Propagate(err, "failed to process config change v2 entry") + } + } + } + + // wait for all entries to be applied before updating the applied index and potentially triggering a snapshot + wg.Wait() + c.appliedIndex = entries[len(entries)-1].Index + + if triggerSnapshot || c.appliedIndex-c.snapshotIndex >= snapshotInterval { + err := c.storage.triggerSnapshot(c.appliedIndex, &c.confState) + if err != nil { + return stacktrace.Propagate(err, "failed to trigger snapshot") + } + + c.snapshotIndex = c.appliedIndex + } + + return nil +} + +// processNormalEntry passes the proposal to the store and waits for the result to be returned before untracking it. +func (c *Consensus) processNormalEntry(data []byte, wg *sync.WaitGroup) error { + if len(data) <= 0 { + return nil + } + + prop := Proposal{} + err := json.Unmarshal(data, &prop) + if err != nil { + return stacktrace.Propagate(err, "failed to unmarshal committed proposal") + } + + //if readOnly proposal and we did not initiate it, skip it (noop) + if prop.ReadOnly && !c.tracker.isPending(prop.ID) { + return nil + } + + applyDoneC := make(chan ProposalResult, 1) + wg.Go(func() { + res := <-applyDoneC + if c.tracker.isPending(prop.ID) { + c.tracker.untrack(prop.ID, res) + } + }) + + ch, ok := c.commitChs[prop.DBName] + if !ok { + return stacktrace.NewError("no commit channel found for %s", prop.DBName) + } + + ch <- EntryCommit{Prop: prop, Done: applyDoneC} return nil } +func (c *Consensus) dispatchSnapshot(snapshotData []byte) error { + var snapshot map[string][]byte + err := json.Unmarshal(snapshotData, &snapshot) + if err != nil { + return stacktrace.Propagate(err, "failed to unmarshal snapshot data") + } + + var eg errgroup.Group + for name, data := range snapshot { + eg.Go(func() error { + ch, ok := c.commitChs[name] + if !ok { + return stacktrace.NewError("no commit channel found for %s", name) + } + + ch <- EntryCommit{SnapshotData: data} + return nil + }) + } + + return eg.Wait() +} + +// raftpb.ConfChange is still used internally by Raft, we just need to apply the change to the node. +// Changes requested by clients are processed by processConfigChangeV2Entry. +func (c *Consensus) processConfigChangeEntry(data []byte) error { + var cc raftpb.ConfChange + err := cc.Unmarshal(data) + if err != nil { + return stacktrace.Propagate(err, "failed to unmarshal config change data") + } + + c.confState = *c.node.ApplyConfChange(cc) + return nil +} + +func (c *Consensus) processConfigChangeV2Entry(data []byte) (bool, error) { + var cc raftpb.ConfChangeV2 + err := cc.Unmarshal(data) + if err != nil { + return false, stacktrace.Propagate(err, "failed to unmarshal config change data") + } + + c.confState = *c.node.ApplyConfChange(cc) + + // TODO - implement config changes when triggered by a proposal + return false, 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) @@ -124,6 +486,10 @@ func (c *Consensus) ReportSnapshot(id uint64, status raft.SnapshotStatus) { } // RegisterStore allows registering a snapshot provider function for a specific store -func (c *Consensus) RegisterStore(name string, provider snapshotProvider) { +// and returns the channel on which committed entries for that store will be sent. +func (c *Consensus) RegisterStore(name string, provider snapshotProvider) chan EntryCommit { c.storage.registerSnapshotProvider(name, provider) + ch := make(chan EntryCommit) + c.commitChs[name] = ch + return ch } diff --git a/pkg/raftstore/consensus/proposal.go b/pkg/raftstore/consensus/proposal.go new file mode 100644 index 000000000..b1243ae30 --- /dev/null +++ b/pkg/raftstore/consensus/proposal.go @@ -0,0 +1,65 @@ +package consensus + +import ( + "sync" + "time" +) + +type EntryCommit struct { + Prop Proposal + Done chan ProposalResult + + SnapshotData []byte +} + +type Proposal struct { + ID string `json:"id"` + DBName string `json:"dbname"` + Timestamp time.Time `json:"timestamp"` + RequestType string `json:"request_type"` + Value []byte `json:"value"` + ReadOnly bool `json:"read_only"` + Parameters map[string][]byte `json:"parameters,omitempty"` +} + +type ProposalResult struct { + Result any + Error error +} + +type proposalsTracker struct { + sync.Mutex + pending map[string]chan ProposalResult +} + +func newProposalsTracker() *proposalsTracker { + return &proposalsTracker{ + pending: make(map[string]chan ProposalResult), + } +} + +func (p *proposalsTracker) isPending(id string) bool { + p.Lock() + defer p.Unlock() + + _, ok := p.pending[id] + return ok +} + +func (p *proposalsTracker) track(id string) chan ProposalResult { + p.Lock() + defer p.Unlock() + + applied := make(chan ProposalResult, 1) + p.pending[id] = applied + return applied +} + +func (p *proposalsTracker) untrack(id string, result ProposalResult) { + p.Lock() + defer p.Unlock() + + applied := p.pending[id] + applied <- result + delete(p.pending, id) +} diff --git a/pkg/raftstore/params/params.go b/pkg/raftstore/params/params.go index a135813d2..3580dad41 100644 --- a/pkg/raftstore/params/params.go +++ b/pkg/raftstore/params/params.go @@ -5,8 +5,10 @@ import ( "net/url" "strconv" "strings" + "time" "github.com/interuss/stacktrace" + "go.etcd.io/raft/v3" ) const ( @@ -14,13 +16,26 @@ const ( // 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,31 @@ 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 + + Insecure bool + CAFile string + CertFile string + KeyFile string } ) @@ -64,22 +100,59 @@ func (c ConnectParameters) PeerMap() (map[uint64]*url.URL, error) { return nil, stacktrace.Propagate(err, "invalid peer URL %s", parts[1]) } + if !c.Insecure && peerURL.Scheme != "https" { + return nil, stacktrace.NewError("invalid peer URL %s: must use https scheme", parts[1]) + } + + if c.Insecure && (peerURL.Scheme != "http" && peerURL.Scheme != "https") { + return nil, stacktrace.NewError("invalid peer URL %s: must use http or https scheme", parts[1]) + } + peers[id] = peerURL } return peers, nil } +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.") + + flag.BoolVar(&connectParameters.Insecure, "raft_insecure", false, "Enable insecure connection to the Raft cluster. If certificates are provided, they are not verified. If no certificates are provided, HTTPS is disabled entirely.") + flag.StringVar(&connectParameters.CAFile, "raft_tls_ca", "", "Path to the CA certificate file (e.g., /path/to/ca.crt). Required if --raft_insecure is not set.") + flag.StringVar(&connectParameters.CertFile, "raft_tls_crt", "", "Path to the node's TLS certificate file (e.g., /path/to/node.crt). Required if --raft_insecure is not set.") + flag.StringVar(&connectParameters.KeyFile, "raft_tls_key", "", "Path to the node's TLS private key file (e.g., /path/to/node.key). Required if --raft_insecure is not set.") } // GetConnectParameters returns a ConnectParameters instance that gets populated from well-known CLI flags. diff --git a/pkg/raftstore/params/params_test.go b/pkg/raftstore/params/params_test.go index 7410bf64e..f26ea3cff 100644 --- a/pkg/raftstore/params/params_test.go +++ b/pkg/raftstore/params/params_test.go @@ -17,22 +17,22 @@ func TestPeerMap(t *testing.T) { }{ { name: "valid single peer", - peers: "1=http://node1:9021", - want: map[uint64]*url.URL{1: mustParseURL("http://node1:9021")}, + peers: "1=https://node1:9021", + want: map[uint64]*url.URL{1: mustParseURL("https://node1:9021")}, }, { name: "valid multiple peers", - peers: "1=http://node1:9021,2=http://node2:9021,3=http://node3:9021", + peers: "1=https://node1:9021,2=https://node2:9021,3=https://node3:9021", want: map[uint64]*url.URL{ - 1: mustParseURL("http://node1:9021"), - 2: mustParseURL("http://node2:9021"), - 3: mustParseURL("http://node3:9021"), + 1: mustParseURL("https://node1:9021"), + 2: mustParseURL("https://node2:9021"), + 3: mustParseURL("https://node3:9021"), }, }, { name: "valid URL with equals sign in query string", - peers: "1=http://node1:9021?token=abc123", - want: map[uint64]*url.URL{1: mustParseURL("http://node1:9021?token=abc123")}, + peers: "1=https://node1:9021?token=abc123", + want: map[uint64]*url.URL{1: mustParseURL("https://node1:9021?token=abc123")}, }, { name: "invalid empty peers string", @@ -46,27 +46,32 @@ func TestPeerMap(t *testing.T) { }, { name: "invalid non-numeric node ID", - peers: "abc=http://node1:9021", + peers: "abc=https://node1:9021", wantError: true, }, { name: "invalid negative node ID", - peers: "-1=http://node1:9021", + peers: "-1=https://node1:9021", wantError: true, }, { name: "mixed valid and invalid entries", - peers: "1=http://node1:9021,badentry", + peers: "1=https://node1:9021,badentry", wantError: true, }, { name: "invalid zero peer ID", - peers: "0=http://node1:9021", + peers: "0=https://node1:9021", wantError: true, }, { name: "duplicate peer IDs", - peers: "1=http://node1:9021,1=http://node2:9021", + peers: "1=https://node1:9021,1=https://node2:9021", + wantError: true, + }, + { + name: "invalid URL scheme", + peers: "1=http://node1:9021", wantError: true, }, } diff --git a/pkg/raftstore/store.go b/pkg/raftstore/store.go index b602d31c3..f69c181fc 100644 --- a/pkg/raftstore/store.go +++ b/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") } @@ -41,7 +41,7 @@ func Init[R any](ctx context.Context, logger *zap.Logger, newRepo func() R) (*St } // TODO: implement - sharedConsensus.RegisterStore("provider", func() ([]byte, error) { + _ = sharedConsensus.RegisterStore("provider", func() ([]byte, error) { return nil, nil })