support cron scheduling (#1695)

motatoes · web-flow · commit b552ebd7556e · 2024-09-05T11:25:59.000+01:00
* support cron scheduling
diff --git a/fly-staging.toml b/fly-staging.toml
@@ -9,7 +9,7 @@ kill_signal = 'SIGINT'
 kill_timeout = '5s'
 
 [env]
-  HOSTNAME = 'https://next-backend-staging.digger.dev'
+  DIGGER_HOSTNAME = 'https://next-backend-staging.digger.dev'
   
 [build]
   dockerfile = 'Dockerfile_next'
diff --git a/fly.toml b/fly.toml
@@ -9,7 +9,7 @@ kill_signal = 'SIGINT'
 kill_timeout = '5s'
 
 [env]
-  HOSTNAME = 'https://next-backend.digger.dev'
+  DIGGER_HOSTNAME = 'https://next-backend.digger.dev'
   
 [build]
   dockerfile = 'Dockerfile_next'
diff --git a/next/controllers/drift.go b/next/controllers/drift.go
@@ -1,9 +1,17 @@
 package controllers
 
 import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"github.com/diggerhq/digger/next/dbmodels"
+	"github.com/diggerhq/digger/next/utils"
 	"github.com/gin-gonic/gin"
 	"log"
 	"net/http"
+	"net/url"
+	"os"
+	"time"
 )
 
 type TriggerDriftRequest struct {
@@ -23,10 +31,78 @@ func (d DiggerController) TriggerDriftDetectionForProject(c *gin.Context) {
 	}
 	projectId := request.ProjectId
 
+	log.Printf("Drift requests for project: %v", projectId)
+	
 	c.JSON(200, gin.H{
 		"status":     "successful",
 		"project_id": projectId,
 	})
 	return
 
 }
+
+func (d DiggerController) TriggerCronForMatchingProjects(c *gin.Context) {
+	webhookSecret := os.Getenv("DIGGER_WEBHOOK_SECRET")
+	diggerHostName := os.Getenv("DIGGER_HOSTNAME")
+
+	driftUrl, err := url.JoinPath(diggerHostName, "_internal/trigger_drift")
+	if err != nil {
+		log.Printf("could not form drift url: %v", err)
+		c.JSON(500, gin.H{"error": "could not form drift url"})
+		return
+	}
+
+	p := dbmodels.DB.Query.Project
+	driftEnabledProjects, err := dbmodels.DB.Query.Project.Where(p.IsDriftDetectionEnabled.Is(true)).Find()
+	if err != nil {
+		log.Printf("could not fetch drift enabled projects: %v", err)
+		c.JSON(500, gin.H{"error": "could not fetch drift enabled projects"})
+		return
+	}
+
+	for _, proj := range driftEnabledProjects {
+		matches, err := utils.MatchesCrontab(proj.DriftCrontab, time.Now())
+		if err != nil {
+			log.Printf("could not check for matching crontab, %v", err)
+			// TODO: send metrics here
+			continue
+		}
+
+		if matches {
+			payload := TriggerDriftRequest{ProjectId: proj.ID}
+
+			// Convert payload to JSON
+			jsonPayload, err := json.Marshal(payload)
+			if err != nil {
+				fmt.Println("Error marshaling JSON:", err)
+				return
+			}
+
+			// Create a new request
+			req, err := http.NewRequest("POST", driftUrl, bytes.NewBuffer(jsonPayload))
+			if err != nil {
+				fmt.Println("Error creating request:", err)
+				return
+			}
+
+			// Set headers
+			req.Header.Set("Content-Type", "application/json")
+			req.Header.Set("Authorization", fmt.Sprintf("Bearer %v", webhookSecret))
+
+			// Send the request
+			client := &http.Client{}
+			resp, err := client.Do(req)
+			if err != nil {
+				fmt.Println("Error sending request:", err)
+				return
+			}
+			defer resp.Body.Close()
+
+			// Get the status code
+			statusCode := resp.StatusCode
+			if statusCode != 200 {
+				log.Printf("got unexpected drift status for project: %v - status: %v", proj.ID, statusCode)
+			}
+		}
+	}
+}
diff --git a/next/controllers/github.go b/next/controllers/github.go
@@ -119,7 +119,7 @@ func GithubAppSetup(c *gin.Context) {
 		Webhook               *githubWebhook    `json:"hook_attributes"`
 	}
 
-	host := os.Getenv("HOSTNAME")
+	host := os.Getenv("DIGGER_HOSTNAME")
 	manifest := &githubAppRequest{
 		Name:        fmt.Sprintf("Digger app %v", rand.Int31()),
 		Description: fmt.Sprintf("Digger hosted at %s", host),
@@ -533,7 +533,7 @@ func ConvertJobsToDiggerJobs(jobType orchestrator_scheduler.DiggerCommand, vcsTy
 	}
 	organisationName := organisation.Title
 
-	backendHostName := os.Getenv("HOSTNAME")
+	backendHostName := os.Getenv("DIGGER_HOSTNAME")
 
 	log.Printf("Number of Jobs: %v\n", len(jobsMap))
 	marshalledJobsMap := map[string][]byte{}
diff --git a/next/controllers/github_after_merge.go b/next/controllers/github_after_merge.go
@@ -24,7 +24,7 @@ func handlePushEventApplyAfterMerge(gh nextutils.GithubClientProvider, payload *
 	requestedBy := *payload.Sender.Login
 	ref := *payload.Ref
 	targetBranch := strings.ReplaceAll(ref, "refs/heads/", "")
-	backendHostName := os.Getenv("HOSTNAME")
+	backendHostName := os.Getenv("DIGGER_HOSTNAME")
 
 	link, err := dbmodels.DB.GetGithubAppInstallationLink(installationId)
 	if err != nil {
diff --git a/next/go.mod b/next/go.mod
@@ -17,9 +17,9 @@ require (
 	github.com/google/go-github/v61 v61.0.0
 	github.com/google/uuid v1.6.0
 	github.com/orandin/slog-gorm v1.3.2
+	github.com/robfig/cron/v3 v3.0.1
 	github.com/samber/lo v1.46.0
 	github.com/samber/slog-gin v1.13.3
-	github.com/stretchr/testify v1.9.0
 	github.com/supabase-community/supabase-go v0.0.4
 	golang.org/x/oauth2 v0.22.0
 	gorm.io/driver/postgres v1.5.9
@@ -105,7 +105,6 @@ require (
 	github.com/creack/pty v1.1.17 // indirect
 	github.com/cyphar/filepath-securejoin v0.2.4 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
-	github.com/diggerhq/digger/cli v0.0.0-20240705091808-75187a7aae8e // indirect
 	github.com/dimchansky/utfbom v1.1.1 // indirect
 	github.com/dineshba/tf-summarize v0.3.10 // indirect
 	github.com/emirpasic/gods v1.18.1 // indirect
@@ -246,6 +245,7 @@ require (
 	github.com/spf13/cast v1.6.0 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
 	github.com/spf13/viper v1.18.2 // indirect
+	github.com/stretchr/testify v1.9.0 // indirect
 	github.com/subosito/gotenv v1.6.0 // indirect
 	github.com/supabase-community/functions-go v0.0.0-20220927045802-22373e6cb51d // indirect
 	github.com/supabase-community/gotrue-go v1.2.0 // indirect
diff --git a/next/go.sum b/next/go.sum
@@ -461,8 +461,6 @@ github.com/dgraph-io/ristretto v0.1.1/go.mod h1:S1GPSBCYCIhmVNfcth17y2zZtQT6wzkz
 github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
 github.com/dgryski/trifles v0.0.0-20200323201526-dd97f9abfb48 h1:fRzb/w+pyskVMQ+UbP35JkH8yB7MYb4q/qhBarqZE6g=
 github.com/dgryski/trifles v0.0.0-20200323201526-dd97f9abfb48/go.mod h1:if7Fbed8SFyPtHLHbg49SI7NAdJiC5WIA09pe59rfAA=
-github.com/diggerhq/digger/cli v0.0.0-20240705091808-75187a7aae8e h1:aRBJ92ZbJc6VQXx6zPihHuQoDotstDTwUi3C8gdbdgw=
-github.com/diggerhq/digger/cli v0.0.0-20240705091808-75187a7aae8e/go.mod h1:+UUif/7rqA5ElbNiYXyu6adjpXcafe5nSrY+IvFoJVA=
 github.com/dimchansky/utfbom v1.1.0/go.mod h1:rO41eb7gLfo8SF1jd9F8HplJm1Fewwi4mQvIirEdv+8=
 github.com/dimchansky/utfbom v1.1.1 h1:vV6w1AhK4VMnhBno/TPVCoK9U/LP0PkLCS9tbxHdi/U=
 github.com/dimchansky/utfbom v1.1.1/go.mod h1:SxdoEBH5qIqFocHMyGOXVAybYJdr71b1Q/j0mACtrfE=
@@ -1137,6 +1135,8 @@ github.com/rcrowley/go-metrics v0.0.0-20200313005456-10cdbea86bc0 h1:MkV+77GLUNo
 github.com/rcrowley/go-metrics v0.0.0-20200313005456-10cdbea86bc0/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
 github.com/rivo/uniseg v0.2.0 h1:S1pD9weZBuJdFmowNwbpi7BJ8TNftyUImj/0WQi72jY=
 github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
+github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
+github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
 github.com/rogpeppe/fastuuid v0.0.0-20150106093220-6724a57986af/go.mod h1:XWv6SoW27p1b0cqNHllgS5HIMJraePCO15w5zCzIWYg=
 github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
diff --git a/next/main.go b/next/main.go
@@ -77,6 +77,9 @@ func main() {
 	r.POST("/github-app-webhook", diggerController.GithubAppWebHook)
 
 	r.POST("/_internal/process_runs_queue", middleware.WebhookAuth(), diggerController.ProcessRunQueueItems)
+	// process all drift crontabs
+	r.POST("/_internal/process_drift", middleware.WebhookAuth(), diggerController.TriggerCronForMatchingProjects)
+	// trigger for specific project
 	r.POST("/_internal/trigger_drift", middleware.WebhookAuth(), diggerController.TriggerDriftDetectionForProject)
 	//authorized := r.Group("/")
 	//authorized.Use(middleware.GetApiMiddleware(), middleware.AccessLevel(dbmodels.CliJobAccessType, dbmodels.AccessPolicyType, models.AdminPolicyType))
diff --git a/next/scripts/cron/process_drift.query b/next/scripts/cron/process_drift.query
@@ -0,0 +1,14 @@
+select
+  cron.schedule(
+    'invoke-function-every-half-minute',
+    '30 seconds',
+    $$
+    select
+      net.http_post(
+          url:='https://{DIGGER_HOSTNAME}/_internal/process_drift',
+          headers:=jsonb_build_object('Content-Type','application/json', 'Authorization', 'Bearer ' || 'abc123'),
+          body:=jsonb_build_object('time', now() ),
+          timeout_milliseconds:=5000
+      ) as request_id;
+    $$
+  );
diff --git a/next/scripts/cron/process_runs_queue.query b/next/scripts/cron/process_runs_queue.query
@@ -0,0 +1,15 @@
+select
+  cron.schedule(
+    'invoke-function-every-half-minute',
+    '30 seconds',
+    $$
+    select
+      net.http_post(
+          url:='https://{DIGGER_HOSTNAME}/_internal/process_runs_queue',
+          headers:='{"Content-Type": "application/json", "Authorization": "Bearer abc123"}'::jsonb,
+          body:='{}'::jsonb
+      ) as request_id;
+    $$
+  );
+
+
diff --git a/next/utils/crontab.go b/next/utils/crontab.go
@@ -0,0 +1,22 @@
+package utils
+
+import (
+	"fmt"
+	"github.com/robfig/cron/v3"
+	"time"
+)
+
+func MatchesCrontab(cronString string, timestamp time.Time) (bool, error) {
+	// Parse the crontab string
+	schedule, err := cron.ParseStandard(cronString)
+	if err != nil {
+		return false, fmt.Errorf("failed to parse crontab string: %w", err)
+	}
+
+	// Round down the timestamp to the nearest minute
+	roundedTime := timestamp.Truncate(time.Minute)
+
+	// Check if the rounded time matches the schedule
+	nextTime := schedule.Next(roundedTime.Add(-time.Minute))
+	return nextTime.Equal(roundedTime), nil
+}
diff --git a/next/utils/crontab_test.go b/next/utils/crontab_test.go
@@ -0,0 +1,31 @@
+package utils
+
+import (
+	"fmt"
+	"github.com/stretchr/testify/assert"
+	"testing"
+	"time"
+)
+
+func TestCrontTabMatching(t *testing.T) {
+	cronString := "*/15 * * * *" // Every 15 minutes
+	timestamp := time.Date(2023, 5, 1, 12, 30, 30, 0, time.UTC)
+
+	matches, err := MatchesCrontab(cronString, timestamp)
+	if err != nil {
+		fmt.Printf("Error: %v\n", err)
+		return
+	}
+	assert.True(t, matches)
+
+	cronString = "*/15 * * * *" // Every 15 minutes
+	timestamp = time.Date(2022, 5, 1, 12, 12, 30, 0, time.UTC)
+
+	matches, err = MatchesCrontab(cronString, timestamp)
+	if err != nil {
+		fmt.Printf("Error: %v\n", err)
+		return
+	}
+	assert.False(t, matches)
+
+}

Original file line number	Diff line number	Diff line change
`@@ -119,7 +119,7 @@ func GithubAppSetup(c *gin.Context) {`
`119`	`119`	Webhook *githubWebhook `json:"hook_attributes"`
`120`	`120`	`}`
`121`	`121`
`122`		`- host := os.Getenv("HOSTNAME")`
	`122`	`+ host := os.Getenv("DIGGER_HOSTNAME")`
`123`	`123`	`manifest := &githubAppRequest{`
`124`	`124`	`Name: fmt.Sprintf("Digger app %v", rand.Int31()),`
`125`	`125`	`Description: fmt.Sprintf("Digger hosted at %s", host),`
`@@ -533,7 +533,7 @@ func ConvertJobsToDiggerJobs(jobType orchestrator_scheduler.DiggerCommand, vcsTy`
`533`	`533`	`}`
`534`	`534`	`organisationName := organisation.Title`
`535`	`535`
`536`		`- backendHostName := os.Getenv("HOSTNAME")`
	`536`	`+ backendHostName := os.Getenv("DIGGER_HOSTNAME")`
`537`	`537`
`538`	`538`	`log.Printf("Number of Jobs: %v\n", len(jobsMap))`
`539`	`539`	`marshalledJobsMap := map[string][]byte{}`