Scheduling works onprem!

backend/go-app/main.go

@@ -42,6 +42,7 @@ import (
 
 	// Random
 	xj "github.com/basgys/goxml2json"
+	newscheduler "github.com/carlescere/scheduler"
 	gyaml "github.com/ghodss/yaml"
 	"github.com/satori/go.uuid"
 	"golang.org/x/crypto/bcrypt"
@@ -231,10 +232,12 @@ type AppInfo struct {
 	DestinationApp ScheduleApp `json:"destinationapp,omitempty" datastore:"destinationapp,noindex"`
 }
 
-// Used for the api integrator
+// May 2020: Reused for onprem schedules - Id, Seconds, WorkflowId and argument
-//Username string `datastore:"Username,noindex"`
 type ScheduleOld struct {
 	Id                   string       `json:"id" datastore:"id"`
+	Seconds              int          `json:"seconds" datastore:"seconds"`
+	WorkflowId           string       `json:"workflow_id datastore:"workflow_id", `
+	Argument             string       `json:"argument" datastore:"argument"`
 	AppInfo              AppInfo      `json:"appinfo" datastore:"appinfo,noindex"`
 	Finished             bool         `json:"finished" finished:"id"`
 	BaseAppLocation      string       `json:"base_app_location" datastore:"baseapplocation,noindex"`
@@ -5624,11 +5627,13 @@ func init() {
 		log.Fatalf("DBclient error during init: %s", err)
 	}
 
+	// Setting stats for backend starts (failure count as well)
 	err = increaseStatisticsField(ctx, "backend_executions", "", 1)
 	if err != nil {
 		log.Printf("Failed increasing local stats: %s", err)
 	}
 
+	// Gets environments and inits if it doesn't exist
 	count, err := getEnvironmentCount()
 	if count == 0 && err == nil {
 		item := Environment{
@@ -5666,6 +5671,35 @@ func init() {
 		iterateAppGithubFolders(fs, dir, "", "testing")
 	}
 
+	// Gets schedules and starts them
+	schedules, err := getAllSchedules(ctx)
+	if err != nil {
+		log.Printf("Failed getting schedules during service init: %s", err)
+	} else {
+		log.Printf("Setting up %d schedule(s)", len(schedules))
+		for _, schedule := range schedules {
+			job := func() {
+				request := &http.Request{
+					Method: "POST",
+					Body:   ioutil.NopCloser(strings.NewReader(schedule.Argument)),
+				}
+
+				_, _, err := handleExecution(schedule.WorkflowId, Workflow{}, request)
+				if err != nil {
+					log.Printf("Failed to execute: %s", err)
+				}
+			}
+
+			jobret, err := newscheduler.Every(schedule.Seconds).Seconds().NotImmediately().Run(job)
+			if err != nil {
+				log.Printf("Failed to schedule workflow: %s", err)
+				// FIXME: what now? lol:w
+			}
+
+			scheduledJobs[schedule.Id] = jobret
+		}
+	}
+
 	log.Printf("Finished INIT")
 
 	r := mux.NewRouter()

backend/go-app/walkoff.go

@@ -27,9 +27,8 @@ import (
 	"github.com/go-git/go-billy/v5/memfs"
 	"github.com/go-git/go-git/v5"
 
-	"github.com/go-git/go-git/v5/storage/memory"
-
 	newscheduler "github.com/carlescere/scheduler"
+	"github.com/go-git/go-git/v5/storage/memory"
 	//"github.com/gorilla/websocket"
 	//"google.golang.org/appengine"
 	//"google.golang.org/appengine/memcache"
@@ -43,6 +42,7 @@ var baseEnvironment = "onprem"
 var cloudname = "cloud"
 
 var defaultLocation = "europe-west2"
+var scheduledJobs = map[string]*newscheduler.Job{}
 
 // To test out firestore before potential merge
 var shuffleTestProject = "shuffle-test-258209"
@@ -384,29 +384,35 @@ 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 {
+	var err error
 	testSplit := strings.Split(frequency, "*")
 	cronJob := ""
+	newfrequency := 0
+
 	if len(testSplit) > 5 {
 		cronJob = frequency
 	} else {
-		newfrequency, err := strconv.Atoi(frequency)
+		newfrequency, err = strconv.Atoi(frequency)
 		if err != nil {
+			log.Printf("Failed to parse time: %s", err)
 			return err
 		}
 
-		_ = newfrequency
-
 		//if int(newfrequency) < 60 {
 		//	cronJob = fmt.Sprintf("*/%s * * * *")
 		//} else if int(newfrequency) <
-		log.Println("FIXME: SHOULD DO Frequency (minutes) to CRON")
 	}
 
-	if len(cronJob) == 0 {
+	// Reverse. Can't handle CRON, only numbers
+	if len(cronJob) > 0 {
 		return errors.New("cronJob isn't formatted correctly")
 	}
 
-	log.Printf("CRON: %s, body: %s", cronJob, string(body))
+	if newfrequency < 1 {
+		return errors.New("Frequency has to be more than 0")
+	}
+
+	//log.Printf("CRON: %s, body: %s", cronJob, string(body))
 
 	// FIXME:
 	// This may run multiple places if multiple servers,
@@ -418,49 +424,42 @@ func createSchedule(ctx context.Context, scheduleId, workflowId, name, frequency
 		}
 
 		_, _, err := handleExecution(workflowId, Workflow{}, request)
-		if err == nil {
+		if err != nil {
 			log.Printf("Failed to execute: %s", err)
 		}
 	}
 
-	// FIXME - Create a real schedule based on cron:
+	log.Printf("Starting frequency: %d", newfrequency)
-	// 1. Parse the cron in a function to match this schedule
+	jobret, err := newscheduler.Every(newfrequency).Seconds().NotImmediately().Run(job)
-	// 2. Make main init check for schedules that aren't running
-	_, err := newscheduler.Every(5).Seconds().NotImmediately().Run(job)
 	if err != nil {
 		log.Printf("Failed to schedule workflow: %s", err)
 		return err
 	}
 
-	return errors.New("ERROR!!")
+	//scheduledJobs = append(scheduledJobs, jobret)
+	scheduledJobs[scheduleId] = jobret
 
-	//log.Printf("REQUEST: %#v", executionRequest)
+	// Doesn't need running/not running. If stopped, we just delete it.
+	timeNow := int64(time.Now().Unix())
+	schedule := ScheduleOld{
+		Id:                   scheduleId,
+		WorkflowId:           workflowId,
+		Argument:             string(body),
+		Seconds:              newfrequency,
+		CreationTime:         timeNow,
+		LastModificationtime: timeNow,
+		LastRuntime:          timeNow,
+	}
 
-	//req := &schedulerpb.CreateJobRequest{
+	err = setSchedule(ctx, schedule)
-	//	Parent: fmt.Sprintf("projects/%s/locations/europe-west2", gceProject),
+	if err != nil {
-	//	Job: &schedulerpb.Job{
+		log.Printf("Failed to set schedule: %s", err)
-	//		Name:        fmt.Sprintf("projects/%s/locations/europe-west2/jobs/schedule_%s", gceProject, scheduleId),
+		return err
-	//		Schedule:    cronJob,
+	}
-	//		Description: name,
+
-	//		Target: &schedulerpb.Job_HttpTarget{
+	// FIXME - Create a real schedule based on cron:
-	//			HttpTarget: &schedulerpb.HttpTarget{
+	// 1. Parse the cron in a function to match this schedule
-	//				Uri:        fmt.Sprintf("https://shuffler.io/api/v1/workflows/%s/execute", workflowId),
+	// 2. Make main init check for schedules that aren't running
-	//				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
 }
@@ -1692,11 +1691,12 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf
 			return WorkflowExecution{}, "Failed getting body", err
 		}
 
+		// This one doesn't really matter.
 		var execution ExecutionRequest
 		err = json.Unmarshal(body, &execution)
 		if err != nil {
-			log.Printf("Failed execution POST unmarshaling: %s", err)
+			//log.Printf("Failed execution POST unmarshaling - still continue: %s", err)
-			return WorkflowExecution{}, "", err
+			//return WorkflowExecution{}, "", err
 		}
 
 		if execution.Start == "" && len(body) > 0 {
@@ -1708,7 +1708,7 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf
 			workflowExecution.ExecutionArgument = execution.ExecutionArgument
 		}
 
-		log.Printf("Execution data: %#v", execution)
+		//log.Printf("Execution data: %#v", execution)
 		if len(execution.Start) == 36 {
 			log.Printf("SHOULD START ON NODE %s", execution.Start)
 			workflow.Start = execution.Start
@@ -1916,7 +1916,7 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf
 				executionRequestWrapper.Data = append(executionRequestWrapper.Data, executionRequest)
 			}
 
-			log.Printf("Execution request: %#v", executionRequest)
+			//log.Printf("Execution request: %#v", executionRequest)
 
 			err = setWorkflowQueue(ctx, executionRequestWrapper, environment)
 			if err != nil {
@@ -2086,7 +2086,117 @@ func stopSchedule(resp http.ResponseWriter, request *http.Request) {
 	return
 }
 
+func stopScheduleGCP(resp http.ResponseWriter, request *http.Request) {
+	cors := handleCors(resp, request)
+	if cors {
+		return
+	}
+
+	user, err := handleApiAuthentication(resp, request)
+	if err != nil {
+		log.Printf("Api authentication failed in schedule workflow: %s", err)
+		resp.WriteHeader(401)
+		resp.Write([]byte(`{"success": false}`))
+		return
+	}
+
+	location := strings.Split(request.URL.String(), "/")
+
+	var fileId string
+	var scheduleId string
+	if location[1] == "api" {
+		if len(location) <= 6 {
+			resp.WriteHeader(401)
+			resp.Write([]byte(`{"success": false}`))
+			return
+		}
+
+		fileId = location[4]
+		scheduleId = location[6]
+	}
+
+	if len(fileId) != 36 {
+		resp.WriteHeader(401)
+		resp.Write([]byte(`{"success": false, "reason": "Workflow ID to stop schedule is not valid"}`))
+		return
+	}
+
+	if len(scheduleId) != 36 {
+		resp.WriteHeader(401)
+		resp.Write([]byte(`{"success": false, "reason": "Schedule ID not valid"}`))
+		return
+	}
+
+	ctx := context.Background()
+	workflow, err := getWorkflow(ctx, fileId)
+	if err != nil {
+		log.Printf("Failed getting the workflow locally: %s", err)
+		resp.WriteHeader(401)
+		resp.Write([]byte(`{"success": false}`))
+		return
+	}
+
+	// FIXME - have a check for org etc too..
+	// FIXME - admin check like this? idk
+	if user.Id != workflow.Owner && user.Role != "admin" && user.Role != "scheduler" {
+		log.Printf("Wrong user (%s) for workflow %s (stop schedule)", user.Username, workflow.ID)
+		resp.WriteHeader(401)
+		resp.Write([]byte(`{"success": false}`))
+		return
+	}
+
+	if len(workflow.Actions) == 0 {
+		workflow.Actions = []Action{}
+	}
+	if len(workflow.Branches) == 0 {
+		workflow.Branches = []Branch{}
+	}
+	if len(workflow.Triggers) == 0 {
+		workflow.Triggers = []Trigger{}
+	}
+	if len(workflow.Errors) == 0 {
+		workflow.Errors = []string{}
+	}
+
+	err = deleteSchedule(ctx, scheduleId)
+	if err != nil {
+		if strings.Contains(err.Error(), "Job not found") {
+			resp.WriteHeader(200)
+			resp.Write([]byte(fmt.Sprintf(`{"success": true}`)))
+		} else {
+			resp.WriteHeader(401)
+			resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed stopping schedule"}`)))
+		}
+		return
+	}
+
+	resp.WriteHeader(200)
+	resp.Write([]byte(fmt.Sprintf(`{"success": true}`)))
+	return
+}
+
 func deleteSchedule(ctx context.Context, id string) error {
+	log.Printf("Should stop schedule %s!", id)
+	//newscheduler "github.com/carlescere/scheduler"
+	log.Printf("Schedules: %#v", scheduledJobs)
+	if value, exists := scheduledJobs[id]; exists {
+		log.Printf("STOP THIS ONE: %s", value)
+		// Looks like this does the trick? Hurr
+		value.Lock()
+		err := DeleteKey(ctx, "schedules", id)
+		if err != nil {
+			log.Printf("Failed to delete schedule: %s", err)
+			return err
+		}
+	} else {
+		// FIXME - allow it to kind of stop anyway?
+		return errors.New("Can't find the schedule.")
+	}
+
+	return nil
+}
+
+func deleteScheduleGCP(ctx context.Context, id string) error {
 	c, err := scheduler.NewCloudSchedulerClient(ctx)
 	if err != nil {
 		log.Printf("%s", err)
@@ -2208,13 +2318,7 @@ func scheduleWorkflow(resp http.ResponseWriter, request *http.Request) {
 		return
 	}
 
-	type tmp struct {
+	scheduleArg, err := json.Marshal(schedule.ExecutionArgument)
-		ExecutionArgument string `json:"execution_argument"`
-	}
-
-	var tmpArg tmp
-	tmpArg.ExecutionArgument = schedule.ExecutionArgument
-	scheduleArg, err := json.Marshal(tmpArg)
 	if err != nil {
 		log.Printf("Failed scheduleArg marshal: %s", err)
 		resp.WriteHeader(http.StatusInternalServerError)
@@ -2222,6 +2326,8 @@ func scheduleWorkflow(resp http.ResponseWriter, request *http.Request) {
 		return
 	}
 
+	log.Printf("Schedulearg: %s", string(scheduleArg))
+
 	err = createSchedule(
 		ctx,
 		schedule.Id,
@@ -3315,7 +3421,7 @@ func getWorkflowExecutions(resp http.ResponseWriter, request *http.Request) {
 	}
 
 	// Query for the specifci workflowId
-	q := datastore.NewQuery("workflowexecution").Filter("workflow_id =", fileId)
+	q := datastore.NewQuery("workflowexecution").Filter("workflow_id =", fileId).Limit(50)
 	var workflowExecutions []WorkflowExecution
 	_, err = dbclient.GetAll(ctx, q, &workflowExecutions)
 	if err != nil {
@@ -3342,6 +3448,18 @@ func getWorkflowExecutions(resp http.ResponseWriter, request *http.Request) {
 	resp.Write(newjson)
 }
 
+func getAllSchedules(ctx context.Context) ([]ScheduleOld, error) {
+	var schedules []ScheduleOld
+	q := datastore.NewQuery("schedules")
+
+	_, err := dbclient.GetAll(ctx, q, &schedules)
+	if err != nil {
+		return []ScheduleOld{}, err
+	}
+
+	return schedules, nil
+}
+
 func getAllWorkflowApps(ctx context.Context) ([]WorkflowApp, error) {
 	var allworkflowapps []WorkflowApp
 	q := datastore.NewQuery("workflowapp")

frontend/src/AngularWorkflow.js

@@ -80,7 +80,6 @@ const splitter = "|~|"
 //const referenceUrl = "https://shuffler.io/functions/webhooks/"
 //const referenceUrl = window.location.origin+"/api/v1/hooks/"
 
-console.log(window.location)
 const AngularWorkflow = (props) => {
   const { globalUrl, isLoggedIn, isLoaded } = props;
 	const referenceUrl = globalUrl+"/api/v1/hooks/"
@@ -1261,11 +1260,11 @@ const AngularWorkflow = (props) => {
 
 	//const submitSchedule = (id, name, frequency, executionArg) => {
 	const submitSchedule = (trigger, triggerindex) => {
-		const cronSplit = workflow.triggers[triggerindex].parameters[0].value.split("*")
+		//const cronSplit = workflow.triggers[triggerindex].parameters[0].value.split("*")
-		if (cronSplit.length <= 5 || cronSplit.length > 6) {
+		//if (cronSplit.length <= 5 || cronSplit.length > 6) {
-			alert.error("Error: Bad cron, example run every 1 minute: */1 * * * *")
+		//	alert.error("Error: Bad cron, example run every 1 minute: */1 * * * *")
-			return
+		//	return
-		}
+		//}
 
 		if (trigger.name.length <= 0) {
 			alert.error("Error: name can't be empty")
@@ -3892,7 +3891,7 @@ const AngularWorkflow = (props) => {
 		if (Object.getOwnPropertyNames(selectedTrigger).length > 0 && workflow.triggers[selectedTriggerIndex] !== undefined) {
 			if (workflow.triggers[selectedTriggerIndex].parameters === undefined || workflow.triggers[selectedTriggerIndex].parameters === null || workflow.triggers[selectedTriggerIndex].parameters.length === 0) {
 				workflow.triggers[selectedTriggerIndex].parameters = []
-				workflow.triggers[selectedTriggerIndex].parameters[0] = {"name": "cron", "value": "*/15 * * * *"}
+				workflow.triggers[selectedTriggerIndex].parameters[0] = {"name": "cron", "value": "120"}
 				workflow.triggers[selectedTriggerIndex].parameters[1] = {"name": "execution_argument", "value": '{"example": {"json": "is cool"}}'}
 				setWorkflow(workflow)
 			}
@@ -3953,7 +3952,7 @@ const AngularWorkflow = (props) => {
 							<div style={{marginTop: "20px", marginBottom: "7px", display: "flex"}}>
 								<div style={{width: "17px", height: "17px", borderRadius: 17 / 2, backgroundColor: "#f85a3e", marginRight: "10px"}}/>
 								<div style={{flex: "10"}}> 
-									<b>Cron: </b> 
+									<b>Run how often (seconds)? </b> 
 								</div>
 							</div>
 							<TextField
@@ -3968,6 +3967,7 @@ const AngularWorkflow = (props) => {
 									},
 								}}
 								fullWidth
+								disabled={workflow.triggers[selectedTriggerIndex].status === "running"}
 								defaultValue={workflow.triggers[selectedTriggerIndex].parameters[0].value}
 								color="primary"
 								placeholder="defaultValue"
@@ -3992,6 +3992,7 @@ const AngularWorkflow = (props) => {
 										fontSize: "1em",
 									},
 								}}
+								disabled={workflow.triggers[selectedTriggerIndex].status === "running"}
 								fullWidth
 								rows="6"
 								multiline

frontend/src/App.js

@@ -38,12 +38,12 @@ import AlertTemplate from "react-alert-template-basic";
 import { positions, Provider } from "react-alert";
 
 // Testing - localhost
-//const globalUrl = "http://192.168.3.6:5001"
+const globalUrl = "http://192.168.3.6:5001"
 console.log("HOST: ", process.env)
 
 
 // Production - backend proxy forwarding in nginx
-const globalUrl = window.location.origin
+//const globalUrl = window.location.origin
 
 const surfaceColor = "#27292D"
 const inputColor = "#383B40"

frontend/src/Workflows.js

@@ -333,18 +333,19 @@ const Workflows = (props) => {
 		}
 
 		return (
-			<Paper square style={paperAppStyle} onClick={() => {
+			<Paper square style={paperAppStyle} onClick={(e) => {
-				if (selectedWorkflow.id !== data.id) {
-					setSelectedWorkflow(data)
-					getWorkflowExecution(data.id)
-				}
 			}}>
 				<div style={{marginLeft: "10px", marginTop: "5px", marginBottom: "5px", width: boxWidth, backgroundColor: boxColor}}>
 				</div>
 				<Grid container style={{margin: "10px 10px 10px 10px", flex: "1"}}>
 					<Grid style={{display: "flex", flexDirection: "column", width: "100%"}}>
 						<Grid item style={{flex: "1", display: "flex"}}>
-							<div style={{flex: "10"}}>
+							<div style={{flex: "10",}} onClick={() => {
+								if (selectedWorkflow.id !== data.id) {
+									setSelectedWorkflow(data)
+									getWorkflowExecution(data.id)
+								}
+							}}>
 								<h3 style={{marginBottom: "0px", marginTop: "10px"}}>{data.name}</h3>
 							</div>
 							<div style={{flex: "1"}}>
@@ -390,17 +391,22 @@ const Workflows = (props) => {
       							</Menu>
 							</div>
 						</Grid>
-						<div style={{display: "flex", flex: "1"}}>
+						<div style={{display: "flex", flex: "1"}} onClick={() => {
+							if (selectedWorkflow.id !== data.id) {
+								setSelectedWorkflow(data)
+								getWorkflowExecution(data.id)
+							}
+						}}>
 							<Grid item style={{flex: "1", justifyContent: "center"}}>
 								<a href={"/workflows/"+data.id}>
 									<Tooltip color="primary" title="Edit workflow" placement="bottom">
-										<Button style={{}} color="primary" variant="outlined" style={{marginRight: 10}} onClick={() => {}}>
+										<Button style={{}} color="primary" variant="text" style={{marginRight: 10}} onClick={() => {}}>
 											<EditIcon style={{marginRight: 10}}/> Edit
 										</Button> 				
 									</Tooltip>
 								</a>
 								<Tooltip color="primary" title="Execute workflow" placement="bottom">
-									<Button style={{}} color="primary" variant="outlined" onClick={() => executeWorkflow(data.id)}>
+									<Button style={{}} color="secondary" variant="text" onClick={() => executeWorkflow(data.id)}>
 										<PlayArrowIcon />
 									</Button> 				
 								</Tooltip>
@@ -840,7 +846,7 @@ const Workflows = (props) => {
 			<div style={{flex: viewSize.executionsView, marginLeft: "10px", marginRight: "10px"}}>
 				<div style={{display: "flex"}}>
 					<div style={{flex: "10"}}>
-						<h2>Executions</h2> 
+						<h2>Executions: {selectedWorkflow.name}</h2> 
 					</div>
 					<div style={{flex: "1"}}>
 						<Button color="primary" style={{marginTop: "20px"}} variant="outlined" onClick={() => {

functions/onprem/orborus/orborus.go

@@ -261,7 +261,9 @@ func main() {
 		// New, abortable version. Should check executionid and remove everything else
 		var toBeRemoved ExecutionRequestWrapper
 		for _, execution := range executionRequests.Data {
-			log.Printf("Argument: %#v", execution.ExecutionArgument)
+			if len(execution.ExecutionArgument) > 0 {
+				log.Printf("Argument: %#v", execution.ExecutionArgument)
+			}
 
 			if execution.Type == "schedule" {
 				log.Printf("SOMETHING ELSE :O: %s", execution.Type)