Merge pull request 'updated imageinference logics' (#290) from tzwang/pcm-coordinator:master into master

Former-commit-id: eebb983c4b6290a88aa508f387cbebfb7f2ed25f

desc/inference/inference.api

@@ -3,6 +3,8 @@ syntax = "v1"
 type (
      /******************image inference*************************/
      DeployInstance {
+          Id string `json:"id"`
+          DeployTaskId string `json:"deployTaskId"`
           InstanceId string `json:"instanceId"`
           InstanceName string `json:"instanceName"`
           AdapterId string `json:"adapterId"`
@@ -13,6 +15,9 @@ type (
           ModelType string `json:"modelType"`
           InferCard string `json:"inferCard"`
           Status string `json:"status"`
+          CreateTime string `json:"createTime"`
+          UpdateTime string `json:"updateTime"`
+          ClusterType string `json:"clusterType"`
      }
 
      /******************image inference*************************/
@@ -30,13 +35,13 @@ type (
      /******************image inference*************************/
 
      ImageInferenceReq {
-          TaskName string `json:"taskName"`
+          TaskName string `form:"taskName"`
-          TaskDesc string `json:"taskDesc"`
+          TaskDesc string `form:"taskDesc"`
-          ModelType string `json:"modelType"`
+          ModelType string `form:"modelType"`
-          Instances []DeployInstance `json:"instances"`
+          InstanceIds []int64 `form:"instanceIds"`
-          Strategy string `json:"strategy,,optional"`
+          Strategy string `form:"strategy,,optional"`
-          StaticWeightMap map[string]int32 `json:"staticWeightMap,optional"`
+          StaticWeightMap map[string]map[string]int32 `form:"staticWeightMap,optional"`
-          Replica     int32     `json:"replicas,optional"`
+          Replica     int32     `form:"replicas,optional"`
      }
 
      ImageInferenceResp {

internal/logic/inference/imageinferencelogic.go

@@ -10,7 +10,9 @@ import (
 	"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/strategy"
 	"gitlink.org.cn/JointCloud/pcm-coordinator/internal/svc"
 	"gitlink.org.cn/JointCloud/pcm-coordinator/internal/types"
+	"gitlink.org.cn/JointCloud/pcm-coordinator/pkg/models"
 	"net/http"
+	"strconv"
 )
 
 type ImageInferenceLogic struct {
@@ -34,21 +36,24 @@ func NewImageInferenceLogic(ctx context.Context, svcCtx *svc.ServiceContext) *Im
 
 func (l *ImageInferenceLogic) ImageInfer(r *http.Request, req *types.ImageInferenceReq) (resp *types.ImageInferenceResp, err error) {
 	resp = &types.ImageInferenceResp{}
-	if len(req.Instances) == 0 {
+	if len(req.InstanceIds) == 0 {
 		return nil, errors.New("instances are empty")
 	}
 
-	opt := &option.InferOption{
+	var instanceList []*models.AiInferDeployInstance
-		TaskName: req.TaskName,
+	for _, id := range req.InstanceIds {
-		TaskDesc: req.TaskDesc,
+		instance, err := l.svcCtx.Scheduler.AiStorages.GetInferDeployInstanceById(id)
-		//AdapterId:       req.AdapterId,
+		if err != nil {
-		//AiClusterIds:    req.AiClusterIds,
+			return nil, err
-		//ModelName:       req.ModelName,
+		}
-		ModelType:       req.ModelType,
+		instanceList = append(instanceList, instance)
-		Strategy:        req.Strategy,
-		StaticWeightMap: req.StaticWeightMap,
 	}
 
+	if len(instanceList) == 0 {
+		return nil, errors.New("instances are empty")
+	}
+
+	// process uploaded images
 	var ts []*imageInference.ImageFile
 
 	uploadedFiles := r.MultipartForm.File
@@ -76,54 +81,66 @@ func (l *ImageInferenceLogic) ImageInfer(r *http.Request, req *types.ImageInfere
 		ts = append(ts, &t)
 	}
 
-	//_, ok := l.svcCtx.Scheduler.AiService.AiCollectorAdapterMap[opt.AdapterId]
+	//single adapter logic
-	//if !ok {
+	if len(req.StaticWeightMap) != 1 {
-	//	return nil, errors.New("AdapterId does not exist")
+		return nil, errors.New("staticWeightMap != 1")
-	//}
-	//
-
-	var cs []*strategy.AssignedCluster
-	var adapterName string
-	if opt.Strategy != "" {
-		var strat strategy.Strategy
-		switch opt.Strategy {
-		case strategy.STATIC_WEIGHT:
-			strat = strategy.NewStaticWeightStrategy(opt.StaticWeightMap, int32(len(ts)))
-			if err != nil {
-				return nil, err
-			}
-		default:
-			return nil, errors.New("no strategy has been chosen")
-		}
-		clusters, err := strat.Schedule()
-		if err != nil {
-			return nil, err
-		}
-
-		if clusters == nil || len(clusters) == 0 {
-			return nil, errors.New("clusters is nil")
-		}
-
-		for i := len(clusters) - 1; i >= 0; i-- {
-			if clusters[i].Replicas == 0 {
-				clusters = append(clusters[:i], clusters[i+1:]...)
-			}
-		}
-
-		name, err := l.svcCtx.Scheduler.AiStorages.GetAdapterNameById(opt.AdapterId)
-		if err != nil {
-			return nil, err
-		}
-		adapterName = name
 	}
 
-	//else {
+	adapterId := strconv.FormatInt(instanceList[0].AdapterId, 10)
-	//	for i, instance := range req.Instances {
+	staticWeightMap, ok := req.StaticWeightMap[adapterId]
-	//
+	if !ok {
-	//	}
+		return nil, errors.New("set staticWeightMap failed")
-	//}
+	}
 
-	imageInfer, err := imageInference.New(imageInference.NewImageClassification(), ts, cs, req.Instances, opt, l.svcCtx.Scheduler.AiStorages, l.svcCtx.Scheduler.AiService.InferenceAdapterMap, adapterName)
+	// create InferOption
+	opt := &option.InferOption{
+		TaskName:  req.TaskName,
+		TaskDesc:  req.TaskDesc,
+		AdapterId: adapterId,
+		//AiClusterIds:    req.AiClusterIds,
+		//ModelName:       req.ModelName,
+		ModelType:       req.ModelType,
+		Strategy:        req.Strategy,
+		StaticWeightMap: staticWeightMap,
+	}
+
+	adapterName, err := l.svcCtx.Scheduler.AiStorages.GetAdapterNameById(opt.AdapterId)
+	if err != nil {
+		return nil, err
+	}
+
+	// set strategy
+	if opt.Strategy != "" {
+		return nil, errors.New("strategy is empty")
+	}
+
+	var strat strategy.Strategy
+	switch opt.Strategy {
+	case strategy.STATIC_WEIGHT:
+		strat = strategy.NewStaticWeightStrategy(opt.StaticWeightMap, int32(len(ts)))
+		if err != nil {
+			return nil, err
+		}
+	default:
+		return nil, errors.New("no strategy has been chosen")
+	}
+	clusters, err := strat.Schedule()
+	if err != nil {
+		return nil, err
+	}
+
+	if clusters == nil || len(clusters) == 0 {
+		return nil, errors.New("clusters is nil")
+	}
+
+	for i := len(clusters) - 1; i >= 0; i-- {
+		if clusters[i].Replicas == 0 {
+			clusters = append(clusters[:i], clusters[i+1:]...)
+		}
+	}
+
+	// create inference struct
+	imageInfer, err := imageInference.New(imageInference.NewImageClassification(), ts, clusters, instanceList, opt, l.svcCtx.Scheduler.AiStorages, l.svcCtx.Scheduler.AiService.InferenceAdapterMap, adapterName)
 	if err != nil {
 		return nil, err
 	}

internal/scheduler/service/inference/imageInference/imageInference.go

@@ -46,7 +46,7 @@ type ImageInference struct {
 	inference    IImageInference
 	files        []*ImageFile
 	clusters     []*strategy.AssignedCluster
-	instances    []types.DeployInstance
+	instances    []*models.AiInferDeployInstance
 	opt          *option.InferOption
 	storage      *database.AiStorage
 	inferAdapter map[string]map[string]inference.ICluster
@@ -58,7 +58,7 @@ func New(
 	inference IImageInference,
 	files []*ImageFile,
 	clusters []*strategy.AssignedCluster,
-	instances []types.DeployInstance,
+	instances []*models.AiInferDeployInstance,
 	opt *option.InferOption,
 	storage *database.AiStorage,
 	inferAdapter map[string]map[string]inference.ICluster,
@@ -131,7 +131,7 @@ func (i *ImageInference) saveTask() (int64, error) {
 		return 0, err
 	}
 
-	i.storage.AddNoticeInfo(i.opt.AdapterId, i.adapterName, "", "", i.opt.TaskName, "create", "任务创建中")
+	i.storage.AddNoticeInfo("", "", "", "", i.opt.TaskName, "create", "任务创建中")
 
 	return id, nil
 }
@@ -197,21 +197,36 @@ func (i *ImageInference) filterClusters() ([]*FilteredCluster, error) {
 	var cs []*FilteredCluster
 	for _, cluster := range i.clusters {
 		var inferurls []*inference.InferUrl
+		var clustertype string
 		for _, instance := range i.instances {
-			if cluster.ClusterId == instance.ClusterId {
+			clusterId := strconv.FormatInt(instance.ClusterId, 10)
+			adapterId := strconv.FormatInt(instance.AdapterId, 10)
+
+			if cluster.ClusterId == clusterId {
 				r := http.Request{}
-				deployInstance, err := i.inferAdapter[instance.AdapterId][instance.ClusterId].GetInferDeployInstance(r.Context(), instance.InstanceId)
+				deployInstance, err := i.inferAdapter[adapterId][clusterId].GetInferDeployInstance(r.Context(), instance.InstanceId)
 				if err != nil {
-					return nil, err
+					continue
 				}
 				var url inference.InferUrl
 				url.Url = deployInstance.InferUrl
 				url.Card = deployInstance.InferCard
 				inferurls = append(inferurls, &url)
+
+				clustertype = deployInstance.ClusterType
 			}
 		}
+		if len(inferurls) == 0 {
+			continue
+		}
+
+		i.inference.AppendRoute(inferurls)
+
 		var f FilteredCluster
 		f.urls = inferurls
+		f.clusterName = cluster.ClusterName
+		f.clusterType = clustertype
+		f.imageNum = cluster.Replicas
 		cs = append(cs, &f)
 	}
 	return cs, nil

internal/scheduler/service/inference/textInference/textToImage.go

@@ -9,7 +9,7 @@ import (
 )
 
 const (
-	TEXTTOIMAGE        = "text-to-image"
+	TEXTTOIMAGE        = "generate_image"
 	TEXTTOIMAGE_AiTYPE = "14"
 )
 

internal/scheduler/service/inference/textInference/textToText.go

@@ -79,9 +79,11 @@ func filterClusters(opt *option.InferOption, storage *database.AiStorage, inferA
 				wg.Done()
 				return
 			}
+
 			for i, _ := range clusterInferUrl.InferUrls {
 				clusterInferUrl.InferUrls[i].Url = clusterInferUrl.InferUrls[i].Url + inference.FORWARD_SLASH + CHAT
 			}
+
 			clusterName, _ := storage.GetClusterNameById(cId)
 
 			var f FilteredCluster

internal/storeLink/octopus.go

@@ -1154,11 +1154,16 @@ func (o *OctopusLink) GetInferDeployInstance(ctx context.Context, id string) (*i
 	if resp.Payload == nil {
 		return nil, errors.New("instance does not exist")
 	}
+
+	url := strings.Replace(resp.Payload.Notebook.Tasks[0].Url, FORWARD_SLASH, "", -1)
+	inferUrl := DOMAIN + url
+
 	ins.InstanceName = resp.Payload.Notebook.Name
 	ins.InstanceId = resp.Payload.Notebook.Id
 	ins.ClusterName = o.platform
 	ins.Status = resp.Payload.Notebook.Status
 	ins.ClusterType = TYPE_OCTOPUS
+	ins.InferUrl = inferUrl
 
 	return ins, nil
 }

internal/types/types.go

@@ -5905,6 +5905,8 @@ type Category struct {
 }
 
 type DeployInstance struct {
+	Id           string `json:"id"`
+	DeployTaskId string `json:"deployTaskId"`
 	InstanceId   string `json:"instanceId"`
 	InstanceName string `json:"instanceName"`
 	AdapterId    string `json:"adapterId"`
@@ -5915,6 +5917,9 @@ type DeployInstance struct {
 	ModelType    string `json:"modelType"`
 	InferCard    string `json:"inferCard"`
 	Status       string `json:"status"`
+	CreateTime   string `json:"createTime"`
+	UpdateTime   string `json:"updateTime"`
+	ClusterType  string `json:"clusterType"`
 }
 
 type ModelTypesResp struct {
@@ -5930,13 +5935,13 @@ type ModelNamesResp struct {
 }
 
 type ImageInferenceReq struct {
-	TaskName        string           `json:"taskName"`
+	TaskName        string                      `form:"taskName"`
-	TaskDesc        string           `json:"taskDesc"`
+	TaskDesc        string                      `form:"taskDesc"`
-	ModelType       string           `json:"modelType"`
+	ModelType       string                      `form:"modelType"`
-	Instances       []DeployInstance `json:"instances"`
+	InstanceIds     []int64                     `form:"instanceIds"`
-	Strategy        string           `json:"strategy,,optional"`
+	Strategy        string                      `form:"strategy,,optional"`
-	StaticWeightMap map[string]int32 `json:"staticWeightMap,optional"`
+	StaticWeightMap map[string]map[string]int32 `form:"staticWeightMap,optional"`
-	Replica         int32            `json:"replicas,optional"`
+	Replica         int32                       `form:"replicas,optional"`
 }
 
 type ImageInferenceResp struct {