diff --git a/pkg/raftstore/README.md b/pkg/raftstore/README.md index 377380c67..048a0f4e4 100644 --- a/pkg/raftstore/README.md +++ b/pkg/raftstore/README.md @@ -68,11 +68,11 @@ The members of a Raft group are bootstrapped from the `--raft_peers` flag on the - **Joining:** Starting a node that joins a running group is not supported yet, see [#1802](https://github.com/interuss/dss/issues/1802). - **Validation:** A configuration change is validated when it is proposed, to fail fast, and again by every node when it is applied. The following changes are rejected: - - adding a node that is already a member or with an invalid peer URL. + - adding a node that is already a member, with an invalid peer URL, or with the peer URL of another member. - removing a node that is not a member, or a voter if the group would be left with fewer than 3 voters. - adding or removing a node ID that was previously removed from the group. - **Removal:** A node stops when it applies its own removal. -- **Persistence:** The peer URLs of the members and the removed node IDs are replicated state: they are rebuilt from the configuration changes of the log and stored in snapshots alongside the application state. +- **Persistence:** The peer URLs of the members and of the removed nodes are replicated state: they are rebuilt from the configuration changes of the log and stored in snapshots alongside the application state. ## Future Work diff --git a/pkg/raftstore/consensus/consensus.go b/pkg/raftstore/consensus/consensus.go index 2d94bb783..6275f0869 100644 --- a/pkg/raftstore/consensus/consensus.go +++ b/pkg/raftstore/consensus/consensus.go @@ -102,6 +102,7 @@ func NewConsensus(ctx context.Context, logger *zap.Logger, connectParams params. membership: membership{ nodeID: connectParams.NodeID, members: map[uint64]member{}, + removed: map[uint64]member{}, }, } @@ -124,7 +125,7 @@ func NewConsensus(ctx context.Context, logger *zap.Logger, connectParams params. if err != nil { return nil, stacktrace.Propagate(err, "failed to decode snapshot") } - err = consensus.membership.restore(initSnapshot.Members, initSnapshot.RemovedIDs) + err = consensus.membership.restore(initSnapshot.Members, initSnapshot.Removed) if err != nil { return nil, stacktrace.Propagate(err, "failed to restore membership") } @@ -356,7 +357,7 @@ func (c *Consensus) startRaftUpdatesConsumer(tickInterval time.Duration, snapsho return } - err = c.membership.restore(env.Members, env.RemovedIDs) + err = c.membership.restore(env.Members, env.Removed) if err != nil { c.logger.Error("failed to restore membership", zap.Error(err)) return @@ -527,28 +528,25 @@ type confChangeContext struct { // 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)}, peerURL) + return c.proposeConfChange(ctx, &raftpb.ConfChangeSingle{Type: raftpb.ConfChangeAddNode.Enum(), NodeId: new(nodeID)}, confChangeContext{member: member{URL: peerURL}}) } // RemoveNode proposes removing nodeID from the Raft group. Its ID is retired and can never be reused. // It blocks until the change is applied or ctx is cancelled. func (c *Consensus) RemoveNode(ctx context.Context, nodeID uint64) error { - return c.proposeConfChange(ctx, &raftpb.ConfChangeSingle{Type: raftpb.ConfChangeRemoveNode.Enum(), NodeId: new(nodeID)}, "") + return c.proposeConfChange(ctx, &raftpb.ConfChangeSingle{Type: raftpb.ConfChangeRemoveNode.Enum(), NodeId: new(nodeID)}, confChangeContext{}) } // 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. -func (c *Consensus) proposeConfChange(ctx context.Context, change *raftpb.ConfChangeSingle, peerURL string) error { - err := c.membership.validateConfChange(change, peerURL) +func (c *Consensus) proposeConfChange(ctx context.Context, change *raftpb.ConfChangeSingle, cctx confChangeContext) error { + err := c.membership.validateConfChange(change, cctx.member) if err != nil { return err } - cctx := confChangeContext{ - ID: uuid.New().String(), - URL: peerURL, - } + cctx.ID = uuid.New().String() ctxBuf, err := json.Marshal(cctx) if err != nil { @@ -598,7 +596,7 @@ func (c *Consensus) applyConfigChangeV2(data []byte) error { if len(cc.GetChanges()) != 1 { err = stacktrace.NewErrorWithCode(dsserr.BadRequest, "expected exactly one change, got %d", len(cc.GetChanges())) } else { - err = c.membership.validateConfChange(cc.GetChanges()[0], cctx.URL) + err = c.membership.validateConfChange(cc.GetChanges()[0], cctx.member) } if err != nil { c.logger.Warn("rejecting config change", zap.String("id", cctx.ID), zap.Error(err)) @@ -642,10 +640,10 @@ func (c *Consensus) updateSnapshotConfState(msgs []*raftpb.Message) { // so that a node catching up via a snapshot fully recovers it, including which IDs must never be reused. // TODO: Use protobuf instead of gob as a follow-up to https://github.com/interuss/dss/pull/1765 type snapshot struct { - // Members maps the ID of every member of the group to its description. + // Members maps the ID of every member of the group to its membership properties. Members map[uint64]member - // RemovedIDs are the IDs of the nodes permanently removed from the group. - RemovedIDs []uint64 + // Removed maps the ID of every node permanently removed from the group to its membership properties when it was removed. + Removed map[uint64]member // AppData is the application snapshot data. AppData []byte } @@ -659,8 +657,8 @@ func (c *Consensus) encodeSnapshot() ([]byte, error) { } env := snapshot{ - Members: c.membership.members, - RemovedIDs: c.membership.removedIDs, + Members: c.membership.members, + Removed: c.membership.removed, // TODO: Avoid encoding the data twice (to be addressed as part of https://github.com/interuss/dss/pull/1765) AppData: appData, } diff --git a/pkg/raftstore/consensus/membership.go b/pkg/raftstore/consensus/membership.go index e39462e87..be9f438c2 100644 --- a/pkg/raftstore/consensus/membership.go +++ b/pkg/raftstore/consensus/membership.go @@ -1,7 +1,6 @@ package consensus import ( - "slices" "sync" dsserr "github.com/interuss/dss/pkg/errors" @@ -26,15 +25,16 @@ type membership struct { // members maps the ID of every member of the group to its description. members map[uint64]member - // removedIDs are the IDs of the nodes removed from the group - // removed IDs can never be reused. - removedIDs []uint64 + // removed maps the ID of every node removed from the group to its description when it was removed. + // Removed IDs can never be reused. + removed map[uint64]member } func (m *membership) isIDRemoved(id uint64) bool { m.RLock() defer m.RUnlock() - return slices.Contains(m.removedIDs, id) + _, ok := m.removed[id] + return ok } // addNode records nodeID as a member described by mem and adds it as a peer of the transport. @@ -48,11 +48,12 @@ func (m *membership) addNode(nodeID uint64, mem member) { } } -// removeNode removes nodeID from the members, retires its ID and removes it from the peers of the transport. +// removeNode moves nodeID from the members to the removed nodes, which retires its ID, and removes it from the peers +// of the transport. func (m *membership) removeNode(nodeID uint64) { m.Lock() + m.removed[nodeID] = m.members[nodeID] delete(m.members, nodeID) - m.removedIDs = append(m.removedIDs, nodeID) m.Unlock() if nodeID != m.nodeID { @@ -60,9 +61,9 @@ func (m *membership) removeNode(nodeID uint64) { } } -// restore replaces the members and the removed IDs and updates the transport peers accordingly. -func (m *membership) restore(members map[uint64]member, removedIDs []uint64) error { - for _, id := range removedIDs { +// restore replaces the members and the removed nodes and updates the transport peers accordingly. +func (m *membership) restore(members map[uint64]member, removed map[uint64]member) error { + for id := range removed { if _, ok := members[id]; ok { return stacktrace.NewError("removed ID %d is still present in the members map", id) } @@ -70,16 +71,19 @@ func (m *membership) restore(members map[uint64]member, removedIDs []uint64) err if members == nil { members = map[uint64]member{} } + if removed == nil { + removed = map[uint64]member{} + } m.Lock() m.members = members - m.removedIDs = removedIDs + m.removed = removed m.Unlock() // It is safe to do this outside the lock because the transport methods as well as addNode and removeNode // are only called by the consumer goroutine. So all transport updates happen one at a time. The lock only protects // the readers on other goroutine (isIDRemoved, validateConfChange etc.) who never call the transport methods. - for _, id := range removedIDs { + for id := range removed { m.transport.RemovePeer(types.ID(id)) } for id, mem := range members { @@ -91,11 +95,13 @@ func (m *membership) restore(members map[uint64]member, removedIDs []uint64) err } // validateConfChange checks that applying change keeps the membership valid. +// When adding a node, mem describes it. // NB: This method must remain deterministic as it is called by every node when the change is applied. // It uses members and not the ConfState, which can be more recent than the members after restoring // a snapshot received from the leader (see Consensus.updateSnapshotConfState). -// A rejected change is reported as a dsserr.BadRequest error. -func (m *membership) validateConfChange(change *raftpb.ConfChangeSingle, peerURL string) error { +// A rejected change is reported as a dsserr.BadRequest error, or a dsserr.AlreadyExists error if the URL of the node +// is already used by a member. +func (m *membership) validateConfChange(change *raftpb.ConfChangeSingle, mem member) error { m.RLock() defer m.RUnlock() @@ -103,7 +109,7 @@ func (m *membership) validateConfChange(change *raftpb.ConfChangeSingle, peerURL return stacktrace.NewErrorWithCode(dsserr.BadRequest, "node ID %d is invalid", change.GetNodeId()) } - if slices.Contains(m.removedIDs, change.GetNodeId()) { + if _, ok := m.removed[change.GetNodeId()]; ok { return stacktrace.NewErrorWithCode(dsserr.BadRequest, "node ID %d was removed from the group and cannot be reused", change.GetNodeId()) } @@ -113,8 +119,13 @@ func (m *membership) validateConfChange(change *raftpb.ConfChangeSingle, peerURL if isMember { return stacktrace.NewErrorWithCode(dsserr.BadRequest, "node %d is already a member of the group", change.GetNodeId()) } - if _, err := types.NewURLs([]string{peerURL}); err != nil { - return stacktrace.NewErrorWithCode(dsserr.BadRequest, "invalid URL %q for node %d: %v", peerURL, change.GetNodeId(), err) + if _, err := types.NewURLs([]string{mem.URL}); err != nil { + return stacktrace.NewErrorWithCode(dsserr.BadRequest, "invalid URL %q for node %d: %v", mem.URL, change.GetNodeId(), err) + } + for id, other := range m.members { + if other.URL == mem.URL { + return stacktrace.NewErrorWithCode(dsserr.AlreadyExists, "the URL %q of node %d is already used by member %d", mem.URL, change.GetNodeId(), id) + } } case raftpb.ConfChangeRemoveNode: diff --git a/pkg/raftstore/consensus/membership_test.go b/pkg/raftstore/consensus/membership_test.go index c2f727014..7ee460346 100644 --- a/pkg/raftstore/consensus/membership_test.go +++ b/pkg/raftstore/consensus/membership_test.go @@ -13,7 +13,7 @@ import ( func TestValidateConfChange(t *testing.T) { voters := []uint64{1, 2, 3, 4} - removed := []uint64{5} + removed := map[uint64]member{5: {URL: "http://node5:9011"}} add := func(id uint64) *raftpb.ConfChangeSingle { return &raftpb.ConfChangeSingle{Type: raftpb.ConfChangeAddNode.Enum(), NodeId: new(id)} } @@ -21,19 +21,24 @@ func TestValidateConfChange(t *testing.T) { return &raftpb.ConfChangeSingle{Type: raftpb.ConfChangeRemoveNode.Enum(), NodeId: new(id)} } + node6 := member{URL: "http://node6:9011"} + tests := []struct { - name string - change *raftpb.ConfChangeSingle - url string - voters []uint64 - wantErr bool + name string + change *raftpb.ConfChangeSingle + node member + voters []uint64 + wantErr bool + wantCode stacktrace.ErrorCode }{ - {name: "add new node", change: add(6), url: "http://node6:9011", voters: voters}, - {name: "add invalid ID", change: add(raft.None), url: "http://node6:9011", voters: voters, wantErr: true}, - {name: "add removed ID", change: add(5), url: "http://node5:9011", voters: voters, wantErr: true}, - {name: "add existing member", change: add(2), url: "http://node2:9011", voters: voters, wantErr: true}, - {name: "add without URL", change: add(6), voters: voters, wantErr: true}, - {name: "add with URL without port", change: add(6), url: "http://node6", voters: voters, wantErr: true}, + {name: "add new node", change: add(6), node: node6, voters: voters}, + {name: "add invalid ID", change: add(raft.None), node: node6, voters: voters, wantErr: true}, + {name: "add removed ID", change: add(5), node: member{URL: "http://node5:9011"}, voters: voters, wantErr: true}, + {name: "add existing member", change: add(2), node: member{URL: "http://node2:9011"}, voters: voters, wantErr: true}, + {name: "add without URL", change: add(6), node: member{}, voters: voters, wantErr: true}, + {name: "add with URL without port", change: add(6), node: member{URL: "http://node6"}, voters: voters, wantErr: true}, + {name: "add with the URL of a member", change: add(6), node: member{URL: "http://node2:9011"}, voters: voters, wantErr: true, wantCode: dsserr.AlreadyExists}, + {name: "add with the URL of a removed node", change: add(6), node: member{URL: "http://node5:9011"}, voters: voters}, {name: "remove member", change: remove(4), voters: voters}, {name: "remove removed ID", change: remove(5), voters: voters, wantErr: true}, {name: "remove non member", change: remove(6), voters: voters, wantErr: true}, @@ -46,13 +51,17 @@ func TestValidateConfChange(t *testing.T) { for _, id := range tt.voters { members[id] = member{URL: fmt.Sprintf("http://node%d:9011", id)} } - m := &membership{members: members, removedIDs: removed} - err := m.validateConfChange(tt.change, tt.url) + m := &membership{members: members, removed: removed} + err := m.validateConfChange(tt.change, tt.node) if !tt.wantErr { require.NoError(t, err) return } - require.Equal(t, dsserr.BadRequest, stacktrace.GetCode(err)) + wantCode := dsserr.BadRequest + if tt.wantCode != 0 { + wantCode = tt.wantCode + } + require.Equal(t, wantCode, stacktrace.GetCode(err)) }) } }