feat(sdk): order queue by project instance (#2983)

* feat(sdk): order queue by project instance

* fix(api): order workflow queue

Author: fsamin

engine/api/workflow/dao_node_run_job.go

@@ -62,7 +62,8 @@ func LoadNodeJobRunQueue(db gorp.SqlExecutor, store cache.Store, rights int, gro
 	from workflow_node_run_job
 	where workflow_node_run_job.queued >= $1
 	and workflow_node_run_job.queued <= $2
-	and workflow_node_run_job.status = ANY(string_to_array($3, ','))`
+	and workflow_node_run_job.status = ANY(string_to_array($3, ','))
+	order by workflow_node_run_job.queued ASC`
 
 	args := []interface{}{*since, *until, strings.Join(statuses, ",")}
 
@@ -84,6 +85,7 @@ func LoadNodeJobRunQueue(db gorp.SqlExecutor, store cache.Store, rights int, gro
 		AND workflow_node_run_job.queued >= $1
 		AND workflow_node_run_job.queued <= $2
 		AND workflow_node_run_job.status = ANY(string_to_array($3, ','))
+		ORDER BY workflow_node_run_job.queued ASC
 		`
 
 		var groupID string

engine/api/workflow/execute_node_run.go

@@ -361,6 +361,7 @@ func addJobsToQueue(ctx context.Context, db gorp.SqlExecutor, stage *sdk.Stage,
 
 		//Create the job run
 		wjob := sdk.WorkflowNodeJobRun{
+			ProjectID:              wr.ProjectID,
 			WorkflowNodeRunID:      run.ID,
 			Start:                  time.Time{},
 			Queued:                 time.Now(),

engine/sql/110_workflow_node_run_job.sql

@@ -0,0 +1,6 @@
+-- +migrate Up
+ALTER TABLE workflow_node_run_job ADD COLUMN project_id BIGINT DEFAULT 0;
+UPDATE workflow_node_run_job SET project_id = 0;
+
+-- +migrate Down
+ALTER TABLE workflow_node_run_job DROP COLUMN project_id;

sdk/cdsclient/client_queue.go

@@ -45,10 +45,11 @@ func (c *client) QueuePolling(ctx context.Context, jobs chan<- sdk.WorkflowNodeJ
 			}
 
 			if jobs != nil {
-				queue := []sdk.WorkflowNodeJobRun{}
+				queue := sdk.WorkflowQueue{}
 				if _, err := c.GetJSON("/queue/workflows", &queue); err != nil {
 					errs <- sdk.WrapError(err, "Unable to load old jobs")
 				}
+				queue.Sort()
 				for _, j := range queue {
 					jobs <- j
 				}

sdk/workflow_run.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"html/template"
 	"net/url"
+	"sort"
 	"strings"
 	"time"
 
@@ -199,6 +200,7 @@ func (w WorkflowNodeRunArtifact) Equal(c WorkflowNodeRunArtifact) bool {
 //WorkflowNodeJobRun represents an job to be run
 //easyjson:json
 type WorkflowNodeJobRun struct {
+	ProjectID              int64              `json:"project_id" db:"project_id"`
 	ID                     int64              `json:"id" db:"id"`
 	WorkflowNodeRunID      int64              `json:"workflow_node_run_id,omitempty" db:"workflow_node_run_id"`
 	Job                    ExecutedJob        `json:"job" db:"-"`
@@ -323,3 +325,22 @@ func (nr WorkflowNodeRun) Report() (string, error) {
 	errE := t.Execute(out, nr)
 	return out.String(), errE
 }
+
+type WorkflowQueue []WorkflowNodeJobRun
+
+func (q WorkflowQueue) Sort() {
+	//Count the number of WorkflowNodeJobRun per project_id
+	n := make(map[int64]int, len(q))
+	for _, j := range q {
+		nb := n[j.ProjectID]
+		nb++
+		n[j.ProjectID] = nb
+	}
+
+	sort.Slice(q, func(i, j int) bool {
+		p1 := n[q[i].ProjectID]
+		p2 := n[q[j].ProjectID]
+		return p1 < p2
+	})
+
+}

sdk/workflow_run_easyjson.go

@@ -105,3 +105,49 @@ func TestWorkflowRunReport(t *testing.T) {
 	assert.NoError(t, err)
 	t.Log(s)
 }
+
+func TestWorkflowQueue_Sort(t *testing.T) {
+	tests := []struct {
+		name     string
+		q        WorkflowQueue
+		expected WorkflowQueue
+	}{
+		{
+			name: "test sort 1",
+			q: WorkflowQueue{
+				{
+					ProjectID: 1,
+					ID:        1,
+				},
+				{
+					ProjectID: 1,
+					ID:        2,
+				},
+				{
+					ProjectID: 2,
+					ID:        3,
+				},
+			},
+			expected: WorkflowQueue{
+				{
+					ProjectID: 2,
+					ID:        3,
+				},
+				{
+					ProjectID: 1,
+					ID:        1,
+				},
+				{
+					ProjectID: 1,
+					ID:        2,
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tt.q.Sort()
+			assert.Equal(t, tt.expected, tt.q)
+		})
+	}
+}