Merge branch 'master' into test

conf/config.go

@@ -9,25 +9,26 @@ import (
 )
 
 type Config struct {
-	SignPrv              string
-	SignPub              string
-	DockerServer         string
-	BenefitAddress       string
-	HeartRespTimeMillis  int64
-	ExternalIp           string
-	SignPublicAddress    common.Address
-	SignPrivateKey       *ecdsa.PrivateKey
-	NmSeed               string  `json:"nm_seed"`
-	HeartRespTimeSecond  int64   `json:"heart_response"`
-	TaskValidatorTime    float64 `json:"task_validator_time"`
-	ContainerNum         int64   `json:"container_num"`
-	NodeManagerNum       int64   `json:"node_manager_num"`
-	ChainID              int64   `json:"chain_id"`
-	ApiUrl               string  `json:"api_url"`
-	ValidatorUrl         string  `json:"validator_url"`
-	OssUrl               string  `json:"oss_url"`
-	WaitLastTaskExecTime int64   `json:"wait_last_task_exec_time"`
-	OpSys                string  `json:"op_sys"`
+	SignPrv                  string
+	SignPub                  string
+	DockerServer             string
+	BenefitAddress           string
+	HeartRespTimeMillis      int64
+	ExternalIp               string
+	SignPublicAddress        common.Address
+	SignPrivateKey           *ecdsa.PrivateKey
+	NmSeed                   string  `json:"nm_seed"`
+	HeartRespTimeSecond      int64   `json:"heart_response"`
+	TaskValidatorTime        float64 `json:"task_validator_time"`
+	ContainerNum             int64   `json:"container_num"`
+	NodeManagerNum           int64   `json:"node_manager_num"`
+	ChainID                  int64   `json:"chain_id"`
+	ApiUrl                   string  `json:"api_url"`
+	ValidatorUrl             string  `json:"validator_url"`
+	OssUrl                   string  `json:"oss_url"`
+	WaitLastTaskExecTime     int64   `json:"wait_last_task_exec_time"`
+	OpSys                    string  `json:"op_sys"`
+	ReplicateImageNameSuffix string  `json:"replicate_image_name_suffix"`
 }
 
 var _cfg *Config = nil

config.json

@@ -9,5 +9,6 @@
   "validator_url": "18.167.203.17:20011",
   "oss_url": "https://tmp-file.aigic.ai/api/v1/upload",
   "wait_last_task_exec_time": 60,
-  "op_sys": "linux"
+  "op_sys": "linux",
+  "replicate_image_name_suffix": "docker.aigic.ai/ai"
 }
\ No newline at end of file

models/const.go

 package models
 
 const (
-	TaskType                 = "taskType"
-	ContainerSign            = "container"
-	MinerSign                = "miner"
-	ReqHash                  = "reqHash"
-	RespHash                 = "respHash"
-	ResultFileExpiresDB      = "expires"
-	ContentType              = "type"
-	RedirectCode             = 303
-	UseFileCache             = "Use-File-Cache"
-	UseRedirect              = "Use-Redirect"
-	Prefer                   = "Prefer"
-	Async                    = "respond-async"
-	MaxExecTime              = "MaxExecTime"
-	HealthCheckAPI           = "/health-check"
-	ReplicateImageNameSuffix = "docker.agicoin.ai/agicoin"
-	READY                    = "READY"
-	ZeroHost                 = "0.0.0.0"
-	ModelPublishStatusYes    = 1
-	ModelPublishStatusNo     = 2
-	DefaultMaxExecTime       = 300
-	DefaultTaskTimer         = 2
+	TaskType              = "taskType"
+	ContainerSign         = "container"
+	MinerSign             = "miner"
+	ReqHash               = "reqHash"
+	RespHash              = "respHash"
+	ResultFileExpiresDB   = "expires"
+	ContentType           = "type"
+	RedirectCode          = 303
+	UseFileCache          = "Use-File-Cache"
+	UseRedirect           = "Use-Redirect"
+	Prefer                = "Prefer"
+	Async                 = "respond-async"
+	MaxExecTime           = "MaxExecTime"
+	HealthCheckAPI        = "/health-check"
+	READY                 = "READY"
+	ZeroHost              = "0.0.0.0"
+	ModelPublishStatusYes = 1
+	ModelPublishStatusNo  = 2
+	DefaultMaxExecTime    = 300
+	DefaultTaskTimer      = 2
 )

nm/task_handler.go

@@ -197,25 +197,6 @@ func (t *TaskWorker) ComputeTaskHandler(taskMsg *nodeManagerV1.PushTaskMessage)
 		log.Info("Container ports:", internalPort)
 		log.WithField("ApiUrl", taskOp.taskCmd.ApiUrl).Info("The image is running")
 	}
-	reqContainerBody := bytes.NewReader(taskOp.taskParam.Body)
-	if len(taskOp.taskParam.Queries) > 0 {
-		queryString := utils.MatchContainerQueryString(taskOp.taskParam.Queries)
-		taskOp.taskCmd.ApiUrl = fmt.Sprintf("%s?%s", taskOp.taskCmd.ApiUrl, queryString)
-		log.WithField("ApiUrl", taskOp.taskCmd.ApiUrl).Info("The task param query str not empty")
-	}
-	taskOp.request, err = http.NewRequest("POST", taskOp.taskCmd.ApiUrl, reqContainerBody)
-	if err != nil {
-		log.WithField("error:", err).Error("New container request failed")
-		taskOp.taskExecResult.TaskExecError = fmt.Sprintf("%s,%s", "Http client new container request failed", err.Error())
-		t.ExecTaskIdIsSuccess.Store(taskMsg.TaskId, true)
-		return
-	}
-	taskOp.request.Header.Set("Content-Type", "application/json")
-	if err = taskOp.validateWebHook(); err != nil {
-		taskOp.taskExecResult.TaskExecError = fmt.Sprintf("%s", err.Error())
-		t.ExecTaskIdIsSuccess.Store(taskMsg.TaskId, true)
-		return
-	}
 	if err = taskOp.waitReqContainerOk(t.DockerOp); err != nil {
 		taskOp.taskExecResult.TaskExecError = fmt.Sprintf("%s", err.Error())
 		t.ExecTaskIdIsSuccess.Store(taskMsg.TaskId, true)
@@ -476,7 +457,7 @@ func (op *TaskOp) waitContainerRunning(handler *TaskWorker, imageId string) erro
 			if !running {
 				continue
 			}
-			if isMatch := strings.HasPrefix(op.taskCmd.ImageName, models.ReplicateImageNameSuffix); isMatch {
+			if isMatch := strings.HasPrefix(op.taskCmd.ImageName, conf.GetConfig().ReplicateImageNameSuffix); isMatch {
 				if isReqSuccess, err := op.checkContainerHealthy(internalIp, internalPort); err != nil {
 					log.WithField("err", err).Errorf("check container healthy failed")
 					return fmt.Errorf("%s-%s", "check container healthy failed", err.Error())
@@ -537,12 +518,25 @@ func (op *TaskOp) waitReqContainerOk(dockerOp *operate.DockerOp) error {
 				log.Errorf("%s", "The maximum execution time for this task has been exceeded")
 				return fmt.Errorf("%s", "The maximum execution time for this task has been exceeded")
 			}
-			newQuest := utils.CloneRequest(op.request)
-			post, err := op.httpClient.Do(newQuest)
+			reqContainerBody := bytes.NewReader(op.taskParam.Body)
+			if len(op.taskParam.Queries) > 0 {
+				queryString := utils.MatchContainerQueryString(op.taskParam.Queries)
+				op.taskCmd.ApiUrl = fmt.Sprintf("%s?%s", op.taskCmd.ApiUrl, queryString)
+				log.WithField("ApiUrl", op.taskCmd.ApiUrl).Info("The task param query str not empty")
+			}
+			op.request, err = http.NewRequest("POST", op.taskCmd.ApiUrl, reqContainerBody)
+			if err != nil {
+				log.WithField("error:", err).Error("New container request failed")
+				return fmt.Errorf("%s,%s", "Http client new container request failed", err.Error())
+			}
+			op.request.Header.Set("Content-Type", "application/json")
+			if err = op.validateWebHook(); err != nil {
+				return fmt.Errorf("%s", err.Error())
+			}
+			post, err := op.httpClient.Do(op.request)
 			if err != nil {
 				log.WithField("error:", err).Error("Http client post request container failed")
 				return fmt.Errorf("%s,%s", "Http client post request container failed", err.Error())
-				//continue
 			}
 			log.WithField("StatusCode", post.StatusCode).WithField("taskId", op.taskMsg.TaskId).Info("Exec task result")
 			if post.StatusCode == http.StatusOK {