entity/cluster.go

@@ -170,14 +170,19 @@ func ConvertCloudbrainOneNotebookResponse(input map[string]interface{}) (*QueryT
 		startTime = 0
 		endTime = 0
 	}
-
-	taskRoles := jobResultPayload.TaskRoles
-	taskRes, _ := models.ConvertToTaskPod(taskRoles[cloudbrain.SubTaskName].(map[string]interface{}))
 	var containerIP, containerID string
-	if taskRes.TaskStatuses != nil && len(taskRes.TaskStatuses) > 0 {
-		containerIP = taskRes.TaskStatuses[0].ContainerIP
-		containerID = taskRes.TaskStatuses[0].ContainerID
+	taskRoles := jobResultPayload.TaskRoles
+	if taskRoles != nil && len(taskRoles) > 0 {
+		subTask := taskRoles[cloudbrain.SubTaskName]
+		if subTask != nil {
+			taskRes, _ := models.ConvertToTaskPod(taskRoles[cloudbrain.SubTaskName].(map[string]interface{}))
+			if taskRes.TaskStatuses != nil && len(taskRes.TaskStatuses) > 0 {
+				containerIP = taskRes.TaskStatuses[0].ContainerIP
+				containerID = taskRes.TaskStatuses[0].ContainerID
+			}
+		}
 	}
+
 	res := &QueryTaskResponse{
 		StartedAt:   timeutil.TimeStamp(startTime),
 		CompletedAt: timeutil.TimeStamp(endTime),

models/cloudbrain.go

@@ -345,7 +345,6 @@ func (task *Cloudbrain) CorrectCreateUnix() {
 		task.CreatedUnix = task.StartTime
 	}
 }
-
 func (task *Cloudbrain) GetAiCenter() string {
 	if task.Type == TypeCloudBrainOne {
 		return AICenterOfCloudBrainOne
@@ -3040,6 +3039,7 @@ func CloudbrainAllStatic(opts *CloudbrainsOptions) ([]*CloudbrainInfo, int64, er
 	return cloudbrains, count, nil
 }
 
+
 func GetLastestNCloudbrain(n int) ([]*Cloudbrain, error) {
 	r := make([]*Cloudbrain, 0)
 	err := x.Where("ai_center!='' or type!=2").Desc("id").Limit(n).Unscoped().Find(&r)
@@ -3049,7 +3049,6 @@ func GetLastestNCloudbrain(n int) ([]*Cloudbrain, error) {
 	return r, nil
 
 }
-
 func CloudbrainAllKanBan(opts *CloudbrainsOptions) ([]*CloudbrainInfo, int64, error) {
 	sess := x.NewSession()
 	defer sess.Close()

modules/modelappservice/modelsevice.go

@@ -45,7 +45,7 @@ func consumerOrder(in <-chan *models.ModelApp, url string) {
 			continue
 		}
 		log.Info("goroutine id=" + fmt.Sprint(goroutine_id) + " wenxin text=" + modelApp.Desc)
-		result, err := modelarts.CreateWenXinJob(modelApp, url)
+		result, err := modelarts.CreateWenXinJobToCD(modelApp, url)
 		if err == nil {
 			if !modelarts.SendPictureReivew(result.Result) {
 				modelApp.Status = -1

modules/modelarts/wenxinresty.go

@@ -1,6 +1,8 @@
 package modelarts
 
 import (
+	"bytes"
+	"crypto/tls"
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
@@ -13,6 +15,7 @@ import (
 
 	"code.gitea.io/gitea/models"
 	"code.gitea.io/gitea/modules/log"
+	"code.gitea.io/gitea/modules/modelarts_gateway/core"
 	"code.gitea.io/gitea/modules/setting"
 )
 
@@ -29,6 +32,66 @@ type WenXinResult struct {
 	Result string `json:"result"`
 }
 
+var (
+	cdHttpClient *http.Client
+)
+
+func getCDHttpClient() *http.Client {
+	if cdHttpClient == nil {
+		cdHttpClient = &http.Client{
+			Timeout:   30 * time.Second,
+			Transport: &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: true}},
+		}
+	}
+	return cdHttpClient
+}
+
+func CreateWenXinJobToCD(modelapp *models.ModelApp, url string) (*WenXinResult, error) {
+	createJobParams := &CreateWenXinParams{
+		Data: WenXinText{
+			Prompt: modelapp.Desc,
+		},
+		Parameters: make(map[string]string),
+	}
+	var result WenXinResult
+
+	client := getCDHttpClient()
+	s := core.Signer{
+		Key:    setting.ModelartsCD.AccessKey,
+		Secret: setting.ModelartsCD.SecretKey,
+	}
+
+	req, _ := json.Marshal(createJobParams)
+	r, _ := http.NewRequest(http.MethodPost, url, ioutil.NopCloser(bytes.NewBuffer(req)))
+	log.Info("send to cd modelarts")
+	r.Header.Add("content-type", "application/json")
+	s.Sign(r)
+	res, err := client.Do(r)
+	if err == nil {
+		if res.StatusCode == 200 {
+			defer res.Body.Close()
+			body, err := ioutil.ReadAll(res.Body)
+			if err != nil {
+				log.Error("ioutil.ReadAll failed: %s", err.Error())
+				return &result, fmt.Errorf("ioutil.ReadAll failed: %s", err.Error())
+			}
+			err = json.Unmarshal(body, &result)
+			if err != nil {
+				log.Error("json.Unmarshal failed: %s", err.Error())
+				return &result, fmt.Errorf("son.Unmarshal failed: %s", err.Error())
+			}
+			return &result, nil
+		} else {
+			log.Info("res.status=" + fmt.Sprint(res.StatusCode))
+			return nil, fmt.Errorf("Service unavailable")
+		}
+	} else {
+		log.Info("error =" + err.Error())
+		return nil, fmt.Errorf("Service unavailable")
+	}
+
+}
+
 func CreateWenXinJob(modelapp *models.ModelApp, url string) (*WenXinResult, error) {
 	createJobParams := &CreateWenXinParams{
 		Data: WenXinText{

routers/api/v1/api.go

@@ -64,6 +64,8 @@ import (
 	"net/http"
 	"strings"
 
+
+
 	cloudbrainService "code.gitea.io/gitea/services/cloudbrain"
 
 	"code.gitea.io/gitea/routers/ai_task"

services/ai_task_service/container_builder/dataset_builder.go

@@ -4,6 +4,7 @@ import (
 	"code.gitea.io/gitea/entity"
 	"code.gitea.io/gitea/models"
 	"code.gitea.io/gitea/modules/log"
+	"code.gitea.io/gitea/modules/setting"
 	"code.gitea.io/gitea/services/ai_task_service/context"
 	"errors"
 	"strings"
@@ -30,7 +31,15 @@ func (b *DatasetBuilder) Build(ctx *context.CreationContext) ([]entity.Container
 	if uuid == "" {
 		return nil, nil
 	}
-	datasetInfos, datasetNames, err := models.GetDatasetInfo(uuid, ctx.Request.ComputeSource.Name)
+	var datasetInfos map[string]models.DatasetInfo
+	var datasetNames string
+	var err error
+	// models.GetDatasetInfo 是使用的以前的方法，所以此处按集群类型适配
+	if ctx.Request.Cluster == models.C2NetCluster {
+		datasetInfos, datasetNames, err = models.GetDatasetInfo(uuid, ctx.Request.ComputeSource.Name)
+	} else {
+		datasetInfos, datasetNames, err = models.GetDatasetInfo(uuid)
+	}
 	if err != nil {
 		log.Error("GetDatasetInfo failed: %v", err)
 		return nil, errors.New("cloudbrain.error.dataset_select")
@@ -41,25 +50,36 @@ func (b *DatasetBuilder) Build(ctx *context.CreationContext) ([]entity.Container
 	}
 	var data []entity.ContainerData
 	for _, datasetInfo := range datasetInfos {
+		name := datasetInfo.FullName
+		//如果不是压缩包，那么文件名是去掉后缀以后的数据集名称
+		if b.Opts.NotArchive {
+			name = datasetInfo.Name
+		}
 		if datasetInfo.Type == models.TypeCloudBrainOne {
+			//如果返回的localPath已经带了实际路径的前缀，需要去除掉以后才是在minio上的objectKey
+			objectKey := datasetInfo.DataLocalPath
+			objectKey = strings.TrimPrefix(objectKey, setting.Attachment.Minio.RealPath)
+			objectKey = strings.TrimPrefix(objectKey, setting.Attachment.Minio.Bucket)
+			objectKey = strings.TrimPrefix(objectKey, "/")
 			data = append(data, entity.ContainerData{
-				Name:          datasetInfo.FullName,
+				Name:          name,
 				Bucket:        minioUploader.GetBucket(),
 				EndPoint:      minioUploader.GetEndpoint(),
-				ObjectKey:     datasetInfo.DataLocalPath,
+				ObjectKey:     objectKey,
 				ReadOnly:      b.Opts.ReadOnly,
-				ContainerPath: b.Opts.ContainerPath + "/" + datasetInfo.FullName,
-				RealPath:      minioUploader.GetRealPath(datasetInfo.DataLocalPath),
+				ContainerPath: b.Opts.ContainerPath + "/" + name,
+				RealPath:      minioUploader.GetRealPath(objectKey),
 			})
 
 		} else {
+			objectKey := datasetInfo.DataLocalPath + datasetInfo.FullName
 			data = append(data, entity.ContainerData{
-				Name:          datasetInfo.FullName,
+				Name:          name,
 				Bucket:        obsUploader.GetBucket(),
 				EndPoint:      obsUploader.GetEndpoint(),
-				ObjectKey:     datasetInfo.DataLocalPath + datasetInfo.FullName,
+				ObjectKey:     objectKey,
 				ReadOnly:      b.Opts.ReadOnly,
-				ContainerPath: b.Opts.ContainerPath + "/" + datasetInfo.FullName,
+				ContainerPath: b.Opts.ContainerPath + "/" + name,
 			})
 		}
 

services/ai_task_service/task/cloudbrain_one_notebook_task.go

@@ -75,6 +75,7 @@ func (g CloudbrainOneNotebookTaskTemplate) GetConfig(opts entity.GetAITaskConfig
 				ContainerPath:     "/dataset",
 				ReadOnly:          true,
 				AcceptStorageType: []entity.StorageType{entity.MINIO},
+				NotArchive:        true,
 			},
 			entity.ContainerPreTrainModel: {
 				ContainerPath:     "/pretrainmodel",

services/ai_task_service/upload/minio.go

@@ -17,7 +17,7 @@ func (m *MinioUploader) GetJobDefaultObjectKeyPrefix(jobName string) string {
 	return setting.CBCodePathPrefix + jobName
 }
 func (m *MinioUploader) GetRealPath(objectKey string) string {
-	return setting.Attachment.Minio.RealPath + setting.Attachment.Minio.Bucket + "/" + objectKey
+	return setting.Attachment.Minio.RealPath + setting.Attachment.Minio.Bucket + "/" + strings.TrimPrefix(objectKey, "/")
 
 }