scheduler

Documentation

Index

• Constants
• Variables
• func AvailableResources(tasks []*tes.Task, res *pbs.Resources) *pbs.Resources
• func GenNodeID(prefix string) string
• func Match(node *pbs.Node, task *tes.Task, predicates []Predicate) bool
• func NoopWorkerFactory(c config.Worker, taskID string, log *logger.Logger) (worker.Worker, error)
• func NotDead(j *tes.Task, n *pbs.Node) error
• func ResourcesFit(t *tes.Task, n *pbs.Node) error
• func SortByAverageScore(offers []*Offer)
• func SubtractResources(t *tes.Task, in *pbs.Resources) *pbs.Resources
• func UpdateNode(ctx context.Context, cli tes.TaskServiceServer, node, existing *pbs.Node) error
• func UpdateNodeState(nodes []*pbs.Node, conf config.Scheduler) []*pbs.Node
• func ZonesFit(t *tes.Task, n *pbs.Node) error
• type Backend
• type Client
  • func NewClient(conf config.Scheduler) (Client, error)
• type ComputeBackend
  • func NewComputeBackend(db Database) *ComputeBackend
  • func (b *ComputeBackend) Submit(task *tes.Task) error
• type Database
• type Node
  • func NewNode(conf config.Config, log *logger.Logger, factory WorkerFactory) (*Node, error)
  • func NewNoopNode(conf config.Config, log *logger.Logger) (*Node, error)
  • func (n *Node) Run(ctx context.Context)
• type Offer
  • func DefaultScheduleAlgorithm(j *tes.Task, nodes []*pbs.Node, weights map[string]float32) *Offer
  • func NewOffer(n *pbs.Node, t *tes.Task, s Scores) *Offer
• type Predicate
  • func NodeHasTag(tag string) Predicate
• type Scaler
• type Scheduler
  • func (s *Scheduler) CheckNodes() error
  • func (s *Scheduler) Run(ctx context.Context) error
  • func (s *Scheduler) Scale(ctx context.Context) error
  • func (s *Scheduler) Schedule(ctx context.Context) error
• type Scores
  • func DefaultScores(w *pbs.Node, t *tes.Task) Scores
  • func (s Scores) Average() float32
  • func (s Scores) Weighted(w map[string]float32) Scores
• type WorkerFactory

Constants

const (
	CPU = "cpu"
	RAM = "ram"
)

Scores keys

Variables

var DefaultPredicates = []Predicate{
	ResourcesFit,
	ZonesFit,
	NotDead,
}

DefaultPredicates is a list of Predicate functions that check the whether a task fits a node.

Functions

func AvailableResources

func AvailableResources(tasks []*tes.Task, res *pbs.Resources) *pbs.Resources

AvailableResources calculates available resources given a list of tasks and base resources.

func GenNodeID

func GenNodeID(prefix string) string

GenNodeID returns a UUID string.

func Match

func Match(node *pbs.Node, task *tes.Task, predicates []Predicate) bool

Match checks whether a task fits a node using the given Predicate list.

func NoopWorkerFactory

func NoopWorkerFactory(c config.Worker, taskID string, log *logger.Logger) (worker.Worker, error)

NoopWorkerFactory returns a new NoopWorker.

func NotDead

func NotDead(j *tes.Task, n *pbs.Node) error

NotDead returns true if the node state is not Dead or Gone.

func ResourcesFit

func ResourcesFit(t *tes.Task, n *pbs.Node) error

ResourcesFit determines whether a task fits a node's resources.

func SortByAverageScore

func SortByAverageScore(offers []*Offer)

SortByAverageScore sorts the given offers by their average score. This modifies the offers list in place.

func SubtractResources

func SubtractResources(t *tes.Task, in *pbs.Resources) *pbs.Resources

SubtractResources subtracts the resources requested by "task" from the node resources "in".

func UpdateNode

func UpdateNode(ctx context.Context, cli tes.TaskServiceServer, node, existing *pbs.Node) error

UpdateNode helps scheduler database backend update a node when PutNode() is called.

func UpdateNodeState

func UpdateNodeState(nodes []*pbs.Node, conf config.Scheduler) []*pbs.Node

UpdateNodeState checks whether a node is dead/gone based on the last time it pinged.

func ZonesFit

func ZonesFit(t *tes.Task, n *pbs.Node) error

ZonesFit determines whether a task's zones fit a node.

Types

type Backend

type Backend interface {
	// GetOffer returns an offer for a task.
	GetOffer(*tes.Task) *Offer
}

Backend is responsible for scheduling a task. It has a single method which is responsible for taking a Task and returning an Offer, or nil if there is no node matching the task request. An Offer includes the ID of the offered node.

Offers include scores which describe how well the task fits the node. Scores may describe a wide variety of metrics: resource usage, packing, startup time, cost, etc. Scores and weights are used to control the behavior of schedulers, and to combine offers from multiple schedulers.

type Client

type Client interface {
	events.EventServiceClient
	pbs.SchedulerServiceClient
	Close()
}

Client is a client for the scheduler and event gRPC services.

func NewClient

func NewClient(conf config.Scheduler) (Client, error)

NewClient returns a new Client instance connected to the scheduler and task logger services at a given address (e.g. "localhost:9090")

type ComputeBackend

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

ComputeBackend represents the funnel scheduler backend.

func NewComputeBackend

func NewComputeBackend(db Database) *ComputeBackend

NewComputeBackend returns a new scheduler ComputeBackend instance.

func (*ComputeBackend) Submit

func (b *ComputeBackend) Submit(task *tes.Task) error

Submit submits a task via gRPC call to the funnel scheduler backend

type Database

type Database interface {
	QueueTask(*tes.Task) error
	ReadQueue(int) []*tes.Task
	ListNodes(context.Context, *pbs.ListNodesRequest) (*pbs.ListNodesResponse, error)
	PutNode(context.Context, *pbs.Node) (*pbs.PutNodeResponse, error)
	DeleteNode(context.Context, *pbs.Node) error
	WriteContext(context.Context, *events.Event) error
}

Database represents the interface to the database used by the scheduler, scaler, etc. Mostly, this exists so it can be mocked during testing.

type Node

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

Node is a structure used for tracking available resources on a compute resource.

func NewNode

func NewNode(conf config.Config, log *logger.Logger, factory WorkerFactory) (*Node, error)

NewNode returns a new Node instance

func NewNoopNode

func NewNoopNode(conf config.Config, log *logger.Logger) (*Node, error)

NewNoopNode returns a new node that doesn't have any side effects (e.g. storage access, docker calls, etc.) which is useful for testing.

func (*Node) Run

func (n *Node) Run(ctx context.Context)

Run runs a node with the given config. This is responsible for communication with the server and starting task workers

type Offer

type Offer struct {
	TaskID string
	Node   *pbs.Node
	Scores Scores
}

Offer describes a node offered by a scheduler for a task. The Scores describe how well the task fits this node, which could be used by other a scheduler to pick the best offer.

func DefaultScheduleAlgorithm

func DefaultScheduleAlgorithm(j *tes.Task, nodes []*pbs.Node, weights map[string]float32) *Offer

DefaultScheduleAlgorithm implements a simple scheduling algorithm that is (currently) common across a few scheduler backends. Given a task, list of nodes, and weights, it returns the best Offer or nil.

func NewOffer

func NewOffer(n *pbs.Node, t *tes.Task, s Scores) *Offer

NewOffer returns a new Offer instance.

type Predicate

type Predicate func(*tes.Task, *pbs.Node) error

Predicate is a function that checks whether a task fits a node.

func NodeHasTag

func NodeHasTag(tag string) Predicate

NodeHasTag returns a predicate function which returns true if the node has the given tag (key in Metadata field).

type Scaler

type Scaler interface {
	// StartNode is where the work is done to start a node instance,
	// for example, calling out to Google Cloud APIs.
	StartNode(*pbs.Node) error
	// ShouldStartNode allows scalers to filter out nodes they are interested in.
	// If "true" is returned, Scaler.StartNode() will be called with this Node.
	ShouldStartNode(*pbs.Node) bool
}

Scaler represents a service that can start node instances, for example the Google Cloud Scheduler backend.

type Scheduler

type Scheduler struct {
	Log     *logger.Logger
	DB      Database
	Conf    config.Scheduler
	Backend Backend
}

Scheduler handles scheduling tasks to nodes and support many backends.

func (*Scheduler) CheckNodes

func (s *Scheduler) CheckNodes() error

CheckNodes is used by the scheduler to check for dead/gone nodes. This is not an RPC endpoint

func (*Scheduler) Run

func (s *Scheduler) Run(ctx context.Context) error

Run starts the scheduling loop. This blocks.

The scheduler will take a chunk of tasks from the queue, request the the configured backend schedule them, and act on offers made by the backend.

func (*Scheduler) Scale

func (s *Scheduler) Scale(ctx context.Context) error

Scale implements some common logic for allowing scheduler backends to poll the database, looking for nodes that need to be started and shutdown.

func (*Scheduler) Schedule

func (s *Scheduler) Schedule(ctx context.Context) error

Schedule does a scheduling iteration. It checks the health of nodes in the database, gets a chunk of tasks from the queue (configurable by config.Scheduler.ScheduleChunk), and calls the given scheduler backend. If the backend returns a valid offer, the task is assigned to the offered node.

type Scores

type Scores map[string]float32

Scores describe how well a task fits a node.

func DefaultScores

func DefaultScores(w *pbs.Node, t *tes.Task) Scores

DefaultScores returns a default set of scores.

func (Scores) Average

func (s Scores) Average() float32

Average returns the average of the scores.

func (Scores) Weighted

func (s Scores) Weighted(w map[string]float32) Scores

Weighted returns a new Scores instance with each score multiplied by the given weights. Weights default to 0.0

type WorkerFactory

type WorkerFactory func(c config.Worker, taskID string, log *logger.Logger) (worker.Worker, error)

WorkerFactory is a function which creates a new worker instance.