feat(stacks): support automated sync for stacks [EE-248] (#5340)

api/scheduler/scheduler.go

@@ -0,0 +1,73 @@
+package scheduler
+
+import (
+	"context"
+	"log"
+	"strconv"
+	"time"
+
+	"github.com/pkg/errors"
+	"github.com/robfig/cron/v3"
+)
+
+type Scheduler struct {
+	crontab     *cron.Cron
+	shutdownCtx context.Context
+}
+
+func NewScheduler(ctx context.Context) *Scheduler {
+	crontab := cron.New(cron.WithChain(cron.Recover(cron.DefaultLogger)))
+	crontab.Start()
+
+	s := &Scheduler{
+		crontab: crontab,
+	}
+
+	if ctx != nil {
+		go func() {
+			<-ctx.Done()
+			s.Shutdown()
+		}()
+	}
+
+	return s
+}
+
+// Shutdown stops the scheduler and waits for it to stop if it is running; otherwise does nothing.
+func (s *Scheduler) Shutdown() error {
+	if s.crontab == nil {
+		return nil
+	}
+
+	log.Println("[DEBUG] Stopping scheduler")
+	ctx := s.crontab.Stop()
+	<-ctx.Done()
+
+	for _, j := range s.crontab.Entries() {
+		s.crontab.Remove(j.ID)
+	}
+
+	err := ctx.Err()
+	if err == context.Canceled {
+		return nil
+	}
+	return err
+}
+
+// StopJob stops the job from being run in the future
+func (s *Scheduler) StopJob(jobID string) error {
+	id, err := strconv.Atoi(jobID)
+	if err != nil {
+		return errors.Wrapf(err, "failed convert jobID %q to int", jobID)
+	}
+	s.crontab.Remove(cron.EntryID(id))
+
+	return nil
+}
+
+// StartJobEvery schedules a new periodic job with a given duration.
+// Returns job id that could be used to stop the given job
+func (s *Scheduler) StartJobEvery(duration time.Duration, job func()) string {
+	entryId := s.crontab.Schedule(cron.Every(duration), cron.FuncJob(job))
+	return strconv.Itoa(int(entryId))
+}

api/scheduler/scheduler_test.go

@@ -0,0 +1,57 @@
+package scheduler
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func Test_CanStartAndTerminate(t *testing.T) {
+	s := NewScheduler(context.Background())
+	s.StartJobEvery(1*time.Minute, func() { fmt.Println("boop") })
+
+	err := s.Shutdown()
+	assert.NoError(t, err, "Shutdown should return no errors")
+	assert.Empty(t, s.crontab.Entries(), "all jobs should have been removed")
+}
+
+func Test_CanTerminateByCancellingContext(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	s := NewScheduler(ctx)
+	s.StartJobEvery(1*time.Minute, func() { fmt.Println("boop") })
+
+	cancel()
+
+	for i := 0; i < 100; i++ {
+		if len(s.crontab.Entries()) == 0 {
+			return
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	t.Fatal("all jobs are expected to be cleaned by now; it might be a timing issue, otherwise implementation defect")
+}
+
+func Test_StartAndStopJob(t *testing.T) {
+	s := NewScheduler(context.Background())
+	defer s.Shutdown()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+
+	var jobOne string
+	var workDone bool
+	jobOne = s.StartJobEvery(time.Second, func() {
+		assert.Equal(t, 1, len(s.crontab.Entries()), "scheduler should have one active job")
+		workDone = true
+
+		s.StopJob(jobOne)
+		cancel()
+	})
+
+	<-ctx.Done()
+	assert.True(t, workDone, "value should been set in the job")
+	assert.Equal(t, 0, len(s.crontab.Entries()), "scheduler should have no active jobs")
+
+}