task

Documentation

Index

• Constants
• Variables
• func ExecTaskAction(action string, task *TaskExecEntity) error
• func PullTasksTicker(switchChan chan bool)
• func ReplyTaskTicker()
• func StartTaskExecEntity(taskExecEntity *TaskExecEntity)
• type CmdMeta
• type Output
• type PullTaskRequestBody
• type PullTaskRespBody
• type ReplyTask
• type ReplyTaskRequestBody
• type ReplyTaskRespBody
• type Task
• type TaskCancel
  • func InstantiateTaskCancel(taskPulled *TaskPulled) *TaskCancel
  • func (tc *TaskCancel) Cancel() error
  • func (tc *TaskCancel) CarryOn() error
  • func (tc *TaskCancel) Pause() error
  • func (tc *TaskCancel) Run(exitCodeChan chan int, outputChan chan *string) error
• type TaskEventProcessor
  • func (p *TaskEventProcessor) Action(action string, fromState string, toState string, args []interface{})
  • func (p *TaskEventProcessor) OnEnter(toState string, args []interface{})
  • func (p *TaskEventProcessor) OnExit(fromState string, args []interface{})
• type TaskExecEntity
  • func InstantiateTaskExecEntity(taskPulled *TaskPulled) *TaskExecEntity
  • func (t *TaskExecEntity) RunAndListen()
• type TaskInterface
  • func InstantiateTaskInterface(taskPulled *TaskPulled) TaskInterface
• type TaskPulled
• type TaskRunShell
  • func InstantiateTaskRunShell(taskPulled *TaskPulled) *TaskRunShell
  • func (trs *TaskRunShell) Cancel() error
  • func (trs *TaskRunShell) CarryOn() error
  • func (trs *TaskRunShell) Pause() error
  • func (trs *TaskRunShell) Run(exitCodeChan chan int, outputChan chan *string) error

Constants

const (
	// TIME_LOCATION_CST represents UTC +08:00
	TIME_LOCATION_CST = "Asia/Shanghai"

	// Reply Task Error NO.
	ERROR_INTERNAL = "asssit.001 Internal Error"

	TASK_COMMAND_INSTALL   = "INSTALL"
	TASK_COMMAND_RUN_SHELL = "RUN_SHELL"
	TASK_COMMAND_CANCEL    = "CANCEL"

	MAX_EVENT_CHAN_SIZE       = 3
	MAX_RETURN_CODE_CHAN_SIZE = 1
	MAX_EXIT_CHAN_SIZE        = 1
	MAX_OUTPUT_CHAN_SIZE      = 1
	MAX_STATES_SIZE           = 20
	MAX_EVENTS_SIZE           = 20

	STATE_CREATED   = "CREATED"
	STATE_RUNNING   = "RUNNING"
	STATE_SLEEPING  = "SLEEPING"
	STATE_TIMEOUT   = "TIMEOUT"
	STATE_FAILED    = "FAILED"
	STATE_SUCCEEDED = "SUCCEEDED"
	STATE_CANCELED  = "CANCELED"

	EVENT_RUN            = "EVENT_RUN"
	EVENT_PAUSE          = "EVENT_PAUSE"
	EVENT_CARRY_ON       = "EVENT_CARRY_ON"
	EVENT_CANCEL         = "EVENT_CANCEL"
	EVENT_TIMEOUT        = "EVENT_TIMEOUT"
	EVENT_RETURN_NONZERO = "EVENT_RETURN_NONZERO"
	EVENT_RETURN_ZERO    = "EVENT_RETURN_ZERO"

	ACTION_RUN            = "ACTION_RUN"
	ACTION_PAUSE          = "ACTION_PAUSE"
	ACTION_CARRY_ON       = "ACTION_CARRY_ON"
	ACTION_CANCEL         = "ACTION_CANCEL"
	ACTION_TIMEOUT        = "ACTION_TIMEOUT"
	ACTION_RETURN_NONZERO = "ACTION_RETURN_NONZERO"
	ACTION_RETURN_ZERO    = "ACTION_RETURN_ZERO"

	CRON_TASKEXEC_MAINTAIN_COUNT = 10
)

const (
	// MAX_STDOUT_IN_BYTES ...
	MAX_STDOUT_IN_BYTES                 = 8192
	MAX_STDERR_IN_BYTES                 = 8192
	EXIT_CODE_ZERO                      = 0
	EXIT_CODE_NON_EXISTENT_WORKING_PATH = -1
	NON_EXISTENT_WORKING_PATH_INFO      = "Working directory does not exist."
)

Variables

var (
	// TaskMap ...
	TaskMap = make(map[string]*Task)
	TaskFSM = initTaskFSM()
)

Functions

func ExecTaskAction

func ExecTaskAction(action string, task *TaskExecEntity) error

ExecTaskAction ...

func PullTasksTicker

func PullTasksTicker(switchChan chan bool)

PullTasksTicker ...

func ReplyTaskTicker

func ReplyTaskTicker()

ReplyTaskTicker ..

func StartTaskExecEntity

func StartTaskExecEntity(taskExecEntity *TaskExecEntity)

StartTaskExecEntity ...

Types

type CmdMeta

type CmdMeta struct {
	WorkPath string `json:"work_path"`
	Command  string `json:"command"`
	// TaskID/InvocationID appears only when Command is CANCEL/STOP/PAUSE
	TaskID       string `json:"task_id"`
	InvocationID string `json:"invocation_id"`
}

CmdMeta ...

type Output

type Output struct {
	ExitCode int    `json:"exit_code"`
	StdOut   string `json:"std_out"`
	StdErr   string `json:"std_err"`
}

Output ...

type PullTaskRequestBody

type PullTaskRequestBody struct {
	InvokeScope string `json:"invoke_scope"`
	Resources   string `json:"resources"`
}

type PullTaskRespBody

type PullTaskRespBody struct {
	Tasks []TaskPulled `json:"tasks"`
}

PullTaskRespBody ...

type ReplyTask

type ReplyTask struct {
	TaskID       string `json:"task_id"`
	InvocationID string `json:"invocation_id"`
	Status       string `json:"status"`
	// ErrNum indicates the internal error in our system, not the return code of certain execution
	ErrNum string `json:"err_no"`
	Output string `json:"output"`
}

ReplyTask ...

type ReplyTaskRequestBody

type ReplyTaskRequestBody struct {
	InstanceID string      `json:"instance_id"`
	Tasks      []ReplyTask `json:"tasks"`
}

ReplyTaskRequestBody ...

type ReplyTaskRespBody

type ReplyTaskRespBody struct {
	SucceededList []string `json:"succeeded_list"`
	FailedList    []string `json:"failed_list"`
}

ReplyTaskRespBody ...

type Task

type Task struct {
	TaskPulled         *TaskPulled
	Cron               *cron.Cron
	TaskExecEntityList *list.List
}

Task ...

type TaskCancel

type TaskCancel struct {
	TaskPulled *TaskPulled
}

TaskCancel the struct type for Run,Pause,Cancel,CarryOn

func InstantiateTaskCancel

func InstantiateTaskCancel(taskPulled *TaskPulled) *TaskCancel

InstantiateTaskCancel ...

func (*TaskCancel) Cancel

func (tc *TaskCancel) Cancel() error

Cancel command Cancel, the process can be killed

func (*TaskCancel) CarryOn

func (tc *TaskCancel) CarryOn() error

CarryOn command CarryOn, the paused task can continue

func (*TaskCancel) Pause

func (tc *TaskCancel) Pause() error

Pause command pause, it may continue to be executed, like CarryOn()

func (*TaskCancel) Run

func (tc *TaskCancel) Run(exitCodeChan chan int, outputChan chan *string) error

Run command execution

type TaskEventProcessor

type TaskEventProcessor struct{}

TaskEventProcessor ...

func (*TaskEventProcessor) Action

func (p *TaskEventProcessor) Action(action string, fromState string, toState string, args []interface{})

Action ...

func (*TaskEventProcessor) OnEnter

func (p *TaskEventProcessor) OnEnter(toState string, args []interface{})

OnEnter ...

func (*TaskEventProcessor) OnExit

func (p *TaskEventProcessor) OnExit(fromState string, args []interface{})

OnExit ...

type TaskExecEntity

type TaskExecEntity struct {
	TaskInterface

	TaskPulled   *TaskPulled
	EventChan    chan string
	ExitCodeChan chan int
	ExitChan     chan bool
	OutputChan   chan *string
	Output       *string
	State        string
	States       *list.List
	Events       *list.List
}

TaskExecEntity ...

func InstantiateTaskExecEntity

func InstantiateTaskExecEntity(taskPulled *TaskPulled) *TaskExecEntity

todo InstantiateTaskInterface 初始化失败场景需要考虑 InstantiateTaskExecEntity ...

func (*TaskExecEntity) RunAndListen

func (t *TaskExecEntity) RunAndListen()

RunAndListen ...

type TaskInterface

type TaskInterface interface {
	// Blocking
	Run(chan int, chan *string) error
	// Unblocking
	Pause() error
	Cancel() error
	CarryOn() error
}

TaskInterface ...

func InstantiateTaskInterface

func InstantiateTaskInterface(taskPulled *TaskPulled) TaskInterface

InstantiateTaskInterface ...

type TaskPulled

type TaskPulled struct {
	TaskID       string  `json:"task_id"`
	InvocationID string  `json:"invocation_id"`
	Name         string  `json:"name"`
	Description  string  `json:"description"`
	Command      string  `json:"command"`
	CmdMeta      CmdMeta `json:"cmd_meta"`
	Cron         string  `json:"cron"`
	Timeout      int     `json:"timeout"`
	Type         string  `json:"type"`
}

TaskPulled ...

type TaskRunShell

type TaskRunShell struct {
	TaskPulled *TaskPulled
	CMD        *exec.Cmd
}

TaskRunShell the struct type for Run,Pause,Cancel,CarryOn

func InstantiateTaskRunShell

func InstantiateTaskRunShell(taskPulled *TaskPulled) *TaskRunShell

InstantiateTaskRunShell ...

func (*TaskRunShell) Cancel

func (trs *TaskRunShell) Cancel() error

Cancel ... command Cancel, the process can be killed

func (*TaskRunShell) CarryOn

func (trs *TaskRunShell) CarryOn() error

CarryOn ... command CarryOn, the paused task can continue

func (*TaskRunShell) Pause

func (trs *TaskRunShell) Pause() error

Pause ... command pause, it may continue to be executed, like CarryOn()

func (*TaskRunShell) Run

func (trs *TaskRunShell) Run(exitCodeChan chan int, outputChan chan *string) error

Run ... command execution