forked from JointCloud/pcm-coordinator
111 lines
2.9 KiB
Go
111 lines
2.9 KiB
Go
package inference
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"github.com/zeromicro/go-zero/core/logx"
|
|
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/common"
|
|
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/service/updater"
|
|
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/svc"
|
|
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/types"
|
|
"gitlink.org.cn/JointCloud/pcm-coordinator/pkg/models"
|
|
"time"
|
|
)
|
|
|
|
type DeployInstanceListLogic struct {
|
|
logx.Logger
|
|
ctx context.Context
|
|
svcCtx *svc.ServiceContext
|
|
}
|
|
|
|
func NewDeployInstanceListLogic(ctx context.Context, svcCtx *svc.ServiceContext) *DeployInstanceListLogic {
|
|
return &DeployInstanceListLogic{
|
|
Logger: logx.WithContext(ctx),
|
|
ctx: ctx,
|
|
svcCtx: svcCtx,
|
|
}
|
|
}
|
|
|
|
func (l *DeployInstanceListLogic) DeployInstanceList(req *types.DeployInstanceListReq) (resp *types.DeployInstanceListResp, err error) {
|
|
limit := req.PageSize
|
|
offset := req.PageSize * (req.PageNum - 1)
|
|
resp = &types.DeployInstanceListResp{}
|
|
|
|
var tasklist []*models.AiDeployInstanceTask
|
|
tx := l.svcCtx.DbEngin.Raw("select * from ai_deploy_instance_task").Scan(&tasklist)
|
|
if tx.Error != nil {
|
|
logx.Errorf(tx.Error.Error())
|
|
return nil, tx.Error
|
|
}
|
|
|
|
//count total
|
|
var total int64
|
|
err = tx.Count(&total).Error
|
|
tx.Limit(limit).Offset(offset)
|
|
|
|
if err != nil {
|
|
return resp, err
|
|
}
|
|
|
|
err = tx.Order("create_time desc").Find(&tasklist).Error
|
|
if err != nil {
|
|
return nil, errors.New(err.Error())
|
|
}
|
|
|
|
deployTasks := l.GenerateDeployTasks(tasklist)
|
|
slices := make([][]*models.AiInferDeployInstance, len(deployTasks))
|
|
for i := 0; i < len(deployTasks); i++ {
|
|
slices[i] = deployTasks[i].Instances
|
|
}
|
|
list := common.ConcatMultipleSlices(slices)
|
|
|
|
if len(list) == 0 {
|
|
return
|
|
}
|
|
|
|
go updater.UpdateDeployInstanceStatusBatch(l.svcCtx, list)
|
|
|
|
ins := list[0]
|
|
for i := range list {
|
|
uTime, _ := time.Parse(time.RFC3339, ins.UpdateTime)
|
|
latest, _ := time.Parse(time.RFC3339, list[i].UpdateTime)
|
|
if latest.After(uTime) {
|
|
ins = list[i]
|
|
}
|
|
}
|
|
|
|
go updater.UpdateDeployInstanceStatus(l.svcCtx, ins, true)
|
|
go updater.UpdateDeployTaskStatus(l.svcCtx)
|
|
|
|
resp.List = &deployTasks
|
|
resp.PageSize = req.PageSize
|
|
resp.PageNum = req.PageNum
|
|
resp.Total = total
|
|
|
|
return
|
|
}
|
|
|
|
func (l *DeployInstanceListLogic) GenerateDeployTasks(tasklist []*models.AiDeployInstanceTask) []*DeployTask {
|
|
var tasks []*DeployTask
|
|
for _, t := range tasklist {
|
|
list, err := l.svcCtx.Scheduler.AiStorages.GetInstanceListByDeployTaskId(t.Id)
|
|
if err != nil {
|
|
logx.Errorf("db GetInstanceListByDeployTaskId error")
|
|
continue
|
|
}
|
|
deployTask := &DeployTask{
|
|
Id: t.Id,
|
|
Name: t.Name,
|
|
Instances: list,
|
|
}
|
|
tasks = append(tasks, deployTask)
|
|
}
|
|
return tasks
|
|
}
|
|
|
|
type DeployTask struct {
|
|
Id int64 `json:"id,string"`
|
|
Name string `json:"name,string"`
|
|
Instances []*models.AiInferDeployInstance `json:"instances,string"`
|
|
}
|