1 files changed, 36 insertions, 28 deletions

diff --git a/weed/server/raft_server.go b/weed/server/raft_server.go
index 2cc8252b8..958680d2b 100644
--- a/weed/server/raft_server.go
+++ b/weed/server/raft_server.go
@@ -3,36 +3,37 @@ package weed_server
 import (
 	"encoding/json"
 	"io/ioutil"
-	"math/rand"
 	"os"
 	"path"
 	"reflect"
 	"sort"
-	"strings"
 	"time"
 
+	"google.golang.org/grpc"
+
+	"github.com/chrislusf/seaweedfs/weed/pb"
+
 	"github.com/chrislusf/raft"
+
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/topology"
-	"github.com/gorilla/mux"
 )
 
 type RaftServer struct {
 	peers      []string // initial peers to join with
 	raftServer raft.Server
 	dataDir    string
-	httpAddr   string
-	router     *mux.Router
+	serverAddr string
 	topo       *topology.Topology
+	*raft.GrpcServer
 }
 
-func NewRaftServer(r *mux.Router, peers []string, httpAddr string, dataDir string, topo *topology.Topology, pulseSeconds int) *RaftServer {
+func NewRaftServer(grpcDialOption grpc.DialOption, peers []string, serverAddr, dataDir string, topo *topology.Topology, pulseSeconds int) *RaftServer {
 	s := &RaftServer{
-		peers:    peers,
-		httpAddr: httpAddr,
-		dataDir:  dataDir,
-		router:   r,
-		topo:     topo,
+		peers:      peers,
+		serverAddr: serverAddr,
+		dataDir:    dataDir,
+		topo:       topo,
 	}
 
 	if glog.V(4) {
@@ -42,41 +43,40 @@ func NewRaftServer(r *mux.Router, peers []string, httpAddr string, dataDir strin
 	raft.RegisterCommand(&topology.MaxVolumeIdCommand{})
 
 	var err error
-	transporter := raft.NewHTTPTransporter("/cluster", time.Second)
-	transporter.Transport.MaxIdleConnsPerHost = 1024
-	glog.V(0).Infof("Starting RaftServer with %v", httpAddr)
+	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"))
 	// Clear old cluster configurations if peers are changed
-	if oldPeers, changed := isPeersChanged(s.dataDir, httpAddr, s.peers); changed {
+	if oldPeers, changed := isPeersChanged(s.dataDir, serverAddr, s.peers); changed {
 		glog.V(0).Infof("Peers Change: %v => %v", oldPeers, s.peers)
-		os.RemoveAll(path.Join(s.dataDir, "conf"))
-		os.RemoveAll(path.Join(s.dataDir, "log"))
-		os.RemoveAll(path.Join(s.dataDir, "snapshot"))
 	}
 
-	s.raftServer, err = raft.NewServer(s.httpAddr, s.dataDir, transporter, nil, topo, "")
+	s.raftServer, err = raft.NewServer(s.serverAddr, s.dataDir, transporter, nil, topo, "")
 	if err != nil {
 		glog.V(0).Infoln(err)
 		return nil
 	}
-	transporter.Install(s.raftServer, s)
 	s.raftServer.SetHeartbeatInterval(500 * time.Millisecond)
 	s.raftServer.SetElectionTimeout(time.Duration(pulseSeconds) * 500 * time.Millisecond)
 	s.raftServer.Start()
 
-	s.router.HandleFunc("/cluster/status", s.statusHandler).Methods("GET")
-
 	for _, peer := range s.peers {
-		s.raftServer.AddPeer(peer, "http://"+peer)
+		s.raftServer.AddPeer(peer, pb.ServerToGrpcAddress(peer))
 	}
-	time.Sleep(time.Duration(1000+rand.Int31n(3000)) * time.Millisecond)
-	if s.raftServer.IsLogEmpty() {
+
+	s.GrpcServer = raft.NewGrpcServer(s.raftServer)
+
+	if s.raftServer.IsLogEmpty() && isTheFirstOne(serverAddr, s.peers) {
 		// Initialize the server by joining itself.
 		glog.V(0).Infoln("Initializing new cluster")
 
 		_, err := s.raftServer.Do(&raft.DefaultJoinCommand{
 			Name:             s.raftServer.Name(),
-			ConnectionString: "http://" + s.httpAddr,
+			ConnectionString: pb.ServerToGrpcAddress(s.serverAddr),
 		})
 
 		if err != nil {
@@ -94,7 +94,7 @@ func (s *RaftServer) Peers() (members []string) {
 	peers := s.raftServer.Peers()
 
 	for _, p := range peers {
-		members = append(members, strings.TrimPrefix(p.ConnectionString, "http://"))
+		members = append(members, p.Name)
 	}
 
 	return
@@ -113,7 +113,7 @@ func isPeersChanged(dir string, self string, peers []string) (oldPeers []string,
 	}
 
 	for _, p := range conf.Peers {
-		oldPeers = append(oldPeers, strings.TrimPrefix(p.ConnectionString, "http://"))
+		oldPeers = append(oldPeers, p.Name)
 	}
 	oldPeers = append(oldPeers, self)
 
@@ -127,3 +127,11 @@ func isPeersChanged(dir string, self string, peers []string) (oldPeers []string,
 	return oldPeers, !reflect.DeepEqual(peers, oldPeers)
 
 }
+
+func isTheFirstOne(self string, peers []string) bool {
+	sort.Strings(peers)
+	if len(peers) <= 0 {
+		return true
+	}
+	return self == peers[0]
+}