scheduler

Documentation

Index

• func AppendEventSequencesFromPreemptedJobs(eventSequences []*armadaevents.EventSequence, ...) ([]*armadaevents.EventSequence, error)
• func AppendEventSequencesFromScheduledJobs(eventSequences []*armadaevents.EventSequence, ...) ([]*armadaevents.EventSequence, error)
• func EventsFromSchedulerResult(result *scheduling.SchedulerResult, time time.Time) ([]*armadaevents.EventSequence, error)
• func JavaStringHash(s string) uint32
• func Run(config schedulerconfig.Configuration) error
• type ExecutorApi
  • func NewExecutorApi(publisher pulsarutils.Publisher[*armadaevents.EventSequence], ...) (*ExecutorApi, error)
  • func (srv *ExecutorApi) LeaseJobRuns(stream executorapi.ExecutorApi_LeaseJobRunsServer) error
  • func (srv *ExecutorApi) ReportEvents(grpcCtx context.Context, list *executorapi.EventList) (*types.Empty, error)
• type MetricsCollector
  • func NewMetricsCollector(jobDb *jobdb.JobDb, queueCache queue.QueueCache, ...) *MetricsCollector
  • func (c *MetricsCollector) Collect(metrics chan<- prometheus.Metric)
  • func (c *MetricsCollector) Describe(out chan<- *prometheus.Desc)
  • func (c *MetricsCollector) Run(ctx *armadacontext.Context) error
• type Publisher
• type PulsarPublisher
  • func NewPulsarPublisher(pulsarClient pulsar.Client, producerOptions pulsar.ProducerOptions, ...) (*PulsarPublisher, error)
  • func (p *PulsarPublisher) PublishMarkers(ctx *armadacontext.Context, groupId uuid.UUID) (uint32, error)
  • func (p *PulsarPublisher) PublishMessages(ctx *armadacontext.Context, events []*armadaevents.EventSequence, ...) error
• type Scheduler
  • func NewScheduler(jobDb *jobdb.JobDb, jobRepository database.JobRepository, ...) (*Scheduler, error)
  • func (s *Scheduler) EnableAssertions()
  • func (s *Scheduler) Run(ctx *armadacontext.Context) error
• type SubmitChecker
  • func NewSubmitChecker(schedulingConfig configuration.SchedulingConfig, ...) *SubmitChecker
  • func (srv *SubmitChecker) Check(ctx *armadacontext.Context, jobs []*jobdb.Job) (map[string]schedulingResult, error)
  • func (srv *SubmitChecker) Run(ctx *armadacontext.Context) error
• type SubmitScheduleChecker

Functions

func AppendEventSequencesFromPreemptedJobs

func AppendEventSequencesFromPreemptedJobs(eventSequences []*armadaevents.EventSequence, jctxs []*schedulercontext.JobSchedulingContext, time time.Time) ([]*armadaevents.EventSequence, error)

func AppendEventSequencesFromScheduledJobs

func AppendEventSequencesFromScheduledJobs(eventSequences []*armadaevents.EventSequence, jctxs []*schedulercontext.JobSchedulingContext) ([]*armadaevents.EventSequence, error)

func EventsFromSchedulerResult

func EventsFromSchedulerResult(result *scheduling.SchedulerResult, time time.Time) ([]*armadaevents.EventSequence, error)

EventsFromSchedulerResult generates necessary EventSequences from the provided SchedulerResult.

func JavaStringHash

func JavaStringHash(s string) uint32

JavaStringHash is the default hashing algorithm used by Pulsar copied from https://github.com/apache/pulsar-client-go/blob/master/pulsar/internal/hash.go

func Run

func Run(config schedulerconfig.Configuration) error

Run sets up a Scheduler application and runs it until a SIGTERM is received

Types

type ExecutorApi

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

ExecutorApi is the gRPC service executors use to synchronise their state with that of the scheduler.

func NewExecutorApi

func NewExecutorApi(publisher pulsarutils.Publisher[*armadaevents.EventSequence],
	jobRepository database.JobRepository,
	executorRepository database.ExecutorRepository,
	allowedPriorities []int32,
	allowedResources []string,
	nodeIdLabel string,
	priorityClassNameOverride *string,
	priorityClasses map[string]priorityTypes.PriorityClass,
	authorizer auth.ActionAuthorizer,
) (*ExecutorApi, error)

func (*ExecutorApi) LeaseJobRuns

func (srv *ExecutorApi) LeaseJobRuns(stream executorapi.ExecutorApi_LeaseJobRunsServer) error

LeaseJobRuns reconciles the state of the executor with that of the scheduler. Specifically it: 1. Stores job and capacity information received from the executor to make it available to the scheduler. 2. Notifies the executor if any of its jobs are no longer active, e.g., due to being preempted by the scheduler. 3. Transfers any jobs scheduled on this executor cluster that the executor don't already have.

func (*ExecutorApi) ReportEvents

func (srv *ExecutorApi) ReportEvents(grpcCtx context.Context, list *executorapi.EventList) (*types.Empty, error)

ReportEvents publishes all eventSequences to Pulsar. The eventSequences are compacted for more efficient publishing.

type MetricsCollector

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

MetricsCollector is a Prometheus Collector that handles scheduler metrics. The metrics themselves are calculated asynchronously every refreshPeriod

func NewMetricsCollector

func NewMetricsCollector(
	jobDb *jobdb.JobDb,
	queueCache queue.QueueCache,
	executorRepository database.ExecutorRepository,
	pools []configuration.PoolConfig,
	refreshPeriod time.Duration,
	floatingResourceTypes *floatingresources.FloatingResourceTypes,
) *MetricsCollector

func (*MetricsCollector) Collect

func (c *MetricsCollector) Collect(metrics chan<- prometheus.Metric)

Collect returns the current state of all metrics of the collector.

func (*MetricsCollector) Describe

func (c *MetricsCollector) Describe(out chan<- *prometheus.Desc)

Describe returns all descriptions of the collector.

func (*MetricsCollector) Run

func (c *MetricsCollector) Run(ctx *armadacontext.Context) error

Run enters s a loop which updates the metrics every refreshPeriod until the supplied context is cancelled

type Publisher

type Publisher interface {
	// PublishMessages will publish the supplied messages. A LeaderToken is provided and the
	// implementor may decide whether to publish based on the status of this token
	PublishMessages(ctx *armadacontext.Context, events []*armadaevents.EventSequence, shouldPublish func() bool) error

	// PublishMarkers publishes a single marker message for each Pulsar partition.  Each marker
	// massage contains the supplied group id, which allows all marker messages for a given call
	// to be identified.  The uint32 returned is the number of messages published
	PublishMarkers(ctx *armadacontext.Context, groupId uuid.UUID) (uint32, error)
}

Publisher is an interface to be implemented by structs that handle publishing messages to pulsar

type PulsarPublisher

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

PulsarPublisher is the default implementation of Publisher

func NewPulsarPublisher

func NewPulsarPublisher(
	pulsarClient pulsar.Client,
	producerOptions pulsar.ProducerOptions,
	maxEventsPerMessage int,
	maxAllowedMessageSize uint,
	sendTimeout time.Duration,
) (*PulsarPublisher, error)

func (*PulsarPublisher) PublishMarkers

func (p *PulsarPublisher) PublishMarkers(ctx *armadacontext.Context, groupId uuid.UUID) (uint32, error)

PublishMarkers sends one pulsar message (containing an armadaevents.PartitionMarker) to each partition of the producer's Pulsar topic.

func (*PulsarPublisher) PublishMessages

func (p *PulsarPublisher) PublishMessages(ctx *armadacontext.Context, events []*armadaevents.EventSequence, shouldPublish func() bool) error

PublishMessages publishes all event sequences to pulsar if shouldPublish() returns true

type Scheduler

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

Scheduler is the main Armada scheduler. It periodically performs the following cycle: 1. Update state from postgres (via the jobRepository). 2. Determine if leader and exit if not. 3. Generate any necessary eventSequences resulting from the state update. 4. Expire any jobs assigned to clusters that have timed out. 5. Schedule jobs. 6. Publish any Armada eventSequences resulting from the scheduling cycle.

func NewScheduler

func NewScheduler(
	jobDb *jobdb.JobDb,
	jobRepository database.JobRepository,
	executorRepository database.ExecutorRepository,
	schedulingAlgo scheduling.SchedulingAlgo,
	leaderController leader.LeaderController,
	publisher Publisher,
	submitChecker SubmitScheduleChecker,
	cyclePeriod time.Duration,
	schedulePeriod time.Duration,
	executorTimeout time.Duration,
	maxAttemptedRuns uint,
	nodeIdLabel string,
	metrics *metrics.Metrics,
) (*Scheduler, error)

func (*Scheduler) EnableAssertions

func (s *Scheduler) EnableAssertions()

func (*Scheduler) Run

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

Run enters the scheduling loop, which will continue until ctx is cancelled.

type SubmitChecker

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

func NewSubmitChecker

func NewSubmitChecker(
	schedulingConfig configuration.SchedulingConfig,
	executorRepository database.ExecutorRepository,
	queueCache queue.QueueCache,
	floatingResourceTypes *floatingresources.FloatingResourceTypes,
	resourceListFactory *internaltypes.ResourceListFactory,
) *SubmitChecker

func (*SubmitChecker) Check

func (srv *SubmitChecker) Check(ctx *armadacontext.Context, jobs []*jobdb.Job) (map[string]schedulingResult, error)

func (*SubmitChecker) Run

func (srv *SubmitChecker) Run(ctx *armadacontext.Context) error

type SubmitScheduleChecker

type SubmitScheduleChecker interface {
	Check(ctx *armadacontext.Context, jobs []*jobdb.Job) (map[string]schedulingResult, error)
}