Repository: adhocore/gronx
Pull request #3 (2021-05-02 08:36:29)

Add task daemon


Patch:
diff --git a/README.md b/README.md
index 018f0b3..76b6972 100644
--- a/README.md
+++ b/README.md
@@ -42,13 +42,105 @@ gron.IsDue(expr, time.Date(2021, time.April, 1, 1, 1, 0, 0, time.UTC)) // true|f
 ```
 
 In a more practical level, you would use this tool to manage and invoke jobs in app itself and not
-mess around with `crontab` for each and every new tasks/jobs. It doesn't yet replace that but rather supplements it.
-There is a plan though [#1](https://github.com/adhocore/gronx/issues/1).
+mess around with `crontab` for each and every new tasks/jobs. ~~It doesn't yet replace that but rather supplements it.
+There is a plan though [#1](https://github.com/adhocore/gronx/issues/1)~~.
 
 In crontab just put one entry with `* * * * *` which points to your Go entry point that uses this tool.
 Then in that entry point you would invoke different tasks if the corresponding Cron expr is due.
 Simple map structure would work for this.
 
+Check the section below for more sophisticated way of managing tasks automatically using `gronx` daemon called `tasker`.
+
+---
+### Go Tasker
+
+Tasker is a task manager that can be programatically used in Golang applications. It runs as a daemon and and invokes tasks scheduled with cron expression:
+```go
+package main
+
+import (
+	"context"
+	"time"
+
+	"github.com/adhocore/gronx/pkg/tasker"
+)
+
+func main() {
+	taskr := tasker.New(tasker.Option{
+		Verbose: true,
+        // optional: defaults to local
+		Tz:      "Asia/Bangkok",
+        // optional: defaults to Golang's default log stream
+		Out:     "/full/path/to/output-file",
+	})
+
+	taskr.Task("* * * * *", func(ctx context.Context) (int, error) { // add task to run every minute
+		// do something ...
+
+		// then return exit code and error, for eg: if everything okay
+		return 0, nil
+	}).Task("*/5 * * * *", func(ctx context.Context) (int, error) { // add task to run every 5 minutes
+		// you can also log the output to Out file as configured in Option above:
+		taskr.Log.Printf("done something in %d s", 2)
+
+		return 0, nil
+	}).Task("@10minutes", tasker.Taskify("command --option val -- args")) // every 10 minute with arbitrary command
+
+	// ... add more tasks
+
+	// optionally if you want tasker to stop after an hour, just pass the duration with Until():
+	taskr.Until(60 * time.Minute)
+
+	// finally run the tasker, it ticks sharply on every minute and runs all the tasks due on that time!
+	// it exits gracefully when ctrl+c is received making sure pending tasks are completed.
+	taskr.Run()
+}
+```
+
+### Task Daemon
+It can also be used as standalone task daemon instead of programmatic usage for Golang application.
+
+First, just install tasker command:
+```sh
+go get -u github.com/adhocore/gronx/cmd/tasker
+```
+
+Then prepare a taskfile ([example](./tests/../test/taskfile.txt)) in crontab format
+(or can even point to existing crontab).
+
+Finally run the task daemon like so
+```
+tasker -file path/to/taskfile
+```
+> You can pass more options to control the behavior of task daemon, see below.
+
+####  Tasker command options:
+```txt
+-file string <required>
+    The task file in crontab format
+-out string
+    The fullpath to file where output from tasks are sent to
+-shell string
+    The shell to use for running tasks (default "/usr/bin/bash")
+-tz string
+    The timezone to use for tasks (default "Local")
+-until int
+    The timeout for task daemon in minutes
+-verbose
+    The verbose mode outputs as much as possible
+```
+
+Examples:
+```sh
+tasker -verbose -file path/to/taskfile -until 120 # run until next 120min (i.e 2hour) with all feedbacks echoed back
+tasker -verbose -file path/to/taskfile -out path/to/output # with all feedbacks echoed to the output file
+tasker -tz America/New_York -file path/to/taskfile -shell zsh # run all tasks using zsh shell based on NY timezone
+```
+
+> File extension of taskfile for (`-file` option) does not matter: can be any or none.
+> The directory for outfile (`-out` option) must exist, file is created by task daemon.
+
+> Same timezone applies for all tasks currently and it might support overriding timezone per task in future release.
 ---
 ### Cron Expression
 
diff --git a/cmd/tasker/main.go b/cmd/tasker/main.go
new file mode 100644
index 0000000..0e0d64a
--- /dev/null
+++ b/cmd/tasker/main.go
@@ -0,0 +1,48 @@
+package main
+
+import (
+	"flag"
+	"log"
+	"os"
+	"time"
+
+	"github.com/adhocore/gronx/pkg/tasker"
+)
+
+func main() {
+	opt := mustGetOption()
+	taskr := tasker.New(opt)
+
+	for _, task := range tasker.MustParseTaskfile(opt) {
+		taskr.Task(task.Expr, tasker.Taskify(task.Cmd, opt))
+	}
+
+	if opt.Until > 0 {
+		taskr.Until(time.Duration(opt.Until) * time.Minute)
+	}
+
+	taskr.Run()
+}
+
+func mustGetOption() tasker.Option {
+	var opt tasker.Option
+
+	flag.StringVar(&opt.File, "file", "", "The task file in crontab format")
+	flag.StringVar(&opt.Tz, "tz", "Local", "The timezone to use for tasks")
+	flag.StringVar(&opt.Shell, "shell", tasker.Shell()[0], "The shell to use for running tasks")
+	flag.StringVar(&opt.Out, "out", "", "The fullpath to file where output from tasks are sent to")
+	flag.BoolVar(&opt.Verbose, "verbose", false, "The verbose mode outputs as much as possible")
+	flag.Int64Var(&opt.Until, "until", 0, "The timeout for task daemon in minutes")
+	flag.Parse()
+
+	if opt.File == "" {
+		flag.Usage()
+		os.Exit(1)
+	}
+
+	if _, err := os.Stat(opt.File); err != nil {
+		log.Fatalf("can't read taskfile: %s", opt.File)
+	}
+
+	return opt
+}
diff --git a/gronx.go b/gronx.go
index b62b1c2..4b57da2 100644
--- a/gronx.go
+++ b/gronx.go
@@ -27,7 +27,8 @@ var expressions = map[string]string{
 	"@30minutes": "0,30 * * * *",
 }
 
-var spaceRe = regexp.MustCompile(`\s+`)
+// SpaceRe is regex for whitespace.
+var SpaceRe = regexp.MustCompile(`\s+`)
 
 func normalize(expr string) []string {
 	expr = strings.Trim(expr, " \t")
@@ -35,7 +36,7 @@ func normalize(expr string) []string {
 		expr = e
 	}
 
-	expr = spaceRe.ReplaceAllString(expr, " ")
+	expr = SpaceRe.ReplaceAllString(expr, " ")
 	expr = literals.Replace(strings.ToUpper(expr))
 
 	return strings.Split(strings.ReplaceAll(expr, "  ", " "), " ")
@@ -54,18 +55,35 @@ func New() Gronx {
 // IsDue checks if cron expression is due for given reference time (or now).
 // It returns bool or error if any.
 func (g *Gronx) IsDue(expr string, ref ...time.Time) (bool, error) {
-	segs := normalize(expr)
-	if len(segs) < 5 || len(segs) > 6 {
-		return false, errors.New("expr should contain 5 to 6 segments separated by space")
-	}
-
 	if len(ref) > 0 {
 		g.C.SetRef(ref[0])
 	} else {
 		g.C.SetRef(time.Now())
 	}
 
-	for pos, seg := range segs {
+	segs, err := Segments(expr)
+	if err != nil {
+		return false, err
+	}
+
+	return g.SegmentsDue(segs)
+}
+
+// Segments splits expr into array array of cron parts.
+// It returns array or error.
+func Segments(expr string) ([]string, error) {
+	segs := normalize(expr)
+	if len(segs) < 5 || len(segs) > 6 {
+		return []string{}, errors.New("expr should contain 5-6 segments separated by space")
+	}
+
+	return segs, nil
+}
+
+// SegmentsDue checks if all cron parts are due.
+// It returns bool. You should use IsDue(expr) instead.
+func (g *Gronx) SegmentsDue(segments []string) (bool, error) {
+	for pos, seg := range segments {
 		if seg == "*" || seg == "?" {
 			continue
 		}
diff --git a/pkg/tasker/parser.go b/pkg/tasker/parser.go
new file mode 100644
index 0000000..7a7df31
--- /dev/null
+++ b/pkg/tasker/parser.go
@@ -0,0 +1,65 @@
+package tasker
+
+import (
+	"bufio"
+	"log"
+	"os"
+	"regexp"
+	"strings"
+
+	"github.com/adhocore/gronx"
+)
+
+// MustParseTaskfile either parses taskfile from given Option.
+// It fails hard in case any error.
+func MustParseTaskfile(opts Option) []Task {
+	file, err := os.Open(opts.File)
+	if err != nil {
+		log.Fatalf("[parser] can't open file: %s", opts.File)
+	}
+	defer file.Close()
+
+	lines := []string{}
+	scan := bufio.NewScanner(file)
+	for scan.Scan() {
+		ln := strings.TrimLeft(scan.Text(), " \t")
+		if ln != "" && ln[0] != '#' {
+			lines = append(lines, ln)
+		}
+	}
+
+	if err := scan.Err(); err != nil {
+		if len(lines) == 0 {
+			log.Fatalf("[parser] error reading taskfile: %v", err)
+		}
+
+		log.Println(err)
+	}
+
+	return linesToTasks(lines)
+}
+
+var cronRe = regexp.MustCompile(`^((?:[^\s]+\s+){5}(?:\d{4})?)(?:\s+)?(.*)`)
+var aliasRe = regexp.MustCompile(`^(@(?:annually|yearly|monthly|weekly|daily|hourly|5minutes|10minutes|15minutes|30minutes|always))(?:\s+)?(.*)`)
+
+func linesToTasks(lines []string) []Task {
+	var tasks []Task
+
+	gron := gronx.New()
+	for _, line := range lines {
+		var match []string
+		if line[0] == '@' {
+			match = aliasRe.FindStringSubmatch(line)
+		} else {
+			match = cronRe.FindStringSubmatch(line)
+		}
+		if len(match) > 2 && gron.IsValid(match[1]) {
+			tasks = append(tasks, Task{strings.Trim(match[1], " \t"), match[2]})
+			continue
+		}
+
+		log.Printf("[parser] can't parse cron expr: %s", line)
+	}
+
+	return tasks
+}
diff --git a/pkg/tasker/tasker.go b/pkg/tasker/tasker.go
new file mode 100644
index 0000000..97271ce
--- /dev/null
+++ b/pkg/tasker/tasker.go
@@ -0,0 +1,323 @@
+package tasker
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"os"
+	"os/exec"
+	"os/signal"
+	"path/filepath"
+	"reflect"
+	"strings"
+	"sync"
+	"syscall"
+	"time"
+
+	"github.com/adhocore/gronx"
+)
+
+// Option is the config options for Tasker.
+type Option struct {
+	File    string
+	Tz      string
+	Shell   string
+	Out     string
+	Until   int64
+	Verbose bool
+}
+
+// TaskFunc is the actual task handler.
+type TaskFunc func(ctx context.Context) (int, error)
+
+// Task wraps a cron expr and its' command.
+type Task struct {
+	Expr string
+	Cmd  string
+}
+
+// Tasker is the task manager.
+type Tasker struct {
+	Log     *log.Logger
+	loc     *time.Location
+	gron    *gronx.Gronx
+	wg      sync.WaitGroup
+	until   time.Time
+	exprs   map[string][]string
+	tasks   map[string]TaskFunc
+	abort   bool
+	timeout bool
+	verbose bool
+}
+
+type result struct {
+	ref  string
+	code int
+	err  error
+}
+
+// New inits a task manager.
+// It returns Tasker.
+func New(opt Option) *Tasker {
+	gron := gronx.New()
+	tasks := make(map[string]TaskFunc)
+	exprs := make(map[string][]string)
+
+	if opt.Tz == "" {
+		opt.Tz = "Local"
+	}
+
+	loc, err := time.LoadLocation(opt.Tz)
+	if err != nil {
+		log.Fatalf("invalid tz location: %s", opt.Tz)
+	}
+
+	logger := log.New(os.Stderr, "", log.LstdFlags)
+	if opt.Out != "" {
+		if _, err := os.Stat(filepath.Dir(opt.Out)); err != nil {
+			log.Fatalf("output dir does not exist: %s", filepath.Base(opt.Out))
+		}
+
+		file, err := os.OpenFile(opt.Out, os.O_CREATE|os.O_WRONLY, 0777)
+		if err != nil {
+			log.Fatalf("can't open output file: %s", opt.Out)
+		}
+
+		logger = log.New(file, "", log.LstdFlags)
+	}
+
+	return &Tasker{Log: logger, loc: loc, gron: &gron, exprs: exprs, tasks: tasks, verbose: opt.Verbose}
+}
+
+// Taskify creates TaskFunc out of plain command wrt given options.
+func Taskify(cmd string, opt Option) TaskFunc {
+	sh := Shell(opt.Shell)
+
+	return func(ctx context.Context) (int, error) {
+		err := exec.Command(sh[0], sh[1], cmd).Run()
+		if err == nil {
+			return 0, nil
+		}
+
+		code := 1
+		if exErr, ok := err.(*exec.ExitError); ok {
+			code = exErr.ExitCode()
+		}
+
+		return code, err
+	}
+}
+
+// Shell gives a pair of shell and arg.
+// It returns array of string.
+func Shell(shell ...string) []string {
+	if os.PathSeparator == '\\' {
+		shell = append(shell, "git-bash.exe -c", "bash.exe -c", "powershell.exe -Command")
+	} else {
+		shell = append(shell, "bash -c", "sh -c", "zsh -c")
+	}
+
+	for _, sh := range shell {
+		arg := "-c"
+		cmd := strings.Split(sh, " -")
+		if len(cmd) > 1 {
+			arg = "-" + cmd[1]
+		}
+		if exc, err := exec.LookPath(cmd[0]); err == nil {
+			return []string{exc, arg}
+		}
+	}
+
+	return []string{"/bin/sh", "-c"}
+}
+
+const taskIDFormat = "[%s][#%d]"
+
+// Task appends new task handler for given cron expr.
+// It returns Tasker (itself) for fluency and bails if expr is invalid.
+func (t *Tasker) Task(expr string, task TaskFunc) *Tasker {
+	segs, err := gronx.Segments(expr)
+	if err != nil {
+		log.Fatalf("invalid cron expr: %+v", err)
+	}
+
+	old, expr := gronx.SpaceRe.ReplaceAllString(expr, " "), strings.Join(segs, " ")
+	if _, ok := t.exprs[expr]; !ok {
+		// Validate expr.
+		if _, err := t.gron.SegmentsDue(segs); err != nil {
+			log.Fatalf("invalid cron expr: %+v", err)
+		}
+
+		t.exprs[expr] = []string{}
+	}
+
+	ref := fmt.Sprintf(taskIDFormat, old, len(t.exprs[expr])+1)
+
+	t.exprs[expr] = append(t.exprs[expr], ref)
+	t.tasks[ref] = task
+
+	return t
+}
+
+// Until sets the cutoff time until which the tasker runs.
+// It returns itself for fluency.
+func (t *Tasker) Until(until interface{}) *Tasker {
+	switch until := until.(type) {
+	case time.Duration:
+		t.until = t.now().Add(until)
+	case time.Time:
+		t.until = until
+	default:
+		log.Fatalf("until must be time.Duration or time.Time, got: %v", reflect.TypeOf(until))
+	}
+
+	return t
+}
+
+func (t *Tasker) now() time.Time {
+	return time.Now().In(t.loc)
+}
+
+// Run runs the task manager.
+func (t *Tasker) Run() {
+	t.doSetup()
+
+	first := true
+	for !t.abort && !t.timeout {
+		ref, willTime := t.tickTimer(first)
+
+		t.gron.C.SetRef(ref)
+		if t.timeout {
+			break
+		}
+
+		tasks := make(map[string]TaskFunc)
+		for expr, refs := range t.exprs {
+			if due, _ := t.gron.SegmentsDue(strings.Split(expr, " ")); !due {
+				continue
+			}
+
+			for _, ref := range refs {
+				tasks[ref] = t.tasks[ref]
+			}
+		}
+
+		if len(tasks) > 0 {
+			t.runTasks(tasks)
+		}
+
+		first = false
+		t.timeout = willTime
+	}
+
+	t.wait()
+}
+
+func (t *Tasker) doSetup() {
+	if len(t.tasks) == 0 {
+		t.Log.Fatal("[tasker] no tasks available")
+	}
+	if !t.until.IsZero() && t.verbose {
+		if t.until.Before(t.now()) {
+			log.Fatalf("[tasker] timeout must be in future")
+		}
+		t.Log.Printf("[tasker] final tick on %s", t.until.Format("2006/01/02 15:04:00"))
+	}
+
+	sig := make(chan os.Signal)
+	signal.Notify(sig, os.Interrupt, syscall.SIGTERM)
+
+	go func() {
+		<-sig
+		t.abort = true
+		t.wait()
+	}()
+}
+
+var tickSec = 60
+
+func (t *Tasker) tickTimer(first bool) (time.Time, bool) {
+	now, timed, willTime := t.now(), !t.until.IsZero(), false
+	if t.timeout || t.abort {
+		return now, willTime
+	}
+
+	wait := tickSec - now.Second()%tickSec
+	if !first && wait == 0 {
+		wait = tickSec
+	}
+
+	if wait < 1 && wait > tickSec {
+		return now, willTime
+	}
+
+	dur := time.Duration(wait) * time.Second
+	next := now.Add(dur)
+	willTime = timed && next.After(t.until)
+	if t.verbose && !willTime {
+		t.Log.Printf("[tasker] next tick on %s", next.Format("2006/01/02 15:04:05"))
+	}
+
+	time.Sleep(dur)
+	t.timeout = timed && next.After(t.until)
+
+	return next, willTime
+}
+
+func (t *Tasker) runTasks(tasks map[string]TaskFunc) {
+	if t.verbose {
+		if t.abort {
+			t.Log.Println("[tasker] completing pending tasks")
+		} else {
+			t.Log.Printf("[tasker] running %d due tasks\n", len(tasks))
+		}
+	}
+
+	ctx := context.TODO()
+	for ref, task := range tasks {
+		t.wg.Add(1)
+		rc := make(chan result)
+
+		go t.doRun(ctx, ref, task, rc)
+		go t.doOut(rc)
+	}
+}
+
+func (t *Tasker) doRun(ctx context.Context, ref string, task TaskFunc, rc chan result) {
+	defer t.wg.Done()
+	if t.abort {
+		return
+	}
+
+	if t.verbose {
+		t.Log.Printf("[tasker] task %s running\n", ref)
+	}
+	code, err := task(ctx)
+
+	rc <- result{ref, code, err}
+}
+
+func (t *Tasker) doOut(rc chan result) {
+	res := <-rc
+	if res.err != nil {
+		t.Log.Printf("[tasker] task %s errored %v", res.ref, res.err)
+	}
+
+	if t.verbose {
+		if res.code == 0 {
+			t.Log.Printf("[tasker] task %s ran successfully", res.ref)
+		} else {
+			t.Log.Printf("[tasker] task %s returned error code: %d", res.ref, res.code)
+		}
+	}
+}
+
+func (t *Tasker) wait() {
+	if !t.abort {
+		t.Log.Println("[tasker] timed out, waiting tasks to complete")
+	} else {
+		t.Log.Println("[tasker] interrupted, waiting tasks to complete")
+	}
+
+	t.wg.Wait()
+}
diff --git a/validator.go b/validator.go
index fa7b992..3437f42 100644
--- a/validator.go
+++ b/validator.go
@@ -71,10 +71,6 @@ func isValidMonthDay(val string, last int, ref time.Time) (bool, error) {
 		return day == last, nil
 	}
 
-	if ref.Weekday() == 0 || ref.Weekday() == 6 {
-		return false, nil
-	}
-
 	pos := strings.Index(val, "W")
 	if pos < 1 {
 		return false, errors.New("invalid offset value: " + val)


Test patch:
diff --git a/.github/workflows/test-action.yml b/.github/workflows/test-action.yml
index 72b55bc..1ab4b6c 100644
--- a/.github/workflows/test-action.yml
+++ b/.github/workflows/test-action.yml
@@ -15,4 +15,4 @@ jobs:
       with:
         go-version: ${{ matrix.go-version }}
     - name: Test
-      run: go test -cover .
+      run: go test -cover ./...
diff --git a/cmd/tasker/main_test.go b/cmd/tasker/main_test.go
new file mode 100644
index 0000000..7209d62
--- /dev/null
+++ b/cmd/tasker/main_test.go
@@ -0,0 +1,24 @@
+package main
+
+import (
+	"os"
+	"testing"
+
+	"github.com/adhocore/gronx/pkg/tasker"
+)
+
+func TestMustGetOption(t *testing.T) {
+	old := os.Args
+	t.Run("Main", func(t *testing.T) {
+		expect := tasker.Option{File: "../../test/taskfile.txt", Out: "../../test/out.txt"}
+		os.Args = append(os.Args, "-verbose", "-file", expect.File, "-out", expect.Out)
+		opt := mustGetOption()
+		os.Args = old
+		if opt.File != expect.File {
+			t.Errorf("file: expected %v, got %v", opt.File, expect.File)
+		}
+		if opt.Out != expect.Out {
+			t.Errorf("out: expected %v, got %v", opt.Out, expect.Out)
+		}
+	})
+}
diff --git a/pkg/tasker/parser_test.go b/pkg/tasker/parser_test.go
new file mode 100644
index 0000000..972ca14
--- /dev/null
+++ b/pkg/tasker/parser_test.go
@@ -0,0 +1,22 @@
+package tasker
+
+import (
+	"testing"
+)
+
+func TestMustParseTaskfile(t *testing.T) {
+	t.Run("MustParseTaskfile", func(t *testing.T) {
+		tasks := MustParseTaskfile(Option{File: "../../test/taskfile.txt"})
+		if len(tasks) != 5 {
+			t.Errorf("should have 5 tasks, got %d", len(tasks))
+		}
+
+		if tasks[0].Expr != "*/1 0/1 * * *" {
+			t.Errorf("expected '*/1 0/1 * * *', got %s", tasks[0].Expr)
+		}
+
+		if tasks[2].Cmd != "echo '[task 3] @always' > test/task3.out" {
+			t.Errorf("expected `echo '[task 3] @always' > test/task3.out`, got %s", tasks[2].Cmd)
+		}
+	})
+}
diff --git a/pkg/tasker/tasker_test.go b/pkg/tasker/tasker_test.go
new file mode 100644
index 0000000..99071f5
--- /dev/null
+++ b/pkg/tasker/tasker_test.go
@@ -0,0 +1,44 @@
+package tasker
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+func TestRun(t *testing.T) {
+	t.Run("Run", func(t *testing.T) {
+		tickSec = 2
+		taskr := New(Option{Verbose: true, Out: "../../test/tasker.out"})
+
+		called := 0
+		taskr.Task("@always", func(ctx context.Context) (int, error) {
+			taskr.Log.Println("task [@always] sleeping 3s")
+			time.Sleep(3 * time.Second)
+			taskr.Log.Println("task [@always] finished")
+			called++
+
+			return 0, nil
+		})
+
+		taskr.Until(5 * time.Second).Run()
+
+		if called != 2 {
+			t.Errorf("task should be run 2 times, ran %d times", called)
+		}
+	})
+}
+
+func TestTaskify(t *testing.T) {
+	t.Run("Taskify", func(t *testing.T) {
+		ctx := context.TODO()
+		code, err := Taskify("echo -n 'taskify' > ../../test/taskify.out; echo 'test' >> ../../test/taskify.out", Option{})(ctx)
+
+		if code != 0 {
+			t.Errorf("expected code 0, got %d", code)
+		}
+		if err != nil {
+			t.Errorf("expected no error, got %v", err)
+		}
+	})
+}
diff --git a/test/taskfile.txt b/test/taskfile.txt
new file mode 100644
index 0000000..75d8787
--- /dev/null
+++ b/test/taskfile.txt
@@ -0,0 +1,12 @@
+# this taskfile contains 5 tasks, lines starting with # are comments
+*/1 0/1 * * * echo '[task 1] */1 0/1 * * *' > test/task1.out
+* * * * * 2021 echo '[task 2] * * * * * 2021' > test/task2.out
+
+# below three are equivalent
+@always echo '[task 3] @always' > test/task3.out
+@always echo '[task 4] @always' > test/task4.out
+*   *  *  * * echo '[task 5] *   *  *  * *' > test/task5.out
+
+# below are invalid
+@invalid
+* * * * *