Started reimplementing scheduler

backend/go-app/go.mod

@@ -8,6 +8,7 @@ require (
 	cloud.google.com/go/pubsub v1.3.1
 	cloud.google.com/go/storage v1.7.0
 	github.com/basgys/goxml2json v1.1.0
+	github.com/carlescere/scheduler v0.0.0-20170109141437-ee74d2f83d82 // indirect
 	github.com/docker/distribution v2.7.1+incompatible // indirect
 	github.com/docker/docker v1.13.1
 	github.com/docker/go-connections v0.4.0

backend/go-app/go.sum

@@ -38,6 +38,8 @@ github.com/anmitsu/go-shlex v0.0.0-20161002113705-648efa622239/go.mod h1:2FmKhYU
 github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5/go.mod h1:wHh0iHkYZB8zMSxRWpUBQtwG5a7fFgvEO+odwuTv2gs=
 github.com/basgys/goxml2json v1.1.0 h1:4ln5i4rseYfXNd86lGEB+Vi652IsIXIvggKM/BhUKVw=
 github.com/basgys/goxml2json v1.1.0/go.mod h1:wH7a5Np/Q4QoECFIU8zTQlZwZkrilY0itPfecMw41Dw=
+github.com/carlescere/scheduler v0.0.0-20170109141437-ee74d2f83d82 h1:9bAydALqAjBfPHd/eAiJBHnMZUYov8m2PkXVr+YGQeI=
+github.com/carlescere/scheduler v0.0.0-20170109141437-ee74d2f83d82/go.mod h1:tyA14J0sA3Hph4dt+AfCjPrYR13+vVodshQSM7km9qw=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
 github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI=
 github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI=

backend/go-app/walkoff.go

@@ -28,6 +28,8 @@ import (
 	"github.com/go-git/go-git/v5"
 
 	"github.com/go-git/go-git/v5/storage/memory"
+
+	newscheduler "github.com/carlescere/scheduler"
 	//"github.com/gorilla/websocket"
 	//"google.golang.org/appengine"
 	//"google.golang.org/appengine/memcache"
@@ -54,6 +56,17 @@ var shuffleTestPath = "./shuffle-test-258209-5a2e8d7e508a.json"
 //	},
 //}
 
+type ExecutionRequest struct {
+	ExecutionId       string   `json:"execution_id"`
+	ExecutionArgument string   `json:"execution_argument"`
+	WorkflowId        string   `json:"workflow_id"`
+	Environments      []string `json:"environments"`
+	Authorization     string   `json:"authorization"`
+	Status            string   `json:"status"`
+	Start             string   `json:"start"`
+	Type              string   `json:"type"`
+}
+
 type Org struct {
 	Name  string `json:"name"`
 	Org   string `json:"org"`
@@ -371,14 +384,7 @@ func getWorkflowQueue(ctx context.Context, id string) (ExecutionRequestWrapper,
 
 // Frequency = cronjob OR minutes between execution
 func createSchedule(ctx context.Context, scheduleId, workflowId, name, frequency string, body []byte) error {
-	c, err := scheduler.NewCloudSchedulerClient(ctx)
-	if err != nil {
-		log.Printf("%s", err)
-		return err
-	}
-
 	testSplit := strings.Split(frequency, "*")
-	log.Println(len(testSplit))
 	cronJob := ""
 	if len(testSplit) > 5 {
 		cronJob = frequency
@@ -400,31 +406,61 @@ func createSchedule(ctx context.Context, scheduleId, workflowId, name, frequency
 		return errors.New("cronJob isn't formatted correctly")
 	}
 
-	req := &schedulerpb.CreateJobRequest{
+	log.Printf("CRON: %s, body: %s", cronJob, string(body))
-		Parent: fmt.Sprintf("projects/%s/locations/europe-west2", gceProject),
+
-		Job: &schedulerpb.Job{
+	// FIXME:
-			Name:        fmt.Sprintf("projects/%s/locations/europe-west2/jobs/schedule_%s", gceProject, scheduleId),
+	// This may run multiple places if multiple servers,
-			Schedule:    cronJob,
+	// but that's a future problem
-			Description: name,
+	job := func() {
-			Target: &schedulerpb.Job_HttpTarget{
+		request := &http.Request{
-				HttpTarget: &schedulerpb.HttpTarget{
+			Method: "POST",
-					Uri:        fmt.Sprintf("https://shuffler.io/api/v1/workflows/%s/execute", workflowId),
+			Body:   ioutil.NopCloser(strings.NewReader(string(body))),
-					HttpMethod: 1,
-					Headers: map[string]string{
-						"Authorization": "",
-					},
-					Body: body,
-				},
-			},
-		},
-		// TODO: Fill request struct fields.
 		}
-	resp, err := c.CreateJob(ctx, req)
+
+		_, _, err := handleExecution(workflowId, Workflow{}, request)
+		if err == nil {
+			log.Printf("Failed to execute: %s", err)
+		}
+	}
+
+	// FIXME - Create a real schedule based on cron:
+	// 1. Parse the cron in a function to match this schedule
+	// 2. Make main init check for schedules that aren't running
+	_, err := newscheduler.Every(5).Seconds().NotImmediately().Run(job)
 	if err != nil {
-		log.Printf("%s", err)
+		log.Printf("Failed to schedule workflow: %s", err)
 		return err
 	}
-	_ = resp
+
+	return errors.New("ERROR!!")
+
+	//log.Printf("REQUEST: %#v", executionRequest)
+
+	//req := &schedulerpb.CreateJobRequest{
+	//	Parent: fmt.Sprintf("projects/%s/locations/europe-west2", gceProject),
+	//	Job: &schedulerpb.Job{
+	//		Name:        fmt.Sprintf("projects/%s/locations/europe-west2/jobs/schedule_%s", gceProject, scheduleId),
+	//		Schedule:    cronJob,
+	//		Description: name,
+	//		Target: &schedulerpb.Job_HttpTarget{
+	//			HttpTarget: &schedulerpb.HttpTarget{
+	//				Uri:        fmt.Sprintf("https://shuffler.io/api/v1/workflows/%s/execute", workflowId),
+	//				HttpMethod: 1,
+	//				Headers: map[string]string{
+	//					"Authorization": "",
+	//				},
+	//				Body: body,
+	//			},
+	//		},
+	//	},
+	//	// TODO: Fill request struct fields.
+	//}
+	//resp, err := c.CreateJob(ctx, req)
+	//if err != nil {
+	//	log.Printf("%s", err)
+	//	return err
+	//}
+	//_ = resp
 
 	return nil
 }
@@ -1443,15 +1479,6 @@ func getWorkflowLocal(fileId string, request *http.Request) ([]byte, error) {
 	return body, nil
 }
 
-type ExecutionRequest struct {
-	ExecutionId       string   `json:"execution_id"`
-	ExecutionArgument string   `json:"execution_argument"`
-	WorkflowId        string   `json:"workflow_id"`
-	Authorization     string   `json:"authorization"`
-	Environments      []string `json:"environments"`
-	Start             string   `json:"start"`
-}
-
 func abortExecution(resp http.ResponseWriter, request *http.Request) {
 	cors := handleCors(resp, request)
 	if cors {

functions/onprem/orborus/orborus.go

@@ -37,11 +37,11 @@ type ExecutionRequestWrapper struct {
 
 type ExecutionRequest struct {
 	ExecutionId       string `json:"execution_id"`
+	ExecutionArgument string `json:"execution_argument"`
 	WorkflowId        string `json:"workflow_id"`
 	Authorization     string `json:"authorization"`
-	ExecutionArgument string   `json:"execution_argument"`
-	Environments      []string `json:"environments"`
 	Status            string `json:"status"`
+	Type              string `json:"type"`
 }
 
 // Deploys the internal worker whenever something happens
@@ -207,6 +207,7 @@ func main() {
 			time.Sleep(time.Duration(sleepTime) * time.Second)
 			continue
 		}
+
 		// FIXME - add check for StatusCode
 		if newresp.StatusCode != 200 {
 			if hasStarted {
@@ -243,7 +244,8 @@ func main() {
 		}
 
 		if hasStarted && len(executionRequests.Data) > 0 {
-			log.Println(string(body))
+			log.Printf("Body: %s", string(body))
+			// Type string `json:"type"`
 		}
 
 		if len(executionRequests.Data) == 0 {
@@ -259,7 +261,13 @@ func main() {
 		// New, abortable version. Should check executionid and remove everything else
 		var toBeRemoved ExecutionRequestWrapper
 		for _, execution := range executionRequests.Data {
-			log.Println(execution.ExecutionArgument)
+			log.Printf("Argument: %#v", execution.ExecutionArgument)
+
+			if execution.Type == "schedule" {
+				log.Printf("SOMETHING ELSE :O: %s", execution.Type)
+				continue
+			}
+
 			if execution.Status == "ABORT" || execution.Status == "FAILED" {
 				log.Printf("Executionstatus issue: ", execution.Status)
 			}