README
¶
Gds
A Raft-based embedded job scheduler for your Go projects.
Features
- Disaster tolerance: if not all of the hosts are down, the job should be fired successfully.
- Validity: only one host runs each job.
- Schedule strategy.
- Automatic panic recovery.
Examples
The Examples folder contains a bunch of code samples you can look into.
Custom job types
All jobs implementations have to satisfy jobs/Job interface. Internal job state such as counters should only change in PostExecution method.
PostExecution method is called on each node in cluster in order to provide consistency.
NextTriggerTime method is idempotent and can be called many times, it have to return zero time.Time if job is finished.
Job struct fields have to be public in order to marshal/unmarshal correctly or covered by Marshal/Unmarshal methods.
Note that all nodes in cluster must register the same JobExecutors because all nodes store full jobs state.
Planned
- Project TODOs
- Remove
isp-logdependency, pass Raft logger to user API - Implement new Job types
- Add sync.Pool to cluster.prepareCommand()
- Library API improvements
Development
Gds uses go modules and Go 1.13
Checks before commit:
golangci-lint run
go test ./... -race
License
MIT
Documentation
¶
Index ¶
- Constants
- Variables
- type ClusterHandler
- func (cl *ClusterHandler) GetHandlers() map[uint64]func(store.WritableState, []byte) (interface{}, error)
- func (cl *ClusterHandler) HandleAcquireJobCommand(state store.WritableState, data []byte) (interface{}, error)
- func (cl *ClusterHandler) HandleAddPeerCommand(state store.WritableState, data []byte) (interface{}, error)
- func (cl *ClusterHandler) HandleDeleteJobCommand(state store.WritableState, data []byte) (interface{}, error)
- func (cl *ClusterHandler) HandleInsertJobCommand(state store.WritableState, data []byte) (interface{}, error)
- func (cl *ClusterHandler) HandleJobExecutedCommand(state store.WritableState, data []byte) (interface{}, error)
- func (cl *ClusterHandler) HandleRemovePeerCommand(state store.WritableState, data []byte) (interface{}, error)
- type JobExecutor
- type RaftAdapter
- type Scheduler
Constants ¶
View Source
const (
DefaultJobExecutionTimeout = 5 * time.Second
)
Variables ¶
View Source
var (
ErrJobTimeoutExceeded = errors.New("job execution timeout exceeded")
)
Functions ¶
This section is empty.
Types ¶
type ClusterHandler ¶
type ClusterHandler struct {
// contains filtered or unexported fields
}
func NewClusterHandler ¶
func NewClusterHandler(typeProvider provider.TypeProvider, executor executor) *ClusterHandler
func (*ClusterHandler) GetHandlers ¶
func (cl *ClusterHandler) GetHandlers() map[uint64]func(store.WritableState, []byte) (interface{}, error)
func (*ClusterHandler) HandleAcquireJobCommand ¶
func (cl *ClusterHandler) HandleAcquireJobCommand(state store.WritableState, data []byte) (interface{}, error)
func (*ClusterHandler) HandleAddPeerCommand ¶
func (cl *ClusterHandler) HandleAddPeerCommand(state store.WritableState, data []byte) (interface{}, error)
func (*ClusterHandler) HandleDeleteJobCommand ¶
func (cl *ClusterHandler) HandleDeleteJobCommand(state store.WritableState, data []byte) (interface{}, error)
func (*ClusterHandler) HandleInsertJobCommand ¶
func (cl *ClusterHandler) HandleInsertJobCommand(state store.WritableState, data []byte) (interface{}, error)
func (*ClusterHandler) HandleJobExecutedCommand ¶
func (cl *ClusterHandler) HandleJobExecutedCommand(state store.WritableState, data []byte) (interface{}, error)
func (*ClusterHandler) HandleRemovePeerCommand ¶
func (cl *ClusterHandler) HandleRemovePeerCommand(state store.WritableState, data []byte) (interface{}, error)
type JobExecutor ¶
type RaftAdapter ¶
type RaftAdapter struct {
Config config.ClusterConfiguration
RaftStore *store.Store
ClusterClient *cluster.Client
HTTPServer *http.Server
EtpServer etp.Server
// contains filtered or unexported fields
}
func NewRaftAdapter ¶
func NewRaftAdapter(cfg config.ClusterConfiguration, handler store.CommandsHandler, typeProvider provider.TypeProvider) (*RaftAdapter, error)
type Scheduler ¶
type Scheduler interface {
WaitCluster(context.Context) bool
Shutdown(ctx context.Context) error
RegisterExecutor(jobType string, executor JobExecutor, newJobFunc provider.NewJobFunc) Scheduler
UnregisterExecutor(jobType string)
ScheduleJob(job jobs.Job) error
GetJob(key string) (*store.JobInfo, error)
DeleteJob(key string) error
GetAllJobs() []store.JobInfo
GetJobsByType(jobType string) []store.JobInfo
}
func NewScheduler ¶
func NewScheduler(config config.ClusterConfiguration) (Scheduler, error)
Source Files
¶
Directories
Click to show internal directories.
Click to hide internal directories.