scheduler

package

Go to main page

Versions in this module

v0

v0.3.45

Jan 13, 2023

Changes in this version

type JobDb

+ func (jobDb *JobDb) BatchDelete(txn *memdb.Txn, ids []string) error

+ func (jobDb *JobDb) GetAll(txn *memdb.Txn) ([]*SchedulerJob, error)

+ func (jobDb *JobDb) GetById(txn *memdb.Txn, id string) (*SchedulerJob, error)

+ func (jobDb *JobDb) ReadTxn() *memdb.Txn

+ func (jobDb *JobDb) WriteTxn() *memdb.Txn

+ type JobIterator interface

+ Next func() (T, error)

+ type JobRepository interface

+ GetExistingJobsByIds func(ids []string) ([]*api.Job, error)

+ GetQueueJobIds func(queueName string) ([]string, error)

+ type JobRepositoryAdapter struct

+ func NewJobRepositoryAdapter(jobRepo repository.JobRepository) *JobRepositoryAdapter

+ func (j *JobRepositoryAdapter) GetJobIterator(ctx context.Context, queue string) (JobIterator[*api.Job], error)

+ func (j *JobRepositoryAdapter) TryLeaseJobs(clusterId string, queue string, jobs []*api.Job) ([]*api.Job, error)

+ type JobRun struct

+ Cancelled bool

+ Executor string

+ Expired bool

+ Failed bool

+ Pending bool

+ Returned bool

+ RunID uuid.UUID

+ Running bool

+ Succeeded bool

+ func (run *JobRun) DeepCopy() *JobRun

+ func (run *JobRun) InTerminalState() bool

+ type LeaderController interface

+ GetToken func() LeaderToken

+ ValidateToken func(tok LeaderToken) bool

+ type LeaderToken struct

+ func InvalidLeaderToken() LeaderToken

+ func NewLeaderToken() LeaderToken

+ type LegacySchedulerJob interface

+ GetAnnotations func() map[string]string

+ GetId func() string

+ GetQueue func() string

+ type Publisher interface

+ PublishMarkers func(ctx context.Context, groupId uuid.UUID) (uint32, error)

+ PublishMessages func(ctx context.Context, events []*armadaevents.EventSequence, token LeaderToken) error

type SchedulerJob

+ CancelRequested bool

+ Cancelled bool

+ Executor string

+ Failed bool

+ Jobset string

+ Node string

+ Queued bool

+ Runs []*JobRun

+ Succeeded bool

+ func (job *SchedulerJob) CurrentRun() *JobRun

+ func (job *SchedulerJob) DeepCopy() *SchedulerJob

+ func (job *SchedulerJob) GetAnnotations() map[string]string

+ func (job *SchedulerJob) GetId() string

+ func (job *SchedulerJob) GetQueue() string

+ func (job *SchedulerJob) InTerminalState() bool

+ func (job *SchedulerJob) NumReturned() uint

+ func (job *SchedulerJob) RunById(id uuid.UUID) *JobRun

type SchedulerJobRepository

+ GetJobIterator func(ctx context.Context, queue string) (JobIterator[T], error)

+ type SchedulingAlgo interface

+ Schedule func(txn *memdb.Txn, jobDb *JobDb) ([]*SchedulerJob, error)

+ type StandaloneLeaderController struct

+ func NewStandaloneLeaderController() *StandaloneLeaderController

+ func (lc *StandaloneLeaderController) GetToken() LeaderToken

+ func (lc *StandaloneLeaderController) ValidateToken(tok LeaderToken) bool

v0.3.44

Dec 14, 2022

v0.3.43

Dec 6, 2022

Changes in this version

+ type Configuration struct

+ BatchDuration time.Duration

+ Metrics configuration.MetricsConfig

+ Postgres configuration.PostgresConfig

+ Pulsar configuration.PulsarConfig

+ PulsarBackoffTime time.Duration

+ PulsarReceiveTimeout time.Duration

+ SubscriptionName string

v0.3.42

Nov 24, 2022

v0.3.41

Nov 23, 2022

v0.3.40

Nov 22, 2022

Changes in this version

+ func GangIdAndCardinalityFromAnnotations(annotations map[string]string, ...) (string, int, bool, error)

+ func GangIdAndCardinalityFromJob(job *api.Job, gangIdAnnotation, gangCardinalityAnnotation string) (string, int, bool, error)

+ func PodRequirementsFromJob(priorityClasses map[string]configuration.PriorityClass, job *api.Job) []*schedulerobjects.PodRequirements

+ func PodRequirementsFromJobs(priorityClasses map[string]configuration.PriorityClass, jobs []*api.Job) []*schedulerobjects.PodRequirements

+ type CandidateGangIterator struct

+ SchedulingRoundReport *SchedulingRoundReport

+ func (it *CandidateGangIterator) Next() ([]*JobSchedulingReport, error)

type LegacyScheduler

+ CandidateGangIterator *CandidateGangIterator

+ GangCardinalityAnnotation string

+ GangIdAnnotation string

type NodeDb

+ func (nodeDb *NodeDb) ClearAllocated() error

+ func (nodeDb *NodeDb) ScheduleMany(reqs []*schedulerobjects.PodRequirements) ([]*PodSchedulingReport, bool, error)

+ func (nodeDb *NodeDb) ScheduleManyWithTxn(txn *memdb.Txn, reqs []*schedulerobjects.PodRequirements) ([]*PodSchedulingReport, bool, error)

+ func (nodeDb *NodeDb) SelectAndBindNodeToPodWithTxn(txn *memdb.Txn, req *schedulerobjects.PodRequirements) (*PodSchedulingReport, error)

+ func (nodeDb *NodeDb) SelectNodeForPodWithTxn(txn *memdb.Txn, req *schedulerobjects.PodRequirements) (*PodSchedulingReport, error)

+ func (nodeDb *NodeDb) TimeOfMostRecentUpsert() time.Time

+ func (nodeDb *NodeDb) Txn(write bool) *memdb.Txn

+ type NodesIterator struct

+ func NewNodesIterator(txn *memdb.Txn) (*NodesIterator, error)

+ func (it *NodesIterator) Next() interface{}

+ func (it *NodesIterator) NextNode() *schedulerobjects.Node

+ func (it *NodesIterator) WatchCh() <-chan struct{}

+ type QueueCandidateGangIterator struct

+ QueueSchedulingRoundReport *QueueSchedulingRoundReport

+ func (it *QueueCandidateGangIterator) Next() ([]*JobSchedulingReport, error)

+ type QueueSchedulingRoundReport struct

+ InitialResourcesByPriority schedulerobjects.QuantityByPriorityAndResourceType

+ NumScheduledJobs int

+ PriorityFactor float64

+ ScheduledResourcesByPriority schedulerobjects.QuantityByPriorityAndResourceType

+ SuccessfulJobSchedulingReports map[uuid.UUID]*JobSchedulingReport

+ UnsuccessfulJobSchedulingReports map[uuid.UUID]*JobSchedulingReport

+ func NewQueueSchedulingRoundReport(priorityFactor float64, ...) *QueueSchedulingRoundReport

+ func (report *QueueSchedulingRoundReport) AddJobSchedulingReport(r *JobSchedulingReport)

+ func (report *QueueSchedulingRoundReport) ClearJobSpecs()

+ type QueuedGangIterator struct

+ func NewQueuedGangIterator(ctx context.Context, it *QueuedJobsIterator, ...) *QueuedGangIterator

+ func (it *QueuedGangIterator) Next() ([]*api.Job, error)

+ type SchedulingConstraints struct

+ ExecutorId string

+ MaxConsecutiveUnschedulableJobs uint

+ MaximalCumulativeResourceFractionPerQueueAndPriority map[int32]map[string]float64

+ MaximalResourceFractionPerQueue map[string]float64

+ MaximalResourceFractionToSchedule map[string]float64

+ MaximalResourceFractionToSchedulePerQueue map[string]float64

+ MaximumJobsToSchedule uint

+ MinimumJobSize schedulerobjects.ResourceList

+ Pool string

+ Priorities []int32

+ PriorityClasses map[string]configuration.PriorityClass

+ ResourceScarcity map[string]float64

+ TotalResources schedulerobjects.ResourceList

+ func SchedulingConstraintsFromSchedulingConfig(executorId, pool string, minimumJobSize schedulerobjects.ResourceList, ...) *SchedulingConstraints

type SchedulingReportsRepository

+ func (repo *SchedulingReportsRepository) AddMany(queueName string, reports []*JobSchedulingReport)

+ func (repo *SchedulingReportsRepository) AddSchedulingRoundReport(report *SchedulingRoundReport)

type SchedulingRoundReport

+ NumScheduledJobs int

+ QueueSchedulingRoundReports map[string]*QueueSchedulingRoundReport

+ ScheduledResourcesByPriority schedulerobjects.QuantityByPriorityAndResourceType

+ type SubmitChecker struct

+ func NewSubmitChecker(executorTimeout time.Duration, ...) *SubmitChecker

+ func (srv *SubmitChecker) Check(reqs []*schedulerobjects.PodRequirements) (bool, string)

+ func (srv *SubmitChecker) CheckApiJobs(jobs []*api.Job) (bool, string)

+ func (srv *SubmitChecker) RegisterNodeDb(executor string, nodeDb *NodeDb)

v0.3.39

Nov 3, 2022

Changes in this version

+ type JobDb struct

+ Db *memdb.MemDB

+ func NewJobDb() (*JobDb, error)

+ func (jobDb *JobDb) Upsert(jobs []*SchedulerJob) error

+ type JobQueueIterator struct

+ func NewJobQueueIterator(txn *memdb.Txn, queue string) (*JobQueueIterator, error)

+ func (it *JobQueueIterator) Next() interface{}

+ func (it *JobQueueIterator) NextJobItem() *SchedulerJob

+ func (it *JobQueueIterator) WatchCh() <-chan struct{}

+ type SchedulerJob struct

+ JobId string

+ NegatedPriorityClassValue int

+ Priority uint32

+ Queue string

+ Timestamp int64

v0.3.38

Oct 31, 2022

v0.3.37

Oct 28, 2022

Changes in this version

type LegacyScheduler

+ CumulativePriorityLimits map[int32]map[string]float64

v0.3.36

Oct 19, 2022

Changes in this version

type LegacyScheduler

+ Pool string

v0.3.35

Oct 18, 2022

Changes in this version

+ func NodeTypesMatchingPod(nodeTypes map[string]*schedulerobjects.NodeType, ...) ([]*schedulerobjects.NodeType, map[string]int, error)

+ func PriorityFromJob(job *api.Job, priorityByPriorityClassName map[string]int32) (priority int32, ok bool)

+ func ResourceListAsWeightedApproximateFloat64(resourceScarcity map[string]float64, rl schedulerobjects.ResourceList) float64

+ func WeightsFromAggregatedUsageByQueue(resourceScarcity map[string]float64, priorityFactorByQueue map[string]float64, ...) map[string]float64

+ type JobSchedulingReport struct

+ ExecutorId string

+ Job *api.Job

+ JobId uuid.UUID

+ PodSchedulingReports []*PodSchedulingReport

+ Req *schedulerobjects.PodRequirements

+ RoundQueueResources schedulerobjects.ResourceList

+ Timestamp time.Time

+ TotalQueueResources schedulerobjects.ResourceList

+ TotalQueueResourcesByPriority schedulerobjects.QuantityByPriorityAndResourceType

+ UnschedulableReason string

+ func (report *JobSchedulingReport) String() string

+ type JobsIterator interface

+ Next func() (*api.Job, error)

+ type LegacyScheduler struct

+ ExecutorId string

+ JobRepository SchedulerJobRepository

+ MinimumJobSize map[string]resource.Quantity

+ NodeDb *NodeDb

+ PriorityFactorByQueue map[string]float64

+ Rand *rand.Rand

+ SchedulingConfig configuration.SchedulingConfig

+ SchedulingRoundReport *SchedulingRoundReport

+ TotalResources schedulerobjects.ResourceList

+ func NewLegacyScheduler(schedulingConfig configuration.SchedulingConfig, executorId string, ...) (*LegacyScheduler, error)

+ func (c *LegacyScheduler) Schedule(ctx context.Context, ...) ([]*api.Job, error)

+ func (sched *LegacyScheduler) String() string

+ type NodeDb struct

+ AssignedByNode map[string]schedulerobjects.AssignedByPriorityAndResourceType

+ Db *memdb.MemDB

+ JobsByNode map[string]map[uuid.UUID]interface{}

+ NodeTypes map[string]*schedulerobjects.NodeType

+ NodesByJob map[uuid.UUID]map[string]interface{}

+ func NewNodeDb(priorities []int32, resourceTypes []string) (*NodeDb, error)

+ func (nodeDb *NodeDb) BindNodeToPod(jobId uuid.UUID, req *schedulerobjects.PodRequirements, ...) error

+ func (nodeDb *NodeDb) MarkJobRunning(jobId uuid.UUID)

+ func (nodeDb *NodeDb) NodeTypesMatchingPod(req *schedulerobjects.PodRequirements) ([]*schedulerobjects.NodeType, map[string]int, error)

+ func (nodeDb *NodeDb) SelectAndBindNodeToPod(jobId uuid.UUID, req *schedulerobjects.PodRequirements) (*PodSchedulingReport, error)

+ func (nodeDb *NodeDb) SelectNodeForPod(jobId uuid.UUID, req *schedulerobjects.PodRequirements) (*PodSchedulingReport, error)

+ func (nodeDb *NodeDb) String() string

+ func (nodeDb *NodeDb) Upsert(nodes []*schedulerobjects.Node) error

+ type NodeItemAvailableResourceIndex struct

+ Priority int32

+ Resource string

+ func (s *NodeItemAvailableResourceIndex) FromArgs(args ...interface{}) ([]byte, error)

+ func (s *NodeItemAvailableResourceIndex) FromObject(raw interface{}) (bool, []byte, error)

+ type NodeTypeResourceIterator struct

+ func NewNodeTypeResourceIterator(txn *memdb.Txn, resource string, priority int32, ...) (*NodeTypeResourceIterator, error)

+ func (it *NodeTypeResourceIterator) Next() interface{}

+ func (it *NodeTypeResourceIterator) NextNodeItem() *schedulerobjects.Node

+ func (it *NodeTypeResourceIterator) WatchCh() <-chan struct{}

+ type NodeTypesResourceIterator struct

+ func NewNodeTypesResourceIterator(txn *memdb.Txn, resource string, priority int32, ...) (*NodeTypesResourceIterator, error)

+ func (it *NodeTypesResourceIterator) Next() interface{}

+ func (it *NodeTypesResourceIterator) NextNodeItem() *schedulerobjects.Node

+ func (it *NodeTypesResourceIterator) WatchCh() <-chan struct{}

+ type NodeTypesResourceIteratorItem struct

+ type NodeTypesResourceIteratorPQ []*NodeTypesResourceIteratorItem

+ func (pq *NodeTypesResourceIteratorPQ) Pop() any

+ func (pq *NodeTypesResourceIteratorPQ) Push(x any)

+ func (pq NodeTypesResourceIteratorPQ) Len() int

+ func (pq NodeTypesResourceIteratorPQ) Less(i, j int) bool

+ func (pq NodeTypesResourceIteratorPQ) Swap(i, j int)

+ type PodSchedulingReport struct

+ DominantResourceType string

+ JobId uuid.UUID

+ Node *schedulerobjects.Node

+ NumExcludedNodeTypesByReason map[string]int

+ NumExcludedNodesByReason map[string]int

+ NumMatchedNodeTypes int

+ Req *schedulerobjects.PodRequirements

+ Score int

+ Timestamp time.Time

+ func (report *PodSchedulingReport) String() string

+ type QueueCandidateJobsIterator struct

+ func NewQueueCandidateJobsIterator(ctx context.Context, queue string, ...) (*QueueCandidateJobsIterator, error)

+ func (it *QueueCandidateJobsIterator) Lease(jobSchedulingReport *JobSchedulingReport)

+ func (it *QueueCandidateJobsIterator) Next() (*JobSchedulingReport, error)

+ type QueueSchedulingReport struct

+ MostRecentSuccessfulJobSchedulingReport *JobSchedulingReport

+ MostRecentUnsuccessfulJobSchedulingReport *JobSchedulingReport

+ Name string

+ func (report *QueueSchedulingReport) String() string

+ type QueuedJobsIterator struct

+ func NewQueuedJobsIterator(ctx context.Context, queue string, repo SchedulerJobRepository) (*QueuedJobsIterator, error)

+ func (it *QueuedJobsIterator) Next() (*api.Job, error)

+ type SchedulerJobRepository interface

+ GetExistingJobsByIds func(jobIds []string) ([]*api.Job, error)

+ GetQueueJobIds func(queue string) ([]string, error)

+ TryLeaseJobs func(clusterId string, queue string, jobs []*api.Job) ([]*api.Job, error)

+ type SchedulingReportsRepository struct

+ MostRecentJobSchedulingReports *lru.Cache

+ MostRecentQueueSchedulingReports *lru.Cache

+ func NewSchedulingReportsRepository(maxQueueSchedulingReports, maxJobSchedulingReports int) *SchedulingReportsRepository

+ func (repo *SchedulingReportsRepository) Add(queueName string, report *JobSchedulingReport)

+ func (repo *SchedulingReportsRepository) GetJobReport(ctx context.Context, jobId *schedulerobjects.JobId) (*schedulerobjects.JobReport, error)

+ func (repo *SchedulingReportsRepository) GetJobSchedulingReport(jobId uuid.UUID) (*JobSchedulingReport, bool)

+ func (repo *SchedulingReportsRepository) GetQueueReport(ctx context.Context, queue *schedulerobjects.Queue) (*schedulerobjects.QueueReport, error)

+ func (repo *SchedulingReportsRepository) GetQueueSchedulingReport(queueName string) (*QueueSchedulingReport, bool)

+ type SchedulingRoundReport struct

+ Executor string

+ Finished time.Time

+ InitialResourcesByQueueAndPriority map[string]schedulerobjects.QuantityByPriorityAndResourceType

+ PriorityFactorByQueue map[string]float64

+ ScheduledResourcesByQueueAndPriority map[string]schedulerobjects.QuantityByPriorityAndResourceType

+ Started time.Time

+ SuccessfulJobSchedulingReportsByQueue map[string]map[uuid.UUID]*JobSchedulingReport

+ TerminationReason string

+ TotalResources schedulerobjects.ResourceList

+ UnsuccessfulJobSchedulingReportsByQueue map[string]map[uuid.UUID]*JobSchedulingReport

+ func NewSchedulingRoundReport(totalResources schedulerobjects.ResourceList, ...) *SchedulingRoundReport

+ func (report *SchedulingRoundReport) String() string

+ func (schedulingRoundReport *SchedulingRoundReport) AddJobSchedulingReport(jobSchedulingReport *JobSchedulingReport)

+ func (schedulingRoundReport *SchedulingRoundReport) AddScheduledResources(queue string, req *schedulerobjects.PodRequirements)

+ func (schedulingRoundReport *SchedulingRoundReport) ClearJobSpecs()

v0.3.34

Sep 22, 2022

v0.3.33

Sep 22, 2022

v0.3.32

Sep 16, 2022

Changes in this version

+ func CopyProtocolUpsert(ctx context.Context, tx pgx.Tx, tableName string, schema string, ...) error

+ func JobErrorsSchema() string

+ func JobRunAssignmentSchema() string

+ func JobRunErrorsSchema() string

+ func JobsSchema() string

+ func LeaderelectionSchema() string

+ func NamesFromRecord(x interface{}) []string

+ func NamesValuesFromRecord(x interface{}) ([]string, []interface{})

+ func NodeInfoSchema() string

+ func PulsarSchema() string

+ func RunsSchema() string

+ func Upsert(ctx context.Context, db *pgxpool.Pool, tableName string, schema string, ...) error

+ func ValuesFromRecord(x interface{}) []interface

+ func WriteDbOp(ctx context.Context, db *pgxpool.Pool, op DbOperation) error

+ type DBTX interface

+ Exec func(context.Context, string, ...interface{}) (pgconn.CommandTag, error)

+ Query func(context.Context, string, ...interface{}) (pgx.Rows, error)

+ QueryRow func(context.Context, string, ...interface{}) pgx.Row

+ type DbOperation interface

+ CanBeAppliedBefore func(DbOperation) bool

+ Merge func(DbOperation) bool

+ func AppendDbOperation(ops []DbOperation, op DbOperation) []DbOperation

+ func DbOpsFromEventInSequence(sequence *armadaevents.EventSequence, i int) ([]DbOperation, error)

+ type DbOperationsBatcher struct

+ In chan *eventutil.EventSequenceWithMessageIds

+ MaxInterval time.Duration

+ MaxMessages int

+ Out chan *DbOperationsWithMessageIds

+ func NewDbOperationsBatcher(in chan *eventutil.EventSequenceWithMessageIds) *DbOperationsBatcher

+ func (srv *DbOperationsBatcher) Run(ctx context.Context) error

+ type DbOperationsWithMessageIds struct

+ MessageIds []pulsar.MessageID

+ Ops []DbOperation

+ type DbOpsWriter struct

+ Consumer pulsar.Consumer

+ Db *pgxpool.Pool

+ In chan *DbOperationsWithMessageIds

+ Logger *logrus.Entry

+ func (srv *DbOpsWriter) Run(ctx context.Context) error

+ type ErrLostLeadership struct

+ Id uuid.UUID

+ func (err *ErrLostLeadership) Error() string

+ type ExecutorApi struct

+ Db *pgxpool.Pool

+ MaxJobsPerCall int32

+ Producer pulsar.Producer

+ func (srv *ExecutorApi) RenewLease(ctx context.Context, req *api.RenewLeaseRequest) (*api.IdList, error)

+ func (srv *ExecutorApi) ReportDone(ctx context.Context, req *api.IdList) (*api.IdList, error)

+ func (srv *ExecutorApi) ReportUsage(ctx context.Context, req *api.ClusterUsageReport) (*types.Empty, error)

+ func (srv *ExecutorApi) ReturnLease(ctx context.Context, req *api.ReturnLeaseRequest) (*types.Empty, error)

+ func (srv *ExecutorApi) StreamingLeaseJobs(stream api.AggregatedQueue_StreamingLeaseJobsServer) error

+ type Ingester struct

+ ConsumerOptions pulsar.ConsumerOptions

+ Db *pgxpool.Pool

+ Logger *logrus.Entry

+ MaxDbOps int

+ MaxWriteInterval time.Duration

+ PulsarClient pulsar.Client

+ func (srv *Ingester) Run(ctx context.Context) error

+ type InsertJobErrors map[int32]*JobError

+ func (a InsertJobErrors) CanBeAppliedBefore(b DbOperation) bool

+ func (a InsertJobErrors) Merge(b DbOperation) bool

+ type InsertJobRunErrors map[int32]*JobRunError

+ func (a InsertJobRunErrors) CanBeAppliedBefore(b DbOperation) bool

+ func (a InsertJobRunErrors) Merge(b DbOperation) bool

+ type InsertJobs map[uuid.UUID]*Job

+ func (a InsertJobs) CanBeAppliedBefore(b DbOperation) bool

+ func (a InsertJobs) Merge(b DbOperation) bool

+ type InsertRunAssignments map[uuid.UUID]*JobRunAssignment

+ func (a InsertRunAssignments) CanBeAppliedBefore(b DbOperation) bool

+ func (a InsertRunAssignments) Merge(b DbOperation) bool

+ type InsertRuns map[uuid.UUID]*Run

+ func (a InsertRuns) CanBeAppliedBefore(b DbOperation) bool

+ func (a InsertRuns) Merge(b DbOperation) bool

+ type Job struct

+ Cancelled bool

+ Failed bool

+ Groups []string

+ JobID uuid.UUID

+ JobSet string

+ LastModified time.Time

+ Priority int64

+ Queue string

+ SchedulingInfo []byte

+ Serial int64

+ SubmitMessage []byte

+ Succeeded bool

+ UserID string

+ type JobError struct

+ Error []byte

+ ID int32

+ JobID uuid.UUID

+ LastModified time.Time

+ Serial int64

+ Terminal bool

+ type JobRunAssignment struct

+ Assignment []byte

+ LastModified time.Time

+ RunID uuid.UUID

+ Serial int64

+ type JobRunError struct

+ Error []byte

+ ID int32

+ LastModified time.Time

+ RunID uuid.UUID

+ Serial int64

+ Terminal bool

+ type JobRuns struct

+ ActiveRuns map[uuid.UUID]*Run

+ InactiveRuns map[uuid.UUID]*Run

+ func NewJobRuns() *JobRuns

+ type JobSetOperation interface

+ AffectsJobSet func(string) bool

+ type LeaderElection struct

+ Db *pgxpool.Pool

+ Id uuid.UUID

+ Interval time.Duration

+ Logger *logrus.Entry

+ Timeout time.Duration

+ func NewLeaderElection(db *pgxpool.Pool) *LeaderElection

+ func (srv *LeaderElection) BecomeLeader(ctx context.Context) error

+ func (srv *LeaderElection) StayLeader(ctx context.Context) error

+ type Leaderelection struct

+ ID uuid.UUID

+ IsLeader bool

+ LastModified time.Time

+ Serial int64

+ type MarkJobSetsCancelled map[string]bool

+ func (a MarkJobSetsCancelled) AffectsJobSet(jobSet string) bool

+ func (a MarkJobSetsCancelled) CanBeAppliedBefore(b DbOperation) bool

+ func (a MarkJobSetsCancelled) Merge(b DbOperation) bool

+ type MarkJobsCancelled map[uuid.UUID]bool

+ func (a MarkJobsCancelled) CanBeAppliedBefore(b DbOperation) bool

+ func (a MarkJobsCancelled) Merge(b DbOperation) bool

+ type MarkJobsFailed map[uuid.UUID]bool

+ func (a MarkJobsFailed) CanBeAppliedBefore(b DbOperation) bool

+ func (a MarkJobsFailed) Merge(b DbOperation) bool

+ type MarkJobsSucceeded map[uuid.UUID]bool

+ func (a MarkJobsSucceeded) CanBeAppliedBefore(b DbOperation) bool

+ func (a MarkJobsSucceeded) Merge(b DbOperation) bool

+ type MarkRunsAsSentByExecutorAndJobIdParams struct

+ Executor string

+ JobIds []uuid.UUID

+ type MarkRunsFailed map[uuid.UUID]bool

+ func (a MarkRunsFailed) CanBeAppliedBefore(b DbOperation) bool

+ func (a MarkRunsFailed) Merge(b DbOperation) bool

+ type MarkRunsRunning map[uuid.UUID]bool

+ func (a MarkRunsRunning) CanBeAppliedBefore(b DbOperation) bool

+ func (a MarkRunsRunning) Merge(b DbOperation) bool

+ type MarkRunsSucceeded map[uuid.UUID]bool

+ func (a MarkRunsSucceeded) CanBeAppliedBefore(b DbOperation) bool

+ func (a MarkRunsSucceeded) Merge(b DbOperation) bool

+ type Nodeinfo struct

+ Executor string

+ ExecutorNodeName string

+ LastModified time.Time

+ Message []byte

+ NodeName string

+ Serial int64

+ type Pulsar struct

+ BatchIdx int32

+ EntryID int64

+ LedgerID int64

+ PartitionIdx int32

+ Topic string

+ type Queries struct

+ func New(db DBTX) *Queries

+ func (q *Queries) GetTopicMessageIds(ctx context.Context, topic string) ([]Pulsar, error)

+ func (q *Queries) MarkJobCancelledById(ctx context.Context, jobID uuid.UUID) error

+ func (q *Queries) MarkJobFailedById(ctx context.Context, jobID uuid.UUID) error

+ func (q *Queries) MarkJobRunCancelledByJobId(ctx context.Context, jobID uuid.UUID) error

+ func (q *Queries) MarkJobRunFailedById(ctx context.Context, runID uuid.UUID) error

+ func (q *Queries) MarkJobRunRunningById(ctx context.Context, runID uuid.UUID) error

+ func (q *Queries) MarkJobRunSucceededById(ctx context.Context, runID uuid.UUID) error

+ func (q *Queries) MarkJobRunsCancelledByJobId(ctx context.Context, jobIds []uuid.UUID) error

+ func (q *Queries) MarkJobRunsCancelledBySet(ctx context.Context, jobSet string) error

+ func (q *Queries) MarkJobRunsCancelledBySets(ctx context.Context, jobSets []string) error

+ func (q *Queries) MarkJobRunsFailedById(ctx context.Context, runIds []uuid.UUID) error

+ func (q *Queries) MarkJobRunsRunningById(ctx context.Context, runIds []uuid.UUID) error

+ func (q *Queries) MarkJobRunsSucceededById(ctx context.Context, runIds []uuid.UUID) error

+ func (q *Queries) MarkJobSucceededById(ctx context.Context, jobID uuid.UUID) error

+ func (q *Queries) MarkJobsCancelledById(ctx context.Context, jobIds []uuid.UUID) error

+ func (q *Queries) MarkJobsCancelledBySet(ctx context.Context, jobSet string) error

+ func (q *Queries) MarkJobsCancelledBySets(ctx context.Context, jobSets []string) error

+ func (q *Queries) MarkJobsFailedById(ctx context.Context, jobIds []uuid.UUID) error

+ func (q *Queries) MarkJobsFailedBySet(ctx context.Context, jobSet string) error

+ func (q *Queries) MarkJobsFailedBySets(ctx context.Context, jobSets []string) error

+ func (q *Queries) MarkJobsSucceededById(ctx context.Context, jobIds []uuid.UUID) error

+ func (q *Queries) MarkJobsSucceededBySet(ctx context.Context, jobSet string) error

+ func (q *Queries) MarkJobsSucceededBySets(ctx context.Context, jobSets []string) error

+ func (q *Queries) MarkRunsAsSent(ctx context.Context, runIds []uuid.UUID) error

+ func (q *Queries) MarkRunsAsSentByExecutorAndJobId(ctx context.Context, arg MarkRunsAsSentByExecutorAndJobIdParams) error

+ func (q *Queries) SelectJobErrorsById(ctx context.Context, jobID uuid.UUID) ([]JobError, error)

+ func (q *Queries) SelectJobsFromIds(ctx context.Context, jobIds []uuid.UUID) ([]Job, error)

+ func (q *Queries) SelectLeader(ctx context.Context) (Leaderelection, error)

+ func (q *Queries) SelectNewActiveJobs(ctx context.Context, serial int64) ([]Job, error)

+ func (q *Queries) SelectNewJobErrors(ctx context.Context, serial int64) ([]JobError, error)

+ func (q *Queries) SelectNewJobs(ctx context.Context, serial int64) ([]Job, error)

+ func (q *Queries) SelectNewNodeInfo(ctx context.Context, serial int64) ([]Nodeinfo, error)

+ func (q *Queries) SelectNewRunAssignments(ctx context.Context, serial int64) ([]JobRunAssignment, error)

+ func (q *Queries) SelectNewRunErrors(ctx context.Context, serial int64) ([]JobRunError, error)

+ func (q *Queries) SelectNewRunsForExecutorWithLimit(ctx context.Context, arg SelectNewRunsForExecutorWithLimitParams) ([]Run, error)

+ func (q *Queries) SelectNewRunsForJobs(ctx context.Context, arg SelectNewRunsForJobsParams) ([]Run, error)

+ func (q *Queries) SelectQueueJobSetFromId(ctx context.Context, jobID uuid.UUID) (SelectQueueJobSetFromIdRow, error)

+ func (q *Queries) SelectQueueJobSetFromIds(ctx context.Context, jobIds []uuid.UUID) ([]SelectQueueJobSetFromIdsRow, error)

+ func (q *Queries) SelectReplicaById(ctx context.Context, id uuid.UUID) (Leaderelection, error)

+ func (q *Queries) SelectRunErrorsById(ctx context.Context, runID uuid.UUID) ([]JobRunError, error)

+ func (q *Queries) SelectRunsFromExecutorAndJobs(ctx context.Context, arg SelectRunsFromExecutorAndJobsParams) ([]Run, error)

+ func (q *Queries) SelectUnsentRunsForExecutor(ctx context.Context, executor string) ([]Run, error)

+ func (q *Queries) UpdateJobPriorityById(ctx context.Context, arg UpdateJobPriorityByIdParams) error

+ func (q *Queries) UpdateJobPriorityByJobSet(ctx context.Context, arg UpdateJobPriorityByJobSetParams) error

+ func (q *Queries) UpsertMessageId(ctx context.Context, arg UpsertMessageIdParams) error

+ func (q *Queries) WithTx(tx pgx.Tx) *Queries

+ type Queue struct

+ Name string

+ Weight float64

+ type Run struct

+ Cancelled bool

+ Executor string

+ Failed bool

+ JobID uuid.UUID

+ JobSet string

+ LastModified time.Time

+ RunID uuid.UUID

+ Running bool

+ SentToExecutor bool

+ Serial int64

+ Succeeded bool

+ type Scheduler struct

+ ActiveJobs map[uuid.UUID]*Job

+ Db *pgxpool.Pool

+ ExecutorAliveDuration time.Duration

+ Executors map[string]time.Time

+ JobErrorsSerial int64

+ JobRunErrorsSerial int64

+ JobRunsAssignmentsSerial int64

+ JobsSerial int64

+ Logger *logrus.Entry

+ Nodes map[string]*Nodeinfo

+ NodesSerial int64

+ Producer pulsar.Producer

+ QueuedJobIds []uuid.UUID

+ RunsByJobId map[uuid.UUID]*JobRuns

+ RunsSerial int64

+ func NewScheduler(producer pulsar.Producer, db *pgxpool.Pool) *Scheduler

+ func (srv *Scheduler) Run(ctx context.Context) error

+ type SelectNewRunsForExecutorWithLimitParams struct

+ Executor string

+ Limit int32

+ type SelectNewRunsForJobsParams struct

+ JobIds []uuid.UUID

+ Serial int64

+ type SelectQueueJobSetFromIdRow struct

+ JobID uuid.UUID

+ JobSet string

+ Queue string

+ type SelectQueueJobSetFromIdsRow struct

+ JobID uuid.UUID

+ JobSet string

+ Queue string

+ type SelectRunsFromExecutorAndJobsParams struct

+ Executor string

+ JobIds []uuid.UUID

+ type UpdateJobPriorities map[uuid.UUID]int64

+ func (a UpdateJobPriorities) CanBeAppliedBefore(b DbOperation) bool

+ func (a UpdateJobPriorities) Merge(b DbOperation) bool

+ type UpdateJobPriorityByIdParams struct

+ JobID uuid.UUID

+ Priority int64

+ type UpdateJobPriorityByJobSetParams struct

+ JobSet string

+ Priority int64

+ type UpdateJobSetPriorities map[string]int64

+ func (a UpdateJobSetPriorities) AffectsJobSet(jobSet string) bool

+ func (a UpdateJobSetPriorities) CanBeAppliedBefore(b DbOperation) bool

+ func (a UpdateJobSetPriorities) Merge(b DbOperation) bool

+ type UpsertMessageIdParams struct

+ BatchIdx int32

+ EntryID int64

+ LedgerID int64

+ PartitionIdx int32

+ Topic string

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL