From 719b757e731ec0b734e18e3b03fb5185087a5d40 Mon Sep 17 00:00:00 2001 From: Mariem Baccari Date: Tue, 3 Nov 2026 16:54:30 +0100 Subject: [PATCH] [raft] Implement the membership of Raft groups in the Raft stores --- pkg/raftstore/consensus/consensus.go | 94 +++++++++++++++++++++++++--- pkg/raftstore/consensus/node.go | 30 +++++++++ pkg/raftstore/consensus/peernode.go | 65 +++++++++++++++++++ pkg/raftstore/params/params.go | 6 ++ pkg/raftstore/store.go | 8 +-- 5 files changed, 190 insertions(+), 13 deletions(-) create mode 100644 pkg/raftstore/consensus/node.go create mode 100644 pkg/raftstore/consensus/peernode.go diff --git a/pkg/raftstore/consensus/consensus.go b/pkg/raftstore/consensus/consensus.go index 6275f0869..992a78e2c 100644 --- a/pkg/raftstore/consensus/consensus.go +++ b/pkg/raftstore/consensus/consensus.go @@ -7,6 +7,7 @@ import ( "encoding/json" "errors" "fmt" + "net/http" "net/url" "sync" "time" @@ -33,10 +34,12 @@ type Consensus struct { logger *zap.Logger nodeID uint64 + url string node raft.Node - transport *rafthttp.Transport - listener *listener + transport *rafthttp.Transport + listener *listener + peerClient *http.Client // sends the requests to other nodes outside of the Raft transport storage *storage getAppSnapshot func() ([]byte, error) @@ -89,6 +92,7 @@ func NewConsensus(ctx context.Context, logger *zap.Logger, connectParams params. logger: logging.WithValuesFromContext(ctx, logger), nodeID: connectParams.NodeID, + url: nodeUrl.String(), node: node, storage: storage, getAppSnapshot: provider, @@ -106,7 +110,7 @@ func NewConsensus(ctx context.Context, logger *zap.Logger, connectParams params. }, } - err = consensus.initTransport(ctx, connectParams.NodeID, connectParams.GroupClusterID, peers, NewTLSInfo(connectParams, logger)) + err = consensus.initTransport(ctx, connectParams.NodeID, connectParams.GroupClusterID, peers, NewTLSInfo(connectParams, logger), connectParams.PeerRequestTimeout) if err != nil { return nil, stacktrace.Propagate(err, "failed to initialize transport") } @@ -149,6 +153,7 @@ func (c *Consensus) Stop(ctx context.Context) { c.transport.Stop() c.logger.Info("transport stopped") + c.peerClient.CloseIdleConnections() c.node.Stop() c.logger.Info("raft node stopped") }) @@ -267,7 +272,7 @@ func peersList(peers map[uint64]*url.URL) ([]raft.Peer, error) { return result, nil } -func (c *Consensus) initTransport(ctx context.Context, nodeID uint64, clusterID uint64, peers map[uint64]*url.URL, tlsInfo etcdtransport.TLSInfo) error { +func (c *Consensus) initTransport(ctx context.Context, nodeID uint64, clusterID uint64, peers map[uint64]*url.URL, tlsInfo etcdtransport.TLSInfo, peerRequestTimeout time.Duration) error { nodeIDStr := fmt.Sprintf("%d", nodeID) if tlsInfo.Empty() { c.logger.Warn("--raft_insecure is set: Raft nodes do not authenticate each other and connection is not encrypted. Don't use this node in production environments.") @@ -303,11 +308,20 @@ func (c *Consensus) initTransport(ctx context.Context, nodeID uint64, clusterID c.transport = transport c.membership.transport = transport + peerRoundTripper, err := rafthttp.NewRoundTripper(tlsInfo, peerRequestTimeout) + if err != nil { + return stacktrace.Propagate(err, "failed to create the client to other nodes") + } + c.peerClient = &http.Client{Transport: peerRoundTripper, Timeout: peerRequestTimeout} + l, err := getListener() if err != nil { return stacktrace.Propagate(err, "failed to get the listener") } - err = l.register(transport.ClusterID, transport.Handler()) + handler := http.NewServeMux() + handler.HandleFunc("GET "+nodePath, c.serveNode) + handler.Handle("/", transport.Handler()) + err = l.register(transport.ClusterID, handler) if err != nil { return stacktrace.Propagate(err, "failed to register the transport to the listener") } @@ -525,10 +539,29 @@ type confChangeContext struct { member } -// AddNode proposes adding nodeID, reachable by its peers at peerURL, as a voter of the Raft group. -// It blocks until the change is applied or ctx is cancelled. -func (c *Consensus) AddNode(ctx context.Context, nodeID uint64, peerURL string) error { - return c.proposeConfChange(ctx, &raftpb.ConfChangeSingle{Type: raftpb.ConfChangeAddNode.Enum(), NodeId: new(nodeID)}, confChangeContext{member: member{URL: peerURL}}) +// AddNode proposes adding the node reachable by its peers at peerURL as a voter of the Raft group. +// The ID of the node is read from the node itself, which must be waiting. +// It blocks until the change is applied or ctx is cancelled, and returns the added node as a member of the group. +// It returns a dsserr.AlreadyExists error if the node is not waiting or does not run this group in this cluster, or +// if peerURL is already used by another member, and a dsserr.Unavailable error if the node cannot be reached or this +// node is not a member of the group. +func (c *Consensus) AddNode(ctx context.Context, peerURL string) (Node, error) { + node, err := c.fetchNode(ctx, peerURL) + if err != nil { + return Node{}, stacktrace.Propagate(err, "failed to fetch node at %s", peerURL) + } + if node.URL != peerURL { + return Node{}, stacktrace.NewErrorWithCode(dsserr.BadRequest, "the node at %s describes itself as being at %s", peerURL, node.URL) + } + if node.State != NodeWaiting { + return Node{}, stacktrace.NewErrorWithCode(dsserr.AlreadyExists, "the node at %s is %s, only a waiting node can be added", peerURL, node.State) + } + + err = c.proposeConfChange(ctx, &raftpb.ConfChangeSingle{Type: raftpb.ConfChangeAddNode.Enum(), NodeId: new(node.ID)}, confChangeContext{member: member{URL: peerURL}}) + if err != nil { + return Node{}, err + } + return Node{ID: node.ID, URL: peerURL, State: NodeMember}, nil } // RemoveNode proposes removing nodeID from the Raft group. Its ID is retired and can never be reused. @@ -537,6 +570,49 @@ func (c *Consensus) RemoveNode(ctx context.Context, nodeID uint64) error { return c.proposeConfChange(ctx, &raftpb.ConfChangeSingle{Type: raftpb.ConfChangeRemoveNode.Enum(), NodeId: new(nodeID)}, confChangeContext{}) } +// Members returns the membership of the Raft group as currently known by this node. +func (c *Consensus) Members() (Membership, error) { + c.membership.RLock() + nodes := make(map[uint64]Node, len(c.membership.members)+len(c.membership.removed)) + for id, m := range c.membership.members { + nodes[id] = Node{ID: id, URL: m.URL, State: NodeMember} + } + for id, m := range c.membership.removed { + nodes[id] = Node{ID: id, URL: m.URL, State: NodeRemoved} + } + c.membership.RUnlock() + + return Membership{ + ClusterID: uint64(c.transport.ClusterID), + LeaderID: c.node.Status().Lead, + Nodes: nodes, + }, nil +} + +// Node describes this node in the Raft group. +func (c *Consensus) Node() (Node, error) { + return Node{ + ID: c.nodeID, + URL: c.url, + State: c.state(), + }, nil +} + +// state returns the state of this node, which is a member of its group from its first start or was removed from it. +func (c *Consensus) state() NodeState { + if c.membership.isIDRemoved(c.nodeID) { + return NodeRemoved + } + return NodeMember +} + +// Bootstrap starts a new Raft group with this node as its only member. +// TODO: The groups are still formed from --raft_peers: this node is a member of its group from its first start, or +// was removed from it, so a dsserr.AlreadyExists error is always returned. +func (c *Consensus) Bootstrap() (Membership, error) { + return Membership{}, stacktrace.NewErrorWithCode(dsserr.AlreadyExists, "the Raft group is formed from --raft_peers, the state of this node is %s", c.state()) +} + // proposeConfChange proposes a single membership change and blocks until it is applied or ctx is cancelled. // Note that Raft only allows one config change in flight: a change proposed while another one is pending // is silently dropped by the leader, in which case this blocks until ctx is cancelled. diff --git a/pkg/raftstore/consensus/node.go b/pkg/raftstore/consensus/node.go new file mode 100644 index 000000000..98994abd3 --- /dev/null +++ b/pkg/raftstore/consensus/node.go @@ -0,0 +1,30 @@ +package consensus + +// NodeState is the state of a DSS node in a Raft group. +type NodeState string + +const ( + NodeWaiting NodeState = "waiting" + NodeMember NodeState = "member" + NodeRemoved NodeState = "removed" +) + +// Node describes a DSS node in a Raft group. +// Its JSON form is what the DSS nodes exchange on the Raft listener. +type Node struct { + ID uint64 `json:"id"` + URL string `json:"url"` + State NodeState `json:"state"` +} + +// Membership describes a Raft group. +type Membership struct { + // ClusterID identifies the group within the cluster. It is identical on all the members of the group. + ClusterID uint64 + // LeaderID is the ID in Nodes of the leader of the group, 0 if it is unknown. + LeaderID uint64 + // Nodes maps the ID of every node that is or was a member of the group to the node. + // A node's state tells whether it is a member or was removed. + // Removed IDs can never be reused. + Nodes map[uint64]Node +} diff --git a/pkg/raftstore/consensus/peernode.go b/pkg/raftstore/consensus/peernode.go new file mode 100644 index 000000000..45b6638f3 --- /dev/null +++ b/pkg/raftstore/consensus/peernode.go @@ -0,0 +1,65 @@ +package consensus + +import ( + "context" + "encoding/json" + "net/http" + "path" + + dsserr "github.com/interuss/dss/pkg/errors" + "github.com/interuss/stacktrace" + "go.etcd.io/etcd/client/pkg/v3/types" + "go.etcd.io/etcd/server/v3/etcdserver/api/rafthttp" + "go.uber.org/zap" +) + +// nodePath is the path at which a node describes itself in a group on the Raft listener. +// A member reads it to learn the ID of a node before adding it. +var nodePath = path.Join(rafthttp.RaftPrefix, "node") + +// serveNode describes this node in the Raft group. +func (c *Consensus) serveNode(w http.ResponseWriter, r *http.Request) { + node, err := c.Node() + if err != nil { + http.Error(w, "failed to describe this node", http.StatusInternalServerError) + return + } + + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(node); err != nil { + c.logger.Warn("failed to send the description of this node", zap.Error(err)) + } +} + +// fetchNode reads the description in this Raft group of the node at peerURL, from its Raft listener. +func (c *Consensus) fetchNode(ctx context.Context, peerURL string) (Node, error) { + if _, err := types.NewURLs([]string{peerURL}); err != nil { + return Node{}, stacktrace.NewErrorWithCode(dsserr.BadRequest, "invalid URL %q: %v", peerURL, err) + } + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, peerURL+nodePath, nil) + if err != nil { + return Node{}, stacktrace.Propagate(err, "failed to create the request to %s", peerURL) + } + req.Header.Set("X-Etcd-Cluster-ID", c.transport.ClusterID.String()) + + resp, err := c.peerClient.Do(req) + if err != nil { + return Node{}, stacktrace.PropagateWithCode(err, dsserr.Unavailable, "failed to reach the node at %s", peerURL) + } + defer func() { _ = resp.Body.Close() }() + + switch resp.StatusCode { + case http.StatusOK: + case http.StatusPreconditionFailed: + return Node{}, stacktrace.NewErrorWithCode(dsserr.AlreadyExists, "the node at %s does not run this Raft group in this cluster", peerURL) + default: + return Node{}, stacktrace.NewErrorWithCode(dsserr.Unavailable, "unexpected status %d from the node at %s", resp.StatusCode, peerURL) + } + + var node Node + if err := json.NewDecoder(resp.Body).Decode(&node); err != nil { + return Node{}, stacktrace.PropagateWithCode(err, dsserr.Unavailable, "invalid response from the node at %s", peerURL) + } + return node, nil +} diff --git a/pkg/raftstore/params/params.go b/pkg/raftstore/params/params.go index 2c268910e..9bba4fa79 100644 --- a/pkg/raftstore/params/params.go +++ b/pkg/raftstore/params/params.go @@ -31,6 +31,8 @@ const ( defaultMaxSizePerMsg = 1024 * 1024 defaultMaxInflightMsgs = 4096 / 8 + + defaultPeerRequestTimeout = 5 * time.Second ) type ( @@ -56,6 +58,9 @@ type ( MaxSizePerMsg uint64 MaxInflightMsgs int + // PeerRequestTimeout bounds the requests sent to other nodes outside of the Raft transport. + PeerRequestTimeout time.Duration + // TLSCertFile, TLSKeyFile and TLSCAFile configure mutual TLS between Raft nodes. They are required unless Insecure is set. TLSCertFile string TLSKeyFile string @@ -152,6 +157,7 @@ func init() { "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.DurationVar(&connectParameters.PeerRequestTimeout, "raft_peer_request_timeout", defaultPeerRequestTimeout, "Timeout of the requests sent to other nodes outside of the Raft transport, e.g. to read the ID of a node being added to a Raft group. Higher values tolerate slower networks but delay reporting that a node cannot be reached.") flag.StringVar(&connectParameters.TLSCertFile, "raft_tls_cert_file", "", "PEM certificate of this node, presented to the other Raft nodes. Required with --raft_tls_key_file and --raft_tls_ca_file for mutual TLS between Raft nodes, unless --raft_insecure is set. Peer URLs must then use https.") flag.StringVar(&connectParameters.TLSKeyFile, "raft_tls_key_file", "", "PEM private key of --raft_tls_cert_file.") diff --git a/pkg/raftstore/store.go b/pkg/raftstore/store.go index faf072df8..13628f092 100644 --- a/pkg/raftstore/store.go +++ b/pkg/raftstore/store.go @@ -31,7 +31,7 @@ type Store[R any] struct { cancel context.CancelFunc registry map[string]store.OperationHandler[R] - Consensus *consensus.Consensus + *consensus.Consensus done chan struct{} } @@ -119,9 +119,9 @@ func (s *Store[R]) Transact(ctx context.Context, request store.OperationRequest) return nil, stacktrace.Propagate(err, "failed to encode op %q", request.OperationID()) } if handler.IsReadOnly { - return s.Consensus.HandleReadRequest(ctx, consensus.RequestType[any](request.OperationID()), payload) + return s.HandleReadRequest(ctx, consensus.RequestType[any](request.OperationID()), payload) } - return s.Consensus.HandleWriteRequest(ctx, consensus.RequestType[any](request.OperationID()), payload) + return s.HandleWriteRequest(ctx, consensus.RequestType[any](request.OperationID()), payload) } // Interact returns the underlying Raft repo which, for every operation, will propose it to Raft and return the results. @@ -132,7 +132,7 @@ func (s *Store[R]) Interact(_ context.Context) (R, error) { // Close shuts down the consensus instance and processCommits loop. // TODO: pass a context to Stop then to consensus.Stop (see issue: https://github.com/interuss/dss/issues/1610). func (s *Store[R]) Close() error { - s.Consensus.Stop(context.Background()) + s.Stop(context.Background()) s.cancel() s.logger.Info("waiting for commit processing goroutine to exit") <-s.done