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.

grampus.go 28 kB

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
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
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
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
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
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
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
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
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
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
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
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857
  1. package repo
  2. import (
  3. "encoding/json"
  4. "errors"
  5. "io/ioutil"
  6. "net/http"
  7. "os"
  8. "path"
  9. "strconv"
  10. "strings"
  11. "time"
  12. "code.gitea.io/gitea/modules/auth"
  13. "code.gitea.io/gitea/modules/git"
  14. "code.gitea.io/gitea/modules/grampus"
  15. "code.gitea.io/gitea/modules/modelarts"
  16. "code.gitea.io/gitea/modules/notification"
  17. "code.gitea.io/gitea/modules/timeutil"
  18. "code.gitea.io/gitea/modules/util"
  19. "github.com/unknwon/com"
  20. "code.gitea.io/gitea/models"
  21. "code.gitea.io/gitea/modules/base"
  22. "code.gitea.io/gitea/modules/cloudbrain"
  23. "code.gitea.io/gitea/modules/context"
  24. "code.gitea.io/gitea/modules/log"
  25. "code.gitea.io/gitea/modules/setting"
  26. )
  27. const (
  28. tplGrampusTrainJobShow base.TplName = "repo/grampus/trainjob/show"
  29. //GPU
  30. tplGrampusTrainJobGPUNew base.TplName = "repo/grampus/trainjob/gpu/new"
  31. //NPU
  32. tplGrampusTrainJobNPUNew base.TplName = "repo/grampus/trainjob/npu/new"
  33. )
  34. func GrampusTrainJobGPUNew(ctx *context.Context) {
  35. ctx.Data["datasetType"] = models.TypeCloudBrainOne
  36. err := grampusTrainJobNewDataPrepare(ctx, grampus.ProcessorTypeGPU)
  37. if err != nil {
  38. ctx.ServerError("get new train-job info failed", err)
  39. return
  40. }
  41. ctx.HTML(http.StatusOK, tplGrampusTrainJobGPUNew)
  42. }
  43. func GrampusTrainJobNPUNew(ctx *context.Context) {
  44. ctx.Data["datasetType"] = models.TypeCloudBrainTwo
  45. err := grampusTrainJobNewDataPrepare(ctx, grampus.ProcessorTypeNPU)
  46. if err != nil {
  47. ctx.ServerError("get new train-job info failed", err)
  48. return
  49. }
  50. ctx.HTML(200, tplGrampusTrainJobNPUNew)
  51. }
  52. func grampusTrainJobNewDataPrepare(ctx *context.Context, processType string) error {
  53. ctx.Data["PageIsCloudBrain"] = true
  54. t := time.Now()
  55. var displayJobName = jobNamePrefixValid(cutString(ctx.User.Name, 5)) + t.Format("2006010215") + strconv.Itoa(int(t.Unix()))[5:]
  56. ctx.Data["display_job_name"] = displayJobName
  57. //get valid images
  58. images, err := grampus.GetImages(processType)
  59. if err != nil {
  60. log.Error("GetImages failed:", err.Error())
  61. } else {
  62. ctx.Data["images"] = images.Infos
  63. }
  64. grampus.InitSpecialPool()
  65. ctx.Data["GPUEnabled"] = true
  66. ctx.Data["NPUEnabled"] = true
  67. includeCenters := make(map[string]struct{})
  68. excludeCenters := make(map[string]struct{})
  69. if grampus.SpecialPools != nil {
  70. for _, pool := range grampus.SpecialPools.Pools {
  71. if pool.IsExclusive {
  72. if !IsUserInOrgPool(ctx.User.ID, pool) {
  73. ctx.Data[pool.Type+"Enabled"] = false
  74. }
  75. } else {
  76. if strings.Contains(strings.ToLower(processType), strings.ToLower(pool.Type)) {
  77. if IsUserInOrgPool(ctx.User.ID, pool) {
  78. for _, center := range pool.Pool {
  79. includeCenters[center.Queue] = struct{}{}
  80. }
  81. } else {
  82. for _, center := range pool.Pool {
  83. excludeCenters[center.Queue] = struct{}{}
  84. }
  85. }
  86. }
  87. }
  88. }
  89. }
  90. //get valid resource specs
  91. specs, err := grampus.GetResourceSpecs(processType)
  92. grampusSpecs := getFilterSpecBySpecialPool(specs, includeCenters, excludeCenters)
  93. if err != nil {
  94. log.Error("GetResourceSpecs failed:", err.Error())
  95. } else {
  96. ctx.Data["flavor_infos"] = grampusSpecs
  97. }
  98. //get branches
  99. branches, _, err := ctx.Repo.GitRepo.GetBranches(0, 0)
  100. if err != nil {
  101. log.Error("GetBranches error:", err.Error())
  102. } else {
  103. ctx.Data["branches"] = branches
  104. }
  105. ctx.Data["branchName"] = ctx.Repo.BranchName
  106. if processType == grampus.ProcessorTypeGPU {
  107. ctx.Data["datasetType"] = models.TypeCloudBrainOne
  108. waitCount := cloudbrain.GetWaitingCloudbrainCount(models.TypeC2Net, models.GPUResource, models.JobTypeTrain)
  109. ctx.Data["WaitCount"] = waitCount
  110. } else if processType == grampus.ProcessorTypeNPU {
  111. ctx.Data["datasetType"] = models.TypeCloudBrainTwo
  112. waitCount := cloudbrain.GetWaitingCloudbrainCount(models.TypeC2Net, models.NPUResource, models.JobTypeTrain)
  113. ctx.Data["WaitCount"] = waitCount
  114. }
  115. return nil
  116. }
  117. func getFilterSpecBySpecialPool(specs *models.GetGrampusResourceSpecsResult, includeCenters map[string]struct{}, excludeCenters map[string]struct{}) []models.GrampusSpec {
  118. if len(includeCenters) == 0 && len(excludeCenters) == 0 {
  119. return specs.Infos
  120. }
  121. var grampusSpecs []models.GrampusSpec
  122. for _, info := range specs.Infos {
  123. if isInIncludeCenters(info, includeCenters) || (len(excludeCenters) != 0 && isNotAllInExcludeCenters(info, excludeCenters)) {
  124. grampusSpecs = append(grampusSpecs, info)
  125. }
  126. }
  127. return grampusSpecs
  128. }
  129. func isInIncludeCenters(grampusSpec models.GrampusSpec, centers map[string]struct{}) bool {
  130. for _, center := range grampusSpec.Centers {
  131. if _, ok := centers[center.ID]; ok {
  132. return true
  133. }
  134. }
  135. return false
  136. }
  137. func isNotAllInExcludeCenters(grampusSpec models.GrampusSpec, centers map[string]struct{}) bool {
  138. for _, center := range grampusSpec.Centers {
  139. if _, ok := centers[center.ID]; !ok {
  140. return true
  141. }
  142. }
  143. return false
  144. }
  145. func IsUserInOrgPool(userId int64, pool *models.SpecialPool) bool {
  146. org, _ := models.GetOrgByName(pool.Org)
  147. if org != nil {
  148. isOrgMember, _ := models.IsOrganizationMember(org.ID, userId)
  149. return isOrgMember
  150. }
  151. return false
  152. }
  153. func grampusParamCheckCreateTrainJob(form auth.CreateGrampusTrainJobForm) error {
  154. if !strings.HasSuffix(strings.TrimSpace(form.BootFile), ".py") {
  155. log.Error("the boot file(%s) must be a python file", form.BootFile)
  156. return errors.New("启动文件必须是python文件")
  157. }
  158. if form.BranchName == "" {
  159. log.Error("the branch must not be null!", form.BranchName)
  160. return errors.New("代码分支不能为空!")
  161. }
  162. return nil
  163. }
  164. func GrampusTrainJobGpuCreate(ctx *context.Context, form auth.CreateGrampusTrainJobForm) {
  165. displayJobName := form.DisplayJobName
  166. jobName := util.ConvertDisplayJobNameToJobName(displayJobName)
  167. uuid := form.Attachment
  168. description := form.Description
  169. bootFile := strings.TrimSpace(form.BootFile)
  170. params := form.Params
  171. repo := ctx.Repo.Repository
  172. codeLocalPath := setting.JobPath + jobName + cloudbrain.CodeMountPath + "/"
  173. codeMinioPath := setting.CBCodePathPrefix + jobName + cloudbrain.CodeMountPath + "/"
  174. dataMinioPath := setting.Attachment.Minio.BasePath + path.Join(uuid[0:1], uuid[1:2]) + "/" + uuid
  175. branchName := form.BranchName
  176. flavorName := form.FlavorName
  177. image := strings.TrimSpace(form.Image)
  178. if !jobNamePattern.MatchString(displayJobName) {
  179. grampusTrainJobNewDataPrepare(ctx, grampus.ProcessorTypeGPU)
  180. ctx.RenderWithErr(ctx.Tr("repo.cloudbrain_jobname_err"), tplGrampusTrainJobGPUNew, &form)
  181. return
  182. }
  183. errStr := checkSpecialPool(ctx, "GPU")
  184. if errStr != "" {
  185. grampusTrainJobNewDataPrepare(ctx, grampus.ProcessorTypeGPU)
  186. ctx.RenderWithErr(errStr, tplGrampusTrainJobGPUNew, &form)
  187. return
  188. }
  189. //check count limit
  190. count, err := models.GetGrampusCountByUserID(ctx.User.ID, string(models.JobTypeTrain), models.GPUResource)
  191. if err != nil {
  192. log.Error("GetGrampusCountByUserID failed:%v", err, ctx.Data["MsgID"])
  193. grampusTrainJobNewDataPrepare(ctx, grampus.ProcessorTypeGPU)
  194. ctx.RenderWithErr("system error", tplGrampusTrainJobGPUNew, &form)
  195. return
  196. } else {
  197. if count >= 1 {
  198. log.Error("the user already has running or waiting task", ctx.Data["MsgID"])
  199. grampusTrainJobNewDataPrepare(ctx, grampus.ProcessorTypeGPU)
  200. ctx.RenderWithErr("you have already a running or waiting task, can not create more", tplGrampusTrainJobGPUNew, &form)
  201. return
  202. }
  203. }
  204. //check param
  205. if err := grampusParamCheckCreateTrainJob(form); err != nil {
  206. log.Error("paramCheckCreateTrainJob failed:(%v)", err, ctx.Data["MsgID"])
  207. grampusTrainJobNewDataPrepare(ctx, grampus.ProcessorTypeGPU)
  208. ctx.RenderWithErr(err.Error(), tplGrampusTrainJobGPUNew, &form)
  209. return
  210. }
  211. //check whether the task name in the project is duplicated
  212. tasks, err := models.GetCloudbrainsByDisplayJobName(repo.ID, string(models.JobTypeTrain), displayJobName)
  213. if err == nil {
  214. if len(tasks) != 0 {
  215. log.Error("the job name did already exist", ctx.Data["MsgID"])
  216. grampusTrainJobNewDataPrepare(ctx, grampus.ProcessorTypeGPU)
  217. ctx.RenderWithErr("the job name did already exist", tplGrampusTrainJobGPUNew, &form)
  218. return
  219. }
  220. } else {
  221. if !models.IsErrJobNotExist(err) {
  222. log.Error("system error, %v", err, ctx.Data["MsgID"])
  223. grampusTrainJobNewDataPrepare(ctx, grampus.ProcessorTypeGPU)
  224. ctx.RenderWithErr("system error", tplGrampusTrainJobGPUNew, &form)
  225. return
  226. }
  227. }
  228. //check dataset
  229. attachment, err := models.GetAttachmentByUUID(uuid)
  230. if err != nil {
  231. log.Error("GetAttachmentByUUID failed:", err.Error(), ctx.Data["MsgID"])
  232. grampusTrainJobNewDataPrepare(ctx, grampus.ProcessorTypeGPU)
  233. ctx.RenderWithErr("dataset is not exist", tplGrampusTrainJobGPUNew, &form)
  234. return
  235. }
  236. //prepare code and out path
  237. _, err = ioutil.ReadDir(codeLocalPath)
  238. if err == nil {
  239. os.RemoveAll(codeLocalPath)
  240. }
  241. if err := downloadZipCode(ctx, codeLocalPath, branchName); err != nil {
  242. log.Error("downloadZipCode failed, server timed out: %s (%v)", repo.FullName(), err, ctx.Data["MsgID"])
  243. grampusTrainJobNewDataPrepare(ctx, grampus.ProcessorTypeGPU)
  244. ctx.RenderWithErr(ctx.Tr("cloudbrain.load_code_failed"), tplGrampusTrainJobGPUNew, &form)
  245. return
  246. }
  247. //todo: upload code (send to file_server todo this work?)
  248. //upload code
  249. if err := uploadCodeToMinio(codeLocalPath+"/", jobName, cloudbrain.CodeMountPath+"/"); err != nil {
  250. log.Error("Failed to uploadCodeToMinio: %s (%v)", repo.FullName(), err, ctx.Data["MsgID"])
  251. grampusTrainJobNewDataPrepare(ctx, grampus.ProcessorTypeGPU)
  252. ctx.RenderWithErr(ctx.Tr("cloudbrain.load_code_failed"), tplGrampusTrainJobGPUNew, &form)
  253. return
  254. }
  255. modelPath := setting.JobPath + jobName + cloudbrain.ModelMountPath + "/"
  256. if err := mkModelPath(modelPath); err != nil {
  257. log.Error("Failed to mkModelPath: %s (%v)", repo.FullName(), err, ctx.Data["MsgID"])
  258. grampusTrainJobNewDataPrepare(ctx, grampus.ProcessorTypeGPU)
  259. ctx.RenderWithErr(ctx.Tr("cloudbrain.load_code_failed"), tplGrampusTrainJobGPUNew, &form)
  260. return
  261. }
  262. //init model readme
  263. if err := uploadCodeToMinio(modelPath, jobName, cloudbrain.ModelMountPath+"/"); err != nil {
  264. log.Error("Failed to uploadCodeToMinio: %s (%v)", repo.FullName(), err, ctx.Data["MsgID"])
  265. grampusTrainJobNewDataPrepare(ctx, grampus.ProcessorTypeGPU)
  266. ctx.RenderWithErr(ctx.Tr("cloudbrain.load_code_failed"), tplGrampusTrainJobGPUNew, &form)
  267. return
  268. }
  269. //prepare command
  270. command, err := generateCommand(repo.Name, grampus.ProcessorTypeGPU, codeMinioPath+cloudbrain.DefaultBranchName+".zip", dataMinioPath, bootFile, params, setting.CBCodePathPrefix+jobName+cloudbrain.ModelMountPath+"/", attachment.Name)
  271. if err != nil {
  272. log.Error("Failed to generateCommand: %s (%v)", displayJobName, err, ctx.Data["MsgID"])
  273. grampusTrainJobNewDataPrepare(ctx, grampus.ProcessorTypeGPU)
  274. ctx.RenderWithErr("Create task failed, internal error", tplGrampusTrainJobGPUNew, &form)
  275. return
  276. }
  277. commitID, _ := ctx.Repo.GitRepo.GetBranchCommitID(branchName)
  278. req := &grampus.GenerateTrainJobReq{
  279. JobName: jobName,
  280. DisplayJobName: displayJobName,
  281. ComputeResource: models.GPUResource,
  282. ProcessType: grampus.ProcessorTypeGPU,
  283. Command: command,
  284. ResourceSpecId: form.FlavorID,
  285. ImageUrl: image,
  286. Description: description,
  287. BootFile: bootFile,
  288. Uuid: uuid,
  289. CommitID: commitID,
  290. BranchName: branchName,
  291. Params: form.Params,
  292. FlavorName: flavorName,
  293. EngineName: image,
  294. DatasetName: attachment.Name,
  295. IsLatestVersion: modelarts.IsLatestVersion,
  296. VersionCount: modelarts.VersionCount,
  297. WorkServerNumber: 1,
  298. }
  299. err = grampus.GenerateTrainJob(ctx, req)
  300. if err != nil {
  301. log.Error("GenerateTrainJob failed:%v", err.Error(), ctx.Data["MsgID"])
  302. grampusTrainJobNewDataPrepare(ctx, grampus.ProcessorTypeGPU)
  303. ctx.RenderWithErr(err.Error(), tplGrampusTrainJobGPUNew, &form)
  304. return
  305. }
  306. ctx.Redirect(setting.AppSubURL + ctx.Repo.RepoLink + "/modelarts/train-job")
  307. }
  308. func checkSpecialPool(ctx *context.Context, resourceType string) string {
  309. grampus.InitSpecialPool()
  310. if grampus.SpecialPools != nil {
  311. for _, pool := range grampus.SpecialPools.Pools {
  312. if pool.IsExclusive && pool.Type == resourceType {
  313. org, _ := models.GetOrgByName(pool.Org)
  314. if org != nil {
  315. isOrgMember, _ := models.IsOrganizationMember(org.ID, ctx.User.ID)
  316. if !isOrgMember {
  317. return ctx.Tr("repo.grampus.no_operate_right")
  318. }
  319. }
  320. }
  321. }
  322. }
  323. return ""
  324. }
  325. func GrampusTrainJobNpuCreate(ctx *context.Context, form auth.CreateGrampusTrainJobForm) {
  326. displayJobName := form.DisplayJobName
  327. jobName := util.ConvertDisplayJobNameToJobName(displayJobName)
  328. uuid := form.Attachment
  329. description := form.Description
  330. bootFile := strings.TrimSpace(form.BootFile)
  331. params := form.Params
  332. repo := ctx.Repo.Repository
  333. codeLocalPath := setting.JobPath + jobName + modelarts.CodePath
  334. codeObsPath := grampus.JobPath + jobName + modelarts.CodePath
  335. dataObsPath := setting.BasePath + path.Join(uuid[0:1], uuid[1:2]) + "/" + uuid + "/"
  336. branchName := form.BranchName
  337. isLatestVersion := modelarts.IsLatestVersion
  338. flavorName := form.FlavorName
  339. versionCount := modelarts.VersionCount
  340. engineName := form.EngineName
  341. if !jobNamePattern.MatchString(displayJobName) {
  342. grampusTrainJobNewDataPrepare(ctx, grampus.ProcessorTypeNPU)
  343. ctx.RenderWithErr(ctx.Tr("repo.cloudbrain_jobname_err"), tplGrampusTrainJobNPUNew, &form)
  344. return
  345. }
  346. errStr := checkSpecialPool(ctx, "NPU")
  347. if errStr != "" {
  348. grampusTrainJobNewDataPrepare(ctx, grampus.ProcessorTypeNPU)
  349. ctx.RenderWithErr(errStr, tplGrampusTrainJobGPUNew, &form)
  350. return
  351. }
  352. //check count limit
  353. count, err := models.GetGrampusCountByUserID(ctx.User.ID, string(models.JobTypeTrain), models.NPUResource)
  354. if err != nil {
  355. log.Error("GetGrampusCountByUserID failed:%v", err, ctx.Data["MsgID"])
  356. grampusTrainJobNewDataPrepare(ctx, grampus.ProcessorTypeNPU)
  357. ctx.RenderWithErr("system error", tplGrampusTrainJobNPUNew, &form)
  358. return
  359. } else {
  360. if count >= 1 {
  361. log.Error("the user already has running or waiting task", ctx.Data["MsgID"])
  362. grampusTrainJobNewDataPrepare(ctx, grampus.ProcessorTypeNPU)
  363. ctx.RenderWithErr("you have already a running or waiting task, can not create more", tplGrampusTrainJobNPUNew, &form)
  364. return
  365. }
  366. }
  367. //check param
  368. if err := grampusParamCheckCreateTrainJob(form); err != nil {
  369. log.Error("paramCheckCreateTrainJob failed:(%v)", err)
  370. grampusTrainJobNewDataPrepare(ctx, grampus.ProcessorTypeNPU)
  371. ctx.RenderWithErr(err.Error(), tplGrampusTrainJobNPUNew, &form)
  372. return
  373. }
  374. //check whether the task name in the project is duplicated
  375. tasks, err := models.GetCloudbrainsByDisplayJobName(repo.ID, string(models.JobTypeTrain), displayJobName)
  376. if err == nil {
  377. if len(tasks) != 0 {
  378. log.Error("the job name did already exist", ctx.Data["MsgID"])
  379. grampusTrainJobNewDataPrepare(ctx, grampus.ProcessorTypeNPU)
  380. ctx.RenderWithErr("the job name did already exist", tplGrampusTrainJobNPUNew, &form)
  381. return
  382. }
  383. } else {
  384. if !models.IsErrJobNotExist(err) {
  385. log.Error("system error, %v", err, ctx.Data["MsgID"])
  386. grampusTrainJobNewDataPrepare(ctx, grampus.ProcessorTypeNPU)
  387. ctx.RenderWithErr("system error", tplGrampusTrainJobNPUNew, &form)
  388. return
  389. }
  390. }
  391. //check dataset
  392. attachment, err := models.GetAttachmentByUUID(uuid)
  393. if err != nil {
  394. log.Error("GetAttachmentByUUID failed:", err.Error(), ctx.Data["MsgID"])
  395. grampusTrainJobNewDataPrepare(ctx, grampus.ProcessorTypeNPU)
  396. ctx.RenderWithErr("dataset is not exist", tplGrampusTrainJobNPUNew, &form)
  397. return
  398. }
  399. //prepare code and out path
  400. _, err = ioutil.ReadDir(codeLocalPath)
  401. if err == nil {
  402. os.RemoveAll(codeLocalPath)
  403. }
  404. if err := downloadZipCode(ctx, codeLocalPath, branchName); err != nil {
  405. log.Error("downloadZipCode failed, server timed out: %s (%v)", repo.FullName(), err)
  406. grampusTrainJobNewDataPrepare(ctx, grampus.ProcessorTypeNPU)
  407. ctx.RenderWithErr(ctx.Tr("cloudbrain.load_code_failed"), tplGrampusTrainJobNPUNew, &form)
  408. return
  409. }
  410. //todo: upload code (send to file_server todo this work?)
  411. if err := obsMkdir(setting.CodePathPrefix + jobName + modelarts.OutputPath); err != nil {
  412. log.Error("Failed to obsMkdir_output: %s (%v)", repo.FullName(), err)
  413. grampusTrainJobNewDataPrepare(ctx, grampus.ProcessorTypeNPU)
  414. ctx.RenderWithErr(ctx.Tr("cloudbrain.load_code_failed"), tplGrampusTrainJobNPUNew, &form)
  415. return
  416. }
  417. if err := uploadCodeToObs(codeLocalPath, jobName, ""); err != nil {
  418. log.Error("Failed to uploadCodeToObs: %s (%v)", repo.FullName(), err)
  419. grampusTrainJobNewDataPrepare(ctx, grampus.ProcessorTypeNPU)
  420. ctx.RenderWithErr(ctx.Tr("cloudbrain.load_code_failed"), tplGrampusTrainJobNPUNew, &form)
  421. return
  422. }
  423. //prepare command
  424. command, err := generateCommand(repo.Name, grampus.ProcessorTypeNPU, codeObsPath+cloudbrain.DefaultBranchName+".zip", dataObsPath+"'"+attachment.Name+"'", bootFile, params, setting.CodePathPrefix+jobName+modelarts.OutputPath, attachment.Name)
  425. if err != nil {
  426. log.Error("Failed to generateCommand: %s (%v)", displayJobName, err, ctx.Data["MsgID"])
  427. grampusTrainJobNewDataPrepare(ctx, grampus.ProcessorTypeNPU)
  428. ctx.RenderWithErr("Create task failed, internal error", tplGrampusTrainJobNPUNew, &form)
  429. return
  430. }
  431. commitID, _ := ctx.Repo.GitRepo.GetBranchCommitID(branchName)
  432. req := &grampus.GenerateTrainJobReq{
  433. JobName: jobName,
  434. DisplayJobName: displayJobName,
  435. ComputeResource: models.NPUResource,
  436. ProcessType: grampus.ProcessorTypeNPU,
  437. Command: command,
  438. ResourceSpecId: form.FlavorID,
  439. ImageId: form.ImageID,
  440. DataUrl: dataObsPath,
  441. Description: description,
  442. CodeObsPath: codeObsPath,
  443. BootFileUrl: codeObsPath + bootFile,
  444. BootFile: bootFile,
  445. WorkServerNumber: form.WorkServerNumber,
  446. Uuid: uuid,
  447. CommitID: commitID,
  448. IsLatestVersion: isLatestVersion,
  449. BranchName: branchName,
  450. Params: form.Params,
  451. FlavorName: flavorName,
  452. EngineName: engineName,
  453. VersionCount: versionCount,
  454. TotalVersionCount: modelarts.TotalVersionCount,
  455. DatasetName: attachment.Name,
  456. }
  457. err = grampus.GenerateTrainJob(ctx, req)
  458. if err != nil {
  459. log.Error("GenerateTrainJob failed:%v", err.Error())
  460. grampusTrainJobNewDataPrepare(ctx, grampus.ProcessorTypeNPU)
  461. ctx.RenderWithErr(err.Error(), tplGrampusTrainJobNPUNew, &form)
  462. return
  463. }
  464. ctx.Redirect(setting.AppSubURL + ctx.Repo.RepoLink + "/modelarts/train-job")
  465. }
  466. func GrampusStopJob(ctx *context.Context) {
  467. var ID = ctx.Params(":jobid")
  468. var resultCode = "0"
  469. var errorMsg = ""
  470. var status = ""
  471. task := ctx.Cloudbrain
  472. for {
  473. if task.Status == string(models.GrampusStatusStopped) || task.Status == string(models.GrampusStatusFailed) || task.Status == string(models.GrampusStatusSucceeded) {
  474. log.Error("the job(%s) has been stopped", task.JobName, ctx.Data["msgID"])
  475. resultCode = "-1"
  476. errorMsg = "system error"
  477. break
  478. }
  479. res, err := grampus.StopJob(task.JobID)
  480. if err != nil {
  481. log.Error("StopJob(%s) failed:%v", task.JobName, err, ctx.Data["msgID"])
  482. resultCode = strconv.Itoa(res.ErrorCode)
  483. errorMsg = res.ErrorMsg
  484. break
  485. }
  486. oldStatus := task.Status
  487. task.Status = string(models.GrampusStatusStopped)
  488. if task.EndTime == 0 {
  489. task.EndTime = timeutil.TimeStampNow()
  490. }
  491. task.ComputeAndSetDuration()
  492. if oldStatus != task.Status {
  493. notification.NotifyChangeCloudbrainStatus(task, oldStatus)
  494. }
  495. err = models.UpdateJob(task)
  496. if err != nil {
  497. log.Error("UpdateJob(%s) failed:%v", task.JobName, err, ctx.Data["msgID"])
  498. resultCode = "-1"
  499. errorMsg = "system error"
  500. break
  501. }
  502. status = task.Status
  503. break
  504. }
  505. ctx.JSON(200, map[string]interface{}{
  506. "result_code": resultCode,
  507. "error_msg": errorMsg,
  508. "status": status,
  509. "id": ID,
  510. "StatusOK": 0,
  511. })
  512. }
  513. func GrampusTrainJobDel(ctx *context.Context) {
  514. var listType = ctx.Query("listType")
  515. if err := deleteGrampusJob(ctx); err != nil {
  516. log.Error("deleteGrampusJob failed: %v", err, ctx.Data["msgID"])
  517. ctx.ServerError(err.Error(), err)
  518. return
  519. }
  520. var isAdminPage = ctx.Query("isadminpage")
  521. var isHomePage = ctx.Query("ishomepage")
  522. if ctx.IsUserSiteAdmin() && isAdminPage == "true" {
  523. ctx.Redirect(setting.AppSubURL + "/admin" + "/cloudbrains")
  524. } else if isHomePage == "true" {
  525. ctx.Redirect(setting.AppSubURL + "/cloudbrains")
  526. } else {
  527. ctx.Redirect(setting.AppSubURL + ctx.Repo.RepoLink + "/modelarts/train-job?listType=" + listType)
  528. }
  529. }
  530. func deleteGrampusJob(ctx *context.Context) error {
  531. task := ctx.Cloudbrain
  532. if task.Status != string(models.GrampusStatusStopped) && task.Status != string(models.GrampusStatusSucceeded) && task.Status != string(models.GrampusStatusFailed) {
  533. log.Error("the job(%s) has not been stopped", task.JobName, ctx.Data["msgID"])
  534. return errors.New("the job has not been stopped")
  535. }
  536. err := models.DeleteJob(task)
  537. if err != nil {
  538. log.Error("DeleteJob failed: %v", err, ctx.Data["msgID"])
  539. return err
  540. }
  541. storageType := models.TypeCloudBrainOne
  542. if task.ComputeResource == models.NPUResource {
  543. storageType = models.TypeCloudBrainTwo
  544. }
  545. deleteJobStorage(task.JobName, storageType)
  546. return nil
  547. }
  548. func GrampusTrainJobShow(ctx *context.Context) {
  549. ctx.Data["PageIsCloudBrain"] = true
  550. var task *models.Cloudbrain
  551. task, err := models.GetCloudbrainByJobIDWithDeleted(ctx.Params(":jobid"))
  552. if err != nil {
  553. log.Error("GetCloudbrainByJobID failed:" + err.Error())
  554. ctx.NotFound(ctx.Req.URL.RequestURI(), nil)
  555. return
  556. }
  557. if task.DeletedAt.IsZero() { //normal record
  558. result, err := grampus.GetJob(task.JobID)
  559. if err != nil {
  560. log.Error("GetJob failed:" + err.Error())
  561. ctx.NotFound(ctx.Req.URL.RequestURI(), nil)
  562. return
  563. }
  564. if result != nil {
  565. if len(result.JobInfo.Tasks[0].CenterID) == 1 && len(result.JobInfo.Tasks[0].CenterName) == 1 {
  566. task.AiCenter = result.JobInfo.Tasks[0].CenterID[0] + "+" + result.JobInfo.Tasks[0].CenterName[0]
  567. }
  568. oldStatus := task.Status
  569. task.Status = grampus.TransTrainJobStatus(result.JobInfo.Status)
  570. if task.Status != result.JobInfo.Status || result.JobInfo.Status == models.GrampusStatusRunning {
  571. task.Duration = result.JobInfo.RunSec
  572. task.TrainJobDuration = models.ConvertDurationToStr(task.Duration)
  573. if task.StartTime == 0 && result.JobInfo.StartedAt > 0 {
  574. task.StartTime = timeutil.TimeStamp(result.JobInfo.StartedAt)
  575. }
  576. if task.EndTime == 0 && models.IsTrainJobTerminal(task.Status) && task.StartTime > 0 {
  577. task.EndTime = task.StartTime.Add(task.Duration)
  578. }
  579. task.CorrectCreateUnix()
  580. if oldStatus != task.Status {
  581. notification.NotifyChangeCloudbrainStatus(task, oldStatus)
  582. }
  583. err = models.UpdateJob(task)
  584. if err != nil {
  585. log.Error("UpdateJob failed:" + err.Error())
  586. }
  587. }
  588. }
  589. }
  590. if len(task.Parameters) > 0 {
  591. var parameters models.Parameters
  592. err := json.Unmarshal([]byte(task.Parameters), &parameters)
  593. if err != nil {
  594. log.Error("Failed to Unmarshal Parameters: %s (%v)", task.Parameters, err)
  595. ctx.ServerError("system error", err)
  596. return
  597. }
  598. if len(parameters.Parameter) > 0 {
  599. paramTemp := ""
  600. for _, Parameter := range parameters.Parameter {
  601. param := Parameter.Label + " = " + Parameter.Value + "; "
  602. paramTemp = paramTemp + param
  603. }
  604. task.Parameters = paramTemp[:len(paramTemp)-2]
  605. } else {
  606. task.Parameters = ""
  607. }
  608. }
  609. taskList := make([]*models.Cloudbrain, 0)
  610. taskList = append(taskList, task)
  611. ctx.Data["version_list_task"] = taskList
  612. ctx.Data["datasetDownload"] = GetCloudBrainDataSetInfo(task.Uuid, false)
  613. ctx.Data["canDownload"] = cloudbrain.CanModifyJob(ctx, task)
  614. ctx.Data["displayJobName"] = task.DisplayJobName
  615. aiCenterInfo := strings.Split(task.AiCenter, "+")
  616. if len(aiCenterInfo) == 2 {
  617. ctx.Data["ai_center"] = aiCenterInfo[1]
  618. }
  619. ctx.HTML(http.StatusOK, tplGrampusTrainJobShow)
  620. }
  621. func GrampusGetLog(ctx *context.Context) {
  622. jobID := ctx.Params(":jobid")
  623. job, err := models.GetCloudbrainByJobID(jobID)
  624. if err != nil {
  625. log.Error("GetCloudbrainByJobID failed: %v", err, ctx.Data["MsgID"])
  626. ctx.ServerError(err.Error(), err)
  627. return
  628. }
  629. content, err := grampus.GetTrainJobLog(job.JobID)
  630. if err != nil {
  631. log.Error("GetTrainJobLog failed: %v", err, ctx.Data["MsgID"])
  632. ctx.ServerError(err.Error(), err)
  633. return
  634. }
  635. ctx.JSON(http.StatusOK, map[string]interface{}{
  636. "JobName": job.JobName,
  637. "Content": content,
  638. })
  639. return
  640. }
  641. func generateCommand(repoName, processorType, codeRemotePath, dataRemotePath, bootFile, paramSrc, outputRemotePath, datasetName string) (string, error) {
  642. var command string
  643. workDir := grampus.NpuWorkDir
  644. if processorType == grampus.ProcessorTypeGPU {
  645. workDir = grampus.GpuWorkDir
  646. }
  647. command += "pwd;cd " + workDir + grampus.CommandPrepareScript
  648. //download code & dataset
  649. if processorType == grampus.ProcessorTypeNPU {
  650. commandDownload := "./downloader_for_obs " + setting.Bucket + " " + codeRemotePath + " " + grampus.CodeArchiveName + " " + dataRemotePath + " '" + datasetName + "';"
  651. command += commandDownload
  652. } else if processorType == grampus.ProcessorTypeGPU {
  653. commandDownload := "./downloader_for_minio " + setting.Grampus.Env + " " + codeRemotePath + " " + grampus.CodeArchiveName + " " + dataRemotePath + " '" + datasetName + "';"
  654. command += commandDownload
  655. }
  656. //check download result
  657. commandCheckRes := "bash -c \"[[ $? -eq 0 ]] && exit 0 || exit -1;\";"
  658. command += commandCheckRes
  659. //unzip code & dataset
  660. toolUnzip := "unzip -q '"
  661. if strings.HasSuffix(datasetName, ".tar.gz") {
  662. toolUnzip = "tar -zxvf '"
  663. }
  664. commandUnzip := "cd " + workDir + "code;unzip -q master.zip;echo \"start to unzip dataset\";cd " + workDir + "dataset;" + toolUnzip + datasetName + "';"
  665. command += commandUnzip
  666. //check unzip result
  667. commandCheckRes = "bash -c \"[[ $? -eq 0 ]] && exit 0 || exit -1;\";"
  668. command += commandCheckRes
  669. command += "echo \"unzip finished;start to exec code;\";"
  670. //exec code
  671. var parameters models.Parameters
  672. var paramCode string
  673. param := make([]models.Parameter, 0)
  674. if len(paramSrc) != 0 {
  675. err := json.Unmarshal([]byte(paramSrc), &parameters)
  676. if err != nil {
  677. log.Error("Failed to Unmarshal params: %s (%v)", paramSrc, err)
  678. return command, err
  679. }
  680. for _, parameter := range parameters.Parameter {
  681. param = append(param, models.Parameter{
  682. Label: parameter.Label,
  683. Value: parameter.Value,
  684. })
  685. paramCode += " --" + parameter.Label + "=" + parameter.Value
  686. }
  687. }
  688. commandCode := "cd " + workDir + "code/" + strings.ToLower(repoName) + ";python " + bootFile + paramCode + ";"
  689. command += commandCode
  690. //get exec result
  691. commandGetRes := "result=$?;"
  692. command += commandGetRes
  693. //upload models
  694. if processorType == grampus.ProcessorTypeNPU {
  695. commandUpload := "cd " + workDir + "script_for_grampus/;./uploader_for_obs " + setting.Bucket + " " + outputRemotePath + " " + workDir + "output/;"
  696. command += commandUpload
  697. } else if processorType == grampus.ProcessorTypeGPU {
  698. commandUpload := "cd " + workDir + "script_for_grampus/;./uploader_for_minio " + setting.Grampus.Env + " " + outputRemotePath + " " + workDir + "output/;"
  699. command += commandUpload
  700. }
  701. //check exec result
  702. commandCheckRes = "bash -c \"[[ $result -eq 0 ]] && exit 0 || exit -1\""
  703. command += commandCheckRes
  704. return command, nil
  705. }
  706. func downloadZipCode(ctx *context.Context, codePath, branchName string) error {
  707. archiveType := git.ZIP
  708. archivePath := codePath
  709. if !com.IsDir(archivePath) {
  710. if err := os.MkdirAll(archivePath, os.ModePerm); err != nil {
  711. log.Error("MkdirAll failed:" + err.Error())
  712. return err
  713. }
  714. }
  715. // Get corresponding commit.
  716. var (
  717. commit *git.Commit
  718. err error
  719. )
  720. gitRepo := ctx.Repo.GitRepo
  721. if err != nil {
  722. log.Error("OpenRepository failed:" + err.Error())
  723. return err
  724. }
  725. if gitRepo.IsBranchExist(branchName) {
  726. commit, err = gitRepo.GetBranchCommit(branchName)
  727. if err != nil {
  728. log.Error("GetBranchCommit failed:" + err.Error())
  729. return err
  730. }
  731. }
  732. archivePath = path.Join(archivePath, grampus.CodeArchiveName)
  733. if !com.IsFile(archivePath) {
  734. if err := commit.CreateArchive(archivePath, git.CreateArchiveOpts{
  735. Format: archiveType,
  736. Prefix: setting.Repository.PrefixArchiveFiles,
  737. }); err != nil {
  738. log.Error("CreateArchive failed:" + err.Error())
  739. return err
  740. }
  741. }
  742. return nil
  743. }