workflow

Documentation

Index

• Constants
• Variables
• func InitWorkflowMirrorPlugin(mgr Manager)
• type Job
  • func (n *Job) GetControlPolicy() flow.ControlPolicy
  • func (n *Job) GetHooks() flow.Hooks
  • func (n *Job) GetMessage() string
  • func (n *Job) GetStatus() fsm.Status
  • func (n *Job) ID() flow.FID
  • func (n *Job) NextBatch(ctx *flow.Context) ([]flow.Task, error)
  • func (n *Job) SetMessage(msg string)
  • func (n *Job) SetStatus(status fsm.Status)
  • func (n *Job) SetStepStatus(stepName flow.TName, status fsm.Status)
  • func (n *Job) Setup(ctx *flow.Context) error
  • func (n *Job) Teardown(ctx *flow.Context)
  • func (n *Job) Type() flow.FType
• type JobStep
  • func (n *JobStep) Do(ctx *flow.Context) error
  • func (n *JobStep) GetMessage() string
  • func (n *JobStep) GetStatus() fsm.Status
  • func (n *JobStep) Name() flow.TName
  • func (n *JobStep) SetMessage(msg string)
  • func (n *JobStep) SetStatus(status fsm.Status)
  • func (n *JobStep) Setup(ctx *flow.Context) error
  • func (n *JobStep) Teardown(ctx *flow.Context)
• type Manager
  • func NewManager(recorder metastore.ScheduledTaskRecorder) (Manager, error)
• type MirrorPlugin
  • func (w *MirrorPlugin) Name() string
  • func (w *MirrorPlugin) Run(ctx context.Context, request *common.Request, params map[string]string) (*common.Response, error)
  • func (w *MirrorPlugin) Type() types.PluginType
  • func (w *MirrorPlugin) Version() string
• type Runner
  • func InitWorkflowRunner(recorder metastore.ScheduledTaskRecorder) (*Runner, error)
  • func (r *Runner) DeleteFlow(flowId flow.FID) error
  • func (r *Runner) DeleteTask(flowId flow.FID, taskName flow.TName) error
  • func (r *Runner) GetFlow(flowId flow.FID) (flow.Flow, error)
  • func (r *Runner) GetFlowMeta(flowId flow.FID) (*flowstorage.FlowMeta, error)
  • func (r *Runner) SaveFlow(flow flow.Flow) error
  • func (r *Runner) SaveTask(flowId flow.FID, task flow.Task) error
  • func (r *Runner) Start(stopCh chan struct{}) error
  • func (r *Runner) WorkFlowHandler(ctx context.Context, wf *types.WorkflowSpec) (*types.WorkflowJob, error)

Constants

const (
	MirrorPluginName    = "workflow"
	MirrorPluginVersion = "1.0"
)

Variables

var (
	ErrJobNotFound = errors.New("job not found")
)

Functions

func InitWorkflowMirrorPlugin

func InitWorkflowMirrorPlugin(mgr Manager)

Types

type Job

type Job struct {
	*types.WorkflowJob
	// contains filtered or unexported fields
}

func (*Job) GetControlPolicy

func (n *Job) GetControlPolicy() flow.ControlPolicy

func (*Job) GetHooks

func (n *Job) GetHooks() flow.Hooks

func (*Job) GetMessage

func (n *Job) GetMessage() string

func (*Job) GetStatus

func (n *Job) GetStatus() fsm.Status

func (*Job) ID

func (n *Job) ID() flow.FID

func (*Job) NextBatch

func (n *Job) NextBatch(ctx *flow.Context) ([]flow.Task, error)

func (*Job) SetMessage

func (n *Job) SetMessage(msg string)

func (*Job) SetStatus

func (n *Job) SetStatus(status fsm.Status)

func (*Job) SetStepStatus

func (n *Job) SetStepStatus(stepName flow.TName, status fsm.Status)

func (*Job) Setup

func (n *Job) Setup(ctx *flow.Context) error

func (*Job) Teardown

func (n *Job) Teardown(ctx *flow.Context)

func (*Job) Type

func (n *Job) Type() flow.FType

type JobStep

type JobStep struct {
	*types.WorkflowJobStep
	// contains filtered or unexported fields
}

func (*JobStep) Do

func (n *JobStep) Do(ctx *flow.Context) error

func (*JobStep) GetMessage

func (n *JobStep) GetMessage() string

func (*JobStep) GetStatus

func (n *JobStep) GetStatus() fsm.Status

func (*JobStep) Name

func (n *JobStep) Name() flow.TName

func (*JobStep) SetMessage

func (n *JobStep) SetMessage(msg string)

func (*JobStep) SetStatus

func (n *JobStep) SetStatus(status fsm.Status)

func (*JobStep) Setup

func (n *JobStep) Setup(ctx *flow.Context) error

func (*JobStep) Teardown

func (n *JobStep) Teardown(ctx *flow.Context)

type Manager

type Manager interface {
	ListWorkflows(ctx context.Context) ([]*types.WorkflowSpec, error)
	GetWorkflow(ctx context.Context, wfId string) (*types.WorkflowSpec, error)
	SaveWorkflow(ctx context.Context, spec *types.WorkflowSpec) (*types.WorkflowSpec, error)
	DeleteWorkflow(ctx context.Context, wfId string) error
	ListJobs(ctx context.Context, wfId string) ([]*types.WorkflowJob, error)

	TriggerWorkflow(ctx context.Context, wfId string) (*types.WorkflowJob, error)
	PauseWorkflowJob(ctx context.Context, jobId string) error
	ResumeWorkflowJob(ctx context.Context, jobId string) error
	CancelWorkflowJob(ctx context.Context, jobId string) error
}

func NewManager

func NewManager(recorder metastore.ScheduledTaskRecorder) (Manager, error)

type MirrorPlugin

type MirrorPlugin struct {
	Manager
}

func (*MirrorPlugin) Name

func (w *MirrorPlugin) Name() string

func (*MirrorPlugin) Run

func (w *MirrorPlugin) Run(ctx context.Context, request *common.Request, params map[string]string) (*common.Response, error)

func (*MirrorPlugin) Type

func (w *MirrorPlugin) Type() types.PluginType

func (*MirrorPlugin) Version

func (w *MirrorPlugin) Version() string

type Runner

type Runner struct {
	*flowcontroller.FlowController

	sync.RWMutex
	// contains filtered or unexported fields
}

func InitWorkflowRunner

func InitWorkflowRunner(recorder metastore.ScheduledTaskRecorder) (*Runner, error)

func (*Runner) DeleteFlow

func (r *Runner) DeleteFlow(flowId flow.FID) error

func (*Runner) DeleteTask

func (r *Runner) DeleteTask(flowId flow.FID, taskName flow.TName) error

func (*Runner) GetFlow

func (r *Runner) GetFlow(flowId flow.FID) (flow.Flow, error)

func (*Runner) GetFlowMeta

func (r *Runner) GetFlowMeta(flowId flow.FID) (*flowstorage.FlowMeta, error)

func (*Runner) SaveFlow

func (r *Runner) SaveFlow(flow flow.Flow) error

func (*Runner) SaveTask

func (r *Runner) SaveTask(flowId flow.FID, task flow.Task) error

func (*Runner) Start

func (r *Runner) Start(stopCh chan struct{}) error

func (*Runner) WorkFlowHandler

func (r *Runner) WorkFlowHandler(ctx context.Context, wf *types.WorkflowSpec) (*types.WorkflowJob, error)