[raft] Implement the membership of Raft groups in the Raft stores

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.

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
+}

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
+}

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.")

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