fix: add task_ai_async

Former-commit-id: 55fdcb01ab8719726afec147431087c2f2d284f8

client/types.go

@@ -136,61 +136,61 @@ type CloudInfo struct {
 }
 
 type AiInfo struct {
-	Id                    int64            `json:"id"` // id
+	Id           int64   `json:"id"` // id
-	AdapterId             int64            `json:"adapterId,omitempty,optional"`
+	AdapterId    int64   `json:"adapterId,omitempty,optional"`
-	AdapterName           string           `json:"adapterName,omitempty,optional"`
+	AdapterName  string  `json:"adapterName,omitempty,optional"`
-	ClusterId             int64            `json:"clusterId,omitempty,optional"`
+	ClusterId    int64   `json:"clusterId,omitempty,optional"`
-	ClusterIds            []int64          `json:"clusterIds,omitempty,optional"`
+	ClusterIds   []int64 `json:"clusterIds,omitempty,optional"`
-	TaskId                int64            `json:"taskId,omitempty"`
+	TaskId       int64   `json:"taskId,omitempty"`
-	TaskName              string           `json:"taskName,omitempty"`
+	ClusterName  string  `json:"clusterName,omitempty,optional"`
-	Replica               int32            `json:"replica,omitempty"`
+	ImageId      string  `json:"imageId,omitempty"`
-	ResourceType          string           `json:"resourceType,omitempty"`
+	ResourceId   string  `json:"resourceId,omitempty"`
-	CpuCoreNum            int32            `json:"cpuCoreNum,omitempty"`
+	AlgorithmId  string  `json:"algorithmId,omitempty"`
-	TaskType              string           `json:"taskType,omitempty"`
+	MetadataName string  `json:"metadataName,omitempty"`
-	DatasetsName          string           `json:"datasetsName,omitempty"`
+	Command      string  `json:"command,omitempty"`
-	ProjectId             string           `json:"project_id,omitempty"`
+	Environments string  `json:"environments,omitempty"`
-	StrategyName          string           `json:"strategyName,omitempty"`
+	Parameters   string  `json:"parameters,omitempty"`
-	ClusterToStaticWeight map[string]int32 `json:"clusterToStaticWeight,omitempty"`
-	Tops                  float64          `json:"tops,omitempty"`
-	ComputeCard           string           `json:"computeCard,omitempty,optional"`
-	CodeType              string           `json:"codeType,omitempty,optional"`
-	ClusterName           string           `json:"clusterName,omitempty,optional"`
-	ModelName             string           `json:"ModelName,omitempty,optional"`
-	AlgorithmName         string           `json:"algorithmName,omitempty,optional"`
-	Strategy              string           `json:"strategy,omitempty"`
 
-	ImageId      string `json:"imageId,omitempty"`
+	Name      string `json:"name,omitempty"`
-	SpecId       string `json:"specId,omitempty"`
+	Status    string `json:"status,omitempty"`
-	DatasetsId   string `json:"datasetsId,omitempty"`
+	StartTime string `json:"startTime,omitempty"`
-	CodeId       string `json:"codeId,omitempty"`
+	//RunningTime int64  `json:"runningTime,omitempty"`
-	ResourceId   string `json:"resourceId,omitempty"`
+	JobId    string `json:"jobId,omitempty"`
-	AlgorithmId  string `json:"algorithmId,omitempty"`
+	FlavorId string `json:"flavorId,omitempty"`
-	MetadataName string `json:"metadataName,omitempty"`
+	//TaskName              string           `json:"taskName,omitempty"`
+	//Replica               int32            `json:"replica,omitempty"`
+	//ResourceType          string           `json:"resourceType,omitempty"`
+	//CpuCoreNum            int32            `json:"cpuCoreNum,omitempty"`
+	//TaskType              string           `json:"taskType,omitempty"`
+	//DatasetsName          string           `json:"datasetsName,omitempty"`
+	//ProjectId             string           `json:"project_id,omitempty"`
+	//StrategyName          string           `json:"strategyName,omitempty"`
+	//ClusterToStaticWeight map[string]int32 `json:"clusterToStaticWeight,omitempty"`
+	//Tops                  float64          `json:"tops,omitempty"`
+	//ComputeCard           string           `json:"computeCard,omitempty,optional"`
+	//CodeType              string           `json:"codeType,omitempty,optional"`
 
-	Cmd          string   `json:"cmd,omitempty"`
+	//ModelName             string           `json:"ModelName,omitempty,optional"`
-	Envs         []string `json:"envs,omitempty"`
+	//AlgorithmName         string           `json:"algorithmName,omitempty,optional"`
-	Params       []string `json:"params,omitempty"`
+	//Strategy              string           `json:"strategy,omitempty"`
-	Environments string   `json:"environments,omitempty"`
+	//Envs         []string `json:"envs,omitempty"`
-	Parameters   string   `json:"parameters,omitempty"`
+	//Params       []string `json:"params,omitempty"`
 
-	Name        string `json:"name,omitempty"`
+	//SpecId       string `json:"specId,omitempty"`
-	Status      string `json:"status,omitempty"`
+	//DatasetsId   string `json:"datasetsId,omitempty"`
-	StartTime   string `json:"startTime,omitempty"`
+	//CodeId       string `json:"codeId,omitempty"`
-	RunningTime int64  `json:"runningTime,omitempty"`
+	//Result      string `json:"result,omitempty"`
-	Result      string `json:"result,omitempty"`
-	JobId       string `json:"jobId,omitempty"`
 
-	Datasets      string `json:"datasets,omitempty"`
+	//Datasets      string `json:"datasets,omitempty"`
-	AlgorithmCode string `json:"algorithmCode,omitempty"`
+	//AlgorithmCode string `json:"algorithmCode,omitempty"`
-	Image         string `json:"image,omitempty"`
+	//Image         string `json:"image,omitempty"`
 
-	CreateTime     string `json:"createTime,omitempty"`
+	//CreateTime     string `json:"createTime,omitempty"`
-	ImageUrl       string `json:"imageUrl,omitempty"`
+	//ImageUrl       string `json:"imageUrl,omitempty"`
-	Command        string `json:"command,omitempty"`
+
-	FlavorId       string `json:"flavorId,omitempty"`
+	//SubscriptionId string `json:"subscriptionId,omitempty"`
-	SubscriptionId string `json:"subscriptionId,omitempty"`
+	//ItemVersionId  string `json:"itemVersionId,omitempty"`
-	ItemVersionId  string `json:"itemVersionId,omitempty"`
+	//ObsUrl         string `json:"obsUrl,omitempty"`
-	ObsUrl         string `json:"obsUrl,omitempty"`
 }
 
 type VmInfo struct {

internal/logic/core/pulltaskinfologic.go

@@ -74,45 +74,12 @@ func (l *PullTaskInfoLogic) PullTaskInfo(req *clientCore.PullTaskInfoReq) (*clie
 		}
 
 	case 1:
-		var aiModelList []models.TaskAiModelarts
+		var aiModelList []models.TaskAiAsynchronous
 		err := findModelList(req.AdapterId, l.svcCtx.DbEngin, &aiModelList)
 		if err != nil {
 			return nil, err
 		}
 		utils.Convert(aiModelList, &resp.AiInfoList)
-		/*if len(resp.AiInfoList) > 0 {
-			for i, aiInfo := range aiModelList {
-				if resp.AiInfoList[i].Environments != "" {
-					// 定义一个map来存储解析后的JSON数据
-					var result map[string]interface{}
-					// 解析JSON字符串
-					err := json.Unmarshal([]byte(resp.AiInfoList[i].Environments), &result)
-					if err != nil {
-						log.Fatalf("Error parsing JSON: %v", err)
-					}
-					// 如果你需要将解析后的map再次转换为JSON字符串，可以使用json.MarshalIndent
-					formattedJSON, err := json.MarshalIndent(result, "", "  ")
-					aiInfo.Environments = string(formattedJSON)
-					fmt.Println(aiInfo.Environments)
-					resp.AiInfoList[i].Environments = aiInfo.Environments
-				}
-				if resp.AiInfoList[i].Parameters != "" {
-					// 定义一个map来存储解析后的JSON数据
-					var result []interface{}
-					// 解析JSON字符串
-					err := json.Unmarshal([]byte(resp.AiInfoList[i].Parameters), &result)
-					if err != nil {
-						log.Fatalf("Error parsing JSON: %v", err)
-					}
-					// 如果你需要将解析后的map再次转换为JSON字符串，可以使用json.MarshalIndent
-					formattedJSON, err := json.MarshalIndent(result, "", "  ")
-					aiInfo.Parameters = string(formattedJSON)
-					fmt.Println(aiInfo.Parameters)
-					resp.AiInfoList[i].Parameters = aiInfo.Parameters
-				}
-
-			}
-		}*/
 	}
 	return &resp, nil
 }

internal/logic/core/pushtaskinfologic.go

@@ -89,8 +89,8 @@ func (l *PushTaskInfoLogic) PushTaskInfo(req *clientCore.PushTaskInfoReq) (*clie
 		}
 	case 1:
 		for _, aiInfo := range req.AiInfoList {
-			l.svcCtx.DbEngin.Exec("update task_ai set status = ?,start_time = ?,project_id = ?,job_id = ? where participant_id = ? and task_id = ? and name = ?",
+			l.svcCtx.DbEngin.Exec("update task_ai_asynchronous set status = ?,start_time = ?,job_id = ? where cluster_id = ? and task_id = ? and name = ?",
-				aiInfo.Status, aiInfo.StartTime, aiInfo.ProjectId, aiInfo.JobId, req.AdapterId, aiInfo.TaskId, aiInfo.Name)
+				aiInfo.Status, aiInfo.StartTime, aiInfo.JobId, aiInfo.ClusterId, aiInfo.TaskId, aiInfo.Name)
 			noticeInfo := clientCore.NoticeInfo{
 				TaskId:      aiInfo.TaskId,
 				AdapterId:   aiInfo.AdapterId,

internal/storeLink/modelarts.go

@@ -22,14 +22,12 @@ import (
 	"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/service/collector"
 	"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/service/inference"
 	"gitlink.org.cn/JointCloud/pcm-coordinator/pkg/constants"
-	"gitlink.org.cn/JointCloud/pcm-coordinator/pkg/models"
 	"gitlink.org.cn/JointCloud/pcm-coordinator/pkg/utils"
 	"gitlink.org.cn/JointCloud/pcm-coordinator/pkg/utils/timeutils"
 	"gitlink.org.cn/JointCloud/pcm-modelarts/client/imagesservice"
 	"gitlink.org.cn/JointCloud/pcm-modelarts/client/modelartsservice"
 	"gitlink.org.cn/JointCloud/pcm-modelarts/modelarts"
 	modelartsclient "gitlink.org.cn/JointCloud/pcm-modelarts/modelarts"
-	"gorm.io/gorm"
 	"log"
 	"mime/multipart"
 	"strconv"
@@ -56,7 +54,6 @@ type ModelArtsLink struct {
 	Version         string
 	ModelId         string
 	ModelType       string
-	DbEngin         *gorm.DB
 }
 
 // Version 结构体表示版本号
@@ -174,20 +171,9 @@ func (m *ModelArtsLink) SubmitTask(ctx context.Context, imageId string, cmd stri
 				NodeCount: 1,
 			},
 		},
-		//Platform: m.platform,
+		Platform: m.platform,
-		Platform: "modelarts-CloudBrain2",
 	}
 	resp, err := m.modelArtsRpc.CreateTrainingJob(ctx, req)
-	aiModelarts := models.TaskAiModelarts{}
-	aiModelarts.ImageId = imageId
-	aiModelarts.FlavorId = resourceId
-	aiModelarts.Cmd = cmd
-	//aiModelarts.TaskId =
-	tx := m.DbEngin.Table("task_ai_modelarts").Create(&aiModelarts)
-	if tx.Error != nil {
-		return tx.Error, nil
-	}
-
 	if err != nil {
 		return nil, err
 	}

pkg/models/taskaiasynchronousmodel.go

@@ -0,0 +1,29 @@
+package models
+
+import "github.com/zeromicro/go-zero/core/stores/sqlx"
+
+var _ TaskAiAsynchronousModel = (*customTaskAiAsynchronousModel)(nil)
+
+type (
+	// TaskAiAsynchronousModel is an interface to be customized, add more methods here,
+	// and implement the added methods in customTaskAiAsynchronousModel.
+	TaskAiAsynchronousModel interface {
+		taskAiAsynchronousModel
+		withSession(session sqlx.Session) TaskAiAsynchronousModel
+	}
+
+	customTaskAiAsynchronousModel struct {
+		*defaultTaskAiAsynchronousModel
+	}
+)
+
+// NewTaskAiAsynchronousModel returns a model for the database table.
+func NewTaskAiAsynchronousModel(conn sqlx.SqlConn) TaskAiAsynchronousModel {
+	return &customTaskAiAsynchronousModel{
+		defaultTaskAiAsynchronousModel: newTaskAiAsynchronousModel(conn),
+	}
+}
+
+func (m *customTaskAiAsynchronousModel) withSession(session sqlx.Session) TaskAiAsynchronousModel {
+	return NewTaskAiAsynchronousModel(sqlx.NewSqlConnFromSession(session))
+}

pkg/models/taskaiasynchronousmodel_gen.go

@@ -0,0 +1,98 @@
+// Code generated by goctl. DO NOT EDIT.
+
+package models
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"strings"
+
+	"github.com/zeromicro/go-zero/core/stores/builder"
+	"github.com/zeromicro/go-zero/core/stores/sqlx"
+	"github.com/zeromicro/go-zero/core/stringx"
+)
+
+var (
+	taskAiAsynchronousFieldNames          = builder.RawFieldNames(&TaskAiAsynchronous{})
+	taskAiAsynchronousRows                = strings.Join(taskAiAsynchronousFieldNames, ",")
+	taskAiAsynchronousRowsExpectAutoSet   = strings.Join(stringx.Remove(taskAiAsynchronousFieldNames, "`create_at`", "`create_time`", "`created_at`", "`update_at`", "`update_time`", "`updated_at`"), ",")
+	taskAiAsynchronousRowsWithPlaceHolder = strings.Join(stringx.Remove(taskAiAsynchronousFieldNames, "`id`", "`create_at`", "`create_time`", "`created_at`", "`update_at`", "`update_time`", "`updated_at`"), "=?,") + "=?"
+)
+
+type (
+	taskAiAsynchronousModel interface {
+		Insert(ctx context.Context, data *TaskAiAsynchronous) (sql.Result, error)
+		FindOne(ctx context.Context, id int64) (*TaskAiAsynchronous, error)
+		Update(ctx context.Context, data *TaskAiAsynchronous) error
+		Delete(ctx context.Context, id int64) error
+	}
+
+	defaultTaskAiAsynchronousModel struct {
+		conn  sqlx.SqlConn
+		table string
+	}
+
+	TaskAiAsynchronous struct {
+		Id          int64  `db:"id"`           // 训练作业资源规格id
+		TaskId      int64  `db:"task_id"`      // 任务id
+		AdapterId   int64  `db:"adapter_id"`   // 执行任务的适配器id
+		AdapterName string `db:"adapter_name"` // 适配器名称
+		ClusterId   int64  `db:"cluster_id"`   // 集群id
+		ClusterName string `db:"cluster_name"` // 集群名称
+		Name        string `db:"name"`         // 任务名
+		Replica     int64  `db:"replica"`      // 执行数
+		JobId       string `db:"job_id"`       // 集群返回任务id
+		StartTime   string `db:"start_time"`   // 开始时间
+		RunningTime string `db:"running_time"` // 运行时间
+		Result      string `db:"result"`       // 运行结果
+		DeletedAt   string `db:"deleted_at"`   // 删除时间
+		ImageId     string `db:"image_id"`     // 镜像id
+		Cmd         string `db:"cmd"`          // 命令行
+		FlavorId    string `db:"flavor_id"`    // 训练作业资源规格id
+		Status      string `db:"status"`       // 任务状态
+	}
+)
+
+func newTaskAiAsynchronousModel(conn sqlx.SqlConn) *defaultTaskAiAsynchronousModel {
+	return &defaultTaskAiAsynchronousModel{
+		conn:  conn,
+		table: "`task_ai_asynchronous`",
+	}
+}
+
+func (m *defaultTaskAiAsynchronousModel) Delete(ctx context.Context, id int64) error {
+	query := fmt.Sprintf("delete from %s where `id` = ?", m.table)
+	_, err := m.conn.ExecCtx(ctx, query, id)
+	return err
+}
+
+func (m *defaultTaskAiAsynchronousModel) FindOne(ctx context.Context, id int64) (*TaskAiAsynchronous, error) {
+	query := fmt.Sprintf("select %s from %s where `id` = ? limit 1", taskAiAsynchronousRows, m.table)
+	var resp TaskAiAsynchronous
+	err := m.conn.QueryRowCtx(ctx, &resp, query, id)
+	switch err {
+	case nil:
+		return &resp, nil
+	case sqlx.ErrNotFound:
+		return nil, ErrNotFound
+	default:
+		return nil, err
+	}
+}
+
+func (m *defaultTaskAiAsynchronousModel) Insert(ctx context.Context, data *TaskAiAsynchronous) (sql.Result, error) {
+	query := fmt.Sprintf("insert into %s (%s) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", m.table, taskAiAsynchronousRowsExpectAutoSet)
+	ret, err := m.conn.ExecCtx(ctx, query, data.Id, data.TaskId, data.AdapterId, data.AdapterName, data.ClusterId, data.ClusterName, data.Name, data.Replica, data.JobId, data.StartTime, data.RunningTime, data.Result, data.DeletedAt, data.ImageId, data.Cmd, data.FlavorId, data.Status)
+	return ret, err
+}
+
+func (m *defaultTaskAiAsynchronousModel) Update(ctx context.Context, data *TaskAiAsynchronous) error {
+	query := fmt.Sprintf("update %s set %s where `id` = ?", m.table, taskAiAsynchronousRowsWithPlaceHolder)
+	_, err := m.conn.ExecCtx(ctx, query, data.TaskId, data.AdapterId, data.AdapterName, data.ClusterId, data.ClusterName, data.Name, data.Replica, data.JobId, data.StartTime, data.RunningTime, data.Result, data.DeletedAt, data.ImageId, data.Cmd, data.FlavorId, data.Status, data.Id)
+	return err
+}
+
+func (m *defaultTaskAiAsynchronousModel) tableName() string {
+	return m.table
+}