zhangweiii
/
pcm-coordinator
forked from JointCloud/pcm-coordinator
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix runtask api types

This commit is contained in:
tzwang 2025-02-26 17:37:32 +08:00
parent db9d21bfa1
commit 0abdf06f2f
2 changed files with 88 additions and 81 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
desc/schedule/pcm-schedule.api
View File

@ -238,7 +238,7 @@ type (
// 启动任务（资源已就绪）:/runTask // 启动任务（资源已就绪）:/runTask
RunTaskReq { RunTaskReq {
TaskID int64 `json:"taskID"` TaskID int64 `json:"taskID"`
ScheduledDatas []*DataScheduleResults `json:"scheduledDatas"` ScheduledDatas []*DataScheduleResults `json:"scheduledDatas,optional"`
} }
DataScheduleResults { DataScheduleResults {

167
internal/logic/schedule/scheduleruntasklogic.go
View File

@ -126,101 +126,108 @@ func (l *ScheduleRunTaskLogic) SaveResult(task *models.Task, results []*schedule
func updateClustersByScheduledDatas(taskId int64, clustersWithDataDistributes *ClustersWithDataDistributes, scheduledDatas []*types.DataScheduleResults) ([]*strategy.AssignedCluster, error) { func updateClustersByScheduledDatas(taskId int64, clustersWithDataDistributes *ClustersWithDataDistributes, scheduledDatas []*types.DataScheduleResults) ([]*strategy.AssignedCluster, error) {
assignedClusters := make([]*strategy.AssignedCluster, 0) assignedClusters := make([]*strategy.AssignedCluster, 0)
// handle pass-in scheduledDatas
for _, cluster := range clustersWithDataDistributes.Clusters { if len(scheduledDatas) == 0 {
for _, data := range scheduledDatas { for _, cluster := range clustersWithDataDistributes.Clusters {
switch data.DataType { assignedClusters = append(assignedClusters, cluster)
case "dataset": }
for _, result := range data.Results { } else {
if !result.Status { // handle pass-in scheduledDatas
continue for _, cluster := range clustersWithDataDistributes.Clusters {
} for _, data := range scheduledDatas {
for _, c := range result.Clusters { switch data.DataType {
if cluster.ClusterId == c.ClusterID { case "dataset":
if c.JsonData == "" { for _, result := range data.Results {
continue if !result.Status {
continue
}
for _, c := range result.Clusters {
if cluster.ClusterId == c.ClusterID {
if c.JsonData == "" {
continue
}
jsonData := struct {
Name string `json:"name"`
Id string `json:"id"`
}{}
err := json.Unmarshal([]byte(c.JsonData), &jsonData)
if err != nil {
return nil, fmt.Errorf("pass-in jsonData convert failed, task %d, cluster %s, datatype %s", taskId, cluster.ClusterId, "dataset")
}
cluster.DatasetId = jsonData.Id
} }
jsonData := struct {
Name string `json:"name"`
Id string `json:"id"`
}{}
err := json.Unmarshal([]byte(c.JsonData), &jsonData)
if err != nil {
return nil, fmt.Errorf("pass-in jsonData convert failed, task %d, cluster %s, datatype %s", taskId, cluster.ClusterId, "dataset")
}
cluster.DatasetId = jsonData.Id
} }
} }
} case "image":
case "image": for _, result := range data.Results {
for _, result := range data.Results { if !result.Status {
if !result.Status { continue
continue }
} for _, c := range result.Clusters {
for _, c := range result.Clusters { if cluster.ClusterId == c.ClusterID {
if cluster.ClusterId == c.ClusterID { if c.JsonData == "" {
if c.JsonData == "" { continue
continue }
jsonData := struct {
Name string `json:"name"`
Id string `json:"id"`
}{}
err := json.Unmarshal([]byte(c.JsonData), &jsonData)
if err != nil {
return nil, fmt.Errorf("pass-in jsonData convert failed, task %d, cluster %s, datatype %s", taskId, cluster.ClusterId, "image")
}
cluster.ImageId = jsonData.Id
} }
jsonData := struct {
Name string `json:"name"`
Id string `json:"id"`
}{}
err := json.Unmarshal([]byte(c.JsonData), &jsonData)
if err != nil {
return nil, fmt.Errorf("pass-in jsonData convert failed, task %d, cluster %s, datatype %s", taskId, cluster.ClusterId, "image")
}
cluster.ImageId = jsonData.Id
} }
} }
} case "code":
case "code": for _, result := range data.Results {
for _, result := range data.Results { if !result.Status {
if !result.Status { continue
continue }
} for _, c := range result.Clusters {
for _, c := range result.Clusters { if cluster.ClusterId == c.ClusterID {
if cluster.ClusterId == c.ClusterID { if c.JsonData == "" {
if c.JsonData == "" { continue
continue }
jsonData := struct {
Name string `json:"name"`
Id string `json:"id"`
}{}
err := json.Unmarshal([]byte(c.JsonData), &jsonData)
if err != nil {
return nil, fmt.Errorf("pass-in jsonData convert failed, task %d, cluster %s, datatype %s", taskId, cluster.ClusterId, "code")
}
cluster.CodeId = jsonData.Id
} }
jsonData := struct {
Name string `json:"name"`
Id string `json:"id"`
}{}
err := json.Unmarshal([]byte(c.JsonData), &jsonData)
if err != nil {
return nil, fmt.Errorf("pass-in jsonData convert failed, task %d, cluster %s, datatype %s", taskId, cluster.ClusterId, "code")
}
cluster.CodeId = jsonData.Id
} }
} }
} case "model":
case "model": for _, result := range data.Results {
for _, result := range data.Results { if !result.Status {
if !result.Status { continue
continue }
} for _, c := range result.Clusters {
for _, c := range result.Clusters { if cluster.ClusterId == c.ClusterID {
if cluster.ClusterId == c.ClusterID { if c.JsonData == "" {
if c.JsonData == "" { continue
continue }
jsonData := struct {
Name string `json:"name"`
Id string `json:"id"`
}{}
err := json.Unmarshal([]byte(c.JsonData), &jsonData)
if err != nil {
return nil, fmt.Errorf("pass-in jsonData convert failed, task %d, cluster %s, datatype %s", taskId, cluster.ClusterId, "model")
}
cluster.ModelId = jsonData.Id
} }
jsonData := struct {
Name string `json:"name"`
Id string `json:"id"`
}{}
err := json.Unmarshal([]byte(c.JsonData), &jsonData)
if err != nil {
return nil, fmt.Errorf("pass-in jsonData convert failed, task %d, cluster %s, datatype %s", taskId, cluster.ClusterId, "model")
}
cluster.ModelId = jsonData.Id
} }
} }
} }
} }
assignedClusters = append(assignedClusters, cluster)
} }
assignedClusters = append(assignedClusters, cluster)
} }
// handle db yaml clustersWithDataDistributes // handle db yaml clustersWithDataDistributes