tinyscheduler

README

tinyscheduler

A tiny cron scheduler for go

Features

• Run multiple jobs concurrently
• Specify Job frequency declaratively. i.e 1s, 3m, 5h, 300ms, etc.
• Prometheus compatible metrics
• Job Timeout
• Jobs Cancellation and Scheduler cleaning up

Installation

go get github.com/adhaamehab/tinyscheduler

Usage

package main

import (
	"fmt"
	"time"

	ts "github.com/adhaamehab/tinyscheduler"
)

func main() {
	jobs := []*ts.Job{
		ts.NewJob("Tick1", "1s", "1s", func() {
			time.Sleep(500 * time.Millisecond)
			fmt.Println("Tick 1")
		}, make([]interface{}, 0)),
		ts.NewJob("Tick2", "3s", "1s", func() { fmt.Println("Tick 2") }, make([]interface{}, 0)),
	}
	s, err := ts.NewScheduler(ts.C{MetricsPort: 9092})
	if err != nil {
		panic(err)
	}
	s.AddMany(jobs)

	// This will block the execution
	// You can use s.StartAsync() to run the scheduler in the background
	s.Start()

}

Components

Scheduler

Scheduler is the core component in tinyscheduler. It gives the user a configurable api to manage the jobs and how they run.It is composed of Executor and Jobs. User has full control over both components (explained later).

Executor

The idea of the executor is to allow scheduler to work different type of execution algorithms. This will enable users to extend the package to run as a distributed scheduler.

Job

A Job to a function is the same idea in K8s Pod & container. A Job is just a wrapper for a go function with some additional meta data about how to run this function in the background.

The downside to the current Job implementation is the ability to cancel a singe Job. Also handling the Job overlapping might be tricky.

Monitoring

tinyscheduler expose a MetricsPort which expose prometheus metrics. This metrics tracks the number of running jobs and executors in addition to the execution time for each individual Job.

User can shut down the MetricsServer using StopMetricsServer() method. But it's yet a limitation to stop that. Using configuration. This should be enhanced later.

Road Map

• Ability to cancel individual jobs.
• Ability to configure metrics server before initializing the Scheduler.
• More default executors. (i.e) Distributed executor, etc.
• User provided logger and log mode.
• Mocking and better test coverage

Documentation

Index

• type C
• type Executor
  • func DefaultExecutor() (*Executor, error)
  • func NewExecutor(name string, jobChan chan *Job, stopChan chan struct{}) (*Executor, error)
  • func (e *Executor) Run()
• type Job
  • func NewJob(name string, frequency string, timeout string, function interface{}, ...) *Job
• type Scheduler
  • func NewScheduler(config C) (*Scheduler, error)
  • func (s *Scheduler) Add(j *Job)
  • func (s *Scheduler) AddMany(jobs []*Job)
  • func (s *Scheduler) CleanUp()
  • func (s *Scheduler) Start()
  • func (s *Scheduler) StartAsync()
  • func (s *Scheduler) Stop()
  • func (s *Scheduler) StopMetricsServer()

Types

type C

type C struct {
	MetricsPort int
	Executor    *Executor
}

type Executor

type Executor struct {
	Name string
	Jobs chan *Job
	Stop chan struct{}
}

Executor describes the worker pool that receives jobs dispatched from Scheduler. Composing the Executor inside Scheduler will give the opportunity to a more plugable executors which mean User can write their own Executor algorithms.

func DefaultExecutor

func DefaultExecutor() (*Executor, error)

func NewExecutor

func NewExecutor(name string, jobChan chan *Job, stopChan chan struct{}) (*Executor, error)

func (*Executor) Run

func (e *Executor) Run()

type Job

type Job struct {
	sync.RWMutex
	Name         string
	Function     interface{}
	Params       []interface{}
	LastRun      time.Time
	NextRun      time.Time
	Count        int           // Number of times the Job ran
	NextDeadline time.Time     // Next deadline for the Job to be running
	Timeout      time.Duration // Max execution time to run the task
	Frequency    time.Duration // Scheduling frequency that the job should be run with
}

Job abstract Background job with its metadata This struct wraps the Function & Params with all the required Data to successfully run the Job in the background It also include locking to avoid updates while scheduling/running which will yield weird behavior only. Time and Frequencies are in seconds only. Adding duration formats will be fairly straight forward since we can abstract this detail later. But using "seconds" only will help simplify the scheduling algorithm

func NewJob

func NewJob(name string, frequency string, timeout string, function interface{}, params []interface{}) *Job

NewJob A custom constructor for Job

type Scheduler

type Scheduler struct {
	ExecutorName  *string
	IsRunning     bool
	Jobs          []*Job
	MetricsServer *http.Server
	// contains filtered or unexported fields
}

Scheduler describe Scheduling API and dispatcher You can provide custom executor by setting `Executor` attribute inside scheduler config object `C` while calling `NewScheduler`

func NewScheduler

func NewScheduler(config C) (*Scheduler, error)

NewScheduler create and return a new scheduler instance You can later start add tasks using `scheduler.Add(j *Job)` method This function also expose custom prometheus metrics using the `/$PORT/-/metrics` endpoint. By default $PORT=9091 but you can override that by passing the port as the first arg to the scheduler passed as a string

func (*Scheduler) Add

func (s *Scheduler) Add(j *Job)

func (*Scheduler) AddMany

func (s *Scheduler) AddMany(jobs []*Job)

func (*Scheduler) CleanUp

func (s *Scheduler) CleanUp()

CleanUp stops all running jobs, remove all scheduled jobs and stops metricsServer

func (*Scheduler) Start

func (s *Scheduler) Start()

Start run the scheduler idempotently and block program execution by default. To run the scheduler in the background use StartAsync

func (*Scheduler) StartAsync

func (s *Scheduler) StartAsync()

StartAsync run the scheduler in the background without blocking the main goroutine

func (*Scheduler) Stop

func (s *Scheduler) Stop()

Stop turn off scheduler and stop the executor

func (*Scheduler) StopMetricsServer

func (s *Scheduler) StopMetricsServer()

StopMetricsServer gracefully shut down running MetricsServer after 2 seconds