job

Documentation

Index

• Constants
• func CreateSchema(db *gorm.DB)
• func RunTestWorker(ctx context.Context, workerIndex int, numWorkers int, numRoaches int, ...)
• type ArchivedInstance
  • func (ai *ArchivedInstance) JobAndInstance() (string, int64)
  • func (ai *ArchivedInstance) TableName() string
• type Artifact
• type Dispatcher
  • func NewDispatcher(db *gorm.DB, currentNodeID string, nodeStalenessThreshold time.Duration, ...) Dispatcher
• type Fetcher
  • func NewFetcher(db *gorm.DB, nodeID string, nodeIndex int, numWorkers int) Fetcher
• type Instance
  • func (ji *Instance) JobAndInstance() (string, int64)
  • func (ji *Instance) TableName() string
• type InstanceSpecifier
• type Job
• type Node
• type Runner
  • func NewRunner(db *gorm.DB, currentNodeID string, nodeStalenessThreshold time.Duration, ...) Runner
• type Scheduler
  • func NewScheduler(db *gorm.DB, numWorkers int) Scheduler

Constants

const (
	// NumJobsPerWorker is the flag for number of jobs each worker performs
	NumJobsPerWorker = "num_jobs_per_worker"
	// JobPeriodScaleMillis is the flag for the scale for the period of the jobs
	JobPeriodScaleMillis = "job_period_scale_millis"
)

const (
	DefaultDatabaseName string = "job"
)

Constants used by testing for interacting with cockroach.

Functions

func CreateSchema

func CreateSchema(db *gorm.DB)

CreateSchema creates the schema for the job test in the specified db, but first drop the tables if they exist.

func RunTestWorker

func RunTestWorker(
	ctx context.Context,
	workerIndex int,
	numWorkers int,
	numRoaches int,
	numJobsPerWorker int,
	jobPeriodScaleMillis int64,
	workerPoolSize int,
	currentTime int64,
	testDuration int64,
	db *gorm.DB,
	testWorkersRunning *sync.WaitGroup,
)

RunTestWorker starts a worker to act as a mock node while r many jobs. Also schedules the specified number of jobs to distribute the work of job scheduling.

Types

type ArchivedInstance

type ArchivedInstance struct {
	JobID          string `gorm:"primary_key"`
	InstanceID     int64  `gorm:"primary_key"`
	Shard          int    `gorm:"index:idx_shard_start"`
	Status         string `sql:"not null"`
	StartTime      int64  `gorm:"index:idx_shard_start"`
	EndTime        int64
	NodeID         *string
	SequenceNumber int64
}

An ArchivedInstance represents one particular instance of a recurring Job that has terminated.

func (*ArchivedInstance) JobAndInstance

func (ai *ArchivedInstance) JobAndInstance() (string, int64)

JobAndInstance returns the job id as a string and the instance id as an int64.

func (*ArchivedInstance) TableName

func (ai *ArchivedInstance) TableName() string

TableName provides a custom table name to gorm.

type Artifact

type Artifact struct {
	ID      string `gorm:"primary_key"`
	Shard   int    `gorm:"index:idx_shard_index_id"`
	IndexID string `gorm:"index:idx_shard_index_id"` // HACK to get 2-d index
}

An Artifact mocks result of the simulated jobs.

type Dispatcher

type Dispatcher interface {
	Dispatch(instance *Instance)
	// contains filtered or unexported methods
}

A Dispatcher provides the ability to run a Instance asynchronously.

func NewDispatcher

func NewDispatcher(
	db *gorm.DB,
	currentNodeID string,
	nodeStalenessThreshold time.Duration,
	workerPoolSize int,
	failureDelayMillis int64,
	numWorkers int,
) Dispatcher

NewDispatcher creates a Dispatcher that uses db to track job state transitions, and that runs/claims jobs as currentNodeID.

type Fetcher

type Fetcher interface {
	Fetch() ([]*Instance, error)
}

A Fetcher fetches JobInstances that need work to be performed on them (e.g., start time has passed and they are unclaimed or belong to a stale node.

func NewFetcher

func NewFetcher(
	db *gorm.DB,
	nodeID string,
	nodeIndex int,
	numWorkers int,
) Fetcher

NewFetcher creates a Fetcher that uses db for persistence operations and has an identity of nodeID.

type Instance

type Instance struct {
	JobID          string `gorm:"primary_key"`
	InstanceID     int64  `gorm:"primary_key"`
	Shard          int    `gorm:"index:idx_shard_start"`
	Status         string `sql:"not null"`
	StartTime      int64  `gorm:"index:idx_shard_start"`
	EndTime        int64
	NodeID         *string
	SequenceNumber int64
}

An Instance represents one particular instance of a recurring Job.

func (*Instance) JobAndInstance

func (ji *Instance) JobAndInstance() (string, int64)

JobAndInstance returns the job id as a string and the instance id as an int64.

func (*Instance) TableName

func (ji *Instance) TableName() string

TableName provides a custom table name to gorm.

type InstanceSpecifier

type InstanceSpecifier interface {
	// JobAndInstance returns the job id as a string and the instance id as an
	// int64.
	JobAndInstance() (string, int64)
}

An InstanceSpecifier provides a common interface for Instance and ArchivedInstance for code reuse in client code that only depends on the ability to get a job id and an instance id.

type Job

type Job struct {
	ID                string `gorm:"primary_key"`
	CurrentInstanceID int64
	PeriodMillis      int64
	JobConfig         string
}

A Job provides a specification of a recurring job to run.

type Node

type Node struct {
	ID             string `gorm:"primary_key"`
	LastUpdateTime int64
}

A Node represents a node that runs a job.

type Runner

type Runner interface {
	Run(instance *Instance)
}

A Runner runs job instances, transitioning each job instance through its states until there is no next state to iterate.

func NewRunner

func NewRunner(
	db *gorm.DB,
	currentNodeID string,
	nodeStalenessThreshold time.Duration,
	failureDelayMillis int64,
	numWorkers int,
) Runner

NewRunner creates a new Runner that uses db for persistence, has currentNodeID as its identity.

type Scheduler

type Scheduler interface {
	// ScheduleRecurringJob schedules a job with the specified id, period, and
	// config, and returns true if a job with the specified id already exists
	// or if it did not exist but was created by this method. If the internal
	// db operations failed or left the db in an ambiguous state, we return false.
	ScheduleRecurringJob(id string, periodMillis int64, jobConfig string) bool
}

A Scheduler schedules jobs to run periodically.

func NewScheduler

func NewScheduler(db *gorm.DB, numWorkers int) Scheduler

NewScheduler creates a Scheduler that is backed by db.