agscheduler

package module
v0.11.0 Latest Latest
Warning

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

 Go to latest Published: Jun 20, 2024 License: MIT Imports: 22 Imported by: 0

README

AGScheduler

test codecov Go Report Card Go Reference GitHub release (with filter) GitHub go.mod Go version (subdirectory of monorepo) license

Advanced Golang Scheduler (AGScheduler) is a task scheduling library for Golang that supports multiple scheduling types, dynamically changing and persistent jobs, job queues, job result collection, event listening, remote call, and cluster

English | 简体中文

Features

  • Supports three scheduling types
    • One-off execution
    • Interval execution
    • Cron-style scheduling
  • Supports multiple job store methods
  • Supports multiple job queues
  • Supports multiple job result backends
    • Memory (Cluster mode is not supported)
    • GORM (any RDBMS supported by GORM works)
    • MongoDB
  • Supports event listening
    • Scheduler event
    • Job event
  • Supports remote call
  • Supports cluster
    • Remote worker nodes
    • Scheduler high availability (Experimental)

Framework

Framework

Installation

go get -u github.com/agscheduler/agscheduler

Usage

package main

import (
	"context"
	"fmt"
	"log/slog"
	"time"

	"github.com/agscheduler/agscheduler"
	"github.com/agscheduler/agscheduler/stores"
)

func printMsg(ctx context.Context, j agscheduler.Job) (result string) {
	slog.Info(fmt.Sprintf("Run job `%s` %s\n\n", j.FullName(), j.Args))
	return
}

func main() {
	agscheduler.RegisterFuncs(
		agscheduler.FuncPkg{Func: printMsg},
	)

	scheduler := &agscheduler.Scheduler{}

	store := &stores.MemoryStore{}
	scheduler.SetStore(store)

	job1 := agscheduler.Job{
		Name:     "Job1",
		Type:     agscheduler.JOB_TYPE_INTERVAL,
		Interval: "2s",
		Timezone: "UTC",
		Func:     printMsg,
		Args:     map[string]any{"arg1": "1", "arg2": "2", "arg3": "3"},
	}
	job1, _ = scheduler.AddJob(job1)
	slog.Info(fmt.Sprintf("%s.\n\n", job1))

	job2 := agscheduler.Job{
		Name:     "Job2",
		Type:     agscheduler.JOB_TYPE_CRON,
		CronExpr: "*/1 * * * *",
		Timezone: "Asia/Shanghai",
		FuncName: "main.printMsg",
		Args:     map[string]any{"arg4": "4", "arg5": "5", "arg6": "6", "arg7": "7"},
	}
	job2, _ = s.AddJob(job2)
	slog.Info(fmt.Sprintf("%s.\n\n", job2))

	job3 := agscheduler.Job{
		Name:     "Job3",
		Type:     agscheduler.JOB_TYPE_DATETIME,
		StartAt:  "2023-09-22 07:30:08",
		Timezone: "America/New_York",
		Func:     printMsg,
		Args:     map[string]any{"arg8": "8", "arg9": "9"},
	}
	job3, _ = s.AddJob(job3)
	slog.Info(fmt.Sprintf("%s.\n\n", job3))

	jobs, _ := s.GetAllJobs()
	slog.Info(fmt.Sprintf("Scheduler get all jobs %s.\n\n", jobs))

	scheduler.Start()

	select {}
}

Register Funcs

Since golang can't serialize functions, you need to register them with RegisterFuncs before scheduler.Start()

Queue

mq := &queues.MemoryQueue{}
broker := &agscheduler.Broker{
	Queues: map[string]agscheduler.QueuePkg{
		"default": {
			Queue:   mq,
			Workers: 2,
		},
	},
}

scheduler.SetBroker(broker)

Result Collection

mb := &backends.MemoryBackend{}
recorder := &agscheduler.Recorder{Backend: mb}

scheduler.SetRecorder(recorder)

job, _ = scheduler.AddJob(job)
records, _ := recorder.GetRecords(job.Id)

Event listening

func jobCallback(ep agscheduler.EventPkg) {
	slog.Info(fmt.Sprintf("Event code: `%d`, job `%s`.\n\n", ep.Event, ep.JobId))
}

......

listener := &agscheduler.Listener{
	Callbacks: []agscheduler.CallbackPkg{
		{
			Callback: jobCallback,
			Event:    agscheduler.EVENT_JOB_ADDED | agscheduler.EVENT_JOB_DELETED,
		},
	},
}

scheduler.SetListener(listener)

gRPC

// Server
grservice := services.GRPCService{
	Scheduler: scheduler,
	Address:   "127.0.0.1:36360",
	// PasswordSha2: "xxxxxx",
}
grservice.Start()

// Client
conn, _ := grpc.NewClient("127.0.0.1:36360", grpc.WithTransportCredentials(insecure.NewCredentials()))
client := pb.NewSchedulerClient(conn)
client.AddJob(ctx, job)

HTTP

// Server
hservice := services.HTTPService{
	Scheduler: scheduler,
	Address:   "127.0.0.1:36370",
	// PasswordSha2: "xxxxxx",
}
hservice.Start()

// Client
mJob := map[string]any{...}
bJob, _ := json.Marshal(mJob)
resp, _ := http.Post("http://127.0.0.1:36370/scheduler/job", "application/json", bytes.NewReader(bJob))

Cluster

// Main Node
cnMain := &agscheduler.ClusterNode{
	Endpoint:     "127.0.0.1:36380",
	EndpointGRPC: "127.0.0.1:36360",
	EndpointHTTP: "127.0.0.1:36370",
	Queue:        "default",
}
schedulerMain.SetClusterNode(ctx, cnMain)
cserviceMain := &services.ClusterService{Cn: cnMain}
cserviceMain.Start()

// Worker Node
cnNode := &agscheduler.ClusterNode{
	EndpointMain: "127.0.0.1:36380",
	Endpoint:     "127.0.0.1:36381",
	EndpointGRPC: "127.0.0.1:36361",
	EndpointHTTP: "127.0.0.1:36371",
	Queue:        "worker",
}
schedulerNode.SetClusterNode(ctx, cnNode)
cserviceNode := &services.ClusterService{Cn: cnNode}
cserviceNode.Start()

Cluster HA (High Availability, Experimental)

// HA requires the following conditions to be met:
//
// 1. The number of HA nodes in the cluster must be odd
// 2. All HA nodes need to connect to the same Store (excluding `MemoryStore`)
// 3. The `Mode` of the `ClusterNode` needs to be set to `HA`
// 4. The main HA node must be started first

// Main HA Node
cnMain := &agscheduler.ClusterNode{..., Mode: "HA"}

// HA Node
cnNode1 := &agscheduler.ClusterNode{..., Mode: "HA"}
cnNode2 := &agscheduler.ClusterNode{..., Mode: "HA"}

// Worker Node
cnNode3 := &agscheduler.ClusterNode{...}

Base API

gRPC Function HTTP Method HTTP Path
GetInfo GET /info
GetFuncs GET /funcs

Scheduler API

gRPC Function HTTP Method HTTP Path
AddJob POST /scheduler/job
GetJob GET /scheduler/job/:id
GetAllJobs GET /scheduler/jobs
UpdateJob PUT /scheduler/job
DeleteJob DELETE /scheduler/job/:id
DeleteAllJobs DELETE /scheduler/jobs
PauseJob POST /scheduler/job/:id/pause
ResumeJob POST /scheduler/job/:id/resume
RunJob POST /scheduler/job/run
ScheduleJob POST /scheduler/job/schedule
Start POST /scheduler/start
Stop POST /scheduler/stop

Broker API

gRPC Function HTTP Method HTTP Path
GetQueues GET /broker/queues

Recorder API

gRPC Function HTTP Method HTTP Path
GetRecords GET /recorder/records/:job_id
GetAllRecords GET /recorder/records
DeleteRecords DELETE /recorder/records/:job_id
DeleteAllRecords DELETE /recorder/records

Cluster API

gRPC Function HTTP Method HTTP Path
GetNodes GET /cluster/nodes

Examples

Complete example

Development

# Clone code
git clone git@github.com:agscheduler/agscheduler.git

# Working directory
cd agscheduler

# Install dependencies
make install

# Up CI services
make up-ci-services

# Run check
make check-all

Cli

cargo install agscheduler-cli

Web

docker run --rm -p 8080:80 ghcr.io/agscheduler/agscheduler-web:latest

Thanks

APScheduler

simple-raft

Documentation

Index

Constants

View Source 
const (
	JOB_TYPE_DATETIME = "datetime"
	JOB_TYPE_INTERVAL = "interval"
	JOB_TYPE_CRON     = "cron"
)

constant indicating a job's type

View Source 
const (
	JOB_STATUS_RUNNING = "running"
	JOB_STATUS_PAUSED  = "paused"
)

constant indicating a job's status

View Source 
const (
	EVENT_SCHEDULER_STARTED event = 1 << iota
	EVENT_SCHEDULER_STOPPED

	EVENT_JOB_ADDED
	EVENT_JOB_UPDATED
	EVENT_JOB_DELETED
	EVENT_ALL_JOBS_DELETED
	EVENT_JOB_PAUSED
	EVENT_JOB_RESUMED
	EVENT_JOB_EXECUTED
	EVENT_JOB_ERROR
	EVENT_JOB_TIMEOUT

	EVENT_ALL event = EVENT_SCHEDULER_STARTED | EVENT_SCHEDULER_STOPPED |
		EVENT_JOB_ADDED | EVENT_JOB_UPDATED |
		EVENT_JOB_DELETED | EVENT_ALL_JOBS_DELETED |
		EVENT_JOB_PAUSED | EVENT_JOB_RESUMED |
		EVENT_JOB_EXECUTED | EVENT_JOB_ERROR | EVENT_JOB_TIMEOUT
)

constant indicating the event.

View Source 
const (
	RECORD_STATUS_RUNNING   = "running"
	RECORD_STATUS_COMPLETED = "completed"
	RECORD_STATUS_ERROR     = "error"
	RECORD_STATUS_TIMEOUT   = "timeout"
)

constant indicating the status of the job record

View Source 
const Version = "0.11.0"

Variables

View Source 
var FuncMap = make(map[string]FuncPkg)

Record the actual path of function and the corresponding function. Since golang can't serialize functions, need to register them with `RegisterFuncs` before using it.

View Source 
var GetBroker = (*Scheduler).getBroker
View Source 
var GetClusterNode = (*Scheduler).getClusterNode
View Source 
var GetListener = (*Scheduler).getListener
View Source 
var GetRecorder = (*Scheduler).getRecorder
View Source 
var GetStore = (*Scheduler).getStore

Functions

func CalcNextRunTime 

func CalcNextRunTime(j Job) (time.Time, error)

Calculate the next run time, different job type will be calculated in different ways, when the job is paused, will return `9999-09-09 09:09:09`.

func FuncMapReadable 

func FuncMapReadable() []map[string]string

func GetNextRunTimeMax 

func GetNextRunTimeMax() (time.Time, error)

func JobMarshal added in v0.8.3 

func JobMarshal(j Job) ([]byte, error)

Serialize Job and convert to Bytes

func JobToPbJobPtr 

func JobToPbJobPtr(j Job) (*pb.Job, error)

Used to gRPC Protobuf

func JobsToPbJobsPtr 

func JobsToPbJobsPtr(js []Job) ([]*pb.Job, error)

Used to gRPC Protobuf

func QueueToPbQueuePtr added in v0.10.0 

func QueueToPbQueuePtr(q map[string]any) (*pb.Queue, error)

Used to gRPC Protobuf

func QueuesToPbQueuesPtr added in v0.10.0 

func QueuesToPbQueuesPtr(qs []map[string]any) ([]*pb.Queue, error)

Used to gRPC Protobuf

func RecordToPbRecordPtr added in v0.8.0 

func RecordToPbRecordPtr(r Record) (*pb.Record, error)

Used to gRPC Protobuf

func RecordsToPbRecordsPtr added in v0.8.0 

func RecordsToPbRecordsPtr(rs []Record) ([]*pb.Record, error)

Used to gRPC Protobuf

func RegisterFuncs 

func RegisterFuncs(fps ...FuncPkg)

Types

type Backend added in v0.8.0 

type Backend interface {
	// Backend name.
	Name() string

	// Initialization functions for each backend,
	// called when the scheduler run `SetBackend`.
	Init() error

	// Record the metadata of the job to this backend.
	RecordMetadata(r Record) error

	// Record the result of the job run to this backend.
	RecordResult(id uint64, status string, result string) error

	// Get records by job id from this backend.
	//  @return records, total, error.
	GetRecords(jId string, page, pageSize int) ([]Record, int64, error)

	// Get all records from this backend.
	//  @return records, total, error.
	GetAllRecords(page, pageSize int) ([]Record, int64, error)

	// Delete records by job id from this backend.
	DeleteRecords(jId string) error

	// Delete all records from this backend.
	DeleteAllRecords() error

	// Clear all resources bound to this backend.
	Clear() error
}

Defines the interface that each backend must implement.

type Broker added in v0.7.0 

type Broker struct {
	// Job queues.
	// def: map[<queue>]QueuePkg
	Queues map[string]QueuePkg
	// contains filtered or unexported fields
}

When using a Broker, job scheduling is done in queue and no longer directly via API calls.

func (*Broker) Clear added in v0.10.0 

func (b *Broker) Clear(queue string) error

func (*Broker) CountJobs added in v0.10.0 

func (b *Broker) CountJobs(queue string) (int, error)

func (*Broker) GetQueues added in v0.10.0 

func (b *Broker) GetQueues() []map[string]any

type CallbackPkg added in v0.11.0 

type CallbackPkg struct {
	Callback func(ep EventPkg)
	Event    event
}

type ClusterNode 

type ClusterNode struct {
	// Main node RPC listening address.
	// If you are the main, `EndpointMain` is the same as `Endpoint`.
	// Default: `127.0.0.1:36380`
	EndpointMain string
	// The unique identifier of this node.
	// RPC listening address.
	// Used to expose the cluster's internal API.
	// Default: `127.0.0.1:36380`
	Endpoint string
	// gRPC listening address.
	// Used to expose the external API.
	// Default: `127.0.0.1:36360`
	EndpointGRPC string
	// HTTP listening address.
	// Used to expose the external API.
	// Default: `127.0.0.1:36370`
	EndpointHTTP string
	// Useful when a job specifies a queue.
	// A queue can correspond to multiple nodes.
	// Default: `default`
	Queue string
	// Node mode, for Scheduler high availability.
	// If the value is `HA`, the node will join the raft group.
	// Default: “, Options `HA`
	Mode string

	// Bind to each other and the Scheduler.
	Scheduler *Scheduler

	// For Scheduler high availability.
	// Bind to each other and the Raft.
	Raft *Raft
	// Used to mark the status of Cluster Scheduler operation.
	SchedulerCanStart bool
	// contains filtered or unexported fields
}

Each node provides `Cluster RPC`, `gRPC`, `HTTP` services, but only the main node starts the scheduler, the other worker nodes register with the main node and then run jobs from the main node via the RPC's `RunJob` API.

func (*ClusterNode) GetEndpointMain 

func (cn *ClusterNode) GetEndpointMain() string

func (*ClusterNode) HANodeMap 

func (cn *ClusterNode) HANodeMap() TypeNodeMap

func (*ClusterNode) IsMainNode 

func (cn *ClusterNode) IsMainNode() bool

func (*ClusterNode) MainNode 

func (cn *ClusterNode) MainNode() map[string]any

func (*ClusterNode) NodeMapCopy 

func (cn *ClusterNode) NodeMapCopy() TypeNodeMap

func (*ClusterNode) NodeMapToPbNodesPtr 

func (cn *ClusterNode) NodeMapToPbNodesPtr() map[string]*pb.Node

Used to gRPC Protobuf

func (*ClusterNode) RPCHeartbeat 

func (cn *ClusterNode) RPCHeartbeat(args *Node, reply *Node)

RPC API

func (*ClusterNode) RPCRegister 

func (cn *ClusterNode) RPCRegister(args *Node, reply *Node)

RPC API

func (*ClusterNode) RegisterNodeRemote 

func (cn *ClusterNode) RegisterNodeRemote(ctx context.Context) error

Used for worker node

After initialization, node need to register with the main node and synchronize cluster node information.

func (*ClusterNode) SetEndpointMain 

func (cn *ClusterNode) SetEndpointMain(endpoint string)

type EventPkg added in v0.11.0 

type EventPkg struct {
	Event event
	JobId string
	Data  any
}

type FuncPkg 

type FuncPkg struct {
	Func func(context.Context, Job) (result string)
	// About this function.
	Info string
}

type FuncUnregisteredError 

type FuncUnregisteredError string

func (FuncUnregisteredError) Error 

func (e FuncUnregisteredError) Error() string

type HeartbeatArgs 

type HeartbeatArgs struct {
	Term           int
	LeaderEndpoint string

	SchedulerCanStart bool
}

type HeartbeatReply 

type HeartbeatReply struct {
	Term int
}

type Job 

type Job struct {
	// The unique identifier of this job, automatically generated.
	// It should not be set manually.
	Id string `json:"id"`
	// User defined.
	Name string `json:"name"`
	// Optional: `JOB_TYPE_DATETIME` | `JOB_TYPE_INTERVAL` | `JOB_TYPE_CRON`
	Type string `json:"type"`
	// It can be used when Type is `JOB_TYPE_DATETIME`.
	// e.g. `2023-09-22 07:30:08`
	StartAt string `json:"start_at"`
	// This field is useless.
	EndAt string `json:"end_at"`
	// It can be used when Type is `JOB_TYPE_INTERVAL`.
	// e.g. `2s`
	Interval string `json:"interval"`
	// It can be used when Type is `JOB_TYPE_CRON`.
	// See `https://en.wikipedia.org/wiki/Cron`.
	// e.g. `*/1 * * * *`
	CronExpr string `json:"cron_expr"`
	// Refer to `time.LoadLocation`.
	// See `https://en.wikipedia.org/wiki/List_of_tz_database_time_zones`
	// Default: `UTC`
	Timezone string `json:"timezone"`
	// The job actually runs the function,
	// and you need to register it through 'RegisterFuncs' before using it.
	// Since it cannot be stored by serialization,
	// when using gRPC or HTTP calls, you should use `FuncName`.
	Func func(context.Context, Job) (result string) `json:"-"`
	// The actual path of `Func`.
	// This field has a higher priority than `Func`
	// e.g. `main.xxxFunc`
	//      `github.com/agscheduler/agscheduler/examples.PrintMsg`
	FuncName string `json:"func_name"`
	// Arguments for `Func`.
	Args map[string]any `json:"args"`
	// The running timeout of `Func`.
	// Default: `1h`
	Timeout string `json:"timeout"`
	// Used in cluster mode, if empty, randomly pick a node to run `Func`,
	// but when broker exist, if empty, randomly pick a queue to run `Func`.
	Queues []string `json:"queues"`

	// Automatic update, not manual setting.
	LastRunTime time.Time `json:"last_run_time"`
	// Automatic update, not manual setting.
	// When the job is paused, this field is set to `9999-09-09 09:09:09`.
	NextRunTime time.Time `json:"next_run_time"`
	// Optional: `JOB_STATUS_RUNNING` | `JOB_STATUS_PAUSED`
	// It should not be set manually.
	Status string `json:"status"`
}

Carry the information of the scheduled job

func JobUnmarshal added in v0.8.3 

func JobUnmarshal(bJ []byte) (Job, error)

Deserialize Bytes and convert to Job

func PbJobPtrToJob 

func PbJobPtrToJob(pbJob *pb.Job) Job

Used to gRPC Protobuf

func PbJobsPtrToJobs 

func PbJobsPtrToJobs(pbJs []*pb.Job) []Job

Used to gRPC Protobuf

func (Job) DeepCopy added in v0.7.0 

func (j Job) DeepCopy() (Job, error)

func (*Job) FullName 

func (j *Job) FullName() string

func (*Job) LastRunTimeWithTimezone 

func (j *Job) LastRunTimeWithTimezone() time.Time

func (*Job) NextRunTimeWithTimezone 

func (j *Job) NextRunTimeWithTimezone() time.Time

func (Job) String 

func (j Job) String() string

type JobNotFoundError 

type JobNotFoundError string

func (JobNotFoundError) Error 

func (e JobNotFoundError) Error() string

type JobSlice 

type JobSlice []Job

`sort.Interface`, sorted by 'NextRunTime', ascend.

func (JobSlice) Len 

func (js JobSlice) Len() int

func (JobSlice) Less 

func (js JobSlice) Less(i, j int) bool

func (JobSlice) Swap 

func (js JobSlice) Swap(i, j int)

type JobTimeoutError 

type JobTimeoutError struct {
	FullName string
	Timeout  string
	Err      error
}

func (*JobTimeoutError) Error 

func (e *JobTimeoutError) Error() string

type Listener added in v0.11.0 

type Listener struct {
	Callbacks []CallbackPkg
}

Event listener.

type Node 

type Node struct {
	EndpointMain string
	Endpoint     string
	EndpointGRPC string
	EndpointHTTP string
	Queue        string
	Mode         string

	NodeMap TypeNodeMap
}

type Queue added in v0.7.0 

type Queue interface {
	// Queue name.
	Name() string

	// Initialization functions for each queue,
	// called when the scheduler run `SetBroker`.
	Init(ctx context.Context) error

	// Push job to this queue.
	PushJob(bJ []byte) error

	// Pull job from this queue.
	PullJob() <-chan []byte

	// Count the number of jobs in this queue.
	//  @return -1, nil, if the queue does not support this feature or error.
	CountJobs() (int, error)

	// Clear all resources bound to this queue.
	Clear() error
}

Defines the interface that each queue must implement.

type QueuePkg added in v0.10.0 

type QueuePkg struct {
	Queue Queue
	// Number of workers.
	// Default: `2`
	Workers int
}

type Raft 

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

func (*Raft) RPCHeartbeat 

func (rf *Raft) RPCHeartbeat(args HeartbeatArgs, reply *HeartbeatReply) error

func (*Raft) RPCRequestVote 

func (rf *Raft) RPCRequestVote(args VoteArgs, reply *VoteReply) error

type Record added in v0.8.0 

type Record struct {
	// Unique Id
	Id uint64 `json:"id"`
	// Job id
	JobId string `json:"job_id"`
	// Job name
	JobName string `json:"job_name"`
	// Optional: `RECORD_STATUS_RUNNING` | `RECORD_STATUS_COMPLETED` | `RECORD_STATUS_ERROR` | `RECORD_STATUS_TIMEOUT`
	Status string `json:"status"`
	// The result of the job run
	Result string `json:"result"`
	// Start time
	StartAt time.Time `json:"start_at"`
	// End time
	EndAt time.Time `json:"end_at"`
}

Carry the information of the job run.

func PbRecordPtrToRecord added in v0.8.0 

func PbRecordPtrToRecord(pbRecord *pb.Record) Record

Used to gRPC Protobuf

func PbRecordsPtrToRecords added in v0.8.0 

func PbRecordsPtrToRecords(pbRs []*pb.Record) []Record

Used to gRPC Protobuf

type RecordSlice added in v0.8.0 

type RecordSlice []Record

`sort.Interface`, sorted by 'StartAt', descend.

func (RecordSlice) Len added in v0.8.0 

func (rs RecordSlice) Len() int

func (RecordSlice) Less added in v0.8.0 

func (rs RecordSlice) Less(i, j int) bool

func (RecordSlice) Swap added in v0.8.0 

func (rs RecordSlice) Swap(i, j int)

type Recorder added in v0.8.0 

type Recorder struct {
	// Record store
	// It should not be used directly.
	Backend Backend
	// contains filtered or unexported fields
}

When using a Recorder, the results of the job runs will be recorded to the specified backend.

func (*Recorder) Clear added in v0.8.0 

func (r *Recorder) Clear() error

func (*Recorder) DeleteAllRecords added in v0.8.0 

func (r *Recorder) DeleteAllRecords() error

func (*Recorder) DeleteRecords added in v0.8.0 

func (r *Recorder) DeleteRecords(jId string) error

func (*Recorder) GetAllRecords added in v0.8.0 

func (r *Recorder) GetAllRecords(page, pageSize int) ([]Record, int64, error)

func (*Recorder) GetRecords added in v0.8.0 

func (r *Recorder) GetRecords(jId string, page, pageSize int) ([]Record, int64, error)

func (*Recorder) RecordMetadata added in v0.8.0 

func (r *Recorder) RecordMetadata(j Job) (id uint64, err error)

func (*Recorder) RecordResult added in v0.8.0 

func (r *Recorder) RecordResult(id uint64, status string, result string) error

type Role 

type Role int
const (
	RAFT_FOLLOWER Role = iota + 1
	RAFT_CANDIDATE
	RAFT_LEADER
)

type Scheduler 

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

In standalone mode, the scheduler only needs to run jobs on a regular basis. In cluster mode, the scheduler also needs to be responsible for allocating jobs to cluster nodes.

func (*Scheduler) AddJob 

func (s *Scheduler) AddJob(j Job) (Job, error)

func (*Scheduler) DeleteAllJobs 

func (s *Scheduler) DeleteAllJobs() error

func (*Scheduler) DeleteJob 

func (s *Scheduler) DeleteJob(id string) error

func (*Scheduler) GetAllJobs 

func (s *Scheduler) GetAllJobs() ([]Job, error)

func (*Scheduler) GetJob 

func (s *Scheduler) GetJob(id string) (Job, error)

func (*Scheduler) HasBroker added in v0.8.0 

func (s *Scheduler) HasBroker() bool

func (*Scheduler) HasListener added in v0.11.0 

func (s *Scheduler) HasListener() bool

func (*Scheduler) HasRecorder added in v0.8.0 

func (s *Scheduler) HasRecorder() bool

func (*Scheduler) Info 

func (s *Scheduler) Info() map[string]any

func (*Scheduler) IsClusterMode 

func (s *Scheduler) IsClusterMode() bool

func (*Scheduler) IsRunning 

func (s *Scheduler) IsRunning() bool

func (*Scheduler) PauseJob 

func (s *Scheduler) PauseJob(id string) (Job, error)

func (*Scheduler) ResumeJob 

func (s *Scheduler) ResumeJob(id string) (Job, error)

func (*Scheduler) RunJob 

func (s *Scheduler) RunJob(j Job) error

func (*Scheduler) ScheduleJob 

func (s *Scheduler) ScheduleJob(j Job) error

Select a worker node or queue.

func (*Scheduler) SetBroker added in v0.7.0 

func (s *Scheduler) SetBroker(ctx context.Context, brk *Broker) error

Bind the broker

func (*Scheduler) SetClusterNode 

func (s *Scheduler) SetClusterNode(ctx context.Context, cn *ClusterNode) error

Bind the cluster node

func (*Scheduler) SetListener added in v0.11.0 

func (s *Scheduler) SetListener(lis *Listener) error

Bind the listener

func (*Scheduler) SetRecorder added in v0.8.0 

func (s *Scheduler) SetRecorder(rec *Recorder) error

Bind the recorder

func (*Scheduler) SetStore 

func (s *Scheduler) SetStore(sto Store) error

Bind the store

func (*Scheduler) Start 

func (s *Scheduler) Start()

In addition to being called manually, it is also called after `AddJob`.

func (*Scheduler) Stop 

func (s *Scheduler) Stop()

In addition to being called manually, there is no job in store that will also be called.

func (*Scheduler) UpdateJob 

func (s *Scheduler) UpdateJob(j Job) (Job, error)

type Store 

type Store interface {
	// Store name.
	Name() string

	// Initialization functions for each store,
	// called when the scheduler run `SetStore`.
	Init() error

	// Add job to this store.
	AddJob(j Job) error

	// Get the job from this store.
	//  @return error `JobNotFoundError` if there are no job.
	GetJob(id string) (Job, error)

	// Get all jobs from this store.
	GetAllJobs() ([]Job, error)

	// Update job in store with a newer version.
	UpdateJob(j Job) error

	// Delete the job from this store.
	DeleteJob(id string) error

	// Delete all jobs from this store.
	DeleteAllJobs() error

	// Get the earliest next run time of all the jobs stored in this store,
	// or `time.Time{}` if there are no job.
	// Used to set the wakeup interval for the scheduler.
	GetNextRunTime() (time.Time, error)

	// Clear all resources bound to this store.
	Clear() error
}

Defines the interface that each store must implement.

type TypeNodeMap 

type TypeNodeMap map[string]map[string]any

type VoteArgs 

type VoteArgs struct {
	Term              int
	CandidateEndpoint string
}

type VoteReply 

type VoteReply struct {
	Term        int
	VoteGranted bool
}

Source Files

View all Source files

Directories

Path Synopsis
backends
cluster
event
grpc
http
queues
stores
proto

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.