skip deltaBeat if dn is zero (#3630)

* skip deltaBeat
https://github.com/seaweedfs/seaweedfs/issues/3629

* fix GrpcPort

* skip url :0

* skip empty DataCenter or Rack

* skip empty heartbeat Ip

* dell msg add DataCenter

* comment todo

* fix
This commit is contained in:
Konstantin Lebedev 2022-09-12 10:31:53 +05:00 committed by GitHub
parent b834027c5a
commit 721c6197f9
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 46 additions and 18 deletions

View file

@ -0,0 +1,5 @@
package master_pb
func (v *VolumeLocation) IsEmptyUrl() bool {
return v.Url == "" || v.Url == ":0"
}

View file

@ -70,6 +70,7 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
} }
message := &master_pb.VolumeLocation{ message := &master_pb.VolumeLocation{
DataCenter: dn.GetDataCenterId(),
Url: dn.Url(), Url: dn.Url(),
PublicUrl: dn.PublicUrl, PublicUrl: dn.PublicUrl,
} }
@ -126,6 +127,10 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
ms.Topo.Sequence.SetMax(heartbeat.MaxFileKey) ms.Topo.Sequence.SetMax(heartbeat.MaxFileKey)
if dn == nil { if dn == nil {
// Skip delta heartbeat for volume server versions better than 3.28 https://github.com/seaweedfs/seaweedfs/pull/3630
if heartbeat.Ip == "" {
continue
} // ToDo must be removed after update major version
dcName, rackName := ms.Topo.Configuration.Locate(heartbeat.Ip, heartbeat.DataCenter, heartbeat.Rack) dcName, rackName := ms.Topo.Configuration.Locate(heartbeat.Ip, heartbeat.DataCenter, heartbeat.Rack)
dc := ms.Topo.GetOrCreateDataCenter(dcName) dc := ms.Topo.GetOrCreateDataCenter(dcName)
rack := dc.GetOrCreateRack(rackName) rack := dc.GetOrCreateRack(rackName)
@ -181,8 +186,10 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
} }
if len(heartbeat.Volumes) > 0 || heartbeat.HasNoVolumes { if len(heartbeat.Volumes) > 0 || heartbeat.HasNoVolumes {
if heartbeat.Ip != "" {
dcName, rackName := ms.Topo.Configuration.Locate(heartbeat.Ip, heartbeat.DataCenter, heartbeat.Rack) dcName, rackName := ms.Topo.Configuration.Locate(heartbeat.Ip, heartbeat.DataCenter, heartbeat.Rack)
ms.Topo.DataNodeRegistration(dcName, rackName, dn) ms.Topo.DataNodeRegistration(dcName, rackName, dn)
}
// process heartbeat.Volumes // process heartbeat.Volumes
stats.MasterReceivedHeartbeatCounter.WithLabelValues("Volumes").Inc() stats.MasterReceivedHeartbeatCounter.WithLabelValues("Volumes").Inc()

View file

@ -160,11 +160,18 @@ func (vs *VolumeServer) doHeartbeat(masterAddress pb.ServerAddress, grpcDialOpti
volumeTickChan := time.Tick(sleepInterval) volumeTickChan := time.Tick(sleepInterval)
ecShardTickChan := time.Tick(17 * sleepInterval) ecShardTickChan := time.Tick(17 * sleepInterval)
dataCenter := vs.store.GetDataCenter()
rack := vs.store.GetRack()
ip := vs.store.Ip
port := uint32(vs.store.Port)
for { for {
select { select {
case volumeMessage := <-vs.store.NewVolumesChan: case volumeMessage := <-vs.store.NewVolumesChan:
deltaBeat := &master_pb.Heartbeat{ deltaBeat := &master_pb.Heartbeat{
Ip: ip,
Port: port,
DataCenter: dataCenter,
Rack: rack,
NewVolumes: []*master_pb.VolumeShortInformationMessage{ NewVolumes: []*master_pb.VolumeShortInformationMessage{
&volumeMessage, &volumeMessage,
}, },
@ -176,6 +183,10 @@ func (vs *VolumeServer) doHeartbeat(masterAddress pb.ServerAddress, grpcDialOpti
} }
case ecShardMessage := <-vs.store.NewEcShardsChan: case ecShardMessage := <-vs.store.NewEcShardsChan:
deltaBeat := &master_pb.Heartbeat{ deltaBeat := &master_pb.Heartbeat{
Ip: ip,
Port: port,
DataCenter: dataCenter,
Rack: rack,
NewEcShards: []*master_pb.VolumeEcShardInformationMessage{ NewEcShards: []*master_pb.VolumeEcShardInformationMessage{
&ecShardMessage, &ecShardMessage,
}, },
@ -188,6 +199,10 @@ func (vs *VolumeServer) doHeartbeat(masterAddress pb.ServerAddress, grpcDialOpti
} }
case volumeMessage := <-vs.store.DeletedVolumesChan: case volumeMessage := <-vs.store.DeletedVolumesChan:
deltaBeat := &master_pb.Heartbeat{ deltaBeat := &master_pb.Heartbeat{
Ip: ip,
Port: port,
DataCenter: dataCenter,
Rack: rack,
DeletedVolumes: []*master_pb.VolumeShortInformationMessage{ DeletedVolumes: []*master_pb.VolumeShortInformationMessage{
&volumeMessage, &volumeMessage,
}, },
@ -199,6 +214,10 @@ func (vs *VolumeServer) doHeartbeat(masterAddress pb.ServerAddress, grpcDialOpti
} }
case ecShardMessage := <-vs.store.DeletedEcShardsChan: case ecShardMessage := <-vs.store.DeletedEcShardsChan:
deltaBeat := &master_pb.Heartbeat{ deltaBeat := &master_pb.Heartbeat{
Ip: ip,
Port: port,
DataCenter: dataCenter,
Rack: rack,
DeletedEcShards: []*master_pb.VolumeEcShardInformationMessage{ DeletedEcShards: []*master_pb.VolumeEcShardInformationMessage{
&ecShardMessage, &ecShardMessage,
}, },
@ -227,12 +246,12 @@ func (vs *VolumeServer) doHeartbeat(masterAddress pb.ServerAddress, grpcDialOpti
case <-vs.stopChan: case <-vs.stopChan:
var volumeMessages []*master_pb.VolumeInformationMessage var volumeMessages []*master_pb.VolumeInformationMessage
emptyBeat := &master_pb.Heartbeat{ emptyBeat := &master_pb.Heartbeat{
Ip: vs.store.Ip, Ip: ip,
Port: uint32(vs.store.Port), Port: port,
PublicUrl: vs.store.PublicUrl, PublicUrl: vs.store.PublicUrl,
MaxFileKey: uint64(0), MaxFileKey: uint64(0),
DataCenter: vs.store.GetDataCenter(), DataCenter: dataCenter,
Rack: vs.store.GetRack(), Rack: rack,
Volumes: volumeMessages, Volumes: volumeMessages,
HasNoVolumes: len(volumeMessages) == 0, HasNoVolumes: len(volumeMessages) == 0,
} }

View file

@ -22,7 +22,6 @@ type topology struct {
type Configuration struct { type Configuration struct {
XMLName xml.Name `xml:"Configuration"` XMLName xml.Name `xml:"Configuration"`
Topo topology `xml:"Topology"` Topo topology `xml:"Topology"`
ip2location map[string]loc // this is not used any more. leave it here for later.
} }
func (c *Configuration) String() string { func (c *Configuration) String() string {
@ -33,12 +32,6 @@ func (c *Configuration) String() string {
} }
func (c *Configuration) Locate(ip string, dcName string, rackName string) (dc string, rack string) { func (c *Configuration) Locate(ip string, dcName string, rackName string) (dc string, rack string) {
if c != nil && c.ip2location != nil {
if loc, ok := c.ip2location[ip]; ok {
return loc.dcName, loc.rackName
}
}
if dcName == "" { if dcName == "" {
dcName = "DefaultDataCenter" dcName = "DefaultDataCenter"
} }

View file

@ -262,6 +262,10 @@ func (mc *MasterClient) tryConnectToMaster(master pb.ServerAddress) (nextHintedL
} }
func (mc *MasterClient) updateVidMap(resp *master_pb.KeepConnectedResponse) { func (mc *MasterClient) updateVidMap(resp *master_pb.KeepConnectedResponse) {
if resp.VolumeLocation.IsEmptyUrl() {
glog.V(0).Infof("updateVidMap ignore short heartbeat: %+v", resp)
return
}
// process new volume location // process new volume location
loc := Location{ loc := Location{
Url: resp.VolumeLocation.Url, Url: resp.VolumeLocation.Url,