raft change from http to grpc

master grpc port is fixed to http port + 10000

1 files changed, 20 insertions, 25 deletions

diff --git a/weed/server/raft_server.go b/weed/server/raft_server.go
index 7afef0b15..4be13810f 100644
--- a/weed/server/raft_server.go
+++ b/weed/server/raft_server.go
@@ -2,36 +2,35 @@ package weed_server
 
 import (
 	"encoding/json"
+	"github.com/chrislusf/seaweedfs/weed/util"
+	"google.golang.org/grpc"
 	"io/ioutil"
 	"os"
 	"path"
 	"reflect"
 	"sort"
-	"strings"
 	"time"
 
 	"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 string, 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) {
@@ -41,43 +40,39 @@ 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
-	transporter.Transport.IdleConnTimeout = time.Second
-	transporter.Transport.ResponseHeaderTimeout = time.Second
-	glog.V(0).Infof("Starting RaftServer with %v", httpAddr)
+	transporter := raft.NewGrpcTransporter(grpcDialOption)
+	glog.V(0).Infof("Starting RaftServer with %v", serverAddr)
 
 	// 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, util.ServerToGrpcAddress(peer, 19333))
 	}
 
-	if s.raftServer.IsLogEmpty() && isTheFirstOne(httpAddr, s.peers) {
+	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: util.ServerToGrpcAddress(s.serverAddr, 19333),
 		})
 
 		if err != nil {
@@ -95,7 +90,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
@@ -114,7 +109,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)