support pipeline pause

gaia.go

@@ -62,7 +62,10 @@ const (
 
 	// RunNotScheduled status
 	RunNotScheduled PipelineRunStatus = "not scheduled"
-
+
+	// PausedScheduled status
+	PausedScheduled PipelineRunStatus = "paused"
+
 	// RunScheduled status
 	RunScheduled PipelineRunStatus = "scheduled"
 

handlers/handler.go

@@ -77,6 +77,8 @@ func (s *GaiaHandler) InitHandlers(e *echo.Echo) error {
 		apiAuthGrp.GET("pipelinerun/:pipelineid", s.deps.PipelineProvider.PipelineGetAllRuns)
 		apiAuthGrp.GET("pipelinerun/:pipelineid/latest", s.deps.PipelineProvider.PipelineGetLatestRun)
 		apiAuthGrp.GET("pipelinerun/:pipelineid/:runid/log", s.deps.PipelineProvider.GetJobLogs)
+		apiAuthGrp.POST("pipelinerun/:pipelineid/:runid/pause", s.deps.PipelineProvider.PipelinePause)
+		apiAuthGrp.POST("pipelinerun/:pipelineid/:runid/unpause", s.deps.PipelineProvider.PipelineUnPause)
 
 		// Secrets
 		apiAuthGrp.GET("secrets", ListSecrets)

providers/pipelines/pipeline_provider.go

@@ -47,6 +47,8 @@ type PipelineProviderer interface {
 	SettingsPollOn(c echo.Context) error
 	SettingsPollOff(c echo.Context) error
 	SettingsPollGet(c echo.Context) error
+	PipelinePause(c echo.Context) error
+	PipelineUnPause(c echo.Context) error
 }
 
 // NewPipelineProvider creates a new provider with the needed dependencies.

providers/pipelines/pipeline_run.go

@@ -18,6 +18,9 @@ import (
 var (
 	// errPipelineRunNotFound is thrown when a pipeline run was not found with the given id
 	errPipelineRunNotFound = errors.New("pipeline run not found with the given id")
+
+	// errPipelineStatusCannotBeChange is thrown when a pipeline status cannot be changed with the given id
+	errPipelineStatusCannotBeChange = errors.New("pipeline status cannot be changed to paused")
 )
 
 // jobLogs represents the json format which is returned
@@ -240,3 +243,99 @@ func (pp *PipelineProvider) GetJobLogs(c echo.Context) error {
 	// Return logs
 	return c.JSON(http.StatusOK, jL)
 }
+
+// PipelinePause pause a running pipeline.
+// @Summary Pause a pipeline run.
+// @Description Pause a pipeline run.
+// @Tags pipelinerun
+// @Accept plain
+// @Produce plain
+// @Security ApiKeyAuth
+// @Param pipelineid query string true "ID of the pipeline"
+// @Param runid query string true "ID of the pipeline run"
+// @Success 200 {object} gaia.PipelineRun "pipeline run"
+// @Failure 400 {string} string "Invalid pipeline id or run id"
+// @Failure 404 {string} string "Pipeline Run not found."
+// @Router /pipelinerun/{pipelineid}/{runid}/pause [post]
+func (pp *PipelineProvider) PipelinePause(c echo.Context) error {
+	// Convert string to int because id is int
+	storeService, _ := services.StorageService()
+	pipelineID, err := strconv.Atoi(c.Param("pipelineid"))
+	if err != nil {
+		return c.String(http.StatusBadRequest, errInvalidPipelineID.Error())
+	}
+
+	// Convert string to int because id is int
+	runID, err := strconv.Atoi(c.Param("runid"))
+	if err != nil {
+		return c.String(http.StatusBadRequest, errPipelineRunNotFound.Error())
+	}
+
+	// Find pipeline run in store
+	pipelineRun, err := storeService.PipelineGetRunByPipelineIDAndID(pipelineID, runID)
+	if err != nil {
+		return c.String(http.StatusInternalServerError, err.Error())
+	}
+
+	// Only when pipelineRun is in the RunScheduled or  RunNotScheduled state can it be changed to the PausedScheduled state
+	if pipelineRun.Status == gaia.RunScheduled || pipelineRun.Status == gaia.RunNotScheduled {
+		pipelineRun.Status = gaia.PausedScheduled
+		err = storeService.PipelinePutRun(pipelineRun)
+		if err != nil {
+			return c.String(http.StatusInternalServerError, err.Error())
+		}
+	} else {
+		return c.String(http.StatusBadRequest, errPipelineStatusCannotBeChange.Error())
+	}
+
+	// Return pipeline run
+	return c.JSON(http.StatusOK, pipelineRun)
+}
+
+// PipelineUnPause unpause a running pipeline.
+// @Summary Unpause a pipeline run.
+// @Description Unpause a pipeline run.
+// @Tags pipelinerun
+// @Accept plain
+// @Produce plain
+// @Security ApiKeyAuth
+// @Param pipelineid query string true "ID of the pipeline"
+// @Param runid query string true "ID of the pipeline run"
+// @Success 200 {object} gaia.PipelineRun "pipeline run"
+// @Failure 400 {string} string "Invalid pipeline id or run id"
+// @Failure 404 {string} string "Pipeline Run not found."
+// @Router /pipelinerun/{pipelineid}/{runid}/unpause [post]
+func (pp *PipelineProvider) PipelineUnPause(c echo.Context) error {
+	// Convert string to int because id is int
+	storeService, _ := services.StorageService()
+	pipelineID, err := strconv.Atoi(c.Param("pipelineid"))
+	if err != nil {
+		return c.String(http.StatusBadRequest, errInvalidPipelineID.Error())
+	}
+
+	// Convert string to int because id is int
+	runID, err := strconv.Atoi(c.Param("runid"))
+	if err != nil {
+		return c.String(http.StatusBadRequest, errPipelineRunNotFound.Error())
+	}
+
+	// Find pipeline run in store
+	pipelineRun, err := storeService.PipelineGetRunByPipelineIDAndID(pipelineID, runID)
+	if err != nil {
+		return c.String(http.StatusInternalServerError, err.Error())
+	}
+
+	// Only when pipelineRun is in the PausedScheduled state can it be changed to the RunNotScheduled state
+	if pipelineRun.Status == gaia.PausedScheduled {
+		pipelineRun.Status = gaia.RunNotScheduled
+		err = storeService.PipelinePutRun(pipelineRun)
+		if err != nil {
+			return c.String(http.StatusInternalServerError, err.Error())
+		}
+	} else {
+		return c.String(http.StatusBadRequest, errPipelineStatusCannotBeChange.Error())
+	}
+
+	// Return pipeline run
+	return c.JSON(http.StatusOK, pipelineRun)
+}

workers/scheduler/gaiascheduler/scheduler.go

@@ -170,6 +170,13 @@ func (s *Scheduler) work() {
 
 // prepareAndExec does the preparation and starts the execution.
 func (s *Scheduler) prepareAndExec(r gaia.PipelineRun) {
+	// Check the pipeline status, if it is PausedScheduled, exit this execution
+	if g, err := s.storeService.PipelineGetRunByID(r.UniqueID); err == nil && g != nil {
+		if g.Status == gaia.PausedScheduled {
+			return
+		}
+	}
+
 	// Mark the scheduled run as running
 	r.Status = gaia.RunRunning
 	r.StartDate = time.Now()
@@ -271,6 +278,11 @@ func (s *Scheduler) schedule() {
 			}
 		}
 
+		// If the pipeline state is PausedScheduled, skip this loop
+		if scheduled[id].Status == gaia.PausedScheduled {
+			continue
+		}
+
 		// If we are a server instance, we will by default give the worker the advantage.
 		// Only in case all workers are busy we will schedule work on the server.
 		workers := s.memDBService.GetAllWorker()