diff --git a/pkgs/prescheduler/calc_score.go b/pkgs/prescheduler/calc_score.go index 50a4d0a..68cc801 100644 --- a/pkgs/prescheduler/calc_score.go +++ b/pkgs/prescheduler/calc_score.go @@ -13,9 +13,9 @@ import ( mgrmq "gitlink.org.cn/cloudream/scheduler/common/pkgs/mq/manager" ) -func (s *DefaultPreScheduler) calcResourceScore(job *schsdk.NormalJobInfo, allCCs map[schsdk.CCID]*candidate) error { +func (s *DefaultPreScheduler) calcResourceScore(jobResource schsdk.JobResourcesInfo, allCCs map[schsdk.CCID]*candidate) error { for _, cc := range allCCs { - res, err := s.calcOneResourceScore(job.Resources, &cc.CC) + res, err := s.calcOneResourceScore(jobResource, &cc.CC) if err != nil { return err } diff --git a/pkgs/prescheduler/default_prescheduler.go b/pkgs/prescheduler/default_prescheduler.go index b30d461..98f6e03 100644 --- a/pkgs/prescheduler/default_prescheduler.go +++ b/pkgs/prescheduler/default_prescheduler.go @@ -183,7 +183,7 @@ func (s *DefaultPreScheduler) ScheduleJobSet(info *schsdk.JobSetInfo) (*jobmod.J // 经过排序后，按顺序生成调度方案 for _, job := range schJobs { if norJob, ok := job.Job.(*schsdk.NormalJobInfo); ok { - scheme, err := s.scheduleForNormalJob(info, job, ccs, jobSetScheme.JobSchemes) + scheme, err := s.scheduleForNormalOrMultiJob(info, job, ccs, jobSetScheme.JobSchemes) if err != nil { return nil, nil, err } @@ -191,7 +191,19 @@ func (s *DefaultPreScheduler) ScheduleJobSet(info *schsdk.JobSetInfo) (*jobmod.J jobSetScheme.JobSchemes[job.Job.GetLocalJobID()] = *scheme // 检查数据文件的配置项，生成上传文件方案 - s.fillNormarlJobLocalUploadScheme(norJob, scheme.TargetCCID, filesUploadSchemes, ccs) + s.fillNormarlJobLocalUploadScheme(norJob.Files, scheme.TargetCCID, filesUploadSchemes, ccs) + } + + if mulJob, ok := job.Job.(*schsdk.MultiInstanceJobInfo); ok { + scheme, err := s.scheduleForNormalOrMultiJob(info, job, ccs, jobSetScheme.JobSchemes) + if err != nil { + return nil, nil, err + } + + jobSetScheme.JobSchemes[job.Job.GetLocalJobID()] = *scheme + + // 检查数据文件的配置项，生成上传文件方案 + s.fillNormarlJobLocalUploadScheme(mulJob.Files, scheme.TargetCCID, filesUploadSchemes, ccs) } // 回源任务目前不需要生成调度方案 @@ -237,13 +249,13 @@ func (s *DefaultPreScheduler) ScheduleJob(instJobInfo *schsdk.InstanceJobInfo) ( job := &schedulingJob{ Job: info, } - scheme, err := s.scheduleForNormalJob2(job, ccs) + scheme, err := s.scheduleForSingleJob(job, ccs) if err != nil { return nil, nil, err } // 检查数据文件的配置项，生成上传文件方案 - s.fillNormarlJobLocalUploadScheme(info, scheme.TargetCCID, filesUploadSchemes, ccs) + s.fillNormarlJobLocalUploadScheme(info.Files, scheme.TargetCCID, filesUploadSchemes, ccs) return scheme, &schsdk.JobFilesUploadScheme{ LocalFileSchemes: lo.Values(filesUploadSchemes), @@ -303,7 +315,7 @@ func (s *DefaultPreScheduler) orderByAfters(jobs []*schedulingJob) ([]*schedulin return orderedJob, true } -func (s *DefaultPreScheduler) scheduleForNormalJob(jobSet *schsdk.JobSetInfo, job *schedulingJob, ccs map[schsdk.CCID]schmod.ComputingCenter, jobSchemes map[string]jobmod.JobScheduleScheme) (*jobmod.JobScheduleScheme, error) { +func (s *DefaultPreScheduler) scheduleForNormalOrMultiJob(jobSet *schsdk.JobSetInfo, job *schedulingJob, ccs map[schsdk.CCID]schmod.ComputingCenter, jobSchemes map[string]jobmod.JobScheduleScheme) (*jobmod.JobScheduleScheme, error) { allCCs := make(map[schsdk.CCID]*candidate) // 初始化备选节点信息 @@ -334,16 +346,28 @@ func (s *DefaultPreScheduler) scheduleForNormalJob(jobSet *schsdk.JobSetInfo, jo allCCs[cc.CCID] = caNode } - norJob := job.Job.(*schsdk.NormalJobInfo) + //norJob := job.Job.(*schsdk.NormalJobInfo) + + var jobFiles *schsdk.JobFilesInfo + var jobResource *schsdk.JobResourcesInfo + + switch runningJob := job.Job.(type) { + case *schsdk.NormalJobInfo: + jobFiles = &runningJob.Files + jobResource = &runningJob.Resources + case *schsdk.MultiInstanceJobInfo: + jobFiles = &runningJob.Files + jobResource = &runningJob.Resources + } // 计算文件占有量得分 - err := s.calcFileScore(norJob.Files, allCCs) + err := s.calcFileScore(*jobFiles, allCCs) if err != nil { return nil, err } // 计算资源余量得分 - err = s.calcResourceScore(norJob, allCCs) + err = s.calcResourceScore(*jobResource, allCCs) if err != nil { return nil, err } @@ -356,11 +380,11 @@ func (s *DefaultPreScheduler) scheduleForNormalJob(jobSet *schsdk.JobSetInfo, jo return nil, ErrNoAvailableScheme } - scheme := s.makeSchemeForNode(norJob, targetNode) + scheme := s.makeSchemeForNode(jobFiles, targetNode) return &scheme, nil } -func (s *DefaultPreScheduler) scheduleForNormalJob2(job *schedulingJob, ccs map[schsdk.CCID]schmod.ComputingCenter) (*jobmod.JobScheduleScheme, error) { +func (s *DefaultPreScheduler) scheduleForSingleJob(job *schedulingJob, ccs map[schsdk.CCID]schmod.ComputingCenter) (*jobmod.JobScheduleScheme, error) { allCCs := make(map[schsdk.CCID]*candidate) // 初始化备选节点信息 @@ -372,16 +396,28 @@ func (s *DefaultPreScheduler) scheduleForNormalJob2(job *schedulingJob, ccs map[ allCCs[cc.CCID] = caNode } - norJob := job.Job.(*schsdk.NormalJobInfo) + //norJob := job.Job.(*schsdk.NormalJobInfo) + + var jobFiles *schsdk.JobFilesInfo + var jobResource *schsdk.JobResourcesInfo + + switch runningJob := job.Job.(type) { + case *schsdk.NormalJobInfo: + jobFiles = &runningJob.Files + jobResource = &runningJob.Resources + case *schsdk.MultiInstanceJobInfo: + jobFiles = &runningJob.Files + jobResource = &runningJob.Resources + } // 计算文件占有量得分 - err := s.calcFileScore(norJob.Files, allCCs) + err := s.calcFileScore(*jobFiles, allCCs) if err != nil { return nil, err } // 计算资源余量得分 - err = s.calcResourceScore(norJob, allCCs) + err = s.calcResourceScore(*jobResource, allCCs) if err != nil { return nil, err } @@ -394,12 +430,12 @@ func (s *DefaultPreScheduler) scheduleForNormalJob2(job *schedulingJob, ccs map[ return nil, ErrNoAvailableScheme } - scheme := s.makeSchemeForNode(norJob, targetNode) + scheme := s.makeSchemeForNode(jobFiles, targetNode) return &scheme, nil } -func (s *DefaultPreScheduler) fillNormarlJobLocalUploadScheme(norJob *schsdk.NormalJobInfo, targetCCID schsdk.CCID, schemes map[string]schsdk.LocalFileUploadScheme, ccs map[schsdk.CCID]schmod.ComputingCenter) { - if localFile, ok := norJob.Files.Dataset.(*schsdk.LocalJobFileInfo); ok { +func (s *DefaultPreScheduler) fillNormarlJobLocalUploadScheme(files schsdk.JobFilesInfo, targetCCID schsdk.CCID, schemes map[string]schsdk.LocalFileUploadScheme, ccs map[schsdk.CCID]schmod.ComputingCenter) { + if localFile, ok := files.Dataset.(*schsdk.LocalJobFileInfo); ok { if _, ok := schemes[localFile.LocalPath]; !ok { cdsNodeID := ccs[targetCCID].CDSNodeID schemes[localFile.LocalPath] = schsdk.LocalFileUploadScheme{ @@ -409,7 +445,7 @@ func (s *DefaultPreScheduler) fillNormarlJobLocalUploadScheme(norJob *schsdk.Nor } } - if localFile, ok := norJob.Files.Code.(*schsdk.LocalJobFileInfo); ok { + if localFile, ok := files.Code.(*schsdk.LocalJobFileInfo); ok { if _, ok := schemes[localFile.LocalPath]; !ok { cdsNodeID := ccs[targetCCID].CDSNodeID schemes[localFile.LocalPath] = schsdk.LocalFileUploadScheme{ @@ -419,7 +455,7 @@ func (s *DefaultPreScheduler) fillNormarlJobLocalUploadScheme(norJob *schsdk.Nor } } - if localFile, ok := norJob.Files.Image.(*schsdk.LocalJobFileInfo); ok { + if localFile, ok := files.Image.(*schsdk.LocalJobFileInfo); ok { if _, ok := schemes[localFile.LocalPath]; !ok { cdsNodeID := ccs[targetCCID].CDSNodeID schemes[localFile.LocalPath] = schsdk.LocalFileUploadScheme{ @@ -430,26 +466,26 @@ func (s *DefaultPreScheduler) fillNormarlJobLocalUploadScheme(norJob *schsdk.Nor } } -func (s *DefaultPreScheduler) makeSchemeForNode(job *schsdk.NormalJobInfo, targetCC *candidate) jobmod.JobScheduleScheme { +func (s *DefaultPreScheduler) makeSchemeForNode(jobFiles *schsdk.JobFilesInfo, targetCC *candidate) jobmod.JobScheduleScheme { scheme := jobmod.JobScheduleScheme{ TargetCCID: targetCC.CC.CCID, } // TODO 根据实际情况选择Move或者Load - if _, ok := job.Files.Dataset.(*schsdk.PackageJobFileInfo); ok && !targetCC.Files.Dataset.IsLoaded { + if _, ok := jobFiles.Dataset.(*schsdk.PackageJobFileInfo); ok && !targetCC.Files.Dataset.IsLoaded { scheme.Dataset.Action = jobmod.ActionLoad } else { scheme.Dataset.Action = jobmod.ActionNo } - if _, ok := job.Files.Code.(*schsdk.PackageJobFileInfo); ok && !targetCC.Files.Code.IsLoaded { + if _, ok := jobFiles.Code.(*schsdk.PackageJobFileInfo); ok && !targetCC.Files.Code.IsLoaded { scheme.Code.Action = jobmod.ActionLoad } else { scheme.Code.Action = jobmod.ActionNo } - if _, ok := job.Files.Image.(*schsdk.PackageJobFileInfo); ok && !targetCC.Files.Image.IsLoaded { + if _, ok := jobFiles.Image.(*schsdk.PackageJobFileInfo); ok && !targetCC.Files.Image.IsLoaded { scheme.Image.Action = jobmod.ActionImportImage } else { scheme.Image.Action = jobmod.ActionNo diff --git a/sdks/scheduler/models.go b/sdks/scheduler/models.go index 8f8e487..45282bc 100644 --- a/sdks/scheduler/models.go +++ b/sdks/scheduler/models.go @@ -58,6 +58,16 @@ type NormalJobInfo struct { Services JobServicesInfo `json:"services"` } +type CommonJobInfo struct { + serder.Metadata `union:"Normal"` + JobInfoBase + Type string `json:"type"` + Files JobFilesInfo `json:"files"` + Runtime JobRuntimeInfo `json:"runtime"` + Resources JobResourcesInfo `json:"resources"` + Services JobServicesInfo `json:"services"` +} + type DataReturnJobInfo struct { serder.Metadata `union:"DataReturn"` JobInfoBase diff --git a/sdks/unifyops/unifyops.go b/sdks/unifyops/unifyops.go index 97e64a7..64bb7ad 100644 --- a/sdks/unifyops/unifyops.go +++ b/sdks/unifyops/unifyops.go @@ -223,41 +223,127 @@ func (c *Client) GetMemoryData(node GetOneResourceDataReq) (*MemoryResourceData, } func (c *Client) GetIndicatorData(node GetOneResourceDataReq) (*[]ResourceData, error) { - url, err := url.JoinPath(c.baseURL, "/cmdb/resApi/getIndicatorData") - if err != nil { - return nil, err + //url, err := url.JoinPath(c.baseURL, "/cmdb/resApi/getIndicatorData") + //if err != nil { + // return nil, err + //} + //resp, err := myhttp.PostJSON(url, myhttp.RequestParam{ + // Body: node, + //}) + //if err != nil { + // return nil, err + //} + // + //contType := resp.Header.Get("Content-Type") + //if strings.Contains(contType, myhttp.ContentTypeJSON) { + // + // var codeResp response[[]map[string]any] + // if err := serder.JSONToObjectStream(resp.Body, &codeResp); err != nil { + // return nil, fmt.Errorf("parsing response: %w", err) + // } + // + // if codeResp.Code != CORRECT_CODE { + // return nil, codeResp.ToError() + // } + // + // var ret []ResourceData + // for _, mp := range codeResp.Data { + // var data ResourceData + // err := serder.MapToObject(mp, &data) + // if err != nil { + // return nil, err + // } + // ret = append(ret, data) + // } + // + // return &ret, nil + //} + // + //return nil, fmt.Errorf("unknow response content type: %s", contType) + + return mockData() +} + +func mockData() (*[]ResourceData, error) { + var ret []ResourceData + + cpuResourceData := CPUResourceData{ + Name: ResourceTypeCPU, + Total: UnitValue[int64]{ + Value: 100, + Unit: "", + }, + Available: UnitValue[int64]{ + Value: 100, + Unit: "", + }, } - resp, err := myhttp.PostJSON(url, myhttp.RequestParam{ - Body: node, - }) - if err != nil { - return nil, err + ret = append(ret, &cpuResourceData) + + npuResourceData := NPUResourceData{ + Name: ResourceTypeNPU, + Total: UnitValue[int64]{ + Value: 100, + Unit: "", + }, + Available: UnitValue[int64]{ + Value: 100, + Unit: "", + }, } - - contType := resp.Header.Get("Content-Type") - if strings.Contains(contType, myhttp.ContentTypeJSON) { - - var codeResp response[[]map[string]any] - if err := serder.JSONToObjectStream(resp.Body, &codeResp); err != nil { - return nil, fmt.Errorf("parsing response: %w", err) - } - - if codeResp.Code != CORRECT_CODE { - return nil, codeResp.ToError() - } - - var ret []ResourceData - for _, mp := range codeResp.Data { - var data ResourceData - err := serder.MapToObject(mp, &data) - if err != nil { - return nil, err - } - ret = append(ret, data) - } - - return &ret, nil + ret = append(ret, &npuResourceData) + + gpuResourceData := GPUResourceData{ + Name: ResourceTypeGPU, + Total: UnitValue[int64]{ + Value: 100, + Unit: "", + }, + Available: UnitValue[int64]{ + Value: 100, + Unit: "", + }, + } + ret = append(ret, &gpuResourceData) + + mluResourceData := MLUResourceData{ + Name: ResourceTypeMLU, + Total: UnitValue[int64]{ + Value: 100, + Unit: "", + }, + Available: UnitValue[int64]{ + Value: 100, + Unit: "", + }, } + ret = append(ret, &mluResourceData) + + storageResourceData := StorageResourceData{ + Name: ResourceTypeStorage, + Total: UnitValue[float64]{ + Value: 100, + Unit: "GB", + }, + Available: UnitValue[float64]{ + Value: 100, + Unit: "GB", + }, + } + ret = append(ret, &storageResourceData) + + memoryResourceData := MemoryResourceData{ + Name: ResourceTypeMemory, + Total: UnitValue[float64]{ + Value: 100, + Unit: "GB", + }, + Available: UnitValue[float64]{ + Value: 100, + Unit: "GB", + }, + } + ret = append(ret, &memoryResourceData) - return nil, fmt.Errorf("unknow response content type: %s", contType) + return &ret, nil } diff --git a/utils/config/config.go b/utils/config/config.go index 7a9890b..4f0d7ae 100644 --- a/utils/config/config.go +++ b/utils/config/config.go @@ -5,6 +5,7 @@ import ( "fmt" "os" "path/filepath" + "strings" "github.com/imdario/mergo" ) @@ -27,6 +28,18 @@ func DefaultLoad(modeulName string, defCfg interface{}) error { return err } + if strings.Contains(execPath, "scheduler") { + execPath = "D:\\Work\\Codes\\new\\workspace\\workspace\\scheduler\\common\\assets\\confs\\" + } + + if strings.Contains(execPath, "storage") { + execPath = "D:\\Work\\Codes\\new\\workspace\\workspace\\storage\\common\\assets\\confs\\" + } + + if strings.Contains(execPath, "gateway") { + execPath = "D:\\Work\\Codes\\new\\workspace\\workspace\\gateway\\assets\\confs\\" + } + // TODO 可以考虑根据环境变量读取不同的配置 configFilePath := filepath.Join(filepath.Dir(execPath), "..", "confs", fmt.Sprintf("%s.config.json", modeulName)) return Load(configFilePath, defCfg)