jobop

Documentation

Index

• type BackoffConfig
  • func NewDefaultBackoffConfig() *BackoffConfig
• type BackoffManager
• type BackoffManagerImpl
  • func NewBackoffManagerImpl(clocker clock.Clock, config *BackoffConfig) *BackoffManagerImpl
  • func (m *BackoffManagerImpl) Allow(jobID string) bool
  • func (m *BackoffManagerImpl) JobFail(jobID string)
  • func (m *BackoffManagerImpl) JobOnline(jobID string)
  • func (m *BackoffManagerImpl) JobTerminate(jobID string)
  • func (m *BackoffManagerImpl) Terminate(jobID string) bool
• type JobBackoff
  • func NewJobBackoff(jobID string, clocker clock.Clock, config *BackoffConfig) *JobBackoff
  • func (b *JobBackoff) Allow() bool
  • func (b *JobBackoff) Fail()
  • func (b *JobBackoff) Success()
  • func (b *JobBackoff) Terminate() bool
• type JobOperator
• type JobOperatorImpl
  • func NewJobOperatorImpl(cli pkgOrm.Client, router operateRouter) *JobOperatorImpl
  • func (oper *JobOperatorImpl) IsJobCanceling(ctx context.Context, jobID string) bool
  • func (oper *JobOperatorImpl) MarkJobCanceled(ctx context.Context, jobID string) error
  • func (oper *JobOperatorImpl) MarkJobCanceling(ctx context.Context, jobID string) error
  • func (oper *JobOperatorImpl) MarkJobNoop(ctx context.Context, jobID string) error
  • func (oper *JobOperatorImpl) Tick(ctx context.Context) error

Types

type BackoffConfig

type BackoffConfig struct {
	ResetInterval   time.Duration `toml:"reset-interval" json:"reset-interval"`
	InitialInterval time.Duration `toml:"initial-interval" json:"initial-interval"`
	MaxInterval     time.Duration `toml:"max-interval" json:"max-interval"`
	Multiplier      float64       `toml:"multiplier" json:"multiplier"`
	MaxTryTime      int           `toml:"max-try-time" json:"max-try-time"`
}

BackoffConfig is used to configure job backoff

func NewDefaultBackoffConfig

func NewDefaultBackoffConfig() *BackoffConfig

NewDefaultBackoffConfig creates a default backoff config

type BackoffManager

type BackoffManager interface {
	Terminate(jobID string) bool
	Allow(jobID string) bool
	JobOnline(jobID string)
	JobFail(jobID string)
	JobTerminate(jobID string)
}

BackoffManager manages backoff of a target job set

type BackoffManagerImpl

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

BackoffManagerImpl manages JobBackoff of all running or retrying jobs, it implement interface BackoffManager.

func NewBackoffManagerImpl

func NewBackoffManagerImpl(clocker clock.Clock, config *BackoffConfig) *BackoffManagerImpl

NewBackoffManagerImpl creates a new backoff manager

func (*BackoffManagerImpl) Allow

func (m *BackoffManagerImpl) Allow(jobID string) bool

Allow checks whether this job can be created now

func (*BackoffManagerImpl) JobFail

func (m *BackoffManagerImpl) JobFail(jobID string)

JobFail means a job is offline(with error) or dispatched with error

func (*BackoffManagerImpl) JobOnline

func (m *BackoffManagerImpl) JobOnline(jobID string)

JobOnline means a job is online

func (*BackoffManagerImpl) JobTerminate

func (m *BackoffManagerImpl) JobTerminate(jobID string)

JobTerminate means a job is finished, canceled or failed

func (*BackoffManagerImpl) Terminate

func (m *BackoffManagerImpl) Terminate(jobID string) bool

Terminate checks whether this job should be terminated, terminated means job manager won't create this job any more.

type JobBackoff

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

JobBackoff is a job backoff manager, it recoreds job online and offline events and determines whether a job can be re-created based on backoff mechanism. The backoff stragegy is as following

• Each time a fail event arrives, the backoff time will be move forward by nextBackoff.
• If a job is success for more than `resetInterval`, the backoff history will be cleared, and backoff time will be re-calculated.

func NewJobBackoff

func NewJobBackoff(jobID string, clocker clock.Clock, config *BackoffConfig) *JobBackoff

NewJobBackoff creates a new job backoff

func (*JobBackoff) Allow

func (b *JobBackoff) Allow() bool

Allow returns whether new request(create job) is allowd

func (*JobBackoff) Fail

func (b *JobBackoff) Fail()

Fail is called when a failure event happens

func (*JobBackoff) Success

func (b *JobBackoff) Success()

Success is called when a success event happens

func (*JobBackoff) Terminate

func (b *JobBackoff) Terminate() bool

Terminate returns whether job should be terminated. It happens when job fails continuously for more than max try times.

type JobOperator

type JobOperator interface {
	MarkJobCanceling(ctx context.Context, jobID string) error
	MarkJobCanceled(ctx context.Context, jobID string) error
	Tick(ctx context.Context) error
	IsJobCanceling(ctx context.Context, jobID string) bool
}

JobOperator abstracts a metastore based job operator, it encapsulates logic to handle JobOp and a Tick API to ensure job moves towards to expected status.

type JobOperatorImpl

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

JobOperatorImpl implements JobOperator

func NewJobOperatorImpl

func NewJobOperatorImpl(cli pkgOrm.Client, router operateRouter) *JobOperatorImpl

NewJobOperatorImpl creates a new JobOperatorImpl

func (*JobOperatorImpl) IsJobCanceling

func (oper *JobOperatorImpl) IsJobCanceling(ctx context.Context, jobID string) bool

IsJobCanceling implements JobOperator

func (*JobOperatorImpl) MarkJobCanceled

func (oper *JobOperatorImpl) MarkJobCanceled(ctx context.Context, jobID string) error

MarkJobCanceled implements JobOperator.MarkJobCanceled

func (*JobOperatorImpl) MarkJobCanceling

func (oper *JobOperatorImpl) MarkJobCanceling(ctx context.Context, jobID string) error

MarkJobCanceling implements JobOperator.MarkJobCanceling

func (*JobOperatorImpl) MarkJobNoop

func (oper *JobOperatorImpl) MarkJobNoop(ctx context.Context, jobID string) error

MarkJobNoop implements JobOperator.MarkJobNoop

func (*JobOperatorImpl) Tick

func (oper *JobOperatorImpl) Tick(ctx context.Context) error

Tick implements JobOperator.Tick