executor

Documentation

Index

• func NewASTCompiler(ctx context.Context, query string, ts CompilerBuilderTimestamps) (flux.Compiler, error)
• func NewExecutor(log *zap.Logger, qs query.QueryService, us PermissionService, ...) (*Executor, *ExecutorMetrics)
• func NewFluxCompiler(ctx context.Context, query string, ts CompilerBuilderTimestamps) (flux.Compiler, error)
• func NewRunCollector(ex *Executor) prometheus.Collector
• func WithFlagger(flagger feature.Flagger) executorOption
• func WithMaxWorkers(n int) executorOption
• func WithNonSystemCompilerBuilder(builder CompilerBuilderFunc) executorOption
• func WithSystemCompilerBuilder(builder CompilerBuilderFunc) executorOption
• type CompilerBuilderFunc
• type CompilerBuilderTimestamps
  • func (ts CompilerBuilderTimestamps) Extern() *ast.File
• type Executor
  • func (e *Executor) Cancel(ctx context.Context, runID platform.ID) error
  • func (e *Executor) Execute(ctx context.Context, id scheduler.ID, scheduledFor time.Time, runAt time.Time) error
  • func (e *Executor) ManualRun(ctx context.Context, id platform.ID, runID platform.ID) (Promise, error)
  • func (e *Executor) PromiseQueueUsage() float64
  • func (e *Executor) PromisedExecute(ctx context.Context, id scheduler.ID, scheduledFor time.Time, runAt time.Time) (Promise, error)
  • func (e *Executor) ResumeCurrentRun(ctx context.Context, id platform.ID, runID platform.ID) (Promise, error)
  • func (e *Executor) RunsActive() int
  • func (e *Executor) SetLimitFunc(l LimitFunc)
  • func (e *Executor) WorkersBusy() float64
• type ExecutorMetrics
  • func NewExecutorMetrics(ex *Executor) *ExecutorMetrics
  • func (em *ExecutorMetrics) FinishRun(task *taskmodel.Task, status taskmodel.RunStatus, runDuration time.Duration)
  • func (em *ExecutorMetrics) LogError(taskType string, err error)
  • func (em *ExecutorMetrics) LogUnrecoverableError(taskID platform.ID, err error)
  • func (em *ExecutorMetrics) PrometheusCollectors() []prometheus.Collector
  • func (em *ExecutorMetrics) StartRun(task *taskmodel.Task, queueDelta time.Duration, runLatency time.Duration)
• type LimitFunc
  • func ConcurrencyLimit(exec *Executor, lang fluxlang.FluxLanguageService) LimitFunc
  • func MultiLimit(limits ...LimitFunc) LimitFunc
• type PermissionService
• type Promise

Functions

func NewASTCompiler

func NewASTCompiler(ctx context.Context, query string, ts CompilerBuilderTimestamps) (flux.Compiler, error)

NewASTCompiler parses a Flux query string into an AST representation.

func NewExecutor

func NewExecutor(log *zap.Logger, qs query.QueryService, us PermissionService, ts taskmodel.TaskService, tcs backend.TaskControlService, opts ...executorOption) (*Executor, *ExecutorMetrics)

NewExecutor creates a new task executor

func NewFluxCompiler

func NewFluxCompiler(ctx context.Context, query string, ts CompilerBuilderTimestamps) (flux.Compiler, error)

NewFluxCompiler wraps a Flux query string in a raw-query representation.

func NewRunCollector

func NewRunCollector(ex *Executor) prometheus.Collector

NewRunCollector returns a collector which exports influxdb process metrics.

func WithFlagger

func WithFlagger(flagger feature.Flagger) executorOption

WithFlagger is an Executor option that allows us to use a feature flagger in the executor

func WithMaxWorkers

func WithMaxWorkers(n int) executorOption

WithMaxWorkers specifies the number of workers used by the Executor.

func WithNonSystemCompilerBuilder

func WithNonSystemCompilerBuilder(builder CompilerBuilderFunc) executorOption

WithNonSystemCompilerBuilder is an Executor option that configures a CompilerBuilderFunc to be used when compiling queries for non-System Tasks (Checks and Notifications).

func WithSystemCompilerBuilder

func WithSystemCompilerBuilder(builder CompilerBuilderFunc) executorOption

WithSystemCompilerBuilder is an Executor option that configures a CompilerBuilderFunc to be used when compiling queries for System Tasks.

Types

type CompilerBuilderFunc

type CompilerBuilderFunc func(ctx context.Context, query string, ts CompilerBuilderTimestamps) (flux.Compiler, error)

CompilerBuilderFunc is a function that yields a new flux.Compiler. The context.Context provided can be assumed to be an authorized context.

type CompilerBuilderTimestamps

type CompilerBuilderTimestamps struct {
	Now           time.Time
	LatestSuccess time.Time
}

CompilerBuilderTimestamps contains timestamps which should be provided along with a Task query.

func (CompilerBuilderTimestamps) Extern

func (ts CompilerBuilderTimestamps) Extern() *ast.File

type Executor

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

Executor it a task specific executor that works with the new scheduler system.

func (*Executor) Cancel

func (e *Executor) Cancel(ctx context.Context, runID platform.ID) error

Cancel a run of a specific task.

func (*Executor) Execute

func (e *Executor) Execute(ctx context.Context, id scheduler.ID, scheduledFor time.Time, runAt time.Time) error

Execute is a executor to satisfy the needs of tasks

func (*Executor) ManualRun

func (e *Executor) ManualRun(ctx context.Context, id platform.ID, runID platform.ID) (Promise, error)

func (*Executor) PromiseQueueUsage

func (e *Executor) PromiseQueueUsage() float64

PromiseQueueUsage returns the percent of the Promise Queue that is currently filled

func (*Executor) PromisedExecute

func (e *Executor) PromisedExecute(ctx context.Context, id scheduler.ID, scheduledFor time.Time, runAt time.Time) (Promise, error)

PromisedExecute begins execution for the tasks id with a specific scheduledFor time. When we execute we will first build a run for the scheduledFor time, We then want to add to the queue anything that was manually queued to run. If the queue is full the call to execute should hang and apply back pressure to the caller We then start a worker to work the newly queued jobs.

func (*Executor) ResumeCurrentRun

func (e *Executor) ResumeCurrentRun(ctx context.Context, id platform.ID, runID platform.ID) (Promise, error)

func (*Executor) RunsActive

func (e *Executor) RunsActive() int

RunsActive returns the current number of workers, which is equivalent to the number of runs actively running

func (*Executor) SetLimitFunc

func (e *Executor) SetLimitFunc(l LimitFunc)

SetLimitFunc sets the limit func for this task executor

func (*Executor) WorkersBusy

func (e *Executor) WorkersBusy() float64

WorkersBusy returns the percent of total workers that are busy

type ExecutorMetrics

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

func NewExecutorMetrics

func NewExecutorMetrics(ex *Executor) *ExecutorMetrics

func (*ExecutorMetrics) FinishRun

func (em *ExecutorMetrics) FinishRun(task *taskmodel.Task, status taskmodel.RunStatus, runDuration time.Duration)

FinishRun adjusts the metrics to indicate a run is no longer in progress for the given task ID.

func (*ExecutorMetrics) LogError

func (em *ExecutorMetrics) LogError(taskType string, err error)

LogError increments the count of errors by error code.

func (*ExecutorMetrics) LogUnrecoverableError

func (em *ExecutorMetrics) LogUnrecoverableError(taskID platform.ID, err error)

LogUnrecoverableError increments the count of unrecoverable errors, which require admin intervention to resolve or deactivate This count is separate from the errors count so that the errors metric can be used to identify only internal, rather than user errors and so that unrecoverable errors can be quickly identified for deactivation

func (*ExecutorMetrics) PrometheusCollectors

func (em *ExecutorMetrics) PrometheusCollectors() []prometheus.Collector

PrometheusCollectors satisfies the prom.PrometheusCollector interface.

func (*ExecutorMetrics) StartRun

func (em *ExecutorMetrics) StartRun(task *taskmodel.Task, queueDelta time.Duration, runLatency time.Duration)

StartRun store the delta time between when a run is due to start and actually starting.

type LimitFunc

type LimitFunc func(*taskmodel.Task, *taskmodel.Run) error

LimitFunc is a function the executor will use to

func ConcurrencyLimit

func ConcurrencyLimit(exec *Executor, lang fluxlang.FluxLanguageService) LimitFunc

ConcurrencyLimit creates a concurrency limit func that uses the executor to determine if the task has exceeded the concurrency limit.

func MultiLimit

func MultiLimit(limits ...LimitFunc) LimitFunc

MultiLimit allows us to create a single limit func that applies more then one limit.

type PermissionService

type PermissionService interface {
	FindPermissionForUser(ctx context.Context, UserID platform.ID) (influxdb.PermissionSet, error)
}

type Promise

type Promise interface {
	ID() platform.ID
	Cancel(ctx context.Context)
	Done() <-chan struct{}
	Error() error
}