pcm-coordinator/api/internal/storeLink/octopus.go

package storeLink

import (
	"context"
	"gitlink.org.cn/jcce-pcm/pcm-coordinator/api/internal/svc"
	"gitlink.org.cn/jcce-pcm/pcm-coordinator/pkg/models"
	"gitlink.org.cn/jcce-pcm/pcm-coordinator/pkg/utils"
	"gitlink.org.cn/jcce-pcm/pcm-participant-octopus/octopus"
	"strings"
)

type OctopusLink struct {
	ctx         context.Context
	svcCtx      *svc.ServiceContext
	pageIndex   int32
	pageSize    int32
	participant *models.ScParticipantPhyInfo
}

const (
	IMG_NAME_PREFIX    = "oct_"
	IMG_VERSION_PREFIX = "version_"
	TASK_NAME_PREFIX   = "trainJob_"
	RESOURCE_POOL      = "common-pool"
)

func NewOctopusLink(ctx context.Context, svcCtx *svc.ServiceContext, participant *models.ScParticipantPhyInfo) *OctopusLink {
	return &OctopusLink{ctx: ctx, svcCtx: svcCtx, participant: participant, pageIndex: 1, pageSize: 100}
}

func (o *OctopusLink) UploadImage(path string) (interface{}, error) {
	// octopus创建镜像
	createReq := &octopus.CreateImageReq{
		Platform: o.participant.Name,
		CreateImage: &octopus.CreateImage{
			SourceType:   1,
			ImageName:    IMG_NAME_PREFIX + utils.RandomString(7),
			ImageVersion: IMG_VERSION_PREFIX + utils.RandomString(7),
		},
	}
	createResp, err := o.svcCtx.OctopusRpc.CreateImage(o.ctx, createReq)
	if err != nil {
		return nil, err
	}

	// octopus上传镜像
	uploadReq := &octopus.UploadImageReq{
		Platform: o.participant.Name,
		ImageId:  createResp.Payload.ImageId,
		Params: &octopus.UploadImageParam{
			Domain:   "",
			FileName: "",
		},
	}
	uploadResp, err := o.svcCtx.OctopusRpc.UploadImage(o.ctx, uploadReq)
	if err != nil {
		return nil, err
	}

	// Todo 实际上传

	//转换成统一返回类型
	resp, err := ConvertType[octopus.UploadImageResp](uploadResp, nil)
	if err != nil {
		return nil, err
	}

	return resp, nil
}

func (o *OctopusLink) DeleteImage(imageId string) (interface{}, error) {
	// octopus删除镜像
	req := &octopus.DeleteImageReq{
		Platform: o.participant.Name,
		ImageId:  imageId,
	}
	resp, err := o.svcCtx.OctopusRpc.DeleteImage(o.ctx, req)
	if err != nil {
		return nil, err
	}

	//转换成统一返回类型
	deleteResp, err := ConvertType[octopus.DeleteImageResp](resp, nil)
	if err != nil {
		return nil, err
	}

	return deleteResp, nil
}

func (o *OctopusLink) QueryImageList() (interface{}, error) {
	// octopus获取镜像列表
	req := &octopus.GetUserImageListReq{
		Platform:  o.participant.Name,
		PageIndex: o.pageIndex,
		PageSize:  o.pageSize,
	}
	resp, err := o.svcCtx.OctopusRpc.GetUserImageList(o.ctx, req)
	if err != nil {
		return nil, err
	}

	//转换成统一返回类型
	imgListResp, err := ConvertType[octopus.GetUserImageListResp](resp, nil)
	if err != nil {
		return nil, err
	}

	return imgListResp, nil
}

func (o *OctopusLink) SubmitTask(imageId string, cmd string, params []string, resourceId string) (interface{}, error) {
	// octopus提交任务
	var prms []*octopus.Parameters
	for _, param := range params {
		var p octopus.Parameters
		s := strings.Split(param, COMMA)
		p.Key = s[0]
		p.Value = s[1]
		prms = append(prms, &p)
	}

	req := &octopus.CreateTrainJobReq{
		Platform: o.participant.Name,
		Params: &octopus.CreateTrainJobParam{
			ImageId:      imageId,
			Name:         TASK_NAME_PREFIX + utils.RandomString(7),
			ResourcePool: RESOURCE_POOL,
			Config: []*octopus.Config{
				{
					Command:               cmd,
					ResourceSpecId:        resourceId,
					MinFailedTaskCount:    1,
					MinSucceededTaskCount: 1,
					TaskNumber:            1,
					Parameters:            prms,
				},
			},
		},
	}
	resp, err := o.svcCtx.OctopusRpc.CreateTrainJob(o.ctx, req)
	if err != nil {
		return nil, err
	}

	//转换成统一返回类型
	submitResp, err := ConvertType[octopus.CreateTrainJobResp](resp, nil)
	if err != nil {
		return nil, err
	}

	return submitResp, nil
}

func (o *OctopusLink) QueryTask(taskId string) (interface{}, error) {
	// octopus获取任务
	req := &octopus.GetTrainJobReq{
		Platform: o.participant.Name,
		Id:       taskId,
	}
	resp, err := o.svcCtx.OctopusRpc.GetTrainJob(o.ctx, req)
	if err != nil {
		return nil, err
	}

	//转换成统一返回类型
	taskResp, err := ConvertType[octopus.GetTrainJobResp](resp, nil)
	if err != nil {
		return nil, err
	}

	return taskResp, nil
}

func (o *OctopusLink) DeleteTask(taskId string) (interface{}, error) {
	// octopus删除任务
	req := &octopus.DeleteTrainJobReq{
		Platform: o.participant.Name,
		JobIds:   []string{taskId},
	}
	resp, err := o.svcCtx.OctopusRpc.DeleteTrainJob(o.ctx, req)
	if err != nil {
		return nil, err
	}

	//转换成统一返回类型
	deleteResp, err := ConvertType[octopus.DeleteTrainJobResp](resp, nil)
	if err != nil {
		return nil, err
	}

	return deleteResp, nil
}

func (o *OctopusLink) QuerySpecs() (interface{}, error) {
	// octopus查询资源规格
	req := &octopus.GetResourceSpecsReq{
		Platform:     o.participant.Name,
		ResourcePool: "common-pool",
	}
	resp, err := o.svcCtx.OctopusRpc.GetResourceSpecs(o.ctx, req)
	if err != nil {
		return nil, err
	}

	//转换成统一返回类型
	specsResp, err := ConvertType[octopus.GetResourceSpecsResp](resp, o.participant)
	if err != nil {
		return nil, err
	}

	return specsResp, nil
}