1 files changed, 32 insertions, 44 deletions

diff --git a/weed/server/raft_server.go b/weed/server/raft_server.go
index 568bfc7b5..d559cb691 100644
--- a/weed/server/raft_server.go
+++ b/weed/server/raft_server.go
@@ -5,8 +5,6 @@ import (
 	"math/rand"
 	"os"
 	"path"
-	"sort"
-	"strings"
 	"time"
 
 	"google.golang.org/grpc"
@@ -19,8 +17,19 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/topology"
 )
 
+type RaftServerOption struct {
+	GrpcDialOption    grpc.DialOption
+	Peers             map[string]pb.ServerAddress
+	ServerAddr        pb.ServerAddress
+	DataDir           string
+	Topo              *topology.Topology
+	RaftResumeState   bool
+	HeartbeatInterval time.Duration
+	ElectionTimeout   time.Duration
+}
+
 type RaftServer struct {
-	peers      []pb.ServerAddress // initial peers to join with
+	peers      map[string]pb.ServerAddress // initial peers to join with
 	raftServer raft.Server
 	dataDir    string
 	serverAddr pb.ServerAddress
@@ -52,12 +61,12 @@ func (s StateMachine) Recovery(data []byte) error {
 	return nil
 }
 
-func NewRaftServer(grpcDialOption grpc.DialOption, peers []pb.ServerAddress, serverAddr pb.ServerAddress, dataDir string, topo *topology.Topology, raftResumeState bool) (*RaftServer, error) {
+func NewRaftServer(option *RaftServerOption) (*RaftServer, error) {
 	s := &RaftServer{
-		peers:      peers,
-		serverAddr: serverAddr,
-		dataDir:    dataDir,
-		topo:       topo,
+		peers:      option.Peers,
+		serverAddr: option.ServerAddr,
+		dataDir:    option.DataDir,
+		topo:       option.Topo,
 	}
 
 	if glog.V(4) {
@@ -67,27 +76,29 @@ func NewRaftServer(grpcDialOption grpc.DialOption, peers []pb.ServerAddress, ser
 	raft.RegisterCommand(&topology.MaxVolumeIdCommand{})
 
 	var err error
-	transporter := raft.NewGrpcTransporter(grpcDialOption)
-	glog.V(0).Infof("Starting RaftServer with %v", serverAddr)
+	transporter := raft.NewGrpcTransporter(option.GrpcDialOption)
+	glog.V(0).Infof("Starting RaftServer with %v", option.ServerAddr)
 
-	if !raftResumeState {
+	// always clear previous log to avoid server is promotable
+	os.RemoveAll(path.Join(s.dataDir, "log"))
+	if !option.RaftResumeState {
 		// 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 {
+	if err := os.MkdirAll(path.Join(s.dataDir, "snapshot"), 0700); err != nil {
 		return nil, err
 	}
 
-	stateMachine := StateMachine{topo: topo}
-	s.raftServer, err = raft.NewServer(string(s.serverAddr), s.dataDir, transporter, stateMachine, topo, "")
+	stateMachine := StateMachine{topo: option.Topo}
+	s.raftServer, err = raft.NewServer(string(s.serverAddr), s.dataDir, transporter, stateMachine, option.Topo, "")
 	if err != nil {
 		glog.V(0).Infoln(err)
 		return nil, err
 	}
-	s.raftServer.SetHeartbeatInterval(time.Duration(300+rand.Intn(150)) * time.Millisecond)
-	s.raftServer.SetElectionTimeout(10 * time.Second)
+	heartbeatInterval := time.Duration(float64(option.HeartbeatInterval) * (rand.Float64()*0.25 + 1))
+	s.raftServer.SetHeartbeatInterval(heartbeatInterval)
+	s.raftServer.SetElectionTimeout(option.ElectionTimeout)
 	if err := s.raftServer.LoadSnapshot(); err != nil {
 		return nil, err
 	}
@@ -95,39 +106,26 @@ func NewRaftServer(grpcDialOption grpc.DialOption, peers []pb.ServerAddress, ser
 		return nil, err
 	}
 
-	for _, peer := range s.peers {
-		if err := s.raftServer.AddPeer(string(peer), peer.ToGrpcAddress()); err != nil {
+	for name, peer := range s.peers {
+		if err := s.raftServer.AddPeer(name, peer.ToGrpcAddress()); err != nil {
 			return nil, err
 		}
 	}
 
 	// Remove deleted peers
 	for existsPeerName := range s.raftServer.Peers() {
-		exists := false
-		var existingPeer pb.ServerAddress
-		for _, peer := range s.peers {
-			if peer.ToGrpcAddress() == existsPeerName {
-				exists, existingPeer = true, peer
-				break
-			}
-		}
-		if exists {
+		if existingPeer, found := s.peers[existsPeerName]; !found {
 			if err := s.raftServer.RemovePeer(existsPeerName); err != nil {
 				glog.V(0).Infoln(err)
 				return nil, err
 			} else {
-				glog.V(0).Infof("removing old peer %s", existingPeer)
+				glog.V(0).Infof("removing old peer: %s", existingPeer)
 			}
 		}
 	}
 
 	s.GrpcServer = raft.NewGrpcServer(s.raftServer)
 
-	if s.raftServer.IsLogEmpty() && isTheFirstOne(serverAddr, s.peers) {
-		// Initialize the server by joining itself.
-		// s.DoJoinCommand()
-	}
-
 	glog.V(0).Infof("current cluster leader: %v", s.raftServer.Leader())
 
 	return s, nil
@@ -143,16 +141,6 @@ func (s *RaftServer) Peers() (members []string) {
 	return
 }
 
-func isTheFirstOne(self pb.ServerAddress, peers []pb.ServerAddress) bool {
-	sort.Slice(peers, func(i, j int) bool {
-		return strings.Compare(string(peers[i]), string(peers[j])) < 0
-	})
-	if len(peers) <= 0 {
-		return true
-	}
-	return self == peers[0]
-}
-
 func (s *RaftServer) DoJoinCommand() {
 
 	glog.V(0).Infoln("Initializing new cluster")