jobrunner

README

JobRunner

This Repo is forked from github.com/bamzi/jobrunner

※This is an enhanced version※

JobRunner is framework for performing work asynchronously, outside of the request flow. It comes with cron to schedule and queue job functions for processing at specified time.

It includes a live monitoring of current schedule and state of active jobs that can be outputed as JSON and it support get job execution history by go channel.

Install

go get github.com/magicLian/jobrunner

Supported Featured

All jobs are processed outside of the request flow

• Support time location in scheduling job
• Support user-defined job name
• Support task status real-time query
• Support to obtain task execution history records through Go channel

Live Monitoring

Basics

    jobrunner.Schedule("* */5 * * * *", DoSomething{}, "DoSomethingJob") // every 5min do something
    jobrunner.Schedule("@every 1h30m10s", ReminderEmails{}, "ReminderEmailsJob")
    jobrunner.Schedule("@midnight", DataStats{}, "DataStatsJob") // every midnight do this..
    jobrunner.Every(16*time.Minute, CleanS3{}, "CleanS3Job") // evey 16 min clean...
    jobrunner.In(10*time.Second, WelcomeEmail{}, "WelcomeEmailJob") // one time job. starts after 10sec
    jobrunner.Now(NowDo{}, "NowJob") // do the job as soon as it's triggered

More Detailed CRON Specs

Setup

standalone

package main

import "github.com/magicLian/jobrunner"

func main() {
    if err := os.Setenv("ZONEINFO", "/tzdata/data.zip"); err != nil {
	panic(err)
    }
    loc,err := time.loadLocation("Asia/Tokyo")
    if err != nil {
        panic(err)
    }
    jobrunner.Start(loc, false)
    jobrunner.Schedule("@every 5s", ReminderEmails{}, "reminderEmails")
}

// Job Specific Functions
type ReminderEmails struct {
    // filtered
}

// ReminderEmails.Run() will get triggered automatically.
func (e ReminderEmails) Run() {
    // Queries the DB
    // Sends some email
    fmt.Printf("Every 5 sec send reminder emails \n")
}

Integrate with Gin

package main

import (
	"fmt"
	"time"

	"github.com/gin-gonic/gin"
	"github.com/magicLian/jobrunner"
)

func main() {
	routes := gin.Default()

	// Resource to return the JSON data
	routes.GET("/jobrunner/json", JobJson)

	jobrunner.Start(nil, false)
	jobrunner.Every(10*time.Minute, DoSomeThing{}, "DoSomeThing")

	routes.Run(":8080")
}

type DoSomeThing struct {
}

func (d DoSomeThing) Run() {
	fmt.Printf("Start to do something")
}

func JobJson(c *gin.Context) {
	// returns a map[string]interface{} that can be marshalled as JSON
	c.JSON(200, jobrunner.StatusJson())
}

Get job execution result through go channel

func main() {
	...
	
	jobrunner.Start(nil, true)
	go getCronExecutionResults()
	jobrunner.Every(10*time.Minute, DoSomeThing{}, "DoSomeThing")

	


	...
}

func getCronExecutionResults(){
	for {
		select {
		case record := <-jobrunner.JobsExecutionStatusChan:
			fmt.Printf("Job exection status result:[%v]", record)
		}
	}
}

Credits

• revel jobs module - Origin of JobRunner
• robfig cron v3 - github.com/robfig/cron/v3

License

MIT

Documentation

Overview

A job runner for executing scheduled or ad-hoc tasks asynchronously from HTTP requests.

It adds a couple of features on top of the Robfig cron package:

1. Protection against job panics. (They print to ERROR instead of take down the process)
2. (Optional) Limit on the number of jobs that may run simulatenously, to limit resource consumption.
3. (Optional) Protection against multiple instances of a single job running concurrently. If one execution runs into the next, the next will be queued.
4. Cron expressions may be defined in app.conf and are reusable across jobs.
5. Job status reporting. [WIP]

Index

• Constants
• Variables
• func Entries() []cron.Entry
• func Every(duration time.Duration, job cron.Job, n string)
• func In(duration time.Duration, job cron.Job, n string)
• func Now(job cron.Job, n string)
• func Remove(id cron.EntryID)
• func Schedule(spec string, job cron.Job, n string) error
• func Start(loc *time.Location, isStoreExecutionStatus bool, v ...int)
• func StatusJson() map[string]interface{}
• func Stop()
• type Func
  • func (r Func) Run()
• type Job
  • func AddJob(job cron.Job) *Job
  • func New(job cron.Job, n string) *Job
  • func (j *Job) Run()
  • func (j *Job) StatusUpdate() string
• type JobStatus
  • func (js *JobStatus) String()
• type StatusData
  • func StatusPage() []StatusData

Constants

const DEFAULT_JOB_POOL_SIZE = 10

const UNNAMED = "(unnamed)"

Variables

var (
	// Singleton instance of the underlying job scheduler.
	MainCron *cron.Cron

	IsStoreExecutionStatus  bool
	JobsExecutionStatusChan chan *JobStatus
)

Functions

func Entries

func Entries() []cron.Entry

Return detailed list of currently running recurring jobs to remove an entry, first retrieve the ID of entry

func Every

func Every(duration time.Duration, job cron.Job, n string)

Run the given job at a fixed interval. The interval provided is the time between the job ending and the job being run again. The time that the job takes to run is not included in the interval.

func In

func In(duration time.Duration, job cron.Job, n string)

Run the given job once, after the given delay.

func Now

func Now(job cron.Job, n string)

Run the given job right now.

func Remove

func Remove(id cron.EntryID)

Remove a specific job from running Get EntryID from the list job entries jobrunner.Entries() If job is in the middle of running, once the process is finished it will be removed

func Schedule

func Schedule(spec string, job cron.Job, n string) error

func Start

func Start(loc *time.Location, isStoreExecutionStatus bool, v ...int)

func StatusJson

func StatusJson() map[string]interface{}

func Stop

func Stop()

Stop ALL active jobs from running at the next scheduled time

Types

type Func

type Func func()

Callers can use jobs.Func to wrap a raw func. (Copying the type to this package makes it more visible)

For example:

jobrunner.Schedule("cron.frequent", jobs.Func(myFunc), "myFunc")

func (Func) Run

func (r Func) Run()

type Job

type Job struct {
	Name string

	Status  string
	Latency string
	// contains filtered or unexported fields
}

func AddJob

func AddJob(job cron.Job) *Job

func New

func New(job cron.Job, n string) *Job

func (*Job) Run

func (j *Job) Run()

func (*Job) StatusUpdate

func (j *Job) StatusUpdate() string

type JobStatus

type JobStatus struct {
	Name      string
	Status    string
	StartTime time.Time
	EndTime   time.Time
}

func (*JobStatus) String

func (js *JobStatus) String()

type StatusData

type StatusData struct {
	Id        cron.EntryID
	JobRunner *Job
	Next      time.Time
	Prev      time.Time
}

func StatusPage

func StatusPage() []StatusData