msg_handler.go

package nm

import (
	"example.com/m/conf"
	"example.com/m/log"
	"example.com/m/models"
	"example.com/m/operate"
	"example.com/m/utils"
	"example.com/m/validator"
	"fmt"
	nodeManagerV1 "github.com/odysseus/odysseus-protocol/gen/proto/go/nodemanager/v1"
	"time"
)

type NodeManagerHandler struct {
	nodeManager   *models.NodeManagerClient
	worker        nodeManagerV1.NodeManagerService_RegisterWorkerClient
	msgRespWorker *RespMsgWorker
	taskMsgWorker *TaskWorker
}

func NewNodeManagerHandler(nodeManager *models.NodeManagerClient, worker nodeManagerV1.NodeManagerService_RegisterWorkerClient, msgRespWorker *RespMsgWorker, taskMsgWorker *TaskWorker) *NodeManagerHandler {
	return &NodeManagerHandler{
		nodeManager:   nodeManager,
		worker:        worker,
		msgRespWorker: msgRespWorker,
		taskMsgWorker: taskMsgWorker,
	}
}

func (n *NodeManagerHandler) DistributionMsgWorker(nodeManagerMsgChan chan *nodeManagerV1.ManagerMessage, proofWorker *validator.ProofWorker) {
	for {
		select {
		case rev := <-nodeManagerMsgChan:
			{
				if !n.nodeManager.Status {
					log.Warn("handlerMsg -> node manager is not running")
					return
				}
				heartbeatReq := rev.GetHeartbeatRequest()
				if heartbeatReq != nil {
					n.nodeManager.UpdateLastHeartTime(time.Now())
					params := utils.BuildParams(heartbeatReq.Timestamp)
					n.msgRespWorker.RegisterMsgResp(n.nodeManager, n.worker, HeartbeatResp, params)
					log.Info("-------------Heart beat req:-------------", heartbeatReq)
					continue
				}

				taskMsg := rev.GetPushTaskMessage()
				if taskMsg != nil {
					go func(msgRespWorker *RespMsgWorker, taskMsgWorker *TaskWorker, taskMsg *nodeManagerV1.PushTaskMessage) {
						isCanExecute, bootUpTime, queueWaitTime, executeTime := taskMsgWorker.GetAckResp(taskMsg)
						ackParams := utils.BuildParams(taskMsg.TaskId, isCanExecute, bootUpTime, queueWaitTime, executeTime)
						msgRespWorker.RegisterMsgResp(n.nodeManager, n.worker, RespTaskAck, ackParams)
						if !isCanExecute {
							log.WithField("taskId", taskMsg.TaskId).Error("The task is not executed")
							return
						}
						taskMsgWorker.Wg.Add(1)
						taskMsgWorker.TaskMsg <- taskMsg
						taskMsgWorker.Wg.Wait()
						taskExecResInterface, _ := taskMsgWorker.LruCache.Get(taskMsg.TaskId)
						//log.WithField("result", taskExecResInterface).Info("lru cache get task result")
						taskExecRes := &models.TaskResult{
							TaskHttpStatusCode: 200,
							TaskRespBody:       nil,
							TaskHttpHeaders:    nil,
							TaskIsSuccess:      false,
							TaskExecTime:       0,
							TaskExecError:      "",
						}
						if taskExecResInterface != nil {
							taskExecRes = taskExecResInterface.(*models.TaskResult)
						}
						isSuccess := taskExecRes.TaskIsSuccess
						containerSign := make([]byte, 0)
						if taskExecRes.TaskRespBody != nil {
							containerSign = taskMsgWorker.DockerOp.GetContainerSign(taskMsg, taskExecRes.TaskRespBody)
							if containerSign == nil || len(containerSign) == 0 {
								log.Error("Container signing failed................")
								isSuccess = false
								taskExecRes.TaskExecError = fmt.Sprintf("%s-%s", "Container sign failed", taskExecRes.TaskExecError)
							}
						} else {
							isSuccess = false
							taskExecRes.TaskExecError = fmt.Sprintf("worker:%s-%s-%s", conf.GetConfig().SignPublicAddress.Hex(), "Task exec error", taskExecRes.TaskExecError)
						}
						_, _, minerSign := taskMsgWorker.GetMinerSign(taskMsg, taskExecRes.TaskRespBody)
						taskResultParams := utils.BuildParams(taskMsg.TaskId, containerSign, minerSign, taskExecRes, isSuccess)
						//taskMsgWorker.LruCache.Add(taskMsg.TaskId+models.TaskType, taskMsg.TaskType)
						//taskMsgWorker.LruCache.Add(taskMsg.TaskId+models.ContainerSign, containerSign)
						//taskMsgWorker.LruCache.Add(taskMsg.TaskId+models.MinerSign, minerSign)
						//taskMsgWorker.LruCache.Add(taskMsg.TaskId+models.ReqHash, reqHash)
						//taskMsgWorker.LruCache.Add(taskMsg.TaskId+models.RespHash, respHash)
						msgRespWorker.RegisterMsgResp(n.nodeManager, n.worker, SubmitResultResp, taskResultParams)
						log.Info("--------------taskMsg--------------:", taskMsg)
					}(n.msgRespWorker, n.taskMsgWorker, taskMsg)
					continue
				}

				nmResultMsg := rev.GetProofTaskResult()
				if nmResultMsg != nil {
					//containerSign, _ := taskMsgWorker.LruCache.Get(nmResultMsg.TaskId + models.ContainerSign)
					//if !ok {
					//	log.Error("taskMsgWorker.LruCache.Get failed: ", nmSignMsg.TaskUuid+models.ContainerSign)
					//}
					//minerSign, _ := taskMsgWorker.LruCache.Get(nmResultMsg.TaskId + models.MinerSign)
					//if !ok {
					//	log.Error("taskMsgWorker.LruCache.Get failed: ", nmSignMsg.TaskUuid+models.MinerSign)
					//}
					//reqHash, _ := taskMsgWorker.LruCache.Get(nmResultMsg.TaskId + models.ReqHash)
					//if !ok {
					//	log.Error("taskMsgWorker.LruCache.Get failed: ", nmSignMsg.TaskUuid+models.ReqHash)
					//}
					//respHash, _ := taskMsgWorker.LruCache.Get(nmResultMsg.TaskId + models.RespHash)
					//if !ok {
					//	log.Error("taskMsgWorker.LruCache.Get failed: ", nmSignMsg.TaskUuid+models.RespHash)
					//}
					//taskType, _ := taskMsgWorker.LruCache.Get(nmResultMsg.TaskId + models.TaskType)
					//proofWorker.ProductProof(nmResultMsg, taskType.(uint64), reqHash.([]byte), respHash.([]byte), containerSign.([]byte), minerSign.([]byte))
					log.WithField("proof", nmResultMsg).Info("Output proof task result")
					continue
				}

				deviceUsageMsg := rev.GetDeviceUsage()
				if deviceUsageMsg != nil {
					n.msgRespWorker.RegisterMsgResp(n.nodeManager, n.worker, DeviceUsageResp, nil)
					log.Info(deviceUsageMsg)
					continue
				}

				nodeInfoMsg := rev.GetNodeInfoRequest()
				if nodeInfoMsg != nil {
					n.msgRespWorker.RegisterMsgResp(n.nodeManager, n.worker, NodeInfoResp, nil)
					log.Info(nodeInfoMsg)
					continue
				}

				statusReqMsg := rev.GetStatusRequest()
				if statusReqMsg != nil {
					n.msgRespWorker.RegisterMsgResp(n.nodeManager, n.worker, StatusResp, nil)
					log.Info(statusReqMsg)
					continue
				}

				goodByeMsg := rev.GetGoodbyeMessage()
				if goodByeMsg != nil {
					reason := goodByeMsg.GetReason()
					log.Infof("Server endpoint:%s , good bye reason : %s", n.nodeManager.Endpoint, reason)
					n.nodeManager.UpdateStatus(false)
					log.Warn("Update nm status is false")
					continue
				}
			}
		}
	}
}

func (n *NodeManagerHandler) MonitorStandardTaskWorker() {
	//ticker := time.NewTicker(time.Second * 30)
	ticker := time.NewTicker(time.Minute * 5)
	for {
		select {
		case <-ticker.C:
			{
				if n.taskMsgWorker.IsExecStandardTask {
					continue
				}
				if !n.taskMsgWorker.IsExecAiTask {
					n.msgRespWorker.RegisterMsgResp(n.nodeManager, n.worker, FetchStandardTaskResp, nil)
					break
				}
			}
		}
	}
}

func (n *NodeManagerHandler) ReportResourceMap(dockerOp *operate.DockerOp) {
	ticker := time.NewTicker(time.Second * 1)
	for {
		select {
		case <-ticker.C:
			if len(dockerOp.ReportModelIds) > 0 {
				bootUpModelIds := make([]uint64, 0)
				containers := dockerOp.ListContainer()
				if containers != nil && len(containers) > 0 {
					for _, container := range containers {
						if container.State == "running" {
							taskId := dockerOp.BootUpModelId[container.Image]
							if taskId != 0 {
								bootUpModelIds = append(bootUpModelIds, taskId)
							}
						}
					}
				}
				params := utils.BuildParams(dockerOp.ReportModelIds, bootUpModelIds)
				n.msgRespWorker.RegisterMsgResp(n.nodeManager, n.worker, SubmitResourceMapRes, params)
				ticker.Reset(time.Minute * 10)
			}
		}
	}
}