raft change from http to grpc

master grpc port is fixed to http port + 10000

weed/command/master.go

@@ -1,21 +1,20 @@
 package command
 
 import (
+	"github.com/chrislusf/raft/protobuf"
+	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
 	"github.com/chrislusf/seaweedfs/weed/security"
+	"github.com/chrislusf/seaweedfs/weed/server"
+	"github.com/chrislusf/seaweedfs/weed/util"
+	"github.com/gorilla/mux"
 	"github.com/spf13/viper"
+	"google.golang.org/grpc/reflection"
 	"net/http"
 	"os"
 	"runtime"
 	"strconv"
 	"strings"
-	"time"
-
-	"github.com/chrislusf/seaweedfs/weed/glog"
-	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
-	"github.com/chrislusf/seaweedfs/weed/server"
-	"github.com/chrislusf/seaweedfs/weed/util"
-	"github.com/gorilla/mux"
-	"google.golang.org/grpc/reflection"
 )
 
 func init() {
@@ -36,7 +35,6 @@ var cmdMaster = &Command{
 
 var (
 	mport                   = cmdMaster.Flag.Int("port", 9333, "http listen port")
-	mGrpcPort               = cmdMaster.Flag.Int("port.grpc", 0, "grpc server listen port, default to http port + 10000")
 	masterIp                = cmdMaster.Flag.String("ip", "localhost", "master <ip>|<server> address")
 	masterBindIp            = cmdMaster.Flag.String("ip.bind", "0.0.0.0", "ip address to bind to")
 	metaFolder              = cmdMaster.Flag.String("mdir", os.TempDir(), "data directory to store meta data")
@@ -92,18 +90,14 @@ func runMaster(cmd *Command, args []string) bool {
 	}
 
 	go func() {
-		time.Sleep(100 * time.Millisecond)
+		// start raftServer
 		myMasterAddress, peers := checkPeers(*masterIp, *mport, *masterPeers)
-		raftServer := weed_server.NewRaftServer(r, peers, myMasterAddress, *metaFolder, ms.Topo, *mpulse)
+		raftServer := weed_server.NewRaftServer(security.LoadClientTLS(viper.Sub("grpc"), "master"),
+			peers, myMasterAddress, *metaFolder, ms.Topo, *mpulse)
 		ms.SetRaftServer(raftServer)
-	}()
 
-	go func() {
 		// starting grpc server
-		grpcPort := *mGrpcPort
+		grpcPort := *mport + 10000
-		if grpcPort == 0 {
-			grpcPort = *mport + 10000
-		}
 		grpcL, err := util.NewListener(*masterBindIp+":"+strconv.Itoa(grpcPort), 0)
 		if err != nil {
 			glog.Fatalf("master failed to listen on grpc port %d: %v", grpcPort, err)
@@ -111,6 +105,7 @@ func runMaster(cmd *Command, args []string) bool {
 		// Create your protocol servers.
 		grpcS := util.NewGrpcServer(security.LoadServerTLS(viper.Sub("grpc"), "master"))
 		master_pb.RegisterSeaweedServer(grpcS, ms)
+		protobuf.RegisterRaftServer(grpcS, raftServer)
 		reflection.Register(grpcS)
 
 		glog.V(0).Infof("Start Seaweed Master %s grpc server at %s:%d", util.VERSION, *masterBindIp, grpcPort)

weed/command/server.go

@@ -1,6 +1,7 @@
 package command
 
 import (
+	"github.com/chrislusf/raft/protobuf"
 	"github.com/chrislusf/seaweedfs/weed/security"
 	"github.com/spf13/viper"
 	"net/http"
@@ -62,7 +63,6 @@ var (
 	serverPeers                   = cmdServer.Flag.String("master.peers", "", "all master nodes in comma separated ip:masterPort list")
 	serverGarbageThreshold        = cmdServer.Flag.Float64("garbageThreshold", 0.3, "threshold to vacuum and reclaim spaces")
 	masterPort                    = cmdServer.Flag.Int("master.port", 9333, "master server http listen port")
-	masterGrpcPort                = cmdServer.Flag.Int("master.port.grpc", 0, "master grpc server listen port, default to http port + 10000")
 	masterMetaFolder              = cmdServer.Flag.String("master.dir", "", "data directory to store meta data, default to same as -dir specified")
 	masterVolumeSizeLimitMB       = cmdServer.Flag.Uint("master.volumeSizeLimitMB", 30*1000, "Master stops directing writes to oversized volumes.")
 	masterVolumePreallocate       = cmdServer.Flag.Bool("master.volumePreallocate", false, "Preallocate disk space for volumes.")
@@ -162,10 +162,8 @@ func runServer(cmd *Command, args []string) bool {
 		}()
 	}
 
-	var raftWaitForMaster sync.WaitGroup
 	var volumeWait sync.WaitGroup
 
-	raftWaitForMaster.Add(1)
 	volumeWait.Add(1)
 
 	go func() {
@@ -183,11 +181,14 @@ func runServer(cmd *Command, args []string) bool {
 		}
 
 		go func() {
+			// start raftServer
+			myMasterAddress, peers := checkPeers(*masterIp, *mport, *masterPeers)
+			raftServer := weed_server.NewRaftServer(security.LoadClientTLS(viper.Sub("grpc"), "master"),
+				peers, myMasterAddress, *metaFolder, ms.Topo, *mpulse)
+			ms.SetRaftServer(raftServer)
+
 			// starting grpc server
-			grpcPort := *masterGrpcPort
+			grpcPort := *masterPort + 10000
-			if grpcPort == 0 {
-				grpcPort = *masterPort + 10000
-			}
 			grpcL, err := util.NewListener(*serverIp+":"+strconv.Itoa(grpcPort), 0)
 			if err != nil {
 				glog.Fatalf("master failed to listen on grpc port %d: %v", grpcPort, err)
@@ -196,22 +197,14 @@ func runServer(cmd *Command, args []string) bool {
 			glog.V(0).Infof("grpc config %+v", viper.Sub("grpc"))
 			grpcS := util.NewGrpcServer(security.LoadServerTLS(viper.Sub("grpc"), "master"))
 			master_pb.RegisterSeaweedServer(grpcS, ms)
+			protobuf.RegisterRaftServer(grpcS, raftServer)
 			reflection.Register(grpcS)
 
 			glog.V(0).Infof("Start Seaweed Master %s grpc server at %s:%d", util.VERSION, *serverIp, grpcPort)
 			grpcS.Serve(grpcL)
 		}()
 
-		go func() {
+		volumeWait.Done()
-			raftWaitForMaster.Wait()
-			time.Sleep(100 * time.Millisecond)
-			myAddress, peers := checkPeers(*serverIp, *masterPort, *serverPeers)
-			raftServer := weed_server.NewRaftServer(r, peers, myAddress, *masterMetaFolder, ms.Topo, *pulseSeconds)
-			ms.SetRaftServer(raftServer)
-			volumeWait.Done()
-		}()
-
-		raftWaitForMaster.Done()
 
 		// start http server
 		httpS := &http.Server{Handler: r}

weed/server/master_ui/templates.go

@@ -41,7 +41,7 @@ var StatusTpl = template.Must(template.New("status").Parse(`<!DOCTYPE html>
                 <td class="col-sm-2 field-label"><label>Other Masters:</label></td>
                 <td class="col-sm-10"><ul class="list-unstyled">
                 {{ range $k, $p := .Peers }}
-                  <li><a href="{{ $p.ConnectionString }}">{{ $p.Name }}</a></li>
+                  <li><a href="http://{{ $p.Name }}">{{ $p.Name }}</a></li>
                 {{ end }}
                 </ul></td>
               </tr>

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
+	serverAddr string
-	router     *mux.Router
 	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,
+		peers:      peers,
-		httpAddr: httpAddr,
+		serverAddr: serverAddr,
-		dataDir:  dataDir,
+		dataDir:    dataDir,
-		router:   r,
+		topo:       topo,
-		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 := raft.NewGrpcTransporter(grpcDialOption)
-	transporter.Transport.MaxIdleConnsPerHost = 1024
+	glog.V(0).Infof("Starting RaftServer with %v", serverAddr)
-	transporter.Transport.IdleConnTimeout = time.Second
-	transporter.Transport.ResponseHeaderTimeout = time.Second
-	glog.V(0).Infof("Starting RaftServer with %v", httpAddr)
 
 	// 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)
 

weed/server/raft_server_handlers.go

@@ -1,21 +0,0 @@
-package weed_server
-
-import (
-	"github.com/chrislusf/seaweedfs/weed/operation"
-	"net/http"
-)
-
-func (s *RaftServer) HandleFunc(pattern string, handler func(http.ResponseWriter, *http.Request)) {
-	s.router.HandleFunc(pattern, handler)
-}
-
-func (s *RaftServer) statusHandler(w http.ResponseWriter, r *http.Request) {
-	ret := operation.ClusterStatusResult{
-		IsLeader: s.topo.IsLeader(),
-		Peers:    s.Peers(),
-	}
-	if leader, e := s.topo.Leader(); e == nil {
-		ret.Leader = leader
-	}
-	writeJsonQuiet(w, r, http.StatusOK, ret)
-}

weed/util/grpc_client_server.go

@@ -83,18 +83,34 @@ func WithCachedGrpcClient(fn func(*grpc.ClientConn) error, address string, opts
 func ParseServerToGrpcAddress(server string, optionalGrpcPort int) (serverGrpcAddress string, err error) {
 	hostnameAndPort := strings.Split(server, ":")
 	if len(hostnameAndPort) != 2 {
-		return "", fmt.Errorf("The server should have hostname:port format: %v", hostnameAndPort)
+		return "", fmt.Errorf("server should have hostname:port format: %v", hostnameAndPort)
 	}
 
-	filerPort, parseErr := strconv.ParseUint(hostnameAndPort[1], 10, 64)
+	port, parseErr := strconv.ParseUint(hostnameAndPort[1], 10, 64)
 	if parseErr != nil {
-		return "", fmt.Errorf("The server port parse error: %v", parseErr)
+		return "", fmt.Errorf("server port parse error: %v", parseErr)
 	}
 
-	filerGrpcPort := int(filerPort) + 10000
+	grpcPort := int(port) + 10000
 	if optionalGrpcPort != 0 {
-		filerGrpcPort = optionalGrpcPort
+		grpcPort = optionalGrpcPort
 	}
 
-	return fmt.Sprintf("%s:%d", hostnameAndPort[0], filerGrpcPort), nil
+	return fmt.Sprintf("%s:%d", hostnameAndPort[0], grpcPort), nil
+}
+
+func ServerToGrpcAddress(server string, defaultGrpcPort int) (serverGrpcAddress string) {
+	hostnameAndPort := strings.Split(server, ":")
+	if len(hostnameAndPort) != 2 {
+		return fmt.Sprintf("%s:%d", server, defaultGrpcPort)
+	}
+
+	port, parseErr := strconv.ParseUint(hostnameAndPort[1], 10, 64)
+	if parseErr != nil {
+		return fmt.Sprintf("%s:%d", hostnameAndPort[0], defaultGrpcPort)
+	}
+
+	grpcPort := int(port) + 10000
+
+	return fmt.Sprintf("%s:%d", hostnameAndPort[0], grpcPort)
 }