gian/cubefs
1
0
Fork 0
mirror of https://github.com/cubefs/cubefs.git synced 2026-08-02 02:00:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix(flashnodeb): asynchronously send flashnode heartbeat message#1000044633

Browse Source 
Signed-off-by: clinx <chenlin1@oppo.com>
This commit is contained in:
clinx 2025-04-07 12:40:45 +08:00 committed by zhumingze1108
parent 5faeca7fd9
commit 72f6196ba6
6 changed files with 103 additions and 79 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
flashnode/flashnode_op.go
View File

@ -81,7 +81,9 @@ func (f *FlashNode) SetTimeout(handleReadTimeout int, readDataNodeTimeout int) {
func (f *FlashNode) opFlashNodeHeartbeat(conn net.Conn, p *proto.Packet) (err error) {
data := p.Data
go responseAckOKToMaster(conn, p)
req := &proto.HeartBeatRequest{}
resp := &proto.FlashNodeHeartbeatResponse{}
adminTask := &proto.AdminTask{
Request: req,
}
@ -90,13 +92,15 @@ func (f *FlashNode) opFlashNodeHeartbeat(conn net.Conn, p *proto.Packet) (err er
if err = decode.Decode(adminTask); err == nil {
f.SetTimeout(req.FlashNodeHandleReadTimeout, req.FlashNodeReadDataNodeTimeout)
} else {
log.LogErrorf("decode HeartBeatRequest error: %s", err.Error())
log.LogWarnf("decode HeartBeatRequest error: %s", err.Error())
resp.Status = proto.TaskFailed
resp.Result = fmt.Sprintf("flashnode(%v) heartbeat decode err(%v)", f.localAddr, err.Error())
goto end
}
resp := &proto.FlashNodeHeartbeatResponse{}
resp.Stat = make([]*proto.FlashNodeDiskCacheStat, 0)
for _, cacheStat := range f.cacheEngine.GetHeartBeatCacheStat() {
stat := &proto.FlashNodeDiskCacheStat{
cacheStat := &proto.FlashNodeDiskCacheStat{
DataPath: cacheStat.DataPath,
Medium: cacheStat.Medium,
Total: cacheStat.Total,
@ -109,11 +113,12 @@ func (f *FlashNode) opFlashNodeHeartbeat(conn net.Conn, p *proto.Packet) (err er
KeyNum: cacheStat.Num,
Status: cacheStat.Status,
}
resp.Stat = append(resp.Stat, stat)
resp.Stat = append(resp.Stat, cacheStat)
}
resp.LimiterStatus = &proto.FlashNodeLimiterStatusInfo{
WriteStatus: proto.FlashNodeLimiterStatus{Status: f.limitWrite.Status(true), DiskNum: len(f.disks), ReadTimeout: f.handleReadTimeout},
ReadStatus: proto.FlashNodeLimiterStatus{Status: f.limitRead.Status(true), DiskNum: len(f.disks), ReadTimeout: f.handleReadTimeout},
}
writeStatus := proto.FlashNodeLimiterStatus{Status: f.limitWrite.Status(true), DiskNum: len(f.disks), ReadTimeout: f.handleReadTimeout}
readStatus := proto.FlashNodeLimiterStatus{Status: f.limitRead.Status(true), DiskNum: len(f.disks), ReadTimeout: f.handleReadTimeout}
resp.LimiterStatus = &proto.FlashNodeLimiterStatusInfo{WriteStatus: writeStatus, ReadStatus: readStatus}
resp.FlashNodeTaskCountLimit = f.taskCountLimit
resp.ManualScanningTasks = make(map[string]*proto.FlashNodeManualTaskResponse)
@ -123,20 +128,14 @@ func (f *FlashNode) opFlashNodeHeartbeat(conn net.Conn, p *proto.Packet) (err er
resp.ManualScanningTasks[scanner.ID] = result
return true
})
reply, err := json.Marshal(resp)
if err != nil {
p.PacketErrorWithBody(proto.OpErr, []byte(err.Error()))
return
resp.Status = proto.TaskSucceeds
end:
adminTask.Response = resp
f.respondToMaster(adminTask)
if log.EnableInfo() {
log.LogInfof("[opMasterHeartbeat] master:%s handleReadTimeout %v(ms) readDataNodeTimeout %v(ms)",
conn.RemoteAddr().String(), req.FlashNodeHandleReadTimeout, req.FlashNodeReadDataNodeTimeout)
}
p.PacketOkWithBody(reply)
if err = p.WriteToConn(conn); err != nil {
log.LogErrorf("ack master response: %s", err.Error())
return err
}
log.LogInfof("[opMasterHeartbeat] master:%s handleReadTimeout %v(ms) readDataNodeTimeout %v(ms)",
conn.RemoteAddr().String(), req.FlashNodeHandleReadTimeout, req.FlashNodeReadDataNodeTimeout)
return
}

2
master/admin_task_manager.go
View File

@ -239,7 +239,7 @@ func (sender *AdminTaskManager) DelTask(t *proto.AdminTask) {
if !ok {
return
}
if t.OpCode != proto.OpMetaNodeHeartbeat && t.OpCode != proto.OpDataNodeHeartbeat && t.OpCode != proto.OpLcNodeHeartbeat {
if t.OpCode != proto.OpMetaNodeHeartbeat && t.OpCode != proto.OpDataNodeHeartbeat && t.OpCode != proto.OpLcNodeHeartbeat && t.OpCode != proto.OpFlashNodeHeartbeat {
log.LogDebugf("action[DelTask] delete task[%v]", t.ToString())
}
delete(sender.TaskMap, t.ID)

37
master/flash_manual_task.go
View File

@ -344,43 +344,6 @@ func (fltMgr *flashManualTaskManager) findMatchTasks(vol string, tid string) []*
return rets
}
func (c *Cluster) handleFlashNodeTaskResponse(nodeAddr string, task *proto.AdminTask) {
if task == nil {
log.LogInfof("flash action[handleFlashNodeTaskResponse] receive addr[%v] task response, but task is nil", nodeAddr)
return
}
log.LogInfof("flash action[handleFlashNodeTaskResponse] receive addr[%v] task: %v", nodeAddr, task.ToString())
var (
err error
flashNode *FlashNode
)
if flashNode, err = c.peekFlashNode(nodeAddr); err != nil {
goto errHandler
}
flashNode.TaskManager.DelTask(task)
if err = unmarshalTaskResponse(task); err != nil {
goto errHandler
}
switch task.OpCode {
case proto.OpFlashNodeScan:
response := task.Response.(*proto.FlashNodeManualTaskResponse)
err = c.handleFlashNodeScanResp(task.OperatorAddr, response)
default:
err = fmt.Errorf(fmt.Sprintf("flash unknown operate code %v", task.OpCode))
goto errHandler
}
if err != nil {
goto errHandler
}
return
errHandler:
log.LogWarnf("flash handleFlashNodeTaskResponse failed, task: %v, err: %v", task.ToString(), err)
}
func (c *Cluster) handleFlashNodeScanResp(nodeAddr string, resp *proto.FlashNodeManualTaskResponse) (err error) {
log.LogDebugf("action[handleFlashNodeScanResp] flashNode[%v] task[%v] Enter", nodeAddr, resp.ID)
defer func() {

83
master/flash_node.go
View File

@ -28,6 +28,7 @@ import (
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/sdk/httpclient"
"github.com/cubefs/cubefs/util"
"github.com/cubefs/cubefs/util/auditlog"
"github.com/cubefs/cubefs/util/errors"
"github.com/cubefs/cubefs/util/exporter"
"github.com/cubefs/cubefs/util/log"
@ -127,9 +128,7 @@ func (flashNode *FlashNode) updateFlashNodeStatHeartbeat(resp *proto.FlashNodeHe
flashNode.Unlock()
}
// TODO: sync with proto.FlashNodeHeartbeatResponse.
func (c *Cluster) syncFlashNodeHeartbeatTasks(tasks []*proto.AdminTask) {
var packet *proto.Packet
func (c *Cluster) addFlashNodeHeartbeatTasks(tasks []*proto.AdminTask) {
for _, t := range tasks {
if t == nil {
continue
@ -139,20 +138,7 @@ func (c *Cluster) syncFlashNodeHeartbeatTasks(tasks []*proto.AdminTask) {
log.LogWarn(fmt.Sprintf("action[syncFlashNodeHeartbeatTasks],nodeAddr:%v,taskID:%v,err:%v", t.OperatorAddr, t.ID, err.Error()))
continue
}
if packet, err = node.TaskManager.syncSendAdminTask(t); err != nil {
log.LogError(fmt.Sprintf("action[syncFlashNodeHeartbeatTasks],nodeAddr:%v,taskID:%v,err:%v", t.OperatorAddr, t.ID, err.Error()))
continue
}
node.setActive()
resp := &proto.FlashNodeHeartbeatResponse{}
err = json.Unmarshal(packet.Data, resp)
if err != nil {
log.LogErrorf("Failed to unmarshal response: %v", err)
continue
}
node.updateFlashNodeStatHeartbeat(resp)
c.handleFlashNodeHeartbeatResp(node, resp)
node.TaskManager.AddTask(t)
}
}
@ -173,7 +159,7 @@ func (c *Cluster) syncFlashNodeSetIOLimitTasks(tasks []*proto.AdminTask) {
}
}
func (c *Cluster) handleFlashNodeHeartbeatResp(flashNode *FlashNode, resp *proto.FlashNodeHeartbeatResponse) {
func (c *Cluster) handleManualTaskProcessing(flashNode *FlashNode, resp *proto.FlashNodeHeartbeatResponse) {
for _, taskRsp := range resp.ManualScanningTasks {
manualTask, ok := c.flashManMgr.LoadManualTaskById(taskRsp.ID)
if !ok {
@ -206,12 +192,14 @@ func (c *Cluster) checkFlashNodeHeartbeat() {
tasks = append(tasks, task)
return true
})
go c.syncFlashNodeHeartbeatTasks(tasks)
c.addFlashNodeHeartbeatTasks(tasks)
}
func (flashNode *FlashNode) checkLiveliness() {
flashNode.Lock()
if time.Since(flashNode.ReportTime) > _defaultNodeTimeoutDuration {
msg := fmt.Sprintf("flashnode[%v] heartbeat lost, last heartbeat time %v", flashNode.Addr, flashNode.ReportTime)
auditlog.LogMasterOp("checkLiveliness", msg, nil)
flashNode.IsActive = false
}
flashNode.Unlock()
@ -678,6 +666,63 @@ func (m *Server) handleFlashNodeTaskResponse(w http.ResponseWriter, r *http.Requ
m.cluster.handleFlashNodeTaskResponse(tr.OperatorAddr, tr)
}
func (c *Cluster) handleFlashNodeTaskResponse(nodeAddr string, task *proto.AdminTask) {
if task == nil {
log.LogInfof("flash action[handleFlashNodeTaskResponse] receive addr[%v] task response, but task is nil", nodeAddr)
return
}
log.LogInfof("flash action[handleFlashNodeTaskResponse] receive addr[%v] task: %v", nodeAddr, task.ToString())
var (
err error
flashNode *FlashNode
)
if flashNode, err = c.peekFlashNode(nodeAddr); err != nil {
goto errHandler
}
flashNode.TaskManager.DelTask(task)
if err = unmarshalTaskResponse(task); err != nil {
goto errHandler
}
switch task.OpCode {
case proto.OpFlashNodeScan:
response := task.Response.(*proto.FlashNodeManualTaskResponse)
err = c.handleFlashNodeScanResp(task.OperatorAddr, response)
case proto.OpFlashNodeHeartbeat:
response := task.Response.(*proto.FlashNodeHeartbeatResponse)
err = c.handleFlashNodeHeartbeatResp(task.OperatorAddr, response)
default:
err = fmt.Errorf(fmt.Sprintf("flash unknown operate code %v", task.OpCode))
goto errHandler
}
if err != nil {
goto errHandler
}
return
errHandler:
log.LogWarnf("flash handleFlashNodeTaskResponse failed, task: %v, err: %v", task.ToString(), err)
}
func (c *Cluster) handleFlashNodeHeartbeatResp(nodeAddr string, resp *proto.FlashNodeHeartbeatResponse) (err error) {
if resp.Status != proto.TaskSucceeds {
Warn(c.Name, fmt.Sprintf("action[handleFlashNodeHeartbeatResp] clusterID[%v] flashNode[%v] heartbeat task failed, err[%v]",
c.Name, nodeAddr, resp.Result))
return
}
var node *FlashNode
if node, err = c.peekFlashNode(nodeAddr); err != nil {
log.LogErrorf("action[handleFlashNodeHeartbeatResp], flashNode[%v], heartbeat error: %v", nodeAddr, err.Error())
return
}
node.setActive()
node.updateFlashNodeStatHeartbeat(resp)
c.handleManualTaskProcessing(node, resp)
return
}
func (c *Cluster) updateFlashNode(flashNode *FlashNode, enable bool) (err error) {
flashNode.Lock()
defer flashNode.Unlock()

19
master/mocktest/flash_server.go
View File

@ -117,9 +117,26 @@ func (mfs *MockFlashServer) serveConn(rc net.Conn) {
switch req.Opcode {
case proto.OpFlashNodeHeartbeat:
err = responseAckOKToMaster(conn, req, nil)
err = mfs.handleHeartbeats(conn, req, adminTask)
Printf("[mocktest] flashnode [%s] report heartbeat to master err:%v\n", mfs.TCPAddr, err)
default:
Printf("[mocktest] flashnode unknown code [%d]\n", req.Opcode)
}
}
func (mfs *MockFlashServer) handleHeartbeats(conn net.Conn, pkg *proto.Packet, task *proto.AdminTask) (err error) {
if err = responseAckOKToMaster(conn, pkg, nil); err != nil {
return
}
resp := &proto.FlashNodeHeartbeatResponse{}
resp.Stat = make([]*proto.FlashNodeDiskCacheStat, 0)
resp.LimiterStatus = &proto.FlashNodeLimiterStatusInfo{}
resp.FlashNodeTaskCountLimit = 8
resp.ManualScanningTasks = make(map[string]*proto.FlashNodeManualTaskResponse)
resp.Status = proto.TaskSucceeds
task.Response = resp
if err = mfs.mc.NodeAPI().ResponseFlashNodeTask(task); err != nil {
return
}
return
}

2
proto/admin_task.go
View File

@ -118,7 +118,7 @@ func (t *AdminTask) isCreateTask() bool {
// IsHeartbeatTask returns if the task is a heartbeat task.
func (t *AdminTask) IsHeartbeatTask() bool {
return t.OpCode == OpDataNodeHeartbeat || t.OpCode == OpMetaNodeHeartbeat || t.OpCode == OpLcNodeHeartbeat
return t.OpCode == OpDataNodeHeartbeat || t.OpCode == OpMetaNodeHeartbeat || t.OpCode == OpLcNodeHeartbeat || t.OpCode == OpFlashNodeHeartbeat
}
// NewAdminTask returns a new adminTask.