manager/client/grampus/grampus.go

@@ -12,6 +12,7 @@ import (
 	"code.gitea.io/gitea/modules/log"
 	"code.gitea.io/gitea/modules/setting"
 	"github.com/go-resty/resty/v2"
+	"github.com/gorilla/websocket"
 )
 
 var (
@@ -597,3 +598,35 @@ sendjob:
 	log.Info("delete grampus task, re=" + result.Info + "  grampus jobId=" + jobID)
 	return &result, nil
 }
+
+func ClickOnce(url string) {
+	client := getRestyClient()
+	log.Info("click url=" + url)
+	res, err := client.R().
+		SetHeader("Content-Type", "application/json").
+		Get(url)
+	if err != nil {
+		log.Info("error=" + err.Error())
+	} else {
+		if res != nil {
+			jsonstr, _ := json.Marshal(res)
+			log.Info("job resp:", jsonstr)
+		}
+	}
+}
+
+func SendMsgToWebsocket(wsurl string, msg string) {
+	//wsurl := strings.Replace(tmpurl, "http", "ws", 1)
+	log.Info("wsurl =" + wsurl)
+	conn, _, err := websocket.DefaultDialer.Dial(wsurl, nil)
+	if err == nil {
+		defer conn.Close()
+		err1 := conn.WriteJSON(msg)
+		//err1 := conn.WriteJSON("{\"msg\":\"send_hash\"}")
+		if err1 != nil {
+			log.Info("websocket send msg error=" + err1.Error())
+		}
+	} else {
+		log.Info("websocket error=" + err.Error())
+	}
+}

modules/setting/setting.go

@@ -641,6 +641,8 @@ var (
 		GPUImageCommonName            string
 		MultiNode                     string
 		MMLSparkMaxTime               int64
+		NoteBookDomainURL             string
+		NoteBookLocalURL              string
 	}{}
 
 	ClearStrategy = struct {
@@ -1965,6 +1967,8 @@ func GetGrampusConfig() {
 
 	Grampus.MMLSparkMaxTime = sec.Key("MMLSparkMaxTime").MustInt64(8 * 3600)
 
+	Grampus.NoteBookDomainURL = sec.Key("NoteBookDomainURL").MustString("https://notebook.openi.org.cn")
+	Grampus.NoteBookLocalURL = sec.Key("NoteBookLocalURL").MustString("http://192.168.242.23")
 }
 
 func SetRadarMapConfig() {

services/ai_task_service/cluster/c2net.go

@@ -180,7 +180,7 @@ func generateCommand(repoName, bootFile, datasetName string) string {
 	}
 	currentTimeNow := time.Now()
 	timePostfix := currentTimeNow.Format("20060102_150405")
-	command += "cd " + copyDatasetPath + ";echo " + timePostfix + " | tee /output/log_" + timePostfix + ".txt; uvicorn " + bootonlyfile + ":app --host 0.0.0.0 --port $OCTOPUS_NOTEBOOK_PORT 2>&1 | tee -a /output/log_" + timePostfix + ".txt;"
+	command += "cd " + copyDatasetPath + ";echo " + timePostfix + " | tee /output/log_" + timePostfix + ".txt; uvicorn " + bootonlyfile + ":app --log-level debug --host 0.0.0.0 --port $OCTOPUS_NOTEBOOK_PORT 2>&1 | tee -a /output/log_" + timePostfix + ".txt;"
 
 	log.Info("comand=" + command)
 	return command
@@ -351,8 +351,28 @@ func (c C2NetClusterAdapter) GetNoteBookUrl(jobId string) (string, error) {
 	if res.Url == "" {
 		return "", errors.New("notebook task error")
 	}
+	onlyForOnlineInference(res.Url, jobId)
 	return res.Url + "?token=" + res.Token, nil
 }
+
+func onlyForOnlineInference(url string, jobId string) {
+	cloudbrainTask, bizErr := models.GetCloudbrainByJobID(jobId)
+	if bizErr == nil {
+		if cloudbrainTask.JobType != string(models.JobTypeOnlineInference) {
+			return
+		}
+	} else {
+		return
+	}
+	if strings.HasPrefix(url, setting.Grampus.NoteBookDomainURL) {
+		tmpurl := strings.Replace(url, setting.Grampus.NoteBookDomainURL, setting.Grampus.NoteBookLocalURL, 1)
+		grampus.ClickOnce(tmpurl)
+		tmpurl += "/queue/join"
+		wsurl := strings.Replace(tmpurl, "http", "ws", 1)
+		grampus.SendMsgToWebsocket(wsurl, "{\"msg\":\"send_hash\"}")
+	}
+}
+
 func (c C2NetClusterAdapter) GetNoteBookOperationProfile(jobId string) (*entity.OperationProfile, error) {
 	if jobId == "" {
 		log.Error("jobid is empty")