change server address from string to a type

1 files changed, 20 insertions, 25 deletions

diff --git a/weed/server/volume_grpc_client_to_master.go b/weed/server/volume_grpc_client_to_master.go
index 770abdab7..2659307fc 100644
--- a/weed/server/volume_grpc_client_to_master.go
+++ b/weed/server/volume_grpc_client_to_master.go
@@ -19,7 +19,7 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/util"
 )
 
-func (vs *VolumeServer) GetMaster() string {
+func (vs *VolumeServer) GetMaster() pb.ServerAddress {
 	return vs.currentMaster
 }
 
@@ -54,7 +54,7 @@ func (vs *VolumeServer) heartbeat() {
 	grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.volume")
 
 	var err error
-	var newLeader string
+	var newLeader pb.ServerAddress
 	for vs.isHeartbeating {
 		for _, master := range vs.SeedMasterNodes {
 			if newLeader != "" {
@@ -63,13 +63,8 @@ func (vs *VolumeServer) heartbeat() {
 				time.Sleep(3 * time.Second)
 				master = newLeader
 			}
-			masterGrpcAddress, parseErr := pb.ParseServerToGrpcAddress(master)
-			if parseErr != nil {
-				glog.V(0).Infof("failed to parse master grpc %v: %v", masterGrpcAddress, parseErr)
-				continue
-			}
 			vs.store.MasterAddress = master
-			newLeader, err = vs.doHeartbeat(master, masterGrpcAddress, grpcDialOption, time.Duration(vs.pulseSeconds)*time.Second)
+			newLeader, err = vs.doHeartbeat(master, grpcDialOption, 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)
@@ -92,25 +87,25 @@ func (vs *VolumeServer) StopHeartbeat() (isAlreadyStopping bool) {
 	return false
 }
 
-func (vs *VolumeServer) doHeartbeat(masterNode, masterGrpcAddress string, grpcDialOption grpc.DialOption, sleepInterval time.Duration) (newLeader string, err error) {
+func (vs *VolumeServer) doHeartbeat(masterAddress pb.ServerAddress, grpcDialOption grpc.DialOption, sleepInterval time.Duration) (newLeader pb.ServerAddress, err error) {
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	grpcConection, err := pb.GrpcDial(ctx, masterGrpcAddress, grpcDialOption)
+	grpcConection, err := pb.GrpcDial(ctx, masterAddress.ToGrpcAddress(), grpcDialOption)
 	if err != nil {
-		return "", fmt.Errorf("fail to dial %s : %v", masterNode, err)
+		return "", fmt.Errorf("fail to dial %s : %v", masterAddress, err)
 	}
 	defer grpcConection.Close()
 
 	client := master_pb.NewSeaweedClient(grpcConection)
 	stream, err := client.SendHeartbeat(ctx)
 	if err != nil {
-		glog.V(0).Infof("SendHeartbeat to %s: %v", masterNode, err)
+		glog.V(0).Infof("SendHeartbeat to %s: %v", masterAddress, err)
 		return "", err
 	}
-	glog.V(0).Infof("Heartbeat to: %v", masterNode)
-	vs.currentMaster = masterNode
+	glog.V(0).Infof("Heartbeat to: %v", masterAddress)
+	vs.currentMaster = masterAddress
 
 	doneChan := make(chan error, 1)
 
@@ -130,9 +125,9 @@ func (vs *VolumeServer) doHeartbeat(masterNode, masterGrpcAddress string, grpcDi
 					}
 				}
 			}
-			if in.GetLeader() != "" && vs.currentMaster != in.GetLeader() {
+			if in.GetLeader() != "" && string(vs.currentMaster) != in.GetLeader() {
 				glog.V(0).Infof("Volume Server found a new master newLeader: %v instead of %v", in.GetLeader(), vs.currentMaster)
-				newLeader = in.GetLeader()
+				newLeader = pb.ServerAddress(in.GetLeader())
 				doneChan <- nil
 				return
 			}
@@ -140,12 +135,12 @@ func (vs *VolumeServer) doHeartbeat(masterNode, masterGrpcAddress string, grpcDi
 	}()
 
 	if err = stream.Send(vs.store.CollectHeartbeat()); err != nil {
-		glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", masterNode, err)
+		glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", masterAddress, err)
 		return "", err
 	}
 
 	if err = stream.Send(vs.store.CollectErasureCodingHeartbeat()); err != nil {
-		glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", masterNode, err)
+		glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", masterAddress, err)
 		return "", err
 	}
 
@@ -162,7 +157,7 @@ func (vs *VolumeServer) doHeartbeat(masterNode, masterGrpcAddress string, grpcDi
 			}
 			glog.V(0).Infof("volume server %s:%d adds volume %d", vs.store.Ip, vs.store.Port, volumeMessage.Id)
 			if err = stream.Send(deltaBeat); err != nil {
-				glog.V(0).Infof("Volume Server Failed to update to master %s: %v", masterNode, err)
+				glog.V(0).Infof("Volume Server Failed to update to master %s: %v", masterAddress, err)
 				return "", err
 			}
 		case ecShardMessage := <-vs.store.NewEcShardsChan:
@@ -174,7 +169,7 @@ func (vs *VolumeServer) doHeartbeat(masterNode, masterGrpcAddress string, grpcDi
 			glog.V(0).Infof("volume server %s:%d adds ec shard %d:%d", vs.store.Ip, vs.store.Port, ecShardMessage.Id,
 				erasure_coding.ShardBits(ecShardMessage.EcIndexBits).ShardIds())
 			if err = stream.Send(deltaBeat); err != nil {
-				glog.V(0).Infof("Volume Server Failed to update to master %s: %v", masterNode, err)
+				glog.V(0).Infof("Volume Server Failed to update to master %s: %v", masterAddress, err)
 				return "", err
 			}
 		case volumeMessage := <-vs.store.DeletedVolumesChan:
@@ -185,7 +180,7 @@ func (vs *VolumeServer) doHeartbeat(masterNode, masterGrpcAddress string, grpcDi
 			}
 			glog.V(0).Infof("volume server %s:%d deletes volume %d", vs.store.Ip, vs.store.Port, volumeMessage.Id)
 			if err = stream.Send(deltaBeat); err != nil {
-				glog.V(0).Infof("Volume Server Failed to update to master %s: %v", masterNode, err)
+				glog.V(0).Infof("Volume Server Failed to update to master %s: %v", masterAddress, err)
 				return "", err
 			}
 		case ecShardMessage := <-vs.store.DeletedEcShardsChan:
@@ -197,20 +192,20 @@ func (vs *VolumeServer) doHeartbeat(masterNode, masterGrpcAddress string, grpcDi
 			glog.V(0).Infof("volume server %s:%d deletes ec shard %d:%d", vs.store.Ip, vs.store.Port, ecShardMessage.Id,
 				erasure_coding.ShardBits(ecShardMessage.EcIndexBits).ShardIds())
 			if err = stream.Send(deltaBeat); err != nil {
-				glog.V(0).Infof("Volume Server Failed to update to master %s: %v", masterNode, err)
+				glog.V(0).Infof("Volume Server Failed to update to master %s: %v", masterAddress, err)
 				return "", err
 			}
 		case <-volumeTickChan:
 			glog.V(4).Infof("volume server %s:%d heartbeat", vs.store.Ip, vs.store.Port)
 			vs.store.MaybeAdjustVolumeMax()
 			if err = stream.Send(vs.store.CollectHeartbeat()); err != nil {
-				glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", masterNode, err)
+				glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", masterAddress, err)
 				return "", err
 			}
 		case <-ecShardTickChan:
 			glog.V(4).Infof("volume server %s:%d ec heartbeat", vs.store.Ip, vs.store.Port)
 			if err = stream.Send(vs.store.CollectErasureCodingHeartbeat()); err != nil {
-				glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", masterNode, err)
+				glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", masterAddress, err)
 				return "", err
 			}
 		case err = <-doneChan:
@@ -229,7 +224,7 @@ func (vs *VolumeServer) doHeartbeat(masterNode, masterGrpcAddress string, grpcDi
 			}
 			glog.V(1).Infof("volume server %s:%d stops and deletes all volumes", vs.store.Ip, vs.store.Port)
 			if err = stream.Send(emptyBeat); err != nil {
-				glog.V(0).Infof("Volume Server Failed to update to master %s: %v", masterNode, err)
+				glog.V(0).Infof("Volume Server Failed to update to master %s: %v", masterAddress, err)
 				return "", err
 			}
 			return