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