1 files changed, 35 insertions, 28 deletions

diff --git a/weed/wdclient/masterclient.go b/weed/wdclient/masterclient.go
index b26853945..e16356e9d 100644
--- a/weed/wdclient/masterclient.go
+++ b/weed/wdclient/masterclient.go
@@ -3,29 +3,32 @@ package wdclient
 import (
 	"context"
 	"fmt"
+	"math/rand"
 	"time"
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
 	"github.com/chrislusf/seaweedfs/weed/util"
-	"math/rand"
+	"google.golang.org/grpc"
 )
 
 type MasterClient struct {
-	ctx           context.Context
-	name          string
-	currentMaster string
-	masters       []string
+	ctx            context.Context
+	name           string
+	currentMaster  string
+	masters        []string
+	grpcDialOption grpc.DialOption
 
 	vidMap
 }
 
-func NewMasterClient(ctx context.Context, clientName string, masters []string) *MasterClient {
+func NewMasterClient(ctx context.Context, grpcDialOption grpc.DialOption, clientName string, masters []string) *MasterClient {
 	return &MasterClient{
-		ctx:     ctx,
-		name:    clientName,
-		masters: masters,
-		vidMap:  newVidMap(),
+		ctx:            ctx,
+		name:           clientName,
+		masters:        masters,
+		grpcDialOption: grpcDialOption,
+		vidMap:         newVidMap(),
 	}
 }
 
@@ -40,7 +43,7 @@ func (mc *MasterClient) WaitUntilConnected() {
 }
 
 func (mc *MasterClient) KeepConnectedToMaster() {
-	glog.V(0).Infof("%s bootstraps with masters %v", mc.name, mc.masters)
+	glog.V(1).Infof("%s bootstraps with masters %v", mc.name, mc.masters)
 	for {
 		mc.tryAllMasters()
 		time.Sleep(time.Second)
@@ -49,28 +52,28 @@ func (mc *MasterClient) KeepConnectedToMaster() {
 
 func (mc *MasterClient) tryAllMasters() {
 	for _, master := range mc.masters {
-		glog.V(0).Infof("Connecting to master %v", master)
-		gprcErr := withMasterClient(master, func(client master_pb.SeaweedClient) error {
+		glog.V(1).Infof("%s Connecting to master %v", mc.name, master)
+		gprcErr := withMasterClient(context.Background(), master, mc.grpcDialOption, func(ctx context.Context, client master_pb.SeaweedClient) error {
 
-			stream, err := client.KeepConnected(context.Background())
+			stream, err := client.KeepConnected(ctx)
 			if err != nil {
-				glog.V(0).Infof("failed to keep connected to %s: %v", master, err)
+				glog.V(0).Infof("%s failed to keep connected to %s: %v", mc.name, master, err)
 				return err
 			}
 
 			if err = stream.Send(&master_pb.ClientListenRequest{Name: mc.name}); err != nil {
-				glog.V(0).Infof("failed to send to %s: %v", master, err)
+				glog.V(0).Infof("%s failed to send to %s: %v", mc.name, master, err)
 				return err
 			}
 
 			if mc.currentMaster == "" {
-				glog.V(0).Infof("Connected to %v", master)
+				glog.V(1).Infof("%s Connected to %v", mc.name, master)
 				mc.currentMaster = master
 			}
 
 			for {
 				if volumeLocation, err := stream.Recv(); err != nil {
-					glog.V(0).Infof("failed to receive from %s: %v", master, err)
+					glog.V(0).Infof("%s failed to receive from %s: %v", mc.name, master, err)
 					return err
 				} else {
 					loc := Location{
@@ -78,9 +81,11 @@ func (mc *MasterClient) tryAllMasters() {
 						PublicUrl: volumeLocation.PublicUrl,
 					}
 					for _, newVid := range volumeLocation.NewVids {
+						glog.V(1).Infof("%s: %s adds volume %d", mc.name, loc.Url, newVid)
 						mc.addLocation(newVid, loc)
 					}
 					for _, deletedVid := range volumeLocation.DeletedVids {
+						glog.V(1).Infof("%s: %s removes volume %d", mc.name, loc.Url, deletedVid)
 						mc.deleteLocation(deletedVid, loc)
 					}
 				}
@@ -96,20 +101,22 @@ func (mc *MasterClient) tryAllMasters() {
 	}
 }
 
-func withMasterClient(master string, fn func(client master_pb.SeaweedClient) error) error {
+func withMasterClient(ctx context.Context, master string, grpcDialOption grpc.DialOption, fn func(ctx context.Context, client master_pb.SeaweedClient) error) error {
 
-	masterGrpcAddress, parseErr := util.ParseServerToGrpcAddress(master, 0)
+	masterGrpcAddress, parseErr := util.ParseServerToGrpcAddress(master)
 	if parseErr != nil {
-		return fmt.Errorf("failed to parse master grpc %v", master)
+		return fmt.Errorf("failed to parse master grpc %v: %v", master, parseErr)
 	}
 
-	grpcConnection, err := util.GrpcDial(masterGrpcAddress)
-	if err != nil {
-		return fmt.Errorf("fail to dial %s: %v", master, err)
-	}
-	defer grpcConnection.Close()
+	return util.WithCachedGrpcClient(ctx, func(grpcConnection *grpc.ClientConn) error {
+		client := master_pb.NewSeaweedClient(grpcConnection)
+		return fn(ctx, client)
+	}, masterGrpcAddress, grpcDialOption)
 
-	client := master_pb.NewSeaweedClient(grpcConnection)
+}
 
-	return fn(client)
+func (mc *MasterClient) WithClient(ctx context.Context, fn func(client master_pb.SeaweedClient) error) error {
+	return withMasterClient(ctx, mc.currentMaster, mc.grpcDialOption, func(ctx context.Context, client master_pb.SeaweedClient) error {
+		return fn(client)
+	})
 }