workflowcontroller

Documentation

Index

• Constants
• 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 SendWorkflowNotifyMessage(task *commonmodels.WorkflowTask, receiver string, status config.Status, ...)
• 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()
• func WorkflowDebugChanKey(workflowName string, taskID int64) string
• func WorkflowDebugLockKey(workflowName string, taskID int64) string
• type CancelMessage
• type CustomStageCtl
  • func NewCustomStageCtl(stage *commonmodels.StageTask, workflowCtx *commonmodels.WorkflowTaskCtx, ...) *CustomStageCtl
  • func (c *CustomStageCtl) AfterRun()
  • func (c *CustomStageCtl) Run(ctx context.Context, concurrency int)
• type ProductServiceDeployInfo
• type StageCtl
• type WorkflowDebugEvent

Constants

const (
	WorkflowDebugEventEnableDebug   = "EnableDebug"
	WorkflowDebugEventDeleteDebug   = "DeleteDebug"
	WorkflowDebugEventSetBreakPoint = "SetBreakpoint"
)

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 SendWorkflowNotifyMessage

func SendWorkflowNotifyMessage(task *commonmodels.WorkflowTask, receiver string, status config.Status, log *zap.SugaredLogger)

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

func WorkflowDebugChanKey

func WorkflowDebugChanKey(workflowName string, taskID int64) string

func WorkflowDebugLockKey

func WorkflowDebugLockKey(workflowName string, taskID int64) string

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) AfterRun

func (c *CustomStageCtl) AfterRun()

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 WorkflowDebugEvent

type WorkflowDebugEvent struct {
	EventType string `json:"event_type"`
	JobName   string `json:"job_name"`
	TaskID    int64  `json:"task_id"`
	Set       bool   `json:"set"`
	Position  string `json:"position"`
}