workflowcontroller

Documentation

Index

• func ApproveStage(workflowName, stageName, userName, userID, comment string, taskID int64, ...) error
• func BlockedTaskQueue() ([]*commonmodels.WorkflowQueue, error)
• func CancelWorkflowTask(userName, workflowName string, taskID int64, logger *zap.SugaredLogger) error
• func ConvertTaskToQueue(task *commonmodels.WorkflowTask) *commonmodels.WorkflowQueue
• func CreateTask(t *commonmodels.WorkflowTask) error
• func GetContextKey(key string) string
• func InitQueue() error
• func InitWorkflowController()
• func ListTasks() []*commonmodels.WorkflowQueue
• func NewWorkflowController(workflowTask *commonmodels.WorkflowTask, logger *zap.SugaredLogger) *workflowCtl
• func NextWaitingTask() (*commonmodels.WorkflowQueue, error)
• func ParallelRunningAndQueuedTasks(currentTask *commonmodels.WorkflowQueue) bool
• func PendingTasks() []*commonmodels.WorkflowQueue
• func Push(t *commonmodels.WorkflowTask) error
• func Remove(taskQueue *commonmodels.WorkflowQueue) error
• func RunStages(ctx context.Context, stages []*commonmodels.StageTask, ...)
• func RunningAndQueuedTasks() []*commonmodels.WorkflowQueue
• func RunningTasks() []*commonmodels.WorkflowQueue
• func UpdateQueue(task *commonmodels.WorkflowTask) bool
• func UpdateTask(t *commonmodels.WorkflowTask) error
• func WorfklowTaskSender()
• type CancelMessage
• type CustomStageCtl
  • func NewCustomStageCtl(stage *commonmodels.StageTask, workflowCtx *commonmodels.WorkflowTaskCtx, ...) *CustomStageCtl
  • func (c *CustomStageCtl) Run(ctx context.Context, concurrency int)
• type StageCtl

Functions

func ApproveStage

func ApproveStage(workflowName, stageName, userName, userID, comment string, taskID int64, approve bool) error

func BlockedTaskQueue

func BlockedTaskQueue() ([]*commonmodels.WorkflowQueue, error)

func CancelWorkflowTask

func CancelWorkflowTask(userName, workflowName string, taskID int64, logger *zap.SugaredLogger) error

func ConvertTaskToQueue

func ConvertTaskToQueue(task *commonmodels.WorkflowTask) *commonmodels.WorkflowQueue

func CreateTask

func CreateTask(t *commonmodels.WorkflowTask) error

CreateTask 接受create task请求, 保存task到数据库, 发送task到queue

func GetContextKey

func GetContextKey(key string) string

func InitQueue

func InitQueue() error

func InitWorkflowController

func InitWorkflowController()

func ListTasks

func ListTasks() []*commonmodels.WorkflowQueue

func NewWorkflowController

func NewWorkflowController(workflowTask *commonmodels.WorkflowTask, logger *zap.SugaredLogger) *workflowCtl

func NextWaitingTask

func NextWaitingTask() (*commonmodels.WorkflowQueue, error)

NextWaitingTask 查询下一个等待的task

func ParallelRunningAndQueuedTasks

func ParallelRunningAndQueuedTasks(currentTask *commonmodels.WorkflowQueue) bool

func PendingTasks

func PendingTasks() []*commonmodels.WorkflowQueue

func Push

func Push(t *commonmodels.WorkflowTask) error

func Remove

func Remove(taskQueue *commonmodels.WorkflowQueue) error

func RunStages

func RunStages(ctx context.Context, stages []*commonmodels.StageTask, workflowCtx *commonmodels.WorkflowTaskCtx, concurrency int, logger *zap.SugaredLogger, ack func())

func RunningAndQueuedTasks

func RunningAndQueuedTasks() []*commonmodels.WorkflowQueue

func RunningTasks

func RunningTasks() []*commonmodels.WorkflowQueue

func UpdateQueue

func UpdateQueue(task *commonmodels.WorkflowTask) bool

func UpdateTask

func UpdateTask(t *commonmodels.WorkflowTask) error

func WorfklowTaskSender

func WorfklowTaskSender()

WorfklowTaskSender 监控warpdrive空闲情况, 如果有空闲, 则发现下一个waiting task给warpdrive 并将task状态设置为queued

Types

type CancelMessage

type CancelMessage struct {
	Revoker      string `json:"revoker"`
	PipelineName string `json:"pipeline_name"`
	TaskID       int64  `json:"task_id"`
	ReqID        string `json:"req_id"`
}

type CustomStageCtl

type CustomStageCtl struct {
	// contains filtered or unexported fields
}

func NewCustomStageCtl

func NewCustomStageCtl(stage *commonmodels.StageTask, workflowCtx *commonmodels.WorkflowTaskCtx, logger *zap.SugaredLogger, ack func()) *CustomStageCtl

func (*CustomStageCtl) Run

func (c *CustomStageCtl) Run(ctx context.Context, concurrency int)

type StageCtl

type StageCtl interface {
	Run(ctx context.Context, concurrency int)
}