workflowrun

package
v0.9.5-alpha Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: May 5, 2019 License: Apache-2.0 Imports: 22 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func GCPodName added in v0.9.3

func GCPodName(wfr string) string

GCPodName generates a pod name for GC pod

func GetExecutionContext added in v0.9.3

func GetExecutionContext(wfr *v1alpha1.WorkflowRun) *v1alpha1.ExecutionContext

GetExecutionContext gets execution context from WorkflowRun, if not found, use the default context in workflow controller configuration.

func IsTrivial added in v0.9.3

func IsTrivial(wf *v1alpha1.Workflow, stage string) bool

IsTrivial returns whether a stage is trivial in a workflow

func IsWorkflowRunTerminated added in v0.9.3

func IsWorkflowRunTerminated(wfr *v1alpha1.WorkflowRun) bool

IsWorkflowRunTerminated judges whether the WorkflowRun has be terminated. Return true if terminated, otherwise return false.

func NextStages

func NextStages(wf *v1alpha1.Workflow, wfr *v1alpha1.WorkflowRun) []string

NextStages determine next stages that can be started to execute. It returns stages that are not started yet but have all depended stages finished.

func ParseTime

func ParseTime(t string) (time.Duration, error)

ParseTime parses time string like '30min', '2h30m' to time.Time

Types

type GCProcessor

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

GCProcessor processes garbage collection for WorkflowRun objects.

func NewGCProcessor

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

NewGCProcessor create new GC processor.

func (*GCProcessor) Add

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

Add WorkflowRun object to GC processor, it will firstly judge whether the WorkflowRun object needs GC, if it's true, it will perform GC on it in the right time.

func (*GCProcessor) Enable

func (p *GCProcessor) Enable()

Enable the processor and start it.

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

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

NewLimitedQueues creates a limited queues for WorkflowRuns, and start auto scan.

func (*LimitedQueues) AddOrRefresh

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

AddOrRefresh adds a WorkflowRun to its corresponding queue, if the queue size exceed the maximum size, the oldest one would be deleted. And if the WorkflowRun already exists in the queue, its 'refresh' time field would be refreshed.

func (*LimitedQueues) AutoScan

func (w *LimitedQueues) AutoScan()

AutoScan scans all WorkflowRuns in the queues regularly, remove abnormal ones with old enough refresh time.

func (*LimitedQueues) Refresh added in v0.9.3

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

Refresh refreshes the WorkflowRun in the queue, the refresh time would be updated.

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

func NewQueue(key string, max int) *LimitedSortedQueue

NewQueue creates a limited sorted queue.

func (*LimitedSortedQueue) Pop

func (q *LimitedSortedQueue) Pop() *Node

Pop pops up a WorkflowRun object from the queue, it's the oldest one that will be popped.

func (*LimitedSortedQueue) PushOrRefresh

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

PushOrRefresh pushes a WorkflowRun object to the queue, it will be inserted in the right place to keep the queue sorted by creation time. If the object already existed in the queue, its refresh time would be updated.

func (*LimitedSortedQueue) Refresh added in v0.9.3

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

Refresh updates refresh time of WorkflowRun in the queue, if the WorkflowRun found in the queue and update successfully, return true, otherwise return false.

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)
	// 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() error
}

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

func NewOperator

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

NewOperator create a new operator.

type TimeoutProcessor

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

TimeoutProcessor manages timeout of WorkflowRun.

func NewTimeoutProcessor

func NewTimeoutProcessor(client clientset.Interface) *TimeoutProcessor

NewTimeoutProcessor creates a timeout manager and run it.

func (*TimeoutProcessor) Add

Add adds a WorkflowRun to the timeout manager.

func (*TimeoutProcessor) Run

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

Run will check timeout of managed WorkflowRun and process items that have expired their time.

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 added in v0.9.3

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

NewWorkloadProcessor ...

func (*WorkloadProcessor) Process added in v0.9.3

func (p *WorkloadProcessor) Process() error

Process processes the stage according to workload type.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL