feat(ui,api,cdsctl): Add filter to queue/workflows on status (#3293)

cli/cdsctl/monitoring.go

@@ -57,6 +57,7 @@ type Termui struct {
 	statusHatcheriesWorkers *cli.ScrollableList
 	status                  *cli.ScrollableList
 	currentURL              string
+	queueTabSelected        int
 }
 
 // Constants for each view of cds ui
@@ -88,6 +89,12 @@ func (ui *Termui) init() {
 	termui.Handle("/sys/kbd/<up>", func(e termui.Event) {
 		ui.monitoringCursorUp()
 	})
+	termui.Handle("/sys/kbd/<left>", func(e termui.Event) {
+		ui.monitoringCursorLeft()
+	})
+	termui.Handle("/sys/kbd/<right>", func(e termui.Event) {
+		ui.monitoringCursorRight()
+	})
 
 	termui.Handle("/sys/kbd/<enter>", func(e termui.Event) {
 		if ui.currentURL != "" {
@@ -240,6 +247,28 @@ func (ui *Termui) monitoringCursorUp() {
 	}
 }
 
+func (ui *Termui) monitoringCursorLeft() {
+	switch ui.selected {
+	case QueueSelected:
+		if 0 < ui.queueTabSelected {
+			ui.queueTabSelected--
+		} else {
+			ui.queueTabSelected = 2
+		}
+	}
+}
+
+func (ui *Termui) monitoringCursorRight() {
+	switch ui.selected {
+	case QueueSelected:
+		if ui.queueTabSelected < 2 {
+			ui.queueTabSelected++
+		} else {
+			ui.queueTabSelected = 0
+		}
+	}
+}
+
 func (ui *Termui) monitoringColorSelected() {
 	ui.queue.BorderFg = termui.ColorDefault
 	ui.statusHatcheriesWorkers.BorderFg = termui.ColorDefault
@@ -375,8 +404,18 @@ func (ui *Termui) computeStatusWorkerModels(workers []sdk.Worker) string {
 }
 
 func (ui *Termui) updateQueue(baseURL string) string {
+	var status []sdk.Status
+	switch ui.queueTabSelected {
+	case 0:
+		status = []sdk.Status{sdk.StatusWaiting}
+	case 1:
+		status = []sdk.Status{sdk.StatusBuilding}
+	case 2:
+		status = []sdk.Status{sdk.StatusWaiting, sdk.StatusBuilding}
+	}
+
 	start := time.Now()
-	wJobs, errw := client.QueueWorkflowNodeJobRun()
+	wJobs, errw := client.QueueWorkflowNodeJobRun(status...)
 	if errw != nil {
 		ui.msg = fmt.Sprintf("[%s](bg-red)", errw.Error())
 		return ""
@@ -394,13 +433,13 @@ func (ui *Termui) updateQueue(baseURL string) string {
 	var maxQueued time.Duration
 
 	items := []string{
-		fmt.Sprintf("[  %s %s%s %s ➤ %s ➤ %s ➤ %s](fg-cyan,bg-default)", pad("since", 9), pad("booked", 27), pad("run", 7), pad("project/workflow", 30), pad("node", 20), pad("triggered by", 17), "requirements"),
+		fmt.Sprintf("[  _ %s %s%s %s ➤ %s ➤ %s ➤ %s](fg-cyan,bg-default)", pad("since", 9), pad("by", 27), pad("run", 7), pad("project/workflow", 30), pad("node", 20), pad("triggered by", 17), "requirements"),
 	}
 
 	var idx int
 	var item string
 	for _, job := range pbJobs {
-		item, maxQueued = ui.updateQueueJob(idx, maxQueued, job.ID, false, job.Parameters, job.Job.Action.Requirements, job.Queued, job.BookedBy, baseURL)
+		item, maxQueued = ui.updateQueueJob(idx, maxQueued, job.ID, false, job.Parameters, job.Job, job.Queued, job.BookedBy, baseURL, job.Status)
 		items = append(items, item)
 		idx++
 	}
@@ -414,20 +453,20 @@ func (ui *Termui) updateQueue(baseURL string) string {
 	msg = fmt.Sprintf("[count queue wf %s](fg-cyan,bg-default) | %s", sdk.Round(elapsed, time.Millisecond).String(), msg)
 
 	for _, job := range wJobs {
-		item, maxQueued = ui.updateQueueJob(idx, maxQueued, job.ID, true, job.Parameters, job.Job.Action.Requirements, job.Queued, job.BookedBy, baseURL)
+		item, maxQueued = ui.updateQueueJob(idx, maxQueued, job.ID, true, job.Parameters, job.Job, job.Queued, job.BookedBy, baseURL, job.Status)
 		items = append(items, item)
 		idx++
 	}
 	ui.queue.Items = items
 
-	t := fmt.Sprintf("Queue:%d - Max Waiting:%s ", nWJobs.Count+int64(len(pbJobs)), sdk.Round(maxQueued, time.Second).String())
+	t := fmt.Sprintf("Queue(%s):%d - Max Waiting:%s ", fmt.Sprintf("%v", status), nWJobs.Count+int64(len(pbJobs)), sdk.Round(maxQueued, time.Second).String())
 	ui.queue.BorderLabel = t
 	return msg
 }
 
-func (ui *Termui) updateQueueJob(idx int, maxQueued time.Duration, id int64, isWJob bool, parameters []sdk.Parameter, requirements []sdk.Requirement, queued time.Time, bookedBy sdk.Hatchery, baseURL string) (string, time.Duration) {
+func (ui *Termui) updateQueueJob(idx int, maxQueued time.Duration, id int64, isWJob bool, parameters []sdk.Parameter, executedJob sdk.ExecutedJob, queued time.Time, bookedBy sdk.Hatchery, baseURL, status string) (string, time.Duration) {
 	req := ""
-	for _, r := range requirements {
+	for _, r := range executedJob.Job.Action.Requirements {
 		req += fmt.Sprintf("%s:%s ", r.Type, r.Value)
 	}
 	prj := getVarsInPbj("cds.project", parameters)
@@ -470,7 +509,9 @@ func (ui *Termui) updateQueueJob(idx int, maxQueued time.Duration, id int64, isW
 		fgColor = "fg-magenta"
 	}
 
-	if bookedBy.ID != 0 {
+	if status == sdk.StatusBuilding.String() {
+		row[1] = pad(fmt.Sprintf(" %s.%s ", executedJob.WorkerName, executedJob.WorkerID), 27)
+	} else if bookedBy.ID != 0 {
 		row[1] = pad(fmt.Sprintf(" %s.%d ", bookedBy.Name, bookedBy.ID), 27)
 	} else {
 		row[1] = pad("", 27)
@@ -479,7 +520,9 @@ func (ui *Termui) updateQueueJob(idx int, maxQueued time.Duration, id int64, isW
 	row[4] = fmt.Sprintf("➤ %s", pad(triggeredBy, 17))
 	row[5] = fmt.Sprintf("➤ %s", req)
 
-	item := fmt.Sprintf("  [%s](%s)[%s %s %s %s %s](%s,bg-default)", row[0], c, row[1], row[2], row[3], row[4], row[5], fgColor)
+	_, color := statusShort(status)
+	color = strings.Replace(color, "fg", "bg", 1)
+	item := fmt.Sprintf("  [ ](%s)[ ](bg-default)[%s](%s)[%s %s %s %s %s](%s,bg-default)", color, row[0], c, row[1], row[2], row[3], row[4], row[5], fgColor)
 
 	if idx == ui.queue.Cursor-1 {
 		ui.currentURL = currentURL

engine/api/router_util.go

@@ -106,6 +106,17 @@ func FormString(r *http.Request, s string) string {
 	return r.FormValue(s)
 }
 
+// QueryStrings returns the list of values for given query param key or nil if key no values.
+func QueryStrings(r *http.Request, key string) ([]string, error) {
+	if err := r.ParseForm(); err != nil {
+		return nil, err
+	}
+	if v, ok := r.Form[key]; ok {
+		return v, nil
+	}
+	return nil, nil
+}
+
 // FormInt return a int from query params
 func FormInt(r *http.Request, s string) (int, error) {
 	stringValue := FormString(r, s)

engine/api/workflow_queue.go

@@ -701,6 +701,15 @@ func (api *API) countWorkflowJobQueueHandler() service.Handler {
 func (api *API) getWorkflowJobQueueHandler() service.Handler {
 	return func(ctx context.Context, w http.ResponseWriter, r *http.Request) error {
 		since, until, limit := getSinceUntilLimitHeader(ctx, w, r)
+
+		status, err := QueryStrings(r, "status")
+		if err != nil {
+			return sdk.NewError(sdk.ErrWrongRequest, err)
+		}
+		if !sdk.StatusValidate(status...) {
+			return sdk.NewError(sdk.ErrWrongRequest, fmt.Errorf("Invalid given status"))
+		}
+
 		groupsID := make([]int64, len(getUser(ctx).Groups))
 		usr := getUser(ctx)
 		for i, g := range usr.Groups {
@@ -714,7 +723,7 @@ func (api *API) getWorkflowJobQueueHandler() service.Handler {
 			usr = nil
 		}
 
-		jobs, err := workflow.LoadNodeJobRunQueue(api.mustDB(), api.Cache, permissions, groupsID, usr, &since, &until, &limit)
+		jobs, err := workflow.LoadNodeJobRunQueue(api.mustDB(), api.Cache, permissions, groupsID, usr, &since, &until, &limit, status...)
 		if err != nil {
 			return sdk.WrapError(err, "getWorkflowJobQueueHandler> Unable to load queue")
 		}

sdk/build.go

@@ -136,6 +136,12 @@ func StatusFromString(in string) Status {
 		return StatusDisabled
 	case StatusSkipped.String():
 		return StatusSkipped
+	case StatusStopped.String():
+		return StatusStopped
+	case StatusWorkerPending.String():
+		return StatusWorkerPending
+	case StatusWorkerRegistering.String():
+		return StatusWorkerRegistering
 	default:
 		return StatusUnknown
 	}
@@ -179,3 +185,13 @@ func StatusIsTerminated(status string) bool {
 		return true
 	}
 }
+
+// StatusValidate returns if given strings are valid status.
+func StatusValidate(status ...string) bool {
+	for _, s := range status {
+		if StatusFromString(s) == StatusUnknown {
+			return false
+		}
+	}
+	return true
+}

sdk/cdsclient/client_queue.go

@@ -9,6 +9,7 @@ import (
 	"io/ioutil"
 	"mime/multipart"
 	"net/http"
+	"net/url"
 	"os"
 	"path/filepath"
 	"runtime"
@@ -146,9 +147,19 @@ func (c *client) QueuePolling(ctx context.Context, jobs chan<- sdk.WorkflowNodeJ
 	}
 }
 
-func (c *client) QueueWorkflowNodeJobRun() ([]sdk.WorkflowNodeJobRun, error) {
+func (c *client) QueueWorkflowNodeJobRun(status ...sdk.Status) ([]sdk.WorkflowNodeJobRun, error) {
 	wJobs := []sdk.WorkflowNodeJobRun{}
-	if _, err := c.GetJSON("/queue/workflows", &wJobs); err != nil {
+
+	url, _ := url.Parse("/queue/workflows")
+	if len(status) > 0 {
+		q := url.Query()
+		for _, s := range status {
+			q.Add("status", s.String())
+		}
+		url.RawQuery = q.Encode()
+	}
+
+	if _, err := c.GetJSON(url.String(), &wJobs); err != nil {
 		return nil, err
 	}
 	return wJobs, nil

sdk/cdsclient/interface.go

@@ -197,7 +197,7 @@ type ProjectVariablesClient interface {
 
 // QueueClient exposes queue related functions
 type QueueClient interface {
-	QueueWorkflowNodeJobRun() ([]sdk.WorkflowNodeJobRun, error)
+	QueueWorkflowNodeJobRun(status ...sdk.Status) ([]sdk.WorkflowNodeJobRun, error)
 	QueueCountWorkflowNodeJobRun(since *time.Time, until *time.Time) (sdk.WorkflowNodeJobRunCount, error)
 	QueuePipelineBuildJob() ([]sdk.PipelineBuildJob, error)
 	QueuePolling(context.Context, chan<- sdk.WorkflowNodeJobRun, chan<- sdk.PipelineBuildJob, chan<- error, time.Duration, int, *int64) error

ui/src/app/model/job.model.ts

@@ -8,6 +8,8 @@ export class Job {
     last_modified: string;
     step_status: Array<StepStatus>;
     warnings: Array<ActionWarning>;
+    worker_name: string;
+    worker_id: string;
 
     // UI parameter
     hasChanged: boolean;