Merge pull request #361 from teharrison/master

io validation update
MG-RAST · May 4, 2015 · 2af852c · 2af852c
2 parents 3e53580 + 2245418
commit 2af852c
Show file tree

Hide file tree

Showing 4 changed files with 18 additions and 15 deletions.
diff --git a/lib/core/core.go b/lib/core/core.go
@@ -281,7 +281,11 @@ func ParseJobTasks(filename string, jid string) (job *Job, err error) {
 	}
 
 	for i := 0; i < len(job.Tasks); i++ {
-		if err := job.Tasks[i].InitTask(job, i); err != nil {
+		if strings.Contains(job.Tasks[i].Id, "_") {
+			// no "_" allowed in inital taskid
+			return nil, errors.New("invalid taskid, may not contain '_'")
+		}
+		if err := job.Tasks[i].InitTask(job); err != nil {
 			return nil, errors.New("error in InitTask: " + err.Error())
 		}
 	}
@@ -370,7 +374,7 @@ func AwfToJob(awf *Workflow, jid string) (job *Job, err error) {
 			parent_id := getParentTask(task.Id, parent)
 			task.DependsOn = append(task.DependsOn, parent_id)
 		}
-		task.InitTask(job, awf_task.TaskId)
+		task.InitTask(job)
 		job.Tasks = append(job.Tasks, task)
 	}
 	job.RemainTasks = len(job.Tasks) - 1

diff --git a/lib/core/db.go b/lib/core/db.go
@@ -19,6 +19,8 @@ func InitJobDB() {
 	cj.EnsureIndex(mgo.Index{Key: []string{"acl.delete"}, Background: true})
 	cj.EnsureIndex(mgo.Index{Key: []string{"id"}, Unique: true})
 	cj.EnsureIndex(mgo.Index{Key: []string{"info.submittime"}, Background: true})
+	cj.EnsureIndex(mgo.Index{Key: []string{"info.pipeline"}, Background: true})
+	cj.EnsureIndex(mgo.Index{Key: []string{"info.user"}, Background: true})
 	cj.EnsureIndex(mgo.Index{Key: []string{"jid"}, Background: true})
 	cj.EnsureIndex(mgo.Index{Key: []string{"state"}, Background: true})
 	cj.EnsureIndex(mgo.Index{Key: []string{"updatetime"}, Background: true})

diff --git a/lib/core/io.go b/lib/core/io.go
@@ -98,7 +98,8 @@ func (io *IO) DataUrl() (dataurl string, err error) {
 		io.Url = fmt.Sprintf("%s/node/%s?download", io.Host, io.Node)
 		return io.Url, nil
 	} else {
-		return "", errors.New("Invalid IO, required fields url or host / node missing")
+	    // empty IO is valid
+		return "", nil
 	}
 }
 

diff --git a/lib/core/task.go b/lib/core/task.go
@@ -6,7 +6,6 @@ import (
 	"github.com/MG-RAST/AWE/lib/conf"
 	"github.com/MG-RAST/AWE/lib/logger"
 	"os/exec"
-	"strconv"
 	"strings"
 	"time"
 )
@@ -65,27 +64,20 @@ func NewTask(job *Job, rank int) *Task {
 }
 
 // fill some info (lacked in input json) for a task
-func (task *Task) InitTask(job *Job, rank int) (err error) {
+func (task *Task) InitTask(job *Job) (err error) {
 	//validate taskid
 	if len(task.Id) == 0 {
 		return errors.New("invalid taskid:" + task.Id)
 	}
+
 	parts := strings.Split(task.Id, "_")
-	if len(parts) == 2 {
-		//is standard taskid (%s_%d), do nothing
-	} else if idInt, err := strconv.Atoi(task.Id); err == nil {
-		//if task.Id is an "integer", it is unmashalled from job.json (submitted by template)
-		//convert to standard taskid
-		if rank != idInt {
-			return errors.New(fmt.Sprintf("invalid job script: task id doesn't match stage %d vs %d", rank, idInt))
-		}
+	if len(parts) == 1 {
+		// is not standard taskid, convert it
 		task.Id = fmt.Sprintf("%s_%s", job.Id, task.Id)
 		for j := 0; j < len(task.DependsOn); j++ {
 			depend := task.DependsOn[j]
 			task.DependsOn[j] = fmt.Sprintf("%s_%s", job.Id, depend)
 		}
-	} else {
-		return errors.New("invalid taskid:" + task.Id)
 	}
 
 	task.Info = job.Info
@@ -125,6 +117,10 @@ func (task *Task) InitTask(job *Job, rank int) (err error) {
 		if _, err = io.DataUrl(); err != nil {
 			return err
 		}
+		// predata IO can not be empty
+		if (io.Url == "") && (io.Node == "-") {
+		    return errors.New("Invalid IO, required fields url or host / node missing")
+		}
 		logger.Debug(2, "inittask predata: host="+io.Host+", node="+io.Node+", url="+io.Url)
 	}