ISO8601 Support

client.go

@@ -7,10 +7,11 @@ import (
 	"fmt"
 	"net/http"
 	"net/url"
-	"strconv"
 	"strings"
 	"time"
 
+	"github.com/senseyeio/duration"
+
 	"github.com/cenkalti/dalga/v2/internal/jobmanager"
 	"github.com/cenkalti/dalga/v2/internal/server"
 	"github.com/cenkalti/dalga/v2/internal/table"
@@ -84,8 +85,11 @@ func (clnt *Client) Schedule(ctx context.Context, path, body string, opts ...Sch
 	}
 
 	values := make(url.Values)
-	if so.Interval > 0 {
-		values.Set("interval", strconv.Itoa(int(so.Interval/time.Second)))
+	if !so.Interval.IsZero() {
+		values.Set("interval", so.Interval.String())
+	}
+	if so.Location != nil {
+		values.Set("location", so.Location.String())
 	}
 	if !so.FirstRun.IsZero() {
 		values.Set("first-run", so.FirstRun.Format(time.RFC3339))
@@ -194,12 +198,34 @@ type ScheduleOptions = jobmanager.ScheduleOptions
 
 type ScheduleOpt func(o *ScheduleOptions)
 
-func WithInterval(d time.Duration) ScheduleOpt {
+func WithInterval(d duration.Duration) ScheduleOpt {
 	return func(o *ScheduleOptions) {
 		o.Interval = d
 	}
 }
 
+func MustWithIntervalString(s string) ScheduleOpt {
+	d, err := duration.ParseISO8601(s)
+	if err != nil {
+		panic(err)
+	}
+	return WithInterval(d)
+}
+
+func WithLocation(l *time.Location) ScheduleOpt {
+	return func(o *ScheduleOptions) {
+		o.Location = l
+	}
+}
+
+func MustWithLocationName(n string) ScheduleOpt {
+	l, err := time.LoadLocation(n)
+	if err != nil {
+		panic(err)
+	}
+	return WithLocation(l)
+}
+
 func WithFirstRun(t time.Time) ScheduleOpt {
 	return func(o *ScheduleOptions) {
 		o.FirstRun = t

client_test.go

@@ -6,7 +6,6 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"testing"
-	"time"
 )
 
 func TestClient(t *testing.T) {
@@ -23,7 +22,8 @@ func TestClient(t *testing.T) {
 
 	config := DefaultConfig
 	config.Endpoint.BaseURL = "http://" + srv.Listener.Addr().String()
-	d, lis, cleanup := newDalga(t, DefaultConfig)
+	config.MySQL.SkipLocked = false
+	d, lis, cleanup := newDalga(t, config)
 	defer cleanup()
 
 	ctx, cancel := context.WithCancel(context.Background())
@@ -45,7 +45,7 @@ func TestClient(t *testing.T) {
 	})
 
 	t.Run("schedule", func(t *testing.T) {
-		if j, err := clnt.Schedule(callCtx, "when", "where", WithInterval(time.Minute)); err != nil {
+		if j, err := clnt.Schedule(callCtx, "when", "where", MustWithIntervalString("PT1M")); err != nil {
 			t.Fatal(err)
 		} else if j.Body != "where" {
 			t.Fatalf("unexpected body: %s", j.Body)

config.go

@@ -8,7 +8,8 @@ import (
 
 var DefaultConfig = Config{
 	Jobs: jobsConfig{
-		RetryInterval: 60,
+		RetryInterval: "PT1M",
+		ScanFrequency: 1000,
 	},
 	MySQL: mysqlConfig{
 		Host:         "127.0.0.1",
@@ -43,7 +44,9 @@ func (c *Config) LoadFromFile(filename string) error {
 
 type jobsConfig struct {
 	RandomizationFactor float64
-	RetryInterval       int
+	RetryInterval       string
+	FixedIntervals      bool
+	ScanFrequency       int
 }
 
 type mysqlConfig struct {

dalga.go

@@ -8,6 +8,9 @@ import (
 	"net"
 	"time"
 
+	"github.com/senseyeio/duration"
+
+	"github.com/cenkalti/dalga/v2/internal/clock"
 	"github.com/cenkalti/dalga/v2/internal/instance"
 	"github.com/cenkalti/dalga/v2/internal/jobmanager"
 	"github.com/cenkalti/dalga/v2/internal/scheduler"
@@ -49,10 +52,16 @@ func New(config Config) (*Dalga, error) {
 	}
 	log.Println("listening", lis.Addr())
 
+	interval, err := duration.ParseISO8601(config.Jobs.RetryInterval)
+	if err != nil {
+		return nil, err
+	}
+
 	t := table.New(db, config.MySQL.Table)
 	t.SkipLocked = config.MySQL.SkipLocked
+	t.FixedIntervals = config.Jobs.FixedIntervals
 	i := instance.New(t)
-	s := scheduler.New(t, i.ID(), config.Endpoint.BaseURL, time.Duration(config.Endpoint.Timeout)*time.Second, time.Duration(config.Jobs.RetryInterval)*time.Second, config.Jobs.RandomizationFactor)
+	s := scheduler.New(t, i.ID(), config.Endpoint.BaseURL, time.Duration(config.Endpoint.Timeout)*time.Second, interval, config.Jobs.RandomizationFactor, time.Millisecond*time.Duration(config.Jobs.ScanFrequency))
 	j := jobmanager.New(t, s)
 	srv := server.New(j, t, i.ID(), lis, 10*time.Second)
 	return &Dalga{
@@ -98,3 +107,8 @@ func (d *Dalga) Run(ctx context.Context) {
 func (d *Dalga) CreateTable() error {
 	return d.table.Create(context.Background())
 }
+
+func (d *Dalga) UseClock(now time.Time) *clock.Clock {
+	d.table.Clk = clock.New(now)
+	return d.table.Clk
+}

dalga_test.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"database/sql"
 	"net/http"
+	"net/http/httptest"
 	"net/url"
 	"strings"
 	"testing"
@@ -20,16 +21,9 @@ func init() {
 const (
 	testBody    = "testBody"
 	testTimeout = 5 * time.Second
-	testAddr    = "127.0.0.1:5000"
 )
 
 func TestSchedule(t *testing.T) {
-	config := DefaultConfig
-	config.MySQL.SkipLocked = false
-
-	d, lis, cleanup := newDalga(t, config)
-	defer cleanup()
-
 	called := make(chan string)
 	endpoint := func(w http.ResponseWriter, r *http.Request) {
 		var buf bytes.Buffer
@@ -38,8 +32,17 @@ func TestSchedule(t *testing.T) {
 		called <- buf.String()
 	}
 
-	http.HandleFunc("/", endpoint)
-	go http.ListenAndServe(testAddr, nil)
+	mux := http.NewServeMux()
+	mux.HandleFunc("/", endpoint)
+	srv := httptest.NewServer(mux)
+	defer srv.Close()
+
+	config := DefaultConfig
+	config.MySQL.SkipLocked = false
+	config.Endpoint.BaseURL = "http://" + srv.Listener.Addr().String() + "/"
+
+	d, lis, cleanup := newDalga(t, config)
+	defer cleanup()
 
 	ctx, cancel := context.WithCancel(context.Background())
 	go d.Run(ctx)

go.mod

@@ -5,6 +5,7 @@ go 1.13
 require (
 	github.com/bmizerany/pat v0.0.0-20170815010413-6226ea591a40
 	github.com/go-sql-driver/mysql v1.5.0
+	github.com/senseyeio/duration v0.0.0-20180430131211-7c2a214ada46
 	gopkg.in/gcfg.v1 v1.2.3
 	gopkg.in/warnings.v0 v0.1.2 // indirect
 )

go.sum

@@ -2,6 +2,8 @@ github.com/bmizerany/pat v0.0.0-20170815010413-6226ea591a40 h1:y4B3+GPxKlrigF1ha
 github.com/bmizerany/pat v0.0.0-20170815010413-6226ea591a40/go.mod h1:8rLXio+WjiTceGBHIoTvn60HIbs7Hm7bcHjyrSqYB9c=
 github.com/go-sql-driver/mysql v1.5.0 h1:ozyZYNQW3x3HtqT1jira07DN2PArx2v7/mN66gGcHOs=
 github.com/go-sql-driver/mysql v1.5.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg=
+github.com/senseyeio/duration v0.0.0-20180430131211-7c2a214ada46 h1:Dz0HrI1AtNSGCE8LXLLqoZU4iuOJXPWndenCsZfstA8=
+github.com/senseyeio/duration v0.0.0-20180430131211-7c2a214ada46/go.mod h1:is8FVkzSi7PYLWEXT5MgWhglFsyyiW8ffxAoJqfuFZo=
 gopkg.in/gcfg.v1 v1.2.3 h1:m8OOJ4ccYHnx2f4gQwpno8nAX5OGOh7RLaaz0pj3Ogs=
 gopkg.in/gcfg.v1 v1.2.3/go.mod h1:yesOnuUOFQAhST5vPY4nbZsb/huCgGGXlipJsBn0b3o=
 gopkg.in/warnings.v0 v0.1.2 h1:wFXVbFY8DY5/xOe1ECiWdKCzZlxgshcYVNkBHstARME=

internal/instance/instance.go

@@ -56,6 +56,6 @@ func (i *Instance) Run(ctx context.Context) {
 func (i *Instance) updateInstance(ctx context.Context) {
 	err := i.table.UpdateInstance(ctx, i.id)
 	if err != nil {
-		log.Print("cannot update instance at db:", err)
+		log.Print("cannot update instance at db: ", err)
 	}
 }

internal/jobmanager/jobmanager.go

@@ -5,6 +5,8 @@ import (
 	"errors"
 	"time"
 
+	"github.com/senseyeio/duration"
+
 	"github.com/cenkalti/dalga/v2/internal/log"
 	"github.com/cenkalti/dalga/v2/internal/scheduler"
 	"github.com/cenkalti/dalga/v2/internal/table"
@@ -19,7 +21,8 @@ type ScheduleOptions struct {
 	OneOff    bool
 	Immediate bool
 	FirstRun  time.Time
-	Interval  time.Duration
+	Location  *time.Location
+	Interval  duration.Duration
 }
 
 var ErrInvalidArgs = errors.New("invalid arguments")
@@ -43,32 +46,32 @@ func (m *JobManager) Schedule(ctx context.Context, path, body string, opt Schedu
 		Body: body,
 	}
 
-	var interval time.Duration
-	var delay time.Duration
+	var interval duration.Duration
+	var delay duration.Duration
 	var nextRun time.Time
 	switch {
 	case opt.OneOff && opt.Immediate: // one-off and immediate
 		// both first-run and interval params must be zero
-		if !opt.FirstRun.IsZero() || opt.Interval != 0 {
+		if !opt.FirstRun.IsZero() || !opt.Interval.IsZero() {
 			return nil, ErrInvalidArgs
 		}
 	case opt.OneOff && !opt.Immediate: // one-off but later
 		// only one of from first-run and interval params must be set
-		if (!opt.FirstRun.IsZero() && opt.Interval != 0) || (opt.FirstRun.IsZero() && opt.Interval == 0) {
+		if (!opt.FirstRun.IsZero() && !opt.Interval.IsZero()) || (opt.FirstRun.IsZero() && opt.Interval.IsZero()) {
 			return nil, ErrInvalidArgs
 		}
-		if opt.Interval != 0 {
+		if !opt.Interval.IsZero() {
 			delay = opt.Interval
 		} else {
 			nextRun = opt.FirstRun
 		}
 	case !opt.OneOff && opt.Immediate: // periodic and immediate
-		if opt.Interval == 0 || !opt.FirstRun.IsZero() {
+		if opt.Interval.IsZero() || !opt.FirstRun.IsZero() {
 			return nil, ErrInvalidArgs
 		}
 		interval = opt.Interval
 	default: // periodic
-		if opt.Interval == 0 {
+		if opt.Interval.IsZero() {
 			return nil, ErrInvalidArgs
 		}
 		interval = opt.Interval
@@ -79,7 +82,7 @@ func (m *JobManager) Schedule(ctx context.Context, path, body string, opt Schedu
 		}
 	}
 	log.Debugln("job is scheduled:", key.Path, key.Body)
-	return m.table.AddJob(ctx, key, interval, delay, nextRun)
+	return m.table.AddJob(ctx, key, interval, delay, opt.Location, nextRun)
 }
 
 // Cancel deletes the job with path and body.

internal/scheduler/scheduler.go

@@ -4,15 +4,16 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
-	"math/rand"
 	"net/http"
 	"strings"
 	"sync/atomic"
 	"time"
 
+	"github.com/go-sql-driver/mysql"
+	"github.com/senseyeio/duration"
+
 	"github.com/cenkalti/dalga/v2/internal/log"
 	"github.com/cenkalti/dalga/v2/internal/table"
-	"github.com/go-sql-driver/mysql"
 )
 
 type Scheduler struct {
@@ -21,18 +22,20 @@ type Scheduler struct {
 	client              http.Client
 	baseURL             string
 	randomizationFactor float64
-	retryInterval       time.Duration
+	retryInterval       duration.Duration
 	runningJobs         int32
+	scanFrequency       time.Duration
 	done                chan struct{}
 }
 
-func New(t *table.Table, instanceID uint32, baseURL string, clientTimeout, retryInterval time.Duration, randomizationFactor float64) *Scheduler {
+func New(t *table.Table, instanceID uint32, baseURL string, clientTimeout time.Duration, retryInterval duration.Duration, randomizationFactor float64, scanFrequency time.Duration) *Scheduler {
 	return &Scheduler{
 		table:               t,
 		instanceID:          instanceID,
 		baseURL:             baseURL,
 		randomizationFactor: randomizationFactor,
 		retryInterval:       retryInterval,
+		scanFrequency:       scanFrequency,
 		done:                make(chan struct{}),
 		client: http.Client{
 			Timeout: clientTimeout,
@@ -62,7 +65,7 @@ func (s *Scheduler) Run(ctx context.Context) {
 		if err == sql.ErrNoRows {
 			log.Debugln("no scheduled jobs in the table")
 			select {
-			case <-time.After(time.Second):
+			case <-time.After(s.scanFrequency):
 			case <-ctx.Done():
 				return
 			}
@@ -75,7 +78,7 @@ func (s *Scheduler) Run(ctx context.Context) {
 		if err != nil {
 			log.Println("error while getting next job:", err)
 			select {
-			case <-time.After(time.Second):
+			case <-time.After(s.scanFrequency):
 			case <-ctx.Done():
 				return
 			}
@@ -92,18 +95,13 @@ func (s *Scheduler) Run(ctx context.Context) {
 	}
 }
 
-func randomize(d time.Duration, f float64) time.Duration {
-	delta := time.Duration(f * float64(d))
-	return d - delta + time.Duration(float64(2*delta)*rand.Float64())
-}
-
 // execute makes a POST request to the endpoint and updates the Job's next run time.
 func (s *Scheduler) execute(ctx context.Context, j *table.Job) error {
 	log.Debugln("executing:", j.String())
 	code, err := s.postJob(ctx, j)
 	if err != nil {
 		log.Printf("error while doing http post for %s: %s", j.String(), err)
-		return s.table.UpdateNextRun(ctx, j.Key, s.retryInterval)
+		return s.table.UpdateNextRun(ctx, j.Key, s.retryInterval, 0)
 	}
 	if j.OneOff() {
 		log.Debugln("deleting one-off job")
@@ -113,17 +111,12 @@ func (s *Scheduler) execute(ctx context.Context, j *table.Job) error {
 		log.Debugln("deleting not found job")
 		return s.table.DeleteJob(ctx, j.Key)
 	}
-	add := j.Interval
-	if s.randomizationFactor > 0 {
-		// Add some randomization to periodic tasks.
-		add = randomize(add, s.randomizationFactor)
-	}
-	return s.table.UpdateNextRun(ctx, j.Key, add)
+	return s.table.UpdateNextRun(ctx, j.Key, j.Interval, s.randomizationFactor)
 }
 
 func (s *Scheduler) postJob(ctx context.Context, j *table.Job) (code int, err error) {
 	url := s.baseURL + j.Path
-	log.Debugln("doing http post to ", url)
+	log.Debugln("doing http post to", url)
 	req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, strings.NewReader(j.Body))
 	if err != nil {
 		return