Merge branch 'master' of https://gitlink.org.cn/JointCloud/pcm-coordinator

Former-commit-id: cfb3f225fc

desc/inference/inference.api

@@ -79,10 +79,8 @@ type (
      TextToTextInferenceReq{
           TaskName string `form:"taskName"`
           TaskDesc string `form:"taskDesc"`
-          ModelName string `form:"modelName"`
           ModelType string `form:"modelType"`
-          AdapterId string `form:"adapterId"`
-          AiClusterIds []string `form:"aiClusterIds"`
+          InstanceId int64 `form:"instanceId"`
      }
      TextToTextInferenceResp{
 
@@ -200,7 +198,7 @@ type (
 
      AdapterAvail {
           AdapterId string `json:"adapterId"`
-          AdapterName string `json:"taskName"`
+          AdapterName string `json:"adapterName"`
           Clusters []*ClusterAvail `json:"clusters"`
      }
 

internal/logic/inference/createdeploytasklogic.go

@@ -95,7 +95,7 @@ func (l *CreateDeployTaskLogic) createDeployInstance(taskId int64, adapterId str
 		return err
 	}
 
-	_, err = l.svcCtx.Scheduler.AiStorages.SaveInferDeployInstance(taskId, ins.InstanceId, ins.InstanceName, aid, adapterName, cid, clusterName, ins.ModelName, ins.ModelType, ins.InferCard)
+	_, err = l.svcCtx.Scheduler.AiStorages.SaveInferDeployInstance(taskId, ins.InstanceId, ins.InstanceName, aid, adapterName, cid, clusterName, ins.ModelName, ins.ModelType, ins.InferCard, ins.ClusterType)
 	if err != nil {
 		return err
 	}

internal/logic/inference/deployinstancelistlogic.go

@@ -52,7 +52,10 @@ func (l *DeployInstanceListLogic) DeployInstanceList(req *types.DeployInstanceLi
 		return nil, errors.New(err.Error())
 	}
 
-	deployTasks := l.GenerateDeployTasks(tasklist)
+	deployTasks, err := l.GenerateDeployTasks(tasklist)
+	if err != nil {
+		return nil, errors.New(err.Error())
+	}
 	slices := make([][]*models.AiInferDeployInstance, len(deployTasks))
 	for i := 0; i < len(deployTasks); i++ {
 		slices[i] = deployTasks[i].Instances
@@ -83,12 +86,20 @@ func (l *DeployInstanceListLogic) DeployInstanceList(req *types.DeployInstanceLi
 	return
 }
 
-func (l *DeployInstanceListLogic) GenerateDeployTasks(tasklist []*models.AiDeployInstanceTask) []*DeployTask {
+func (l *DeployInstanceListLogic) GenerateDeployTasks(tasklist []*models.AiDeployInstanceTask) ([]*DeployTask, error) {
 	var tasks []*DeployTask
 	for _, t := range tasklist {
 		list, err := l.svcCtx.Scheduler.AiStorages.GetInstanceListByDeployTaskId(t.Id)
 		if err != nil {
 			logx.Errorf("db GetInstanceListByDeployTaskId error")
+			return nil, errors.New(err.Error())
+		}
+		if len(list) == 0 {
+			err := l.svcCtx.Scheduler.AiStorages.DeleteDeployTaskById(t.Id)
+			if err != nil {
+				logx.Errorf("db DeleteByDeployTaskId error")
+				return nil, errors.New(err.Error())
+			}
 			continue
 		}
 		deployTask := &DeployTask{
@@ -99,12 +110,12 @@ func (l *DeployInstanceListLogic) GenerateDeployTasks(tasklist []*models.AiDeplo
 		}
 		tasks = append(tasks, deployTask)
 	}
-	return tasks
+	return tasks, nil
 }
 
 type DeployTask struct {
 	Id        int64                           `json:"id,string"`
-	Name      string                          `json:"name,string"`
-	Desc      string                          `json:"desc,string"`
+	Name      string                          `json:"name"`
+	Desc      string                          `json:"desc"`
 	Instances []*models.AiInferDeployInstance `json:"instances,string"`
 }

internal/logic/inference/texttotextinferencelogic.go

@@ -9,6 +9,7 @@ import (
 	"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/service/inference/textInference"
 	"gitlink.org.cn/JointCloud/pcm-coordinator/internal/svc"
 	"gitlink.org.cn/JointCloud/pcm-coordinator/internal/types"
+	"strconv"
 )
 
 type TextToTextInferenceLogic struct {
@@ -27,27 +28,31 @@ func NewTextToTextInferenceLogic(ctx context.Context, svcCtx *svc.ServiceContext
 
 func (l *TextToTextInferenceLogic) TextToTextInference(req *types.TextToTextInferenceReq) (resp *types.TextToTextInferenceResp, err error) {
 	resp = &types.TextToTextInferenceResp{}
-	opt := &option.InferOption{
-		TaskName:     req.TaskName,
-		TaskDesc:     req.TaskDesc,
-		AdapterId:    req.AdapterId,
-		AiClusterIds: req.AiClusterIds,
-		ModelName:    req.ModelName,
-		ModelType:    req.ModelType,
+
+	instance, err := l.svcCtx.Scheduler.AiStorages.GetInferDeployInstanceById(req.InstanceId)
+	if err != nil {
+		return nil, err
 	}
+	if instance == nil {
+		return nil, errors.New("instance is empty ")
+	}
+
+	adapterId := strconv.FormatInt(instance.AdapterId, 10)
 
-	_, ok := l.svcCtx.Scheduler.AiService.AiCollectorAdapterMap[opt.AdapterId]
-	if !ok {
-		return nil, errors.New("AdapterId does not exist")
+	opt := &option.InferOption{
+		TaskName:  req.TaskName,
+		TaskDesc:  req.TaskDesc,
+		ModelType: req.ModelType,
+		AdapterId: adapterId,
 	}
 
 	adapterName, err := l.svcCtx.Scheduler.AiStorages.GetAdapterNameById(opt.AdapterId)
 
-	inType, err := textInference.NewTextToText(opt, l.svcCtx.Scheduler.AiStorages, l.svcCtx.Scheduler.AiService.InferenceAdapterMap)
+	infer, err := textInference.NewTextToText(opt, l.svcCtx.Scheduler.AiStorages, l.svcCtx.Scheduler.AiService.InferenceAdapterMap, instance)
 	if err != nil {
 		return nil, err
 	}
-	textInfer, err := textInference.New(inType, opt, l.svcCtx.Scheduler.AiStorages, l.svcCtx.Scheduler.AiService.InferenceAdapterMap, adapterName)
+	textInfer, err := textInference.New(infer, opt, l.svcCtx.Scheduler.AiStorages, adapterName)
 	if err != nil {
 		return nil, err
 	}

internal/scheduler/database/aiStorage.go

@@ -374,7 +374,7 @@ func (s *AiStorage) AddNoticeInfo(adapterId string, adapterName string, clusterI
 }
 
 func (s *AiStorage) SaveInferDeployInstance(taskId int64, instanceId string, instanceName string, adapterId int64,
-	adapterName string, clusterId int64, clusterName string, modelName string, modelType string, inferCard string) (int64, error) {
+	adapterName string, clusterId int64, clusterName string, modelName string, modelType string, inferCard string, clusterType string) (int64, error) {
 	startTime := time.Now().Format(time.RFC3339)
 	// 构建主任务结构体
 	insModel := models.AiInferDeployInstance{
@@ -388,7 +388,8 @@ func (s *AiStorage) SaveInferDeployInstance(taskId int64, instanceId string, ins
 		ModelName:            modelName,
 		ModelType:            modelType,
 		InferCard:            inferCard,
-		Status:               constants.Stopped,
+		ClusterType:          clusterType,
+		Status:               constants.Deploying,
 		CreateTime:           startTime,
 		UpdateTime:           startTime,
 	}
@@ -464,6 +465,15 @@ func (s *AiStorage) UpdateDeployTask(task *models.AiDeployInstanceTask, needUpda
 	return nil
 }
 
+func (s *AiStorage) DeleteDeployTaskById(id int64) error {
+	tx := s.DbEngin.Delete(&models.AiDeployInstanceTask{}, id)
+	if tx.Error != nil {
+		logx.Errorf(tx.Error.Error())
+		return tx.Error
+	}
+	return nil
+}
+
 func (s *AiStorage) UpdateDeployTaskById(id int64) error {
 	task, err := s.GetDeployTaskById(id)
 	if err != nil {

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

@@ -17,30 +17,28 @@ type FilteredCluster struct {
 	urls        []*inference.InferUrl
 	clusterId   string
 	clusterName string
+	clusterType string
 }
 
 type TextInference struct {
-	inference    ITextInference
-	opt          *option.InferOption
-	storage      *database.AiStorage
-	inferAdapter map[string]map[string]inference.ICluster
-	errMap       map[string]string
-	adapterName  string
+	inference   ITextInference
+	opt         *option.InferOption
+	storage     *database.AiStorage
+	errMap      map[string]string
+	adapterName string
 }
 
 func New(
 	inference ITextInference,
 	opt *option.InferOption,
 	storage *database.AiStorage,
-	inferAdapter map[string]map[string]inference.ICluster,
 	adapterName string) (*TextInference, error) {
 	return &TextInference{
-		inference:    inference,
-		opt:          opt,
-		storage:      storage,
-		inferAdapter: inferAdapter,
-		adapterName:  adapterName,
-		errMap:       make(map[string]string),
+		inference:   inference,
+		opt:         opt,
+		storage:     storage,
+		adapterName: adapterName,
+		errMap:      make(map[string]string),
 	}, nil
 }
 

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

@@ -22,11 +22,12 @@ type TextToText struct {
 	opt          *option.InferOption
 	storage      *database.AiStorage
 	inferAdapter map[string]map[string]inference.ICluster
+	instance     *models.AiInferDeployInstance
 	cs           []*FilteredCluster
 }
 
-func NewTextToText(opt *option.InferOption, storage *database.AiStorage, inferAdapter map[string]map[string]inference.ICluster) (*TextToText, error) {
-	cs, err := filterClusters(opt, storage, inferAdapter)
+func NewTextToText(opt *option.InferOption, storage *database.AiStorage, inferAdapter map[string]map[string]inference.ICluster, instance *models.AiInferDeployInstance) (*TextToText, error) {
+	cs, err := filterClusters(inferAdapter, instance)
 	if err != nil {
 		return nil, err
 	}
@@ -64,7 +65,35 @@ func (tt *TextToText) SaveAiTask(id int64, adapterName string) error {
 	return nil
 }
 
-func filterClusters(opt *option.InferOption, storage *database.AiStorage, inferAdapter map[string]map[string]inference.ICluster) ([]*FilteredCluster, error) {
+func filterClusters(inferAdapter map[string]map[string]inference.ICluster, instance *models.AiInferDeployInstance) ([]*FilteredCluster, error) {
+	var cs []*FilteredCluster
+	var inferurls []*inference.InferUrl
+	clusterId := strconv.FormatInt(instance.ClusterId, 10)
+	adapterId := strconv.FormatInt(instance.AdapterId, 10)
+	r := http.Request{}
+	deployInstance, err := inferAdapter[adapterId][clusterId].GetInferDeployInstance(r.Context(), instance.InstanceId)
+	if err != nil {
+		return nil, err
+	}
+	var url inference.InferUrl
+	url.Url = deployInstance.InferUrl + inference.FORWARD_SLASH + CHAT
+	url.Card = deployInstance.InferCard
+	inferurls = append(inferurls, &url)
+
+	clusterType := deployInstance.ClusterType
+	clusterName := deployInstance.ClusterName
+
+	var f FilteredCluster
+	f.urls = inferurls
+	f.clusterId = clusterId
+	f.clusterName = clusterName
+	f.clusterType = clusterType
+	cs = append(cs, &f)
+
+	return cs, nil
+}
+
+func filterClustersTemp(opt *option.InferOption, storage *database.AiStorage, inferAdapter map[string]map[string]inference.ICluster) ([]*FilteredCluster, error) {
 	var wg sync.WaitGroup
 	var ch = make(chan *FilteredCluster, len(opt.AiClusterIds))
 	var cs []*FilteredCluster

internal/storeLink/octopus.go

@@ -1158,10 +1158,14 @@ func (o *OctopusLink) GetInferDeployInstance(ctx context.Context, id string) (*i
 	url := strings.Replace(resp.Payload.Notebook.Tasks[0].Url, FORWARD_SLASH, "", -1)
 	inferUrl := DOMAIN + url
 
+	var modelType string
+	var modelName string
 	var card string
 	if resp.Payload.Notebook.Desc != "" {
 		str := strings.Split(resp.Payload.Notebook.Desc, FORWARD_SLASH)
 		if len(str) == 3 {
+			modelType = str[0]
+			modelName = str[1]
 			card = str[2]
 		}
 	}
@@ -1171,6 +1175,8 @@ func (o *OctopusLink) GetInferDeployInstance(ctx context.Context, id string) (*i
 	ins.ClusterName = o.platform
 	ins.Status = resp.Payload.Notebook.Status
 	ins.ClusterType = TYPE_OCTOPUS
+	ins.ModelType = modelType
+	ins.ModelName = modelName
 	ins.InferUrl = inferUrl
 	ins.InferCard = card
 

internal/types/types.go

@@ -5977,12 +5977,10 @@ type InferenceResult struct {
 }
 
 type TextToTextInferenceReq struct {
-	TaskName     string   `form:"taskName"`
-	TaskDesc     string   `form:"taskDesc"`
-	ModelName    string   `form:"modelName"`
-	ModelType    string   `form:"modelType"`
-	AdapterId    string   `form:"adapterId"`
-	AiClusterIds []string `form:"aiClusterIds"`
+	TaskName   string `form:"taskName"`
+	TaskDesc   string `form:"taskDesc"`
+	ModelType  string `form:"modelType"`
+	InstanceId int64  `form:"instanceId"`
 }
 
 type TextToTextInferenceResp struct {
@@ -6097,7 +6095,7 @@ type GetAdaptersByModelResp struct {
 
 type AdapterAvail struct {
 	AdapterId   string          `json:"adapterId"`
-	AdapterName string          `json:"taskName"`
+	AdapterName string          `json:"adapterName"`
 	Clusters    []*ClusterAvail `json:"clusters"`
 }