feat(tasks): stop all tasks

Author: fiftin

.dredd/hooks/main.go

@@ -126,6 +126,7 @@ func main() {
 	h.Before("template > /api/project/{project_id}/templates/{template_id} > Get template > 200 > application/json", capabilityWrapper("template"))
 	h.Before("template > /api/project/{project_id}/templates/{template_id} > Updates template > 204 > application/json", capabilityWrapper("template"))
 	h.Before("template > /api/project/{project_id}/templates/{template_id} > Removes template > 204 > application/json", capabilityWrapper("template"))
+	h.Before("template > /api/project/{project_id}/templates/{template_id}/stop_all_tasks > Stop all active tasks of template > 204 > application/json", capabilityWrapper("template"))
 
 	h.Before("task > /api/project/{project_id}/tasks > Starts a job > 201 > application/json", capabilityWrapper("template"))
 	h.Before("task > /api/project/{project_id}/tasks/last > Get last 200 Tasks related to current project > 200 > application/json", capabilityWrapper("template"))

api-docs.yml

@@ -2437,6 +2437,25 @@ paths:
         204:
           description: template removed
 
+  /project/{project_id}/templates/{template_id}/stop_all_tasks:
+    parameters:
+      - $ref: "#/parameters/project_id"
+      - $ref: "#/parameters/template_id"
+    post:
+      tags:
+        - template
+      summary: Stop all active tasks of template
+      responses:
+        201:
+          description: tasks stopped
+    delete:
+      tags:
+        - template
+      summary: Removes template
+      responses:
+        204:
+          description: template removed
+
 
   # project schedules
   /project/{project_id}/schedules/{schedule_id}:

api/projects/tasks.go

@@ -4,15 +4,16 @@ import (
 	"bytes"
 	"encoding/json"
 	"errors"
+	"net/http"
+	"strconv"
+	"time"
+
 	"github.com/semaphoreui/semaphore/api/helpers"
 	"github.com/semaphoreui/semaphore/db"
 	"github.com/semaphoreui/semaphore/pkg/common_errors"
 	"github.com/semaphoreui/semaphore/services/tasks"
 	"github.com/semaphoreui/semaphore/util"
 	log "github.com/sirupsen/logrus"
-	"net/http"
-	"strconv"
-	"time"
 )
 
 type TaskController struct {
@@ -412,5 +413,19 @@ func GetTaskStats(w http.ResponseWriter, r *http.Request) {
 }
 
 func (c *TaskController) StopAllTasks(w http.ResponseWriter, r *http.Request) {
+	project := helpers.GetFromContext(r, "project").(db.Project)
+	tpl := helpers.GetFromContext(r, "template").(db.Template)
+
+	var stopObj struct {
+		Force bool `json:"force"`
+	}
+
+	// optional body; ignore bind error and default Force=false
+	if ok := helpers.Bind(w, r, &stopObj); !ok {
+		helpers.WriteErrorStatus(w, "Not allowed", http.StatusBadRequest)
+		return
+	}
+
+	taskPool(r).StopTasksByTemplate(project.ID, tpl.ID, stopObj.Force)
 	w.WriteHeader(http.StatusNoContent)
 }

db/Store.go

@@ -52,6 +52,7 @@ type RetrieveQueryParams struct {
 	SortInverted bool
 	Filter       string
 	Ownership    OwnershipFilter
+	TaskFilter   *TaskFilter
 }
 
 type ObjectReferrer struct {
@@ -171,6 +172,7 @@ type TaskFilter struct {
 	Start  *time.Time `json:"start"`
 	End    *time.Time `json:"end"`
 	UserID *int       `json:"user_id"`
+	Status []task_logger.TaskStatus
 }
 
 type TaskStat struct {

db/sql/task.go

@@ -2,10 +2,11 @@ package sql
 
 import (
 	"encoding/json"
-	"github.com/Masterminds/squirrel"
-	"github.com/semaphoreui/semaphore/db"
 	"math/rand"
 	"time"
+
+	"github.com/Masterminds/squirrel"
+	"github.com/semaphoreui/semaphore/db"
 )
 
 func (d *SqlDb) CreateTaskStage(stage db.TaskStage) (res db.TaskStage, err error) {
@@ -243,6 +244,10 @@ func (d *SqlDb) getTasks(projectID int, templateID *int, taskIDs []int, params d
 		LeftJoin("`user` on task.user_id=`user`.id").
 		OrderBy("id desc")
 
+	if params.TaskFilter != nil && len(params.TaskFilter.Status) > 0 {
+		q = q.Where(squirrel.Eq{"status": params.TaskFilter.Status})
+	}
+
 	if templateID == nil {
 		q = q.Where("tpl.project_id=?", projectID)
 	} else {

pkg/task_logger/task_logger.go

@@ -20,6 +20,18 @@ const (
 	TaskFailStatus          TaskStatus = "error"
 )
 
+func UnfinishedTaskStatuses() []TaskStatus {
+	return []TaskStatus{
+		TaskWaitingStatus,
+		TaskStartingStatus,
+		TaskWaitingConfirmation,
+		TaskConfirmed,
+		TaskRejected,
+		TaskRunningStatus,
+		TaskStoppingStatus,
+	}
+}
+
 func (s TaskStatus) IsValid() bool {
 	switch s {
 	case TaskWaitingStatus,

services/tasks/TaskPool.go

@@ -422,6 +422,81 @@ func (p *TaskPool) StopTask(targetTask db.Task, forceStop bool) error {
 	return nil
 }
 
+// StopTasksByTemplate stops all active (queued or running) tasks that belong to
+// the specified project and template. If forceStop is true, tasks are marked as
+// stopped immediately and running tasks are killed; otherwise tasks are marked
+// as stopping and will gracefully transition to stopped.
+func (p *TaskPool) StopTasksByTemplate(projectID int, templateID int, forceStop bool) {
+	// Handle queued tasks
+	for _, t := range p.state.QueueRange() {
+		if t == nil {
+			continue
+		}
+		if t.Task.ProjectID != projectID || t.Task.TemplateID != templateID {
+			continue
+		}
+		if t.Task.Status.IsFinished() {
+			continue
+		}
+		if forceStop {
+			t.SetStatus(task_logger.TaskStoppedStatus)
+		} else {
+			t.SetStatus(task_logger.TaskStoppingStatus)
+		}
+		// Queued tasks will be dequeued and immediately finalize to Stopped in run()
+	}
+
+	// Handle running tasks
+	for _, t := range p.state.RunningRange() {
+		if t == nil {
+			continue
+		}
+		if t.Task.ProjectID != projectID || t.Task.TemplateID != templateID {
+			continue
+		}
+		if t.Task.Status.IsFinished() {
+			continue
+		}
+		prevStatus := t.Task.Status
+		if forceStop {
+			t.SetStatus(task_logger.TaskStoppedStatus)
+		} else {
+			t.SetStatus(task_logger.TaskStoppingStatus)
+		}
+		if prevStatus == task_logger.TaskRunningStatus {
+			t.kill()
+		}
+	}
+
+	// Update tasks in DB that are neither queued nor running but still active
+	// (e.g., created but not present in this instance's memory state).
+	if tasks, err := p.store.GetTemplateTasks(projectID, templateID, db.RetrieveQueryParams{
+		TaskFilter: &db.TaskFilter{
+			Status: task_logger.UnfinishedTaskStatuses(),
+		},
+	}); err == nil {
+		for _, twt := range tasks {
+
+			// if task is managed locally (queued/running), it was handled above
+			if p.GetTask(twt.Task.ID) != nil {
+				continue
+			}
+
+			// mark non-local task as stopped and write event for history
+			tr := NewTaskRunner(twt.Task, p, "", p.keyInstallationService)
+			if err := tr.populateDetails(); err != nil {
+				log.Error(err)
+				continue
+			}
+
+			tr.SetStatus(task_logger.TaskStoppedStatus)
+			tr.createTaskEvent()
+		}
+	} else {
+		log.Error(err)
+	}
+}
+
 // GetQueuedTasks returns a snapshot of tasks currently queued
 func (p *TaskPool) GetQueuedTasks() []*TaskRunner {
 	return p.state.QueueRange()