updated createinferencetasklogic

internal/logic/inference/createinferencetasklogic.go

@@ -2,7 +2,10 @@ package inference
 
 import (
 	"context"
+	"encoding/json"
+	"errors"
 	"fmt"
+	"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/entity"
 	"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/service/utils/task"
 	"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/strategy"
 
@@ -33,7 +36,7 @@ func (l *CreateInferenceTaskLogic) CreateInferenceTask(req *types.CreateInferenc
 		return nil, err
 	}
 
-	err = task.ValidateJobResources(req.JobResources, "")
+	err = task.ValidateJobResources(req.JobResources, "inference")
 	if err != nil {
 		return nil, err
 	}
@@ -53,8 +56,54 @@ func (l *CreateInferenceTaskLogic) CreateInferenceTask(req *types.CreateInferenc
 	return
 }
 
-func generateClusterInfosForTaskCreation() ([]*strategy.AssignedCluster, error) {
-	return nil, nil
+func generateClustersForTaskCreation(distributes types.DataDistribute, taskId string) ([]*strategy.AssignedCluster, error) {
+	var assignedClusters []*strategy.AssignedCluster
+	clusterMap := make(map[string]*strategy.AssignedCluster)
+
+	for _, distribute := range distributes.Model {
+		if len(distribute.Clusters) == 0 {
+			return nil, fmt.Errorf("model distribute: must specify at least one cluster")
+		}
+
+		for _, c := range distribute.Clusters {
+			if c.ClusterID == "" {
+				return nil, fmt.Errorf("model distribute: clusterId can not be empty")
+			}
+			cluster := &strategy.AssignedCluster{}
+			cluster.ClusterId = c.ClusterID
+			jsonData := entity.JsonData{}
+			err := json.Unmarshal([]byte(c.JsonData), &jsonData)
+			if err != nil {
+				return nil, fmt.Errorf("jsonData convert failed, task %d, cluster %s, datatype: %s", taskId, cluster.ClusterId, "Model")
+			}
+			if jsonData.Id == "" {
+				continue
+			}
+			cluster.ModelId = jsonData.Id
+			clusterMap[c.ClusterID] = cluster
+		}
+
+	}
+
+	for _, distribute := range distributes.Code {
+		if len(distribute.Clusters) == 0 {
+			return nil, errors.New("Code distribute must specify at least one cluster")
+		}
+		if distribute.Clusters[0].ClusterID == "" {
+			return nil, errors.New("Code distribute clusterId can not be empty")
+		}
+	}
+
+	for _, distribute := range distributes.Image {
+		if len(distribute.Clusters) == 0 {
+			return nil, errors.New("Image distribute must specify at least one cluster")
+		}
+		if distribute.Clusters[0].ClusterID == "" {
+			return nil, errors.New("Image distribute clusterId can not be empty")
+		}
+	}
+
+	return assignedClusters, nil
 }
 
 func validateRequest(resources types.JobResources, clusters []*strategy.AssignedCluster) error {

internal/logic/schedule/scheduleruntasklogic.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"github.com/zeromicro/go-zero/core/logx"
+	"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/entity"
 	"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/schedulers"
 	"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/schedulers/option"
 	"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/service/executor"
@@ -147,10 +148,7 @@ func updateClustersByScheduledDatas(taskId int64, clustersWithDataDistributes *C
 								if c.JsonData == "" {
 									continue
 								}
-								jsonData := struct {
-									Name string `json:"name"`
-									Id   string `json:"id"`
-								}{}
+								jsonData := entity.JsonData{}
 								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")
@@ -169,10 +167,7 @@ func updateClustersByScheduledDatas(taskId int64, clustersWithDataDistributes *C
 								if c.JsonData == "" {
 									continue
 								}
-								jsonData := struct {
-									Name string `json:"name"`
-									Id   string `json:"id"`
-								}{}
+								jsonData := entity.JsonData{}
 								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")
@@ -191,10 +186,7 @@ func updateClustersByScheduledDatas(taskId int64, clustersWithDataDistributes *C
 								if c.JsonData == "" {
 									continue
 								}
-								jsonData := struct {
-									Name string `json:"name"`
-									Id   string `json:"id"`
-								}{}
+								jsonData := entity.JsonData{}
 								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")
@@ -213,10 +205,7 @@ func updateClustersByScheduledDatas(taskId int64, clustersWithDataDistributes *C
 								if c.JsonData == "" {
 									continue
 								}
-								jsonData := struct {
-									Name string `json:"name"`
-									Id   string `json:"id"`
-								}{}
+								jsonData := entity.JsonData{}
 								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")
@@ -240,10 +229,7 @@ func updateClustersByScheduledDatas(taskId int64, clustersWithDataDistributes *C
 						if c.JsonData == "" {
 							continue
 						}
-						jsonData := struct {
-							Name string `json:"name"`
-							Id   string `json:"id"`
-						}{}
+						jsonData := entity.JsonData{}
 						err := json.Unmarshal([]byte(c.JsonData), &jsonData)
 						if err != nil {
 							return nil, fmt.Errorf("db yaml jsonData convert failed, task %d, cluster %s, datatype: %s", taskId, cluster.ClusterId, "dataset")
@@ -261,10 +247,7 @@ func updateClustersByScheduledDatas(taskId int64, clustersWithDataDistributes *C
 						if c.JsonData == "" {
 							continue
 						}
-						jsonData := struct {
-							Name string `json:"name"`
-							Id   string `json:"id"`
-						}{}
+						jsonData := entity.JsonData{}
 						err := json.Unmarshal([]byte(c.JsonData), &jsonData)
 						if err != nil {
 							return nil, fmt.Errorf("db yaml jsonData convert failed, task %d, cluster %s, datatype: %s", taskId, cluster.ClusterId, "image")
@@ -286,10 +269,7 @@ func updateClustersByScheduledDatas(taskId int64, clustersWithDataDistributes *C
 						if c.JsonData == "" {
 							continue
 						}
-						jsonData := struct {
-							Name string `json:"name"`
-							Id   string `json:"id"`
-						}{}
+						jsonData := entity.JsonData{}
 						err := json.Unmarshal([]byte(c.JsonData), &jsonData)
 						if err != nil {
 							return nil, fmt.Errorf("db yaml jsonData convert failed, task %d, cluster %s, datatype: %s", taskId, cluster.ClusterId, "code")
@@ -308,10 +288,7 @@ func updateClustersByScheduledDatas(taskId int64, clustersWithDataDistributes *C
 						if c.JsonData == "" {
 							continue
 						}
-						jsonData := struct {
-							Name string `json:"name"`
-							Id   string `json:"id"`
-						}{}
+						jsonData := entity.JsonData{}
 						err := json.Unmarshal([]byte(c.JsonData), &jsonData)
 						if err != nil {
 							return nil, fmt.Errorf("jsonData convert failed, task %d, cluster %s, datatype: %s", taskId, cluster.ClusterId, "model")

internal/scheduler/entity/entity.go

@@ -11,3 +11,8 @@ type Participant struct {
 	Name           string
 	Participant_id int64
 }
+
+type JsonData struct {
+	Name string `json:"name"`
+	Id   string `json:"id"`
+}