add nm check

config.toml

 endpoint="127.0.0.1:10002"
 metrics_port = 28012
 routines = 1
+max_nm_update_ex = 40
 
 [redis]
 addr="127.0.0.1:6379"

config/Config.go

@@ -26,12 +26,13 @@ type MysqlConfig struct {
 }
 
 type Config struct {
-	Endpoint   string      `json:"endpoint" toml:"endpoint"`
-	MetricPort int         `json:"metrics_port" toml:"metrics_port"`
-	Routines   int         `json:"routines" toml:"routines"`
-	Redis      RedisConfig `json:"redis" toml:"redis"`
-	Kafka      KafkaConfig `json:"kafka" toml:"kafka"`
-	DbConfig   MysqlConfig `json:"mysql" toml:"mysql"`
+	Endpoint      string      `json:"endpoint" toml:"endpoint"`
+	MetricPort    int         `json:"metrics_port" toml:"metrics_port"`
+	Routines      int         `json:"routines" toml:"routines"`
+	MaxNmUpdateEx int         `json:"max_nm_update_ex" toml:"max_nm_update_ex"`
+	Redis         RedisConfig `json:"redis" toml:"redis"`
+	Kafka         KafkaConfig `json:"kafka" toml:"kafka"`
+	DbConfig      MysqlConfig `json:"mysql" toml:"mysql"`
 }
 
 var _cfg *Config = nil

server/nmstatus.go

+package server
+
+import (
+	"context"
+	"github.com/odysseus/scheduler/config"
+	"github.com/redis/go-redis/v9"
+	"time"
+)
+
+type NMRegistryInfo struct {
+	Pubkey    string `redis:"pubkey"`
+	Timestamp int64  `redis:"timestamp"`
+	Endpoint  string `redis:"endpoint"`
+	Addr      string `redis:"addr"`
+}
+
+// managerUpdateTm got nm latest update time.
+func managerLatestUpdateTm(rdb *redis.Client, manager string) int64 {
+	var invalid int64 = 1000000000
+	allnm, err := allNodeManager(rdb)
+	if err != nil {
+		return invalid
+	}
+	if _, ok := allnm[manager]; ok {
+		return allnm[manager].Timestamp - time.Now().Unix()
+	}
+	return invalid
+
+}
+
+func allNodeManager(rdb *redis.Client) (map[string]NMRegistryInfo, error) {
+	var ret = make(map[string]NMRegistryInfo)
+	var tsExpired = 100
+	keys, err := rdb.Keys(context.Background(), config.NODE_MANAGER_SET+"*").Result()
+	if err != nil {
+		return nil, err
+	}
+	for _, key := range keys {
+		res := rdb.HGetAll(context.Background(), key)
+		var info NMRegistryInfo
+		if err := res.Scan(&info); err != nil {
+			continue
+		}
+		if time.Now().Unix()-info.Timestamp > int64(tsExpired) {
+			// heart beat expired, ignore this nodemanager
+			continue
+		}
+		ret[info.Endpoint] = info
+	}
+	return ret, nil
+}

server/worker.go

@@ -71,13 +71,18 @@ func newManagerClient(endpoint string) (omanager.NodeManagerServiceClient, error
 	return manager, nil
 
 }
-func DispatchTask(worker Worker, task *odysseus.TaskContent) error {
+func DispatchTask(rdb *redis.Client, worker Worker, task *odysseus.TaskContent) error {
 	l := log.WithField("task-id", task.TaskId)
 	l.WithFields(log.Fields{
 		"worker":      worker.workerid,
 		"managerList": worker.managers,
 	}).Debug("dispatch task to worker")
 	for _, manager := range worker.managers {
+		if latest := managerLatestUpdateTm(rdb, manager); latest > int64(config.GetConfig().MaxNmUpdateEx) {
+			l.WithField("manager", manager).Warn("ignore the manager because of not update tm")
+			continue
+		}
+
 		client, err := newManagerClient(manager)
 		if err != nil {
 			l.WithFields(log.Fields{