Merge pull request #3741 from yichengq/receive-restore

etcdserver: restore KV snapshot when receiving snapshot

etcdserver/server.go

@@ -493,6 +493,23 @@ func (s *EtcdServer) run() {
 						apply.snapshot.Metadata.Index, appliedi)
 				}
 
+				if s.cfg.V3demo {
+					if err := s.kv.Close(); err != nil {
+						plog.Panicf("close KV error: %v", err)
+					}
+					snapfn, err := s.r.raftStorage.snapStore.getSnapFilePath(apply.snapshot.Metadata.Index)
+					if err != nil {
+						plog.Panicf("get snapshot file path error: %v", err)
+					}
+					fn := path.Join(s.cfg.StorageDir(), databaseFilename)
+					if err := os.Rename(snapfn, fn); err != nil {
+						plog.Panicf("rename snapshot file error: %v", err)
+					}
+					s.kv = dstorage.New(fn)
+					if err := s.kv.Restore(); err != nil {
+						plog.Panicf("restore KV error: %v", err)
+					}
+				}
 				if err := s.store.Recovery(apply.snapshot.Data); err != nil {
 					plog.Panicf("recovery store error: %v", err)
 				}

etcdserver/server_test.go

@@ -836,6 +836,7 @@ func TestRecvSnapshot(t *testing.T) {
 	cl := newCluster("abc")
 	cl.SetStore(store.New())
 	s := &EtcdServer{
+		cfg: &ServerConfig{},
 		r: raftNode{
 			Node:        n,
 			transport:   &nopTransporter{},
@@ -871,6 +872,7 @@ func TestApplySnapshotAndCommittedEntries(t *testing.T) {
 	cl.SetStore(store.New())
 	storage := newRaftStorage()
 	s := &EtcdServer{
+		cfg: &ServerConfig{},
 		r: raftNode{
 			Node:        n,
 			storage:     &storageRecorder{},