runner

Documentation

Overview

Package runner implements running a job.

Index

• Constants
• type Factory
  • func NewFactory(jf job.Factory, rmc rm.Client) Factory
• type Repo
  • func NewRepo() Repo
• type Return
• type Runner
  • func NewRunner(pJob proto.Job, realJob job.Job, reqId string, prevTries, totalTries uint, ...) Runner
• type Status

Constants

const (
	// Number of times to attempt sending a job log to the RM.
	JOB_LOG_TRIES = 5
	// Time to wait between attempts to send a job log to RM.
	JOB_LOG_RETRY_WAIT = 500 * time.Millisecond
)

Types

type Factory

type Factory interface {
	Make(job proto.Job, requestId string, prevTries, totalTries uint) (Runner, error)
}

A Factory makes a Runner for one job. There are two try counts: prevTries and totalTries. prevTries is a gauge from [0, 1+retry], where retry is the retry count from the request spec. The prevTries count is per-sequence try, which is why it can reset to zero on sequence retry (handled by a chain.Reaper). On suspend/resume, jobs are stopped and the try on which it's stopped doesn't count, so prevTries is decremented by 1 on resume to retry. The totalTries count is a monotonically increasing global counter of how many times the job was run. This count is used for the proto.JobLog.Try field which cannot repeat a number because the job_log table primary key is <request_id, job_id, try>.

func NewFactory

func NewFactory(jf job.Factory, rmc rm.Client) Factory

NewRunnerFactory makes a RunnerFactory.

type Repo

type Repo interface {
	Set(jobId string, runner Runner)
	Get(jobId string) Runner
	Remove(jobId string)
	Items() map[string]Runner
	Count() int
}

Repo is a small wrapper around a concurrent map that provides the ability to store and retrieve Runners in a thread-safe way.

func NewRepo

func NewRepo() Repo

type Return

type Return struct {
	FinalState byte // Final proto.STATE_*. Determines if/how chain continues running.
	Tries      uint // Number of tries this run, not including any previous tries
}

type Runner

type Runner interface {
	// Run runs the job, blocking until it has completed or when Stop is called.
	// If the job fails, Run will retry it as many times as the job is configured
	// to be retried. After each run attempt, a Job Log is created and sent to
	// the RM. When the job successfully completes, or reaches the maximum number
	// of retry attempts, Run returns the final state of the job.
	Run(jobData map[string]interface{}) Return

	// Stop stops the job if it's running. The job is responsible for stopping
	// quickly because Stop blocks while waiting for the job to stop.
	Stop() error

	// Status returns the job try count and real-time status. The runner handles
	// the try count. The underlying job.Job must handle async, real-time status
	// requests while running.
	Status() Status
}

A Runner runs and manages one job in a job chain. The job must implement the job.Job interface.

func NewRunner

func NewRunner(pJob proto.Job, realJob job.Job, reqId string, prevTries, totalTries uint, rmc rm.Client) Runner

NewRunner takes a proto.Job struct and its corresponding job.Job interface, and returns a Runner.

type Status

type Status struct {
	Job       proto.Job
	StartedAt time.Time // set once when Runner created
	Try       uint      // total tries, not current sequence try (proto.JobLog.Try)
	Status    string    // real-time job status (job.Job.Status())
	Sleeping  bool      // if sleeping between tries
}