lsm/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2025-08-01 17:46:55 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge pull request #3256 from shichanglin5/fix_keep_connected

Browse Source 
fixed `volume xx not found` caused by missing VolumeLocation events
This commit is contained in:
Chris Lu 2022-06-30 00:10:09 -07:00 committed by GitHub
commit 413cb3d389
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 38 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

68
weed/wdclient/masterclient.go
View File

@ -168,15 +168,19 @@ func (mc *MasterClient) tryConnectToMaster(master pb.ServerAddress) (nextHintedL
} }
// check if it is the leader to determine whether to reset the vidMap // check if it is the leader to determine whether to reset the vidMap
if resp.VolumeLocation != nil && resp.VolumeLocation.Leader != "" && string(master) != resp.VolumeLocation.Leader { if resp.VolumeLocation != nil {
glog.V(0).Infof("master %v redirected to leader %v", master, resp.VolumeLocation.Leader) if resp.VolumeLocation.Leader != "" && string(master) != resp.VolumeLocation.Leader {
nextHintedLeader = pb.ServerAddress(resp.VolumeLocation.Leader) glog.V(0).Infof("master %v redirected to leader %v", master, resp.VolumeLocation.Leader)
stats.MasterClientConnectCounter.WithLabelValues(stats.RedirectedToleader).Inc() nextHintedLeader = pb.ServerAddress(resp.VolumeLocation.Leader)
return nil stats.MasterClientConnectCounter.WithLabelValues(stats.RedirectedToleader).Inc()
return nil
}
mc.vidMap = newVidMap("")
mc.updateVidMap(resp)
} else {
mc.vidMap = newVidMap("")
} }
mc.currentMaster = master mc.currentMaster = master
mc.vidMap = newVidMap("")
for { for {
resp, err := stream.Recv() resp, err := stream.Recv()
@ -195,29 +199,7 @@ func (mc *MasterClient) tryConnectToMaster(master pb.ServerAddress) (nextHintedL
return nil return nil
} }
// process new volume location mc.updateVidMap(resp)
loc := Location{
Url: resp.VolumeLocation.Url,
PublicUrl: resp.VolumeLocation.PublicUrl,
DataCenter: resp.VolumeLocation.DataCenter,
GrpcPort: int(resp.VolumeLocation.GrpcPort),
}
for _, newVid := range resp.VolumeLocation.NewVids {
glog.V(1).Infof("%s.%s: %s masterClient adds volume %d", mc.FilerGroup, mc.clientType, loc.Url, newVid)
mc.addLocation(newVid, loc)
}
for _, deletedVid := range resp.VolumeLocation.DeletedVids {
glog.V(1).Infof("%s.%s: %s masterClient removes volume %d", mc.FilerGroup, mc.clientType, loc.Url, deletedVid)
mc.deleteLocation(deletedVid, loc)
}
for _, newEcVid := range resp.VolumeLocation.NewEcVids {
glog.V(1).Infof("%s.%s: %s masterClient adds ec volume %d", mc.FilerGroup, mc.clientType, loc.Url, newEcVid)
mc.addEcLocation(newEcVid, loc)
}
for _, deletedEcVid := range resp.VolumeLocation.DeletedEcVids {
glog.V(1).Infof("%s.%s: %s masterClient removes ec volume %d", mc.FilerGroup, mc.clientType, loc.Url, deletedEcVid)
mc.deleteEcLocation(deletedEcVid, loc)
}
} }
if resp.ClusterNodeUpdate != nil { if resp.ClusterNodeUpdate != nil {
@ -245,6 +227,32 @@ func (mc *MasterClient) tryConnectToMaster(master pb.ServerAddress) (nextHintedL
return return
} }
func (mc *MasterClient) updateVidMap(resp *master_pb.KeepConnectedResponse) {
// process new volume location
loc := Location{
Url: resp.VolumeLocation.Url,
PublicUrl: resp.VolumeLocation.PublicUrl,
DataCenter: resp.VolumeLocation.DataCenter,
GrpcPort: int(resp.VolumeLocation.GrpcPort),
}
for _, newVid := range resp.VolumeLocation.NewVids {
glog.V(1).Infof("%s.%s: %s masterClient adds volume %d", mc.FilerGroup, mc.clientType, loc.Url, newVid)
mc.addLocation(newVid, loc)
}
for _, deletedVid := range resp.VolumeLocation.DeletedVids {
glog.V(1).Infof("%s.%s: %s masterClient removes volume %d", mc.FilerGroup, mc.clientType, loc.Url, deletedVid)
mc.deleteLocation(deletedVid, loc)
}
for _, newEcVid := range resp.VolumeLocation.NewEcVids {
glog.V(1).Infof("%s.%s: %s masterClient adds ec volume %d", mc.FilerGroup, mc.clientType, loc.Url, newEcVid)
mc.addEcLocation(newEcVid, loc)
}
for _, deletedEcVid := range resp.VolumeLocation.DeletedEcVids {
glog.V(1).Infof("%s.%s: %s masterClient removes ec volume %d", mc.FilerGroup, mc.clientType, loc.Url, deletedEcVid)
mc.deleteEcLocation(deletedEcVid, loc)
}
}
func (mc *MasterClient) WithClient(streamingMode bool, fn func(client master_pb.SeaweedClient) error) error { func (mc *MasterClient) WithClient(streamingMode bool, fn func(client master_pb.SeaweedClient) error) error {
return util.Retry("master grpc", func() error { return util.Retry("master grpc", func() error {
for mc.currentMaster == "" { for mc.currentMaster == "" {