| // Copyright 2015 The etcd Authors |
| // |
| // Licensed under the Apache License, Version 2.0 (the "License"); |
| // you may not use this file except in compliance with the License. |
| // You may obtain a copy of the License at |
| // |
| // http://www.apache.org/licenses/LICENSE-2.0 |
| // |
| // Unless required by applicable law or agreed to in writing, software |
| // distributed under the License is distributed on an "AS IS" BASIS, |
| // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| // See the License for the specific language governing permissions and |
| // limitations under the License. |
| |
| package main |
| |
| import ( |
| "context" |
| "fmt" |
| "log" |
| "net/http" |
| "net/url" |
| "os" |
| "strconv" |
| "time" |
| |
| "github.com/coreos/etcd/etcdserver/stats" |
| "github.com/coreos/etcd/pkg/fileutil" |
| "github.com/coreos/etcd/pkg/types" |
| "github.com/coreos/etcd/raft" |
| "github.com/coreos/etcd/raft/raftpb" |
| "github.com/coreos/etcd/rafthttp" |
| "github.com/coreos/etcd/snap" |
| "github.com/coreos/etcd/wal" |
| "github.com/coreos/etcd/wal/walpb" |
| ) |
| |
| // A key-value stream backed by raft |
| type raftNode struct { |
| proposeC <-chan string // proposed messages (k,v) |
| confChangeC <-chan raftpb.ConfChange // proposed cluster config changes |
| commitC chan<- *string // entries committed to log (k,v) |
| errorC chan<- error // errors from raft session |
| |
| id int // client ID for raft session |
| peers []string // raft peer URLs |
| join bool // node is joining an existing cluster |
| waldir string // path to WAL directory |
| snapdir string // path to snapshot directory |
| getSnapshot func() ([]byte, error) |
| lastIndex uint64 // index of log at start |
| |
| confState raftpb.ConfState |
| snapshotIndex uint64 |
| appliedIndex uint64 |
| |
| // raft backing for the commit/error channel |
| node raft.Node |
| raftStorage *raft.MemoryStorage |
| wal *wal.WAL |
| |
| snapshotter *snap.Snapshotter |
| snapshotterReady chan *snap.Snapshotter // signals when snapshotter is ready |
| |
| snapCount uint64 |
| transport *rafthttp.Transport |
| stopc chan struct{} // signals proposal channel closed |
| httpstopc chan struct{} // signals http server to shutdown |
| httpdonec chan struct{} // signals http server shutdown complete |
| } |
| |
| var defaultSnapCount uint64 = 10000 |
| |
| // newRaftNode initiates a raft instance and returns a committed log entry |
| // channel and error channel. Proposals for log updates are sent over the |
| // provided the proposal channel. All log entries are replayed over the |
| // commit channel, followed by a nil message (to indicate the channel is |
| // current), then new log entries. To shutdown, close proposeC and read errorC. |
| func newRaftNode(id int, peers []string, join bool, getSnapshot func() ([]byte, error), proposeC <-chan string, |
| confChangeC <-chan raftpb.ConfChange) (<-chan *string, <-chan error, <-chan *snap.Snapshotter) { |
| |
| commitC := make(chan *string) |
| errorC := make(chan error) |
| |
| rc := &raftNode{ |
| proposeC: proposeC, |
| confChangeC: confChangeC, |
| commitC: commitC, |
| errorC: errorC, |
| id: id, |
| peers: peers, |
| join: join, |
| waldir: fmt.Sprintf("raftexample-%d", id), |
| snapdir: fmt.Sprintf("raftexample-%d-snap", id), |
| getSnapshot: getSnapshot, |
| snapCount: defaultSnapCount, |
| stopc: make(chan struct{}), |
| httpstopc: make(chan struct{}), |
| httpdonec: make(chan struct{}), |
| |
| snapshotterReady: make(chan *snap.Snapshotter, 1), |
| // rest of structure populated after WAL replay |
| } |
| go rc.startRaft() |
| return commitC, errorC, rc.snapshotterReady |
| } |
| |
| func (rc *raftNode) saveSnap(snap raftpb.Snapshot) error { |
| // must save the snapshot index to the WAL before saving the |
| // snapshot to maintain the invariant that we only Open the |
| // wal at previously-saved snapshot indexes. |
| walSnap := walpb.Snapshot{ |
| Index: snap.Metadata.Index, |
| Term: snap.Metadata.Term, |
| } |
| if err := rc.wal.SaveSnapshot(walSnap); err != nil { |
| return err |
| } |
| if err := rc.snapshotter.SaveSnap(snap); err != nil { |
| return err |
| } |
| return rc.wal.ReleaseLockTo(snap.Metadata.Index) |
| } |
| |
| func (rc *raftNode) entriesToApply(ents []raftpb.Entry) (nents []raftpb.Entry) { |
| if len(ents) == 0 { |
| return |
| } |
| firstIdx := ents[0].Index |
| if firstIdx > rc.appliedIndex+1 { |
| log.Fatalf("first index of committed entry[%d] should <= progress.appliedIndex[%d] 1", firstIdx, rc.appliedIndex) |
| } |
| if rc.appliedIndex-firstIdx+1 < uint64(len(ents)) { |
| nents = ents[rc.appliedIndex-firstIdx+1:] |
| } |
| return nents |
| } |
| |
| // publishEntries writes committed log entries to commit channel and returns |
| // whether all entries could be published. |
| func (rc *raftNode) publishEntries(ents []raftpb.Entry) bool { |
| for i := range ents { |
| switch ents[i].Type { |
| case raftpb.EntryNormal: |
| if len(ents[i].Data) == 0 { |
| // ignore empty messages |
| break |
| } |
| s := string(ents[i].Data) |
| select { |
| case rc.commitC <- &s: |
| case <-rc.stopc: |
| return false |
| } |
| |
| case raftpb.EntryConfChange: |
| var cc raftpb.ConfChange |
| cc.Unmarshal(ents[i].Data) |
| rc.confState = *rc.node.ApplyConfChange(cc) |
| switch cc.Type { |
| case raftpb.ConfChangeAddNode: |
| if len(cc.Context) > 0 { |
| rc.transport.AddPeer(types.ID(cc.NodeID), []string{string(cc.Context)}) |
| } |
| case raftpb.ConfChangeRemoveNode: |
| if cc.NodeID == uint64(rc.id) { |
| log.Println("I've been removed from the cluster! Shutting down.") |
| return false |
| } |
| rc.transport.RemovePeer(types.ID(cc.NodeID)) |
| } |
| } |
| |
| // after commit, update appliedIndex |
| rc.appliedIndex = ents[i].Index |
| |
| // special nil commit to signal replay has finished |
| if ents[i].Index == rc.lastIndex { |
| select { |
| case rc.commitC <- nil: |
| case <-rc.stopc: |
| return false |
| } |
| } |
| } |
| return true |
| } |
| |
| func (rc *raftNode) loadSnapshot() *raftpb.Snapshot { |
| snapshot, err := rc.snapshotter.Load() |
| if err != nil && err != snap.ErrNoSnapshot { |
| log.Fatalf("raftexample: error loading snapshot (%v)", err) |
| } |
| return snapshot |
| } |
| |
| // openWAL returns a WAL ready for reading. |
| func (rc *raftNode) openWAL(snapshot *raftpb.Snapshot) *wal.WAL { |
| if !wal.Exist(rc.waldir) { |
| if err := os.Mkdir(rc.waldir, 0750); err != nil { |
| log.Fatalf("raftexample: cannot create dir for wal (%v)", err) |
| } |
| |
| w, err := wal.Create(rc.waldir, nil) |
| if err != nil { |
| log.Fatalf("raftexample: create wal error (%v)", err) |
| } |
| w.Close() |
| } |
| |
| walsnap := walpb.Snapshot{} |
| if snapshot != nil { |
| walsnap.Index, walsnap.Term = snapshot.Metadata.Index, snapshot.Metadata.Term |
| } |
| log.Printf("loading WAL at term %d and index %d", walsnap.Term, walsnap.Index) |
| w, err := wal.Open(rc.waldir, walsnap) |
| if err != nil { |
| log.Fatalf("raftexample: error loading wal (%v)", err) |
| } |
| |
| return w |
| } |
| |
| // replayWAL replays WAL entries into the raft instance. |
| func (rc *raftNode) replayWAL() *wal.WAL { |
| log.Printf("replaying WAL of member %d", rc.id) |
| snapshot := rc.loadSnapshot() |
| w := rc.openWAL(snapshot) |
| _, st, ents, err := w.ReadAll() |
| if err != nil { |
| log.Fatalf("raftexample: failed to read WAL (%v)", err) |
| } |
| rc.raftStorage = raft.NewMemoryStorage() |
| if snapshot != nil { |
| rc.raftStorage.ApplySnapshot(*snapshot) |
| } |
| rc.raftStorage.SetHardState(st) |
| |
| // append to storage so raft starts at the right place in log |
| rc.raftStorage.Append(ents) |
| // send nil once lastIndex is published so client knows commit channel is current |
| if len(ents) > 0 { |
| rc.lastIndex = ents[len(ents)-1].Index |
| } else { |
| rc.commitC <- nil |
| } |
| return w |
| } |
| |
| func (rc *raftNode) writeError(err error) { |
| rc.stopHTTP() |
| close(rc.commitC) |
| rc.errorC <- err |
| close(rc.errorC) |
| rc.node.Stop() |
| } |
| |
| func (rc *raftNode) startRaft() { |
| if !fileutil.Exist(rc.snapdir) { |
| if err := os.Mkdir(rc.snapdir, 0750); err != nil { |
| log.Fatalf("raftexample: cannot create dir for snapshot (%v)", err) |
| } |
| } |
| rc.snapshotter = snap.New(rc.snapdir) |
| rc.snapshotterReady <- rc.snapshotter |
| |
| oldwal := wal.Exist(rc.waldir) |
| rc.wal = rc.replayWAL() |
| |
| rpeers := make([]raft.Peer, len(rc.peers)) |
| for i := range rpeers { |
| rpeers[i] = raft.Peer{ID: uint64(i + 1)} |
| } |
| c := &raft.Config{ |
| ID: uint64(rc.id), |
| ElectionTick: 10, |
| HeartbeatTick: 1, |
| Storage: rc.raftStorage, |
| MaxSizePerMsg: 1024 * 1024, |
| MaxInflightMsgs: 256, |
| } |
| |
| if oldwal { |
| rc.node = raft.RestartNode(c) |
| } else { |
| startPeers := rpeers |
| if rc.join { |
| startPeers = nil |
| } |
| rc.node = raft.StartNode(c, startPeers) |
| } |
| |
| rc.transport = &rafthttp.Transport{ |
| ID: types.ID(rc.id), |
| ClusterID: 0x1000, |
| Raft: rc, |
| ServerStats: stats.NewServerStats("", ""), |
| LeaderStats: stats.NewLeaderStats(strconv.Itoa(rc.id)), |
| ErrorC: make(chan error), |
| } |
| |
| rc.transport.Start() |
| for i := range rc.peers { |
| if i+1 != rc.id { |
| rc.transport.AddPeer(types.ID(i+1), []string{rc.peers[i]}) |
| } |
| } |
| |
| go rc.serveRaft() |
| go rc.serveChannels() |
| } |
| |
| // stop closes http, closes all channels, and stops raft. |
| func (rc *raftNode) stop() { |
| rc.stopHTTP() |
| close(rc.commitC) |
| close(rc.errorC) |
| rc.node.Stop() |
| } |
| |
| func (rc *raftNode) stopHTTP() { |
| rc.transport.Stop() |
| close(rc.httpstopc) |
| <-rc.httpdonec |
| } |
| |
| func (rc *raftNode) publishSnapshot(snapshotToSave raftpb.Snapshot) { |
| if raft.IsEmptySnap(snapshotToSave) { |
| return |
| } |
| |
| log.Printf("publishing snapshot at index %d", rc.snapshotIndex) |
| defer log.Printf("finished publishing snapshot at index %d", rc.snapshotIndex) |
| |
| if snapshotToSave.Metadata.Index <= rc.appliedIndex { |
| log.Fatalf("snapshot index [%d] should > progress.appliedIndex [%d] + 1", snapshotToSave.Metadata.Index, rc.appliedIndex) |
| } |
| rc.commitC <- nil // trigger kvstore to load snapshot |
| |
| rc.confState = snapshotToSave.Metadata.ConfState |
| rc.snapshotIndex = snapshotToSave.Metadata.Index |
| rc.appliedIndex = snapshotToSave.Metadata.Index |
| } |
| |
| var snapshotCatchUpEntriesN uint64 = 10000 |
| |
| func (rc *raftNode) maybeTriggerSnapshot() { |
| if rc.appliedIndex-rc.snapshotIndex <= rc.snapCount { |
| return |
| } |
| |
| log.Printf("start snapshot [applied index: %d | last snapshot index: %d]", rc.appliedIndex, rc.snapshotIndex) |
| data, err := rc.getSnapshot() |
| if err != nil { |
| log.Panic(err) |
| } |
| snap, err := rc.raftStorage.CreateSnapshot(rc.appliedIndex, &rc.confState, data) |
| if err != nil { |
| panic(err) |
| } |
| if err := rc.saveSnap(snap); err != nil { |
| panic(err) |
| } |
| |
| compactIndex := uint64(1) |
| if rc.appliedIndex > snapshotCatchUpEntriesN { |
| compactIndex = rc.appliedIndex - snapshotCatchUpEntriesN |
| } |
| if err := rc.raftStorage.Compact(compactIndex); err != nil { |
| panic(err) |
| } |
| |
| log.Printf("compacted log at index %d", compactIndex) |
| rc.snapshotIndex = rc.appliedIndex |
| } |
| |
| func (rc *raftNode) serveChannels() { |
| snap, err := rc.raftStorage.Snapshot() |
| if err != nil { |
| panic(err) |
| } |
| rc.confState = snap.Metadata.ConfState |
| rc.snapshotIndex = snap.Metadata.Index |
| rc.appliedIndex = snap.Metadata.Index |
| |
| defer rc.wal.Close() |
| |
| ticker := time.NewTicker(100 * time.Millisecond) |
| defer ticker.Stop() |
| |
| // send proposals over raft |
| go func() { |
| var confChangeCount uint64 = 0 |
| |
| for rc.proposeC != nil && rc.confChangeC != nil { |
| select { |
| case prop, ok := <-rc.proposeC: |
| if !ok { |
| rc.proposeC = nil |
| } else { |
| // blocks until accepted by raft state machine |
| rc.node.Propose(context.TODO(), []byte(prop)) |
| } |
| |
| case cc, ok := <-rc.confChangeC: |
| if !ok { |
| rc.confChangeC = nil |
| } else { |
| confChangeCount += 1 |
| cc.ID = confChangeCount |
| rc.node.ProposeConfChange(context.TODO(), cc) |
| } |
| } |
| } |
| // client closed channel; shutdown raft if not already |
| close(rc.stopc) |
| }() |
| |
| // event loop on raft state machine updates |
| for { |
| select { |
| case <-ticker.C: |
| rc.node.Tick() |
| |
| // store raft entries to wal, then publish over commit channel |
| case rd := <-rc.node.Ready(): |
| rc.wal.Save(rd.HardState, rd.Entries) |
| if !raft.IsEmptySnap(rd.Snapshot) { |
| rc.saveSnap(rd.Snapshot) |
| rc.raftStorage.ApplySnapshot(rd.Snapshot) |
| rc.publishSnapshot(rd.Snapshot) |
| } |
| rc.raftStorage.Append(rd.Entries) |
| rc.transport.Send(rd.Messages) |
| if ok := rc.publishEntries(rc.entriesToApply(rd.CommittedEntries)); !ok { |
| rc.stop() |
| return |
| } |
| rc.maybeTriggerSnapshot() |
| rc.node.Advance() |
| |
| case err := <-rc.transport.ErrorC: |
| rc.writeError(err) |
| return |
| |
| case <-rc.stopc: |
| rc.stop() |
| return |
| } |
| } |
| } |
| |
| func (rc *raftNode) serveRaft() { |
| url, err := url.Parse(rc.peers[rc.id-1]) |
| if err != nil { |
| log.Fatalf("raftexample: Failed parsing URL (%v)", err) |
| } |
| |
| ln, err := newStoppableListener(url.Host, rc.httpstopc) |
| if err != nil { |
| log.Fatalf("raftexample: Failed to listen rafthttp (%v)", err) |
| } |
| |
| err = (&http.Server{Handler: rc.transport.Handler()}).Serve(ln) |
| select { |
| case <-rc.httpstopc: |
| default: |
| log.Fatalf("raftexample: Failed to serve rafthttp (%v)", err) |
| } |
| close(rc.httpdonec) |
| } |
| |
| func (rc *raftNode) Process(ctx context.Context, m raftpb.Message) error { |
| return rc.node.Step(ctx, m) |
| } |
| func (rc *raftNode) IsIDRemoved(id uint64) bool { return false } |
| func (rc *raftNode) ReportUnreachable(id uint64) {} |
| func (rc *raftNode) ReportSnapshot(id uint64, status raft.SnapshotStatus) {} |