gitlab-com-seaweedfs/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2024-01-19 02:48:24 +00:00
Code Issues Projects Releases Wiki Activity

volume: the variable for the master node may be stale?

Browse source 
related to https://github.com/chrislusf/seaweedfs/issues/1414
This commit is contained in:
Chris Lu 2020-08-06 09:48:54 -07:00
parent 4ecfa9879d
commit 93ea0801ea
2 changed files with 12 additions and 27 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

19
weed/server/master_grpc_server.go
View file

@ -19,14 +19,13 @@ import (
func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServer) error { func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServer) error {
var dn *topology.DataNode var dn *topology.DataNode
t := ms.Topo
defer func() { defer func() {
if dn != nil { if dn != nil {
// if the volume server disconnects and reconnects quickly // if the volume server disconnects and reconnects quickly
// the unregister and register can race with each other // the unregister and register can race with each other
t.UnRegisterDataNode(dn) ms.Topo.UnRegisterDataNode(dn)
glog.V(0).Infof("unregister disconnected volume server %s:%d", dn.Ip, dn.Port) glog.V(0).Infof("unregister disconnected volume server %s:%d", dn.Ip, dn.Port)
message := &master_pb.VolumeLocation{ message := &master_pb.VolumeLocation{
@ -62,11 +61,11 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
return err return err
} }
t.Sequence.SetMax(heartbeat.MaxFileKey) ms.Topo.Sequence.SetMax(heartbeat.MaxFileKey)
if dn == nil { if dn == nil {
dcName, rackName := t.Configuration.Locate(heartbeat.Ip, heartbeat.DataCenter, heartbeat.Rack) dcName, rackName := ms.Topo.Configuration.Locate(heartbeat.Ip, heartbeat.DataCenter, heartbeat.Rack)
dc := t.GetOrCreateDataCenter(dcName) dc := ms.Topo.GetOrCreateDataCenter(dcName)
rack := dc.GetOrCreateRack(rackName) rack := dc.GetOrCreateRack(rackName)
dn = rack.GetOrCreateDataNode(heartbeat.Ip, dn = rack.GetOrCreateDataNode(heartbeat.Ip,
int(heartbeat.Port), heartbeat.PublicUrl, int(heartbeat.Port), heartbeat.PublicUrl,
@ -102,12 +101,12 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
message.DeletedVids = append(message.DeletedVids, volInfo.Id) message.DeletedVids = append(message.DeletedVids, volInfo.Id)
} }
// update master internal volume layouts // update master internal volume layouts
t.IncrementalSyncDataNodeRegistration(heartbeat.NewVolumes, heartbeat.DeletedVolumes, dn) ms.Topo.IncrementalSyncDataNodeRegistration(heartbeat.NewVolumes, heartbeat.DeletedVolumes, dn)
} }
if len(heartbeat.Volumes) > 0 || heartbeat.HasNoVolumes { if len(heartbeat.Volumes) > 0 || heartbeat.HasNoVolumes {
// process heartbeat.Volumes // process heartbeat.Volumes
newVolumes, deletedVolumes := t.SyncDataNodeRegistration(heartbeat.Volumes, dn) newVolumes, deletedVolumes := ms.Topo.SyncDataNodeRegistration(heartbeat.Volumes, dn)
for _, v := range newVolumes { for _, v := range newVolumes {
glog.V(0).Infof("master see new volume %d from %s", uint32(v.Id), dn.Url()) glog.V(0).Infof("master see new volume %d from %s", uint32(v.Id), dn.Url())
@ -122,7 +121,7 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
if len(heartbeat.NewEcShards) > 0 || len(heartbeat.DeletedEcShards) > 0 { if len(heartbeat.NewEcShards) > 0 || len(heartbeat.DeletedEcShards) > 0 {
// update master internal volume layouts // update master internal volume layouts
t.IncrementalSyncDataNodeEcShards(heartbeat.NewEcShards, heartbeat.DeletedEcShards, dn) ms.Topo.IncrementalSyncDataNodeEcShards(heartbeat.NewEcShards, heartbeat.DeletedEcShards, dn)
for _, s := range heartbeat.NewEcShards { for _, s := range heartbeat.NewEcShards {
message.NewVids = append(message.NewVids, s.Id) message.NewVids = append(message.NewVids, s.Id)
@ -138,7 +137,7 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
if len(heartbeat.EcShards) > 0 || heartbeat.HasNoEcShards { if len(heartbeat.EcShards) > 0 || heartbeat.HasNoEcShards {
glog.V(1).Infof("master recieved ec shards from %s: %+v", dn.Url(), heartbeat.EcShards) glog.V(1).Infof("master recieved ec shards from %s: %+v", dn.Url(), heartbeat.EcShards)
newShards, deletedShards := t.SyncDataNodeEcShards(heartbeat.EcShards, dn) newShards, deletedShards := ms.Topo.SyncDataNodeEcShards(heartbeat.EcShards, dn)
// broadcast the ec vid changes to master clients // broadcast the ec vid changes to master clients
for _, s := range newShards { for _, s := range newShards {
@ -163,7 +162,7 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
} }
// tell the volume servers about the leader // tell the volume servers about the leader
newLeader, err := t.Leader() newLeader, err := ms.Topo.Leader()
if err != nil { if err != nil {
glog.Warningf("SendHeartbeat find leader: %v", err) glog.Warningf("SendHeartbeat find leader: %v", err)
return err return err

20
weed/server/volume_grpc_client_to_master.go
View file

@ -2,7 +2,6 @@ package weed_server
import ( import (
"fmt" "fmt"
"net"
"time" "time"
"google.golang.org/grpc" "google.golang.org/grpc"
@ -87,12 +86,12 @@ func (vs *VolumeServer) doHeartbeat(masterNode, masterGrpcAddress string, grpcDi
vs.store.SetVolumeSizeLimit(in.GetVolumeSizeLimit()) vs.store.SetVolumeSizeLimit(in.GetVolumeSizeLimit())
if vs.store.MaybeAdjustVolumeMax() { if vs.store.MaybeAdjustVolumeMax() {
if err = stream.Send(vs.store.CollectHeartbeat()); err != nil { 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", vs.currentMaster, err)
} }
} }
} }
if in.GetLeader() != "" && masterNode != in.GetLeader() && !isSameIP(in.GetLeader(), masterNode) { if in.GetLeader() != "" && vs.currentMaster != in.GetLeader() {
glog.V(0).Infof("Volume Server found a new master newLeader: %v instead of %v", in.GetLeader(), masterNode) glog.V(0).Infof("Volume Server found a new master newLeader: %v instead of %v", in.GetLeader(), vs.currentMaster)
newLeader = in.GetLeader() newLeader = in.GetLeader()
doneChan <- nil doneChan <- nil
return return
@ -185,16 +184,3 @@ func (vs *VolumeServer) doHeartbeat(masterNode, masterGrpcAddress string, grpcDi
} }
} }
} }
func isSameIP(ip string, host string) bool {
ips, err := net.LookupIP(host)
if err != nil {
return false
}
for _, t := range ips {
if ip == t.String() {
return true
}
}
return false
}