From 8c82fb7e5f0604953d0f8430f3752c0b2bfcada8 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=D0=A3=D1=81=D1=82=D1=8E=D0=B6=D0=B0=D0=BD=D0=B8=D0=BD=20?=
 =?UTF-8?q?=D0=90=D0=BD=D1=82=D0=BE=D0=BD=20=D0=90=D0=BB=D0=B5=D0=BA=D1=81?=
 =?UTF-8?q?=D0=B0=D0=BD=D0=B4=D1=80=D0=BE=D0=B2=D0=B8=D1=87?=
 <ustuzhanin@tochka.com>
Date: Fri, 2 Oct 2020 23:01:20 +0500
Subject: [PATCH] fix: restore raft state

---
 weed/command/master.go     |  8 +++--
 weed/server/raft_server.go | 61 +++++++++++++++++++++++++++++++-------
 2 files changed, 55 insertions(+), 14 deletions(-)

diff --git a/weed/command/master.go b/weed/command/master.go
index 144962f63..3cf8cc89b 100644
--- a/weed/command/master.go
+++ b/weed/command/master.go
@@ -41,6 +41,7 @@ type MasterOptions struct {
 	disableHttp        *bool
 	metricsAddress     *string
 	metricsIntervalSec *int
+	removeRaftState    *bool
 }
 
 func init() {
@@ -59,6 +60,7 @@ func init() {
 	m.disableHttp = cmdMaster.Flag.Bool("disableHttp", false, "disable http requests, only gRPC operations are allowed.")
 	m.metricsAddress = cmdMaster.Flag.String("metrics.address", "", "Prometheus gateway address <host>:<port>")
 	m.metricsIntervalSec = cmdMaster.Flag.Int("metrics.intervalSeconds", 15, "Prometheus push interval in seconds")
+	m.removeRaftState = cmdMaster.Flag.Bool("raft.removeState", true, "remove raft state on start master server")
 }
 
 var cmdMaster = &Command{
@@ -118,10 +120,10 @@ func startMaster(masterOption MasterOptions, masterWhiteList []string) {
 		glog.Fatalf("Master startup error: %v", e)
 	}
 	// start raftServer
-	raftServer := weed_server.NewRaftServer(security.LoadClientTLS(util.GetViper(), "grpc.master"),
-		peers, myMasterAddress, util.ResolvePath(*masterOption.metaFolder), ms.Topo, 5)
+	raftServer, err := weed_server.NewRaftServer(security.LoadClientTLS(util.GetViper(), "grpc.master"),
+		peers, myMasterAddress, util.ResolvePath(*masterOption.metaFolder), ms.Topo, 5, *masterOption.removeRaftState)
 	if raftServer == nil {
-		glog.Fatalf("please verify %s is writable, see https://github.com/chrislusf/seaweedfs/issues/717", *masterOption.metaFolder)
+		glog.Fatalf("please verify %s is writable, see https://github.com/chrislusf/seaweedfs/issues/717: %s", *masterOption.metaFolder, err)
 	}
 	ms.SetRaftServer(raftServer)
 	r.HandleFunc("/cluster/status", raftServer.StatusHandler).Methods("GET")
diff --git a/weed/server/raft_server.go b/weed/server/raft_server.go
index 958680d2b..0679215a1 100644
--- a/weed/server/raft_server.go
+++ b/weed/server/raft_server.go
@@ -28,7 +28,31 @@ type RaftServer struct {
 	*raft.GrpcServer
 }
 
-func NewRaftServer(grpcDialOption grpc.DialOption, peers []string, serverAddr, dataDir string, topo *topology.Topology, pulseSeconds int) *RaftServer {
+type StateMachine struct {
+	raft.StateMachine
+	topo *topology.Topology
+}
+
+func (s StateMachine) Save() ([]byte, error) {
+	state := topology.MaxVolumeIdCommand{
+		MaxVolumeId: s.topo.GetMaxVolumeId(),
+	}
+	glog.V(1).Infof("Save raft state %+v", state)
+	return json.Marshal(state)
+}
+
+func (s StateMachine) Recovery(data []byte) error {
+	state := topology.MaxVolumeIdCommand{}
+	err := json.Unmarshal(data, &state)
+	if err != nil {
+		return err
+	}
+	glog.V(1).Infof("Recovery raft state %+v", state)
+	s.topo.UpAdjustMaxVolumeId(state.MaxVolumeId)
+	return nil
+}
+
+func NewRaftServer(grpcDialOption grpc.DialOption, peers []string, serverAddr, dataDir string, topo *topology.Topology, pulseSeconds int, cleanState bool) (*RaftServer, error) {
 	s := &RaftServer{
 		peers:      peers,
 		serverAddr: serverAddr,
@@ -46,26 +70,41 @@ func NewRaftServer(grpcDialOption grpc.DialOption, peers []string, serverAddr, d
 	transporter := raft.NewGrpcTransporter(grpcDialOption)
 	glog.V(0).Infof("Starting RaftServer with %v", serverAddr)
 
-	// always clear previous metadata
-	os.RemoveAll(path.Join(s.dataDir, "conf"))
-	os.RemoveAll(path.Join(s.dataDir, "log"))
-	os.RemoveAll(path.Join(s.dataDir, "snapshot"))
+	if cleanState {
+		// always clear previous metadata
+		os.RemoveAll(path.Join(s.dataDir, "conf"))
+		os.RemoveAll(path.Join(s.dataDir, "log"))
+		os.RemoveAll(path.Join(s.dataDir, "snapshot"))
+	}
+	if err := os.MkdirAll(path.Join(s.dataDir, "snapshot"), 0600); err != nil {
+		return nil, err
+	}
+
 	// Clear old cluster configurations if peers are changed
 	if oldPeers, changed := isPeersChanged(s.dataDir, serverAddr, s.peers); changed {
 		glog.V(0).Infof("Peers Change: %v => %v", oldPeers, s.peers)
 	}
 
-	s.raftServer, err = raft.NewServer(s.serverAddr, s.dataDir, transporter, nil, topo, "")
+	stateMachine := StateMachine{topo: topo}
+	s.raftServer, err = raft.NewServer(s.serverAddr, s.dataDir, transporter, stateMachine, topo, "")
 	if err != nil {
 		glog.V(0).Infoln(err)
-		return nil
+		return nil, err
 	}
 	s.raftServer.SetHeartbeatInterval(500 * time.Millisecond)
 	s.raftServer.SetElectionTimeout(time.Duration(pulseSeconds) * 500 * time.Millisecond)
-	s.raftServer.Start()
+	if err := s.raftServer.LoadSnapshot(); err != nil {
+		return nil, err
+	}
+	if err := s.raftServer.Start(); err != nil {
+		return nil, err
+	}
 
 	for _, peer := range s.peers {
-		s.raftServer.AddPeer(peer, pb.ServerToGrpcAddress(peer))
+		if err := s.raftServer.AddPeer(peer, pb.ServerToGrpcAddress(peer)); err != nil {
+			return nil, err
+		}
+
 	}
 
 	s.GrpcServer = raft.NewGrpcServer(s.raftServer)
@@ -81,13 +120,13 @@ func NewRaftServer(grpcDialOption grpc.DialOption, peers []string, serverAddr, d
 
 		if err != nil {
 			glog.V(0).Infoln(err)
-			return nil
+			return nil, err
 		}
 	}
 
 	glog.V(0).Infof("current cluster leader: %v", s.raftServer.Leader())
 
-	return s
+	return s, nil
 }
 
 func (s *RaftServer) Peers() (members []string) {