evaluation

Documentation

Index

• Variables
• type BrokerStats
  • func (s *BrokerStats) IsEmpty() bool
• type InMemoryBroker
  • func NewInMemoryBroker(params InMemoryBrokerParams) (*InMemoryBroker, error)
  • func (b *InMemoryBroker) Ack(evalID, receiptHandle string) error
  • func (b *InMemoryBroker) Cancelable(batchSize int) []*models.Evaluation
  • func (b *InMemoryBroker) Dequeue(types []string, timeout time.Duration) (*models.Evaluation, string, error)
  • func (b *InMemoryBroker) Enabled() bool
  • func (b *InMemoryBroker) Enqueue(evaluation *models.Evaluation) error
  • func (b *InMemoryBroker) EnqueueAll(evals map[*models.Evaluation]string) error
  • func (b *InMemoryBroker) Inflight(evalID string) (string, bool)
  • func (b *InMemoryBroker) InflightExtend(evalID, receiptHandle string) error
  • func (b *InMemoryBroker) Nack(evalID, receiptHandle string) error
  • func (b *InMemoryBroker) SetEnabled(enabled bool)
  • func (b *InMemoryBroker) Stats() *BrokerStats
• type InMemoryBrokerParams
• type PendingEvaluations
  • func (p PendingEvaluations) Len() int
  • func (p PendingEvaluations) Less(i, j int) bool
  • func (p *PendingEvaluations) MarkForCancel() []*models.Evaluation
  • func (p *PendingEvaluations) Pop() interface{}
  • func (p *PendingEvaluations) Push(e interface{})
  • func (p PendingEvaluations) Swap(i, j int)
• type ReadyEvaluations
  • func (r ReadyEvaluations) Len() int
  • func (r ReadyEvaluations) Less(i, j int) bool
  • func (r ReadyEvaluations) Peek() *models.Evaluation
  • func (r *ReadyEvaluations) Pop() interface{}
  • func (r *ReadyEvaluations) Push(e interface{})
  • func (r ReadyEvaluations) Swap(i, j int)
• type SchedulerStats
  • func (s *SchedulerStats) IsEmpty() bool
• type WatchHandler
  • func NewWatchHandler(broker orchestrator.EvaluationBroker) *WatchHandler
  • func (h *WatchHandler) HandleEvent(ctx context.Context, event watcher.Event) error

Variables

var (
	// ErrNotInflight is returned if an evaluation is not inflight
	ErrNotInflight = errors.New("evaluation is not inflight")

	// ErrReceiptHandleMismatch is the outstanding eval has a different receiptHandle
	ErrReceiptHandleMismatch = errors.New("evaluation receiptHandle does not match")

	// ErrNackTimeoutReached is returned if an expired evaluation is reset
	ErrNackTimeoutReached = errors.New("evaluation visibility timeout reached")
)

Types

type BrokerStats

type BrokerStats struct {
	TotalReady      int
	TotalInflight   int
	TotalPending    int
	TotalWaiting    int
	TotalCancelable int
	DelayedEvals    map[string]*models.Evaluation
	ByScheduler     map[string]*SchedulerStats
}

BrokerStats returns all the stats about the broker

func (*BrokerStats) IsEmpty

func (s *BrokerStats) IsEmpty() bool

IsEmpty returns true if the stats are zero

type InMemoryBroker

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

InMemoryBroker The broker is designed to be entirely in-memory.

func NewInMemoryBroker

func NewInMemoryBroker(params InMemoryBrokerParams) (*InMemoryBroker, error)

NewInMemoryBroker creates a new evaluation broker. This is parameterized with:

• VisibilityTimeout used for messages. If not acknowledged before this time we assume a Nack and attempt to redeliver.
• MaxReceiveCount which prevents a failing eval from being endlessly delivered.
• InitialNackDelay which is the delay before making a first-time Nacked evaluation available again
• SubsequentNackDelay is the compounding delay before making evaluations available again, after the first Nack.

func (*InMemoryBroker) Ack

func (b *InMemoryBroker) Ack(evalID, receiptHandle string) error

func (*InMemoryBroker) Cancelable

func (b *InMemoryBroker) Cancelable(batchSize int) []*models.Evaluation

Cancelable retrieves a batch of previously-pending evaluations that are now stale and ready to mark for canceling. The eval RPC will call this with a batch size set to avoid sending overly large raft messages.

func (*InMemoryBroker) Dequeue

func (b *InMemoryBroker) Dequeue(types []string, timeout time.Duration) (*models.Evaluation, string, error)

func (*InMemoryBroker) Enabled

func (b *InMemoryBroker) Enabled() bool

Enabled is used to check if the broker is enabled.

func (*InMemoryBroker) Enqueue

func (b *InMemoryBroker) Enqueue(evaluation *models.Evaluation) error

func (*InMemoryBroker) EnqueueAll

func (b *InMemoryBroker) EnqueueAll(evals map[*models.Evaluation]string) error

func (*InMemoryBroker) Inflight

func (b *InMemoryBroker) Inflight(evalID string) (string, bool)

func (*InMemoryBroker) InflightExtend

func (b *InMemoryBroker) InflightExtend(evalID, receiptHandle string) error

func (*InMemoryBroker) Nack

func (b *InMemoryBroker) Nack(evalID, receiptHandle string) error

func (*InMemoryBroker) SetEnabled

func (b *InMemoryBroker) SetEnabled(enabled bool)

SetEnabled is used to control if the broker is enabled.

func (*InMemoryBroker) Stats

func (b *InMemoryBroker) Stats() *BrokerStats

Stats is used to query the state of the broker

type InMemoryBrokerParams

type InMemoryBrokerParams struct {
	VisibilityTimeout time.Duration
	MaxReceiveCount   int
	// contains filtered or unexported fields
}

type PendingEvaluations

type PendingEvaluations []*models.Evaluation

PendingEvaluations is a list of pending Evaluations for a given job. We implement the container/heap interface so that this is a priority queue.

func (PendingEvaluations) Len

func (p PendingEvaluations) Len() int

Len is for the sorting interface

func (PendingEvaluations) Less

func (p PendingEvaluations) Less(i, j int) bool

Less is for the sorting interface. We flip the check so that the "min" in the min-heap is the element with the highest priority or highest modify index

func (*PendingEvaluations) MarkForCancel

func (p *PendingEvaluations) MarkForCancel() []*models.Evaluation

MarkForCancel is used to clear the pending list of all but the one with the highest modify index and highest priority. It returns a slice of cancelable evals so that Eval.Ack RPCs can write batched raft entries to cancel them. This must be called inside the broker's lock.

func (*PendingEvaluations) Pop

func (p *PendingEvaluations) Pop() interface{}

Pop implements the heap interface and is used to remove an evaluation from the slice

func (*PendingEvaluations) Push

func (p *PendingEvaluations) Push(e interface{})

Push implements the heap interface and is used to add a new evaluation to the slice

func (PendingEvaluations) Swap

func (p PendingEvaluations) Swap(i, j int)

Swap is for the sorting interface

type ReadyEvaluations

type ReadyEvaluations []*models.Evaluation

ReadyEvaluations is a list of ready Evaluations across multiple jobs. We implement the container/heap interface so that this is a priority queue.

func (ReadyEvaluations) Len

func (r ReadyEvaluations) Len() int

Len is for the sorting interface

func (ReadyEvaluations) Less

func (r ReadyEvaluations) Less(i, j int) bool

Less is for the sorting interface. We flip the check so that the "min" in the min-heap is the element with the highest priority

func (ReadyEvaluations) Peek

func (r ReadyEvaluations) Peek() *models.Evaluation

Peek is used to peek at the next element that would be popped

func (*ReadyEvaluations) Pop

func (r *ReadyEvaluations) Pop() interface{}

Pop is used to remove an evaluation from the slice

func (*ReadyEvaluations) Push

func (r *ReadyEvaluations) Push(e interface{})

Push is used to add a new evaluation to the slice

func (ReadyEvaluations) Swap

func (r ReadyEvaluations) Swap(i, j int)

Swap is for the sorting interface

type SchedulerStats

type SchedulerStats struct {
	Ready    int
	Inflight int
}

SchedulerStats returns the stats per scheduler

func (*SchedulerStats) IsEmpty

func (s *SchedulerStats) IsEmpty() bool

IsEmpty returns true if the scheduler stats are zero

type WatchHandler

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

WatchHandler processes evaluation events from the event store and enqueues them into the evaluation broker.

func NewWatchHandler

func NewWatchHandler(broker orchestrator.EvaluationBroker) *WatchHandler

func (*WatchHandler) HandleEvent

func (h *WatchHandler) HandleEvent(ctx context.Context, event watcher.Event) error

HandleEvent processes evaluation events and enqueues new evaluations into the broker. It only processes creation events since deletions are handled by the broker itself.