Manager: supporting Task.JobPriority

Tasks are sorted by job priority first, and then by task priority.

packages/flamenco-manager-go/src/flamenco-manager/flamenco/documents.go

@@ -12,21 +12,22 @@ type Command struct {
 }
 
 type Task struct {
-	Id       bson.ObjectId   `bson:"_id,omitempty" json:"_id,omitempty"`
-	Etag     string          `bson:"_etag,omitempty" json:"_etag,omitempty"`
-	Job      bson.ObjectId   `bson:"job,omitempty" json:"job"`
-	Manager  bson.ObjectId   `bson:"manager,omitempty" json:"manager"`
-	Project  bson.ObjectId   `bson:"project,omitempty" json:"project"`
-	User     bson.ObjectId   `bson:"user,omitempty" json:"user"`
-	Name     string          `bson:"name" json:"name"`
-	Status   string          `bson:"status" json:"status"`
-	Priority int             `bson:"priority" json:"priority"`
-	JobType  string          `bson:"job_type" json:"job_type"`
-	Commands []Command       `bson:"commands" json:"commands"`
-	Log      string          `bson:"log,omitempty" json:"log,omitempty"`
-	Activity string          `bson:"activity,omitempty" json:"activity,omitempty"`
-	Parents  []bson.ObjectId `bson:"parents,omitempty" json:"parents,omitempty"`
-	Worker   string          `bson:"worker,omitempty" json:"worker,omitempty"`
+	Id          bson.ObjectId   `bson:"_id,omitempty" json:"_id,omitempty"`
+	Etag        string          `bson:"_etag,omitempty" json:"_etag,omitempty"`
+	Job         bson.ObjectId   `bson:"job,omitempty" json:"job"`
+	Manager     bson.ObjectId   `bson:"manager,omitempty" json:"manager"`
+	Project     bson.ObjectId   `bson:"project,omitempty" json:"project"`
+	User        bson.ObjectId   `bson:"user,omitempty" json:"user"`
+	Name        string          `bson:"name" json:"name"`
+	Status      string          `bson:"status" json:"status"`
+	Priority    int             `bson:"priority" json:"priority"`
+	JobPriority int             `bson:"job_priority" json:"job_priority"`
+	JobType     string          `bson:"job_type" json:"job_type"`
+	Commands    []Command       `bson:"commands" json:"commands"`
+	Log         string          `bson:"log,omitempty" json:"log,omitempty"`
+	Activity    string          `bson:"activity,omitempty" json:"activity,omitempty"`
+	Parents     []bson.ObjectId `bson:"parents,omitempty" json:"parents,omitempty"`
+	Worker      string          `bson:"worker,omitempty" json:"worker,omitempty"`
 
 	// Internal bookkeeping
 	WorkerId       *bson.ObjectId `bson:"worker_id,omitempty" json:"-"`

packages/flamenco-manager-go/src/flamenco-manager/flamenco/scheduler.go

@@ -168,6 +168,7 @@ func (ts *TaskScheduler) fetchTaskFromQueueOrManager(
 		M{"$project": M{"task": 1}},
 		// 9: Sort by priority, with highest prio first. If prio is equal, use newest task.
 		M{"$sort": bson.D{
+			{"task.job_priority", -1},
 			{"task.priority", -1},
 			{"task._id", 1},
 		}},

packages/flamenco-manager-go/src/flamenco-manager/flamenco/scheduler_test.go

@@ -141,6 +141,31 @@ func (s *SchedulerTestSuite) TestSchedulerOrderByPriority(t *check.C) {
 	assert.Equal(t, task2.Id.Hex(), json_task.Id.Hex())
 }
 
+func (s *SchedulerTestSuite) TestSchedulerOrderByJobPriority(t *check.C) {
+	// Store task in DB.
+	task1 := ConstructTestTaskWithPrio("1aaaaaaaaaaaaaaaaaaaaaaa", "sleeping", 50)
+	task1.JobPriority = 10
+	if err := s.db.C("flamenco_tasks").Insert(task1); err != nil {
+		t.Fatal("Unable to insert test task1", err)
+	}
+	task2 := ConstructTestTaskWithPrio("2aaaaaaaaaaaaaaaaaaaaaaa", "sleeping", 100)
+	task2.JobPriority = 5
+	if err := s.db.C("flamenco_tasks").Insert(task2); err != nil {
+		t.Fatal("Unable to insert test task 2", err)
+	}
+
+	// Perform HTTP request to the scheduler.
+	resp_rec := httptest.NewRecorder()
+	request, _ := http.NewRequest("GET", "/task", nil)
+	ar := &auth.AuthenticatedRequest{Request: *request, Username: s.worker_lnx.Id.Hex()}
+	s.sched.ScheduleTask(resp_rec, ar)
+
+	// We should have gotten task 1, because its job has the highest priority.
+	json_task := Task{}
+	parseJson(t, resp_rec, 200, &json_task)
+	assert.Equal(t, task1.Id.Hex(), json_task.Id.Hex())
+}
+
 /**
  * The failure case, where the TaskScheduler cannot reach the Server to check
  * the task for updates, is already implicitly handled in the TestVariableReplacement