Merge pull request #1 from chrislusf/master

merge from chrislus/seaweedfs

docker/Dockerfile

@@ -15,7 +15,9 @@ EXPOSE 8080
 EXPOSE 18888
 # filer server http port
 EXPOSE 8888
-# master server shared gprc+http port
+# master server shared gprc port
+EXPOSE 19333
+# master server shared http port
 EXPOSE 9333
 # s3 server http port
 EXPOSE 8333

docker/Dockerfile.go_build

@@ -9,7 +9,9 @@ EXPOSE 8080
 EXPOSE 18888
 # filer server http port
 EXPOSE 8888
-# master server shared gprc+http port
+# master server shared gprc port
+EXPOSE 19333
+# master server shared http port
 EXPOSE 9333
 # s3 server http port
 EXPOSE 8333

docker/dev-compose.yml

@@ -7,6 +7,7 @@ services:
       dockerfile: Dockerfile.go_build
     ports:
     - 9333:9333
+    - 19333:19333
     command: "master"
   volume:
     build:

docker/seaweedfs-compose.yml

@@ -5,6 +5,7 @@ services:
     image: chrislusf/seaweedfs # use a remote image
     ports:
     - 9333:9333
+    - 19333:19333
     command: "master"
   volume:
     image: chrislusf/seaweedfs # use a remote image

weed/command/master.go

@@ -13,7 +13,6 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/server"
 	"github.com/chrislusf/seaweedfs/weed/util"
 	"github.com/gorilla/mux"
-	"github.com/soheilhy/cmux"
 	"google.golang.org/grpc/reflection"
 )
 
@@ -32,6 +31,7 @@ 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")
@@ -79,7 +79,7 @@ func runMaster(cmd *Command, args []string) bool {
 
 	glog.V(0).Infoln("Start Seaweed Master", util.VERSION, "at", listeningAddress)
 
-	listener, e := util.NewListener(listeningAddress, 0)
+	masterListener, e := util.NewListener(listeningAddress, 0)
 	if e != nil {
 		glog.Fatalf("Master startup error: %v", e)
 	}
@@ -91,23 +91,28 @@ func runMaster(cmd *Command, args []string) bool {
 		ms.SetRaftServer(raftServer)
 	}()
 
-	// start grpc and http server
-	m := cmux.New(listener)
-
-	grpcL := m.Match(cmux.HTTP2HeaderField("content-type", "application/grpc"))
-	httpL := m.Match(cmux.Any())
-
+	go func() {
+		// starting grpc server
+		grpcPort := *mGrpcPort
+		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)
+		}
 		// Create your protocol servers.
 		grpcS := util.NewGrpcServer()
 		master_pb.RegisterSeaweedServer(grpcS, ms)
 		reflection.Register(grpcS)
 
-	httpS := &http.Server{Handler: r}
-
-	go grpcS.Serve(grpcL)
-	go httpS.Serve(httpL)
+		glog.V(0).Infof("Start Seaweed Master %s grpc server at %s:%d", util.VERSION, *masterBindIp, grpcPort)
+		grpcS.Serve(grpcL)
+	}()
 
-	if err := m.Serve(); err != nil {
+	// start http server
+	httpS := &http.Server{Handler: r}
+	if err := httpS.Serve(masterListener); err != nil {
 		glog.Fatalf("master server failed to serve: %v", err)
 	}
 

weed/command/server.go

@@ -15,7 +15,6 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/server"
 	"github.com/chrislusf/seaweedfs/weed/util"
 	"github.com/gorilla/mux"
-	"github.com/soheilhy/cmux"
 	"google.golang.org/grpc/reflection"
 )
 
@@ -62,6 +61,7 @@ var (
 	serverSecureKey               = cmdServer.Flag.String("secure.secret", "", "secret to encrypt Json Web Token(JWT)")
 	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.")
@@ -179,6 +179,25 @@ func runServer(cmd *Command, args []string) bool {
 			glog.Fatalf("Master startup error: %v", e)
 		}
 
+		go func() {
+			// starting grpc server
+			grpcPort := *masterGrpcPort
+			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)
+			}
+			// Create your protocol servers.
+			grpcS := util.NewGrpcServer()
+			master_pb.RegisterSeaweedServer(grpcS, ms)
+			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() {
 			raftWaitForMaster.Wait()
 			time.Sleep(100 * time.Millisecond)
@@ -190,23 +209,9 @@ func runServer(cmd *Command, args []string) bool {
 
 		raftWaitForMaster.Done()
 
-		// start grpc and http server
-		m := cmux.New(masterListener)
-
-		grpcL := m.Match(cmux.HTTP2HeaderField("content-type", "application/grpc"))
-		httpL := m.Match(cmux.Any())
-
-		// Create your protocol servers.
-		grpcS := util.NewGrpcServer()
-		master_pb.RegisterSeaweedServer(grpcS, ms)
-		reflection.Register(grpcS)
-
+		// start http server
 		httpS := &http.Server{Handler: r}
-
-		go grpcS.Serve(grpcL)
-		go httpS.Serve(httpL)
-
-		if err := m.Serve(); err != nil {
+		if err := httpS.Serve(masterListener); err != nil {
 			glog.Fatalf("master server failed to serve: %v", err)
 		}
 

weed/operation/grpc_client.go

@@ -44,9 +44,14 @@ func toVolumeServerGrpcAddress(volumeServer string) (grpcAddress string, err err
 
 func withMasterServerClient(masterServer string, fn func(masterClient master_pb.SeaweedClient) error) error {
 
+	masterGrpcAddress, parseErr := util.ParseServerToGrpcAddress(masterServer, 0)
+	if parseErr != nil {
+		return fmt.Errorf("failed to parse master grpc %v", masterServer)
+	}
+
 	return util.WithCachedGrpcClient(func(grpcConnection *grpc.ClientConn) error {
 		client := master_pb.NewSeaweedClient(grpcConnection)
 		return fn(client)
-	}, masterServer)
+	}, masterGrpcAddress)
 
 }

weed/operation/upload_content.go

@@ -13,7 +13,6 @@ import (
 	"net/textproto"
 	"path/filepath"
 	"strings"
-	"time"
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/security"
@@ -31,10 +30,9 @@ var (
 )
 
 func init() {
-	client = &http.Client{
-		Transport: &http.Transport{MaxIdleConnsPerHost: 1024},
-		Timeout:   5 * time.Second,
-	}
+	client = &http.Client{Transport: &http.Transport{
+		MaxIdleConnsPerHost: 1024,
+	}}
 }
 
 var fileNameEscaper = strings.NewReplacer("\\", "\\\\", "\"", "\\\"")

weed/server/raft_server.go

@@ -42,7 +42,7 @@ func NewRaftServer(r *mux.Router, peers []string, httpAddr string, dataDir strin
 	raft.RegisterCommand(&topology.MaxVolumeIdCommand{})
 
 	var err error
-	transporter := raft.NewHTTPTransporter("/cluster", 0)
+	transporter := raft.NewHTTPTransporter("/cluster", time.Second)
 	transporter.Transport.MaxIdleConnsPerHost = 1024
 	glog.V(0).Infof("Starting RaftServer with %v", httpAddr)
 

weed/server/volume_grpc_client_to_master.go

@@ -27,7 +27,12 @@ func (vs *VolumeServer) heartbeat() {
 			if newLeader != "" {
 				master = newLeader
 			}
-			newLeader, err = vs.doHeartbeat(master, time.Duration(vs.pulseSeconds)*time.Second)
+			masterGrpcAddress, parseErr := util.ParseServerToGrpcAddress(master, 0)
+			if parseErr != nil {
+				glog.V(0).Infof("failed to parse master grpc %v", masterGrpcAddress)
+				continue
+			}
+			newLeader, err = vs.doHeartbeat(master, masterGrpcAddress, time.Duration(vs.pulseSeconds)*time.Second)
 			if err != nil {
 				glog.V(0).Infof("heartbeat error: %v", err)
 				time.Sleep(time.Duration(vs.pulseSeconds) * time.Second)
@@ -36,9 +41,9 @@ func (vs *VolumeServer) heartbeat() {
 	}
 }
 
-func (vs *VolumeServer) doHeartbeat(masterNode string, sleepInterval time.Duration) (newLeader string, err error) {
+func (vs *VolumeServer) doHeartbeat(masterNode, masterGrpcAddress string, sleepInterval time.Duration) (newLeader string, err error) {
 
-	grpcConection, err := util.GrpcDial(masterNode)
+	grpcConection, err := util.GrpcDial(masterGrpcAddress)
 	if err != nil {
 		return "", fmt.Errorf("fail to dial %s : %v", masterNode, err)
 	}

weed/util/constants.go

@@ -1,5 +1,5 @@
 package util
 
 const (
-	VERSION = "1.20"
+	VERSION = "1.23"
 )

weed/util/grpc_client_server.go

@@ -2,6 +2,8 @@ package util
 
 import (
 	"fmt"
+	"strconv"
+	"strings"
 	"sync"
 	"time"
 
@@ -25,8 +27,8 @@ func NewGrpcServer() *grpc.Server {
 }
 
 func GrpcDial(address string, opts ...grpc.DialOption) (*grpc.ClientConn, error) {
-	opts = append(opts, grpc.WithBlock())
-	opts = append(opts, grpc.WithTimeout(time.Duration(5*time.Second)))
+	// opts = append(opts, grpc.WithBlock())
+	// opts = append(opts, grpc.WithTimeout(time.Duration(5*time.Second)))
 	opts = append(opts, grpc.WithInsecure())
 	opts = append(opts, grpc.WithKeepaliveParams(keepalive.ClientParameters{
 		Time:    30 * time.Second, // client ping server if no activity for this long
@@ -64,3 +66,22 @@ func WithCachedGrpcClient(fn func(*grpc.ClientConn) error, address string, opts
 
 	return err
 }
+
+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)
+	}
+
+	filerPort, parseErr := strconv.ParseUint(hostnameAndPort[1], 10, 64)
+	if parseErr != nil {
+		return "", fmt.Errorf("The server port parse error: %v", parseErr)
+	}
+
+	filerGrpcPort := int(filerPort) + 10000
+	if optionalGrpcPort != 0 {
+		filerGrpcPort = optionalGrpcPort
+	}
+
+	return fmt.Sprintf("%s:%d", hostnameAndPort[0], filerGrpcPort), nil
+}

weed/wdclient/masterclient.go

@@ -49,8 +49,9 @@ func (mc *MasterClient) KeepConnectedToMaster() {
 
 func (mc *MasterClient) tryAllMasters() {
 	for _, master := range mc.masters {
-		glog.V(0).Infof("Connecting to %v", master)
-		withMasterClient(master, func(client master_pb.SeaweedClient) error {
+		glog.V(0).Infof("Connecting to master %v", master)
+		gprcErr := withMasterClient(master, func(client master_pb.SeaweedClient) error {
+
 			stream, err := client.KeepConnected(context.Background())
 			if err != nil {
 				glog.V(0).Infof("failed to keep connected to %s: %v", master, err)
@@ -88,13 +89,22 @@ func (mc *MasterClient) tryAllMasters() {
 
 		})
 
+		if gprcErr != nil {
+			glog.V(0).Infof("%s failed to connect with master %v: %v", mc.name, master, gprcErr)
+		}
+
 		mc.currentMaster = ""
 	}
 }
 
 func withMasterClient(master string, fn func(client master_pb.SeaweedClient) error) error {
 
-	grpcConnection, err := util.GrpcDial(master)
+	masterGrpcAddress, parseErr := util.ParseServerToGrpcAddress(master, 0)
+	if parseErr != nil {
+		return fmt.Errorf("failed to parse master grpc %v", master)
+	}
+
+	grpcConnection, err := util.GrpcDial(masterGrpcAddress)
 	if err != nil {
 		return fmt.Errorf("fail to dial %s: %v", master, err)
 	}