You can not select more than 25 topics Topics must start with a chinese character,a letter or number, can include dashes ('-') and can be up to 35 characters long.

cloudbrain.go 27 kB

4 years ago
5 years ago
4 years ago
5 years ago
5 years ago
4 years ago
5 years ago
4 years ago
4 years ago
4 years ago
5 years ago
4 years ago
5 years ago
5 years ago
5 years ago
5 years ago
4 years ago
5 years ago
4 years ago
5 years ago
5 years ago
5 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
5 years ago
4 years ago
5 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
5 years ago
4 years ago
3 years ago
5 years ago
4 years ago
4 years ago
5 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
3 years ago
3 years ago
4 years ago
4 years ago
5 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
5 years ago
5 years ago
5 years ago
4 years ago
4 years ago
5 years ago
4 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
4 years ago
5 years ago
5 years ago
5 years ago
4 years ago
5 years ago
5 years ago
5 years ago
5 years ago
4 years ago
5 years ago
5 years ago
5 years ago
3 years ago
3 years ago
3 years ago
5 years ago
3 years ago
5 years ago
3 years ago
5 years ago
3 years ago
3 years ago
3 years ago
5 years ago
3 years ago
3 years ago
3 years ago
3 years ago
5 years ago
4 years ago
4 years ago
3 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
3 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
5 years ago
5 years ago
4 years ago
4 years ago
5 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929
  1. package repo
  2. import (
  3. "bufio"
  4. "encoding/json"
  5. "errors"
  6. "fmt"
  7. "io"
  8. "net/http"
  9. "os"
  10. "regexp"
  11. "sort"
  12. "strconv"
  13. "strings"
  14. "time"
  15. "code.gitea.io/gitea/modules/modelarts"
  16. "code.gitea.io/gitea/modules/git"
  17. "code.gitea.io/gitea/modules/storage"
  18. "code.gitea.io/gitea/models"
  19. "code.gitea.io/gitea/modules/auth"
  20. "code.gitea.io/gitea/modules/base"
  21. "code.gitea.io/gitea/modules/cloudbrain"
  22. "code.gitea.io/gitea/modules/context"
  23. "code.gitea.io/gitea/modules/log"
  24. "code.gitea.io/gitea/modules/setting"
  25. )
  26. const (
  27. tplCloudBrainIndex base.TplName = "repo/cloudbrain/index"
  28. tplCloudBrainNew base.TplName = "repo/cloudbrain/new"
  29. tplCloudBrainShow base.TplName = "repo/cloudbrain/show"
  30. tplCloudBrainShowModels base.TplName = "repo/cloudbrain/models/index"
  31. )
  32. var (
  33. gpuInfos *models.GpuInfos
  34. categories *models.Categories
  35. )
  36. var jobNamePattern = regexp.MustCompile(`^[a-z0-9][a-z0-9-_]{1,34}[a-z0-9-]$`)
  37. // MustEnableDataset check if repository enable internal cb
  38. func MustEnableCloudbrain(ctx *context.Context) {
  39. if !ctx.Repo.CanRead(models.UnitTypeCloudBrain) {
  40. ctx.NotFound("MustEnableCloudbrain", nil)
  41. return
  42. }
  43. }
  44. func cutString(str string, lens int) string {
  45. if len(str) < lens {
  46. return str
  47. }
  48. return str[:lens]
  49. }
  50. func jobNamePrefixValid(s string) string {
  51. lowStr := strings.ToLower(s)
  52. re := regexp.MustCompile(`[^a-z0-9_\\-]+`)
  53. removeSpecial := re.ReplaceAllString(lowStr, "")
  54. re = regexp.MustCompile(`^[_\\-]+`)
  55. return re.ReplaceAllString(removeSpecial, "")
  56. }
  57. func cloudBrainNewDataPrepare(ctx *context.Context) error {
  58. ctx.Data["PageIsCloudBrain"] = true
  59. t := time.Now()
  60. var jobName = jobNamePrefixValid(cutString(ctx.User.Name, 5)) + t.Format("2006010215") + strconv.Itoa(int(t.Unix()))[5:]
  61. ctx.Data["job_name"] = jobName
  62. result, err := cloudbrain.GetImages()
  63. if err != nil {
  64. ctx.Data["error"] = err.Error()
  65. log.Error("cloudbrain.GetImages failed:", err.Error(), ctx.Data["MsgID"])
  66. }
  67. for i, payload := range result.Payload.ImageInfo {
  68. if strings.HasPrefix(result.Payload.ImageInfo[i].Place, "192.168") {
  69. result.Payload.ImageInfo[i].PlaceView = payload.Place[strings.Index(payload.Place, "/"):len(payload.Place)]
  70. } else {
  71. result.Payload.ImageInfo[i].PlaceView = payload.Place
  72. }
  73. }
  74. ctx.Data["images"] = result.Payload.ImageInfo
  75. resultPublic, err := cloudbrain.GetPublicImages()
  76. if err != nil {
  77. ctx.Data["error"] = err.Error()
  78. log.Error("cloudbrain.GetPublicImages failed:", err.Error(), ctx.Data["MsgID"])
  79. }
  80. for i, payload := range resultPublic.Payload.ImageInfo {
  81. if strings.HasPrefix(resultPublic.Payload.ImageInfo[i].Place, "192.168") {
  82. resultPublic.Payload.ImageInfo[i].PlaceView = payload.Place[strings.Index(payload.Place, "/"):len(payload.Place)]
  83. } else {
  84. resultPublic.Payload.ImageInfo[i].PlaceView = payload.Place
  85. }
  86. }
  87. ctx.Data["public_images"] = resultPublic.Payload.ImageInfo
  88. attachs, err := models.GetAllUserAttachments(ctx.User.ID)
  89. if err != nil {
  90. log.Error("GetAllUserAttachments failed: %v", err, ctx.Data["MsgID"])
  91. return err
  92. }
  93. ctx.Data["attachments"] = attachs
  94. ctx.Data["command"] = cloudbrain.Command
  95. ctx.Data["code_path"] = cloudbrain.CodeMountPath
  96. ctx.Data["dataset_path"] = cloudbrain.DataSetMountPath
  97. ctx.Data["model_path"] = cloudbrain.ModelMountPath
  98. ctx.Data["benchmark_path"] = cloudbrain.BenchMarkMountPath
  99. ctx.Data["is_benchmark_enabled"] = setting.IsBenchmarkEnabled
  100. if categories == nil {
  101. json.Unmarshal([]byte(setting.BenchmarkCategory), &categories)
  102. }
  103. ctx.Data["benchmark_categories"] = categories.Category
  104. if gpuInfos == nil {
  105. json.Unmarshal([]byte(setting.GpuTypes), &gpuInfos)
  106. }
  107. ctx.Data["gpu_types"] = gpuInfos.GpuInfo
  108. if cloudbrain.ResourceSpecs == nil {
  109. json.Unmarshal([]byte(setting.ResourceSpecs), &cloudbrain.ResourceSpecs)
  110. }
  111. ctx.Data["resource_specs"] = cloudbrain.ResourceSpecs.ResourceSpec
  112. ctx.Data["snn4imagenet_path"] = cloudbrain.Snn4imagenetMountPath
  113. ctx.Data["is_snn4imagenet_enabled"] = setting.IsSnn4imagenetEnabled
  114. ctx.Data["brainscore_path"] = cloudbrain.BrainScoreMountPath
  115. ctx.Data["is_brainscore_enabled"] = setting.IsBrainScoreEnabled
  116. return nil
  117. }
  118. func CloudBrainNew(ctx *context.Context) {
  119. err := cloudBrainNewDataPrepare(ctx)
  120. if err != nil {
  121. ctx.ServerError("get new cloudbrain info failed", err)
  122. return
  123. }
  124. ctx.HTML(200, tplCloudBrainNew)
  125. }
  126. func CloudBrainCreate(ctx *context.Context, form auth.CreateCloudBrainForm) {
  127. ctx.Data["PageIsCloudBrain"] = true
  128. jobName := form.JobName
  129. image := form.Image
  130. command := form.Command
  131. uuid := form.Attachment
  132. jobType := form.JobType
  133. gpuQueue := form.GpuType
  134. codePath := setting.JobPath + jobName + cloudbrain.CodeMountPath
  135. resourceSpecId := form.ResourceSpecId
  136. if !jobNamePattern.MatchString(jobName) {
  137. ctx.RenderWithErr(ctx.Tr("repo.cloudbrain_jobname_err"), tplCloudBrainNew, &form)
  138. return
  139. }
  140. if jobType != string(models.JobTypeBenchmark) && jobType != string(models.JobTypeDebug) && jobType != string(models.JobTypeSnn4imagenet) && jobType != string(models.JobTypeBrainScore) {
  141. log.Error("jobtype error:", jobType, ctx.Data["MsgID"])
  142. cloudBrainNewDataPrepare(ctx)
  143. ctx.RenderWithErr("jobtype error", tplCloudBrainNew, &form)
  144. return
  145. }
  146. count, err := models.GetCloudbrainCountByUserID(ctx.User.ID)
  147. if err != nil {
  148. log.Error("GetCloudbrainCountByUserID failed:%v", err, ctx.Data["MsgID"])
  149. cloudBrainNewDataPrepare(ctx)
  150. ctx.RenderWithErr("system error", tplCloudBrainNew, &form)
  151. return
  152. } else {
  153. if count >= 1 {
  154. log.Error("the user already has running or waiting task", ctx.Data["MsgID"])
  155. cloudBrainNewDataPrepare(ctx)
  156. ctx.RenderWithErr("you have already a running or waiting task, can not create more", tplCloudBrainNew, &form)
  157. return
  158. }
  159. }
  160. _, err = models.GetCloudbrainByName(jobName)
  161. if err == nil {
  162. log.Error("the job name did already exist", ctx.Data["MsgID"])
  163. cloudBrainNewDataPrepare(ctx)
  164. ctx.RenderWithErr("the job name did already exist", tplCloudBrainNew, &form)
  165. return
  166. } else {
  167. if !models.IsErrJobNotExist(err) {
  168. log.Error("system error, %v", err, ctx.Data["MsgID"])
  169. cloudBrainNewDataPrepare(ctx)
  170. ctx.RenderWithErr("system error", tplCloudBrainNew, &form)
  171. return
  172. }
  173. }
  174. repo := ctx.Repo.Repository
  175. downloadCode(repo, codePath)
  176. uploadCodeToMinio(codePath+"/", jobName, cloudbrain.CodeMountPath+"/")
  177. modelPath := setting.JobPath + jobName + cloudbrain.ModelMountPath + "/"
  178. mkModelPath(modelPath)
  179. uploadCodeToMinio(modelPath, jobName, cloudbrain.ModelMountPath+"/")
  180. benchmarkPath := setting.JobPath + jobName + cloudbrain.BenchMarkMountPath
  181. if setting.IsBenchmarkEnabled && jobType == string(models.JobTypeBenchmark) {
  182. var gpuType string
  183. for _, gpuInfo := range gpuInfos.GpuInfo {
  184. if gpuInfo.Queue == gpuQueue {
  185. gpuType = gpuInfo.Value
  186. }
  187. }
  188. downloadRateCode(repo, jobName, setting.BenchmarkOwner, setting.BrainScoreName, benchmarkPath, form.BenchmarkCategory, gpuType)
  189. uploadCodeToMinio(benchmarkPath+"/", jobName, cloudbrain.BenchMarkMountPath+"/")
  190. }
  191. snn4imagenetPath := setting.JobPath + jobName + cloudbrain.Snn4imagenetMountPath
  192. if setting.IsSnn4imagenetEnabled && jobType == string(models.JobTypeSnn4imagenet) {
  193. downloadRateCode(repo, jobName, setting.Snn4imagenetOwner, setting.Snn4imagenetName, snn4imagenetPath, "", "")
  194. uploadCodeToMinio(snn4imagenetPath+"/", jobName, cloudbrain.Snn4imagenetMountPath+"/")
  195. }
  196. brainScorePath := setting.JobPath + jobName + cloudbrain.BrainScoreMountPath
  197. if setting.IsBrainScoreEnabled && jobType == string(models.JobTypeBrainScore) {
  198. downloadRateCode(repo, jobName, setting.BrainScoreOwner, setting.BrainScoreName, brainScorePath, "", "")
  199. uploadCodeToMinio(brainScorePath+"/", jobName, cloudbrain.BrainScoreMountPath+"/")
  200. }
  201. err = cloudbrain.GenerateTask(ctx, jobName, image, command, uuid, storage.GetMinioPath(jobName, cloudbrain.CodeMountPath+"/"),
  202. storage.GetMinioPath(jobName, cloudbrain.ModelMountPath+"/"),
  203. storage.GetMinioPath(jobName, cloudbrain.BenchMarkMountPath+"/"), storage.GetMinioPath(jobName, cloudbrain.Snn4imagenetMountPath+"/"),
  204. storage.GetMinioPath(jobName, cloudbrain.BrainScoreMountPath+"/"), jobType, gpuQueue, resourceSpecId)
  205. if err != nil {
  206. cloudBrainNewDataPrepare(ctx)
  207. ctx.RenderWithErr(err.Error(), tplCloudBrainNew, &form)
  208. return
  209. }
  210. ctx.Redirect(setting.AppSubURL + ctx.Repo.RepoLink + "/debugjob")
  211. }
  212. func CloudBrainRestart(ctx *context.Context) {
  213. /*
  214. todo: 是否启用事务?
  215. 启用:处理时间可能比较长,容易出现大事务
  216. 不启用:容易出现脏数据
  217. */
  218. var jobID = ctx.Params(":jobid")
  219. var resultCode = "0"
  220. var errorMsg = ""
  221. var status = ""
  222. for {
  223. task, err := models.GetCloudbrainByJobID(jobID)
  224. if err != nil {
  225. log.Error("GetCloudbrainByJobID(%s) failed:%v", jobID, err.Error(), ctx.Data["MsgID"])
  226. resultCode = "-1"
  227. errorMsg = "system error"
  228. break
  229. }
  230. if task.Status != string(models.JobStopped) && task.Status != string(models.JobSucceeded) && task.Status != string(models.JobFailed) {
  231. log.Error("the job(%s) is not stopped", task.JobName, ctx.Data["MsgID"])
  232. resultCode = "-1"
  233. errorMsg = "the job is not stopped"
  234. break
  235. }
  236. count, err := models.GetCloudbrainCountByUserID(ctx.User.ID)
  237. if err != nil {
  238. log.Error("GetCloudbrainCountByUserID failed:%v", err, ctx.Data["MsgID"])
  239. resultCode = "-1"
  240. errorMsg = "system error"
  241. break
  242. } else {
  243. if count >= 1 {
  244. log.Error("the user already has running or waiting task", ctx.Data["MsgID"])
  245. resultCode = "-1"
  246. errorMsg = "the user already has running or waiting task"
  247. break
  248. }
  249. }
  250. err = cloudbrain.RestartTask(ctx, task)
  251. if err != nil {
  252. log.Error("RestartTask failed:%v", err.Error(), ctx.Data["MsgID"])
  253. resultCode = "-1"
  254. errorMsg = "system error"
  255. break
  256. }
  257. status = task.Status
  258. jobID = task.JobID
  259. break
  260. }
  261. ctx.JSON(200, map[string]string{
  262. "result_code": resultCode,
  263. "error_msg": errorMsg,
  264. "status": status,
  265. "job_id": jobID,
  266. })
  267. }
  268. func CloudBrainShow(ctx *context.Context) {
  269. ctx.Data["PageIsCloudBrain"] = true
  270. var jobID = ctx.Params(":jobid")
  271. task, err := models.GetCloudbrainByJobID(jobID)
  272. if err != nil {
  273. ctx.Data["error"] = err.Error()
  274. }
  275. result, err := cloudbrain.GetJob(jobID)
  276. if err != nil {
  277. ctx.Data["error"] = err.Error()
  278. }
  279. if result != nil {
  280. jobRes, _ := models.ConvertToJobResultPayload(result.Payload)
  281. jobRes.Resource.Memory = strings.ReplaceAll(jobRes.Resource.Memory, "Mi", "MB")
  282. taskRoles := jobRes.TaskRoles
  283. if jobRes.JobStatus.State != string(models.JobFailed) {
  284. taskRes, _ := models.ConvertToTaskPod(taskRoles[cloudbrain.SubTaskName].(map[string]interface{}))
  285. ctx.Data["taskRes"] = taskRes
  286. task.Status = taskRes.TaskStatuses[0].State
  287. task.ContainerID = taskRes.TaskStatuses[0].ContainerID
  288. task.ContainerIp = taskRes.TaskStatuses[0].ContainerIP
  289. err = models.UpdateJob(task)
  290. if err != nil {
  291. ctx.Data["error"] = err.Error()
  292. }
  293. } else {
  294. task.Status = jobRes.JobStatus.State
  295. taskRes := models.TaskPod{TaskStatuses: []models.TaskStatuses{
  296. {
  297. State: jobRes.JobStatus.State,
  298. },
  299. }}
  300. ctx.Data["taskRes"] = taskRes
  301. jobRes.JobStatus.StartTime = time.Unix(int64(task.CreatedUnix), 0).Format("2006-01-02 15:04:05")
  302. jobRes.JobStatus.EndTime = time.Unix(int64(task.UpdatedUnix), 0).Format("2006-01-02 15:04:05")
  303. }
  304. ctx.Data["result"] = jobRes
  305. }
  306. ctx.Data["task"] = task
  307. ctx.Data["jobID"] = jobID
  308. ctx.HTML(200, tplCloudBrainShow)
  309. }
  310. func CloudBrainDebug(ctx *context.Context) {
  311. var jobID = ctx.Params(":jobid")
  312. if !ctx.IsSigned {
  313. log.Error("the user has not signed in")
  314. ctx.Error(http.StatusForbidden, "", "the user has not signed in")
  315. return
  316. }
  317. task, err := models.GetCloudbrainByJobID(jobID)
  318. if err != nil {
  319. ctx.ServerError("GetCloudbrainByJobID failed", err)
  320. return
  321. }
  322. debugUrl := setting.DebugServerHost + "jpylab_" + task.JobID + "_" + task.SubTaskName
  323. ctx.Redirect(debugUrl)
  324. }
  325. func CloudBrainCommitImage(ctx *context.Context, form auth.CommitImageCloudBrainForm) {
  326. var jobID = ctx.Params(":jobid")
  327. if !ctx.IsSigned {
  328. log.Error("the user has not signed in")
  329. ctx.Error(http.StatusForbidden, "", "the user has not signed in")
  330. return
  331. }
  332. task, err := models.GetCloudbrainByJobID(jobID)
  333. if err != nil {
  334. ctx.JSON(200, map[string]string{
  335. "result_code": "-1",
  336. "error_msg": "GetCloudbrainByJobID failed",
  337. })
  338. return
  339. }
  340. err = cloudbrain.CommitImage(jobID, models.CommitImageParams{
  341. Ip: task.ContainerIp,
  342. TaskContainerId: task.ContainerID,
  343. ImageDescription: form.Description,
  344. ImageTag: form.Tag,
  345. })
  346. if err != nil {
  347. log.Error("CommitImage(%s) failed:%v", task.JobName, err.Error(), ctx.Data["msgID"])
  348. ctx.JSON(200, map[string]string{
  349. "result_code": "-1",
  350. "error_msg": "CommitImage failed",
  351. })
  352. return
  353. }
  354. ctx.JSON(200, map[string]string{
  355. "result_code": "0",
  356. "error_msg": "",
  357. })
  358. }
  359. func CloudBrainStop(ctx *context.Context) {
  360. var jobID = ctx.Params(":jobid")
  361. var resultCode = "0"
  362. var errorMsg = ""
  363. var status = ""
  364. for {
  365. task, err := models.GetCloudbrainByJobID(jobID)
  366. if err != nil {
  367. log.Error("GetCloudbrainByJobID(%s) failed:%v", task.JobName, err, ctx.Data["msgID"])
  368. resultCode = "-1"
  369. errorMsg = "system error"
  370. break
  371. }
  372. if task.Status == string(models.JobStopped) || task.Status == string(models.JobFailed) {
  373. log.Error("the job(%s) has been stopped", task.JobName, ctx.Data["msgID"])
  374. resultCode = "-1"
  375. errorMsg = "system error"
  376. break
  377. }
  378. err = cloudbrain.StopJob(jobID)
  379. if err != nil {
  380. log.Error("StopJob(%s) failed:%v", task.JobName, err, ctx.Data["msgID"])
  381. resultCode = "-1"
  382. errorMsg = "system error"
  383. break
  384. }
  385. task.Status = string(models.JobStopped)
  386. err = models.UpdateJob(task)
  387. if err != nil {
  388. log.Error("UpdateJob(%s) failed:%v", task.JobName, err, ctx.Data["msgID"])
  389. resultCode = "-1"
  390. errorMsg = "system error"
  391. break
  392. }
  393. status = task.Status
  394. break
  395. }
  396. ctx.JSON(200, map[string]string{
  397. "result_code": resultCode,
  398. "error_msg": errorMsg,
  399. "status": status,
  400. "job_id": jobID,
  401. })
  402. }
  403. func StopJobsByUserID(userID int64) {
  404. cloudBrains, err := models.GetCloudbrainsNeededStopByUserID(userID)
  405. if err != nil {
  406. log.Warn("Failed to get cloudBrain info", err)
  407. return
  408. }
  409. StopJobs(cloudBrains)
  410. }
  411. func StopJobsByRepoID(repoID int64) {
  412. cloudBrains, err := models.GetCloudbrainsNeededStopByRepoID(repoID)
  413. if err != nil {
  414. log.Warn("Failed to get cloudBrain info", err)
  415. return
  416. }
  417. StopJobs(cloudBrains)
  418. }
  419. /**
  420. */
  421. func StopJobs(cloudBrains []*models.Cloudbrain) {
  422. for _, taskInfo := range cloudBrains {
  423. if taskInfo.Type == models.TypeCloudBrainOne {
  424. err := retry(3, time.Second*30, func() error {
  425. return cloudbrain.StopJob(taskInfo.JobID)
  426. })
  427. logErrorAndUpdateJobStatus(err, taskInfo)
  428. } else {
  429. if taskInfo.JobType == string(models.JobTypeTrain) {
  430. err := retry(3, time.Second*30, func() error {
  431. _, err := modelarts.StopTrainJob(taskInfo.JobID, strconv.FormatInt(taskInfo.VersionID, 10))
  432. return err
  433. })
  434. logErrorAndUpdateJobStatus(err, taskInfo)
  435. } else {
  436. param := models.NotebookAction{
  437. Action: models.ActionStop,
  438. }
  439. err := retry(3, time.Second*30, func() error {
  440. _, err := modelarts.ManageNotebook(taskInfo.JobID, param)
  441. return err
  442. })
  443. logErrorAndUpdateJobStatus(err, taskInfo)
  444. }
  445. }
  446. }
  447. }
  448. func retry(attempts int, sleep time.Duration, f func() error) (err error) {
  449. for i := 0; i < attempts; i++ {
  450. if i > 0 {
  451. log.Warn("retrying after error:", err)
  452. time.Sleep(sleep)
  453. }
  454. err = f()
  455. if err == nil {
  456. return nil
  457. }
  458. }
  459. return fmt.Errorf("after %d attempts, last error: %s", attempts, err)
  460. }
  461. func logErrorAndUpdateJobStatus(err error, taskInfo *models.Cloudbrain) {
  462. if err != nil {
  463. log.Warn("Failed to stop cloudBrain job:"+taskInfo.JobID, err)
  464. } else {
  465. taskInfo.Status = string(models.JobStopped)
  466. err = models.UpdateJob(taskInfo)
  467. if err != nil {
  468. log.Warn("UpdateJob failed", err)
  469. }
  470. }
  471. }
  472. func CloudBrainDel(ctx *context.Context) {
  473. var jobID = ctx.Params(":jobid")
  474. task, err := models.GetCloudbrainByJobID(jobID)
  475. if err != nil {
  476. ctx.ServerError("GetCloudbrainByJobID failed", err)
  477. return
  478. }
  479. if task.Status != string(models.JobStopped) && task.Status != string(models.JobFailed) {
  480. log.Error("the job(%s) has not been stopped", task.JobName, ctx.Data["msgID"])
  481. ctx.ServerError("the job has not been stopped", errors.New("the job has not been stopped"))
  482. return
  483. }
  484. err = models.DeleteJob(task)
  485. if err != nil {
  486. ctx.ServerError("DeleteJob failed", err)
  487. return
  488. }
  489. ctx.Redirect(setting.AppSubURL + ctx.Repo.RepoLink + "/debugjob")
  490. }
  491. func CloudBrainShowModels(ctx *context.Context) {
  492. ctx.Data["PageIsCloudBrain"] = true
  493. jobID := ctx.Params(":jobid")
  494. parentDir := ctx.Query("parentDir")
  495. dirArray := strings.Split(parentDir, "/")
  496. task, err := models.GetCloudbrainByJobID(jobID)
  497. if err != nil {
  498. log.Error("no such job!", ctx.Data["msgID"])
  499. ctx.ServerError("no such job:", err)
  500. return
  501. }
  502. //get dirs
  503. dirs, err := GetModelDirs(task.JobName, parentDir)
  504. if err != nil {
  505. log.Error("GetModelDirs failed:%v", err.Error(), ctx.Data["msgID"])
  506. ctx.ServerError("GetModelDirs failed:", err)
  507. return
  508. }
  509. var fileInfos []storage.FileInfo
  510. err = json.Unmarshal([]byte(dirs), &fileInfos)
  511. if err != nil {
  512. log.Error("json.Unmarshal failed:%v", err.Error(), ctx.Data["msgID"])
  513. ctx.ServerError("json.Unmarshal failed:", err)
  514. return
  515. }
  516. for i, fileInfo := range fileInfos {
  517. temp, _ := time.Parse("2006-01-02 15:04:05", fileInfo.ModTime)
  518. fileInfos[i].ModTime = temp.Local().Format("2006-01-02 15:04:05")
  519. }
  520. sort.Slice(fileInfos, func(i, j int) bool {
  521. return fileInfos[i].ModTime > fileInfos[j].ModTime
  522. })
  523. ctx.Data["Path"] = dirArray
  524. ctx.Data["Dirs"] = fileInfos
  525. ctx.Data["task"] = task
  526. ctx.Data["JobID"] = jobID
  527. ctx.HTML(200, tplCloudBrainShowModels)
  528. }
  529. func GetPublicImages(ctx *context.Context) {
  530. getImages(ctx, cloudbrain.Public)
  531. }
  532. func GetCustomImages(ctx *context.Context) {
  533. getImages(ctx, cloudbrain.Custom)
  534. }
  535. func getImages(ctx *context.Context, imageType string) {
  536. log.Info("Get images begin")
  537. page := ctx.QueryInt("page")
  538. size := ctx.QueryInt("size")
  539. name := ctx.Query("name")
  540. getImagesResult, err := cloudbrain.GetImagesPageable(page, size, imageType, name)
  541. if err != nil {
  542. log.Error("Can not get images:%v", err)
  543. ctx.JSON(http.StatusOK, models.GetImagesPayload{
  544. Count: 0,
  545. TotalPages: 0,
  546. ImageInfo: []*models.ImageInfo{},
  547. })
  548. } else {
  549. ctx.JSON(http.StatusOK, getImagesResult.Payload)
  550. }
  551. log.Info("Get images end")
  552. }
  553. func GetModelDirs(jobName string, parentDir string) (string, error) {
  554. var req string
  555. modelActualPath := storage.GetMinioPath(jobName, cloudbrain.ModelMountPath+"/")
  556. if parentDir == "" {
  557. req = "baseDir=" + modelActualPath
  558. } else {
  559. req = "baseDir=" + modelActualPath + "&parentDir=" + parentDir
  560. }
  561. return getDirs(req)
  562. }
  563. func CloudBrainDownloadModel(ctx *context.Context) {
  564. parentDir := ctx.Query("parentDir")
  565. fileName := ctx.Query("fileName")
  566. jobName := ctx.Query("jobName")
  567. filePath := "jobs/" + jobName + "/model/" + parentDir
  568. url, err := storage.Attachments.PresignedGetURL(filePath, fileName)
  569. if err != nil {
  570. log.Error("PresignedGetURL failed: %v", err.Error(), ctx.Data["msgID"])
  571. ctx.ServerError("PresignedGetURL", err)
  572. return
  573. }
  574. http.Redirect(ctx.Resp, ctx.Req.Request, url, http.StatusMovedPermanently)
  575. }
  576. func GetRate(ctx *context.Context) {
  577. var jobID = ctx.Params(":jobid")
  578. job, err := models.GetCloudbrainByJobID(jobID)
  579. if err != nil {
  580. ctx.ServerError("GetCloudbrainByJobID failed", err)
  581. return
  582. }
  583. if job.JobType == string(models.JobTypeBenchmark) {
  584. ctx.Redirect(setting.BenchmarkServerHost + "?username=" + ctx.User.Name)
  585. } else if job.JobType == string(models.JobTypeSnn4imagenet) {
  586. ctx.Redirect(setting.Snn4imagenetServerHost)
  587. } else if job.JobType == string(models.JobTypeBrainScore) {
  588. ctx.Redirect(setting.BrainScoreServerHost)
  589. } else {
  590. log.Error("JobType error:%s", job.JobType, ctx.Data["msgID"])
  591. }
  592. }
  593. func downloadCode(repo *models.Repository, codePath string) error {
  594. if err := git.Clone(repo.RepoPath(), codePath, git.CloneRepoOptions{}); err != nil {
  595. log.Error("Failed to clone repository: %s (%v)", repo.FullName(), err)
  596. return err
  597. }
  598. configFile, err := os.OpenFile(codePath+"/.git/config", os.O_RDWR, 0666)
  599. if err != nil {
  600. log.Error("open file(%s) failed:%v", codePath+"/,git/config", err)
  601. return err
  602. }
  603. defer configFile.Close()
  604. pos := int64(0)
  605. reader := bufio.NewReader(configFile)
  606. for {
  607. line, err := reader.ReadString('\n')
  608. if err != nil {
  609. if err == io.EOF {
  610. log.Error("not find the remote-url")
  611. return nil
  612. } else {
  613. log.Error("read error: %v", err)
  614. return err
  615. }
  616. }
  617. if strings.Contains(line, "url") && strings.Contains(line, ".git") {
  618. originUrl := "\turl = " + repo.CloneLink().HTTPS + "\n"
  619. if len(line) > len(originUrl) {
  620. originUrl += strings.Repeat(" ", len(line)-len(originUrl))
  621. }
  622. bytes := []byte(originUrl)
  623. _, err := configFile.WriteAt(bytes, pos)
  624. if err != nil {
  625. log.Error("WriteAt failed:%v", err)
  626. return err
  627. }
  628. break
  629. }
  630. pos += int64(len(line))
  631. }
  632. return nil
  633. }
  634. func downloadRateCode(repo *models.Repository, taskName, rateOwnerName, rateRepoName, codePath, benchmarkCategory, gpuType string) error {
  635. err := os.MkdirAll(codePath, os.ModePerm)
  636. if err != nil {
  637. log.Error("mkdir codePath failed", err.Error())
  638. return err
  639. }
  640. repoExt, err := models.GetRepositoryByOwnerAndName(rateOwnerName, rateRepoName)
  641. if err != nil {
  642. log.Error("GetRepositoryByOwnerAndName(%s) failed", rateRepoName, err.Error())
  643. return err
  644. }
  645. if err := git.Clone(repoExt.RepoPath(), codePath, git.CloneRepoOptions{}); err != nil {
  646. log.Error("Failed to clone repository: %s (%v)", repoExt.FullName(), err)
  647. return err
  648. }
  649. fileName := codePath + cloudbrain.TaskInfoName
  650. f, err := os.OpenFile(fileName, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, os.ModePerm)
  651. if err != nil {
  652. log.Error("OpenFile failed", err.Error())
  653. return err
  654. }
  655. defer f.Close()
  656. data, err := json.Marshal(models.TaskInfo{
  657. Username: repo.Owner.Name,
  658. TaskName: taskName,
  659. CodeName: repo.Name,
  660. BenchmarkCategory: strings.Split(benchmarkCategory, ","),
  661. CodeLink: strings.TrimSuffix(repo.CloneLink().HTTPS, ".git"),
  662. GpuType: gpuType,
  663. })
  664. if err != nil {
  665. log.Error("json.Marshal failed", err.Error())
  666. return err
  667. }
  668. _, err = f.Write(data)
  669. if err != nil {
  670. log.Error("WriteString failed", err.Error())
  671. return err
  672. }
  673. return nil
  674. }
  675. func uploadCodeToMinio(codePath, jobName, parentDir string) error {
  676. files, err := readDir(codePath)
  677. if err != nil {
  678. log.Error("readDir(%s) failed: %s", codePath, err.Error())
  679. return err
  680. }
  681. for _, file := range files {
  682. if file.IsDir() {
  683. if err = uploadCodeToMinio(codePath+file.Name()+"/", jobName, parentDir+file.Name()+"/"); err != nil {
  684. log.Error("uploadCodeToMinio(%s) failed: %s", file.Name(), err.Error())
  685. return err
  686. }
  687. } else {
  688. destObject := setting.CBCodePathPrefix + jobName + parentDir + file.Name()
  689. sourceFile := codePath + file.Name()
  690. err = storage.Attachments.UploadObject(destObject, sourceFile)
  691. if err != nil {
  692. log.Error("UploadObject(%s) failed: %s", file.Name(), err.Error())
  693. return err
  694. }
  695. }
  696. }
  697. return nil
  698. }
  699. func mkModelPath(modelPath string) error {
  700. err := os.MkdirAll(modelPath, os.ModePerm)
  701. if err != nil {
  702. log.Error("MkdirAll(%s) failed:%v", modelPath, err)
  703. return err
  704. }
  705. fileName := modelPath + "README"
  706. f, err := os.OpenFile(fileName, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, os.ModePerm)
  707. if err != nil {
  708. log.Error("OpenFile failed", err.Error())
  709. return err
  710. }
  711. defer f.Close()
  712. _, err = f.WriteString("You can put the model file into this directory and download it by the web page.")
  713. if err != nil {
  714. log.Error("WriteString failed", err.Error())
  715. return err
  716. }
  717. return nil
  718. }
  719. func SyncCloudbrainStatus() {
  720. cloudBrains, err := models.GetCloudBrainUnStoppedJob()
  721. if err != nil {
  722. log.Error("GetCloudBrainUnStoppedJob failed:", err.Error())
  723. return
  724. }
  725. for _, task := range cloudBrains {
  726. if task.Type == models.TypeCloudBrainOne {
  727. result, err := cloudbrain.GetJob(task.JobID)
  728. if err != nil {
  729. log.Error("GetJob(%s) failed:%v", task.JobName, err)
  730. continue
  731. }
  732. if result != nil {
  733. jobRes, _ := models.ConvertToJobResultPayload(result.Payload)
  734. taskRoles := jobRes.TaskRoles
  735. taskRes, _ := models.ConvertToTaskPod(taskRoles[cloudbrain.SubTaskName].(map[string]interface{}))
  736. task.Status = taskRes.TaskStatuses[0].State
  737. if task.Status != string(models.JobWaiting) {
  738. task.Duration = time.Now().Unix() - taskRes.TaskStatuses[0].StartAt.Unix()
  739. err = models.UpdateJob(task)
  740. if err != nil {
  741. log.Error("UpdateJob(%s) failed:%v", task.JobName, err)
  742. }
  743. if task.Duration >= setting.MaxDuration {
  744. log.Info("begin to stop job(%s), because of the duration", task.JobName)
  745. err = cloudbrain.StopJob(task.JobID)
  746. if err != nil {
  747. log.Error("StopJob(%s) failed:%v", task.JobName, err)
  748. continue
  749. }
  750. task.Status = string(models.JobStopped)
  751. err = models.UpdateJob(task)
  752. if err != nil {
  753. log.Error("UpdateJob(%s) failed:%v", task.JobName, err)
  754. }
  755. }
  756. }
  757. }
  758. } else if task.Type == models.TypeCloudBrainTwo {
  759. if task.JobType == string(models.JobTypeDebug) {
  760. result, err := modelarts.GetJob(task.JobID)
  761. if err != nil {
  762. log.Error("GetJob(%s) failed:%v", task.JobName, err)
  763. continue
  764. }
  765. if result != nil {
  766. task.Status = result.Status
  767. err = models.UpdateJob(task)
  768. if err != nil {
  769. log.Error("UpdateJob(%s) failed:%v", task.JobName, err)
  770. continue
  771. }
  772. }
  773. } else if task.JobType == string(models.JobTypeTrain) {
  774. result, err := modelarts.GetTrainJob(task.JobID, strconv.FormatInt(task.VersionID, 10))
  775. if err != nil {
  776. log.Error("GetTrainJob(%s) failed:%v", task.JobName, err)
  777. continue
  778. }
  779. if result != nil {
  780. task.Status = modelarts.TransTrainJobStatus(result.IntStatus)
  781. task.Duration = result.Duration
  782. task.TrainJobDuration = result.TrainJobDuration
  783. err = models.UpdateJob(task)
  784. if err != nil {
  785. log.Error("UpdateJob(%s) failed:%v", task.JobName, err)
  786. continue
  787. }
  788. }
  789. } else {
  790. log.Error("task.JobType(%s) is error:%s", task.JobName, task.JobType)
  791. }
  792. } else {
  793. log.Error("task.Type(%s) is error:%d", task.JobName, task.Type)
  794. }
  795. }
  796. return
  797. }