add mongo

Makefile

@@ -33,4 +33,4 @@ demoworker:
 clean:
 	rm -fr build/*
 docker:
-	docker build -t manager:${TAG} .
+	docker build --no-cache -t manager:${TAG} .

config.toml

@@ -9,6 +9,7 @@ worker_signature_expired_time=1
 
 [mongodb]
 url="mongodb://0.0.0.0:27017"
+database="ai"
 user="admin"
 password="admin"
 

config/Config.go

@@ -30,9 +30,10 @@ type RedisConfig struct {
 }
 
 type MongoDbConfig struct {
-	Url    string `json:"url" toml:"url"`
-	User   string `json:"user" toml:"user"`
-	Passwd string `json:"password" toml:"password"`
+	Url      string `json:"url" toml:"url"`
+	Database string `json:"database" toml:"database"`
+	User     string `json:"user" toml:"user"`
+	Passwd   string `json:"password" toml:"password"`
 }
 
 type TickerConfig struct {

server/node.go

@@ -200,3 +200,7 @@ func (n *Node) PayForFee(uid int64, fee int64) error {
 func (n *Node) Cache() *cachedata.CacheData {
 	return n.cache
 }
+
+func (n *Node) Mongo() *mongo.Client {
+	return n.mogo
+}

server/workerManager.go

@@ -19,6 +19,7 @@ import (
 	"github.com/odysseus/service-registry/registry"
 	"github.com/redis/go-redis/v9"
 	log "github.com/sirupsen/logrus"
+	"go.mongodb.org/mongo-driver/mongo"
 	"golang.org/x/crypto/sha3"
 	"math/big"
 	"regexp"
@@ -54,6 +55,7 @@ type NodeInterface interface {
 	Sign(hash []byte) ([]byte, error)
 	Cache() *cachedata.CacheData
 	PayForFee(uid int64, fee int64) error
+	Mongo() *mongo.Client
 }
 
 type WorkerManager struct {
@@ -68,8 +70,8 @@ type WorkerManager struct {
 	wkRwLock   sync.RWMutex
 	quit       chan struct{}
 
-	node NodeInterface
-	std  *standardlib.StandardTasks
+	node                    NodeInterface
+	std                     *standardlib.StandardTasks
 	workerInfoOperator      *operator.WorkerInfoOperator
 	workerInstalledOperator *operator.WorkerInstalledOperator
 	workerRunningOperator   *operator.WorkerRunningOperator
@@ -77,14 +79,17 @@ type WorkerManager struct {
 
 func NewWorkerManager(rdb *redis.Client, node NodeInterface) *WorkerManager {
 	return &WorkerManager{
-		heartBeat: make(map[int64]int64),
-		workerReg: make(map[int64]*registry.Registry),
-		workers:   make(map[int64]*Worker),
-		workid:    make(map[string]*Worker),
-		quit:      make(chan struct{}),
-		rdb:       rdb,
-		node:      node,
-		std:       standardlib.NewStandardTasks(),
+		heartBeat:               make(map[int64]int64),
+		workerReg:               make(map[int64]*registry.Registry),
+		workers:                 make(map[int64]*Worker),
+		workid:                  make(map[string]*Worker),
+		quit:                    make(chan struct{}),
+		rdb:                     rdb,
+		node:                    node,
+		std:                     standardlib.NewStandardTasks(),
+		workerInfoOperator:      operator.NewDBWorker(node.Mongo(), config.GetConfig().Mongo.Database),
+		workerInstalledOperator: operator.NewDBWorkerInstalled(node.Mongo(), config.GetConfig().Mongo.Database),
+		workerRunningOperator:   operator.NewDBWorkerRunning(node.Mongo(), config.GetConfig().Mongo.Database),
 	}
 }
 

server/workerstatu.go

@@ -7,6 +7,8 @@ import (
 	"fmt"
 	"github.com/gomodule/redigo/redis"
 	"github.com/google/uuid"
+	"github.com/odysseus/mogo/operator"
+	"github.com/odysseus/mogo/types"
 	"github.com/odysseus/nodemanager/config"
 	log "github.com/sirupsen/logrus"
 	"strconv"
@@ -76,6 +78,16 @@ func (wm *WorkerManager) AddWorkerFirst(worker *Worker) error {
 			}
 		}
 	}
+	_, err := wm.workerInfoOperator.InsertWorker(context.Background(), &operator.WorkerInfo{
+		WorkerId: worker.WorkerAccount().String(),
+		NodeInfo: types.PbToNodeInfo(worker.info.nodeInfo.Info),
+		Models:   types.PbToModelInfo(worker.info.nodeInfo.Models),
+		Hardware: types.PbToHardwareInfo(worker.info.nodeInfo.Hardware),
+	})
+	if err != nil {
+		log.WithError(err).Error("insert worker info failed")
+		return err
+	}
 
 	return nil
 }
@@ -185,6 +197,11 @@ func (wm *WorkerManager) InActiveWorker(worker *Worker) {
 		wm.rdb.Del(context.Background(), workerBootedResourceInfoKey(worker))
 		if worker.info.nodeInfo != nil {
 			wm.rmWorkerFromSets(worker, worker.info.nodeInfo.Info.BenefitAddress)
+			n, err := wm.workerInfoOperator.DeleteByWorkerId(context.Background(), worker.WorkerAccount().String())
+			if err != nil {
+				log.WithError(err).Error("delete worker info failed")
+			}
+			log.Debugf("delete worker info count %d", n)
 		}
 	}
 }