Skip to content

Commit

Permalink
refactor seaweedfs#3616 (seaweedfs#3625)
Browse files Browse the repository at this point in the history
  • Loading branch information
@kmlebedev
kmlebedev committed Sep 8, 2022
1 parent 7de1129 commit 31d2f77
Showing 1 changed file with 106 additions and 106 deletions.
212 changes: 106 additions & 106 deletions weed/server/master_grpc_server.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -104,136 +104,136 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
return err
}

if ms.Topo.IsLeader() {
ms.Topo.Sequence.SetMax(heartbeat.MaxFileKey)

if dn == nil {
dcName, rackName := ms.Topo.Configuration.Locate(heartbeat.Ip, heartbeat.DataCenter, heartbeat.Rack)
dc := ms.Topo.GetOrCreateDataCenter(dcName)
rack := dc.GetOrCreateRack(rackName)
dn = rack.GetOrCreateDataNode(heartbeat.Ip, int(heartbeat.Port), int(heartbeat.GrpcPort), heartbeat.PublicUrl, heartbeat.MaxVolumeCounts)
glog.V(0).Infof("added volume server %d: %v:%d %v", dn.Counter, heartbeat.GetIp(), heartbeat.GetPort(), heartbeat.LocationUuids)
uuidlist, err := ms.RegisterUuids(heartbeat)
if err != nil {
if stream_err := stream.Send(&master_pb.HeartbeatResponse{
DuplicatedUuids: uuidlist,
}); stream_err != nil {
glog.Warningf("SendHeartbeat.Send DuplicatedDirectory response to %s:%d %v", dn.Ip, dn.Port, stream_err)
return stream_err
}
return err
if !ms.Topo.IsLeader() {
// tell the volume servers about the leader
newLeader, err := ms.Topo.Leader()
if err != nil {
glog.Warningf("SendHeartbeat find leader: %v", err)
return err
}
if err := stream.Send(&master_pb.HeartbeatResponse{
Leader: string(newLeader),
}); err != nil {
if dn != nil {
glog.Warningf("SendHeartbeat.Send response to %s:%d %v", dn.Ip, dn.Port, err)
} else {
glog.Warningf("SendHeartbeat.Send response %v", err)
}
return err
}
continue
}

if err := stream.Send(&master_pb.HeartbeatResponse{
VolumeSizeLimit: uint64(ms.option.VolumeSizeLimitMB) * 1024 * 1024,
}); err != nil {
glog.Warningf("SendHeartbeat.Send volume size to %s:%d %v", dn.Ip, dn.Port, err)
return err
ms.Topo.Sequence.SetMax(heartbeat.MaxFileKey)
if dn == nil {
dcName, rackName := ms.Topo.Configuration.Locate(heartbeat.Ip, heartbeat.DataCenter, heartbeat.Rack)
dc := ms.Topo.GetOrCreateDataCenter(dcName)
rack := dc.GetOrCreateRack(rackName)
dn = rack.GetOrCreateDataNode(heartbeat.Ip, int(heartbeat.Port), int(heartbeat.GrpcPort), heartbeat.PublicUrl, heartbeat.MaxVolumeCounts)
glog.V(0).Infof("added volume server %d: %v:%d %v", dn.Counter, heartbeat.GetIp(), heartbeat.GetPort(), heartbeat.LocationUuids)
uuidlist, err := ms.RegisterUuids(heartbeat)
if err != nil {
if stream_err := stream.Send(&master_pb.HeartbeatResponse{
DuplicatedUuids: uuidlist,
}); stream_err != nil {
glog.Warningf("SendHeartbeat.Send DuplicatedDirectory response to %s:%d %v", dn.Ip, dn.Port, stream_err)
return stream_err
}
stats.MasterReceivedHeartbeatCounter.WithLabelValues("dataNode").Inc()
dn.Counter++
return err
}

dn.AdjustMaxVolumeCounts(heartbeat.MaxVolumeCounts)
if err := stream.Send(&master_pb.HeartbeatResponse{
VolumeSizeLimit: uint64(ms.option.VolumeSizeLimitMB) * 1024 * 1024,
}); err != nil {
glog.Warningf("SendHeartbeat.Send volume size to %s:%d %v", dn.Ip, dn.Port, err)
return err
}
stats.MasterReceivedHeartbeatCounter.WithLabelValues("dataNode").Inc()
dn.Counter++
}

glog.V(4).Infof("master received heartbeat %s", heartbeat.String())
stats.MasterReceivedHeartbeatCounter.WithLabelValues("total").Inc()
dn.AdjustMaxVolumeCounts(heartbeat.MaxVolumeCounts)

message := &master_pb.VolumeLocation{
Url: dn.Url(),
PublicUrl: dn.PublicUrl,
DataCenter: dn.GetDataCenterId(),
}
if len(heartbeat.NewVolumes) > 0 {
stats.FilerRequestCounter.WithLabelValues("newVolumes").Inc()
}
if len(heartbeat.DeletedVolumes) > 0 {
stats.FilerRequestCounter.WithLabelValues("deletedVolumes").Inc()
glog.V(4).Infof("master received heartbeat %s", heartbeat.String())
stats.MasterReceivedHeartbeatCounter.WithLabelValues("total").Inc()

message := &master_pb.VolumeLocation{
Url: dn.Url(),
PublicUrl: dn.PublicUrl,
DataCenter: dn.GetDataCenterId(),
}
if len(heartbeat.NewVolumes) > 0 {
stats.FilerRequestCounter.WithLabelValues("newVolumes").Inc()
}
if len(heartbeat.DeletedVolumes) > 0 {
stats.FilerRequestCounter.WithLabelValues("deletedVolumes").Inc()
}
if len(heartbeat.NewVolumes) > 0 || len(heartbeat.DeletedVolumes) > 0 {
// process delta volume ids if exists for fast volume id updates
for _, volInfo := range heartbeat.NewVolumes {
message.NewVids = append(message.NewVids, volInfo.Id)
}
if len(heartbeat.NewVolumes) > 0 || len(heartbeat.DeletedVolumes) > 0 {
// process delta volume ids if exists for fast volume id updates
for _, volInfo := range heartbeat.NewVolumes {
message.NewVids = append(message.NewVids, volInfo.Id)
}
for _, volInfo := range heartbeat.DeletedVolumes {
message.DeletedVids = append(message.DeletedVids, volInfo.Id)
}
// update master internal volume layouts
ms.Topo.IncrementalSyncDataNodeRegistration(heartbeat.NewVolumes, heartbeat.DeletedVolumes, dn)
for _, volInfo := range heartbeat.DeletedVolumes {
message.DeletedVids = append(message.DeletedVids, volInfo.Id)
}
// update master internal volume layouts
ms.Topo.IncrementalSyncDataNodeRegistration(heartbeat.NewVolumes, heartbeat.DeletedVolumes, dn)
}

if len(heartbeat.Volumes) > 0 || heartbeat.HasNoVolumes {
dcName, rackName := ms.Topo.Configuration.Locate(heartbeat.Ip, heartbeat.DataCenter, heartbeat.Rack)
ms.Topo.DataNodeRegistration(dcName, rackName, dn)
if len(heartbeat.Volumes) > 0 || heartbeat.HasNoVolumes {
dcName, rackName := ms.Topo.Configuration.Locate(heartbeat.Ip, heartbeat.DataCenter, heartbeat.Rack)
ms.Topo.DataNodeRegistration(dcName, rackName, dn)

// process heartbeat.Volumes
stats.MasterReceivedHeartbeatCounter.WithLabelValues("Volumes").Inc()
newVolumes, deletedVolumes := ms.Topo.SyncDataNodeRegistration(heartbeat.Volumes, dn)
// process heartbeat.Volumes
stats.MasterReceivedHeartbeatCounter.WithLabelValues("Volumes").Inc()
newVolumes, deletedVolumes := ms.Topo.SyncDataNodeRegistration(heartbeat.Volumes, dn)

for _, v := range newVolumes {
glog.V(0).Infof("master see new volume %d from %s", uint32(v.Id), dn.Url())
message.NewVids = append(message.NewVids, uint32(v.Id))
}
for _, v := range deletedVolumes {
glog.V(0).Infof("master see deleted volume %d from %s", uint32(v.Id), dn.Url())
message.DeletedVids = append(message.DeletedVids, uint32(v.Id))
}
for _, v := range newVolumes {
glog.V(0).Infof("master see new volume %d from %s", uint32(v.Id), dn.Url())
message.NewVids = append(message.NewVids, uint32(v.Id))
}
for _, v := range deletedVolumes {
glog.V(0).Infof("master see deleted volume %d from %s", uint32(v.Id), dn.Url())
message.DeletedVids = append(message.DeletedVids, uint32(v.Id))
}
}

if len(heartbeat.NewEcShards) > 0 || len(heartbeat.DeletedEcShards) > 0 {
stats.MasterReceivedHeartbeatCounter.WithLabelValues("newEcShards").Inc()
// update master internal volume layouts
ms.Topo.IncrementalSyncDataNodeEcShards(heartbeat.NewEcShards, heartbeat.DeletedEcShards, dn)
if len(heartbeat.NewEcShards) > 0 || len(heartbeat.DeletedEcShards) > 0 {
stats.MasterReceivedHeartbeatCounter.WithLabelValues("newEcShards").Inc()
// update master internal volume layouts
ms.Topo.IncrementalSyncDataNodeEcShards(heartbeat.NewEcShards, heartbeat.DeletedEcShards, dn)

for _, s := range heartbeat.NewEcShards {
message.NewEcVids = append(message.NewEcVids, s.Id)
}
for _, s := range heartbeat.DeletedEcShards {
if dn.HasEcShards(needle.VolumeId(s.Id)) {
continue
}
message.DeletedEcVids = append(message.DeletedEcVids, s.Id)
for _, s := range heartbeat.NewEcShards {
message.NewEcVids = append(message.NewEcVids, s.Id)
}
for _, s := range heartbeat.DeletedEcShards {
if dn.HasEcShards(needle.VolumeId(s.Id)) {
continue
}

message.DeletedEcVids = append(message.DeletedEcVids, s.Id)
}

if len(heartbeat.EcShards) > 0 || heartbeat.HasNoEcShards {
stats.MasterReceivedHeartbeatCounter.WithLabelValues("ecShards").Inc()
glog.V(4).Infof("master received ec shards from %s: %+v", dn.Url(), heartbeat.EcShards)
newShards, deletedShards := ms.Topo.SyncDataNodeEcShards(heartbeat.EcShards, dn)
}

// broadcast the ec vid changes to master clients
for _, s := range newShards {
message.NewEcVids = append(message.NewEcVids, uint32(s.VolumeId))
}
for _, s := range deletedShards {
if dn.HasVolumesById(s.VolumeId) {
continue
}
message.DeletedEcVids = append(message.DeletedEcVids, uint32(s.VolumeId))
}
if len(heartbeat.EcShards) > 0 || heartbeat.HasNoEcShards {
stats.MasterReceivedHeartbeatCounter.WithLabelValues("ecShards").Inc()
glog.V(4).Infof("master received ec shards from %s: %+v", dn.Url(), heartbeat.EcShards)
newShards, deletedShards := ms.Topo.SyncDataNodeEcShards(heartbeat.EcShards, dn)

// broadcast the ec vid changes to master clients
for _, s := range newShards {
message.NewEcVids = append(message.NewEcVids, uint32(s.VolumeId))
}
if len(message.NewVids) > 0 || len(message.DeletedVids) > 0 || len(message.NewEcVids) > 0 || len(message.DeletedEcVids) > 0 {
ms.broadcastToClients(&master_pb.KeepConnectedResponse{VolumeLocation: message})
for _, s := range deletedShards {
if dn.HasVolumesById(s.VolumeId) {
continue
}
message.DeletedEcVids = append(message.DeletedEcVids, uint32(s.VolumeId))
}
}

// tell the volume servers about the leader
newLeader, err := ms.Topo.Leader()
if err != nil {
glog.Warningf("SendHeartbeat find leader: %v", err)
return err
}
if err := stream.Send(&master_pb.HeartbeatResponse{
Leader: string(newLeader),
}); err != nil {
if dn != nil {
glog.Warningf("SendHeartbeat.Send response to %s:%d %v", dn.Ip, dn.Port, err)
} else {
glog.Warningf("SendHeartbeat.Send response %v", err)
}
return err
if len(message.NewVids) > 0 || len(message.DeletedVids) > 0 || len(message.NewEcVids) > 0 || len(message.DeletedEcVids) > 0 {
ms.broadcastToClients(&master_pb.KeepConnectedResponse{VolumeLocation: message})
}
}
}
Expand Down

0 comments on commit 31d2f77

Please sign in to comment.