diff --git a/pkg/scheduler/helpers.go b/pkg/scheduler/helpers.go index d76da26c1..2b2b07d3e 100644 --- a/pkg/scheduler/helpers.go +++ b/pkg/scheduler/helpers.go @@ -8,11 +8,19 @@ import ( ) // OnInterval parses all given strings as crontab expressions (ii) and returns true if any of them matches current time -func OnInterval(ee ...string) bool { +func OnInterval(ii ...string) bool { + match, err := onInterval(now(), ii...) + if err != nil { + sentry.CaptureException(err) + } + return match +} + +func onInterval(now time.Time, ii ...string) (bool, error) { var ( // This function will likely to be called exactly on a minute (00 sec) or a few milliseconds after // Round it up to the smallest unit that cronexpr package supports - currTime = now().Truncate(time.Second) + currTime = now.Truncate(time.Second) // For cron expression reference we need to subtract 1ns // this will cause Next() fn to include next nanosecond (if it matches) @@ -20,41 +28,42 @@ func OnInterval(ee ...string) bool { ) // At least one of the given expressions should match - for _, e := range ee { - exp, err := cronexpr.Parse(e) + for _, i := range ii { + exp, err := cronexpr.Parse(i) if err != nil { - sentry.CaptureException(err) - return false + return false, err } - if currTime.Equal(exp.Next(cronRef)) { - return true - } + return currTime.Equal(exp.Next(cronRef)), nil } - return false + return false, nil } // OnTimestamp parses all given strings as RFC3339 timestamps and returns true if any of them matches current time func OnTimestamp(tt ...string) bool { + match, err := onTimestamp(now(), tt...) + if err != nil { + sentry.CaptureException(err) + } + return match +} + +func onTimestamp(now time.Time, tt ...string) (bool, error) { var ( // This function will likely to be called exactly on a minute (00 sec) or a few milliseconds after // Round it up to the smallest unit that cronexpr package supports - currTime = now().Truncate(time.Second) + currTime = now.Truncate(time.Second) ) for _, t := range tt { ts, err := time.Parse(time.RFC3339, t) - if err != nil { - sentry.CaptureException(err) - return false + return false, err } - if currTime.Equal(ts.Round(time.Second)) { - return true - } + return currTime.Equal(ts.Round(time.Second)), nil } - return false + return false, nil } diff --git a/pkg/scheduler/helpers_test.go b/pkg/scheduler/helpers_test.go new file mode 100644 index 000000000..4f22f45f3 --- /dev/null +++ b/pkg/scheduler/helpers_test.go @@ -0,0 +1,141 @@ +package scheduler + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" +) + +func TestOnInterval(t *testing.T) { + cases := []struct { + name string + now string + ii []string + match bool + err bool + }{ + { + "empty", + time.Now().Format(time.RFC3339), + []string{}, + false, + false, + }, + { + "nil", + time.Now().Format(time.RFC3339), + nil, + false, + false, + }, + { + "next minute", + "2019-10-10T10:11:00Z", + []string{"* * * * *"}, + true, + false, + }, + { + "every 5 minutes", + "2019-10-10T10:15:00Z", + []string{"*/5 * * * *"}, + true, + false, + }, + { + "not full minute", + "2019-10-10T10:15:50Z", + []string{"* * * * *"}, + false, + false, + }, + { + "invalid format", + "2019-10-10T10:15:50Z", + []string{":P"}, + false, + true, + }, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + now, err := time.Parse(time.RFC3339, c.now) + assert.NoError(t, err) + + match, err := onInterval(now, c.ii...) + if !c.err { + assert.NoError(t, err) + } + + assert.Equal(t, c.match, match) + }) + } + + // touch exported func + OnInterval("* * * * *") + OnInterval(":P") +} + +func TestOnTimestamp(t *testing.T) { + cases := []struct { + name string + now string + ii []string + match bool + err bool + }{ + { + "empty", + time.Now().Format(time.RFC3339), + []string{}, + false, + false, + }, + { + "nil", + time.Now().Format(time.RFC3339), + nil, + false, + false, + }, + { + "must match", + "2019-10-10T10:11:00Z", + []string{"2019-10-10T10:11:00Z"}, + true, + false, + }, + { + "must not match", + "2019-10-10T10:15:00Z", + []string{"2020-10-10T10:15:00Z"}, + false, + false, + }, + { + "invalid format", + "2019-10-10T10:15:00Z", + []string{":P"}, + false, + true, + }, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + now, err := time.Parse(time.RFC3339, c.now) + assert.NoError(t, err) + + match, err := onTimestamp(now, c.ii...) + if !c.err { + assert.NoError(t, err) + } + + assert.Equal(t, c.match, match) + }) + } + + // touch exported func + OnTimestamp("2019-10-10T10:15:00Z") + OnTimestamp(":P") +} diff --git a/pkg/scheduler/service.go b/pkg/scheduler/service.go index d9d956e7e..c8f4ed2fb 100644 --- a/pkg/scheduler/service.go +++ b/pkg/scheduler/service.go @@ -18,7 +18,7 @@ type ( dispatcher dispatcher // Simple chan to control if service is running or not - active chan bool + ticker *time.Ticker } dispatcher interface { @@ -28,6 +28,9 @@ type ( const ( defaultInterval = time.Minute + + // There should not be more than 2 per each service ( * 2 [interval + timestamp]) + maxEvents = 16 ) var ( @@ -40,8 +43,8 @@ var ( // Setup configures global scheduling service func Setup(log *zap.Logger, d dispatcher, interval time.Duration) { if gScheduler != nil { - // shut it down - gScheduler.active <- false + // shut down current global scheduler + gScheduler.Stop() } gScheduler = NewService(log, d, interval) @@ -61,6 +64,7 @@ func NewService(log *zap.Logger, d dispatcher, interval time.Duration) *service log: log.Named("scheduler"), interval: interval, dispatcher: d, + events: make([]eventbus.Event, 0, maxEvents), } return svc @@ -71,19 +75,31 @@ func (svc *service) OnTick(events ...eventbus.Event) { svc.events = append(svc.events, events...) } +func (svc *service) Stop() { + if svc.ticker == nil { + svc.log.Debug("already stopped") + } else { + svc.log.Debug("stopping") + svc.ticker.Stop() + svc.ticker = nil + } +} + // Run starts event scheduler service -func (svc service) Start(ctx context.Context) { - if svc.active != nil { +func (svc *service) Start(ctx context.Context) { + if svc.ticker != nil { + svc.log.Debug("already started") return } - svc.active = make(chan bool, 1) + svc.ticker = &time.Ticker{} + go func() { defer sentry.Recover() nextTick := now().Truncate(svc.interval).Add(svc.interval) - svc.log.Info( + svc.log.Debug( "starting", zap.Time("delay", nextTick), zap.Duration("interval", svc.interval), @@ -91,30 +107,34 @@ func (svc service) Start(ctx context.Context) { // Wait until start of the next interval time.Sleep(nextTick.Sub(now())) - svc.log.Info("started") + svc.log.Debug("started") // start with first interval svc.dispatch(ctx) - ticker := time.NewTicker(svc.interval) - defer ticker.Stop() - defer svc.log.Info("stopped") + svc.ticker = time.NewTicker(svc.interval) + loop: for { select { - case <-svc.active: - svc.log.Info("unactivated") - return - case <-ctx.Done(): - svc.log.Info("done") - return - case <-ticker.C: + case <-svc.ticker.C: svc.dispatch(ctx) + + case <-ctx.Done(): + svc.log.Debug("done") + break loop } } + defer svc.log.Debug("stopped") + svc.ticker.Stop() + svc.ticker = nil }() } +func (svc service) Started() (started bool) { + return svc.ticker != nil +} + func (svc service) dispatch(ctx context.Context) { for _, ev := range svc.events { go func(ev eventbus.Event) { diff --git a/pkg/scheduler/service_test.go b/pkg/scheduler/service_test.go new file mode 100644 index 000000000..fcb60d310 --- /dev/null +++ b/pkg/scheduler/service_test.go @@ -0,0 +1,33 @@ +package scheduler + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" + "go.uber.org/zap" + + "github.com/cortezaproject/corteza-server/pkg/eventbus" +) + +func TestMainServiceFunctions(t *testing.T) { + r := require.New(t) + + const ( + loopInterval = time.Millisecond + actionWait = loopInterval * 10 + ) + + r.Nil(gScheduler) + Setup(zap.NewNop(), eventbus.New(), time.Second) + r.NotNil(gScheduler) + r.False(gScheduler.Started()) + r.Equal(gScheduler, Service()) + Service().Start(context.Background()) + time.Sleep(actionWait) + r.True(gScheduler.Started()) + gScheduler.Stop() + time.Sleep(actionWait) + r.False(gScheduler.Started()) +}