etcdserver, raft: cleanup raft.Storage interface and raft.MemoryStora…

contrib/raftexample/raft.go

@@ -56,7 +56,7 @@ type raftNode struct {
 
 	// raft backing for the commit/error channel
 	node        raft.Node
-	raftStorage *raft.MemoryStorage
+	raftStorage raft.Storage
 	wal         *wal.WAL
 
 	snapshotter      *raftsnap.Snapshotter
@@ -230,7 +230,7 @@ func (rc *raftNode) replayWAL() *wal.WAL {
 	if err != nil {
 		log.Fatalf("raftexample: failed to read WAL (%v)", err)
 	}
-	rc.raftStorage = raft.NewMemoryStorage()
+	rc.raftStorage = raft.NewStorage()
 	if snapshot != nil {
 		rc.raftStorage.ApplySnapshot(*snapshot)
 	}

etcdserver/raft.go

@@ -121,7 +121,7 @@ type raftNodeConfig struct {
 	// to check if msg receiver is removed from cluster
 	isIDRemoved func(id uint64) bool
 	raft.Node
-	raftStorage *raft.MemoryStorage
+	raftStorage raft.Storage
 	storage     Storage
 	heartbeat   time.Duration // for logging
 	// transport specifies the transport to send and receive msgs to members.
@@ -380,7 +380,7 @@ func advanceTicksForElection(n raft.Node, electionTicks int) {
 	}
 }
 
-func startNode(cfg ServerConfig, cl *membership.RaftCluster, ids []types.ID) (id types.ID, n raft.Node, s *raft.MemoryStorage, w *wal.WAL) {
+func startNode(cfg ServerConfig, cl *membership.RaftCluster, ids []types.ID) (id types.ID, n raft.Node, s raft.Storage, w *wal.WAL) {
 	var err error
 	member := cl.MemberByName(cfg.Name)
 	metadata := pbutil.MustMarshal(
@@ -402,7 +402,7 @@ func startNode(cfg ServerConfig, cl *membership.RaftCluster, ids []types.ID) (id
 	}
 	id = member.ID
 	plog.Infof("starting member %s in cluster %s", id, cl.ID())
-	s = raft.NewMemoryStorage()
+	s = raft.NewStorage()
 	c := &raft.Config{
 		ID:              uint64(id),
 		ElectionTick:    cfg.ElectionTicks,
@@ -421,7 +421,7 @@ func startNode(cfg ServerConfig, cl *membership.RaftCluster, ids []types.ID) (id
 	return id, n, s, w
 }
 
-func restartNode(cfg ServerConfig, snapshot *raftpb.Snapshot) (types.ID, *membership.RaftCluster, raft.Node, *raft.MemoryStorage, *wal.WAL) {
+func restartNode(cfg ServerConfig, snapshot *raftpb.Snapshot) (types.ID, *membership.RaftCluster, raft.Node, raft.Storage, *wal.WAL) {
 	var walsnap walpb.Snapshot
 	if snapshot != nil {
 		walsnap.Index, walsnap.Term = snapshot.Metadata.Index, snapshot.Metadata.Term
@@ -431,7 +431,7 @@ func restartNode(cfg ServerConfig, snapshot *raftpb.Snapshot) (types.ID, *member
 	plog.Infof("restarting member %s in cluster %s at commit index %d", id, cid, st.Commit)
 	cl := membership.NewCluster("")
 	cl.SetID(cid)
-	s := raft.NewMemoryStorage()
+	s := raft.NewStorage()
 	if snapshot != nil {
 		s.ApplySnapshot(*snapshot)
 	}
@@ -455,7 +455,7 @@ func restartNode(cfg ServerConfig, snapshot *raftpb.Snapshot) (types.ID, *member
 	return id, cl, n, s, w
 }
 
-func restartAsStandaloneNode(cfg ServerConfig, snapshot *raftpb.Snapshot) (types.ID, *membership.RaftCluster, raft.Node, *raft.MemoryStorage, *wal.WAL) {
+func restartAsStandaloneNode(cfg ServerConfig, snapshot *raftpb.Snapshot) (types.ID, *membership.RaftCluster, raft.Node, raft.Storage, *wal.WAL) {
 	var walsnap walpb.Snapshot
 	if snapshot != nil {
 		walsnap.Index, walsnap.Term = snapshot.Metadata.Index, snapshot.Metadata.Term
@@ -487,7 +487,7 @@ func restartAsStandaloneNode(cfg ServerConfig, snapshot *raftpb.Snapshot) (types
 	plog.Printf("forcing restart of member %s in cluster %s at commit index %d", id, cid, st.Commit)
 	cl := membership.NewCluster("")
 	cl.SetID(cid)
-	s := raft.NewMemoryStorage()
+	s := raft.NewStorage()
 	if snapshot != nil {
 		s.ApplySnapshot(*snapshot)
 	}

etcdserver/raft_test.go

@@ -156,7 +156,7 @@ func TestStopRaftWhenWaitingForApplyDone(t *testing.T) {
 	r := newRaftNode(raftNodeConfig{
 		Node:        n,
 		storage:     mockstorage.NewStorageRecorder(""),
-		raftStorage: raft.NewMemoryStorage(),
+		raftStorage: raft.NewStorage(),
 		transport:   rafthttp.NewNopTransporter(),
 	})
 	srv := &EtcdServer{r: *r}
@@ -183,7 +183,7 @@ func TestConfgChangeBlocksApply(t *testing.T) {
 	r := newRaftNode(raftNodeConfig{
 		Node:        n,
 		storage:     mockstorage.NewStorageRecorder(""),
-		raftStorage: raft.NewMemoryStorage(),
+		raftStorage: raft.NewStorage(),
 		transport:   rafthttp.NewNopTransporter(),
 	})
 	srv := &EtcdServer{r: *r}

etcdserver/server.go

@@ -261,7 +261,7 @@ func NewServer(cfg ServerConfig) (srv *EtcdServer, err error) {
 	var (
 		w  *wal.WAL
 		n  raft.Node
-		s  *raft.MemoryStorage
+		s  raft.Storage
 		id types.ID
 		cl *membership.RaftCluster
 	)

etcdserver/server_test.go

@@ -172,7 +172,7 @@ func TestApplyRepeat(t *testing.T) {
 	cl.AddMember(&membership.Member{ID: 1234})
 	r := newRaftNode(raftNodeConfig{
 		Node:        n,
-		raftStorage: raft.NewMemoryStorage(),
+		raftStorage: raft.NewStorage(),
 		storage:     mockstorage.NewStorageRecorder(""),
 		transport:   rafthttp.NewNopTransporter(),
 	})
@@ -679,7 +679,7 @@ func TestDoProposal(t *testing.T) {
 		r := newRaftNode(raftNodeConfig{
 			Node:        newNodeCommitter(),
 			storage:     mockstorage.NewStorageRecorder(""),
-			raftStorage: raft.NewMemoryStorage(),
+			raftStorage: raft.NewStorage(),
 			transport:   rafthttp.NewNopTransporter(),
 		})
 		srv := &EtcdServer{
@@ -849,7 +849,7 @@ func TestSyncTrigger(t *testing.T) {
 	tk := &time.Ticker{C: st}
 	r := newRaftNode(raftNodeConfig{
 		Node:        n,
-		raftStorage: raft.NewMemoryStorage(),
+		raftStorage: raft.NewStorage(),
 		transport:   rafthttp.NewNopTransporter(),
 		storage:     mockstorage.NewStorageRecorder(""),
 	})
@@ -903,7 +903,7 @@ func TestSnapshot(t *testing.T) {
 		os.RemoveAll(tmpPath)
 	}()
 
-	s := raft.NewMemoryStorage()
+	s := raft.NewStorage()
 	s.Append([]raftpb.Entry{{Index: 1}})
 	st := mockstore.NewRecorderStream()
 	p := mockstorage.NewStorageRecorderStream("")
@@ -972,7 +972,7 @@ func TestSnapshotOrdering(t *testing.T) {
 		t.Fatalf("couldn't make snap dir (%v)", err)
 	}
 
-	rs := raft.NewMemoryStorage()
+	rs := raft.NewStorage()
 	p := mockstorage.NewStorageRecorderStream(testdir)
 	tr, snapDoneC := rafthttp.NewSnapTransporter(snapdir)
 	r := newRaftNode(raftNodeConfig{
@@ -1039,7 +1039,7 @@ func TestTriggerSnap(t *testing.T) {
 	p := mockstorage.NewStorageRecorderStream("")
 	r := newRaftNode(raftNodeConfig{
 		Node:        newNodeCommitter(),
-		raftStorage: raft.NewMemoryStorage(),
+		raftStorage: raft.NewStorage(),
 		storage:     p,
 		transport:   rafthttp.NewNopTransporter(),
 	})
@@ -1098,7 +1098,7 @@ func TestConcurrentApplyAndSnapshotV3(t *testing.T) {
 		t.Fatalf("Couldn't make snap dir (%v)", err)
 	}
 
-	rs := raft.NewMemoryStorage()
+	rs := raft.NewStorage()
 	tr, snapDoneC := rafthttp.NewSnapTransporter(testdir)
 	r := newRaftNode(raftNodeConfig{
 		isIDRemoved: func(id uint64) bool { return cl.IsIDRemoved(types.ID(id)) },
@@ -1187,7 +1187,7 @@ func TestAddMember(t *testing.T) {
 	cl.SetStore(st)
 	r := newRaftNode(raftNodeConfig{
 		Node:        n,
-		raftStorage: raft.NewMemoryStorage(),
+		raftStorage: raft.NewStorage(),
 		storage:     mockstorage.NewStorageRecorder(""),
 		transport:   rafthttp.NewNopTransporter(),
 	})
@@ -1228,7 +1228,7 @@ func TestRemoveMember(t *testing.T) {
 	cl.AddMember(&membership.Member{ID: 1234})
 	r := newRaftNode(raftNodeConfig{
 		Node:        n,
-		raftStorage: raft.NewMemoryStorage(),
+		raftStorage: raft.NewStorage(),
 		storage:     mockstorage.NewStorageRecorder(""),
 		transport:   rafthttp.NewNopTransporter(),
 	})
@@ -1268,7 +1268,7 @@ func TestUpdateMember(t *testing.T) {
 	cl.AddMember(&membership.Member{ID: 1234})
 	r := newRaftNode(raftNodeConfig{
 		Node:        n,
-		raftStorage: raft.NewMemoryStorage(),
+		raftStorage: raft.NewStorage(),
 		storage:     mockstorage.NewStorageRecorder(""),
 		transport:   rafthttp.NewNopTransporter(),
 	})

raft/README.md

@@ -57,7 +57,7 @@ The primary object in raft is a Node. Either start a Node from scratch using raf
 
 To start a three-node cluster
 ```go
-  storage := raft.NewMemoryStorage()
+  storage := raft.NewStorage()
   c := &Config{
     ID:              0x01,
     ElectionTick:    10,
@@ -87,7 +87,7 @@ To allow a new node to join this cluster, do not pass in any peers. First, add t
 
 To restart a node from previous state:
 ```go
-  storage := raft.NewMemoryStorage()
+  storage := raft.NewStorage()
 
   // Recover the in-memory storage from persistent snapshot, state and entries.
   storage.ApplySnapshot(snapshot)

raft/doc.go

@@ -32,7 +32,7 @@ using raft.StartNode or start a Node from some initial state using raft.RestartN
 
 To start a node from scratch:
 
-  storage := raft.NewMemoryStorage()
+  storage := raft.NewStorage()
   c := &Config{
     ID:              0x01,
     ElectionTick:    10,
@@ -45,7 +45,7 @@ To start a node from scratch:
 
 To restart a node from previous state:
 
-  storage := raft.NewMemoryStorage()
+  storage := raft.NewStorage()
 
   // recover the in-memory storage from persistent
   // snapshot, state and entries.

raft/log_test.go

@@ -45,7 +45,7 @@ func TestFindConflict(t *testing.T) {
 	}
 
 	for i, tt := range tests {
-		raftLog := newLog(NewMemoryStorage(), raftLogger)
+		raftLog := newLog(newMemoryStorage(), raftLogger)
 		raftLog.append(previousEnts...)
 
 		gconflict := raftLog.findConflict(tt.ents)
@@ -57,7 +57,7 @@ func TestFindConflict(t *testing.T) {
 
 func TestIsUpToDate(t *testing.T) {
 	previousEnts := []pb.Entry{{Index: 1, Term: 1}, {Index: 2, Term: 2}, {Index: 3, Term: 3}}
-	raftLog := newLog(NewMemoryStorage(), raftLogger)
+	raftLog := newLog(newMemoryStorage(), raftLogger)
 	raftLog.append(previousEnts...)
 	tests := []struct {
 		lastIndex uint64
@@ -123,7 +123,7 @@ func TestAppend(t *testing.T) {
 	}
 
 	for i, tt := range tests {
-		storage := NewMemoryStorage()
+		storage := newMemoryStorage()
 		storage.Append(previousEnts)
 		raftLog := newLog(storage, raftLogger)
 
@@ -236,7 +236,7 @@ func TestLogMaybeAppend(t *testing.T) {
 	}
 
 	for i, tt := range tests {
-		raftLog := newLog(NewMemoryStorage(), raftLogger)
+		raftLog := newLog(newMemoryStorage(), raftLogger)
 		raftLog.append(previousEnts...)
 		raftLog.committed = commit
 		func() {
@@ -280,7 +280,7 @@ func TestCompactionSideEffects(t *testing.T) {
 	lastIndex := uint64(1000)
 	unstableIndex := uint64(750)
 	lastTerm := lastIndex
-	storage := NewMemoryStorage()
+	storage := newMemoryStorage()
 	for i = 1; i <= unstableIndex; i++ {
 		storage.Append([]pb.Entry{{Term: uint64(i), Index: uint64(i)}})
 	}
@@ -356,7 +356,7 @@ func TestHasNextEnts(t *testing.T) {
 		{5, false},
 	}
 	for i, tt := range tests {
-		storage := NewMemoryStorage()
+		storage := newMemoryStorage()
 		storage.ApplySnapshot(snap)
 		raftLog := newLog(storage, raftLogger)
 		raftLog.append(ents...)
@@ -389,7 +389,7 @@ func TestNextEnts(t *testing.T) {
 		{5, nil},
 	}
 	for i, tt := range tests {
-		storage := NewMemoryStorage()
+		storage := newMemoryStorage()
 		storage.ApplySnapshot(snap)
 		raftLog := newLog(storage, raftLogger)
 		raftLog.append(ents...)
@@ -417,7 +417,7 @@ func TestUnstableEnts(t *testing.T) {
 
 	for i, tt := range tests {
 		// append stable entries to storage
-		storage := NewMemoryStorage()
+		storage := newMemoryStorage()
 		storage.Append(previousEnts[:tt.unstable-1])
 
 		// append unstable entries to raftlog
@@ -459,7 +459,7 @@ func TestCommitTo(t *testing.T) {
 					}
 				}
 			}()
-			raftLog := newLog(NewMemoryStorage(), raftLogger)
+			raftLog := newLog(newMemoryStorage(), raftLogger)
 			raftLog.append(previousEnts...)
 			raftLog.committed = commit
 			raftLog.commitTo(tt.commit)
@@ -482,7 +482,7 @@ func TestStableTo(t *testing.T) {
 		{3, 1, 1}, // bad index
 	}
 	for i, tt := range tests {
-		raftLog := newLog(NewMemoryStorage(), raftLogger)
+		raftLog := newLog(newMemoryStorage(), raftLogger)
 		raftLog.append([]pb.Entry{{Index: 1, Term: 1}, {Index: 2, Term: 2}}...)
 		raftLog.stableTo(tt.stablei, tt.stablet)
 		if raftLog.unstable.offset != tt.wunstable {
@@ -517,7 +517,7 @@ func TestStableToWithSnap(t *testing.T) {
 		{snapi - 1, snapt + 1, []pb.Entry{{Index: snapi + 1, Term: snapt}}, snapi + 1},
 	}
 	for i, tt := range tests {
-		s := NewMemoryStorage()
+		s := newMemoryStorage()
 		s.ApplySnapshot(pb.Snapshot{Metadata: pb.SnapshotMetadata{Index: snapi, Term: snapt}})
 		raftLog := newLog(s, raftLogger)
 		raftLog.append(tt.newEnts...)
@@ -553,7 +553,7 @@ func TestCompaction(t *testing.T) {
 				}
 			}()
 
-			storage := NewMemoryStorage()
+			storage := newMemoryStorage()
 			for i := uint64(1); i <= tt.lastIndex; i++ {
 				storage.Append([]pb.Entry{{Index: i}})
 			}
@@ -581,7 +581,7 @@ func TestLogRestore(t *testing.T) {
 	index := uint64(1000)
 	term := uint64(1000)
 	snap := pb.SnapshotMetadata{Index: index, Term: term}
-	storage := NewMemoryStorage()
+	storage := newMemoryStorage()
 	storage.ApplySnapshot(pb.Snapshot{Metadata: snap})
 	raftLog := newLog(storage, raftLogger)
 
@@ -605,7 +605,7 @@ func TestLogRestore(t *testing.T) {
 func TestIsOutOfBounds(t *testing.T) {
 	offset := uint64(100)
 	num := uint64(100)
-	storage := NewMemoryStorage()
+	storage := newMemoryStorage()
 	storage.ApplySnapshot(pb.Snapshot{Metadata: pb.SnapshotMetadata{Index: offset}})
 	l := newLog(storage, raftLogger)
 	for i := uint64(1); i <= num; i++ {
@@ -688,7 +688,7 @@ func TestTerm(t *testing.T) {
 	offset := uint64(100)
 	num := uint64(100)
 
-	storage := NewMemoryStorage()
+	storage := newMemoryStorage()
 	storage.ApplySnapshot(pb.Snapshot{Metadata: pb.SnapshotMetadata{Index: offset, Term: 1}})
 	l := newLog(storage, raftLogger)
 	for i = 1; i < num; i++ {
@@ -718,7 +718,7 @@ func TestTermWithUnstableSnapshot(t *testing.T) {
 	storagesnapi := uint64(100)
 	unstablesnapi := storagesnapi + 5
 
-	storage := NewMemoryStorage()
+	storage := newMemoryStorage()
 	storage.ApplySnapshot(pb.Snapshot{Metadata: pb.SnapshotMetadata{Index: storagesnapi, Term: 1}})
 	l := newLog(storage, raftLogger)
 	l.restore(pb.Snapshot{Metadata: pb.SnapshotMetadata{Index: unstablesnapi, Term: 1}})
@@ -752,7 +752,7 @@ func TestSlice(t *testing.T) {
 	half := offset + num/2
 	halfe := pb.Entry{Index: half, Term: half}
 
-	storage := NewMemoryStorage()
+	storage := newMemoryStorage()
 	storage.ApplySnapshot(pb.Snapshot{Metadata: pb.SnapshotMetadata{Index: offset}})
 	for i = 1; i < num/2; i++ {
 		storage.Append([]pb.Entry{{Index: offset + i, Term: offset + i}})

raft/node_bench_test.go

@@ -25,7 +25,7 @@ func BenchmarkOneNode(b *testing.B) {
 	defer cancel()
 
 	n := newNode()
-	s := NewMemoryStorage()
+	s := newMemoryStorage()
 	r := newTestRaft(1, []uint64{1}, 10, 1, s)
 	go n.run(r)