diff --git a/.env b/.env
index 7e073bd9..5a78c4d8 100644
--- a/.env
+++ b/.env
@@ -30,6 +30,7 @@ SHUFFLE_FILE_LOCATION=./shuffle-files
SHUFFLE_ENCRYPTION_MODIFIER=
# Other configs
+BASE_URL=http://shuffle-backend:5001
BACKEND_HOSTNAME=shuffle-backend
BACKEND_PORT=5001
FRONTEND_PORT=3001
diff --git a/backend/app_sdk/app_base.py b/backend/app_sdk/app_base.py
index b725ae9f..7e238e76 100644
--- a/backend/app_sdk/app_base.py
+++ b/backend/app_sdk/app_base.py
@@ -4,16 +4,18 @@ import sys
import re
import time
import json
+import liquid
import logging
-import requests
-import urllib.parse
-import http.client
import urllib3
import hashlib
-from liquid import Liquid
-import liquid
import zipfile
+import requests
+import http.client
+import urllib.parse
from io import BytesIO
+from liquid import Liquid
+
+runtime = os.getenv("SHUFFLE_SWARM_CONFIG", "")
class AppBase:
__version__ = None
@@ -27,7 +29,7 @@ class AppBase:
# apikey is for the user / org
# authorization is for the specific workflow
- self.url = os.getenv("CALLBACK_URL", "https://shuffler.io")
+ self.url = os.getenv("CALLBACK_URL", "https://shuffler.io")
self.base_url = os.getenv("BASE_URL", "https://shuffler.io")
self.action = os.getenv("ACTION", "")
self.original_action = os.getenv("ACTION", "")
@@ -51,8 +53,10 @@ class AppBase:
try:
self.action = json.loads(self.action)
self.original_action = json.loads(self.action)
- except:
- self.logger.info("[WARNING] Failed parsing action as JSON")
+ except Exception as e:
+ self.logger.info(f"[WARNING] Failed parsing action as JSON (init): {e}. NOT important if running apps with webserver")
+
+ #print(f"ACTION: {self.action}")
if len(self.base_url) == 0:
self.base_url = self.url
@@ -80,14 +84,14 @@ class AppBase:
# I wonder if this actually works
self.logger.info(f"[DEBUG] Before last stream result")
url = "%s%s" % (self.base_url, stream_path)
- #self.logger.info("[INFO] URL (URL): %s" % url)
+ self.logger.info("[INFO] URL FOR RESULT (URL): %s" % url)
try:
ret = requests.post(url, headers=headers, json=action_result)
#self.logger.info(f"[DEBUG] Result: {ret.status_code}")
#if ret.status_code != 200:
# self.logger.info(f"[DEBUG] Shuffle Response: {ret.text}")
- self.logger.info(f"[DEBUG] Successful request: Status= {ret.status_code} & Response= {ret.text}")
+ self.logger.info(f"""[DEBUG] Successful request result request: Status= {ret.status_code} & Response= {ret.text}. Action status: {action_result["status"]}""")
except requests.exceptions.ConnectionError as e:
self.logger.info(f"[DEBUG] Unexpected ConnectionError happened: {e}")
return
@@ -520,8 +524,10 @@ class AppBase:
}
self.send_result(self.action_result, {"Content-Type": "application/json", "Authorization": "Bearer %s" % self.authorization}, "/api/v1/streams")
- exit()
- #return
+ if runtime != "run":
+ exit()
+ else:
+ return
else:
#subparams = new_params
#self.logger.info(f"NEW PARAMS: {new_params}")
@@ -849,8 +855,8 @@ class AppBase:
self.logger.info("IDS TO RETURN: %s" % file_ids)
return file_ids
- async def execute_action(self, action):
-
+ #async def execute_action(self, action):
+ def execute_action(self, action):
# !!! Let this line stay - its used for some horrible codegeneration / stitching !!! #
#STARTCOPY
stream_path = "/api/v1/streams"
@@ -880,19 +886,19 @@ class AppBase:
}
if len(self.action) == 0:
- self.logger.info("ACTION env not defined")
+ self.logger.info("[WARNING] ACTION env not defined")
self.action_result["result"] = "Error in setup ENV: ACTION not defined"
self.send_result(self.action_result, headers, stream_path)
return
if len(self.authorization) == 0:
- self.logger.info("AUTHORIZATION env not defined")
+ self.logger.info("[WARING] AUTHORIZATION env not defined")
self.action_result["result"] = "Error in setup ENV: AUTHORIZATION not defined"
self.send_result(self.action_result, headers, stream_path)
return
if len(self.current_execution_id) == 0:
- self.logger.info("EXECUTIONID env not defined")
+ self.logger.info("[WARNING] EXECUTIONID env not defined")
self.action_result["result"] = "Error in setup ENV: EXECUTIONID not defined"
self.send_result(self.action_result, headers, stream_path)
return
@@ -918,7 +924,7 @@ class AppBase:
# Verify whether there are any parameters with ACTION_RESULT required
# If found, we get the full results list from backend
fullexecution = {}
- if len(self.full_execution) == 0:
+ if isinstance(self.full_execution, str) and len(self.full_execution) == 0:
self.logger.info("[DEBUG] NO EXECUTION - LOADING!")
try:
tmpdata = {
@@ -937,8 +943,8 @@ class AppBase:
fullexecution = ret.json()
else:
try:
- self.logger.info("Error: Data: ", ret.json())
- self.logger.info("Error with status code for results. Crashing because ACTION_RESULTS or WORKFLOW_VARIABLE can't be handled. Status: %d" % ret.status_code)
+ self.logger.info("[DEBUG] Error: Data: ", ret.json())
+ self.logger.info("[DEBUG] Error with status code for results. Crashing because ACTION_RESULTS or WORKFLOW_VARIABLE can't be handled. Status: %d" % ret.status_code)
except json.decoder.JSONDecodeError:
pass
@@ -946,11 +952,12 @@ class AppBase:
self.send_result(self.action_result, headers, stream_path)
return
except requests.exceptions.ConnectionError as e:
- self.logger.info("Connectionerror: %s" % e)
+ self.logger.info("[DEBUG] FullExec Connectionerror: %s" % e)
self.action_result["result"] = "Connection error during startup: %s" % e
self.send_result(self.action_result, headers, stream_path)
return
else:
+ self.logger.info(f"[DEBUG] Setting execution to default value with type {type(self.full_execution)}")
try:
fullexecution = json.loads(self.full_execution)
except json.decoder.JSONDecodeError as e:
@@ -1147,6 +1154,11 @@ class AppBase:
except TypeError:
return data, False
+ # Because liquid can handle ALL of this now.
+ # Implemented for >0.9.25
+ self.logger.info("[DEBUG] Skipping parser because use of its been deprecated >0.9.25 due to Liquid implementation")
+ return data, False
+
wrappers = ["int", "number", "lower", "upper", "trim", "strip", "split", "parse", "len", "length", "lenght", "join", "replace"]
if not any(wrapper in data for wrapper in wrappers):
@@ -1219,7 +1231,7 @@ class AppBase:
if isinstance(data, str) and len(data) > 4:
if (data[0] == "{" or data[0] == "[") and (data[len(data)-1] == "]" or data[len(data)-1] == "}"):
- self.logger.info("Skipping parser because use of {[ and ]}")
+ self.logger.info("[DEBUG] Skipping parser because use of {[ and ]}")
return data
newdata = []
@@ -1587,10 +1599,11 @@ class AppBase:
# Sending self as it's not a normal function
def parse_liquid(template, self):
-
- #self.logger.info("Inside liquid with glob: %s" % globals())
+
+ errors = False
+ error_msg = ""
try:
- if len(template) > 5000000:
+ if len(template) > 10000000:
self.logger.info("[DEBUG] Skipping liquid - size too big (%d)" % len(template))
return template
@@ -1609,21 +1622,41 @@ class AppBase:
# Can't handle self yet (?)
ret = run.render(**globals())
return ret
- #try:
- #run = Liquid(template)
- #return ret
- #except liquid.exceptions.LiquidSyntaxError as e:
- # run = Liquid(template, {'mode': 'python'})
- # ret = run.render(**globals())
- # return ret
- #except liquid.exceptions.LiquidRenderError as e:
- # self.logger.info("Render error: %s" % e)
except jinja2.exceptions.TemplateNotFound as e:
- self.logger.info("[ERROR] Template error: %s" % e)
+ self.logger.info(f"[ERROR] Liquid Template error: {e}")
+ error = True
+ error_msg = e
except jinja2.exceptions.TemplateSyntaxError as e:
- self.logger.info("[ERROR] Syntax error: %s" % e)
- except:
- self.logger.info("[ERROR] General exception for liquid")
+ self.logger.info(f"[ERROR] Liquid Syntax error: {e}")
+ error = True
+ error_msg = e
+ except Exception as e:
+ self.logger.info(f"[ERROR] General exception for liquid: {e}")
+ error = True
+ error_msg = e
+
+ if error == True:
+ self.action_result["status"] = "FAILURE"
+ data = {
+ "success": False,
+ "input": template,
+ "reason": f"Failed to parse LiquidPy: {error_msg}",
+ }
+ try:
+ self.action_result["result"] = json.dumps(data)
+ except Exception as e:
+ self.action_result["result"] = f"Failed to parse LiquidPy: {error_msg}"
+ print("[WARNING] Failed to set LiquidPy result")
+
+ self.action_result["completed_at"] = int(time.time())
+ self.send_result(self.action_result, headers, stream_path)
+
+ self.logger.info(f"[ERROR] Sent FAILURE response to backend due to : {e}")
+
+ if runtime == "run":
+ return template
+ else:
+ os.exit()
return template
@@ -1983,11 +2016,11 @@ class AppBase:
except KeyError:
continue
- self.logger.info("Relevant conditions: %s" % branch["conditions"])
+ self.logger.info("[DEBUG] Relevant conditions: %s" % branch["conditions"])
successful_conditions = []
failed_conditions = []
for condition in branch["conditions"]:
- self.logger.info("Getting condition value of %s" % condition)
+ self.logger.info("[DEBUG] Getting condition value of %s" % condition)
# Parse all values first here
sourcevalue = condition["source"]["value"]
@@ -2083,7 +2116,7 @@ class AppBase:
if " " in actionname:
actionname.replace(" ", "_", -1)
-
+ #print(action)
#if action.generated:
# actionname = actionname.lower()
@@ -2091,13 +2124,14 @@ class AppBase:
try:
func = getattr(self, actionname, None)
if func == None:
- self.logger.debug(f"Failed executing {actionname} because func is None.")
+ self.logger.debug(f"[DEBUG] Failed executing {actionname} because func is None.")
self.action_result["status"] = "FAILURE"
self.action_result["result"] = "Function %s doesn't exist." % actionname
elif callable(func):
try:
if len(action["parameters"]) < 1:
- result = await func()
+ #result = await func()
+ result = func()
else:
# Potentially parse JSON here
# FIXME - add potential authentication as first parameter(s) here
@@ -2438,7 +2472,7 @@ class AppBase:
# This part has fucked over so many random JSON usages because of weird paranthesis parsing
value = parse_wrapper_start(value, self)
- self.logger.info("[DEBUG] Post return: %s" % value)
+ #self.logger.info("[DEBUG] Post return: %s" % value)
#self.logger.info("POST data value: %s" % value)
params[parameter["name"]] = value
@@ -2529,11 +2563,12 @@ class AppBase:
#newres = ""
while True:
try:
- newres = await func(**params)
+ #newres = await func(**params)
+ newres = func(**params)
break
except TypeError as e:
newres = ""
- self.logger.info(f"[DEBUG] Got exec error: {errorstring}")
+ self.logger.info(f"[DEBUG] Got exec error: {e}")
errorstring = f"{e}"
if "got an unexpected keyword argument" in errorstring:
fieldsplit = errorstring.split("'")
@@ -2604,7 +2639,8 @@ class AppBase:
self.logger.info("[INFO] Running WITHOUT outer loop (looping)")
json_object = False
- results = await self.run_recursed_items(func, multi_parameters, {})
+ #results = await self.run_recursed_items(func, multi_parameters, {})
+ results = self.run_recursed_items(func, multi_parameters, {})
if isinstance(results, dict) or isinstance(results, list):
json_object = True
@@ -2799,34 +2835,141 @@ class AppBase:
logger = logging.getLogger(f"{cls.__name__}")
logger.setLevel(logging.DEBUG)
- #self.logger.info("Started execution: %s!!" % cls)
- #self.logger.info("Action: %s" % action)
- #if isinstance(cls, object):
- # self.action = cls
- app = cls(redis=None, logger=logger, console_logger=logger)
- if isinstance(action, str):
- print("[DEBUG] Normal execution. Action is a string.")
- elif isinstance(action, object):
- print("[DEBUG] OBJECT execution. Action is NOT a string.")
- app.action = action
+ ##############################################
- try:
- app.authorization = action["authorization"]
- app.current_execution_id = action["execution_id"]
- except:
- pass
+ exposed_port = os.getenv("SHUFFLE_APP_EXPOSED_PORT", "")
+ logger.info(f"[DEBUG] \"{runtime}\" - run indicates microservices. Port: \"{exposed_port}\"")
+ if runtime == "run" and exposed_port != "":
+ # Base port is 33334. Exposed port may differ based on discovery from Worker
+ port = int(exposed_port)
+ logger.info(f"[DEBUG] Starting webserver on port {port} (same as exposed port)")
+ from flask import Flask, request
+ from waitress import serve
+ import asyncio
+
+ flask_app = Flask(__name__)
+
+ @flask_app.route("/api/v1/run", methods=["POST"])
+ #async def execute():
+ def execute():
+ if request.method == "POST":
+ #print(request.get_json(force=True))
+ #print("DATA: ", request.data)
+ requestdata = {}
+ try:
+ requestdata = json.loads(request.data)
+ except Exception as e:
+ return {
+ "success": False,
+ "reason": f"Invalid Action data {e}",
+ }
+
+ #logger.info(f"[DEBUG] Datatype: {type(requestdata)}: {requestdata}")
- try:
- app.url = action["url"]
- except:
- pass
+ # Remaking class for each request
+ #print(f"APP: {app}")
+
+ app = cls(redis=None, logger=logger, console_logger=logger)
+ try:
+ #asyncio.run(AppBase.run(action=requestdata), debug=True)
+ #value = json.dumps(value)
+ try:
+ app.full_execution = json.dumps(requestdata["workflow_execution"])
+ except Exception as e:
+ logger.info(f"Failed parsing full execution from workflow_execution: {e}")
+ try:
+ app.action = requestdata["action"]
+ except:
+ logger.info("Failed parsing action")
- try:
- app.base_url = action["base_url"]
- except:
- pass
+ try:
+ app.authorization = requestdata["authorization"]
+ app.current_execution_id = requestdata["execution_id"]
+ except:
+ logger.info("Failed parsing auth and exec id")
+
+ # BASE URL (backend)
+ try:
+ app.url = requestdata["url"]
+ logger.info(f"BACKEND URL: {app.url}")
+ except:
+ logger.info("Failed parsing url")
+
+ # URL (worker)
+ try:
+ app.base_url = requestdata["base_url"]
+ logger.info(f"WORKER URL: {app.base_url}")
+ except:
+ logger.info("Failed parsing base url")
+
+ #await
+ app.execute_action(app.action)
+ logger.info("\n\n[DEBUG] Done awaiting app action running\n\n")
+ except Exception as e:
+ return {
+ "success": False,
+ "reason": f"Problem in execution {e}",
+ }
+
+ return {
+ "success": True,
+ "reason": "App successfully finished",
+ }
+ else:
+ return {
+ "success": False,
+ "reason": f"HTTP method {request.method} not allowed",
+ }
+
+ logger.info(f"[DEBUG] Serving on port {port}")
+ serve(
+ flask_app,
+ host="0.0.0.0",
+ port=port,
+ threads=8,
+ channel_timeout=30,
+ expose_tracebacks=True,
+ asyncore_use_poll=True,
+ )
+ #######################
else:
- self.logger.info("ACTION TYPE (unhandled): %s" % type(action))
+ # Has to start like this due to imports in other apps
+ # Move it outside everything?
+ app = cls(redis=None, logger=logger, console_logger=logger)
+ logger.info(f"[DEBUG] Action: {action}")
+
+ if isinstance(action, str):
+ logger.info("[DEBUG] Normal execution (env var). Action is a string.")
+ elif isinstance(action, object):
+ logger.info("[DEBUG] OBJECT execution (cloud). Action is NOT a string.")
+ app.action = action
- await app.execute_action(app.action)
+ try:
+ app.authorization = action["authorization"]
+ app.current_execution_id = action["execution_id"]
+ except:
+ pass
+
+ # BASE URL (worker)
+ try:
+ app.url = action["url"]
+ except:
+ pass
+
+ # Callback URL (backend)
+ try:
+ app.base_url = action["base_url"]
+ except:
+ pass
+ else:
+ self.logger.info("ACTION TYPE (unhandled): %s" % type(action))
+
+ #await app.execute_action(app.action)
+ app.execute_action(app.action)
+
+ #app.run(host="0.0.0.0", port=33334)
+
+if __name__ == "__main__":
+ import asyncio
+ asyncio.run(AppBase.run(), debug=True)
diff --git a/backend/app_sdk/build.sh b/backend/app_sdk/build.sh
index bc44f74b..072777b4 100644
--- a/backend/app_sdk/build.sh
+++ b/backend/app_sdk/build.sh
@@ -3,7 +3,7 @@
### DEFAULT
NAME=shuffle-app_sdk
-VERSION=0.9.25
+VERSION=0.9.30
docker rmi docker.pkg.github.com/frikky/shuffle/$NAME:$VERSION --force
docker build . -f Dockerfile -t frikky/shuffle:app_sdk -t frikky/$NAME:$VERSION -t docker.pkg.github.com/frikky/shuffle/$NAME:$VERSION -t ghcr.io/frikky/$NAME:$VERSION -t ghcr.io/frikky/$NAME:nightly
diff --git a/backend/app_sdk/requirements.txt b/backend/app_sdk/requirements.txt
index 79cf5ea2..3cba831d 100644
--- a/backend/app_sdk/requirements.txt
+++ b/backend/app_sdk/requirements.txt
@@ -1,4 +1,7 @@
urllib3==1.26.5
requests==2.25.1
MarkupSafe==2.0.1
-liquidpy==0.7.1
+liquidpy==0.7.2
+flask[async]==2.0.2
+waitress==2.0.0
+#flask==1.1.2
diff --git a/backend/go-app/go.mod b/backend/go-app/go.mod
index 16f22e0f..e76a9b50 100644
--- a/backend/go-app/go.mod
+++ b/backend/go-app/go.mod
@@ -1,50 +1,32 @@
-module shuffle
+module main
-go 1.13
-
-replace github.com/shuffle/shuffle-shared => ../../../../git/shuffle-shared
+go 1.15
+//replace github.com/shuffle/shuffle-shared => ../../../../git/shuffle-shared
//replace github.com/frikky/kin-openapi => ../../../../git/kin-openapi
+//replace github.com/frikky/go-elasticsearch => ../../../../git/go-elasticsearch
require (
- cloud.google.com/go v0.75.0
- cloud.google.com/go/datastore v1.4.0
- cloud.google.com/go/pubsub v1.3.1
- cloud.google.com/go/storage v1.12.0
- github.com/Masterminds/semver v1.5.0 // indirect
- github.com/RobotsAndPencils/go-saml v0.0.0-20170520135329-fb13cb52a46b // indirect
- github.com/algolia/algoliasearch-client-go/v3 v3.18.1 // indirect
+ cloud.google.com/go/datastore v1.6.0
+ cloud.google.com/go/pubsub v1.17.0
+ cloud.google.com/go/storage v1.18.1
github.com/basgys/goxml2json v1.1.0
- github.com/bradfitz/slice v0.0.0-20180809154707-2b758aa73013 // indirect
github.com/carlescere/scheduler v0.0.0-20170109141437-ee74d2f83d82
- github.com/docker/distribution v2.7.1+incompatible // indirect
- github.com/docker/docker v20.10.3-0.20210216175712-646072ed6524+incompatible
- github.com/docker/go-connections v0.4.0
- github.com/docker/go-units v0.4.0 // indirect
- github.com/elastic/go-elasticsearch/v7 v7.13.1 // indirect
- github.com/frikky/kin-openapi v0.39.0
- github.com/frikky/shuffle-shared v0.1.15
- github.com/fsouza/go-dockerclient v1.7.2
+ github.com/docker/docker v20.10.9+incompatible
+ github.com/frikky/kin-openapi v0.40.0
+ github.com/fsouza/go-dockerclient v1.7.4
github.com/ghodss/yaml v1.0.0
- github.com/go-git/go-billy/v5 v5.0.0
- github.com/go-git/go-git/v5 v5.0.0
- github.com/google/go-github/v28 v28.1.1
- github.com/gorilla/handlers v1.4.2 // indirect
+ github.com/go-git/go-billy/v5 v5.3.1
+ github.com/go-git/go-git/v5 v5.4.2
github.com/gorilla/mux v1.8.0
- github.com/h2non/filetype v1.0.12
- github.com/kardianos/osext v0.0.0-20190222173326-2bc1f35cddc0 // indirect
- github.com/nu7hatch/gouuid v0.0.0-20131221200532-179d4d0c4d8d // indirect
- github.com/patrickmn/go-cache v2.1.0+incompatible
+ github.com/h2non/filetype v1.1.1
github.com/satori/go.uuid v1.2.0
- github.com/shuffle/shuffle-shared v0.1.15
+ github.com/shuffle/shuffle-shared v0.1.27
go4.org v0.0.0-20201209231011-d4a079459e60 // indirect
- golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9
- golang.org/x/oauth2 v0.0.0-20210113160501-8b1d76fa0423
- google.golang.org/api v0.36.0
+ golang.org/x/crypto v0.0.0-20210921155107-089bfa567519
+ google.golang.org/api v0.58.0
google.golang.org/appengine v1.6.7
- google.golang.org/genproto v0.0.0-20210113195801-ae06605f4595
- google.golang.org/grpc v1.34.1
+ google.golang.org/grpc v1.41.0
gopkg.in/src-d/go-git.v4 v4.13.1
- gopkg.in/yaml.v2 v2.4.0
gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b
)
diff --git a/backend/go-app/main.go b/backend/go-app/main.go
index cdece1bb..36ee3399 100644
--- a/backend/go-app/main.go
+++ b/backend/go-app/main.go
@@ -1955,9 +1955,18 @@ func handleWebhookCallback(resp http.ResponseWriter, request *http.Request) {
// 1. Get callback data
// 2. Load the configuration
// 3. Execute the workflow
- cors := shuffle.HandleCors(resp, request)
- if cors {
- return
+ //cors := shuffle.HandleCors(resp, request)
+ //if cors {
+ // return
+ //}
+
+ if request.Method != "POST" {
+ request.Method = "POST"
+ }
+
+ if request.Body == nil {
+ stringReader := strings.NewReader("")
+ request.Body = ioutil.NopCloser(stringReader)
}
path := strings.Split(request.URL.String(), "/")
@@ -3837,7 +3846,7 @@ func runInitEs(ctx context.Context) {
//}
} else {
- log.Printf("[DEBUG] There are %d org(s).", len(activeOrgs))
+ log.Printf("[DEBUG] Found %d org(s) in total.", len(activeOrgs))
if len(activeOrgs) == 1 {
if len(activeOrgs[0].Users) == 0 {
@@ -5639,6 +5648,7 @@ func initHandlers() {
log.Printf("[DEBUG] Initialized Shuffle database connection. Setting up environment.")
if elasticConfig == "elasticsearch" {
+ time.Sleep(5 * time.Second)
go runInitEs(ctx)
} else {
go runInit(ctx)
@@ -5732,7 +5742,7 @@ func initHandlers() {
// Triggers
r.HandleFunc("/api/v1/hooks/new", shuffle.HandleNewHook).Methods("POST", "OPTIONS")
- r.HandleFunc("/api/v1/hooks/{key}", handleWebhookCallback).Methods("POST", "OPTIONS")
+ r.HandleFunc("/api/v1/hooks/{key}", handleWebhookCallback).Methods("POST", "GET", "PATCH", "PUT", "DELETE", "OPTIONS")
r.HandleFunc("/api/v1/hooks/{key}/delete", shuffle.HandleDeleteHook).Methods("DELETE", "OPTIONS")
// OpenAPI configuration
diff --git a/backend/go-app/walkoff.go b/backend/go-app/walkoff.go
index 7078110e..2475e57a 100644
--- a/backend/go-app/walkoff.go
+++ b/backend/go-app/walkoff.go
@@ -569,7 +569,7 @@ func handleGetWorkflowqueueConfirm(resp http.ResponseWriter, request *http.Reque
// FIXME: Add authentication?
id := request.Header.Get("Org-Id")
if len(id) == 0 {
- log.Printf("No Org-Id header set - confirm")
+ log.Printf("[ERROR] No Org-Id header set - confirm")
resp.WriteHeader(401)
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Specify the org-id header."}`)))
return
@@ -579,7 +579,7 @@ func handleGetWorkflowqueueConfirm(resp http.ResponseWriter, request *http.Reque
ctx := context.Background()
executionRequests, err := shuffle.GetWorkflowQueue(ctx, id)
if err != nil {
- log.Printf("(1) Failed reading body for workflowqueue: %s", err)
+ log.Printf("[WARNING] (1) Failed reading body for workflowqueue: %s", err)
resp.WriteHeader(401)
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Entity parsing error - confirm"}`)))
return
@@ -676,6 +676,19 @@ func handleGetWorkflowqueue(resp http.ResponseWriter, request *http.Request) {
}
ctx := context.Background()
+ env, err := shuffle.GetEnvironment(ctx, id, "")
+ timeNow := time.Now().Unix()
+ if err == nil && len(env.Id) > 0 && len(env.Name) > 0 {
+ if time.Now().Unix() > env.Edited+60 {
+ env.RunningIp = request.RemoteAddr
+ env.Checkin = timeNow
+ err = shuffle.SetEnvironment(ctx, env)
+ if err != nil {
+ log.Printf("[WARNING] Failed updating environment: %s", err)
+ }
+ }
+ }
+
executionRequests, err := shuffle.GetWorkflowQueue(ctx, id)
if err != nil {
// Skipping as this comes up over and over
@@ -685,11 +698,48 @@ func handleGetWorkflowqueue(resp http.ResponseWriter, request *http.Request) {
return
}
+ // Checking and updating the environment related to the first execution
if len(executionRequests.Data) == 0 {
executionRequests.Data = []shuffle.ExecutionRequest{}
} else {
- //log.Printf("[INFO] Executionrequests (%s): %d", id, len(executionRequests.Data))
- //log.Printf("IDS: %#v", executionRequests.Data[0].ExecutionId)
+ //log.Printf("In workflowqueue with %d", len(executionRequests.Data))
+
+ // Try again :)
+ if len(env.Id) == 0 && len(env.Name) == 0 {
+ orgId := ""
+ for _, requestData := range executionRequests.Data {
+ execution, err := shuffle.GetWorkflowExecution(ctx, requestData.ExecutionId)
+ if err == nil {
+ if len(execution.ExecutionOrg) > 0 {
+ orgId = execution.ExecutionOrg
+ break
+ }
+ }
+ }
+
+ if len(orgId) > 0 {
+ env, err := shuffle.GetEnvironment(ctx, id, orgId)
+ if err != nil {
+ log.Printf("[WARNING] No env found matching %s - continuing without updating orborus anyway: %s", id, err)
+ //resp.WriteHeader(401)
+ //resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "No env found matching %s"}`, id)))
+ //return
+ } else {
+ if timeNow > env.Edited+60 {
+ env.RunningIp = request.RemoteAddr
+ env.Checkin = timeNow
+ err = shuffle.SetEnvironment(ctx, env)
+ if err != nil {
+ log.Printf("[WARNING] Failed updating environment: %s", err)
+ }
+ }
+ }
+ }
+ }
+
+ if len(executionRequests.Data) > 10 {
+ executionRequests.Data = executionRequests.Data[0:9]
+ }
}
newjson, err := json.Marshal(executionRequests)
@@ -849,7 +899,7 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
var trigger shuffle.Trigger
err = json.Unmarshal([]byte(actionResult.Result), &trigger)
if err != nil {
- log.Printf("Failed unmarshaling actionresult for user input: %s", err)
+ log.Printf("[WARNING] Failed unmarshaling actionresult for user input: %s", err)
resp.WriteHeader(401)
resp.Write([]byte(`{"success": false}`))
return
@@ -862,21 +912,21 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
err := handleUserInput(trigger, orgId, workflowExecution.Workflow.ID, workflowExecution.ExecutionId)
if err != nil {
- log.Printf("Failed userinput handler: %s", err)
+ log.Printf("[WARNING] Failed userinput handler: %s", err)
actionResult.Result = fmt.Sprintf("Cloud error: %s", err)
workflowExecution.Results = append(workflowExecution.Results, actionResult)
workflowExecution.Status = "ABORTED"
err = shuffle.SetWorkflowExecution(ctx, *workflowExecution, true)
if err != nil {
- log.Printf("Failed to set execution during wait")
+ log.Printf("[WARNING] Failed to set execution during wait: %s", err)
} else {
- log.Printf("Successfully set the execution to waiting.")
+ log.Printf("[INFO] Successfully set the execution %s to waiting.", workflowExecution.ExecutionId)
}
resp.WriteHeader(401)
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Error: %s"}`, err)))
} else {
- log.Printf("Successful userinput handler")
+ log.Printf("[INFO] Successful userinput handler")
resp.WriteHeader(200)
resp.Write([]byte(fmt.Sprintf(`{"success": true, "reason": "CLOUD IS DONE"}`)))
@@ -886,7 +936,7 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
workflowExecution.Status = actionResult.Status
err = shuffle.SetWorkflowExecution(ctx, *workflowExecution, true)
if err != nil {
- log.Printf("Failed ")
+ log.Printf("[WARNING] Failed setting userinput: %s", err)
} else {
log.Printf("Successfully set the execution to waiting.")
}
@@ -1168,8 +1218,6 @@ func getWorkflowLocal(fileId string, request *http.Request) ([]byte, error) {
return body, nil
}
-//// New execution with firestore
-
func handleExecution(id string, workflow shuffle.Workflow, request *http.Request) (shuffle.WorkflowExecution, string, error) {
//go func() {
// log.Printf("\n\nPRE TIME: %s\n\n", time.Now().Format("2006-01-02 15:04:05"))
@@ -1233,725 +1281,38 @@ func handleExecution(id string, workflow shuffle.Workflow, request *http.Request
return shuffle.WorkflowExecution{}, fmt.Sprintf(`workflow %s is invalid`, workflow.ID), errors.New("Failed getting workflow")
}
- workflowBytes, err := json.Marshal(workflow)
+ workflowExecution, execInfo, _, err := shuffle.PrepareWorkflowExecution(ctx, workflow, request)
if err != nil {
- log.Printf("Failed workflow unmarshal in execution: %s", err)
+ log.Printf("[WARNING] Failed in prepareExecution: %s", err)
return shuffle.WorkflowExecution{}, "", err
}
- //log.Println(workflow)
- var workflowExecution shuffle.WorkflowExecution
- err = json.Unmarshal(workflowBytes, &workflowExecution.Workflow)
+ err = imageCheckBuilder(execInfo.ImageNames)
if err != nil {
- log.Printf("Failed execution unmarshaling: %s", err)
- return shuffle.WorkflowExecution{}, "Failed unmarshal during execution", err
- }
-
- makeNew := true
- start, startok := request.URL.Query()["start"]
- if request.Method == "POST" {
- body, err := ioutil.ReadAll(request.Body)
- if err != nil {
- log.Printf("[ERROR] Failed request POST read: %s", err)
- return shuffle.WorkflowExecution{}, "Failed getting body", err
- }
-
- // This one doesn't really matter.
- log.Printf("[INFO] Running POST execution with body of length %d for workflow %s", len(string(body)), workflowExecution.Workflow.ID)
-
- if len(body) >= 4 {
- if body[0] == 34 && body[len(body)-1] == 34 {
- body = body[1 : len(body)-1]
- }
- if body[0] == 34 && body[len(body)-1] == 34 {
- body = body[1 : len(body)-1]
- }
- }
-
- sourceAuth, sourceAuthOk := request.URL.Query()["source_auth"]
- if sourceAuthOk {
- //log.Printf("\n\n\nSETTING SOURCE WORKFLOW AUTH TO %s!!!\n\n\n", sourceAuth[0])
- workflowExecution.ExecutionSourceAuth = sourceAuth[0]
- } else {
- //log.Printf("Did NOT get source workflow")
- }
-
- sourceNode, sourceNodeOk := request.URL.Query()["source_node"]
- if sourceNodeOk {
- //log.Printf("\n\n\nSETTING SOURCE WORKFLOW NODE TO %s!!!\n\n\n", sourceNode[0])
- workflowExecution.ExecutionSourceNode = sourceNode[0]
- } else {
- //log.Printf("Did NOT get source workflow")
- }
-
- //workflowExecution.ExecutionSource = "default"
- sourceWorkflow, sourceWorkflowOk := request.URL.Query()["source_workflow"]
- if sourceWorkflowOk {
- //log.Printf("Got source workflow %s", sourceWorkflow)
- workflowExecution.ExecutionSource = sourceWorkflow[0]
- } else {
- //log.Printf("Did NOT get source workflow")
- }
-
- sourceExecution, sourceExecutionOk := request.URL.Query()["source_execution"]
- if sourceExecutionOk {
- //log.Printf("[INFO] Got source execution%s", sourceExecution)
- workflowExecution.ExecutionParent = sourceExecution[0]
- } else {
- //log.Printf("Did NOT get source execution")
- }
-
- if len(string(body)) < 50 {
- //log.Println(body)
- // String in string
- //log.Println(body)
-
- //if string(body)[0] == "\"" && string(body)[string(body)
- log.Printf("[DEBUG] Body: %s", string(body))
- }
-
- var execution shuffle.ExecutionRequest
- err = json.Unmarshal(body, &execution)
- if err != nil {
- log.Printf("[WARNING] Failed execution POST unmarshaling - continuing anyway: %s", err)
- //return shuffle.WorkflowExecution{}, "", err
- }
-
- if execution.Start == "" && len(body) > 0 {
- execution.ExecutionArgument = string(body)
- }
-
- // FIXME - this should have "execution_argument" from executeWorkflow frontend
- //log.Printf("EXEC: %#v", execution)
- if len(execution.ExecutionArgument) > 0 {
- workflowExecution.ExecutionArgument = execution.ExecutionArgument
- }
-
- if len(execution.ExecutionSource) > 0 {
- workflowExecution.ExecutionSource = execution.ExecutionSource
- }
-
- //log.Printf("Execution data: %#v", execution)
- if len(execution.Start) == 36 && len(workflow.Actions) > 0 {
- log.Printf("[INFO] Should start execution on node %s", execution.Start)
- workflowExecution.Start = execution.Start
-
- found := false
- for _, action := range workflow.Actions {
- if action.ID == execution.Start {
- found = true
- break
- }
- }
-
- if !found {
- log.Printf("[ERROR] Action %s was NOT found! Exiting execution.", execution.Start)
- return shuffle.WorkflowExecution{}, fmt.Sprintf("Startnode %s was not found in actions", workflow.Start), errors.New(fmt.Sprintf("Startnode %s was not found in actions", workflow.Start))
- }
- } else if len(execution.Start) > 0 {
- //log.Printf("[INFO] !")
- //log.Printf("[ERROR] START ACTION %s IS WRONG ID LENGTH %d!", execution.Start, len(execution.Start))
- //return shuffle.WorkflowExecution{}, fmt.Sprintf("Startnode %s was not found in actions", execution.Start), errors.New(fmt.Sprintf("Startnode %s was not found in actions", execution.Start))
- }
-
- if len(execution.ExecutionId) == 36 {
- workflowExecution.ExecutionId = execution.ExecutionId
- } else {
- sessionToken := uuid.NewV4()
- workflowExecution.ExecutionId = sessionToken.String()
- }
- } else {
- // Check for parameters of start and ExecutionId
- // This is mostly used for user input trigger
-
- answer, answerok := request.URL.Query()["answer"]
- referenceId, referenceok := request.URL.Query()["reference_execution"]
- if answerok && referenceok {
- // If answer is false, reference execution with result
- log.Printf("[INFO] Answer is OK AND reference is OK!")
- if answer[0] == "false" {
- log.Printf("Should update reference and return, no need for further execution!")
-
- // Get the reference execution
- oldExecution, err := shuffle.GetWorkflowExecution(ctx, referenceId[0])
- if err != nil {
- log.Printf("Failed getting execution (execution) %s: %s", referenceId[0], err)
- return shuffle.WorkflowExecution{}, fmt.Sprintf("Failed getting execution ID %s because it doesn't exist.", referenceId[0]), err
- }
-
- if oldExecution.Workflow.ID != id {
- log.Println("Wrong workflowid!")
- return shuffle.WorkflowExecution{}, fmt.Sprintf("Bad ID %s", referenceId), errors.New("Bad ID")
- }
-
- newResults := []shuffle.ActionResult{}
- //log.Printf("%#v", oldExecution.Results)
- for _, result := range oldExecution.Results {
- log.Printf("%s - %s", result.Action.ID, start[0])
- if result.Action.ID == start[0] {
- note, noteok := request.URL.Query()["note"]
- if noteok {
- result.Result = fmt.Sprintf("User note: %s", note[0])
- } else {
- result.Result = fmt.Sprintf("User clicked %s", answer[0])
- }
-
- // Stopping the whole thing
- result.CompletedAt = int64(time.Now().Unix())
- result.Status = "ABORTED"
- oldExecution.Status = result.Status
- oldExecution.Result = result.Result
- oldExecution.LastNode = result.Action.ID
- }
-
- newResults = append(newResults, result)
- }
-
- oldExecution.Results = newResults
- err = shuffle.SetWorkflowExecution(ctx, *oldExecution, true)
- if err != nil {
- log.Printf("Error saving workflow execution actionresult setting: %s", err)
- return shuffle.WorkflowExecution{}, fmt.Sprintf("Failed setting workflowexecution actionresult in execution: %s", err), err
- }
-
- return shuffle.WorkflowExecution{}, "", nil
- }
- }
-
- if referenceok {
- log.Printf("Handling an old execution continuation!")
- // Will use the old name, but still continue with NEW ID
- oldExecution, err := shuffle.GetWorkflowExecution(ctx, referenceId[0])
- if err != nil {
- log.Printf("Failed getting execution (execution) %s: %s", referenceId[0], err)
- return shuffle.WorkflowExecution{}, fmt.Sprintf("Failed getting execution ID %s because it doesn't exist.", referenceId[0]), err
- }
-
- workflowExecution = *oldExecution
- }
-
- if len(workflowExecution.ExecutionId) == 0 {
- sessionToken := uuid.NewV4()
- workflowExecution.ExecutionId = sessionToken.String()
- } else {
- log.Printf("Using the same executionId as before: %s", workflowExecution.ExecutionId)
- makeNew = false
- }
-
- // Don't override workflow defaults
- }
-
- if startok {
- //log.Printf("\n\n[INFO] Setting start to %s based on query!\n\n", start[0])
- //workflowExecution.Workflow.Start = start[0]
- workflowExecution.Start = start[0]
- }
-
- // FIXME - regex uuid, and check if already exists?
- if len(workflowExecution.ExecutionId) != 36 {
- log.Printf("Invalid uuid: %s", workflowExecution.ExecutionId)
- return shuffle.WorkflowExecution{}, "Invalid uuid", err
- }
-
- // FIXME - find owner of workflow
- // FIXME - get the actual workflow itself and build the request
- // MAYBE: Don't send the workflow within the pubsub, as this requires more data to be sent
- // Check if a worker already exists for company, else run one with:
- // locations, project IDs and subscription names
-
- // When app is executed:
- // Should update with status execution (somewhere), which will trigger the next node
- // IF action.type == internal, we need the internal watcher to be running and executing
- // This essentially means the WORKER has to be the responsible party for new actions in the INTERNAL landscape
- // Results are ALWAYS posted back to cloud@execution_id?
- if makeNew {
- workflowExecution.Type = "workflow"
- //workflowExecution.Stream = "tmp"
- //workflowExecution.WorkflowQueue = "tmp"
- //workflowExecution.SubscriptionNameNodestream = "testcompany-nodestream"
- //workflowExecution.Locations = []string{"europe-west2"}
- workflowExecution.ProjectId = gceProject
- workflowExecution.WorkflowId = workflow.ID
- workflowExecution.StartedAt = int64(time.Now().Unix())
- workflowExecution.CompletedAt = 0
- workflowExecution.Authorization = uuid.NewV4().String()
-
- // Status for the entire workflow.
- workflowExecution.Status = "EXECUTING"
- }
-
- if len(workflowExecution.ExecutionSource) == 0 {
- log.Printf("[INFO] No execution source (trigger) specified. Setting to default")
- workflowExecution.ExecutionSource = "default"
- } else {
- log.Printf("[INFO] Execution source is %s for execution ID %s in workflow %s", workflowExecution.ExecutionSource, workflowExecution.ExecutionId, workflowExecution.Workflow.ID)
- }
-
- workflowExecution.ExecutionVariables = workflow.ExecutionVariables
- if len(workflowExecution.Start) == 0 && len(workflowExecution.Workflow.Start) > 0 {
- workflowExecution.Start = workflowExecution.Workflow.Start
- }
-
- startnodeFound := false
- newStartnode := ""
- for _, item := range workflowExecution.Workflow.Actions {
- if item.ID == workflowExecution.Start {
- startnodeFound = true
- }
-
- if item.IsStartNode {
- newStartnode = item.ID
- }
- }
-
- if !startnodeFound {
- log.Printf("[INFO] Couldn't find startnode %s. Remapping to %#v", workflowExecution.Start, newStartnode)
-
- if len(newStartnode) > 0 {
- workflowExecution.Start = newStartnode
- } else {
- return shuffle.WorkflowExecution{}, fmt.Sprintf("Startnode couldn't be found"), errors.New("Startnode isn't defined in this workflow..")
- }
- }
-
- childNodes := shuffle.FindChildNodes(workflowExecution, workflowExecution.Start)
-
- topic := "workflows"
- startFound := false
- // FIXME - remove this?
- newActions := []shuffle.Action{}
- defaultResults := []shuffle.ActionResult{}
-
- allAuths := []shuffle.AppAuthenticationStorage{}
- for _, action := range workflowExecution.Workflow.Actions {
- //action.LargeImage = ""
- if action.ID == workflowExecution.Start {
- startFound = true
- }
- //log.Println(action.Environment)
-
- if action.Environment == "" {
- return shuffle.WorkflowExecution{}, fmt.Sprintf("Environment is not defined for %s", action.Name), errors.New("Environment not defined!")
- }
-
- // FIXME: Authentication parameters
- if len(action.AuthenticationId) > 0 {
- if len(allAuths) == 0 {
- allAuths, err = shuffle.GetAllWorkflowAppAuth(ctx, workflow.ExecutingOrg.Id)
- if err != nil {
- log.Printf("Api authentication failed in get all app auth: %s", err)
- return shuffle.WorkflowExecution{}, fmt.Sprintf("Api authentication failed in get all app auth: %s", err), err
- }
- }
-
- curAuth := shuffle.AppAuthenticationStorage{Id: ""}
- for _, auth := range allAuths {
- if auth.Id == action.AuthenticationId {
- curAuth = auth
- break
- }
- }
-
- if len(curAuth.Id) == 0 {
- return shuffle.WorkflowExecution{}, fmt.Sprintf("Auth ID %s doesn't exist", action.AuthenticationId), errors.New(fmt.Sprintf("Auth ID %s doesn't exist", action.AuthenticationId))
- }
-
- if curAuth.Encrypted {
- setField := true
- newFields := []shuffle.AuthenticationStore{}
- for _, field := range curAuth.Fields {
- parsedKey := fmt.Sprintf("%s_%d_%s_%s", curAuth.OrgId, curAuth.Created, curAuth.Label, field.Key)
- newValue, err := shuffle.HandleKeyDecryption(field.Value, parsedKey)
- if err != nil {
- log.Printf("[WARNING] Failed decryption for %s: %s", field.Key, err)
- setField = false
- break
- }
-
- field.Value = newValue
- newFields = append(newFields, field)
- }
-
- if setField {
- curAuth.Fields = newFields
- }
- } else {
- log.Printf("[INFO] AUTH IS NOT ENCRYPTED - attempting encrypting!")
- err = shuffle.SetWorkflowAppAuthDatastore(ctx, curAuth, curAuth.Id)
- if err != nil {
- log.Printf("[WARNING] Failed running encryption during execution: %s", err)
- }
- }
-
- newParams := []shuffle.WorkflowAppActionParameter{}
- if strings.ToLower(curAuth.Type) == "oauth2" {
- log.Printf("[DEBUG] Should replace auth parameters (Oauth2)")
-
- for _, param := range curAuth.Fields {
- if param.Key == "expiration" {
- continue
- }
-
- newParams = append(newParams, shuffle.WorkflowAppActionParameter{
- Name: param.Key,
- Value: param.Value,
- })
- }
-
- for _, param := range action.Parameters {
- //log.Printf("Param: %#v", param)
- if param.Configuration {
- continue
- }
-
- newParams = append(newParams, param)
- }
- } else {
- // Rebuild params with the right data. This is to prevent issues on the frontend
- for _, param := range action.Parameters {
-
- for _, authparam := range curAuth.Fields {
- if param.Name == authparam.Key {
- param.Value = authparam.Value
- //log.Printf("Name: %s - value: %s", param.Name, param.Value)
- //log.Printf("Name: %s - value: %s\n", param.Name, param.Value)
- break
- }
- }
-
- newParams = append(newParams, param)
- }
- }
-
- action.Parameters = newParams
- }
-
- action.LargeImage = ""
- if len(action.Label) == 0 {
- action.Label = action.ID
- }
- //log.Printf("LABEL: %s", action.Label)
- newActions = append(newActions, action)
-
- // If the node is NOT found, it's supposed to be set to SKIPPED,
- // as it's not a childnode of the startnode
- // This is a configuration item for the workflow itself.
- if len(workflowExecution.Results) > 0 {
- defaultResults = []shuffle.ActionResult{}
- for _, result := range workflowExecution.Results {
- if result.Status == "WAITING" {
- result.Status = "FINISHED"
- result.Result = "Continuing"
- }
-
- defaultResults = append(defaultResults, result)
- }
- } else if len(workflowExecution.Results) == 0 && !workflowExecution.Workflow.Configuration.StartFromTop {
- found := false
- for _, nodeId := range childNodes {
- if nodeId == action.ID {
- //log.Printf("Found %s", action.ID)
- found = true
- }
- }
-
- if !found {
- if action.ID == workflowExecution.Start {
- continue
- }
-
- //log.Printf("[WARNING] Set %s to SKIPPED as it's NOT a childnode of the startnode.", action.ID)
- curaction := shuffle.Action{
- AppName: action.AppName,
- AppVersion: action.AppVersion,
- Label: action.Label,
- Name: action.Name,
- ID: action.ID,
- }
- //action
- //curaction.Parameters = []
- defaultResults = append(defaultResults, shuffle.ActionResult{
- Action: curaction,
- ExecutionId: workflowExecution.ExecutionId,
- Authorization: workflowExecution.Authorization,
- Result: "Skipped because it's not under the startnode",
- StartedAt: 0,
- CompletedAt: 0,
- Status: "SKIPPED",
- })
- }
- }
- }
-
- removeTriggers := []string{}
- for triggerIndex, trigger := range workflowExecution.Workflow.Triggers {
- //log.Printf("[INFO] ID: %s vs %s", trigger.ID, workflowExecution.Start)
- if trigger.ID == workflowExecution.Start {
- if trigger.AppName == "User Input" {
- startFound = true
- break
- }
- }
-
- if trigger.AppName == "User Input" || trigger.AppName == "Shuffle Workflow" {
- found := false
- for _, node := range childNodes {
- if node == trigger.ID {
- found = true
- break
- }
- }
-
- if !found {
- //log.Printf("SHOULD SET TRIGGER %s TO BE SKIPPED", trigger.ID)
-
- curaction := shuffle.Action{
- AppName: "shuffle-subflow",
- AppVersion: trigger.AppVersion,
- Label: trigger.Label,
- Name: trigger.Name,
- ID: trigger.ID,
- }
-
- defaultResults = append(defaultResults, shuffle.ActionResult{
- Action: curaction,
- ExecutionId: workflowExecution.ExecutionId,
- Authorization: workflowExecution.Authorization,
- Result: "Skipped because it's not under the startnode",
- StartedAt: 0,
- CompletedAt: 0,
- Status: "SKIPPED",
- })
- } else {
- // Replaces trigger with the subflow
- //if trigger.AppName == "Shuffle Workflow" {
- // replaceActions := false
- // workflowAction := ""
- // for _, param := range trigger.Parameters {
- // if param.Name == "argument" && !strings.Contains(param.Value, ".#") {
- // replaceActions = true
- // }
-
- // if param.Name == "startnode" {
- // workflowAction = param.Value
- // }
- // }
-
- // if replaceActions {
- // replacementNodes, newBranches, lastnode := shuffle.GetReplacementNodes(ctx, workflowExecution, trigger, trigger.Label)
- // log.Printf("REPLACEMENTS: %d, %d", len(replacementNodes), len(newBranches))
- // if len(replacementNodes) > 0 {
- // for _, action := range replacementNodes {
- // found := false
-
- // for subActionIndex, subaction := range newActions {
- // if subaction.ID == action.ID {
- // found = true
- // //newActions[subActionIndex].Name = action.Name
- // newActions[subActionIndex].Label = action.Label
- // break
- // }
- // }
-
- // if !found {
- // action.SubAction = true
- // newActions = append(newActions, action)
- // }
-
- // // Check if it's already set to have a value
- // for resultIndex, result := range defaultResults {
- // if result.Action.ID == action.ID {
- // defaultResults = append(defaultResults[:resultIndex], defaultResults[resultIndex+1:]...)
- // break
- // }
- // }
- // }
-
- // for _, branch := range newBranches {
- // workflowExecution.Workflow.Branches = append(workflowExecution.Workflow.Branches, branch)
- // }
-
- // // Append branches:
- // // parent -> new inner node (FIRST one)
- // for branchIndex, branch := range workflowExecution.Workflow.Branches {
- // if branch.DestinationID == trigger.ID {
- // log.Printf("REPLACE DESTINATION WITH %s!!", workflowAction)
- // workflowExecution.Workflow.Branches[branchIndex].DestinationID = workflowAction
- // }
-
- // if branch.SourceID == trigger.ID {
- // log.Printf("REPLACE SOURCE WITH LASTNODE %s!!", lastnode)
- // workflowExecution.Workflow.Branches[branchIndex].SourceID = lastnode
- // }
- // }
-
- // // Remove the trigger
- // removeTriggers = append(removeTriggers, workflowExecution.Workflow.Triggers[triggerIndex].ID)
- // }
-
- // log.Printf("NEW ACTION LENGTH %d, RESULT: %d, Triggers: %d, BRANCHES: %d", len(newActions), len(defaultResults), len(workflowExecution.Workflow.Triggers), len(workflowExecution.Workflow.Branches))
- // }
- //}
- _ = triggerIndex
- }
- }
- }
-
- //newTriggers := []shuffle.Trigger{}
- //for _, trigger := range workflowExecution.Workflow.Triggers {
- // found := false
- // for _, triggerId := range removeTriggers {
- // if trigger.ID == triggerId {
- // found = true
- // break
- // }
- // }
-
- // if found {
- // log.Printf("[WARNING] Removed trigger %s during execution", trigger.ID)
- // continue
- // }
-
- // newTriggers = append(newTriggers, trigger)
- //}
- //workflowExecution.Workflow.Triggers = newTriggers
- _ = removeTriggers
-
- if !startFound {
- if len(workflowExecution.Start) == 0 && len(workflowExecution.Workflow.Start) > 0 {
- workflowExecution.Start = workflow.Start
- } else if len(workflowExecution.Workflow.Actions) > 0 {
- workflowExecution.Start = workflowExecution.Workflow.Actions[0].ID
- } else {
- log.Printf("[ERROR] Startnode %s doesn't exist!!", workflowExecution.Start)
- return shuffle.WorkflowExecution{}, fmt.Sprintf("Workflow action %s doesn't exist in workflow", workflowExecution.Start), errors.New(fmt.Sprintf(`Workflow start node "%s" doesn't exist. Exiting!`, workflowExecution.Start))
- }
- }
-
- //log.Printf("EXECUTION START: %s", workflowExecution.Start)
-
- // Verification for execution environments
- workflowExecution.Results = defaultResults
- workflowExecution.Workflow.Actions = newActions
- onpremExecution := true
- environments := []string{}
-
- if len(workflowExecution.ExecutionOrg) == 0 && len(workflow.ExecutingOrg.Id) > 0 {
- workflowExecution.ExecutionOrg = workflow.ExecutingOrg.Id
- }
-
- var allEnvs []shuffle.Environment
- if len(workflowExecution.ExecutionOrg) > 0 {
- //log.Printf("[INFO] Executing ORG: %s", workflowExecution.ExecutionOrg)
-
- allEnvironments, err := shuffle.GetEnvironments(ctx, workflowExecution.ExecutionOrg)
- if err != nil {
- log.Printf("Failed finding environments: %s", err)
- return shuffle.WorkflowExecution{}, fmt.Sprintf("Workflow environments not found for this org"), errors.New(fmt.Sprintf("Workflow environments not found for this org"))
- }
-
- for _, curenv := range allEnvironments {
- if curenv.Archived {
- continue
- }
-
- allEnvs = append(allEnvs, curenv)
- }
- } else {
- log.Printf("[ERROR] No org identified for execution of %s. Returning", workflowExecution.Workflow.ID)
- return shuffle.WorkflowExecution{}, "No org identified for execution", errors.New("No org identified for execution")
- }
-
- if len(allEnvs) == 0 {
- log.Printf("[ERROR] No active environments found for org: %s", workflowExecution.ExecutionOrg)
- return shuffle.WorkflowExecution{}, "No active environments found", errors.New(fmt.Sprintf("No active env found for org %s", workflowExecution.ExecutionOrg))
- }
-
- // Check if the actions are children of the startnode?
- imageNames := []string{}
- cloudExec := false
- for _, action := range workflowExecution.Workflow.Actions {
- // Verify if the action environment exists and append
- found := false
- for _, env := range allEnvs {
- if env.Name == action.Environment {
- found = true
-
- if env.Type == "cloud" {
- cloudExec = true
- } else if env.Type == "onprem" {
- onpremExecution = true
- } else {
- log.Printf("[ERROR] No handler for environment type %s", env.Type)
- return shuffle.WorkflowExecution{}, "No active environments found", errors.New(fmt.Sprintf("No handler for environment type %s", env.Type))
- }
- break
- }
- }
-
- if !found {
- log.Printf("[ERROR] Couldn't find environment %s. Maybe it's inactive?", action.Environment)
- return shuffle.WorkflowExecution{}, "Couldn't find the environment", errors.New(fmt.Sprintf("Couldn't find env %s in org %s", action.Environment, workflowExecution.ExecutionOrg))
- }
-
- found = false
- for _, env := range environments {
- if env == action.Environment {
-
- found = true
- break
- }
- }
-
- // Check if the app exists?
- newName := action.AppName
- newName = strings.ReplaceAll(newName, " ", "-")
- imageNames = append(imageNames, fmt.Sprintf("%s:%s_%s", baseDockerName, newName, action.AppVersion))
-
- if !found {
- environments = append(environments, action.Environment)
- }
- }
-
- err = imageCheckBuilder(imageNames)
- if err != nil {
- log.Printf("[ERROR] Failed building the required images from %#v: %s", imageNames, err)
+ log.Printf("[ERROR] Failed building the required images from %#v: %s", execInfo.ImageNames, err)
return shuffle.WorkflowExecution{}, "Failed building missing Docker images", err
}
- //b, err := json.Marshal(workflowExecution)
- //if err == nil {
- // log.Printf("LEN: %d", len(string(b)))
- // //workflowExecution.ExecutionOrg.SyncFeatures = Org{}
- //}
-
- workflowExecution.Workflow.ExecutingOrg = shuffle.OrgMini{
- Id: workflowExecution.Workflow.ExecutingOrg.Id,
- }
- workflowExecution.Workflow.Org = []shuffle.OrgMini{
- workflowExecution.Workflow.ExecutingOrg,
- }
-
//Org []Org `json:"org,omitempty" datastore:"org"`
err = shuffle.SetWorkflowExecution(ctx, workflowExecution, true)
if err != nil {
- log.Printf("[WARNING] Error saving workflow execution for updates %s: %s", topic, err)
+ log.Printf("[WARNING] Error saving workflow execution for updates %s", err)
return shuffle.WorkflowExecution{}, fmt.Sprintf("Failed setting workflowexecution: %s", err), err
}
// Adds queue for onprem execution
// FIXME - add specifics to executionRequest, e.g. specific environment (can run multi onprem)
- if onpremExecution {
+ if execInfo.OnpremExecution {
// FIXME - tmp name based on future companyname-companyId
// This leads to issues with overlaps. Should set limits and such instead
- for _, environment := range environments {
+ for _, environment := range execInfo.Environments {
log.Printf("[INFO] Execution: %s should execute onprem with execution environment \"%s\". Workflow: %s", workflowExecution.ExecutionId, environment, workflowExecution.Workflow.ID)
executionRequest := shuffle.ExecutionRequest{
ExecutionId: workflowExecution.ExecutionId,
WorkflowId: workflowExecution.Workflow.ID,
Authorization: workflowExecution.Authorization,
- Environments: environments,
+ Environments: execInfo.Environments,
}
//executionRequestWrapper, err := getWorkflowQueue(ctx, environment)
@@ -1972,7 +1333,7 @@ func handleExecution(id string, workflow shuffle.Workflow, request *http.Request
}
// Verifies and runs cloud executions
- if cloudExec {
+ if execInfo.CloudExec {
featuresList, err := handleVerifyCloudsync(workflowExecution.ExecutionOrg)
if !featuresList.Workflows.Active || err != nil {
log.Printf("Error: %s", err)
@@ -2684,7 +2045,7 @@ func getWorkflowApps(resp http.ResponseWriter, request *http.Request) {
// Double unmarshal because of user apps
newbody, err := json.Marshal(newapps)
if err != nil {
- log.Printf("Failed unmarshalling all newapps: %s", err)
+ log.Printf("[ERROR] Failed unmarshalling all newapps: %s", err)
resp.WriteHeader(401)
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed unpacking workflow apps"}`)))
return
@@ -4050,9 +3411,19 @@ func LoadSpecificApps(resp http.ResponseWriter, request *http.Request) {
if tmpBody.ForceUpdate {
dockercli, err := dockerclient.NewEnvClient()
if err == nil {
- _, err := dockercli.ImagePull(ctx, "frikky/shuffle:app_sdk", types.ImagePullOptions{})
- if err != nil {
- log.Printf("[WARNING] Failed to download apps with the new App SDK: %s", err)
+
+ appSdk := os.Getenv("SHUFFLE_APP_SDK_VERSION")
+ if len(appSdk) == 0 {
+ _, err := dockercli.ImagePull(ctx, "frikky/shuffle:app_sdk", types.ImagePullOptions{})
+ if err != nil {
+ log.Printf("[WARNING] Failed to download new App SDK: %s", err)
+ }
+ } else {
+ _, err := dockercli.ImagePull(ctx, fmt.Sprintf("%s/%s/shuffle-app_sdk:%s", "ghcr.io", "frikky", appSdk), types.ImagePullOptions{})
+ if err != nil {
+ log.Printf("[WARNING] Failed to download new App SDK %s: %s", err)
+ }
+
}
} else {
log.Printf("[WARNING] Failed to download apps with the new App SDK because of docker cli: %s", err)
diff --git a/docker-compose.yml b/docker-compose.yml
index 88fe59f2..494635d1 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -2,7 +2,7 @@ version: '3'
services:
frontend:
#build: ./frontend
- image: ghcr.io/frikky/shuffle-frontend:latest
+ image: ghcr.io/frikky/shuffle-frontend:nightly
container_name: shuffle-frontend
hostname: shuffle-frontend
ports:
@@ -17,7 +17,7 @@ services:
- backend
backend:
#build: ./backend
- image: ghcr.io/frikky/shuffle-backend:latest
+ image: ghcr.io/frikky/shuffle-backend:nightly
container_name: shuffle-backend
hostname: ${BACKEND_HOSTNAME}
# Here for debugging:
@@ -39,8 +39,7 @@ services:
#- opensearch #Not necessary because dependancy is handled within the backend itself instead
#- database
orborus:
- #build: ./functions/onprem/orborus
- image: ghcr.io/frikky/shuffle-orborus:latest
+ image: ghcr.io/frikky/shuffle-orborus:nightly
container_name: shuffle-orborus
hostname: shuffle-orborus
networks:
@@ -48,8 +47,8 @@ services:
volumes:
- /var/run/docker.sock:/var/run/docker.sock
environment:
- - SHUFFLE_APP_SDK_VERSION=latest
- - SHUFFLE_WORKER_VERSION=latest
+ - SHUFFLE_APP_SDK_VERSION=0.8.97
+ - SHUFFLE_WORKER_VERSION=0.9.30
- ORG_ID=${ORG_ID}
- ENVIRONMENT_NAME=${ENVIRONMENT_NAME}
- BASE_URL=http://${OUTER_HOSTNAME}:${BACKEND_PORT}
diff --git a/frontend/Dockerfile b/frontend/Dockerfile
index 12ba0b27..5c29245a 100644
--- a/frontend/Dockerfile
+++ b/frontend/Dockerfile
@@ -17,6 +17,7 @@ COPY ./src /usr/src/app/src/
COPY ./*.sh /usr/src/app/
COPY ./*.json /usr/src/app/
+RUN rm -rf /usr/src/app/node_modules/webpack
RUN yarn build
# Production environment
diff --git a/frontend/package.json b/frontend/package.json
index 4b9a7814..6e606427 100644
--- a/frontend/package.json
+++ b/frontend/package.json
@@ -1,9 +1,10 @@
{
"name": "shuffler",
"homepage": "https://shuffler.io",
- "version": "0.9.24",
+ "version": "0.9.25",
"private": true,
"dependencies": {
+ "@babel/core": "^7.15.8",
"@material-ui/core": "^4.5.2",
"@material-ui/data-grid": "^4.0.0-alpha.22",
"@material-ui/icons": "^4.11.2",
@@ -13,8 +14,8 @@
"@uiw/react-codemirror": "^3.2.1",
"@use-it/interval": "^1.0.0",
"babel-eslint": "^10.1.0",
- "class-transformer": "^0.3.1",
- "create-react-app": "^2.0.3",
+ "class-transformer": "^0.4.0",
+ "create-react-app": "^4.0.3",
"cytoscape": "^3.11.0",
"cytoscape-clipboard": "^2.2.1",
"cytoscape-cxtmenu": "^3.1.1",
@@ -62,7 +63,7 @@
"shellwords": "^0.1.1",
"simplebar": "^4.2.3",
"styled-components": "^4.4.0",
- "webpack": "4.44.2",
+ "webpack": "^4.44.2",
"websocket": "^1.0.30",
"yaml": "^1.7.2",
"yamljs": "^0.3.0",
diff --git a/frontend/public/images/btn_google_light_focus_ios.svg b/frontend/public/images/btn_google_light_focus_ios.svg
new file mode 100644
index 00000000..1f3ee4ff
--- /dev/null
+++ b/frontend/public/images/btn_google_light_focus_ios.svg
@@ -0,0 +1,44 @@
+
+
\ No newline at end of file
diff --git a/frontend/src/components/Header.js b/frontend/src/components/Header.js
index 449c057b..113ba374 100644
--- a/frontend/src/components/Header.js
+++ b/frontend/src/components/Header.js
@@ -276,7 +276,7 @@ const Header = props => {
setAnchorEl(event.currentTarget);
}}>