pcm-coordinator/internal/logic/core/jobtotallogic.go

/*

 Copyright (c) [2023] [pcm]
 [pcm-coordinator] is licensed under Mulan PSL v2.
 You can use this software according to the terms and conditions of the Mulan PSL v2.
 You may obtain a copy of Mulan PSL v2 at:
          http://license.coscl.org.cn/MulanPSL2
 THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 EITHER EXPaRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 See the Mulan PSL v2 for more details.

*/

package core

import (
	"context"
	"github.com/zeromicro/go-zero/core/logx"
	"gitlink.org.cn/JointCloud/pcm-coordinator/internal/svc"
	"gitlink.org.cn/JointCloud/pcm-coordinator/internal/types"
	"gitlink.org.cn/JointCloud/pcm-coordinator/pkg/helper/enum"
	"gitlink.org.cn/JointCloud/pcm-coordinator/pkg/models"
	"gitlink.org.cn/JointCloud/pcm-coordinator/pkg/utils/httputils"
	"k8s.io/apimachinery/pkg/util/json"
)

type JobTotalLogic struct {
	logx.Logger
	ctx    context.Context
	svcCtx *svc.ServiceContext
}

type Job struct {
	TotalSize int     `json:"totalSize"`
	OtJobs    []OtJob `json:"otJobs"`
}

type OtJob struct {
	Name   string `json:"name"`
	Status string `json:"status"`
	Tasks  []Task `json:"tasks"`
}

type Task struct {
	CenterName []string `json:"centerName"`
}

func NewJobTotalLogic(ctx context.Context, svcCtx *svc.ServiceContext) *JobTotalLogic {
	return &JobTotalLogic{
		Logger: logx.WithContext(ctx),
		ctx:    ctx,
		svcCtx: svcCtx,
	}
}

func (l *JobTotalLogic) JobTotal() (resp *types.JobTotalResp, err error) {
	// 获取任务时间信息
	resp = &types.JobTotalResp{}
	bytes, err := httputils.HttpGet("GET", "http://grampus.openi.org.cn/openapi/v1/sharescreen/computepower/alljobinfo")
	if err != nil {
		return nil, err
	}
	json.Unmarshal(bytes, resp)

	// 获取其他任务信息
	jobs := &Job{}
	jobBytes, err := httputils.HttpGet("GET", "http://grampus.openi.org.cn/openapi/v1/sharescreen/trainjob?pageIndex=1&pageSize=10")
	if err != nil {
		return nil, err
	}
	json.Unmarshal(jobBytes, jobs)

	for _, job := range jobs.OtJobs {
		trainJob := types.TrainJob{
			Name:          job.Name,
			Status:        enum.ExternalStatus(job.Status).String(),
			Strategy:      0,
			SynergyStatus: "未协同",
		}
		if job.Tasks[0].CenterName != nil {
			trainJob.ParticipantName = job.Tasks[0].CenterName[0]
		}
		resp.TrainJobs = append(resp.TrainJobs, trainJob)
	}

	var tasks []models.Task
	tx := l.svcCtx.DbEngin.Find(&tasks)
	if tx.Error != nil {
		logx.Error(err)
		return nil, tx.Error
	}
	if len(tasks) == 0 {
		return nil, nil
	}
	for _, task := range tasks {
		var participantName string
		tx := l.svcCtx.DbEngin.Raw("SELECT name from sc_participant_phy_info  where id  in (SELECT CONCAT_WS(',',GROUP_CONCAT(DISTINCT h.participant_id) ,GROUP_CONCAT(DISTINCT  a.participant_id) ,GROUP_CONCAT(DISTINCT c.participant_id))as service_name from task t left join hpc h on t.id  = h.task_id left join cloud c  on t.id  = c.task_id left join ai a on t.id  = a.task_id where t.id  = ?)", task.Id).Scan(&participantName)
		if tx.Error != nil {
			logx.Error(err)
			return nil, tx.Error
		}
		// 承接方转义
		resp.TrainJobs = append(resp.TrainJobs, types.TrainJob{
			ParticipantName: participantName,
			Name:            task.Name,
			Strategy:        int(task.Strategy),
			SynergyStatus:   enum.SynergyStatus(task.SynergyStatus).String(),
			Status:          task.Status,
		})

	}
	return resp, nil
}