orchestrator

README

go-orchestrator

Orchestrate worker nodes by splitting out the work amongst the cluster.

Documentation

Index

• type Communicator
• type Logger
• type Orchestrator
  • func New(c Communicator, opts ...OrchestratorOption) *Orchestrator
  • func (o *Orchestrator) AddTask(task interface{}, opts ...TaskOption)
  • func (o *Orchestrator) AddWorker(worker interface{})
  • func (o *Orchestrator) LastActual() []WorkerState
  • func (o *Orchestrator) ListExpectedTasks() []Task
  • func (o *Orchestrator) NextTerm(ctx context.Context)
  • func (o *Orchestrator) Rebalance(toAdd map[interface{}]int, toRemove, actual map[interface{}][]interface{}) (map[interface{}]int, map[interface{}][]interface{})
  • func (o *Orchestrator) RemoveTask(task interface{})
  • func (o *Orchestrator) RemoveWorker(worker interface{})
  • func (o *Orchestrator) UpdateTasks(tasks []Task)
  • func (o *Orchestrator) UpdateWorkers(workers []interface{})
• type OrchestratorOption
  • func WithCommunicatorTimeout(t time.Duration) OrchestratorOption
  • func WithLogger(l Logger) OrchestratorOption
  • func WithStats(f func(TermStats)) OrchestratorOption
• type Task
• type TaskOption
  • func WithTaskInstances(i int) TaskOption
• type TermStats
• type WorkerState

Types

type Communicator

type Communicator interface {
	// List returns the workload from the given worker.
	List(ctx context.Context, worker interface{}) ([]interface{}, error)

	// Add adds the given task to the worker. The error only logged (for now).
	// It is assumed that if the worker returns an error trying to update, the
	// next term will fix the problem and move the task elsewhere.
	Add(ctx context.Context, worker, task interface{}) error

	// Removes the given task from the worker. The error is only logged (for
	// now). It is assumed that if the worker is returning an error, then it
	// is either not doing the task because the worker is down, or there is a
	// network partition and a future term will fix the problem.
	Remove(ctx context.Context, worker, task interface{}) error
}

Communicator manages the intra communication between the Orchestrator and the node cluster. Each method must be safe to call on many go-routines. The given context represents the state of the term. Therefore, the Communicator is expected to cancel immediately if the context is done.

type Logger

type Logger interface {
	// Print calls l.Output to print to the logger. Arguments are handled in
	// the manner of fmt.Print.
	Printf(format string, v ...interface{})
}

Logger is used to write information.

type Orchestrator

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

Orchestrator stores the expected workload and reaches out to the cluster to see what the actual workload is. It then tries to fix the delta.

The expected task list can be altered via AddTask, RemoveTask and UpdateTasks. Each method is safe to be called on multiple go-routines.

func New

func New(c Communicator, opts ...OrchestratorOption) *Orchestrator

New creates a new Orchestrator.

func (*Orchestrator) AddTask

func (o *Orchestrator) AddTask(task interface{}, opts ...TaskOption)

AddTask adds a new task to the expected workload. The update will not take affect until the next term. It is safe to invoke AddTask, RemoveTask and UpdateTasks on multiple go-routines.

func (*Orchestrator) AddWorker

func (o *Orchestrator) AddWorker(worker interface{})

AddWorker adds a worker to the known worker cluster. The update will not take affect until the next term. It is safe to invoke AddWorker, RemoveWorkers and UpdateWorkers on multiple go-routines.

func (*Orchestrator) LastActual

func (o *Orchestrator) LastActual() []WorkerState

LastActual returns the actual from the last term. It will return nil before the first term.

func (*Orchestrator) ListExpectedTasks

func (o *Orchestrator) ListExpectedTasks() []Task

ListExpectedTasks returns the curent list of the expected tasks.

func (*Orchestrator) NextTerm

func (o *Orchestrator) NextTerm(ctx context.Context)

NextTerm reaches out to the cluster to gather to actual workload. It then attempts to fix the delta between actual and expected. The lifecycle of the term is managed by the given context.

func (*Orchestrator) Rebalance

func (o *Orchestrator) Rebalance(
	toAdd map[interface{}]int,
	toRemove,
	actual map[interface{}][]interface{},
) (map[interface{}]int, map[interface{}][]interface{})

rebalance will rebalance tasks across the workers. If any worker has too many tasks, it will be added to the remove map, and added to the returned add slice.

func (*Orchestrator) RemoveTask

func (o *Orchestrator) RemoveTask(task interface{})

RemoveTask removes a task from the expected workload. The update will not take affect until the next term. It is safe to invoke AddTask, RemoveTask and UpdateTasks on multiple go-routines.

func (*Orchestrator) RemoveWorker

func (o *Orchestrator) RemoveWorker(worker interface{})

RemoveWorker removes a worker from the known worker cluster. The update will not take affect until the next term. It is safe to invoke AddWorker, RemoveWorkers and UpdateWorkers on multiple go-routines.

func (*Orchestrator) UpdateTasks

func (o *Orchestrator) UpdateTasks(tasks []Task)

UpdateTasks overwrites the expected task list. The update will not take affect until the next term. It is safe to invoke AddTask, RemoveTask and UpdateTasks on multiple go-routines.

func (*Orchestrator) UpdateWorkers

func (o *Orchestrator) UpdateWorkers(workers []interface{})

UpdateWorkers overwrites the expected worker list. The update will not take affect until the next term. It is safe to invoke AddWorker, RemoveWorker and UpdateWorkers on multiple go-routines.

type OrchestratorOption

type OrchestratorOption func(*Orchestrator)

OrchestratorOption configures an Orchestrator.

func WithCommunicatorTimeout

func WithCommunicatorTimeout(t time.Duration) OrchestratorOption

WithCommunicatorTimeout sets the timeout for the communication to respond. Defaults to 10 seconds.

func WithLogger

func WithLogger(l Logger) OrchestratorOption

WithLogger sets the logger for the Orchestrator. Defaults to silent logger.

func WithStats

func WithStats(f func(TermStats)) OrchestratorOption

WithStats sets the stats handler for the Orchestrator. The stats handler is invoked for each term, with what the Orchestrator wrote to the Communicator.

type Task

type Task struct {
	Name      interface{}
	Instances int
}

Task stores the required information for a task.

type TaskOption

type TaskOption func(*Task)

TaskOption is used to configure a task when it is being added.

func WithTaskInstances

func WithTaskInstances(i int) TaskOption

WithTaskInstances configures the number of tasks. Defaults to 1.

type TermStats

type TermStats struct {
	// WorkerCount is the number of workers that responded without an error
	// to a List request.
	WorkerCount int
}

TermStats is the information about the last processed term. It is passed to a stats handler. See WithStats().

type WorkerState

type WorkerState struct {
	// Name is the given name of a worker.
	Name interface{}

	// Tasks is the task names the worker is servicing.
	Tasks []interface{}
}

WorkerState stores the state of a worker.