update nodemanager

config.toml

-remote_host="192.168.1.112"
-local_host="192.168.1.112"
+remote_host="192.168.2.30"
+local_host="192.168.2.30"
 port=10001
 metrics_port = 28010
 private_key = "E671C143A110C239B563F702E9F4017CA6B2B2912F675EED9AA4FED684EB30CC"
-standard_task_file = "/Users/luxq/work/wuban/nodemanager/standardtask.json"
+standard_task_file = "standardtask.json"
 
 [redis]
 addr="127.0.0.1:6379"
@@ -11,11 +11,16 @@ password="123456"
 db=0
 
 [mysql]
-host="192.168.1.211"
+#host="192.168.1.211"
+#port=3306
+#user="root"
+#password="12345678"
+#database="liuxuzhong"
+host="127.0.0.1"
 port=3306
-user="root"
-password="12345678"
-database="liuxuzhong"
+user="ai"
+password="RFnnKHRar5xk7TEF"
+database="ai"
 
 [kafka]
 brokers="127.0.0.1:9092"
@@ -28,3 +33,4 @@ heart_beat = 10
 status_ticker = 10
 device_info_ticker = 120
 device_usage_ticker = 100
+worker_task_expire_ticker = 60

config/Config.go

@@ -3,6 +3,7 @@ package config
 import (
 	"fmt"
 	"github.com/BurntSushi/toml"
+	"github.com/odysseus/nodemanager/utils"
 	log "github.com/sirupsen/logrus"
 	"io/ioutil"
 )
@@ -29,10 +30,11 @@ type RedisConfig struct {
 }
 
 type TickerConfig struct {
-	HeartBeat         int `json:"heart_beat" toml:"heart_beat"`
-	StatusTicker      int `json:"status_ticker" toml:"status_ticker"`
-	DeviceInfoTicker  int `json:"device_info_ticker" toml:"device_info_ticker"`
-	DeviceUsageTicker int `json:"device_usage_ticker" toml:"device_usage_ticker"`
+	HeartBeat              int `json:"heart_beat" toml:"heart_beat"`
+	StatusTicker           int `json:"status_ticker" toml:"status_ticker"`
+	DeviceInfoTicker       int `json:"device_info_ticker" toml:"device_info_ticker"`
+	DeviceUsageTicker      int `json:"device_usage_ticker" toml:"device_usage_ticker"`
+	WorkerTaskExpireTicker int `json:"worker_task_expire_ticker" toml:"worker_task_expire_ticker"`
 }
 
 type Config struct {
@@ -64,6 +66,13 @@ func ParseConfig(path string) (*Config, error) {
 		log.Error("unmarshal config failed", "err", err)
 		panic(err)
 	}
+	if _cfg.RemoteHost == "" {
+		ip, err := utils.PublicIp()
+		if err != nil {
+			log.Error("get public ip failed", "err", err)
+		}
+		_cfg.RemoteHost = ip
+	}
 	return _cfg, nil
 }
 

config/constant.go

@@ -10,4 +10,5 @@ const (
 	WORKER_DEVICE_STATUS_PREFIX = "worker_device_status_"
 	WORKER_USAGE_INFO_PREFIX    = "worker_usage_info_"
 	WORKER_RESOURCE_INFO_PREFIX = "worker_resource_info_"
+	WORKER_LAST_TASK_TM_PREFIX  = "worker_last_task_tm_"
 )

docker/docker-compose.yml

@@ -7,7 +7,7 @@ services:
         mysql:
                 container_name: "ai-mysql"
                 ports :
-                        - "33306:3306"
+                        - "3306:3306"
                 environment:                        
                         - MYSQL_ROOT_PASSWORD=RFnnKHRar5xk7TEF
                         - MYSQL_USER=ai

server/workerManager.go

@@ -28,6 +28,7 @@ var (
 	Succeed                = errors.New("succeed")
 	ErrWorkerExist         = errors.New("worker exist")
 	ErrHeartBeatExpired    = errors.New("worker heartbeat expired")
+	ErrLongtimeNoTask      = errors.New("worker long time no task")
 	ErrInvalidMessageValue = errors.New("invalid message value")
 )
 
@@ -368,6 +369,9 @@ func (wm *WorkerManager) manageWorker(worker *Worker) error {
 				if err == nil {
 					// add task to cache.
 					worker.recentTask.Add(task.TaskId, task)
+					if e := wm.setWorkerLastTaskTime(worker, time.Now().Unix()); e != nil {
+						log.WithField("worker", worker.uuid).WithError(e).Error("set worker last task time failed")
+					}
 				}
 				log.WithField("worker", worker.uuid).Info("dispatch task to worker")
 
@@ -440,6 +444,19 @@ func (wm *WorkerManager) handleWorkerMsg(worker *Worker) {
 				worker.quit <- ErrHeartBeatExpired
 				return
 			}
+			if lastTaskTm, err := wm.getWorkerLastTaskTime(worker); err != nil {
+				log.WithError(err).Error("get worker last task time failed")
+			} else {
+				expire := config.GetConfig().Tickers.WorkerTaskExpireTicker
+				if expire <= 0 {
+					expire = 60 // default value
+				}
+				if time.Now().Unix()-lastTaskTm > int64(expire) {
+					log.WithField("worker-uuid", worker.uuid).Error("worker last task time expired")
+					worker.quit <- ErrLongtimeNoTask
+					return
+				}
+			}
 		default:
 			wmsg, err := worker.stream.Recv()
 			if err != nil {

server/workerstatu.go

@@ -93,6 +93,7 @@ func (wm *WorkerManager) AddWorkerToQueue(worker *Worker) {
 				}
 			} else {
 				// else if nmlist is not empty, clear and add self to it.
+				worker.nonce = nonce
 				wm.rdb.Del(context.Background(), workerStatusKey(worker))
 				wm.UpdateWorkerActive(worker)
 			}
@@ -196,6 +197,14 @@ func (wm *WorkerManager) getWorkerSets(benefit string) ([]string, error) {
 	return list, err
 }
 
+func (wm *WorkerManager) getWorkerLastTaskTime(worker *Worker) (int64, error) {
+	return wm.rdb.Get(context.Background(), workerLastTaskTmKey(worker)).Int64()
+}
+
+func (wm *WorkerManager) setWorkerLastTaskTime(worker *Worker, tm int64) error {
+	return wm.rdb.Set(context.Background(), workerLastTaskTmKey(worker), tm, 0).Err()
+}
+
 func (wm *WorkerManager) checkWhiteList(worker *Worker, benefit string) error {
 	wh, err := wm.node.cache.GetWhWithAddr(benefit)
 	if err != nil {
@@ -253,6 +262,10 @@ func workerStatusKey(w *Worker) string {
 	return fmt.Sprintf("%s_%s", config.WORKER_STATUS_PREFIX, id)
 }
 
+func workerLastTaskTmKey(w *Worker) string {
+	return config.WORKER_LAST_TASK_TM_PREFIX + w.workerAddr
+}
+
 func workerId(w *Worker) string {
 	return fmt.Sprintf("%s_%d", w.workerAddr, w.nonce)
 }