controller

Documentation

Index

• Constants
• type AdmissionController
  • func NewAdmissionController(invocations *store.Invocations, api *api.Invocation, maxParallel int) *AdmissionController
  • func (c *AdmissionController) Run()
• type InvocationController
  • func NewInvocationController(invocationID string, executor *executor.LocalExecutor, ...) *InvocationController
  • func (c *InvocationController) Eval(ctx context.Context, processValue *ctrl.Event) ctrl.Result
• type InvocationMetaController
  • func NewInvocationMetaController(executor *executor.LocalExecutor, invocations *store.Invocations, ...) *InvocationMetaController
  • func (c *InvocationMetaController) Close() error
  • func (c *InvocationMetaController) Run()
• type InvocationNotificationSensor
  • func NewInvocationNotificationSensor(invocations *store.Invocations) *InvocationNotificationSensor
  • func (s *InvocationNotificationSensor) Close() error
  • func (s *InvocationNotificationSensor) Run(evalQueue ctrl.EvalQueue)
  • func (s *InvocationNotificationSensor) Start(evalQueue ctrl.EvalQueue) error
• type InvocationStorePollSensor
  • func NewInvocationStorePollSensor(invocations *store.Invocations, interval time.Duration, api *api.Invocation) *InvocationStorePollSensor
  • func (s *InvocationStorePollSensor) Poll(evalQueue ctrl.EvalQueue)
• type QueuedInvocationPollSensor
  • func NewQueuedInvocationPollSensor(invocations *store.Invocations, api *api.Invocation, interval time.Duration, ...) *QueuedInvocationPollSensor
  • func (s *QueuedInvocationPollSensor) Poll(ctrl.EvalQueue)
• type QueuedInvocationSensor
  • func NewQueuedInvocationSensor(invocations *store.Invocations, api *api.Invocation, ...) *QueuedInvocationSensor
  • func (s *QueuedInvocationSensor) Close() error
  • func (s *QueuedInvocationSensor) Run(ctrl.EvalQueue)
  • func (s *QueuedInvocationSensor) Start(eq ctrl.EvalQueue) error
• type StalenessPollSensor
  • func NewStalenessPollSensor(system *ctrl.System, ...) *StalenessPollSensor
  • func (s *StalenessPollSensor) Poll(queue ctrl.EvalQueue)
• type WorkflowController
  • func NewWorkflowController(api *api.Workflow, executor *executor.LocalExecutor, workflowID string) *WorkflowController
  • func (c *WorkflowController) Eval(ctx context.Context, processValue *ctrl.Event) ctrl.Result
• type WorkflowMetaController
  • func NewWorkflowMetaController(api *api.Workflow, workflows *store.Workflows, ...) *WorkflowMetaController
  • func (c *WorkflowMetaController) Close() error
  • func (c *WorkflowMetaController) Run()
• type WorkflowNotificationSensor
  • func NewWorkflowNotificationSensor(workflows *store.Workflows) *WorkflowNotificationSensor
  • func (s *WorkflowNotificationSensor) Close() error
  • func (s *WorkflowNotificationSensor) Run(evalQueue ctrl.EvalQueue)
  • func (s *WorkflowNotificationSensor) Start(evalQueue ctrl.EvalQueue) error
• type WorkflowStorePollSensor
  • func NewWorkflowStorePollSensor(workflows *store.Workflows, interval time.Duration) *WorkflowStorePollSensor
  • func (s *WorkflowStorePollSensor) Poll(evalQueue ctrl.EvalQueue)

Constants

const (
	DefaultMaxRuntime = 10 * time.Minute

	InvocationControllerLabel = "invocation"
)

const (
	EventRefresh = "refresh"

	WorkflowControllerType = "workflow"
)

const (
	POLL_INTERVAL = time.Millisecond * 500
)

Types

type AdmissionController

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

func NewAdmissionController

func NewAdmissionController(invocations *store.Invocations, api *api.Invocation, maxParallel int) *AdmissionController

func (*AdmissionController) Run

func (c *AdmissionController) Run()

type InvocationController

type InvocationController struct {
	StateStore *expr.Store // Future: just grab the initial state of the parent, instead of constantly rebuilding it.
	// contains filtered or unexported fields
}

InvocationController is the controller for ensuring the processing of a single workflow invocation.

func NewInvocationController

func NewInvocationController(invocationID string, executor *executor.LocalExecutor, invocationAPI *api.Invocation,
	taskAPI *api.Task, scheduler *scheduler.InvocationScheduler, stateStore *expr.Store,
	span opentracing.Span, logger *logrus.Entry, consentAPI *api.Consent, provAPI *api.Provenance) *InvocationController

func (*InvocationController) Eval

func (c *InvocationController) Eval(ctx context.Context, processValue *ctrl.Event) ctrl.Result

type InvocationMetaController

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

InvocationMetaController is the component responsible for the full integration of the invocations reconciliation loop.

Specifically, the meta-controller is responsible for the following: - It starts or registers the sensors to ensure that events are routed to this control system. - It manages all of the workflow controllers. - It provides an executor pool for controllers to submit their tasks to.

func NewInvocationMetaController

func NewInvocationMetaController(executor *executor.LocalExecutor, invocations *store.Invocations,
	invocationAPI *api.Invocation, taskAPI *api.Task, scheduler *scheduler.InvocationScheduler, stateStore *expr.Store,
	cachePollInterval time.Duration, consentAPI *api.Consent, provAPI *api.Provenance) *InvocationMetaController

func (*InvocationMetaController) Close

func (c *InvocationMetaController) Close() error

func (*InvocationMetaController) Run

func (c *InvocationMetaController) Run()

type InvocationNotificationSensor

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

InvocationNotificationSensor watches the invocations store notifications for workflow events.

func NewInvocationNotificationSensor

func NewInvocationNotificationSensor(invocations *store.Invocations) *InvocationNotificationSensor

func (*InvocationNotificationSensor) Close

func (s *InvocationNotificationSensor) Close() error

func (*InvocationNotificationSensor) Run

func (s *InvocationNotificationSensor) Run(evalQueue ctrl.EvalQueue)

func (*InvocationNotificationSensor) Start

func (s *InvocationNotificationSensor) Start(evalQueue ctrl.EvalQueue) error

type InvocationStorePollSensor

type InvocationStorePollSensor struct {
	*ctrl.PollSensor
	// contains filtered or unexported fields
}

InvocationStorePollSensor polls the invocations store on a set interval.

func NewInvocationStorePollSensor

func NewInvocationStorePollSensor(invocations *store.Invocations, interval time.Duration, api *api.Invocation) *InvocationStorePollSensor

func (*InvocationStorePollSensor) Poll

func (s *InvocationStorePollSensor) Poll(evalQueue ctrl.EvalQueue)

type QueuedInvocationPollSensor

type QueuedInvocationPollSensor struct {
	*ctrl.PollSensor
	// contains filtered or unexported fields
}

func NewQueuedInvocationPollSensor

func NewQueuedInvocationPollSensor(invocations *store.Invocations, api *api.Invocation,
	interval time.Duration, monitor *monitor.InvocationMonitor, maxParallel int) *QueuedInvocationPollSensor

func (*QueuedInvocationPollSensor) Poll

func (s *QueuedInvocationPollSensor) Poll(ctrl.EvalQueue)

type QueuedInvocationSensor

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

func NewQueuedInvocationSensor

func NewQueuedInvocationSensor(invocations *store.Invocations, api *api.Invocation, monitor *monitor.InvocationMonitor, maxParallel int) *QueuedInvocationSensor

func (*QueuedInvocationSensor) Close

func (s *QueuedInvocationSensor) Close() error

func (*QueuedInvocationSensor) Run

func (s *QueuedInvocationSensor) Run(ctrl.EvalQueue)

func (*QueuedInvocationSensor) Start

func (s *QueuedInvocationSensor) Start(eq ctrl.EvalQueue) error

type StalenessPollSensor

type StalenessPollSensor struct {
	*ctrl.PollSensor
	// contains filtered or unexported fields
}

func NewStalenessPollSensor

func NewStalenessPollSensor(system *ctrl.System, stateFetcher func(ctrlKey string) (fes.Aggregate, fes.Entity, error),
	interval time.Duration, maxStaleness time.Duration) *StalenessPollSensor

func (*StalenessPollSensor) Poll

func (s *StalenessPollSensor) Poll(queue ctrl.EvalQueue)

type WorkflowController

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

WorkflowController is the controller for ensuring the processing of a single workflow.

func NewWorkflowController

func NewWorkflowController(api *api.Workflow, executor *executor.LocalExecutor, workflowID string) *WorkflowController

func (*WorkflowController) Eval

func (c *WorkflowController) Eval(ctx context.Context, processValue *ctrl.Event) ctrl.Result

type WorkflowMetaController

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

WorkflowMetaController is the component responsible for the full integration of the workflows reconciliation loop.

Specifically, the meta-controller is responsible for the following: - It starts or registers the sensors to ensure that events are routed to this control system. - It manages all of the workflow controllers. - It provides an executor pool for controllers to submit their tasks to.

func NewWorkflowMetaController

func NewWorkflowMetaController(api *api.Workflow, workflows *store.Workflows, executor *executor.LocalExecutor,
	storePollInterval time.Duration) *WorkflowMetaController

func (*WorkflowMetaController) Close

func (c *WorkflowMetaController) Close() error

func (*WorkflowMetaController) Run

func (c *WorkflowMetaController) Run()

type WorkflowNotificationSensor

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

WorkflowNotificationSensor watches the workflow store notifications for workflow events.

func NewWorkflowNotificationSensor

func NewWorkflowNotificationSensor(workflows *store.Workflows) *WorkflowNotificationSensor

func (*WorkflowNotificationSensor) Close

func (s *WorkflowNotificationSensor) Close() error

func (*WorkflowNotificationSensor) Run

func (s *WorkflowNotificationSensor) Run(evalQueue ctrl.EvalQueue)

func (*WorkflowNotificationSensor) Start

func (s *WorkflowNotificationSensor) Start(evalQueue ctrl.EvalQueue) error

type WorkflowStorePollSensor

type WorkflowStorePollSensor struct {
	*ctrl.PollSensor
	// contains filtered or unexported fields
}

WorkflowStorePollSensor polls the workflows store on a set interval.

func NewWorkflowStorePollSensor

func NewWorkflowStorePollSensor(workflows *store.Workflows, interval time.Duration) *WorkflowStorePollSensor

func (*WorkflowStorePollSensor) Poll

func (s *WorkflowStorePollSensor) Poll(evalQueue ctrl.EvalQueue)