floraachy
/
pcm-coordinator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix：create vm server 

Former-commit-id: d45c7c3228f7cedeaac94d7175d220855cf39c46
This commit is contained in:
qiwang 2024-05-11 08:58:57 +08:00
parent a57e0b7ba8
commit 54dd6a7eb2
6 changed files with 206 additions and 84 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
api/desc/core/pcm-core.api
View File

@ -203,15 +203,30 @@ type (
type ( type (
commitVmTaskReq { commitVmTaskReq {
// Name string `json:"name"` Name string `json:"name"`
// NsID string `json:"nsID"` AdapterIds []string `json:"adapterIds,optional"`
ClusterIds []string `json:"clusterIds"`
Strategy string `json:"strategy"`
StaticWeightMap map[string]int32 `json:"staticWeightMap,optional"`
MinCount int64 `json:"min_count,optional"`
ImageRef int64 `json:"imageRef,optional"`
FlavorRef int64 `json:"flavorRef,optional"`
Uuid int64 `json:"uuid,optional"`
//Replicas int64 `json:"replicas,string"`
VmName string `json:"vm_name,optional"`
// Replicas int64 `json:"replicas,optional"` // Replicas int64 `json:"replicas,optional"`
// MatchLabels map[string]string `json:"matchLabels,optional"` // MatchLabels map[string]string `json:"matchLabels,optional"`
// AdapterId string `json:"adapterId,optional"`
// ClusterType string `json:"clusterType,optional"` // ClusterType string `json:"clusterType,optional"`
// //Virtual Machine Section // //Virtual Machine Section
CreateMulServer []CreateMulDomainServer `json:"createMulServer,optional"` //CreateMulServer []CreateMulDomainServer `json:"createMulServer,optional"`
VmOption *VmOption `json:"vmOption,optional"` //VmOption *VmOption `json:"vmOption,optional"`
}
TaskVm {
ImageRef string `json:"imageRef"`
FlavorRef string `json:"flavorRef"`
Uuid string `json:"uuid"`
Platform string `json:"platform"`
} }
VmOption { VmOption {
AdapterId string `json:"adapterId"` AdapterId string `json:"adapterId"`
@ -225,23 +240,6 @@ type (
MatchLabels map[string]string `json:"matchLabels,optional"` MatchLabels map[string]string `json:"matchLabels,optional"`
StaticWeightMap map[string]int32 `json:"staticWeightMap,optional"` StaticWeightMap map[string]int32 `json:"staticWeightMap,optional"`
CreateMulServer []CreateMulDomainServer `json:"createMulServer,optional"` CreateMulServer []CreateMulDomainServer `json:"createMulServer,optional"`
// Id int64 `json:"id"`
// ParticipantId int64 `json:"participantId"`
// TaskId int64 `json:"taskId"`
// AdapterId int64 `json:"adapterId"`
// ClusterId int64 `json:"clusterId"`
// FlavorRef string `json:"flavorRef"`
// ImageRef string `json:"imageRef"`
// Status string `json:"status"`
// Platform string `json:"platform"`
// Description string `json:"description"`
// AvailabilityZone string `json:"availabilityZone"`
// MinCount int64 `json:"minCount"`
// Uuid string `json:"uuid"`
// StartTime string `json:"startTime"`
// RunningTime string `json:"runningTime"`
// Result string `json:"result"`
// DeletedAt string `json:"deletedAt"`
} }
CreateMulDomainServer { CreateMulDomainServer {

7
api/internal/handler/routes.go
View File

@ -1190,6 +1190,11 @@ func RegisterHandlers(server *rest.Server, serverCtx *svc.ServiceContext) {
Path: "/schedule/ai/getAlgorithms/:adapterId/:resourceType/:taskType/:dataset", Path: "/schedule/ai/getAlgorithms/:adapterId/:resourceType/:taskType/:dataset",
Handler: schedule.ScheduleGetAlgorithmsHandler(serverCtx), Handler: schedule.ScheduleGetAlgorithmsHandler(serverCtx),
}, },
{
Method: http.MethodGet,
Path: "/schedule/ai/getJobLog/:adapterId/:clusterId/:taskId/:instanceNum",
Handler: schedule.ScheduleGetAiJobLogLogHandler(serverCtx),
},
{ {
Method: http.MethodPost, Method: http.MethodPost,
Path: "/schedule/submit", Path: "/schedule/submit",
@ -1294,7 +1299,7 @@ func RegisterHandlers(server *rest.Server, serverCtx *svc.ServiceContext) {
}, },
{ {
Method: http.MethodPost, Method: http.MethodPost,
Path: "/core/syncClusterAlert", Path: "/monitoring/syncClusterAlert",
Handler: monitoring.SyncClusterAlertHandler(serverCtx), Handler: monitoring.SyncClusterAlertHandler(serverCtx),
}, },
{ {

193
api/internal/logic/core/commitvmtasklogic.go
View File

@ -3,12 +3,14 @@ package core
import ( import (
"context" "context"
"fmt" "fmt"
clientCore "gitlink.org.cn/JointCloud/pcm-coordinator/api/client"
"gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/scheduler/schedulers" "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/scheduler/schedulers"
"gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/scheduler/schedulers/option" "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/scheduler/schedulers/option"
"gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/svc" "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/internal/types"
"gitlink.org.cn/JointCloud/pcm-coordinator/pkg/constants" "gitlink.org.cn/JointCloud/pcm-coordinator/pkg/constants"
"gitlink.org.cn/JointCloud/pcm-coordinator/pkg/models" "gitlink.org.cn/JointCloud/pcm-coordinator/pkg/models"
"gitlink.org.cn/JointCloud/pcm-coordinator/pkg/utils"
"strconv" "strconv"
"time" "time"
@ -31,9 +33,27 @@ func NewCommitVmTaskLogic(ctx context.Context, svcCtx *svc.ServiceContext) *Comm
func (l *CommitVmTaskLogic) CommitVmTask(req *types.CommitVmTaskReq) (resp *types.CommitVmTaskResp, err error) { func (l *CommitVmTaskLogic) CommitVmTask(req *types.CommitVmTaskReq) (resp *types.CommitVmTaskResp, err error) {
// todo: add your logic here and delete this line // todo: add your logic here and delete this line
/*var ImageRef string
var FlavorRef string
var NetworkRef string*/
resp = &types.CommitVmTaskResp{} resp = &types.CommitVmTaskResp{}
tx := l.svcCtx.DbEngin.Begin()
//Building the main task structure //Building the main task structure
opt := &option.VmOption{ defer func() {
if p := recover(); p != nil {
tx.Rollback()
logx.Error(p)
} else if tx.Error != nil {
logx.Info("rollback, error", tx.Error)
tx.Rollback()
} else {
tx = tx.Commit()
logx.Info("commit success")
}
}()
//TODO adapter
adapterId, _ := strconv.ParseUint(req.AdapterIds[0], 10, 64)
/* opt := &option.VmOption{
AdapterId: req.VmOption.AdapterId, AdapterId: req.VmOption.AdapterId,
Replicas: req.VmOption.Replicas, Replicas: req.VmOption.Replicas,
Strategy: req.VmOption.Strategy, Strategy: req.VmOption.Strategy,
@ -43,32 +63,21 @@ func (l *CommitVmTaskLogic) CommitVmTask(req *types.CommitVmTaskReq) (resp *type
StaticWeightMap: req.VmOption.StaticWeightMap, StaticWeightMap: req.VmOption.StaticWeightMap,
Name: req.VmOption.Name, Name: req.VmOption.Name,
CommitTime: time.Now(), CommitTime: time.Now(),
} }*/
taskModel := models.Task{
Status: constants.Saved,
Name: req.VmOption.Name,
CommitTime: time.Now(),
Description: "vm task",
}
// Save task data to database
tx := l.svcCtx.DbEngin.Create(&taskModel)
if tx.Error != nil {
return nil, tx.Error
}
//var clusters []*models.VmModel var clusters []*models.VmModel
//err2 := l.svcCtx.DbEngin.Raw("SELECT * FROM `t_cluster` where adapter_id in ? and id in ?", req.VmOption.AdapterId, req.VmOption.VmClusterIds).Scan(&clusters).Error err2 := l.svcCtx.DbEngin.Raw("SELECT * FROM `t_cluster` where adapter_id in ? and id in ?", req.AdapterIds, req.ClusterIds).Scan(&clusters).Error
//if err2 != nil { if err2 != nil {
// logx.Errorf("CommitGeneralTask() => sql execution error: %v", err) logx.Errorf("CommitGeneralTask() => sql execution error: %v", err)
// //return errors.Errorf("the cluster does not match the drive resources. Check the data"), nil //return errors.Errorf("the cluster does not match the drive resources. Check the data"), nil
//} }
taskVm := models.TaskVm{} taskVm := models.TaskVm{}
//TODO 执行策略返回集群跟 Replica //TODO 执行策略返回集群跟 Replica
/*opt := &option.VmOption{} opt := &option.VmOption{}
utils.Convert(&req, &opt)*/ utils.Convert(&req, &opt)
// 2、Initialize scheduler // 2、Initialize scheduler
vmSchdl, err := schedulers.NewVmScheduler(l.ctx, "", l.svcCtx.Scheduler, opt, l.svcCtx.DbEngin, l.svcCtx.PromClient) vmSchdl, _ := schedulers.NewVmScheduler(l.ctx, "", l.svcCtx.Scheduler, opt, l.svcCtx.DbEngin, l.svcCtx.PromClient)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -76,43 +85,139 @@ func (l *CommitVmTaskLogic) CommitVmTask(req *types.CommitVmTaskReq) (resp *type
// 3、Return scheduling results // 3、Return scheduling results
results, err := l.svcCtx.Scheduler.AssignAndSchedule(vmSchdl) results, err := l.svcCtx.Scheduler.AssignAndSchedule(vmSchdl)
if err != nil { if err != nil {
logx.Errorf("AssignAndSchedule() => execution error: %v", err)
return nil, err return nil, err
} }
rs := (results).([]*schedulers.VmResult) rs := (results).([]*schedulers.VmResult)
for _, r := range rs { var synergyStatus int64
for _, CreateMulServer := range req.CreateMulServer { if len(rs) > 1 {
if r.Replica > 0 && r.ClusterId == CreateMulServer.ClusterId { synergyStatus = 1
fmt.Println("", req.CreateMulServer)
var clusterIds []int64
l.svcCtx.DbEngin.Raw("SELECT id FROM `t_cluster` where adapter_id = ? ", req.VmOption.AdapterId).Scan(&clusterIds)
if len(clusterIds) == 0 || clusterIds == nil {
return nil, nil
} }
adapterId, _ := strconv.ParseUint(req.VmOption.AdapterId, 10, 64)
taskVm.AdapterId = int64(adapterId) var strategy int64
clusterId, _ := strconv.ParseUint(r.ClusterId, 10, 64) sqlStr := `select t_dict_item.item_value
taskVm.ClusterId = int64(clusterId) from t_dict
taskVm.Name = req.VmOption.Name left join t_dict_item on t_dict.id = t_dict_item.dict_id
taskVm.TaskId = taskModel.Id where item_text = ?
clusterId, _ = strconv.ParseUint(r.ClusterId, 10, 64) and t_dict.dict_code = 'schedule_Strategy'`
taskVm.ClusterId = int64(clusterId) //查询调度策略
err = tx.Raw(sqlStr, req.Strategy).Scan(&strategy).Error
taskModel := models.Task{
Id: utils.GenSnowflakeID(),
Status: constants.Saved,
Name: req.Name,
CommitTime: time.Now(),
Description: "vm task",
AdapterTypeDict: 0,
SynergyStatus: synergyStatus,
Strategy: strategy,
}
var taskVms models.TaskVm
var VmObject types.TaskVm
for _, r := range rs {
for _, clusterId := range req.ClusterIds {
if r.Replica > 0 && r.ClusterId == clusterId {
fmt.Println("", clusterId)
sql := `SELECT vi.image_id as imageRef,vf.flavor_id as flavorRef,vn.network_id as uuid,vi.cluster_name as platform FROM
vm_flavor vf
LEFT JOIN vm_image vi ON vf.cluster_id = vi.cluster_id
LEFT JOIN vm_network vn ON vf.cluster_id = vn.cluster_id
WHERE
vi.cluster_id = ?
AND vf.public_flavor_id = ?
AND vi.public_image_id = ?
AND vn.public_network_id = ?`
// err2 := l.svcCtx.DbEngin.Raw(sql, clusterId, req.FlavorRef, req.ImageRef, req.Uuid).Scan(&taskVm).Error
txVm := l.svcCtx.DbEngin.Raw(sql, clusterId, req.FlavorRef, req.ImageRef, req.Uuid).Scan(&VmObject)
if txVm.Error != nil {
logx.Error(err)
return nil, txVm.Error
}
if err2 != nil {
logx.Errorf("CommitGeneralTask() => sql execution error: %v", err)
//return errors.Errorf("the cluster does not match the drive resources. Check the data"), nil
}
taskVms.Name = req.Name
taskVm.Status = "Saved" taskVm.Status = "Saved"
taskVm.StartTime = time.Now().String() taskVm.StartTime = time.Now().String()
taskVm.MinCount = CreateMulServer.Min_count taskVm.MinCount = req.MinCount
taskVm.ImageRef = CreateMulServer.ImageRef /* sqlImage := "SELECT image_id FROM `vm_image_dict` vm left join vm_image vi on vm.id=vi.public_image_id where cluster_id =? AND public_image_id = ?"
taskVm.FlavorRef = CreateMulServer.FlavorRef txImage := l.svcCtx.DbEngin.Raw(sqlImage, clusterId, req.ImageRef).Scan(&ImageRef)
taskVm.Uuid = CreateMulServer.Uuid if txImage.Error != nil {
taskVm.Platform = CreateMulServer.Platform logx.Error(err)
return nil, txImage.Error
}*/
taskVm.ImageRef = VmObject.ImageRef
/* sqlFlavor := "SELECT * FROM `vm_flavor_dict` vm left join vm_flavor vf on vm.id=vf.public_flavor_id where cluster_id =? AND public_flavor_id = ?"
txFlavor := l.svcCtx.DbEngin.Raw(sqlFlavor, clusterId, req.FlavorRef).Scan(&FlavorRef)
if txFlavor.Error != nil {
logx.Error(err)
return nil, txFlavor.Error
}*/
taskVm.FlavorRef = VmObject.FlavorRef
/* sqlNetwork := "SELECT * FROM `vm_network_dict` vm left join vm_network vi on vm.id=vi.public_network_id where cluster_id =? AND public_network_id = ?"
txNetwork := l.svcCtx.DbEngin.Raw(sqlNetwork, clusterId, req.Uuid).Scan(&NetworkRef)
if txNetwork.Error != nil {
logx.Error(err)
return nil, txNetwork.Error
}*/
taskVm.Uuid = VmObject.Uuid
taskVm.Platform = VmObject.Platform
tx = l.svcCtx.DbEngin.Create(&taskVm) tx = l.svcCtx.DbEngin.Create(&taskVm)
if tx.Error != nil { if tx.Error != nil {
return nil, tx.Error return nil, tx.Error
} }
//var clusterIds []int64
//l.svcCtx.DbEngin.Raw("SELECT id FROM `t_cluster` where adapter_id = ? ", req.VmOption.AdapterId).Scan(&clusterIds)
//if len(clusterIds) == 0 || clusterIds == nil {
// return nil, nil
//}
//adapterId, _ := strconv.ParseUint(req.VmOption.AdapterId, 10, 64)
//taskVm.AdapterId = int64(adapterId)
//clusterId, _ = strconv.ParseUint(r.ClusterId, 10, 64)
//taskVm.ClusterId = int64(clusterId)
//taskVm.Status = "Saved"
//taskVm.StartTime = time.Now().String()
//taskVm.ImageRef = CreateMulServer.ImageRef
//taskVm.FlavorRef = CreateMulServer.FlavorRef
//taskVm.Uuid = CreateMulServer.Uuid
//taskVm.Platform = CreateMulServer.Platform
//tx = l.svcCtx.DbEngin.Create(&taskVm)
//if tx.Error != nil {
// return nil, tx.Error
//}
} }
} }
} }
adapterName := ""
tx.Table("t_adapter").Select("name").Where("id=?", adapterId).Find(&adapterName)
noticeInfo := clientCore.NoticeInfo{
AdapterId: int64(adapterId),
AdapterName: adapterName,
NoticeType: "create",
TaskName: req.Name,
Incident: "任务创建中",
CreatedTime: time.Now(),
}
db := tx.Table("task").Create(&taskModel)
db = tx.Table("task_cloud").Create(&taskVm)
db = tx.Table("t_notice").Create(&noticeInfo)
if db.Error != nil {
logx.Errorf("Task creation failure, err: %v", db.Error)
}
//db = tx.Table("t_notice").Create(&noticeInfo)
// Save task data to database
//tf := l.svcCtx.DbEngin.Create(&taskModel)
//if tf.Error != nil {
// return nil, tf.Error
//}
//tn := l.svcCtx.DbEngin.Create(&noticeInfo)
//if tn.Error != nil {
// return nil, tn.Error
//}
resp.Code = 200 resp.Code = 200
resp.Msg = "Success" resp.Msg = "Success"
return resp, nil return resp, nil
} }

2
api/internal/scheduler/schedulers/vmScheduler.go
View File

@ -83,7 +83,7 @@ func (vm *VmScheduler) PickOptimalStrategy() (strategy.Strategy, error) {
return strategy, nil*/ return strategy, nil*/
case strategy.STATIC_WEIGHT: case strategy.STATIC_WEIGHT:
//todo resources should match cluster StaticWeightMap //todo resources should match cluster StaticWeightMap
strategy := strategy.NewStaticWeightStrategy(vm.option.ClusterToStaticWeight, 1) strategy := strategy.NewStaticWeightStrategy(vm.option.StaticWeightMap, 1)
return strategy, nil return strategy, nil
} }

19
api/internal/types/types.go
View File

@ -182,8 +182,23 @@ type TaskYaml struct {
} }
type CommitVmTaskReq struct { type CommitVmTaskReq struct {
CreateMulServer []CreateMulDomainServer `json:"createMulServer,optional"` Name string `json:"name"`
VmOption *VmOption `json:"vmOption,optional"` AdapterIds []string `json:"adapterIds,optional"`
ClusterIds []string `json:"clusterIds"`
Strategy string `json:"strategy"`
StaticWeightMap map[string]int32 `json:"staticWeightMap,optional"`
MinCount int64 `json:"min_count,optional"`
ImageRef int64 `json:"imageRef,optional"`
FlavorRef int64 `json:"flavorRef,optional"`
Uuid int64 `json:"uuid,optional"`
VmName string `json:"vm_name,optional"`
}
type TaskVm struct {
ImageRef string `json:"imageRef"`
FlavorRef string `json:"flavorRef"`
Uuid string `json:"uuid"`
Platform string `json:"platform"`
} }
type VmOption struct { type VmOption struct {

7
pkg/models/taskvmmodel_gen.go
View File

@ -36,7 +36,6 @@ type (
TaskVm struct { TaskVm struct {
Id int64 `db:"id"` // id Id int64 `db:"id"` // id
ParticipantId int64 `db:"participant_id"` // p端id
TaskId int64 `db:"task_id"` // 任务id TaskId int64 `db:"task_id"` // 任务id
Name string `db:"name"` // 虚拟机名称 Name string `db:"name"` // 虚拟机名称
AdapterId int64 `db:"adapter_id"` // 执行任务的适配器id AdapterId int64 `db:"adapter_id"` // 执行任务的适配器id
@ -91,14 +90,14 @@ func (m *defaultTaskVmModel) FindOne(ctx context.Context, id int64) (*TaskVm, er
} }
func (m *defaultTaskVmModel) Insert(ctx context.Context, data *TaskVm) (sql.Result, error) { func (m *defaultTaskVmModel) Insert(ctx context.Context, data *TaskVm) (sql.Result, error) {
query := fmt.Sprintf("insert into %s (%s) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", m.table, taskVmRowsExpectAutoSet) 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) ret, err := m.conn.ExecCtx(ctx, query, 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 return ret, err
} }
func (m *defaultTaskVmModel) Update(ctx context.Context, data *TaskVm) error { func (m *defaultTaskVmModel) Update(ctx context.Context, data *TaskVm) error {
query := fmt.Sprintf("update %s set %s where `id` = ?", m.table, taskVmRowsWithPlaceHolder) 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) _, err := m.conn.ExecCtx(ctx, query, 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 return err
} }