scheduler

Documentation

Index

• Constants
• Variables
• type BatchRateLimiter
  • func NewBatchRateLimiter(params BatchRateLimiterParams) *BatchRateLimiter
  • func (b *BatchRateLimiter) Apply(ctx context.Context, plan *models.Plan, totalNeeded int) int
• type BatchRateLimiterParams
• type BatchServiceJobScheduler
  • func NewBatchServiceJobScheduler(params BatchServiceJobSchedulerParams) *BatchServiceJobScheduler
  • func (b *BatchServiceJobScheduler) Process(ctx context.Context, evaluation *models.Evaluation) (err error)
• type BatchServiceJobSchedulerParams
• type DaemonJobScheduler
  • func NewDaemonJobScheduler(params DaemonJobSchedulerParams) *DaemonJobScheduler
  • func (b *DaemonJobScheduler) Process(ctx context.Context, evaluation *models.Evaluation) (err error)
• type DaemonJobSchedulerParams
• type ExecutionRateLimiter
• type NoopRateLimiter
  • func NewNoopRateLimiter() *NoopRateLimiter
  • func (n *NoopRateLimiter) Apply(ctx context.Context, plan *models.Plan, totalNeeded int) int
• type OpsJobScheduler
  • func NewOpsJobScheduler(params OpsJobSchedulerParams) *OpsJobScheduler
  • func (b *OpsJobScheduler) Process(ctx context.Context, evaluation *models.Evaluation) (err error)
• type OpsJobSchedulerParams

Constants

const (
	AttrJobType       = "job_type"
	AttrEvalType      = "eval_type"
	AttrSchedulerType = "scheduler_type"

	AttrOperationPartGetJob      = "get_job"
	AttrOperationPartGetExecs    = "get_executions"
	AttrOperationPartGetNodes    = "get_node_infos"
	AttrOperationPartMatchNodes  = "match_nodes"
	AttrOperationPartProcessPlan = "process_plan"

	AttrOutcomeKey              = attribute.Key("outcome")
	AttrOutcomeSuccess          = "success"
	AttrOutcomeFailure          = "failure"
	AttrOutcomeAlreadyTerminal  = "already_terminal"
	AttrOutcomeExhaustedRetries = "exhausted_retries"
	AttrOutcomeQueueing         = "queueing"
	AttrOutcomeTimeout          = "timeout"
	AttrOutcomeQueueTimeout     = "queue_timeout"
)

Common attribute keys

Variables

var (
	Meter = otel.GetMeterProvider().Meter("scheduler")
)

Types

type BatchRateLimiter

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

BatchRateLimiter limits the number of executions that can be created in a single evaluation and creates delayed evaluations for remaining executions

func NewBatchRateLimiter

func NewBatchRateLimiter(params BatchRateLimiterParams) *BatchRateLimiter

func (*BatchRateLimiter) Apply

func (b *BatchRateLimiter) Apply(ctx context.Context, plan *models.Plan, totalNeeded int) int

type BatchRateLimiterParams

type BatchRateLimiterParams struct {
	// MaxExecutionsPerEval limits the number of new executions that can be created in a single evaluation
	MaxExecutionsPerEval int
	// ExecutionLimitBackoff is the duration to wait before creating a new evaluation when hitting execution limits
	ExecutionLimitBackoff time.Duration
	// Clock is used for time-based operations. If not provided, system clock is used.
	Clock clock.Clock
}

type BatchServiceJobScheduler

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

BatchServiceJobScheduler handles scheduling of batch and service jobs, with support for partitioned execution. Each job can specify N parallel executions via its Count field, where each execution is assigned a unique partition index from 0 to Count-1.

Partitioning Logic: - The job.Count field determines the number of partitions (N) - Each execution is assigned a unique PartitionIndex (0 to Count-1) - The scheduler ensures exactly one active execution per partition - When an execution fails, its partition becomes available for retry

Execution Lifecycle Per Partition: 1. Initial scheduling: assign execution to an available node with PartitionIndex 2. If execution fails: partition becomes available for retry on another node 3. If execution succeeds:

• Batch jobs: partition is marked complete, no more executions
• Service jobs: partition must maintain one running execution

Example with Count = 3: - Creates three partitions (0, 1, 2) - Each partition runs independently - If partition 1 fails, it can be retried while 0 and 2 continue running - Batch job completes when all three partitions have successful executions - Service job runs continuously with one execution per partition

func NewBatchServiceJobScheduler

func NewBatchServiceJobScheduler(params BatchServiceJobSchedulerParams) *BatchServiceJobScheduler

func (*BatchServiceJobScheduler) Process

func (b *BatchServiceJobScheduler) Process(ctx context.Context, evaluation *models.Evaluation) (err error)

Process handles the scheduling logic for partitioned jobs: 1. Tracks existing executions per partition 2. Handles failed executions and retries while maintaining partition assignment 3. Ensures each partition (0 to job.Count-1) has exactly one active execution 4. For batch jobs: monitors completion of all partitions 5. For service jobs: maintains continuous execution per partition

type BatchServiceJobSchedulerParams

type BatchServiceJobSchedulerParams struct {
	JobStore      jobstore.Store
	Planner       orchestrator.Planner
	NodeSelector  orchestrator.NodeSelector
	RetryStrategy orchestrator.RetryStrategy
	QueueBackoff  time.Duration
	// RateLimiter controls the rate at which new executions are created
	// If not provided, a NoopRateLimiter is used
	RateLimiter ExecutionRateLimiter
	// Clock is the clock used for time-based operations.
	// If not provided, the system clock is used.
	Clock clock.Clock
}

type DaemonJobScheduler

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

func NewDaemonJobScheduler

func NewDaemonJobScheduler(params DaemonJobSchedulerParams) *DaemonJobScheduler

func (*DaemonJobScheduler) Process

func (b *DaemonJobScheduler) Process(ctx context.Context, evaluation *models.Evaluation) (err error)

type DaemonJobSchedulerParams

type DaemonJobSchedulerParams struct {
	JobStore     jobstore.Store
	Planner      orchestrator.Planner
	NodeSelector orchestrator.NodeSelector
	// RateLimiter controls the rate at which new executions are created
	// If not provided, a NoopRateLimiter is used
	RateLimiter ExecutionRateLimiter
}

DaemonJobScheduler is a scheduler for batch jobs that run until completion

type ExecutionRateLimiter

type ExecutionRateLimiter interface {
	// Apply checks if the number of executions exceeds the limit and optionally creates
	// a delayed evaluation for remaining executions. Returns the number of executions
	// that should be created in this evaluation.
	Apply(ctx context.Context, plan *models.Plan, totalNeeded int) int
}

ExecutionRateLimiter provides rate limiting functionality for job executions

type NoopRateLimiter

type NoopRateLimiter struct{}

NoopRateLimiter is a rate limiter that does not impose any limits

func NewNoopRateLimiter

func NewNoopRateLimiter() *NoopRateLimiter

func (*NoopRateLimiter) Apply

func (n *NoopRateLimiter) Apply(ctx context.Context, plan *models.Plan, totalNeeded int) int

type OpsJobScheduler

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

func NewOpsJobScheduler

func NewOpsJobScheduler(params OpsJobSchedulerParams) *OpsJobScheduler

func (*OpsJobScheduler) Process

func (b *OpsJobScheduler) Process(ctx context.Context, evaluation *models.Evaluation) (err error)

type OpsJobSchedulerParams

type OpsJobSchedulerParams struct {
	JobStore     jobstore.Store
	Planner      orchestrator.Planner
	NodeSelector orchestrator.NodeSelector
	// RateLimiter controls the rate at which new executions are created
	// If not provided, a NoopRateLimiter is used
	RateLimiter ExecutionRateLimiter
	// Clock is the clock used for time-based operations.
	// If not provided, the system clock is used.
	Clock clock.Clock
}

OpsJobScheduler is a scheduler for batch jobs that run until completion