workflowrun

type AttemptAction ¶ added in v1.0.0

type AttemptAction string

AttemptAction defines the action while try to run a new workflowRun

const (
	// AttemptActionStart represents the WorkflowRun can start to run
	AttemptActionStart AttemptAction = "Start"
	// AttemptActionQueued represents the WorkflowRun is queued
	AttemptActionQueued AttemptAction = "Queued"
	// AttemptActionFailed represents the WorkflowRun will fail directly due to queue full
	AttemptActionFailed AttemptAction = "Failed"
)

type EventUpdater ¶ added in v1.1.0

type EventUpdater interface {
	UpdateEvents(stage string, events []v1alpha1.StageEvent) error
}

EventUpdater update events to the stage status of corresponding workflowRun's status. Event with same name will be override.

func NewEventUpdater(client clientset.Interface, wfrNamespace, wfrName string) EventUpdater

type GCProcessor ¶

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

GCProcessor processes garbage collection for WorkflowRun objects.

func NewGCProcessor(client k8s.Interface, enabled bool) *GCProcessor

func (p *GCProcessor) Add(wfr *v1alpha1.WorkflowRun)

func (p *GCProcessor) Enable()

type LimitedQueues ¶

type LimitedQueues struct {
	// Maximum queue size, it indicates maximum number of WorkflowRuns to retain for each Workflow.
	MaxQueueSize int
	// Workflow queue map. It use Workflow name and namespace as the key, and manage Queue for each
	// Workflow.
	Queues map[string]*LimitedSortedQueue
	// k8s client used to clean old WorkflowRun
	Client clientset.Interface
}

LimitedQueues manages WorkflowRun queue for each Workflow. Queue for each Workflow is limited to a given maximum size, if new WorkflowRun created, the oldest one would be removed.

func NewLimitedQueues(client clientset.Interface, maxSize int) *LimitedQueues

func (w *LimitedQueues) AddOrRefresh(wfr *v1alpha1.WorkflowRun)

func (w *LimitedQueues) AutoScan()

func (w *LimitedQueues) Refresh(wfr *v1alpha1.WorkflowRun)

type LimitedSortedQueue ¶

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

LimitedSortedQueue is a sorted fixed length queue implemented with single linked list. Note that each queue would have a sentinel node to assist the implementation, it's a dummy node, and won't be counted in the queue size. So an empty queue would have head pointed to dummy node, with queue size 0.

func NewQueue(key string, max int) *LimitedSortedQueue

func (q *LimitedSortedQueue) Pop() *Node

func (q *LimitedSortedQueue) PushOrRefresh(wfr *v1alpha1.WorkflowRun)

func (q *LimitedSortedQueue) Refresh(wfr *v1alpha1.WorkflowRun) bool

type Node ¶

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

Node represents a WorkflowRun in the queue. The 'next' link to next node in the queue, and the 'refresh' stands the last time this node is refreshed.

'refresh' here is used to deal with a rarely occurred case: when one WorkflowRun got deleted in etcd, but workflow controller didn't get notified. Workflow controller would perform resync with etcd regularly, (5 minutes by default), during resync every WorkflowRun in the queue would be refreshed, it one WorkflowRun is deleted in etcd abnormally, it wouldn't get refreshed in the queue, so we can judge by the refresh time for this case.

When we found a node that hasn't be refreshed for a long time (for example, twice the resync period), then we remove this node from the queue.

type Operator ¶

type Operator interface {
	// Get Recorder
	GetRecorder() record.EventRecorder
	// Get WorkflowRun instance.
	GetWorkflowRun() *v1alpha1.WorkflowRun
	// Update WorkflowRun, mainly the status.
	Update() error
	// Update stage status.
	UpdateStageStatus(stage string, status *v1alpha1.Status)
	// Update stage pod info.
	UpdateStagePodInfo(stage string, podInfo *v1alpha1.PodInfo)
	// Update stage outputs, they are key-value results from stage execution
	UpdateStageOutputs(stage string, keyValues []v1alpha1.KeyValue)
	// Decide overall status of the WorkflowRun from stage status.
	OverallStatus() (*v1alpha1.Status, error)
	// Garbage collection on the WorkflowRun based on GC policy configured
	// in Workflow Controller. Pod and data on PV would be cleaned.
	// - 'lastTry' indicates whether this is the last time to perform GC,
	// if set to true, the WorkflowRun status will be marked as cleaned regardless
	// whether the GC action succeeded or not.
	// - 'wfrDeletion' indicates whether the GC is performed because of WorkflowRun deleted.
	GC(lastTry, wfrDeletion bool) error
	// Run next stages in the Workflow and resolve overall status.
	Reconcile() (controller.Result, error)
	// ResolveGlobalVariables resolves global variables from workflow.
	ResolveGlobalVariables()
}

Operator is used to perform operations on a WorkflowRun instance, such as update status, run next stages, garbage collection, etc.

func NewOperator(clusterClient kubernetes.Interface, client k8s.Interface, wfr interface{}, namespace string) (Operator, error)

type ParallelismController ¶ added in v1.0.0

type ParallelismController interface {
	// AttemptNew tries to run a new WorkflowRun, and returns the corresponding action.
	AttemptNew(ns, wf, wfr string) AttemptAction
	// MarkFinished mark a WorkflowRun execution finished
	MarkFinished(ns, wf, wfr string)
}

ParallelismController is an interface to manage parallelism of WorkflowRun executions

func NewParallelismController(parallelismConfig *controller.ParallelismConfig) ParallelismController

type PodEventWatcher ¶ added in v1.1.0

type PodEventWatcher interface {
	// Invoke Work in background goroutine, this is a blocking method.
	Work(stage, podNamespace, podName string)
}

PodEventWatcher watches Kubernetes events for a pod and records Warning type events to WorkflowRun status.

type TimeoutProcessor ¶

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

TimeoutProcessor manages timeout of WorkflowRun.

func NewTimeoutProcessor(client k8s.Interface) *TimeoutProcessor

func (m *TimeoutProcessor) AddIfNotExist(wfr *v1alpha1.WorkflowRun) error

func (m *TimeoutProcessor) Run(interval time.Duration)

type WorkloadProcessor ¶ added in v0.9.3

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

WorkloadProcessor processes stage workload. There are kinds of workload supported: pod, delegation. With pod, Cyclone would create a pod to run the stage. With delegation, Cyclone would send a POST request to the given URL in the workload spec.

func NewWorkloadProcessor(clusterClient kubernetes.Interface, client k8s.Interface, wf *v1alpha1.Workflow, wfr *v1alpha1.WorkflowRun, stage *v1alpha1.Stage, wfrOperator Operator) *WorkloadProcessor

func (p *WorkloadProcessor) Process() error