From c66d7b2fcd33237d15089e87f2c8949f3f966a15 Mon Sep 17 00:00:00 2001 From: tzwang Date: Thu, 5 Dec 2024 22:11:23 +0800 Subject: [PATCH] added createtask api --- desc/pcm.api | 3 + desc/schedule/pcm-schedule.api | 55 +++++++++++++++++++ go.mod | 2 +- go.sum | 4 +- internal/handler/routes.go | 5 ++ .../schedule/schedulecreatetaskhandler.go | 28 ++++++++++ .../logic/schedule/schedulecreatetasklogic.go | 31 +++++++++++ internal/scheduler/common/common.go | 7 +++ .../imageInference/imageInference.go | 10 +--- internal/storeLink/openi.go | 54 ++++++++++++++++-- internal/types/types.go | 53 ++++++++++++++++++ 11 files changed, 236 insertions(+), 16 deletions(-) create mode 100644 internal/handler/schedule/schedulecreatetaskhandler.go create mode 100644 internal/logic/schedule/schedulecreatetasklogic.go diff --git a/desc/pcm.api b/desc/pcm.api index 2613ea2c..49002681 100644 --- a/desc/pcm.api +++ b/desc/pcm.api @@ -940,6 +940,9 @@ service pcm { @handler QueryResourcesHandler get /schedule/queryResources (QueryResourcesReq) returns (QueryResourcesResp) + + @handler ScheduleCreateTaskHandler + post /schedule/createTask (CreateTaskReq) returns (CreateTaskResp) } @server( diff --git a/desc/schedule/pcm-schedule.api b/desc/schedule/pcm-schedule.api index 0f8e807c..de18ef60 100644 --- a/desc/schedule/pcm-schedule.api +++ b/desc/schedule/pcm-schedule.api @@ -151,6 +151,7 @@ type ( Balance float64 `json:"balance"` } + // 调度资源信息:/queryResources QueryResourcesReq{ AdapterId string `form:"adapterId"` } @@ -158,4 +159,58 @@ type ( QueryResourcesResp{ Data interface{} `json:"data"` } + + // 创建任务（下发资源）:/createTask + CreateTaskReq{ + Resources []*CreateTaskResource `json:"resources"` + DataDistributes DataDistribute `json:"dataDistributes"` + } + + CreateTaskResource { + Type string `json:"type"` + Req interface{} `json:"req"` + } + + DataDistribute { + Dataset []*DatasetDistribute `json:"dataset"` + Code []*CodeDistribute `json:"code"` + Image []*ImageDistribute `json:"image"` + Model []*ModelDistribute `json:"model"` + } + + DatasetDistribute { + DataName string `json:"dataName"` + PackageID int64 `json:"packageID"` + Clusters []int64 `json:"clusters"` + } + + CodeDistribute { + DataName string `json:"dataName"` + PackageID int64 `json:"packageID"` + Clusters []int64 `json:"clusters"` + } + + ImageDistribute { + DataName string `json:"dataName"` + PackageID int64 `json:"packageID"` + Clusters []int64 `json:"clusters"` + } + + ModelDistribute { + DataName string `json:"dataName"` + PackageID int64 `json:"packageID"` + Clusters []int64 `json:"clusters"` + } + + CreateTaskResp { + TaskID int64 `json:"taskID"` + ScheduleDatas []*ScheduleData `json:"scheduleDatas"` + } + + ScheduleData { + DataType string `json:"dataType"` // 数据类型：数据集、镜像.. + PackageID int64 `json:"packageID"` + StorageType string `json:"storageType"` // 目的地类型：如代码仓、JCS + ClusterIDs []int64 `json:"clusterIDs"` + } ) \ No newline at end of file diff --git a/go.mod b/go.mod index c3e2dd54..bde8a642 100644 --- a/go.mod +++ b/go.mod @@ -22,7 +22,7 @@ require ( gitlink.org.cn/JointCloud/pcm-hpc v0.0.0-20241125115811-72f3568255a4 gitlink.org.cn/JointCloud/pcm-modelarts v0.0.0-20240918011543-482dcd609877 gitlink.org.cn/JointCloud/pcm-octopus v0.0.0-20240817071412-44397870b110 - gitlink.org.cn/JointCloud/pcm-openi v0.0.0-20241205101654-eeb26c591155 + gitlink.org.cn/JointCloud/pcm-openi v0.0.0-20241205140615-e1b443983040 gitlink.org.cn/JointCloud/pcm-openstack v0.0.0-20240403033338-e7edabad4203 gitlink.org.cn/JointCloud/pcm-slurm v0.0.0-20240301080743-8b94bbaf57f5 go.opentelemetry.io/otel/trace v1.31.0 diff --git a/go.sum b/go.sum index 552586af..61f53d93 100644 --- a/go.sum +++ b/go.sum @@ -532,8 +532,8 @@ gitlink.org.cn/JointCloud/pcm-modelarts v0.0.0-20240918011543-482dcd609877 h1:a+ gitlink.org.cn/JointCloud/pcm-modelarts v0.0.0-20240918011543-482dcd609877/go.mod h1:/eOmBFZKWGoabG3sRVkVvIbLwsd2631k4jkUBR6x1AA= gitlink.org.cn/JointCloud/pcm-octopus v0.0.0-20240817071412-44397870b110 h1:GaXwr5sgDh0raHjUf9IewTvnRvajYea7zbLsaerYyXo= gitlink.org.cn/JointCloud/pcm-octopus v0.0.0-20240817071412-44397870b110/go.mod h1:QOD5+/l2D+AYBjF2h5T0mdJyfGAmF78QmeKdbBXbjLQ= -gitlink.org.cn/JointCloud/pcm-openi v0.0.0-20241205101654-eeb26c591155 h1:Vcm7KZH6EnsEYAtUTuDRJX2gXf7OhMZmzgnqc56mbtY= -gitlink.org.cn/JointCloud/pcm-openi v0.0.0-20241205101654-eeb26c591155/go.mod h1:oDJrr/TNbUCaVjI+RaOrUtGawD7UPAvp7U/oVgT2Dhc= +gitlink.org.cn/JointCloud/pcm-openi v0.0.0-20241205140615-e1b443983040 h1:dWUf3LDZO3l37FPswxf6vC+hzNT2Ym/G0PXvJDHg+w8= +gitlink.org.cn/JointCloud/pcm-openi v0.0.0-20241205140615-e1b443983040/go.mod h1:oDJrr/TNbUCaVjI+RaOrUtGawD7UPAvp7U/oVgT2Dhc= gitlink.org.cn/JointCloud/pcm-openstack v0.0.0-20240403033338-e7edabad4203 h1:s6PsZ1+bev294IWdZRlV7mnOwI1+UzFcldVW/BqhQzI= gitlink.org.cn/JointCloud/pcm-openstack v0.0.0-20240403033338-e7edabad4203/go.mod h1:i2rrbMQ+Fve345BY9Heh4MUqVTAimZQElQhzzRee5B8= gitlink.org.cn/JointCloud/pcm-slurm v0.0.0-20240301080743-8b94bbaf57f5 h1:+/5vnzkJBfMRnya1NrhOzlroUtRa5ePiYbPKlHLoLV0= diff --git a/internal/handler/routes.go b/internal/handler/routes.go index fcb77838..ef3ae436 100644 --- a/internal/handler/routes.go +++ b/internal/handler/routes.go @@ -1182,6 +1182,11 @@ func RegisterHandlers(server *rest.Server, serverCtx *svc.ServiceContext) { Path: "/schedule/queryResources", Handler: schedule.QueryResourcesHandler(serverCtx), }, + { + Method: http.MethodPost, + Path: "/schedule/createTask", + Handler: schedule.ScheduleCreateTaskHandler(serverCtx), + }, }, rest.WithPrefix("/pcm/v1"), ) diff --git a/internal/handler/schedule/schedulecreatetaskhandler.go b/internal/handler/schedule/schedulecreatetaskhandler.go new file mode 100644 index 00000000..5f9b5fa9 --- /dev/null +++ b/internal/handler/schedule/schedulecreatetaskhandler.go @@ -0,0 +1,28 @@ +package schedule + +import ( + "net/http" + + "github.com/zeromicro/go-zero/rest/httpx" + "gitlink.org.cn/JointCloud/pcm-coordinator/internal/logic/schedule" + "gitlink.org.cn/JointCloud/pcm-coordinator/internal/svc" + "gitlink.org.cn/JointCloud/pcm-coordinator/internal/types" +) + +func ScheduleCreateTaskHandler(svcCtx *svc.ServiceContext) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + var req types.CreateTaskReq + if err := httpx.Parse(r, &req); err != nil { + httpx.ErrorCtx(r.Context(), w, err) + return + } + + l := schedule.NewScheduleCreateTaskLogic(r.Context(), svcCtx) + resp, err := l.ScheduleCreateTask(&req) + if err != nil { + httpx.ErrorCtx(r.Context(), w, err) + } else { + httpx.OkJsonCtx(r.Context(), w, resp) + } + } +} diff --git a/internal/logic/schedule/schedulecreatetasklogic.go b/internal/logic/schedule/schedulecreatetasklogic.go new file mode 100644 index 00000000..80cd13c0 --- /dev/null +++ b/internal/logic/schedule/schedulecreatetasklogic.go @@ -0,0 +1,31 @@ +package schedule + +import ( + "context" + "fmt" + + "gitlink.org.cn/JointCloud/pcm-coordinator/internal/svc" + "gitlink.org.cn/JointCloud/pcm-coordinator/internal/types" + + "github.com/zeromicro/go-zero/core/logx" +) + +type ScheduleCreateTaskLogic struct { + logx.Logger + ctx context.Context + svcCtx *svc.ServiceContext +} + +func NewScheduleCreateTaskLogic(ctx context.Context, svcCtx *svc.ServiceContext) *ScheduleCreateTaskLogic { + return &ScheduleCreateTaskLogic{ + Logger: logx.WithContext(ctx), + ctx: ctx, + svcCtx: svcCtx, + } +} + +func (l *ScheduleCreateTaskLogic) ScheduleCreateTask(req *types.CreateTaskReq) (resp *types.CreateTaskResp, err error) { + // todo: add your logic here and delete this line + fmt.Println(req) + return +} diff --git a/internal/scheduler/common/common.go b/internal/scheduler/common/common.go index 3447b5f5..eafa966c 100644 --- a/internal/scheduler/common/common.go +++ b/internal/scheduler/common/common.go @@ -15,6 +15,7 @@ package common import ( + "github.com/go-resty/resty/v2" "math" "math/rand" "time" @@ -115,3 +116,9 @@ func ConcatMultipleSlices[T any](slices [][]T) []T { return result } + +func GetRestyRequest(timeoutSeconds int64) *resty.Request { + client := resty.New().SetTimeout(time.Duration(timeoutSeconds) * time.Second) + request := client.R() + return request +} diff --git a/internal/scheduler/service/inference/imageInference/imageInference.go b/internal/scheduler/service/inference/imageInference/imageInference.go index ed0a375d..51a079c5 100644 --- a/internal/scheduler/service/inference/imageInference/imageInference.go +++ b/internal/scheduler/service/inference/imageInference/imageInference.go @@ -3,8 +3,8 @@ package imageInference import ( "encoding/json" "errors" - "github.com/go-resty/resty/v2" "github.com/zeromicro/go-zero/core/logx" + "gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/common" "gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/database" "gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/schedulers/option" "gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/service/inference" @@ -476,7 +476,7 @@ func getInferResult(url string, file multipart.File, fileName string, clusterId return r, nil default: var res Res - req := GetRestyRequest(20) + req := common.GetRestyRequest(20) _, err := req. SetFileReader("file", fileName, file). SetResult(&res). @@ -513,12 +513,6 @@ func getInferResultModelarts(url string, file multipart.File, fileName string) ( return res.Result, nil } -func GetRestyRequest(timeoutSeconds int64) *resty.Request { - client := resty.New().SetTimeout(time.Duration(timeoutSeconds) * time.Second) - request := client.R() - return request -} - type Res struct { Result string `json:"result"` } diff --git a/internal/storeLink/openi.go b/internal/storeLink/openi.go index 648d7b14..b360679d 100644 --- a/internal/storeLink/openi.go +++ b/internal/storeLink/openi.go @@ -3,15 +3,36 @@ package storeLink import ( "context" "gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/service/collector" + "gitlink.org.cn/JointCloud/pcm-openi/common" + "gitlink.org.cn/JointCloud/pcm-openi/model" +) + +const ( + DEBUG = "DEBUG" + TRAIN = "TRAIN" + INFERENCE = "INFERENCE" + C2NET = "C2Net" + TESTREPO = "testrepo" +) + +// compute source +var ( + ComputeSource = []string{"GPU", "NPU", "GCU", "MLU", "DCU", "CPU", "ILUVATAR-GPGPU", "METAX-GPGPU"} ) type OpenI struct { - Host string + participantId int64 + host string + userName string + accessToken string } -func NewOpenI(host string) *OpenI { +func NewOpenI(host string, id int64, name string, token string) *OpenI { return &OpenI{ - Host: host, + host: host, + participantId: id, + userName: name, + accessToken: token, } } @@ -21,8 +42,31 @@ func (o OpenI) GetResourceStats(ctx context.Context) (*collector.ResourceStats, } func (o OpenI) GetDatasetsSpecs(ctx context.Context) ([]*collector.DatasetsSpecs, error) { - //TODO implement me - panic("implement me") + var specs []*collector.DatasetsSpecs + var url string + + for _ = range ComputeSource { + param := model.TaskCreationRequiredParam{} + resp := model.TaskCreationRequired{} + go func() { + req := common.GetRestyRequest(common.TIMEOUT) + _, err := req. + SetBody(param). + SetQueryParam(common.ACCESSTOKEN, o.accessToken). + SetResult(&resp). + Get(url) + + if err != nil { + return + } + + spec := &collector.DatasetsSpecs{} + specs = append(specs, spec) + }() + + } + + return specs, nil } func (o OpenI) GetAlgorithms(ctx context.Context) ([]*collector.Algorithm, error) { diff --git a/internal/types/types.go b/internal/types/types.go index 1c922a6a..981f8a0d 100644 --- a/internal/types/types.go +++ b/internal/types/types.go @@ -5878,6 +5878,59 @@ type QueryResourcesResp struct { Data interface{} `json:"data"` } +type CreateTaskReq struct { + Resources []*CreateTaskResource `json:"resources"` + DataDistributes DataDistribute `json:"dataDistributes"` +} + +type CreateTaskResource struct { + Type string `json:"type"` + Req interface{} `json:"req"` +} + +type DataDistribute struct { + Dataset []*DatasetDistribute `json:"dataset"` + Code []*CodeDistribute `json:"code"` + Image []*ImageDistribute `json:"image"` + Model []*ModelDistribute `json:"model"` +} + +type DatasetDistribute struct { + DataName string `json:"dataName"` + PackageID int64 `json:"packageID"` + Clusters []int64 `json:"clusters"` +} + +type CodeDistribute struct { + DataName string `json:"dataName"` + PackageID int64 `json:"packageID"` + Clusters []int64 `json:"clusters"` +} + +type ImageDistribute struct { + DataName string `json:"dataName"` + PackageID int64 `json:"packageID"` + Clusters []int64 `json:"clusters"` +} + +type ModelDistribute struct { + DataName string `json:"dataName"` + PackageID int64 `json:"packageID"` + Clusters []int64 `json:"clusters"` +} + +type CreateTaskResp struct { + TaskID int64 `json:"taskID"` + ScheduleDatas []*ScheduleData `json:"scheduleDatas"` +} + +type ScheduleData struct { + DataType string `json:"dataType"` // 数据类型：数据集、镜像.. + PackageID int64 `json:"packageID"` + StorageType string `json:"storageType"` // 目的地类型：如代码仓、JCS + ClusterIDs []int64 `json:"clusterIDs"` +} + type CreateAlertRuleReq struct { CLusterId string `json:"clusterId"` ClusterName string `json:"clusterName"`