Merge branch 'master' of https://gitlink.org.cn/zhouqunjie/pcm-coordinator

# Conflicts:
#	api/internal/types/types.go


Former-commit-id: a15861b50777adaaaea13ac07f12e4b7bb3f33bc

api/client/types.go

@@ -124,7 +124,8 @@ type VmInfo struct {
 	BlockUuid           string `json:"block_uuid,omitempty"`
 	SourceType          string `json:"source_type,omitempty"`
 	DeleteOnTermination bool   `json:"delete_on_termination,omitempty"`
-	State               string `json:"state,omitempty"`
+	Status              string `json:"Status,omitempty"`
+	StartTime           string `json:"startTime,omitempty"`
 }
 
 type ResourceStats struct {

api/desc/core/pcm-core.api

@@ -130,35 +130,19 @@ type (
         NsID string `json:"nsID"`
         Replicas int64 `json:"replicas,optional"`
         MatchLabels map[string]string `json:"matchLabels,optional"`
-        servers []ServerCommit `json:"servers,optional"`
-        platform string `json:"platform,optional"`
         AdapterId   string            `json:"adapterId,optional"`
         ClusterType string            `json:"clusterType,optional"`
+        //Virtual Machine Section
+        CreateMulServer []CreateMulDomainServer `json:"createMulServer,optional"`
     }
-    ServerCommit {
+    CreateMulDomainServer {
-        allCardRunTime string `json:"allCardRunTime"`
+        Platform string `json:"platform,optional"`
-        flavorRef string `json:"flavorRef,optional"`
         name string `json:"name,optional"`
+        min_count int64 `json:"min_count,optional"`
         imageRef string `json:"imageRef,optional"`
-        accessIPv4 string `json:"accessIPv4,optional"`
+        flavorRef string `json:"flavorRef,optional"`
-        accessIPv6 string `json:"accessIPv6,optional"`
-        adminPass string `json:"adminPass,optional"`
-        availability_zone string `json:"availability_zone,optional"`
-        key_name string `json:"key_name,optional"`
-        hostname string `json:"hostname,optional"`
-        host string `json:"host,optional"`
-        networks []Networks `json:"networks,optional"`
-    }
-    Networks {
-        uuid string `json:"uuid,optional"`
-        port string `json:"port,optional"`
-        fixed_ip string `json:"fixed_ip,optional"`
-        tag string `json:"tag,optional"`
-    }
-    Block_device_mapping_v2Commit {
         uuid string `json:"uuid,optional"`
     }
-
     commitVmTaskResp {
        // VmTask []VmTask `json:"vmTask" copier:"VmTask"`
         TaskId int64  `json:"taskId"`

api/desc/participant/pcm-participant.api

@@ -105,7 +105,10 @@ type VmInfo {
     BlockUuid           string `json:"block_uuid,omitempty"`
     SourceType          string `json:"source_type,omitempty"`
     DeleteOnTermination bool   `json:"delete_on_termination,omitempty"`
-    State               string `json:"state,omitempty"`
+    Status              string `json:"status,omitempty"`
+    MinCount            string `json:"min_count,omitempty"`
+    Platform            string `json:"platform,omitempty"`
+    Uuid                string `json:"uuid,omitempty"`
 }
 
 type PushTaskInfoReq {

api/desc/schedule/pcm-schedule.api

@@ -19,6 +19,7 @@ type (
     ScheduleResult {
         ClusterId string `json:"clusterId"`
         TaskId string `json:"taskId"`
+        Strategy string `json:"strategy"`
         Replica int32 `json:"replica"`
         Msg string `json:"msg"`
     }

api/internal/logic/core/commitvmtasklogic.go

@@ -2,13 +2,12 @@ package core
 
 import (
 	"context"
-	"gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/mqs"
+	"fmt"
 	"gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/svc"
 	"gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/types"
-	"gitlink.org.cn/JointCloud/pcm-coordinator/api/pkg/response"
 	"gitlink.org.cn/JointCloud/pcm-coordinator/pkg/constants"
 	"gitlink.org.cn/JointCloud/pcm-coordinator/pkg/models"
-	tool "gitlink.org.cn/JointCloud/pcm-coordinator/pkg/utils"
+	"math/rand"
 	"time"
 
 	"github.com/zeromicro/go-zero/core/logx"
@@ -35,7 +34,6 @@ func (l *CommitVmTaskLogic) CommitVmTask(req *types.CommitVmTaskReq) (resp *type
 		Status:     constants.Saved,
 		Name:       req.Name,
 		CommitTime: time.Now(),
-		NsID:       req.NsID,
 	}
 	// Save task data to database
 	tx := l.svcCtx.DbEngin.Create(&taskModel)
@@ -43,28 +41,38 @@ func (l *CommitVmTaskLogic) CommitVmTask(req *types.CommitVmTaskReq) (resp *type
 		return nil, tx.Error
 	}
 
-	var clusterIds []int64
+	for _, CreateMulServer := range req.CreateMulServer {
-	l.svcCtx.DbEngin.Raw("SELECT id FROM `t_cluster` where adapter_id = ? and label = ?", req.AdapterId, req.ClusterType).Scan(&clusterIds)
+		fmt.Println("", req.CreateMulServer)
+		var clusterIds []int64
+		l.svcCtx.DbEngin.Raw("SELECT id FROM `t_cluster` where adapter_id = ? and label = ?", req.AdapterId, req.ClusterType).Scan(&clusterIds)
 
-	if len(clusterIds) == 0 || clusterIds == nil {
+		if len(clusterIds) == 0 || clusterIds == nil {
-		return nil, nil
+			return nil, nil
+		}
+
+		vmInfo := models.TaskVm{
+			TaskId:    taskModel.Id,
+			ClusterId: clusterIds[rand.Intn(len(clusterIds))],
+			Name:      taskModel.Name,
+			Status:    "Saved",
+			StartTime: time.Now().String(),
+			MinCount:  CreateMulServer.Min_count,
+			ImageRef:  CreateMulServer.ImageRef,
+			FlavorRef: CreateMulServer.FlavorRef,
+			Uuid:      CreateMulServer.Uuid,
+			Platform:  CreateMulServer.Platform,
+		}
+
+		tx = l.svcCtx.DbEngin.Create(&vmInfo)
+		if tx.Error != nil {
+			return nil, tx.Error
+		}
+		resp = &types.CommitVmTaskResp{
+			Code:   200,
+			Msg:    "success",
+			TaskId: taskModel.Id,
+		}
 	}
 
-	vm := models.Vm{}
-	tool.Convert(req, &vm)
-	mqInfo := response.TaskInfo{
-		TaskId:      taskModel.Id,
-		TaskType:    "vm",
-		MatchLabels: req.MatchLabels,
-		NsID:        req.NsID,
-	}
-	//req.TaskId = taskModel.Id
-	mqs.InsQueue.Beta.Add(&mqInfo)
-	tx = l.svcCtx.DbEngin.Create(&mqInfo)
-	resp = &types.CommitVmTaskResp{
-		Code:   200,
-		Msg:    "success",
-		TaskId: taskModel.Id,
-	}
 	return resp, nil
 }

api/internal/logic/core/pulltaskinfologic.go

@@ -67,6 +67,13 @@ func (l *PullTaskInfoLogic) PullTaskInfo(req *clientCore.PullTaskInfoReq) (*clie
 			return nil, err
 		}
 		utils.Convert(aiModelList, &resp.AiInfoList)
+	case 3:
+		var vmModelList []models.TaskVm
+		err := findModelList(req.AdapterId, l.svcCtx.DbEngin, &vmModelList)
+		if err != nil {
+			return nil, err
+		}
+		utils.Convert(vmModelList, &resp.VmInfoList)
 	}
 	return &resp, nil
 }

api/internal/logic/core/pushtaskinfologic.go

@@ -49,6 +49,12 @@ func (l *PushTaskInfoLogic) PushTaskInfo(req *clientCore.PushTaskInfoReq) (*clie
 				aiInfo.Status, aiInfo.StartTime, aiInfo.ProjectId, aiInfo.JobId, req.AdapterId, aiInfo.TaskId, aiInfo.Name)
 			syncTask(l.svcCtx.DbEngin, aiInfo.TaskId)
 		}
+	case 3:
+		for _, vmInfo := range req.VmInfoList {
+			l.svcCtx.DbEngin.Exec("update task_vm set status = ?,start_time = ?  where participant_id = ? and task_id = ? and name = ?",
+				vmInfo.Status, vmInfo.StartTime, req.AdapterId, vmInfo.TaskId, vmInfo.Name)
+			syncTask(l.svcCtx.DbEngin, vmInfo.TaskId)
+		}
 	}
 
 	return &resp, nil

api/internal/logic/schedule/schedulesubmitlogic.go

@@ -55,6 +55,7 @@ func (l *ScheduleSubmitLogic) ScheduleSubmit(req *types.ScheduleReq) (resp *type
 			scheResult := &types.ScheduleResult{}
 			scheResult.ClusterId = r.ClusterId
 			scheResult.TaskId = r.TaskId
+			scheResult.Strategy = r.Strategy
 			scheResult.Replica = r.Replica
 			scheResult.Msg = r.Msg
 			resp.Results = append(resp.Results, scheResult)

api/internal/scheduler/schedulers/aiScheduler.go

@@ -43,6 +43,7 @@ type AiScheduler struct {
 type AiResult struct {
 	TaskId    string
 	ClusterId string
+	Strategy  string
 	Replica   int32
 	Msg       string
 }
@@ -154,6 +155,7 @@ func (as *AiScheduler) AssignTask(clusters []*strategy.AssignedCluster) (interfa
 			result, _ := convertType(resp)
 			result.Replica = c.Replicas
 			result.ClusterId = c.ClusterId
+			result.Strategy = as.option.StrategyName
 
 			ch <- result
 			wg.Done()
@@ -167,7 +169,7 @@ func (as *AiScheduler) AssignTask(clusters []*strategy.AssignedCluster) (interfa
 		errs = append(errs, e)
 	}
 
-	if len(errs) != len(clusters) {
+	if len(errs) == len(clusters) {
 		return nil, errors.New("submit task failed")
 	}
 

api/internal/scheduler/service/aiService.go

@@ -28,18 +28,18 @@ func InitAiClusterMap(octopusRpc octopusclient.Octopus, modelArtsRpc modelartsse
 		case OCTOPUS:
 			id, _ := strconv.ParseInt(c.Id, 10, 64)
 			octopus := storeLink.NewOctopusLink(octopusRpc, c.Nickname, id)
-			collectorMap[c.Nickname] = octopus
+			collectorMap[c.Id] = octopus
-			executorMap[c.Nickname] = octopus
+			executorMap[c.Id] = octopus
 		case MODELARTS:
 			id, _ := strconv.ParseInt(c.Id, 10, 64)
 			modelarts := storeLink.NewModelArtsLink(modelArtsRpc, modelArtsImgRpc, c.Nickname, id)
-			collectorMap[c.Nickname] = modelarts
+			collectorMap[c.Id] = modelarts
-			executorMap[c.Nickname] = modelarts
+			executorMap[c.Id] = modelarts
 		case SHUGUANGAI:
 			id, _ := strconv.ParseInt(c.Id, 10, 64)
 			sgai := storeLink.NewShuguangAi(aCRpc, c.Nickname, id)
-			collectorMap[c.Nickname] = sgai
+			collectorMap[c.Id] = sgai
-			executorMap[c.Nickname] = sgai
+			executorMap[c.Id] = sgai
 		}
 	}
 

api/internal/types/types.go

@@ -111,40 +111,22 @@ type TaskYaml struct {
 }
 
 type CommitVmTaskReq struct {
-	Name        string            `json:"name"`
+	Name            string                  `json:"name"`
-	NsID        string            `json:"nsID"`
+	NsID            string                  `json:"nsID"`
-	Replicas    int64             `json:"replicas,optional"`
+	Replicas        int64                   `json:"replicas,optional"`
-	MatchLabels map[string]string `json:"matchLabels,optional"`
+	MatchLabels     map[string]string       `json:"matchLabels,optional"`
-	Servers     []ServerCommit    `json:"servers,optional"`
+	AdapterId       string                  `json:"adapterId,optional"`
-	Platform    string            `json:"platform,optional"`
+	ClusterType     string                  `json:"clusterType,optional"`
-	AdapterId   string            `json:"adapterId,optional"`
+	CreateMulServer []CreateMulDomainServer `json:"createMulServer,optional"`
-	ClusterType string            `json:"clusterType,optional"`
 }
 
-type ServerCommit struct {
+type CreateMulDomainServer struct {
-	AllCardRunTime    string     `json:"allCardRunTime"`
+	Platform  string `json:"platform,optional"`
-	FlavorRef         string     `json:"flavorRef,optional"`
+	Name      string `json:"name,optional"`
-	Name              string     `json:"name,optional"`
+	Min_count int64  `json:"min_count,optional"`
-	ImageRef          string     `json:"imageRef,optional"`
+	ImageRef  string `json:"imageRef,optional"`
-	AccessIPv4        string     `json:"accessIPv4,optional"`
+	FlavorRef string `json:"flavorRef,optional"`
-	AccessIPv6        string     `json:"accessIPv6,optional"`
+	Uuid      string `json:"uuid,optional"`
-	AdminPass         string     `json:"adminPass,optional"`
-	Availability_zone string     `json:"availability_zone,optional"`
-	Key_name          string     `json:"key_name,optional"`
-	Hostname          string     `json:"hostname,optional"`
-	Host              string     `json:"host,optional"`
-	Networks          []Networks `json:"networks,optional"`
-}
-
-type Networks struct {
-	Uuid     string `json:"uuid,optional"`
-	Port     string `json:"port,optional"`
-	Fixed_ip string `json:"fixed_ip,optional"`
-	Tag      string `json:"tag,optional"`
-}
-
-type Block_device_mapping_v2Commit struct {
-	Uuid string `json:"uuid,optional"`
 }
 
 type CommitVmTaskResp struct {
@@ -5289,6 +5271,7 @@ type ScheduleResp struct {
 type ScheduleResult struct {
 	ClusterId string `json:"clusterId"`
 	TaskId    string `json:"taskId"`
+	Strategy  string `json:"strategy"`
 	Replica   int32  `json:"replica"`
 	Msg       string `json:"msg"`
 }

pkg/models/taskvmmodel.go

@@ -0,0 +1,24 @@
+package models
+
+import "github.com/zeromicro/go-zero/core/stores/sqlx"
+
+var _ TaskVmModel = (*customTaskVmModel)(nil)
+
+type (
+	// TaskVmModel is an interface to be customized, add more methods here,
+	// and implement the added methods in customTaskVmModel.
+	TaskVmModel interface {
+		taskVmModel
+	}
+
+	customTaskVmModel struct {
+		*defaultTaskVmModel
+	}
+)
+
+// NewTaskVmModel returns a model for the database table.
+func NewTaskVmModel(conn sqlx.SqlConn) TaskVmModel {
+	return &customTaskVmModel{
+		defaultTaskVmModel: newTaskVmModel(conn),
+	}
+}

pkg/models/taskvmmodel_gen.go

@@ -0,0 +1,107 @@
+// 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/sqlc"
+	"github.com/zeromicro/go-zero/core/stores/sqlx"
+	"github.com/zeromicro/go-zero/core/stringx"
+)
+
+var (
+	taskVmFieldNames          = builder.RawFieldNames(&TaskVm{})
+	taskVmRows                = strings.Join(taskVmFieldNames, ",")
+	taskVmRowsExpectAutoSet   = strings.Join(stringx.Remove(taskVmFieldNames, "`id`", "`create_at`", "`create_time`", "`created_at`", "`update_at`", "`update_time`", "`updated_at`"), ",")
+	taskVmRowsWithPlaceHolder = strings.Join(stringx.Remove(taskVmFieldNames, "`id`", "`create_at`", "`create_time`", "`created_at`", "`update_at`", "`update_time`", "`updated_at`"), "=?,") + "=?"
+)
+
+type (
+	taskVmModel interface {
+		Insert(ctx context.Context, data *TaskVm) (sql.Result, error)
+		FindOne(ctx context.Context, id int64) (*TaskVm, error)
+		Update(ctx context.Context, data *TaskVm) error
+		Delete(ctx context.Context, id int64) error
+	}
+
+	defaultTaskVmModel struct {
+		conn  sqlx.SqlConn
+		table string
+	}
+
+	TaskVm struct {
+		Id               int64  `db:"id"`             // id
+		ParticipantId    int64  `db:"participant_id"` // p端id
+		TaskId           int64  `db:"task_id"`        // 任务id
+		Name             string `db:"name"`           // 虚拟机名称
+		AdapterId        int64  `db:"adapter_id"`     // 执行任务的适配器id
+		ClusterId        int64  `db:"cluster_id"`     // 执行任务的集群id
+		FlavorRef        string `db:"flavor_ref"`     // 规格索引
+		ImageRef         string `db:"image_ref"`      // 镜像索引
+		Status           string `db:"status"`         // 状态
+		Platform         string `db:"platform"`       // 平台
+		Description      string `db:"description"`    // 描述
+		AvailabilityZone string `db:"availability_zone"`
+		MinCount         int64  `db:"min_count"`    // 数量
+		Uuid             string `db:"uuid"`         // 网络id
+		StartTime        string `db:"start_time"`   // 开始时间
+		RunningTime      string `db:"running_time"` // 运行时间
+		Result           string `db:"result"`       // 运行结果
+		DeletedAt        string `db:"deleted_at"`   // 删除时间
+	}
+)
+
+func newTaskVmModel(conn sqlx.SqlConn) *defaultTaskVmModel {
+	return &defaultTaskVmModel{
+		conn:  conn,
+		table: "`task_vm`",
+	}
+}
+
+func (m *defaultTaskVmModel) withSession(session sqlx.Session) *defaultTaskVmModel {
+	return &defaultTaskVmModel{
+		conn:  sqlx.NewSqlConnFromSession(session),
+		table: "`task_vm`",
+	}
+}
+
+func (m *defaultTaskVmModel) 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 *defaultTaskVmModel) FindOne(ctx context.Context, id int64) (*TaskVm, error) {
+	query := fmt.Sprintf("select %s from %s where `id` = ? limit 1", taskVmRows, m.table)
+	var resp TaskVm
+	err := m.conn.QueryRowCtx(ctx, &resp, query, id)
+	switch err {
+	case nil:
+		return &resp, nil
+	case sqlc.ErrNotFound:
+		return nil, ErrNotFound
+	default:
+		return nil, err
+	}
+}
+
+func (m *defaultTaskVmModel) Insert(ctx context.Context, data *TaskVm) (sql.Result, error) {
+	query := fmt.Sprintf("insert into %s (%s) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", m.table, taskVmRowsExpectAutoSet)
+	ret, err := m.conn.ExecCtx(ctx, query, data.ParticipantId, data.TaskId, data.Name, data.AdapterId, data.ClusterId, data.FlavorRef, data.ImageRef, data.Status, data.Platform, data.Description, data.AvailabilityZone, data.MinCount, data.Uuid, data.StartTime, data.RunningTime, data.Result, data.DeletedAt)
+	return ret, err
+}
+
+func (m *defaultTaskVmModel) Update(ctx context.Context, data *TaskVm) error {
+	query := fmt.Sprintf("update %s set %s where `id` = ?", m.table, taskVmRowsWithPlaceHolder)
+	_, err := m.conn.ExecCtx(ctx, query, data.ParticipantId, data.TaskId, data.Name, data.AdapterId, data.ClusterId, data.FlavorRef, data.ImageRef, data.Status, data.Platform, data.Description, data.AvailabilityZone, data.MinCount, data.Uuid, data.StartTime, data.RunningTime, data.Result, data.DeletedAt, data.Id)
+	return err
+}
+
+func (m *defaultTaskVmModel) tableName() string {
+	return m.table
+}