Merge pull request #447 from MG-RAST/develop

teharrison · web-flow · commit 5f357d9818f4 · 2016-07-14T10:03:12.000-05:00
version 0.9.27
diff --git a/RELEASE_NOTES.txt b/RELEASE_NOTES.txt
@@ -1,3 +1,10 @@
+# v0.9.27
+
+- improved log / error messages
+- added 'report' option to job GET, returns a summary of job with stdout, stderr, and worknotes for each workunit ran
+- changed max jid to be stored in mongodb and not on filesystem
+  - this is backwards compatible, will use jid on filesystem for first time if no value is in mongodb
+
 # v0.9.26
 
 - added 'full' option to job DELETE, actually deletes job from mongoDB and filesystem
diff --git a/VERSION b/VERSION
@@ -1 +1 @@
-0.9.26
+0.9.27
diff --git a/awe-server/awe-server.go b/awe-server/awe-server.go
@@ -239,10 +239,7 @@ func main() {
 		fmt.Println("Done")
 	}
 
-	if conf.DEBUG_LEVEL > 0 {
-		fmt.Println("init max job number (jid)...")
-	}
-	//init max job number (jid)
+	// init max job number (jid), backwards compatible with jobid file
 	if err := core.QMgr.InitMaxJid(); err != nil {
 		fmt.Fprintf(os.Stderr, "ERROR from InitMaxJid : %v\n", err)
 		os.Exit(1)
diff --git a/lib/conf/conf.go b/lib/conf/conf.go
@@ -19,9 +19,11 @@ var GIT_COMMIT_HASH string // use -ldflags "-X github.com/MG-RAST/AWE/lib/conf.G
 const BasePriority int = 1
 
 const DB_COLL_JOBS string = "Jobs"
+const DB_COLL_JID string = "MaxID"
 const DB_COLL_PERF string = "Perf"
 const DB_COLL_CGS string = "ClientGroups"
 const DB_COLL_USERS string = "Users"
+const JOB_ID_START int = 10000
 
 //default index type used for intermediate data
 const DEFAULT_INDEX string = "chunkrecord"
diff --git a/lib/controller/jobController.go b/lib/controller/jobController.go
@@ -205,6 +205,16 @@ func (cr *JobController) Read(id string, cx *goweb.Context) {
 		return
 	}
 
+	if query.Has("report") {
+		jobLogs, err := job.GetJobLogs()
+		if err != nil {
+			logger.Error("Err@GetJobLogs: " + id + ":" + err.Error())
+			cx.RespondWithErrorMessage("job logs not found: "+id, http.StatusBadRequest)
+		}
+		cx.RespondWithData(jobLogs)
+		return
+	}
+
 	if core.QMgr.IsJobRegistered(id) {
 		job.Registered = true
 	} else {
@@ -677,7 +687,7 @@ func (cr *JobController) Update(id string, cx *goweb.Context) {
 			cx.RespondWithErrorMessage("fail to recompute job: "+id+" "+err.Error(), http.StatusBadRequest)
 			return
 		}
-		cx.RespondWithData("job recompute started: " + id)
+		cx.RespondWithData("job recompute started at task " + stage + ": " + id)
 		return
 	}
 	if query.Has("resubmit") { // to recompute a job from the beginning, all tasks will be computed
diff --git a/lib/core/core.go b/lib/core/core.go
@@ -652,7 +652,6 @@ func getParentTask(taskid string, origin int) string {
 	return taskid
 }
 
-//
 func contains(list []string, elem string) bool {
 	for _, t := range list {
 		if t == elem {
@@ -662,14 +661,6 @@ func contains(list []string, elem string) bool {
 	return false
 }
 
-func jidIncr(jid string) (newjid string) {
-	if jidint, err := strconv.Atoi(jid); err == nil {
-		jidint += 1
-		return strconv.Itoa(jidint)
-	}
-	return jid
-}
-
 //functions for REST API communication  (=deprecated=)
 //notify AWE server a workunit is finished with status either "failed" or "done", and with perf statistics if "done"
 func NotifyWorkunitProcessed(work *Workunit, perf *WorkPerf) (err error) {
diff --git a/lib/core/db.go b/lib/core/db.go
@@ -53,6 +53,21 @@ func InitClientGroupDB() {
 	cc.EnsureIndex(mgo.Index{Key: []string{"token"}, Unique: true})
 }
 
+// if max job id does not exist, set to start value
+func initMaxJidDB(startjid int) (err error) {
+	session := db.Connection.Session.Copy()
+	defer session.Close()
+	cc := session.DB(conf.MONGODB_DATABASE).C(conf.DB_COLL_JID)
+	cc.EnsureIndex(mgo.Index{Key: []string{"name"}, Unique: true})
+	if _, jerr := dbFindMaxJobID(); jerr != nil {
+		initjid := &JobID{"jid", startjid}
+		if err := dbUpsert(initjid); err != nil {
+			return err
+		}
+	}
+	return
+}
+
 func dbDelete(q bson.M, coll string) (err error) {
 	session := db.Connection.Session.Copy()
 	defer session.Close()
@@ -80,6 +95,9 @@ func dbUpsert(t interface{}) (err error) {
 	case *ClientGroup:
 		c := session.DB(conf.MONGODB_DATABASE).C(conf.DB_COLL_CGS)
 		_, err = c.Upsert(bson.M{"id": t.Id}, &t)
+	case *JobID:
+		c := session.DB(conf.MONGODB_DATABASE).C(conf.DB_COLL_JID)
+		_, err = c.Upsert(bson.M{"name": t.Name}, &t)
 	default:
 		fmt.Printf("invalid database entry type\n")
 	}
@@ -181,6 +199,16 @@ func dbFindSortClientGroups(q bson.M, results *ClientGroups, options map[string]
 	return
 }
 
+func dbFindMaxJobID() (jid *JobID, err error) {
+	session := db.Connection.Session.Copy()
+	defer session.Close()
+	c := session.DB(conf.MONGODB_DATABASE).C(conf.DB_COLL_JID)
+	if err = c.Find(bson.M{}).One(&jid); err == nil {
+		return jid, nil
+	}
+	return nil, err
+}
+
 func LoadJob(id string) (job *Job, err error) {
 	job = new(Job)
 	session := db.Connection.Session.Copy()
diff --git a/lib/core/io.go b/lib/core/io.go
@@ -32,7 +32,7 @@ type IO struct {
 	ShockIndex    string                 `bson:"shockindex" json:"shockindex"` // on input it indicates that Shock node has to be indexed by AWE server
 	AttrFile      string                 `bson:"attrfile" json:"attrfile"`
 	NoFile        bool                   `bson:"nofile" json:"nofile"`
-	Delete        bool                   `bson:"delete" json:"delete"`
+	Delete        bool                   `bson:"delete" json:"delete"` // speficies that this is a temorary node, to be deleted from shock on job completion
 	Type          string                 `bson:"type" json:"type"`
 	NodeAttr      map[string]interface{} `bson:"nodeattr" json:"nodeattr"` // specifies attribute data to be stored in shock node (output only)
 	FormOptions   map[string]string      `bson:"formoptions" json:"formoptions"`
diff --git a/lib/core/job.go b/lib/core/job.go
@@ -78,6 +78,21 @@ type JobMin struct {
 	UserAttr      map[string]string `bson:"userattr" json:"userattr"`
 }
 
+type JobLog struct {
+	Id         string     `bson:"id" json:"id"`
+	State      string     `bson:"state" json:"state"`
+	UpdateTime time.Time  `bson:"updatetime" json:"updatetime"`
+	Notes      string     `bson:"notes" json:"notes"`
+	LastFailed string     `bson:"lastfailed" json:"lastfailed"`
+	Resumed    int        `bson:"resumed" json:"resumed"`
+	Tasks      []*TaskLog `bson:"tasks" json:"tasks"`
+}
+
+type JobID struct {
+	Name string `bson:"name" json:"name"`
+	Max  int    `bson:"max" json:"max"`
+}
+
 //set job's uuid
 func (job *Job) setId() {
 	job.Id = uuid.New()
@@ -302,6 +317,20 @@ func (job *Job) GetPrivateEnv(taskid string) (env map[string]string) {
 	return
 }
 
+func (job *Job) GetJobLogs() (jlog *JobLog, err error) {
+	jlog = new(JobLog)
+	jlog.Id = job.Id
+	jlog.State = job.State
+	jlog.UpdateTime = job.UpdateTime
+	jlog.Notes = job.Notes
+	jlog.LastFailed = job.LastFailed
+	jlog.Resumed = job.Resumed
+	for _, task := range job.Tasks {
+		jlog.Tasks = append(jlog.Tasks, task.GetTaskLogs())
+	}
+	return
+}
+
 func ReloadFromDisk(path string) (err error) {
 	id := filepath.Base(path)
 	jobbson, err := ioutil.ReadFile(path + "/" + id + ".bson")
diff --git a/lib/core/servermgr.go b/lib/core/servermgr.go
@@ -33,11 +33,11 @@ type ServerMgr struct {
 	taskMap   map[string]*Task
 	actJobs   map[string]*JobPerf
 	susJobs   map[string]bool
-	jsReq     chan bool   //channel for job submission request (JobController -> qmgr.Handler)
-	jsAck     chan string //channel for return an assigned job number in response to jsReq  (qmgr.Handler -> JobController)
-	taskIn    chan *Task  //channel for receiving Task (JobController -> qmgr.Handler)
-	coSem     chan int    //semaphore for checkout (mutual exclusion between different clients)
-	nextJid   string      //next jid that will be assigned to newly submitted job
+	jsReq     chan bool  //channel for job submission request (JobController -> qmgr.Handler)
+	jsAck     chan int   //channel for return an assigned job number in response to jsReq  (qmgr.Handler -> JobController)
+	taskIn    chan *Task //channel for receiving Task (JobController -> qmgr.Handler)
+	coSem     chan int   //semaphore for checkout (mutual exclusion between different clients)
+	nextJid   int        //next jid that will be assigned to newly submitted job
 }
 
 func NewServerMgr() *ServerMgr {
@@ -53,11 +53,11 @@ func NewServerMgr() *ServerMgr {
 		},
 		taskMap: map[string]*Task{},
 		jsReq:   make(chan bool),
-		jsAck:   make(chan string),
+		jsAck:   make(chan int),
 		taskIn:  make(chan *Task, 1024),
 		actJobs: map[string]*JobPerf{},
 		susJobs: map[string]bool{},
-		nextJid: "",
+		nextJid: 0,
 	}
 }
 
@@ -68,7 +68,7 @@ func (qm *ServerMgr) JidHandle() {
 		<-qm.jsReq
 		jid := qm.getNextJid()
 		qm.jsAck <- jid
-		logger.Debug(2, fmt.Sprintf("qmgr:receive a job submission request, assigned jid=%s", jid))
+		logger.Debug(2, fmt.Sprintf("qmgr:receive a job submission request, assigned jid=%d", jid))
 	}
 }
 
@@ -336,47 +336,22 @@ func (qm *ServerMgr) listTasks() (ids []string) {
 //--------server methods-------
 
 func (qm *ServerMgr) InitMaxJid() (err error) {
+	// this is for backwards compatibility with jid on filysystem
+	startjid := conf.JOB_ID_START
 	jidfile := conf.DATA_PATH + "/maxjid"
-
-	if _, err := os.Stat(jidfile); err != nil {
-
-		f, err := os.Create(jidfile)
-		if err != nil {
-			fmt.Fprintf(os.Stderr, fmt.Sprintf("error creating jidfile ", err.Error())) // logger does not work
-			logger.Error(fmt.Sprintf("error creating jidfile ", err.Error()))
-			return err
-		}
-		f.WriteString("10000")
-		qm.nextJid = "10001"
-		f.Close()
-	} else {
-
-		buf, err := ioutil.ReadFile(jidfile)
-		if err != nil {
-			if conf.DEBUG_LEVEL > 0 {
-				fmt.Println("error ioutil.ReadFile(jidfile)")
-			}
-			return err
-		}
+	if buf, ferr := ioutil.ReadFile(jidfile); ferr == nil {
 		bufstr := strings.TrimSpace(string(buf))
-
-		maxjid, err := strconv.Atoi(bufstr)
-		if err != nil {
-			if conf.DEBUG_LEVEL > 0 {
-				fmt.Println(fmt.Sprintf("error strconv.Atoi(bufstr), bufstr=\"%s\"", bufstr))
-			}
-			fmt.Fprintf(os.Stderr, fmt.Sprintf("Could not convert \"%s\" into int", bufstr)) // logger does not work
-			logger.Error(fmt.Sprintf("Could not convert \"%s\" into int", bufstr))
-			return err
+		if jid, serr := strconv.Atoi(bufstr); serr == nil {
+			startjid = jid
 		}
-
-		qm.nextJid = strconv.Itoa(maxjid + 1)
-
 	}
-	if conf.DEBUG_LEVEL > 0 {
-		fmt.Println("in InitMaxJid C")
+	// set in mongoDB
+	if err := initMaxJidDB(startjid); err != nil {
+		return err
 	}
-	logger.Debug(2, fmt.Sprintf("qmgr:jid initialized, next jid=%s", qm.nextJid))
+	qm.nextJid = startjid + 1
+
+	logger.Debug(2, fmt.Sprintf("qmgr:jid initialized, next jid=%d", qm.nextJid))
 	return
 }
 
@@ -1012,19 +987,18 @@ func (qm *ServerMgr) ShowTasks() {
 //---job methods---
 func (qm *ServerMgr) JobRegister() (jid string, err error) {
 	qm.jsReq <- true
-	jid = <-qm.jsAck
-
-	if jid == "" {
+	id := <-qm.jsAck
+	if id == 0 {
 		return "", errors.New("failed to assign a job number for the newly submitted job")
 	}
-	return jid, nil
+	return strconv.Itoa(id), nil
 }
 
-func (qm *ServerMgr) getNextJid() (jid string) {
+func (qm *ServerMgr) getNextJid() (jid int) {
 	jid = qm.nextJid
-	jidfile := conf.DATA_PATH + "/maxjid"
-	ioutil.WriteFile(jidfile, []byte(jid), 0644)
-	qm.nextJid = jidIncr(qm.nextJid)
+	nextjid := &JobID{"jid", jid}
+	dbUpsert(nextjid)
+	qm.nextJid += 1
 	return jid
 }
 
diff --git a/lib/core/task.go b/lib/core/task.go
@@ -75,6 +75,14 @@ type TaskDep struct {
 	ClientGroups      string            `bson:"clientgroups" json:"clientgroups"`
 }
 
+type TaskLog struct {
+	Id            string     `bson:"taskid" json:"taskid"`
+	State         string     `bson:"state" json:"state"`
+	TotalWork     int        `bson:"totalwork" json:"totalwork"`
+	CompletedDate time.Time  `bson:"completedDate" json:"completeddate"`
+	Workunits     []*WorkLog `bson:"workunits" json:"workunits"`
+}
+
 func NewTask(job *Job, rank int) *Task {
 	return &Task{
 		Id:         fmt.Sprintf("%s_%d", job.Id, rank),
@@ -173,14 +181,14 @@ func (task *Task) CreateIndex() (err error) {
 		if len(io.ShockIndex) > 0 {
 			idxinfo, err := io.GetIndexInfo()
 			if err != nil {
-				errMsg := "could not retrieve index info from input shock node, taskid=" + task.Id
+				errMsg := "could not retrieve index info from input shock node, taskid=" + task.Id + ", error=" + err.Error()
 				logger.Error("error: " + errMsg)
 				return errors.New(errMsg)
 			}
 
 			if _, ok := idxinfo[io.ShockIndex]; !ok {
 				if err := ShockPutIndex(io.Host, io.Node, io.ShockIndex, task.Info.DataToken); err != nil {
-					errMsg := "failed to create index on shock node for taskid=" + task.Id
+					errMsg := "failed to create index on shock node for taskid=" + task.Id + ", error=" + err.Error()
 					logger.Error("error: " + errMsg)
 					return errors.New(errMsg)
 				}
@@ -238,21 +246,21 @@ func (task *Task) InitPartIndex() (err error) {
 	idxinfo, err := input_io.GetIndexInfo()
 	if err != nil {
 		task.setTotalWork(1)
-		logger.Error("warning: invalid file info, taskid=" + task.Id)
+		logger.Error("warning: invalid file info, taskid=" + task.Id + ", error=" + err.Error())
 		return nil
 	}
 
 	idxtype := conf.DEFAULT_INDEX
 	if _, ok := idxinfo[idxtype]; !ok { //if index not available, create index
 		if err := ShockPutIndex(input_io.Host, input_io.Node, idxtype, task.Info.DataToken); err != nil {
 			task.setTotalWork(1)
-			logger.Error("warning: fail to create index on shock for taskid=" + task.Id)
+			logger.Error("warning: fail to create index on shock for taskid=" + task.Id + ", error=" + err.Error())
 			return nil
 		}
 		totalunits, err = input_io.TotalUnits(idxtype) //get index info again
 		if err != nil {
 			task.setTotalWork(1)
-			logger.Error("warning: fail to get index units, taskid=" + task.Id + ":" + err.Error())
+			logger.Error("warning: fail to get index units, taskid=" + task.Id + ", error=" + err.Error())
 			return nil
 		}
 	} else { //index existing, use it directly
@@ -316,6 +324,22 @@ func (task *Task) ParseWorkunit() (wus []*Workunit, err error) {
 	return
 }
 
+func (task *Task) GetTaskLogs() (tlog *TaskLog) {
+	tlog = new(TaskLog)
+	tlog.Id = task.Id
+	tlog.State = task.State
+	tlog.TotalWork = task.TotalWork
+	tlog.CompletedDate = task.CompletedDate
+	if task.TotalWork == 1 {
+		tlog.Workunits = append(tlog.Workunits, NewWorkLog(task.Id, 0))
+	} else {
+		for i := 1; i <= task.TotalWork; i++ {
+			tlog.Workunits = append(tlog.Workunits, NewWorkLog(task.Id, i))
+		}
+	}
+	return
+}
+
 func (task *Task) Skippable() bool {
 	// For a task to be skippable, it should meet
 	// the following requirements (this may change
diff --git a/lib/core/workunit.go b/lib/core/workunit.go
