const maxSizePerMsg … const maxInflightMsgs … var raftStatusMu … var raftStatus … func init() { … } type apply … type raftNode … type raftNodeConfig … func newRaftNode(cfg raftNodeConfig) *raftNode { … } // raft.Node does not have locks in Raft package func (r *raftNode) tick() { … } func (r *raftNode) getLatestTickTs() time.Time { … } // start prepares and starts raftNode in a new goroutine. It is no longer safe // to modify the fields after it has been started. func (r *raftNode) start(rh *raftReadyHandler) { … } // For a cluster with only one member, the raft may send both the // unstable entries and committed entries to etcdserver, and there // may have overlapped log entries between them. // // etcd responds to the client once it finishes (actually partially) // the applying workflow. But when the client receives the response, // it doesn't mean etcd has already successfully saved the data, // including BoltDB and WAL, because: // 1. etcd commits the boltDB transaction periodically instead of on each request; // 2. etcd saves WAL entries in parallel with applying the committed entries. // // Accordingly, it might run into a situation of data loss when the etcd crashes // immediately after responding to the client and before the boltDB and WAL // successfully save the data to disk. // Note that this issue can only happen for clusters with only one member. // // For clusters with multiple members, it isn't an issue, because etcd will // not commit & apply the data before it being replicated to majority members. // When the client receives the response, it means the data must have been applied. // It further means the data must have been committed. // Note: for clusters with multiple members, the raft will never send identical // unstable entries and committed entries to etcdserver. // // Refer to https://github.com/etcd-io/etcd/issues/14370. func shouldWaitWALSync(rd raft.Ready) bool { … } func updateCommittedIndex(ap *apply, rh *raftReadyHandler) { … } func (r *raftNode) processMessages(ms []raftpb.Message) []raftpb.Message { … } func (r *raftNode) apply() chan apply { … } func (r *raftNode) stop() { … } func (r *raftNode) onStop() { … } // for testing func (r *raftNode) pauseSending() { … } func (r *raftNode) resumeSending() { … } // advanceTicks advances ticks of Raft node. // This can be used for fast-forwarding election // ticks in multi data-center deployments, thus // speeding up election process. func (r *raftNode) advanceTicks(ticks int) { … } func startNode(cfg config.ServerConfig, cl *membership.RaftCluster, ids []types.ID) (id types.ID, n raft.Node, s *raft.MemoryStorage, w *wal.WAL) { … } func restartNode(cfg config.ServerConfig, snapshot *raftpb.Snapshot) (types.ID, *membership.RaftCluster, raft.Node, *raft.MemoryStorage, *wal.WAL) { … } func restartAsStandaloneNode(cfg config.ServerConfig, snapshot *raftpb.Snapshot) (types.ID, *membership.RaftCluster, raft.Node, *raft.MemoryStorage, *wal.WAL) { … } // getIDs returns an ordered set of IDs included in the given snapshot and // the entries. The given snapshot/entries can contain three kinds of // ID-related entry: // - ConfChangeAddNode, in which case the contained ID will be added into the set. // - ConfChangeRemoveNode, in which case the contained ID will be removed from the set. // - ConfChangeAddLearnerNode, in which the contained ID will be added into the set. func getIDs(lg *zap.Logger, snap *raftpb.Snapshot, ents []raftpb.Entry) []uint64 { … } // createConfigChangeEnts creates a series of Raft entries (i.e. // EntryConfChange) to remove the set of given IDs from the cluster. The ID // `self` is _not_ removed, even if present in the set. // If `self` is not inside the given ids, it creates a Raft entry to add a // default member with the given `self`. func createConfigChangeEnts(lg *zap.Logger, ids []uint64, self uint64, term, index uint64) []raftpb.Entry { … }
Codebase Browser
kubernetes