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 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 RunningWorkflowTasks(name string) ([]*commonmodels.WorkflowQueue, error)
• func UpdateQueue(task *commonmodels.WorkflowTask) bool
• func UpdateTask(t *commonmodels.WorkflowTask) error
• func WaitForApproveWorkflowTasks(name string) ([]*commonmodels.WorkflowQueue, error)
• func WaitingTasks() ([]*commonmodels.WorkflowQueue, 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 ProductServiceDeployInfo
• type StageCtl
• type WorkflowTaskWithLock
  • func GetWorkflowTaskInMap(workflowName string, taskID int64) *WorkflowTaskWithLock

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 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 RunningWorkflowTasks

func RunningWorkflowTasks(name string) ([]*commonmodels.WorkflowQueue, error)

func UpdateQueue

func UpdateQueue(task *commonmodels.WorkflowTask) bool

func UpdateTask

func UpdateTask(t *commonmodels.WorkflowTask) error

func WaitForApproveWorkflowTasks

func WaitForApproveWorkflowTasks(name string) ([]*commonmodels.WorkflowQueue, error)

func WaitingTasks

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

WaitingTasks 查询所有等待的task

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 ProductServiceDeployInfo

type ProductServiceDeployInfo struct {
	ProductName     string
	EnvName         string
	ServiceName     string
	Uninstall       bool
	ServiceRevision int
	VariableYaml    string
	Containers      []*models.Container
}

type StageCtl

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

type WorkflowTaskWithLock

type WorkflowTaskWithLock struct {
	WorkflowTask *models.WorkflowTask
	Ack          func()
	sync.RWMutex
}

func GetWorkflowTaskInMap

func GetWorkflowTaskInMap(workflowName string, taskID int64) *WorkflowTaskWithLock