add service registry

server/nmstatus.go

 package server
 
 import (
-	"context"
-	"github.com/odysseus/scheduler/config"
+	"encoding/json"
+	"github.com/odysseus/service-registry/common"
+	"github.com/odysseus/service-registry/query"
+	"github.com/odysseus/service-registry/registry"
 	"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()
+func (n *Node) allNodeManager() map[string]registry.RegistryInfo {
+	querier := query.NewQuery(registry.RedisConnParam{
+		Addr:     n.conf.Redis.Addr,
+		Password: n.conf.Redis.Password,
+		DbIndex:  n.conf.Redis.DbIndex,
+	})
+	result, err := querier.Select(common.SERVICE_NODE_MANAGER).List()
 	if err != nil {
-		return nil, err
+		return nil
 	}
-	for _, key := range keys {
-		res := rdb.HGetAll(context.Background(), key)
-		var info NMRegistryInfo
-		if err := res.Scan(&info); err != nil {
+	res := make(map[string]registry.RegistryInfo, 0)
+	for _, v := range result {
+		var info registry.RegistryInfo
+		var nminfo query.NodeManagerInfo
+		if err := json.Unmarshal([]byte(v), &info); err != nil {
 			continue
 		}
-		if time.Now().Unix()-info.Timestamp > int64(tsExpired) {
-			// heart beat expired, ignore this nodemanager
+		if err := json.Unmarshal([]byte(info.Detail), &nminfo); err != nil {
 			continue
+		} else {
+			res[nminfo.EndPoint] = info
 		}
-		ret[info.Endpoint] = info
 	}
-	return ret, nil
+	return res
+}
+
+// managerUpdateTm got nm latest update time.
+func (n *Node) managerLatestUpdateTm(rdb *redis.Client, manager_endpoint string) int64 {
+	var invalid int64 = 1000000000
+	allnm := n.allNodeManager()
+
+	if _, ok := allnm[manager_endpoint]; ok {
+		return allnm[manager_endpoint].Timestamp - time.Now().Unix()
+	}
+	return invalid
 }

server/node.go

@@ -32,21 +32,24 @@ type Node struct {
 	cache         *cachedata.CacheData
 	wg            sync.WaitGroup
 	status        string
+	reg           *registry.Registry
 }
 
 func (n *Node) ServiceType() common.ServiceType {
 	return common.SERVICE_SCHEDULER
 }
 
-func (n *Node) Endpoint() string {
+func (n *Node) Status() string {
+	return n.status
+}
+
+func (n *Node) Instance() string {
 	hname, _ := os.Hostname()
 	return fmt.Sprintf("%s:%d", hname, n.conf.Endpoint)
 }
 
 func (n *Node) DetailInfo() (json.RawMessage, error) {
 	info := query.SchedulerInfo{}
-	info.Status = n.status
-
 	return json.Marshal(info)
 }
 
@@ -102,12 +105,13 @@ func (n *Node) startAllTask() error {
 }
 
 func (n *Node) register() {
-	registry := registry.NewRegistry(registry.RedisConnParam{
+	reg := registry.NewRegistry(registry.RedisConnParam{
 		Addr:     n.conf.Redis.Addr,
 		Password: n.conf.Redis.Password,
 		DbIndex:  n.conf.Redis.DbIndex,
 	}, n)
-	registry.Start()
+	n.reg = reg
+	reg.Start()
 }
 
 func (n *Node) Loop(idx int) {
@@ -205,7 +209,7 @@ func (n *Node) Loop(idx int) {
 						l.WithError(err).Error("pop worker failed")
 						continue
 					}
-					err = DispatchTask(n.rdb, worker, task)
+					err = n.DispatchTask(n.rdb, worker, task)
 					if err != nil {
 						l.WithError(err).Error("dispatch task failed")
 						continue

server/worker.go

@@ -75,14 +75,14 @@ func newManagerClient(endpoint string) (omanager.NodeManagerServiceClient, error
 	return manager, nil
 
 }
-func DispatchTask(rdb *redis.Client, worker Worker, task *odysseus.TaskContent) error {
+func (n *Node) 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) {
+		if latest := n.managerLatestUpdateTm(rdb, manager); latest > int64(config.GetConfig().MaxNmUpdateEx) {
 			l.WithField("manager", manager).Warn("ignore the manager because of not update tm")
 			continue
 		}