144 lines
3.3 KiB
Go
144 lines
3.3 KiB
Go
package main
|
|
|
|
import (
|
|
"fmt"
|
|
"os"
|
|
"time"
|
|
|
|
_ "google.golang.org/grpc/balancer/grpclb"
|
|
|
|
"gitlink.org.cn/cloudream/common/pkgs/logger"
|
|
schglb "gitlink.org.cn/cloudream/scheduler/common/globals"
|
|
"gitlink.org.cn/cloudream/scheduler/common/pkgs/db"
|
|
mgrmq "gitlink.org.cn/cloudream/scheduler/common/pkgs/mq/manager"
|
|
"gitlink.org.cn/cloudream/scheduler/manager/internal/advisormgr"
|
|
"gitlink.org.cn/cloudream/scheduler/manager/internal/config"
|
|
"gitlink.org.cn/cloudream/scheduler/manager/internal/executormgr"
|
|
"gitlink.org.cn/cloudream/scheduler/manager/internal/jobmgr"
|
|
mqsvc "gitlink.org.cn/cloudream/scheduler/manager/internal/mq"
|
|
)
|
|
|
|
func main() {
|
|
err := config.Init()
|
|
if err != nil {
|
|
fmt.Printf("init config failed, err: %s", err.Error())
|
|
os.Exit(1)
|
|
}
|
|
|
|
err = logger.Init(&config.Cfg().Logger)
|
|
if err != nil {
|
|
fmt.Printf("init logger failed, err: %s", err.Error())
|
|
os.Exit(1)
|
|
}
|
|
|
|
schglb.InitMQPool(&config.Cfg().RabbitMQ)
|
|
schglb.InitCloudreamStoragePool(&config.Cfg().CloudreamStorage)
|
|
schglb.InitRcloneConfig(config.Cfg().CDSRclone.CDSRcloneID, "")
|
|
|
|
exeMgr, err := executormgr.NewManager(time.Duration(config.Cfg().ReportTimeoutSecs) * time.Second)
|
|
if err != nil {
|
|
fmt.Printf("new executor manager: %s", err.Error())
|
|
os.Exit(1)
|
|
}
|
|
|
|
advMgr, err := advisormgr.NewManager(time.Duration(config.Cfg().ReportTimeoutSecs) * time.Second)
|
|
if err != nil {
|
|
fmt.Printf("new advisor manager: %s", err.Error())
|
|
os.Exit(1)
|
|
}
|
|
|
|
db, err := db.NewDB(&config.Cfg().DB)
|
|
if err != nil {
|
|
fmt.Printf("new db: %s", err.Error())
|
|
os.Exit(1)
|
|
}
|
|
|
|
nodeSvc := jobmgr.NewNodeService()
|
|
jobMgr, err := jobmgr.NewManager(exeMgr, advMgr, db, nodeSvc)
|
|
if err != nil {
|
|
fmt.Printf("new job manager: %s", err.Error())
|
|
os.Exit(1)
|
|
}
|
|
|
|
svc, err := mqsvc.NewService(jobMgr, exeMgr, advMgr, db)
|
|
if err != nil {
|
|
fmt.Printf("new service: %s", err.Error())
|
|
os.Exit(1)
|
|
}
|
|
|
|
mqSvr, err := mgrmq.NewServer(svc, &config.Cfg().RabbitMQ)
|
|
if err != nil {
|
|
logger.Fatalf("new manager mq server: %s", err.Error())
|
|
}
|
|
|
|
mqSvr.OnError(func(err error) {
|
|
logger.Warnf("manager server err: %s", err.Error())
|
|
})
|
|
|
|
// 启动服务
|
|
go serveJobManager(jobMgr)
|
|
|
|
//go serveExecutorManager(exeMgr)
|
|
go exeMgr.Serve()
|
|
|
|
go serveAdvisorManager(advMgr)
|
|
|
|
go serveMQServer(mqSvr)
|
|
|
|
forever := make(chan bool)
|
|
<-forever
|
|
}
|
|
|
|
func serveJobManager(mgr *jobmgr.Manager) {
|
|
logger.Info("start serving job manager")
|
|
|
|
err := mgr.Serve()
|
|
if err != nil {
|
|
logger.Errorf("job manager stopped with error: %s", err.Error())
|
|
}
|
|
|
|
logger.Info("job manager stopped")
|
|
}
|
|
|
|
func serveMQServer(server *mgrmq.Server) {
|
|
logger.Info("start serving mq server")
|
|
|
|
ch := server.Start()
|
|
if ch == nil {
|
|
logger.Errorf("RabbitMQ logEvent is nil")
|
|
os.Exit(1)
|
|
}
|
|
|
|
for {
|
|
val, err := ch.Receive()
|
|
if err != nil {
|
|
logger.Errorf("command server stopped with error: %s", err.Error())
|
|
break
|
|
}
|
|
|
|
switch val := val.(type) {
|
|
case error:
|
|
logger.Errorf("rabbitmq connect with error: %v", val)
|
|
case int:
|
|
if val == 1 {
|
|
break
|
|
}
|
|
}
|
|
}
|
|
logger.Info("command server stopped")
|
|
|
|
// TODO 仅简单结束了程序
|
|
os.Exit(1)
|
|
}
|
|
|
|
func serveAdvisorManager(mgr *advisormgr.Manager) {
|
|
logger.Info("start serving advisor manager")
|
|
|
|
err := mgr.Serve()
|
|
if err != nil {
|
|
logger.Errorf("advisor manager stopped with error: %s", err.Error())
|
|
}
|
|
|
|
logger.Info("advisor manager stopped")
|
|
}
|