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

modified aioption and submit func 

Former-commit-id: c7dce696d7d090075f4395bb6945a8c06bf77236
This commit is contained in:
tzwang 2024-02-21 17:04:03 +08:00
parent 7e70ac5693
commit c1855def29
9 changed files with 55 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
api/internal/logic/storelink/submitlinktasklogic.go
View File

@ -67,7 +67,7 @@ func (l *SubmitLinkTaskLogic) SubmitLinkTask(req *types.SubmitLinkTaskReq) (resp
envs = append(envs, env) envs = append(envs, env)
} }
} }
task, err := storelink.ILinkage.SubmitTask(req.ImageId, req.Cmd, envs, params, req.ResourceId, "pytorch") task, err := storelink.ILinkage.SubmitTask(req.ImageId, req.Cmd, envs, params, req.ResourceId, "", "", "pytorch")
if err != nil { if err != nil {
return nil, err return nil, err
} }

11
api/internal/scheduler/schedulers/option/aiOption.go
View File

@ -9,11 +9,12 @@ type AiOption struct {
ClusterToStaticWeight map[string]int32 ClusterToStaticWeight map[string]int32
CodeType string CodeType string
ImageId string ImageId string
SpecId string SpecId string
//DatasetsId string DatasetsId string
CodeId string CodeId string
ResourceId string ResourceId string
AlgorithmId string
Cmd string Cmd string
Envs []string Envs []string

10
api/internal/scheduler/service/aiService.go
View File

@ -16,11 +16,11 @@ const (
var ( var (
AiTypeMap = map[string]string{ AiTypeMap = map[string]string{
"hanwuji": OCTOPUS, "hanwuji": OCTOPUS,
"suiyan": OCTOPUS, //"suiyan": OCTOPUS,
"sailingsi": OCTOPUS, //"sailingsi": OCTOPUS,
"modelarts-CloudBrain2": MODELARTS, //"modelarts-CloudBrain2": MODELARTS,
"shuguangAi": SHUGUANGAI, "shuguangAi": SHUGUANGAI,
} }
) )

10
api/internal/scheduler/strategy/strategy.go
View File

@ -1,10 +1,12 @@
package strategy package strategy
const ( const (
REPLICATION = "replication" REPLICATION = "replication"
RESOURCES_PRICING = "resourcesPricing" RESOURCES_PRICING = "resourcesPricing"
STATIC_WEIGHT = "staticWeight" STATIC_WEIGHT = "staticWeight"
DYNAMIC_RESOURCES = "dynamicResources" DYNAMIC_RESOURCES = "dynamicResources"
DATA_LOCALITY = "dataLocality" //感知数据位置，数据调度和计算调度协同，近数据调度
ENERGY_CONSUMPTION = "energyConsumption" //根据各集群总体能耗水平调度作业，优先选择能耗低的集群调度作业
) )
var ( var (

4
api/internal/storeLink/modelarts.go
View File

@ -63,7 +63,7 @@ func (m *ModelArtsLink) QueryImageList() (interface{}, error) {
return resp, nil return resp, nil
} }
func (m *ModelArtsLink) SubmitTask(imageId string, cmd string, envs []string, params []string, resourceId string, aiType string) (interface{}, error) { func (m *ModelArtsLink) SubmitTask(imageId string, cmd string, envs []string, params []string, resourceId string, datasetsId string, algorithmId string, aiType string) (interface{}, error) {
// modelArts提交任务 // modelArts提交任务
environments := make(map[string]string) environments := make(map[string]string)
parameters := make([]*modelarts.ParametersTrainJob, 0) parameters := make([]*modelarts.ParametersTrainJob, 0)
@ -162,7 +162,7 @@ func (m *ModelArtsLink) Execute(option *option.AiOption) (interface{}, error) {
if err != nil { if err != nil {
return nil, err return nil, err
} }
task, err := m.SubmitTask(option.ImageId, option.Cmd, option.Envs, option.Params, option.ResourceId, option.TaskType) task, err := m.SubmitTask(option.ImageId, option.Cmd, option.Envs, option.Params, option.ResourceId, option.DatasetsId, option.AlgorithmId, option.TaskType)
if err != nil { if err != nil {
return nil, err return nil, err
} }

34
api/internal/storeLink/octopus.go
View File

@ -108,7 +108,7 @@ func (o *OctopusLink) QueryImageList() (interface{}, error) {
return resp, nil return resp, nil
} }
func (o *OctopusLink) SubmitTask(imageId string, cmd string, envs []string, params []string, resourceId string, aiType string) (interface{}, error) { func (o *OctopusLink) SubmitTask(imageId string, cmd string, envs []string, params []string, resourceId string, datasetsId string, algorithmId string, aiType string) (interface{}, error) {
// octopus提交任务 // octopus提交任务
// python参数 // python参数
@ -227,7 +227,7 @@ func (o *OctopusLink) Execute(option *option.AiOption) (interface{}, error) {
if err != nil { if err != nil {
return nil, err return nil, err
} }
task, err := o.SubmitTask(option.ImageId, option.Cmd, option.Envs, option.Params, option.ResourceId, option.TaskType) task, err := o.SubmitTask(option.ImageId, option.Cmd, option.Envs, option.Params, option.ResourceId, option.DatasetsId, option.AlgorithmId, option.TaskType)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -239,6 +239,10 @@ func (o *OctopusLink) GenerateSubmitParams(option *option.AiOption) error {
if err != nil { if err != nil {
return err return err
} }
err = o.generateDatasetsId(option)
if err != nil {
return err
}
err = o.generateImageId(option) err = o.generateImageId(option)
if err != nil { if err != nil {
return err return err
@ -259,10 +263,34 @@ func (o *OctopusLink) GenerateSubmitParams(option *option.AiOption) error {
} }
func (o *OctopusLink) generateResourceId(option *option.AiOption) error { func (o *OctopusLink) generateResourceId(option *option.AiOption) error {
return nil return nil
} }
func (o *OctopusLink) generateDatasetsId(option *option.AiOption) error {
if option.DatasetsName == "" {
return errors.New("DatasetsName not set")
}
req := &octopus.GetMyDatasetListReq{
Platform: o.platform,
PageIndex: o.pageIndex,
PageSize: o.pageSize,
}
resp, err := o.svcCtx.OctopusRpc.GetMyDatasetList(o.ctx, req)
if err != nil {
return err
}
if !resp.Success {
return errors.New("failed to get DatasetsId")
}
for _, dataset := range resp.Payload.Datasets {
if dataset.Name == option.DatasetsName {
option.DatasetsId = dataset.Id
return nil
}
}
return errors.New("failed to get DatasetsId")
}
func (o *OctopusLink) generateImageId(option *option.AiOption) error { func (o *OctopusLink) generateImageId(option *option.AiOption) error {
return nil return nil

2
api/internal/storeLink/shuguangHpc.go
View File

@ -144,7 +144,7 @@ func (s ShuguangHpc) QueryImageList() (interface{}, error) {
return nil, nil return nil, nil
} }
func (s ShuguangHpc) SubmitTask(imageId string, cmd string, envs []string, params []string, resourceId string, aiType string) (interface{}, error) { func (s ShuguangHpc) SubmitTask(imageId string, cmd string, envs []string, params []string, resourceId string, datasetsId string, algorithmId string, aiType string) (interface{}, error) {
// shuguangHpc提交任务 // shuguangHpc提交任务
//判断是否resourceId匹配自定义资源Id //判断是否resourceId匹配自定义资源Id

4
api/internal/storeLink/shuguangai.go
View File

@ -141,7 +141,7 @@ func (s *ShuguangAi) SubmitTensorflowTask(imageId string, cmd string, envs []str
return nil, nil return nil, nil
} }
func (s *ShuguangAi) SubmitTask(imageId string, cmd string, envs []string, params []string, resourceId string, aiType string) (interface{}, error) { func (s *ShuguangAi) SubmitTask(imageId string, cmd string, envs []string, params []string, resourceId string, datasetsId string, algorithmId string, aiType string) (interface{}, error) {
// shuguangAi提交任务 // shuguangAi提交任务
switch aiType { switch aiType {
case PYTORCH_TASK: case PYTORCH_TASK:
@ -248,7 +248,7 @@ func (s *ShuguangAi) Execute(option *option.AiOption) (interface{}, error) {
if err != nil { if err != nil {
return nil, err return nil, err
} }
task, err := s.SubmitTask(option.ImageId, option.Cmd, option.Envs, option.Params, option.ResourceId, option.TaskType) task, err := s.SubmitTask(option.ImageId, option.Cmd, option.Envs, option.Params, option.ResourceId, option.DatasetsId, option.AlgorithmId, option.TaskType)
if err != nil { if err != nil {
return nil, err return nil, err
} }

2
api/internal/storeLink/storeLink.go
View File

@ -33,7 +33,7 @@ type Linkage interface {
UploadImage(path string) (interface{}, error) UploadImage(path string) (interface{}, error)
DeleteImage(imageId string) (interface{}, error) DeleteImage(imageId string) (interface{}, error)
QueryImageList() (interface{}, error) QueryImageList() (interface{}, error)
SubmitTask(imageId string, cmd string, envs []string, params []string, resourceId string, aiType string) (interface{}, error) SubmitTask(imageId string, cmd string, envs []string, params []string, resourceId string, datasetsId string, algorithmId string, aiType string) (interface{}, error)
QueryTask(taskId string) (interface{}, error) QueryTask(taskId string) (interface{}, error)
QuerySpecs() (interface{}, error) QuerySpecs() (interface{}, error)
DeleteTask(taskId string) (interface{}, error) DeleteTask(taskId string) (interface{}, error)