aboutsummaryrefslogtreecommitdiff
path: root/weed/wdclient
diff options
context:
space:
mode:
authorhilimd <68371223+hilimd@users.noreply.github.com>2020-10-10 13:07:57 +0800
committerGitHub <noreply@github.com>2020-10-10 13:07:57 +0800
commit411e49f96494629fa3da334e8dc7cc15bd4d19cd (patch)
tree1756f0a2f86c871a6db67df7ecbd509c9df32233 /weed/wdclient
parentac162fc85769cb1b2a1f8694f9644eae7d0ce6c8 (diff)
parent4a15e9c830de1b654515308e5be8380ffa34aefa (diff)
downloadseaweedfs-411e49f96494629fa3da334e8dc7cc15bd4d19cd.tar.xz
seaweedfs-411e49f96494629fa3da334e8dc7cc15bd4d19cd.zip
Merge pull request #23 from chrislusf/master
sync
Diffstat (limited to 'weed/wdclient')
-rw-r--r--weed/wdclient/masterclient.go30
-rw-r--r--weed/wdclient/vid_map.go51
2 files changed, 44 insertions, 37 deletions
diff --git a/weed/wdclient/masterclient.go b/weed/wdclient/masterclient.go
index 3d23d8f13..7b0f73ce9 100644
--- a/weed/wdclient/masterclient.go
+++ b/weed/wdclient/masterclient.go
@@ -52,6 +52,32 @@ func (mc *MasterClient) KeepConnectedToMaster() {
}
}
+func (mc *MasterClient) FindLeader(myMasterAddress string) (leader string) {
+ for _, master := range mc.masters {
+ if master == myMasterAddress {
+ continue
+ }
+ if grpcErr := pb.WithMasterClient(master, mc.grpcDialOption, func(client master_pb.SeaweedClient) error {
+ ctx, cancel := context.WithTimeout(context.Background(), 120*time.Millisecond)
+ defer cancel()
+ resp, err := client.GetMasterConfiguration(ctx, &master_pb.GetMasterConfigurationRequest{})
+ if err != nil {
+ return err
+ }
+ leader = resp.Leader
+ return nil
+ }); grpcErr != nil {
+ glog.V(0).Infof("connect to %s: %v", master, grpcErr)
+ }
+ if leader != "" {
+ glog.V(0).Infof("existing leader is %s", leader)
+ return
+ }
+ }
+ glog.V(0).Infof("No existing leader found!")
+ return
+}
+
func (mc *MasterClient) tryAllMasters() {
nextHintedLeader := ""
for _, master := range mc.masters {
@@ -75,7 +101,7 @@ func (mc *MasterClient) tryConnectToMaster(master string) (nextHintedLeader stri
stream, err := client.KeepConnected(ctx)
if err != nil {
- glog.V(0).Infof("%s masterClient failed to keep connected to %s: %v", mc.clientType, master, err)
+ glog.V(1).Infof("%s masterClient failed to keep connected to %s: %v", mc.clientType, master, err)
return err
}
@@ -118,7 +144,7 @@ func (mc *MasterClient) tryConnectToMaster(master string) (nextHintedLeader stri
})
if gprcErr != nil {
- glog.V(0).Infof("%s masterClient failed to connect with master %v: %v", mc.clientType, master, gprcErr)
+ glog.V(1).Infof("%s masterClient failed to connect with master %v: %v", mc.clientType, master, gprcErr)
}
return
}
diff --git a/weed/wdclient/vid_map.go b/weed/wdclient/vid_map.go
index 97df49cb6..cee2da6e1 100644
--- a/weed/wdclient/vid_map.go
+++ b/weed/wdclient/vid_map.go
@@ -44,38 +44,36 @@ func (vc *vidMap) getLocationIndex(length int) (int, error) {
return int(atomic.AddInt32(&vc.cursor, 1)) % length, nil
}
-func (vc *vidMap) LookupVolumeServerUrl(vid string) (serverUrl string, err error) {
+func (vc *vidMap) LookupVolumeServerUrl(vid string) (serverUrls []string, err error) {
id, err := strconv.Atoi(vid)
if err != nil {
glog.V(1).Infof("Unknown volume id %s", vid)
- return "", err
+ return nil, err
}
- return vc.GetRandomLocation(uint32(id))
-}
-
-func (vc *vidMap) LookupFileId(fileId string) (fullUrl string, err error) {
- parts := strings.Split(fileId, ",")
- if len(parts) != 2 {
- return "", errors.New("Invalid fileId " + fileId)
+ locations, found := vc.GetLocations(uint32(id))
+ if !found {
+ return nil, fmt.Errorf("volume %d not found", id)
}
- serverUrl, lookupError := vc.LookupVolumeServerUrl(parts[0])
- if lookupError != nil {
- return "", lookupError
+ for _, loc := range locations {
+ serverUrls = append(serverUrls, loc.Url)
}
- return "http://" + serverUrl + "/" + fileId, nil
+ return
}
-func (vc *vidMap) LookupVolumeServer(fileId string) (volumeServer string, err error) {
+func (vc *vidMap) LookupFileId(fileId string) (fullUrls []string, err error) {
parts := strings.Split(fileId, ",")
if len(parts) != 2 {
- return "", errors.New("Invalid fileId " + fileId)
+ return nil, errors.New("Invalid fileId " + fileId)
}
- serverUrl, lookupError := vc.LookupVolumeServerUrl(parts[0])
+ serverUrls, lookupError := vc.LookupVolumeServerUrl(parts[0])
if lookupError != nil {
- return "", lookupError
+ return nil, lookupError
}
- return serverUrl, nil
+ for _, serverUrl := range serverUrls {
+ fullUrls = append(fullUrls, "http://"+serverUrl+"/"+fileId)
+ }
+ return
}
func (vc *vidMap) GetVidLocations(vid string) (locations []Location, err error) {
@@ -99,23 +97,6 @@ func (vc *vidMap) GetLocations(vid uint32) (locations []Location, found bool) {
return
}
-func (vc *vidMap) GetRandomLocation(vid uint32) (serverUrl string, err error) {
- vc.RLock()
- defer vc.RUnlock()
-
- locations := vc.vid2Locations[vid]
- if len(locations) == 0 {
- return "", fmt.Errorf("volume %d not found", vid)
- }
-
- index, err := vc.getLocationIndex(len(locations))
- if err != nil {
- return "", fmt.Errorf("volume %d: %v", vid, err)
- }
-
- return locations[index].Url, nil
-}
-
func (vc *vidMap) addLocation(vid uint32, location Location) {
vc.Lock()
defer vc.Unlock()