jobs

package
v0.12.1 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Dec 13, 2022 License: AGPL-3.0 Imports: 19 Imported by: 0

Documentation

Overview

+kubebuilder:validation:Optional

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func Module

func Module() fx.Option

Module is a fx module that provides default jobs for internal service components.

Types

type BasicJob

type BasicJob struct {
	JobFunc JobCallback
	JobBase
}

BasicJob is a basic job that every other job builds on.

func (*BasicJob) Execute

func (job *BasicJob) Execute(ctx context.Context) (proto.Message, error)

Execute executes the job.

func (*BasicJob) JobWatchers

func (job *BasicJob) JobWatchers() JobWatchers

JobWatchers returns the job watchers.

func (*BasicJob) Name

func (job *BasicJob) Name() string

Name returns the name of the job.

type BodyProvider

type BodyProvider func() io.Reader

BodyProvider allows the users to provide a body to the HTTP jobs. For example for posting a payload as a job.

type GroupWatcher

type GroupWatcher interface {
	OnJobRegistered(name string)
	OnJobDeregistered(name string)
	OnJobScheduled(name string)
	OnJobCompleted(name string, status *statusv1.Status, jobStats JobStats)
}

GroupWatcher is used for tracking completion of JobGroup.

type GroupWatchers

type GroupWatchers []GroupWatcher

GroupWatchers is a collection of GroupWatcher.

func (GroupWatchers) OnJobCompleted

func (gws GroupWatchers) OnJobCompleted(name string, status *statusv1.Status, jobStats JobStats)

OnJobCompleted calls OnJobCompleted for each GroupWatcher in the collection.

func (GroupWatchers) OnJobDeregistered

func (gws GroupWatchers) OnJobDeregistered(name string)

OnJobDeregistered calls OnJobDeregistered for each GroupWatcher in the collection.

func (GroupWatchers) OnJobRegistered

func (gws GroupWatchers) OnJobRegistered(name string)

OnJobRegistered calls OnJobRegistered for each GroupWatcher in the collection.

func (GroupWatchers) OnJobScheduled

func (gws GroupWatchers) OnJobScheduled(name string)

OnJobScheduled calls OnJobScheduled for each GroupWatcher in the collection.

type HTTPJob

type HTTPJob struct {
	BasicJob
	// contains filtered or unexported fields
}

HTTPJob wraps a basic job along with HTTPJobConfig to execute an HTTP job.

func NewHTTPJob

func NewHTTPJob(config HTTPJobConfig) *HTTPJob

NewHTTPJob creates a new HTTPJob.

func (*HTTPJob) Execute

func (job *HTTPJob) Execute(ctx context.Context) (proto.Message, error)

Execute executes the job.

func (*HTTPJob) JobWatchers

func (job *HTTPJob) JobWatchers() JobWatchers

JobWatchers returns the job watchers for the job.

func (*HTTPJob) Name

func (job *HTTPJob) Name() string

Name returns the name of the job.

type HTTPJobConfig

type HTTPJobConfig struct {
	Client         *http.Client
	Body           BodyProvider
	URL            string
	Method         string
	ExpectedBody   string
	Name           string
	ExpectedStatus int
}

HTTPJobConfig is the configuration for an HTTP job.

type Job

type Job interface {
	// Returns the name
	Name() string
	// Executes the job
	Execute(ctx context.Context) (proto.Message, error)
	// JobWatchers
	JobWatchers() JobWatchers
}

Job interface and basic job implementation.

type JobBase

type JobBase struct {
	JobName string
	JWS     JobWatchers
}

JobBase is the base job implementation.

func (JobBase) JobWatchers

func (job JobBase) JobWatchers() JobWatchers

JobWatchers returns the job watchers.

func (JobBase) Name

func (job JobBase) Name() string

Name returns the name of the job.

type JobCallback

type JobCallback func(context.Context) (proto.Message, error)

JobCallback is the callback function that is called after a job is executed.

type JobConfig

type JobConfig struct {
	// Initial delay to start the job. Zero value will schedule the job immediately. Negative value will wait for next scheduled interval.
	InitialDelay config.Duration `json:"initial_delay" default:"0s"`

	// Time period between job executions. Zero or negative value means that the job will never execute periodically.
	ExecutionPeriod config.Duration `json:"execution_period" default:"10s"`

	// Execution timeout
	ExecutionTimeout config.Duration `json:"execution_timeout" validate:"gte=0s" default:"5s"`

	// Sets whether the job is initially healthy
	InitiallyHealthy bool `json:"initially_healthy" default:"false"`
}

JobConfig is config for Job swagger:model +kubebuilder:object:generate=true

func (*JobConfig) DeepCopy added in v0.1.2

func (in *JobConfig) DeepCopy() *JobConfig

DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new JobConfig.

func (*JobConfig) DeepCopyInto added in v0.1.2

func (in *JobConfig) DeepCopyInto(out *JobConfig)

DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.

type JobGroup

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

JobGroup tracks a group of jobs. It is responsible for scheduling jobs and keeping track of their statuses.

func NewJobGroup

func NewJobGroup(
	statusRegistry status.Registry,
	maxConcurrentJobs int,
	schedulerMode SchedulerMode,
	gws GroupWatchers,
) (*JobGroup, error)

NewJobGroup creates a new JobGroup.

func (*JobGroup) DeregisterAll

func (jg *JobGroup) DeregisterAll()

DeregisterAll deregisters all Jobs from the JobGroup.

func (*JobGroup) DeregisterJob

func (jg *JobGroup) DeregisterJob(name string) error

DeregisterJob deregisters a Job from the JobGroup. It returns an error if the job is not registered. It also stops the job's executor.

func (*JobGroup) GetStatusRegistry added in v0.4.0

func (jg *JobGroup) GetStatusRegistry() status.Registry

GetStatusRegistry returns the registry of the JobGroup.

func (*JobGroup) IsHealthy

func (jg *JobGroup) IsHealthy() bool

IsHealthy returns true if the job is healthy.

func (*JobGroup) JobInfo

func (jg *JobGroup) JobInfo(name string) *JobInfo

JobInfo returns the information related to a job with given name.

func (*JobGroup) RegisterJob

func (jg *JobGroup) RegisterJob(job Job, config JobConfig) error

RegisterJob registers a new Job in a JobGroup. It returns an error if the job is already registered. It also starts the job's executor.

func (*JobGroup) Results

func (jg *JobGroup) Results() (*statusv1.GroupStatus, bool)

Results returns the results of all jobs in the JobGroup.

func (*JobGroup) Start

func (jg *JobGroup) Start() error

Start starts the JobGroup.

func (*JobGroup) Stop

func (jg *JobGroup) Stop() error

Stop stops the JobGroup.

func (*JobGroup) TriggerJob

func (jg *JobGroup) TriggerJob(name string)

TriggerJob triggers a Job in the JobGroup.

type JobGroupConfig

type JobGroupConfig struct {
	SchedulerConfig `json:",inline"`
}

JobGroupConfig holds configuration for JobGroup. swagger:model +kubebuilder:object:generate=true

func (*JobGroupConfig) DeepCopy added in v0.1.2

func (in *JobGroupConfig) DeepCopy() *JobGroupConfig

DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new JobGroupConfig.

func (*JobGroupConfig) DeepCopyInto added in v0.1.2

func (in *JobGroupConfig) DeepCopyInto(out *JobGroupConfig)

DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.

type JobGroupConstructor

type JobGroupConstructor struct {
	// Name of the job group - config key is <name> and statuses are updated under <name>.<job>
	Name string
	// Config key --  if it is empty then it is <name>.scheduler
	Key           string
	GW            GroupWatchers
	DefaultConfig JobGroupConfig
	SchedulerMode SchedulerMode
}

JobGroupConstructor holds fields to create annotated instances of JobGroup.

func (JobGroupConstructor) Annotate

func (jgc JobGroupConstructor) Annotate() fx.Option

Annotate provides annotated instances of JobGroup.

type JobInfo

type JobInfo struct {
	LastRunTime time.Time
	NextRunTime time.Time
	RunCount    int
}

JobInfo contains information such as run count, last run time, etc. for a Job.

type JobStats

type JobStats struct {
	Duration time.Duration
}

JobStats holds fields to track job statistics.

type JobWatcher

type JobWatcher interface {
	OnJobScheduled()
	OnJobCompleted(status *statusv1.Status, stats JobStats)
}

JobWatcher is used for tracking completion of Job.

type JobWatchers

type JobWatchers []JobWatcher

JobWatchers is a collection of JobWatcher.

func (JobWatchers) OnJobCompleted

func (jws JobWatchers) OnJobCompleted(status *statusv1.Status, jobStats JobStats)

OnJobCompleted calls OnJobCompleted for each JobWatcher in the collection.

func (JobWatchers) OnJobScheduled

func (jws JobWatchers) OnJobScheduled()

OnJobScheduled calls OnJobScheduled for each JobWatcher in the collection.

type MultiJob

type MultiJob struct {
	JobBase
	// contains filtered or unexported fields
}

MultiJob runs multiple jobs in asynchronous manner.

func NewMultiJob

func NewMultiJob(registry status.Registry, jws JobWatchers, gws GroupWatchers) *MultiJob

NewMultiJob creates a new instance of MultiJob.

func (*MultiJob) DeregisterAll

func (mj *MultiJob) DeregisterAll()

DeregisterAll removes all jobs from the MultiJob.

func (*MultiJob) DeregisterJob

func (mj *MultiJob) DeregisterJob(name string) error

DeregisterJob deregisters a job with the MultiJob.

func (*MultiJob) Execute

func (mj *MultiJob) Execute(ctx context.Context) (proto.Message, error)

Execute executes all jobs, collects that results, and returns the aggregated status.

func (*MultiJob) JobWatchers

func (mj *MultiJob) JobWatchers() JobWatchers

JobWatchers returns the list of job watchers.

func (*MultiJob) Name

func (mj *MultiJob) Name() string

Name returns the name of the job.

func (*MultiJob) RegisterJob

func (mj *MultiJob) RegisterJob(job Job) error

RegisterJob registers a job with the MultiJob.

type MultiJobConfig

type MultiJobConfig struct {
	JobConfig
}

MultiJobConfig holds configuration for MultiJob. swagger:model

type MultiJobConstructor

type MultiJobConstructor struct {
	DefaultConfig MultiJobConfig
	Name          string
	JobGroupName  string
	JWS           JobWatchers
	GWS           GroupWatchers
}

MultiJobConstructor holds fields to create annotated instance of MultiJob.

func (MultiJobConstructor) Annotate

func (mjc MultiJobConstructor) Annotate() fx.Option

Annotate provides annotated instance of MultiJob.

type SchedulerConfig

type SchedulerConfig struct {
	// Limits how many jobs can be running at the same time. This is useful when running resource intensive jobs and a precise start time is not critical. 0 = no limit.
	MaxConcurrentJobs int `json:"max_concurrent_jobs" validate:"gte=0" default:"0"`
}

SchedulerConfig holds configuration for job Scheduler. swagger:model +kubebuilder:object:generate=true

func (*SchedulerConfig) DeepCopy added in v0.1.2

func (in *SchedulerConfig) DeepCopy() *SchedulerConfig

DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new SchedulerConfig.

func (*SchedulerConfig) DeepCopyInto added in v0.1.2

func (in *SchedulerConfig) DeepCopyInto(out *SchedulerConfig)

DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.

type SchedulerMode

type SchedulerMode int8

SchedulerMode configures the scheduler's behavior when concurrency limit is applied.

const (
	// RescheduleMode - the default is that if a limit on maximum
	// concurrent jobs is set and the limit is reached, a job will
	// skip it's run and try again on the next occurrence in the schedule.
	RescheduleMode SchedulerMode = iota
	// WaitMode - if a limit on maximum concurrent jobs is set
	// and the limit is reached, a job will wait to try and run
	// until a spot in the limit is freed up.
	//
	// Note: this mode can produce unpredictable results as
	// job execution order isn't guaranteed. For example, a job that
	// executes frequently may pile up in the wait queue and be executed
	// many times back to back when the queue opens.
	WaitMode
)

type SelfChecksIn

type SelfChecksIn struct {
	fx.In

	Liveness  *MultiJob `name:"liveness.service"`
	Readiness *MultiJob `name:"readiness.service"`
}

SelfChecksIn holds parameters for RegisterSelfChecks.

Jump to

Keyboard shortcuts

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