diff --git a/server/cmd/main.go b/server/cmd/main.go index b561e95..6ae3df9 100644 --- a/server/cmd/main.go +++ b/server/cmd/main.go @@ -11,6 +11,12 @@ import ( "syscall" "time" + // Embeds the IANA zone database in the binary. Load-bearing: server/Dockerfile + // builds on Alpine, which ships no zoneinfo, so without this + // time.LoadLocation("Europe/London") fails in production and every workflow + // schedule silently falls back to UTC. + _ "time/tzdata" + "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/api" "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/auth" "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/bus" @@ -18,6 +24,7 @@ import ( grpcserver "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/grpc" "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/monitorsched" "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/services" + "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/workflowsched" "github.com/gin-gonic/gin" ) @@ -179,6 +186,10 @@ func serve() { services.StartAuditSweeper(jobCtx) services.StartReaper(jobCtx) monitorsched.Start(jobCtx) + workflowsched.Start(jobCtx, workflowsched.Deps{ + TriggerWorkflow: services.TriggerWorkflow, + LogEvent: services.LogEvent, + }) ticker := time.NewTicker(2 * time.Minute) defer ticker.Stop() diff --git a/server/internal/api/workflows.go b/server/internal/api/workflows.go index 2f2d6f9..62bd5cc 100644 --- a/server/internal/api/workflows.go +++ b/server/internal/api/workflows.go @@ -13,7 +13,9 @@ import ( "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/auth" "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/models" "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/services" + "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/workflowsched" "github.com/gin-gonic/gin" + "go.mongodb.org/mongo-driver/v2/mongo" ) func registerWorkflowRoutes(g *gin.RouterGroup) { @@ -34,6 +36,8 @@ func registerWorkflowRoutes(g *gin.RouterGroup) { g.DELETE("/workflows/:id", deleteWorkflow) g.POST("/workflows/:id/run", runWorkflow) g.GET("/workflows/:id/runs", listWorkflowRuns) + g.PUT("/workflows/:id/schedule", putWorkflowSchedule) + g.GET("/workflows/:id/schedule/preview", previewWorkflowSchedule) g.GET("/runs/:runId", getRun) g.POST("/runs/:runId/cancel", cancelRun) @@ -386,3 +390,51 @@ func cancelRun(c *gin.Context) { services.LogEvent(auth.InstanceID(c), "workflow.run_cancelled", actorFromCtx(c), "", c.Param("runId"), "run cancelled") c.JSON(http.StatusOK, gin.H{"cancelled": true}) } + +func putWorkflowSchedule(c *gin.Context) { + var body models.Schedule + if err := c.ShouldBindJSON(&body); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid body"}) + return + } + + instanceID := auth.InstanceID(c) + next, err := services.SetSchedule(instanceID, c.Param("id"), &body) + if err != nil { + if errors.Is(err, workflowsched.ErrBadSchedule) { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + if errors.Is(err, mongo.ErrNoDocuments) { + c.JSON(http.StatusNotFound, gin.H{"error": "workflow not found"}) + return + } + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + + services.LogEvent(instanceID, "workflow.schedule_updated", actorFromCtx(c), "", c.Param("id"), + fmt.Sprintf("schedule %q %s enabled=%v", body.Cron, body.TZ, body.Enabled)) + c.JSON(http.StatusOK, gin.H{"schedule": body, "next_run_at": next}) +} + +// previewWorkflowSchedule exists so the browser and the scheduler agree on +// what a cron string means. A client-side cron parser that disagrees with the +// server by one field is a bug found in production, at night. +func previewWorkflowSchedule(c *gin.Context) { + expr := c.Query("cron") + tz := c.Query("tz") + + occurrences := make([]time.Time, 0, 3) + from := time.Now() + for i := 0; i < 3; i++ { + next, err := workflowsched.NextOccurrence(expr, tz, from) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + occurrences = append(occurrences, next) + from = next + } + c.JSON(http.StatusOK, gin.H{"occurrences": occurrences}) +} diff --git a/server/internal/workflowsched/sched.go b/server/internal/workflowsched/sched.go new file mode 100644 index 0000000..da57cfb --- /dev/null +++ b/server/internal/workflowsched/sched.go @@ -0,0 +1,150 @@ +package workflowsched + +import ( + "context" + "log" + "time" + + "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/db" + "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/models" + "go.mongodb.org/mongo-driver/v2/bson" + "go.mongodb.org/mongo-driver/v2/mongo/options" +) + +const tickInterval = 30 * time.Second + +// Deps are the service functions the loop needs. They are injected rather than +// imported because services already imports this package for NextOccurrence, +// and a package cannot import its own importer. +type Deps struct { + TriggerWorkflow func(instanceID, workflowID, actor string) (string, error) + LogEvent func(instanceID, eventType, actor, serverID, keyID, details string) +} + +// Start runs the scheduler until ctx is cancelled. It is called inside +// bus.RunAsLeader("housekeeping", …) alongside monitorsched and the sweepers: +// one role, one lock. N replicas each running this loop would fire every +// scheduled workflow N times. +func Start(ctx context.Context, deps Deps) { + go func() { + ticker := time.NewTicker(tickInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + tick(ctx, deps) + } + } + }() +} + +func tick(ctx context.Context, deps Deps) { + now := time.Now() + + cur, err := db.Col("workflows").Find(ctx, bson.M{ + "schedule.enabled": true, + "next_run_at": bson.M{"$lte": now}, + }) + if err != nil { + log.Printf("workflowsched: find due: %v", err) + return + } + defer cur.Close(ctx) + + var due []models.Workflow + if err := cur.All(ctx, &due); err != nil { + log.Printf("workflowsched: decode due: %v", err) + return + } + + for _, wf := range due { + if ctx.Err() != nil { + return + } + process(ctx, deps, wf, now) + } +} + +func process(ctx context.Context, deps Deps, wf models.Workflow, now time.Time) { + if wf.NextRunAt == nil || wf.Schedule == nil { + return + } + dueAt := *wf.NextRunAt + + next, err := NextOccurrence(wf.Schedule.Cron, wf.Schedule.TZ, now) + if err != nil { + // A schedule that no longer parses cannot be advanced, and leaving + // next_run_at in the past would spin this loop every 30 seconds + // forever. Disable it and say so. + log.Printf("workflowsched: workflow %s has an unusable schedule, disabling: %v", wf.WorkflowID, err) + disable(ctx, deps, wf, err.Error()) + return + } + + // The claim. Matching on the current next_run_at as well as the id means a + // second process reaching this document after another has claimed it + // matches nothing and does nothing. This — not the leader lock — is what + // makes a double fire impossible; the lock only keeps it cheap. + res, err := db.Col("workflows").UpdateOne(ctx, + bson.M{"workflow_id": wf.WorkflowID, "next_run_at": dueAt}, + bson.M{"$set": bson.M{"next_run_at": next}}, + ) + if err != nil { + log.Printf("workflowsched: claim %s: %v", wf.WorkflowID, err) + return + } + if res.MatchedCount == 0 { + return // claimed elsewhere + } + + switch Decide(dueAt, now, hasActiveRun(ctx, wf.InstanceID, wf.WorkflowID)) { + case SkipMissed: + recordSkip(ctx, deps, wf, string(SkipMissed), dueAt, now) + case SkipRunning: + recordSkip(ctx, deps, wf, string(SkipRunning), dueAt, now) + case Fire: + if _, err := deps.TriggerWorkflow(wf.InstanceID, wf.WorkflowID, "schedule"); err != nil { + log.Printf("workflowsched: trigger %s: %v", wf.WorkflowID, err) + recordSkip(ctx, deps, wf, "error: "+err.Error(), dueAt, now) + return + } + _, _ = db.Col("workflows").UpdateOne(ctx, + bson.M{"workflow_id": wf.WorkflowID}, + bson.M{"$set": bson.M{"last_run_at": now}, "$unset": bson.M{"last_skipped": ""}}, + ) + deps.LogEvent(wf.InstanceID, "workflow.scheduled_run", "schedule", "", "", + "workflow "+wf.Name+" started on schedule") + } +} + +func hasActiveRun(ctx context.Context, instanceID, workflowID string) bool { + err := db.Col("workflow_runs").FindOne(ctx, bson.M{ + "instance_id": instanceID, + "workflow_id": workflowID, + "status": "running", + }, options.FindOne().SetProjection(bson.M{"_id": 1})).Err() + return err == nil +} + +func recordSkip(ctx context.Context, deps Deps, wf models.Workflow, reason string, due, at time.Time) { + _, _ = db.Col("workflows").UpdateOne(ctx, + bson.M{"workflow_id": wf.WorkflowID}, + bson.M{"$set": bson.M{"last_skipped": models.Skip{Reason: reason, Due: due, At: at}}}, + ) + deps.LogEvent(wf.InstanceID, "workflow.schedule_skipped", "schedule", "", "", + "workflow "+wf.Name+" skipped "+due.Format(time.RFC3339)+": "+reason) +} + +func disable(ctx context.Context, deps Deps, wf models.Workflow, reason string) { + _, _ = db.Col("workflows").UpdateOne(ctx, + bson.M{"workflow_id": wf.WorkflowID}, + bson.M{ + "$set": bson.M{"schedule.enabled": false}, + "$unset": bson.M{"next_run_at": ""}, + }, + ) + deps.LogEvent(wf.InstanceID, "workflow.schedule_disabled", "schedule", "", "", + "workflow "+wf.Name+" schedule disabled: "+reason) +}