diff options
Diffstat (limited to 'weed/wdclient/masterclient.go')
| -rw-r--r-- | weed/wdclient/masterclient.go | 140 |
1 files changed, 77 insertions, 63 deletions
diff --git a/weed/wdclient/masterclient.go b/weed/wdclient/masterclient.go index f58c28504..4c066d535 100644 --- a/weed/wdclient/masterclient.go +++ b/weed/wdclient/masterclient.go @@ -2,30 +2,35 @@ package wdclient import ( "context" - "fmt" + "math/rand" "time" + "google.golang.org/grpc" + "github.com/chrislusf/seaweedfs/weed/glog" + "github.com/chrislusf/seaweedfs/weed/pb" "github.com/chrislusf/seaweedfs/weed/pb/master_pb" - "github.com/chrislusf/seaweedfs/weed/util" - "math/rand" ) type MasterClient struct { - ctx context.Context - name string - currentMaster string - masters []string + clientType string + clientHost string + grpcPort uint32 + currentMaster string + masters []string + grpcDialOption grpc.DialOption vidMap } -func NewMasterClient(ctx context.Context, clientName string, masters []string) *MasterClient { +func NewMasterClient(grpcDialOption grpc.DialOption, clientType string, clientHost string, clientGrpcPort uint32, masters []string) *MasterClient { return &MasterClient{ - ctx: ctx, - name: clientName, - masters: masters, - vidMap: newVidMap(), + clientType: clientType, + clientHost: clientHost, + grpcPort: clientGrpcPort, + masters: masters, + grpcDialOption: grpcDialOption, + vidMap: newVidMap(), } } @@ -40,7 +45,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 masterClient bootstraps with masters %v", mc.clientType, mc.masters) for { mc.tryAllMasters() time.Sleep(time.Second) @@ -48,69 +53,78 @@ func (mc *MasterClient) KeepConnectedToMaster() { } func (mc *MasterClient) tryAllMasters() { + nextHintedLeader := "" for _, master := range mc.masters { - glog.V(0).Infof("Connecting to master %v", master) - gprcErr := withMasterClient(master, func(client master_pb.SeaweedClient) error { - stream, err := client.KeepConnected(context.Background()) + nextHintedLeader = mc.tryConnectToMaster(master) + for nextHintedLeader != "" { + nextHintedLeader = mc.tryConnectToMaster(nextHintedLeader) + } + + mc.currentMaster = "" + mc.vidMap = newVidMap() + } +} + +func (mc *MasterClient) tryConnectToMaster(master string) (nextHintedLeader string) { + glog.V(1).Infof("%s masterClient Connecting to master %v", mc.clientType, master) + gprcErr := pb.WithMasterClient(master, mc.grpcDialOption, func(client master_pb.SeaweedClient) error { + + stream, err := client.KeepConnected(context.Background()) + if err != nil { + glog.V(0).Infof("%s masterClient failed to keep connected to %s: %v", mc.clientType, master, err) + return err + } + + if err = stream.Send(&master_pb.KeepConnectedRequest{Name: mc.clientType, GrpcPort: mc.grpcPort}); err != nil { + glog.V(0).Infof("%s masterClient failed to send to %s: %v", mc.clientType, master, err) + return err + } + + glog.V(1).Infof("%s masterClient Connected to %v", mc.clientType, master) + mc.currentMaster = master + + for { + volumeLocation, err := stream.Recv() if err != nil { - glog.V(0).Infof("failed to keep connected to %s: %v", master, err) + glog.V(0).Infof("%s masterClient failed to receive from %s: %v", mc.clientType, 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) - return err + // maybe the leader is changed + if volumeLocation.Leader != "" { + glog.V(0).Infof("redirected to leader %v", volumeLocation.Leader) + nextHintedLeader = volumeLocation.Leader + return nil } - for { - if volumeLocation, err := stream.Recv(); err != nil { - glog.V(0).Infof("failed to receive from %s: %v", master, err) - return err - } else { - loc := Location{ - Url: volumeLocation.Url, - PublicUrl: volumeLocation.PublicUrl, - } - for _, newVid := range volumeLocation.NewVids { - mc.addLocation(newVid, loc) - } - for _, deletedVid := range volumeLocation.DeletedVids { - mc.deleteLocation(deletedVid, loc) - } - - if mc.currentMaster == "" { - glog.V(0).Infof("Connected to %v", master) - mc.currentMaster = master - } - - } + // process new volume location + loc := Location{ + Url: volumeLocation.Url, + PublicUrl: volumeLocation.PublicUrl, + } + for _, newVid := range volumeLocation.NewVids { + glog.V(1).Infof("%s: %s masterClient adds volume %d", mc.clientType, loc.Url, newVid) + mc.addLocation(newVid, loc) + } + for _, deletedVid := range volumeLocation.DeletedVids { + glog.V(1).Infof("%s: %s masterClient removes volume %d", mc.clientType, loc.Url, deletedVid) + mc.deleteLocation(deletedVid, loc) } - - }) - - if gprcErr != nil { - glog.V(0).Infof("%s failed to connect with master %v: %v", mc.name, master, gprcErr) } - mc.currentMaster = "" + }) + if gprcErr != nil { + glog.V(0).Infof("%s masterClient failed to connect with master %v: %v", mc.clientType, master, gprcErr) } + return } -func withMasterClient(master string, fn func(client master_pb.SeaweedClient) error) error { - - 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) +func (mc *MasterClient) WithClient(fn func(client master_pb.SeaweedClient) error) error { + for mc.currentMaster == "" { + time.Sleep(3 * time.Second) } - defer grpcConnection.Close() - - client := master_pb.NewSeaweedClient(grpcConnection) - - return fn(client) + return pb.WithMasterClient(mc.currentMaster, mc.grpcDialOption, func(client master_pb.SeaweedClient) error { + return fn(client) + }) } |