From b2dfe8d8f991cdaab0a839f2f9c3608d4f2fb522 Mon Sep 17 00:00:00 2001 From: yashsinghcodes Date: Mon, 14 Jul 2025 20:33:50 +0530 Subject: [PATCH] added cron support --- backend/go-app/go.mod | 3 +++ backend/go-app/main.go | 23 ++++++++++++++----- backend/go-app/walkoff.go | 48 +++++++++++++++++++++++++++------------ 3 files changed, 54 insertions(+), 20 deletions(-) diff --git a/backend/go-app/go.mod b/backend/go-app/go.mod index 3c21ac4c..d76f453f 100644 --- a/backend/go-app/go.mod +++ b/backend/go-app/go.mod @@ -74,6 +74,7 @@ require ( github.com/felixge/httpsnoop v1.0.4 // indirect github.com/frikky/schemaless v0.0.17 // indirect github.com/fxamacker/cbor/v2 v2.7.0 // indirect + github.com/go-co-op/gocron v1.37.0 // indirect github.com/go-git/gcfg v1.5.1-0.20230307220236-3a3c6141e376 // indirect github.com/go-jose/go-jose/v4 v4.0.5 // indirect github.com/go-logr/logr v1.4.2 // indirect @@ -118,6 +119,7 @@ require ( github.com/pjbgf/sha1cd v0.3.2 // indirect github.com/pkg/errors v0.9.1 // indirect github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 // indirect + github.com/robfig/cron/v3 v3.0.1 // indirect github.com/sashabaranov/go-openai v1.40.1 // indirect github.com/sendgrid/rest v2.6.9+incompatible // indirect github.com/sendgrid/sendgrid-go v3.16.1+incompatible // indirect @@ -141,6 +143,7 @@ require ( go.opentelemetry.io/otel/sdk/metric v1.36.0 // indirect go.opentelemetry.io/otel/trace v1.36.0 // indirect go.opentelemetry.io/proto/otlp v1.5.0 // indirect + go.uber.org/atomic v1.9.0 // indirect go4.org v0.0.0-20230225012048-214862532bf5 // indirect golang.org/x/net v0.40.0 // indirect golang.org/x/oauth2 v0.30.0 // indirect diff --git a/backend/go-app/main.go b/backend/go-app/main.go index fdd9dc05..3e8a024e 100755 --- a/backend/go-app/main.go +++ b/backend/go-app/main.go @@ -4034,6 +4034,7 @@ func runInitEs(ctx context.Context) { // FIXME: This should ONLY run on one backend instance schedules, err := shuffle.GetAllSchedules(ctx, "ALL") + log.Printf("Schedules %s", schedules) if err != nil { log.Printf("[WARNING] Failed getting schedules during service init: %s", err) } else { @@ -4078,14 +4079,23 @@ func runInitEs(ctx context.Context) { //log.Printf("Schedule: %#v", schedule) //log.Printf("Schedule time: every %d seconds", schedule.Seconds) - jobret, err := newscheduler.Every(schedule.Seconds).Seconds().NotImmediately().Run(job(schedule)) - if err != nil { - log.Printf("[ERROR] Failed to start schedule for workflow %s: %s", schedule.WorkflowId, err) + if schedule.Seconds == 0 && len(schedule.Frequency) > 0 { + cronJob, err := CronScheduler.Cron(schedule.Frequency).Do(job(schedule)) + if err != nil { + log.Printf("[ERROR] Failed to start schedule for workflow %s: %s", schedule.WorkflowId, err) + } else { + log.Printf("[DEBUG] Successfully started schedule for workflow %s", schedule.WorkflowId) + } + cronJobs[schedule.Id] = cronJob } else { - log.Printf("[DEBUG] Successfully started schedule for workflow %s", schedule.WorkflowId) + jobret, err := newscheduler.Every(schedule.Seconds).Seconds().NotImmediately().Run(job(schedule)) + if err != nil { + log.Printf("[ERROR] Failed to start schedule for workflow %s: %s", schedule.WorkflowId, err) + } else { + log.Printf("[DEBUG] Successfully started schedule for workflow %s", schedule.WorkflowId) + } + scheduledJobs[schedule.Id] = jobret } - - scheduledJobs[schedule.Id] = jobret } } @@ -5090,6 +5100,7 @@ func handleAppZipUpload(resp http.ResponseWriter, request *http.Request) { func initHandlers() { var err error ctx := context.Background() + CronScheduler.StartAsync() log.Printf("[DEBUG] Starting Shuffle backend - initializing database connection") //requestCache = cache.New(5*time.Minute, 10*time.Minute) diff --git a/backend/go-app/walkoff.go b/backend/go-app/walkoff.go index f2d8a118..eb174870 100755 --- a/backend/go-app/walkoff.go +++ b/backend/go-app/walkoff.go @@ -26,6 +26,7 @@ import ( uuid "github.com/satori/go.uuid" newscheduler "github.com/carlescere/scheduler" + "github.com/go-co-op/gocron" "github.com/frikky/kin-openapi/openapi3" "github.com/go-git/go-billy/v5" "github.com/go-git/go-billy/v5/memfs" @@ -42,17 +43,22 @@ var baseEnvironment = "onprem" var cloudname = "cloud" var scheduledJobs = map[string]*newscheduler.Job{} +var cronJobs = map[string]*gocron.Job{} var scheduledOrgs = map[string]*newscheduler.Job{} +var CronScheduler = gocron.NewScheduler(time.UTC) + // Frequency = cronjob OR minutes between execution func createSchedule(ctx context.Context, scheduleId, workflowId, name, startNode, frequency, orgId string, body []byte) error { var err error testSplit := strings.Split(frequency, "*") cronJob := "" + isCron := false newfrequency := 0 if len(testSplit) > 5 { cronJob = frequency + isCron = true } else { newfrequency, err = strconv.Atoi(frequency) if err != nil { @@ -65,12 +71,7 @@ func createSchedule(ctx context.Context, scheduleId, workflowId, name, startNode //} else if int(newfrequency) < } - // Reverse. Can't handle CRON, only numbers - if len(cronJob) > 0 { - return errors.New("cronJob isn't formatted correctly") - } - - if newfrequency < 1 { + if newfrequency < 1 && !isCron { return errors.New("Frequency has to be more than 0") } @@ -96,17 +97,27 @@ func createSchedule(ctx context.Context, scheduleId, workflowId, name, startNode } } - log.Printf("[INFO] Starting frequency for execution: %d", newfrequency) + log.Printf("[INFO] Starting frequency for execution: %s", frequency) - //jobret, err := newscheduler.Every(newfrequency).Seconds().NotImmediately().Run(job) - jobret, err := newscheduler.Every(newfrequency).Seconds().Run(job) - if err != nil { - log.Printf("Failed to schedule workflow: %s", err) - return err + if isCron { + cronJob, err := CronScheduler.Cron(cronJob).Do(job) + if err != nil { + log.Printf("[ERROR] Failed to start schedule with cron(%s): %s", cronJob, err) + } + + cronJobs[scheduleId] = cronJob + } else { + //jobret, err := newscheduler.Every(newfrequency).Seconds().NotImmediately().Run(job) + jobret, err := newscheduler.Every(newfrequency).Seconds().Run(job) + if err != nil { + log.Printf("Failed to schedule workflow: %s", err) + return err + } + + scheduledJobs[scheduleId] = jobret } //scheduledJobs = append(scheduledJobs, jobret) - scheduledJobs[scheduleId] = jobret // Doesn't need running/not running. If stopped, we just delete it. timeNow := int64(time.Now().Unix()) @@ -117,6 +128,7 @@ func createSchedule(ctx context.Context, scheduleId, workflowId, name, startNode Argument: string(body), WrappedArgument: bodyWrapper, Seconds: newfrequency, + Frequency: frequency, CreationTime: timeNow, LastModificationtime: timeNow, LastRuntime: timeNow, @@ -1541,7 +1553,15 @@ func deleteSchedule(ctx context.Context, id string) error { value.Lock() } else { // FIXME - allow it to kind of stop anyway? - return errors.New("Can't find the schedule.") + if j, ok := cronJobs[id]; ok { + err := CronScheduler.RemoveByID(j) + if err != nil { + log.Printf("[ERROR] Failed to remove the scheduler %s", err) + return err + } + } else { + return errors.New("Can't find the schedule.") + } } }