pcm-coordinator/adaptor/PCM-HPC/PCM-AC/rpc/hpcac.go

package main

import (
	"PCM/adaptor/PCM-CORE/rpc/pcmcoreclient"
	"PCM/adaptor/PCM-HPC/PCM-AC/rpc/hpcAC"
	"PCM/adaptor/PCM-HPC/PCM-AC/rpc/internal/config"
	"PCM/adaptor/PCM-HPC/PCM-AC/rpc/internal/logic"
	"PCM/adaptor/PCM-HPC/PCM-AC/rpc/internal/server"
	"PCM/adaptor/PCM-HPC/PCM-AC/rpc/internal/svc"
	"PCM/common/tool"
	"context"
	"flag"
	"github.com/zeromicro/go-zero/core/conf"
	"github.com/zeromicro/go-zero/core/logx"
	"github.com/zeromicro/go-zero/core/service"
	"github.com/zeromicro/go-zero/zrpc"
	"google.golang.org/grpc"
	"google.golang.org/grpc/reflection"
)

var configFile = flag.String("f", "adaptor/PCM-HPC/PCM-AC/rpc/etc/hpcac.yaml", "the config file")

func main() {
	flag.Parse()

	var c config.Config
	conf.MustLoad(*configFile, &c)
	// start log component
	logx.MustSetup(c.LogConf)
	ctx := svc.NewServiceContext(c)
	ctx.Cron.Start()
	s := zrpc.MustNewServer(c.RpcServerConf, func(grpcServer *grpc.Server) {
		hpcAC.RegisterHpcACServer(grpcServer, server.NewHpcACServer(ctx))

		if c.Mode == service.DevMode || c.Mode == service.TestMode {
			reflection.Register(grpcServer)
		}
	})
	defer s.Stop()

	logx.Infof("Starting rpc server at %s...\n", c.ListenOn)
	initCron(ctx)
	s.Start()
}

func initCron(svc *svc.ServiceContext) {
	submitJobLogic := logic.NewSubmitJobLogic(context.Background(), svc)
	listLogic := logic.NewListJobLogic(context.Background(), svc)
	svc.Cron.AddFunc("*/5  *  *  *  *  ?", func() {
		syncInfoReq := pcmcoreclient.SyncInfoReq{
			Kind:        "hpc",
			ServiceName: "ac",
		}
		// 查询core端分发下来的任务列表
		infoList, err := queryCoreInfoList(svc)
		if err != nil {
			logx.Error(err)
			return
		}
		// 提交任务
		submitJob(infoList, submitJobLogic)
		// 查询运行中的任务列表同步信息
		listReq := hpcAC.ListJobReq{}
		listJob, err := listLogic.ListJob(&listReq)
		if err != nil {
			logx.Error(err)
			return
		}
		for index1, _ := range infoList.HpcInfoList {
			for index2, _ := range listJob.Jobs {
				if listJob.Jobs[index2].JobName == infoList.HpcInfoList[index1].Name {
					infoList.HpcInfoList[index1].StartTime = listJob.Jobs[index2].JobStartTime
					infoList.HpcInfoList[index1].RunningTime = int64(tool.RunTimeToSeconds(listJob.Jobs[index2].JobRunTime))
					if listJob.Jobs[index2].JobStatus == "statR" {
						infoList.HpcInfoList[index1].Status = "Running"
					}
					if listJob.Jobs[index2].JobStatus == "statC" {
						infoList.HpcInfoList[index1].Status = "Completed"
					}
				}
			}
		}
		// 同步信息到core端
		if len(infoList.HpcInfoList) != 0 {
			syncInfoReq.HpcInfoList = infoList.HpcInfoList
			svc.PcmCoreRpc.SyncInfo(context.Background(), &syncInfoReq)
		}
	})
}

func submitJob(infoList *pcmcoreclient.InfoListResp, submitJobLogic *logic.SubmitJobLogic) {
	for index, _ := range infoList.HpcInfoList {
		if infoList.HpcInfoList[index].Status == "Saved" {
			submitReq := hpcAC.SubmitJobReq{
				Appname:         "BASE",
				Apptype:         "BASIC",
				StrJobManagerID: 1638523853,
				MapAppJobInfo: &hpcAC.MapAppJobInfo{
					GAP_CMD_FILE:     "sleep 10",
					GAP_NNODE:        "1",
					GAP_SUBMIT_TYPE:  "cmd",
					GAP_JOB_NAME:     infoList.HpcInfoList[index].Name,
					GAP_WORK_DIR:     infoList.HpcInfoList[index].WorkDir,
					GAP_QUEUE:        "debug2",
					GAP_NPROC:        "1",
					GAP_APPNAME:      "BASE",
					GAP_WALL_TIME:    infoList.HpcInfoList[index].WallTime,
					GAP_STD_OUT_FILE: "/public/home/zhijiang/test/testjob1/std.out.%j",
					GAP_STD_ERR_FILE: " /public/home/zhijiang/test/testjob1/std.err.%j",
				},
			}
			jobResult, _ := submitJobLogic.SubmitJob(&submitReq)
			if jobResult.Code == "0" {
				infoList.HpcInfoList[index].Status = "Pending"
				infoList.HpcInfoList[index].JobId = jobResult.Data
			} else {
				infoList.HpcInfoList[index].Result = "Failed"
				infoList.HpcInfoList[index].Result = jobResult.Msg
			}
		}
	}
}

func queryCoreInfoList(svc *svc.ServiceContext) (*pcmcoreclient.InfoListResp, error) {
	infoReq := pcmcoreclient.InfoListReq{
		Kind:        "hpc",
		ServiceName: "ac",
	}
	infoList, err := svc.PcmCoreRpc.InfoList(context.Background(), &infoReq)
	if err != nil {
		return nil, err
	}
	return infoList, nil
}