| 129 | } |
| 130 | |
| 131 | func (n *Node) run() error { |
| 132 | // The node is already running |
| 133 | if !n.isRunning.CompareAndSwap(false, true) { |
| 134 | return nil |
| 135 | } |
| 136 | n.shutdown = make(chan struct{}) |
| 137 | |
| 138 | peers := make([]raft.Peer, len(n.config.Peers)) |
| 139 | for i, peer := range n.config.Peers { |
| 140 | peers[i] = raft.Peer{ |
| 141 | ID: uint64(i + 1), |
| 142 | Context: []byte(peer), |
| 143 | } |
| 144 | } |
| 145 | raftConfig := &raft.Config{ |
| 146 | ID: n.config.ID, |
| 147 | HeartbeatTick: n.config.HeartbeatSeconds, |
| 148 | ElectionTick: n.config.ElectionSeconds, |
| 149 | MaxInflightMsgs: 128, |
| 150 | MaxSizePerMsg: 10 * 1024 * 1024, // 10 MiB |
| 151 | Storage: n.dataStore.raftStorage, |
| 152 | Logger: Logger{SugaredLogger: n.logger.Sugar()}, |
| 153 | } |
| 154 | |
| 155 | // WAL existing check must be done before replayWAL since it will create a new WAL if not exists |
| 156 | walExists := n.dataStore.walExists() |
| 157 | snapshot, err := n.dataStore.replayWAL() |
| 158 | if err != nil { |
| 159 | return err |
| 160 | } |
| 161 | n.appliedIndex = snapshot.Metadata.Index |
| 162 | n.snapshotIndex = snapshot.Metadata.Index |
| 163 | n.confState = snapshot.Metadata.ConfState |
| 164 | |
| 165 | if n.config.ClusterState == ClusterStateExisting || walExists { |
| 166 | n.raftNode = raft.RestartNode(raftConfig) |
| 167 | } else { |
| 168 | n.raftNode = raft.StartNode(raftConfig, peers) |
| 169 | } |
| 170 | |
| 171 | if err := n.runTransport(); err != nil { |
| 172 | return err |
| 173 | } |
| 174 | n.watchLeaderChange() |
| 175 | return n.runRaftMessages() |
| 176 | } |
| 177 | |
| 178 | func (n *Node) runTransport() error { |
| 179 | logger := logger.Get() |