Merge branch 'launch' into launch

This commit is contained in:
Frikky
2021-10-31 22:23:06 +01:00
committed by GitHub
24 changed files with 929 additions and 2659 deletions
+19
View File
@@ -0,0 +1,19 @@
# Security Policy
## Supported Versions
Shuffle is currently still in beta, but we aim to support older version with critical severity issues, but do advise you to stay up to date with Major versions.
| Version | Supported |
| ------- | ------------------ |
| >=0.9.0 | :white_check_mark: |
| < 0.9.0 | :x: |
## Reporting a Vulnerability
Reporting a vulnerability can either be done to (frikky@shuffler.io)[mailto:frikky@shuffler.io] or [through the contact page on our website](https://shuffler.io/contact)
Security.txt: https://shuffler.io/.well_known/security.txt
When a >medium severity vulnerability is discovered, expect it to be fixed ASAP - please nag us until it is. Security is a top priority, and we expect you to keep us accountable.
In the case it makes sense, we'll further create a security advisory, and publish a new CVE for your new glorious finding.
+176 -55
View File
@@ -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:
@@ -1645,7 +1652,11 @@ class AppBase:
self.send_result(self.action_result, headers, stream_path)
self.logger.info(f"[ERROR] Sent FAILURE response to backend due to : {e}")
os.exit()
if runtime == "run":
return template
else:
os.exit()
return template
@@ -2005,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"]
@@ -2105,7 +2116,7 @@ class AppBase:
if " " in actionname:
actionname.replace(" ", "_", -1)
#print(action)
#if action.generated:
# actionname = actionname.lower()
@@ -2113,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
@@ -2460,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
@@ -2551,7 +2563,8 @@ class AppBase:
#newres = ""
while True:
try:
newres = await func(**params)
#newres = await func(**params)
newres = func(**params)
break
except TypeError as e:
newres = ""
@@ -2626,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
@@ -2821,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)
+2 -1
View File
@@ -3,7 +3,7 @@
### DEFAULT
NAME=shuffle-app_sdk
VERSION=0.9.26
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
@@ -16,6 +16,7 @@ docker build . -f Dockerfile -t frikky/shuffle:app_sdk -t frikky/$NAME:$VERSION
docker push frikky/shuffle:app_sdk
docker push ghcr.io/frikky/$NAME:$VERSION
docker push ghcr.io/frikky/$NAME:nightly
docker push ghcr.io/frikky/$NAME:latest
#### BLACKARCH ###
NAME=shuffle-app_sdk_kali
+3
View File
@@ -2,3 +2,6 @@ urllib3==1.26.5
requests==2.25.1
MarkupSafe==2.0.1
liquidpy==0.7.2
flask[async]==2.0.2
waitress==2.0.0
#flask==1.1.2
+2 -2
View File
@@ -2,7 +2,7 @@ module main
go 1.15
replace github.com/shuffle/shuffle-shared => ../../../../git/shuffle-shared
//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
@@ -21,7 +21,7 @@ require (
github.com/gorilla/mux v1.8.0
github.com/h2non/filetype v1.1.1
github.com/satori/go.uuid v1.2.0
github.com/shuffle/shuffle-shared v0.1.18
github.com/shuffle/shuffle-shared v0.1.27
go4.org v0.0.0-20201209231011-d4a079459e60 // indirect
golang.org/x/crypto v0.0.0-20210921155107-089bfa567519
google.golang.org/api v0.58.0
+3 -3
View File
@@ -3084,7 +3084,7 @@ func buildSwaggerApp(resp http.ResponseWriter, body []byte, user shuffle.User) {
basePath, err := shuffle.BuildStructure(swagger, newmd5)
if err != nil {
log.Printf("Failed to build base structure: %s", err)
log.Printf("[WARNING] Failed to build base structure: %s", err)
resp.WriteHeader(500)
resp.Write([]byte(`{"success": false, "reason": "Failed building baseline structure"}`))
return
@@ -3103,7 +3103,7 @@ func buildSwaggerApp(resp http.ResponseWriter, body []byte, user shuffle.User) {
// Can't overwrite existing normal app
workflowApps, err := shuffle.GetPrioritizedApps(ctx, user)
if err != nil {
log.Printf("Failed getting all workflow apps from database to verify: %s", err)
log.Printf("[WARNING] Failed getting all workflow apps from database to verify: %s", err)
resp.WriteHeader(401)
resp.Write([]byte(`{"success": false, "reason": "Failed to verify existence"}`))
return
@@ -3123,7 +3123,7 @@ func buildSwaggerApp(resp http.ResponseWriter, body []byte, user shuffle.User) {
err = shuffle.DumpApi(basePath, api)
if err != nil {
log.Printf("Failed dumping yaml: %s", err)
log.Printf("[WARNING] Failed dumping yaml: %s", err)
resp.WriteHeader(500)
resp.Write([]byte(`{"success": false, "reason": "Failed dumping yaml"}`))
return
+1 -2
View File
@@ -39,7 +39,6 @@ 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:nightly
container_name: shuffle-orborus
hostname: shuffle-orborus
@@ -49,7 +48,7 @@ services:
- /var/run/docker.sock:/var/run/docker.sock
environment:
- SHUFFLE_APP_SDK_VERSION=0.8.97
- SHUFFLE_WORKER_VERSION=latest
- SHUFFLE_WORKER_VERSION=0.9.30
- ORG_ID=${ORG_ID}
- ENVIRONMENT_NAME=${ENVIRONMENT_NAME}
- BASE_URL=http://${OUTER_HOSTNAME}:${BACKEND_PORT}
+1 -1
View File
@@ -2320,7 +2320,7 @@ const Admin = (props) => {
:
<Tooltip title={"Go to workflow"} style={{}} aria-label={"Download"}>
<span>
<a style={{textDecoration: "none", color: "#f85a3e"}} href={`/workflows/${file.workflow_id}`} target="_blank">
<a rel="noopener noreferrer" style={{textDecoration: "none", color: "#f85a3e"}} href={`/workflows/${file.workflow_id}`} target="_blank">
<IconButton disabled={file.workflow_id === "global"}>
<OpenInNewIcon style={{color: file.workflow_id !== "global" ? "white" : "grey",}} />
</IconButton>
+2 -1
View File
@@ -1723,6 +1723,7 @@ const AngularWorkflow = (props) => {
}
workflow.start = parentNode.data('id')
setLastSaved(true)
parentNode.data('isStartNode', true)
}
@@ -7296,7 +7297,7 @@ const AngularWorkflow = (props) => {
}
const executionModal =
<Drawer anchor={"right"} open={executionModalOpen} onClose={() => setExecutionModalOpen(false)} style={{resize: "both", overflow: "auto", zIndex: 10005}} PaperProps={{style: {resize: "both", overflow: "auto", minWidth: 400, maxWidth: 400, backgroundColor: "#1F2023", color: "white", fontSize: 18, zIndex: 10005}}}>
<Drawer anchor={"right"} open={executionModalOpen} onClose={() => setExecutionModalOpen(false)} style={{resize: "both", overflow: "auto", zIndex: 10005}} PaperProps={{style: {resize: "both", overflow: "auto", minWidth: 420, maxWidth: 420, backgroundColor: "#1F2023", color: "white", fontSize: 18, zIndex: 10005}}}>
{executionModalView === 0 ?
<div style={{padding: 25, }}>
<Breadcrumbs aria-label="breadcrumb" separator="" style={{color: "white", fontSize: 16}}>
+7 -1
View File
@@ -426,10 +426,16 @@ const AppCreator = (props) => {
return newitem
}
const base64_decode = (str) => {
return decodeURIComponent(atob(str).split('').map(function(c) {
return '%' + ('00' + c.charCodeAt(0).toString(16)).slice(-2);
}).join(''))
}
// Sets the data up as it should be at later points
// This is the data FROM the database, not what's being saved
const parseIncomingOpenapiData = (data) => {
const parsedapp = data.openapi === undefined ? data : JSON.parse(atob(data.openapi))
const parsedapp = data.openapi === undefined ? data : JSON.parse(base64_decode(data.openapi))
data = parsedapp.body === undefined ? parsedapp : parsedapp.body
var jsonvalid = false
+6 -6
View File
@@ -695,7 +695,7 @@ const Apps = (props) => {
</Select>
: null }
{isCloud ?
<a href={"https://shuffler.io/apps/"+selectedApp.id} style={{textDecoration: "none", color: "#f85a3e"}} target="_blank">
<a rel="noopener noreferrer" href={"https://shuffler.io/apps/"+selectedApp.id} style={{textDecoration: "none", color: "#f85a3e"}} target="_blank">
<IconButton style={{top: -10, right: 0, position: "absolute", color: "#f85a3e"}} >
<OpenInNewIcon style={{}} />
</IconButton>
@@ -848,10 +848,10 @@ const Apps = (props) => {
<Paper square style={uploadViewPaperStyle}>
<div style={{width: "100%", margin: 25}}>
<h2>App Creator</h2>
<a href="https://shuffler.io/docs/apps" style={{textDecoration: "none", color: "#f85a3e"}} target="_blank">How it works</a>
&nbsp;- <a href="https://github.com/frikky/security-openapis" style={{textDecoration: "none", color: "#f85a3e"}} target="_blank">Security API's</a>
&nbsp;- <a href="https://github.com/APIs-guru/openapi-directory/tree/main/APIs" style={{textDecoration: "none", color: "#f85a3e"}} target="_blank">OpenAPI directory</a>
&nbsp;- <a href="https://editor.swagger.io/" style={{textDecoration: "none", color: "#f85a3e"}} target="_blank">OpenAPI Validator</a>
<a rel="noopener noreferrer" href="https://shuffler.io/docs/apps" style={{textDecoration: "none", color: "#f85a3e"}} target="_blank">How it works</a>
&nbsp;- <a rel="noopener noreferrer" href="https://github.com/frikky/security-openapis" style={{textDecoration: "none", color: "#f85a3e"}} target="_blank">Security API's</a>
&nbsp;- <a rel="noopener noreferrer" href="https://github.com/APIs-guru/openapi-directory/tree/main/APIs" style={{textDecoration: "none", color: "#f85a3e"}} target="_blank">OpenAPI directory</a>
&nbsp;- <a rel="noopener noreferrer" href="https://editor.swagger.io/" style={{textDecoration: "none", color: "#f85a3e"}} target="_blank">OpenAPI Validator</a>
<div/>
<Typography variant="body2" color="textSecondary">
Apps interact with eachother in workflows. They are created with the app creator, using OpenAPI specification or manually in python. The links above are references to OpenAPI tools and other app repositories. There's thousands of them.
@@ -1061,7 +1061,7 @@ const Apps = (props) => {
<Paper square style={uploadViewPaperStyle}>
<Typography style={{margin: 10, }}>
<span>
<a href={"https://shuffler.io/search"} style={{textDecoration: "none", color: "#f85a3e"}} target="_blank">
<a rel="noopener noreferrer" href={"https://shuffler.io/search"} style={{textDecoration: "none", color: "#f85a3e"}} target="_blank">
Click here
</a> to search ALL apps, not just your activated ones.
</span>
+2 -2
View File
@@ -304,7 +304,7 @@ const Docs = (props) => {
<div style={{flex: 3, display: "flex", vAlign: "center",}}>
{mobile ? null :
<Typography style={{display: "inline", marginTop: 6, }}>
<a rel="norefferer" target="_blank" href={selectedMeta.link} target="_blank" style={{textDecoration: "none", color: "#f85a3e"}}>
<a rel="noopener noreferrer" target="_blank" href={selectedMeta.link} target="_blank" style={{textDecoration: "none", color: "#f85a3e"}}>
<Button style={{}} variant="outlined">
<EditIcon /> &nbsp;&nbsp;Edit
</Button>
@@ -323,7 +323,7 @@ const Docs = (props) => {
<div style={{margin: 10, height: "100%", display: "inline",}}>
{selectedMeta.contributors.slice(0,7).map((data, index) => {
return (
<a rel="norefferer" target="_blank" href={data.url} target="_blank" style={{textDecoration: "none", color: "#f85a3e"}}>
<a rel="noopener noreferrer" target="_blank" href={data.url} target="_blank" style={{textDecoration: "none", color: "#f85a3e"}}>
<Tooltip title={data.url} placement="bottom">
<img alt={data.url} src={data.image} style={{marginTop: 5, marginRight: 10, height: 40, borderRadius: 40, }} />
</Tooltip>
+46 -235
View File
@@ -1,15 +1,15 @@
import React, { useEffect} from 'react';
import { useInterval } from 'react-powerhooks';
import { makeStyles } from '@material-ui/core/styles';
import { useTheme } from '@material-ui/core/styles';
import {Badge, Avatar, Grid, Paper, Tooltip, Divider, Button, TextField, FormControl, IconButton, Menu, MenuItem, FormControlLabel, Chip, Switch, Typography, Zoom, CircularProgress, Dialog, DialogTitle, DialogActions, DialogContent} from '@material-ui/core';
import {GridOn as GridOnIcon, List as ListIcon, Close as CloseIcon, Compare as CompareIcon, Maximize as MaximizeIcon, Minimize as MinimizeIcon, AddCircle as AddCircleIcon, Toc as TocIcon, Send as SendIcon, Search as SearchIcon, FileCopy as FileCopyIcon, Delete as DeleteIcon, BubbleChart as BubbleChartIcon, Restore as RestoreIcon, Cached as CachedIcon, GetApp as GetAppIcon, Apps as AppsIcon, Edit as EditIcon, MoreVert as MoreVertIcon, PlayArrow as PlayArrowIcon, Add as AddIcon, Publish as PublishIcon, CloudUpload as CloudUploadIcon, CloudDownload as CloudDownloadIcon} from '@material-ui/icons';
import NestedMenuItem from "material-ui-nested-menu-item";
//import {Search as SearchIcon, ArrowUpward as ArrowUpwardIcon, Visibility as VisibilityIcon, Done as DoneIcon, Close as CloseIcon, Error as ErrorIcon, FindReplace as FindreplaceIcon, ArrowLeft as ArrowLeftIcon, Cached as CachedIcon, DirectionsRun as DirectionsRunIcon, Add as AddIcon, Polymer as PolymerIcon, FormatListNumbered as FormatListNumberedIcon, Create as CreateIcon, PlayArrow as PlayArrowIcon, AspectRatio as AspectRatioIcon, MoreVert as MoreVertIcon, Apps as AppsIcon, Schedule as ScheduleIcon, FavoriteBorder as FavoriteBorderIcon, Pause as PauseIcon, Delete as DeleteIcon, AddCircleOutline as AddCircleOutlineIcon, Save as SaveIcon, KeyboardArrowLeft as KeyboardArrowLeftIcon, KeyboardArrowRight as KeyboardArrowRightIcon, ArrowBack as ArrowBackIcon, Settings as SettingsIcon, LockOpen as LockOpenIcon, ExpandMore as ExpandMoreIcon, VpnKey as VpnKeyIcon} from '@material-ui/icons';
//https://next.material-ui.com/components/material-icons/
import {DataGrid, GridToolbarContainer, GridDensitySelector, GridToolbar} from '@material-ui/data-grid';
import {DataGrid, GridToolbar} from '@material-ui/data-grid';
//import JSONPretty from 'react-json-pretty';
//import JSONPrettyMon from 'react-json-pretty/dist/monikai'
@@ -19,33 +19,14 @@ import {Link} from 'react-router-dom';
import { useAlert } from "react-alert";
import ChipInput from 'material-ui-chip-input'
import { v4 as uuidv4 } from 'uuid';
import CytoscapeWrapper from '../components/RenderCytoscape'
//import mobileImage from '../assets/img/mobile.svg';
//import bagImage from '../assets/img/bag.svg';
//import bookImage from '../assets/img/book.svg';
const inputColor = "#383B40"
const surfaceColor = "#27292D"
const svgSize = 24
const imagesize = 22
const flexContainerStyle = {
display: "flex",
flexDirection: "row",
justifyContent: "left",
alignContent: "space-between",
}
const flexBoxStyle = {
height: 125,
borderRadius: 4,
boxSizing: "border-box",
letterSpacing: "0.4px",
color: "#D6791E",
margin: 10,
flex: 1,
}
const useStyles = makeStyles((theme) => ({
datagrid: {
@@ -258,42 +239,22 @@ export const GetIconInfo = (action) => {
console.log(`MISSING PATH FOR ${selectedKey} (find in scope): `, selectedItem.originalIcon.type.type)
}
if (selectedItem.originalIcon === undefined || selectedItem.originalIcon === "" && (selectedItem.icon !== "" && selectedItem.icon !== undefined)) {
if ((selectedItem.originalIcon === undefined || selectedItem.originalIcon === "") && (selectedItem.icon !== "" && selectedItem.icon !== undefined)) {
const svg_pin = <svg width={svgSize} height={svgSize} viewBox={`0 0 ${svgSize} ${svgSize}`} version="1.1" xmlns="http://www.w3.org/2000/svg"><path d={selectedItem.icon} fill={selectedItem.iconColor}></path></svg>
//const svgpin_Url = encodeURI("data:image/svg+xml;utf-8," + svg_pin)
selectedItem.originalIcon = svg_pin
//parsedIcons[selectedKey].originalIcon = svg_pin
//console.log("ADDED ICON: ", svg_pin)
}
return selectedItem
}
//const activeWorkflowStyle = {backgroundColor: "#FFF5EE"}
//const notificationStyle = {backgroundColor: "#E5F9FF"}
//const activeWorkflowStyle = {backgroundColor: "#3d3f43"}
const availableWorkflowStyle = {backgroundColor: "#3d3f43"}
const notificationStyle = {backgroundColor: "#3d3f43"}
const activeWorkflowStyle = {backgroundColor: "#3d3f43"}
const fontSize_16 = {fontSize: "16px",}
const counterStyle = {fontSize: "36px",fontWeight:"bold"}
const blockRightStyle = {textAlign: "right",padding: "20px 20px 0px 0px",width:"100%"}
const chipStyle = {
backgroundColor: "#3d3f43", height: 30, marginRight: 5, paddingLeft: 5, paddingRight: 5, height: 28, cursor: "pointer", borderColor: "#3d3f43", color: "white",
backgroundColor: "#3d3f43", marginRight: 5, paddingLeft: 5, paddingRight: 5, height: 28, cursor: "pointer", borderColor: "#3d3f43", color: "white",
}
const flexContentStyle = {
display: "flex",
flexDirection: "row"
}
const iconStyle = {
width: "75px",
height: "75px",
padding: "20px"
}
export const validateJson = (showResult) => {
//showResult = showResult.split(" None").join(" \"None\"")
@@ -302,15 +263,13 @@ export const validateJson = (showResult) => {
var jsonvalid = true
try {
const tmp = String(JSON.parse(showResult))
if (!showResult.includes("{") && !showResult.includes("[")) {
jsonvalid = false
}
} catch (e) {
showResult = showResult.split("\'").join("\"")
showResult = showResult.split("'").join("\"")
try {
const tmp = String(JSON.parse(showResult))
if (!showResult.includes("{") && !showResult.includes("[")) {
jsonvalid = false
}
@@ -328,7 +287,7 @@ export const validateJson = (showResult) => {
}
const Workflows = (props) => {
const { globalUrl, isLoggedIn, isLoaded, removeCookie, cookies, userdata} = props;
const { globalUrl, isLoggedIn, isLoaded, userdata} = props;
document.title = "Shuffle - Workflows"
const theme = useTheme();
const alert = useAlert()
@@ -338,19 +297,14 @@ const Workflows = (props) => {
const referenceUrl = globalUrl+"/api/v1/hooks/"
var upload = ""
const [file, setFile] = React.useState("");
const [workflows, setWorkflows] = React.useState([]);
const [filteredWorkflows, setFilteredWorkflows] = React.useState([]);
const [selectedWorkflow, setSelectedWorkflow] = React.useState({});
const [selectedExecution, setSelectedExecution] = React.useState({});
const [workflowExecutions, setWorkflowExecutions] = React.useState([]);
const [firstrequest, setFirstrequest] = React.useState(true)
const [workflowDone, setWorkflowDone] = React.useState(false)
const [, setTrackingId] = React.useState("")
const [selectedWorkflowId, setSelectedWorkflowId] = React.useState("")
const [collapseJson, setCollapseJson] = React.useState(false)
const [field1, setField1] = React.useState("")
const [field2, setField2] = React.useState("")
const [downloadUrl, setDownloadUrl] = React.useState("https://github.com/frikky/shuffle-workflows")
@@ -364,14 +318,11 @@ const Workflows = (props) => {
const [newWorkflowDescription, setNewWorkflowDescription] = React.useState("");
const [newWorkflowTags, setNewWorkflowTags] = React.useState([]);
const [showExtraOptions, setShowExtraOptions] = React.useState(true);
const [defaultReturnValue, setDefaultReturnValue] = React.useState("");
const [update, setUpdate] = React.useState("test");
const [deleteModalOpen, setDeleteModalOpen] = React.useState(false);
const [publishModalOpen, setPublishModalOpen] = React.useState(false);
const [editingWorkflow, setEditingWorkflow] = React.useState({})
const [executionLoading, setExecutionLoading] = React.useState(false)
const [importLoading, setImportLoading] = React.useState(false)
const [isDropzone, setIsDropzone] = React.useState(false);
const [view, setView] = React.useState("grid")
@@ -422,8 +373,6 @@ const Workflows = (props) => {
})
}
//console.log("FOUND: ", found)
//if (found) {
if (found.every(v => v === true)) {
newWorkflows.push(curWorkflow)
continue
@@ -463,9 +412,7 @@ const Workflows = (props) => {
}
//console.log("Removing filter index", index)
newfilters.splice(index, 1)
//console.log("FILTER LENGTH: ", filters.length)
if (newfilters.length === 0) {
newfilters = []
@@ -473,7 +420,6 @@ const Workflows = (props) => {
} else {
setFilters(newfilters)
}
//console.log("FILTERS: ", newfilters)
findWorkflow(newfilters)
}
@@ -625,14 +571,14 @@ const Workflows = (props) => {
}
// Initialize the workflow itself
const ret = setNewWorkflow(data.name, data.description, data.tags, data.default_return_value, {}, false)
setNewWorkflow(data.name, data.description, data.tags, data.default_return_value, {}, false)
.then((response) => {
if (response !== undefined) {
// SET THE FULL THING
data.id = response.id
// Actually create it
const ret = setNewWorkflow(data.name, data.description, data.tags, data.default_return_value, data, false)
setNewWorkflow(data.name, data.description, data.tags, data.default_return_value, data, false)
.then((response) => {
if (response !== undefined) {
alert.success(`Successfully imported ${data.name}`)
@@ -653,10 +599,10 @@ const Workflows = (props) => {
useEffect(() => {
if (isDropzone) {
//redirectOpenApi();
setIsDropzone(false);
}
}, [isDropzone]);
const getAvailableWorkflows = () => {
fetch(globalUrl+"/api/v1/workflows", {
@@ -683,9 +629,6 @@ const Workflows = (props) => {
return response.json()
})
.then((responseJson) => {
setSelectedExecution({})
setWorkflowExecutions([])
//console.log(responseJson)
if (responseJson !== undefined) {
setWorkflows(responseJson)
@@ -720,16 +663,14 @@ const Workflows = (props) => {
return
}
if (responseJson.length > 0){
//setSelectedWorkflow(responseJson[0])
//getWorkflowExecution(responseJson[0].id)
}
})
.catch(error => {
alert.error(error.toString())
});
}
// eslint-disable-next-line react-hooks/exhaustive-deps
useEffect(() => {
if (workflows.length <= 0 && firstrequest) {
const tmpView = localStorage.getItem('view');
@@ -740,7 +681,7 @@ const Workflows = (props) => {
setFirstrequest(false)
getAvailableWorkflows()
}
})
})
const viewStyle = {
color: "#ffffff",
@@ -749,7 +690,6 @@ const Workflows = (props) => {
minWidth: 1024,
maxWidth: 1024,
margin: "auto",
/*maxHeight: "90vh",*/
}
const emptyWorkflowStyle = {
@@ -769,14 +709,6 @@ const Workflows = (props) => {
}
const scrollStyle = {
marginTop: "10px",
overflow: "scroll",
height: "90%",
overflowX: "hidden",
overflowY: "auto",
}
const paperAppContainer = {
display: "flex",
flexWrap: 'wrap',
@@ -813,38 +745,6 @@ const Workflows = (props) => {
justifyContent: "space-between",
}
const executeWorkflow = (id) => {
alert.show("Executing workflow "+id)
setTrackingId(id)
fetch(globalUrl+"/api/v1/workflows/"+id+"/execute", {
method: 'GET',
headers: {
'Content-Type': 'application/json',
'Accept': 'application/json',
},
credentials: "include",
})
.then((response) => {
if (response.status !== 200) {
console.log("Status not 200 for WORKFLOW EXECUTION :O!")
}
return response.json()
})
.then((responseJson) => {
if (!responseJson.success) {
alert.error(responseJson.reason)
}
})
.catch(error => {
alert.error(error.toString())
});
}
function sleep (time) {
return new Promise((resolve) => setTimeout(resolve, time));
}
const exportAllWorkflows = () => {
for (var key in workflows) {
exportWorkflow(workflows[key], false)
@@ -861,13 +761,12 @@ const Workflows = (props) => {
}
}
if (trigger.status == "running") {
if (trigger.status === "running") {
trigger.status = "stopped"
}
const newId = uuidv4()
if (trigger.trigger_type === "WEBHOOK") {
const hookname = "webhook_"+newId
if (trigger.parameters !== undefined && trigger.parameters !== null && trigger.parameters.length === 2) {
trigger.parameters[0].value = referenceUrl+"webhook_"+trigger.id
trigger.parameters[1].value = "webhook_"+trigger.id
@@ -897,7 +796,7 @@ const Workflows = (props) => {
}
if (data.actions !== null && data.actions !== undefined) {
for (var key in data.actions) {
for (key in data.actions) {
data.actions[key].authentication_id = ""
for (var subkey in data.actions[key].parameters) {
@@ -914,7 +813,7 @@ const Workflows = (props) => {
}
const newId = uuidv4()
for (var branchkey in data.branches) {
for ( branchkey in data.branches) {
const branch = data.branches[branchkey]
if (branch.source_id === data.actions[key].id) {
branch.source_id = newId
@@ -929,14 +828,13 @@ const Workflows = (props) => {
data.start = newId
}
//data.actions[key].environment = isCloud ? "cloud" : "Shuffle"
data.actions[key].environment = ""
data.actions[key].id = newId
}
}
if (data.workflow_variables !== null && data.workflow_variables !== undefined) {
for (var key in data.workflow_variables) {
for (key in data.workflow_variables) {
const param = data.workflow_variables[key]
if (param.name.includes("key") || param.name.includes("user") || param.name.includes("pass") || param.name.includes("api") || param.name.includes("auth") || param.name.includes("secret") || param.name.includes("email")) {
param.value = ""
@@ -983,7 +881,6 @@ const Workflows = (props) => {
// }
//}
}
//return
// Add correct ID's for triggers
// Add mag
@@ -1042,8 +939,7 @@ const Workflows = (props) => {
data.id = ""
data.name = data.name+"_copy"
data = deduplicateIds(data)
//console.log("COPIED DATA: ", data)
//return
fetch(globalUrl+"/api/v1/workflows", {
method: 'POST',
@@ -1061,7 +957,7 @@ const Workflows = (props) => {
}
return response.json()
})
.then((responseJson) => {
.then(() => {
setTimeout(() => {
getAvailableWorkflows()
}, 1000)
@@ -1091,7 +987,7 @@ const Workflows = (props) => {
return response.json()
})
.then((responseJson) => {
.then(() => {
setTimeout(() => {
getAvailableWorkflows()
}, 1000)
@@ -1107,7 +1003,7 @@ const Workflows = (props) => {
}
const NewWorkflowPaper = (props) => {
const NewWorkflowPaper = () => {
const [hover, setHover] = React.useState(false)
const innerColor = "rgba(255,255,255,0.3)"
@@ -1143,10 +1039,7 @@ const Workflows = (props) => {
const [open, setOpen] = React.useState(false);
const [anchorEl, setAnchorEl] = React.useState(null);
var boxWidth = "2px"
if (selectedWorkflow.id === data.id) {
boxWidth = "4px"
}
var boxColor = "#FECC00"
if (data.is_valid) {
@@ -1168,8 +1061,7 @@ const Workflows = (props) => {
}
const actions = data.actions !== null ? data.actions.length : 0
const [triggers, schedules, webhooks, subflows] = getWorkflowMeta(data)
const [triggers, subflows] = getWorkflowMeta(data)
const workflowMenuButtons = <Menu
id="long-menu"
@@ -1182,7 +1074,6 @@ const Workflows = (props) => {
}}
>
<MenuItem style={{backgroundColor: inputColor, color: "white"}} onClick={() => {
//console.log("DATA:" ,data)
setModalOpen(true)
setEditingWorkflow(data)
setNewWorkflowName(data.name)
@@ -1271,14 +1162,20 @@ const Workflows = (props) => {
const foundOrg = userdata.orgs.find(org => org.id === data["org_id"])
if (foundOrg !== undefined && foundOrg !== null) {
//position: "absolute", bottom: 5, right: -5,
const imageStyle = {width: imagesize, height: imagesize, pointerEvents: "none", marginRight: 10, marginLeft: data.creator_org !== undefined && data.creator_org.length > 0 ? 20 : 0, borderRadius: 10, border: foundOrg.id === userdata.active_org.id ? `3px solid ${boxColor}` : null, cursor: "pointer", marginRight: 10, }
const imageStyle = {
width: imagesize,
height: imagesize,
pointerEvents: "none",
marginLeft: data.creator_org !== undefined && data.creator_org.length > 0 ? 20 : 0,
borderRadius: 10,
border: foundOrg.id === userdata.active_org.id ? `3px solid ${boxColor}` : null,
cursor: "pointer",
marginRight: 10 }
//<Tooltip title={`Org: ${foundOrg.name}`} placement="bottom">
image = foundOrg.image === "" ?
<img alt={foundOrg.name} src={theme.palette.defaultImage} style={imageStyle} />
:
<img alt={foundOrg.name} src={foundOrg.image} style={imageStyle} onClick={() => {
//setFilteredWorkflows(newWorkflows)
}}/>
orgName = foundOrg.name
@@ -1295,8 +1192,6 @@ const Workflows = (props) => {
<Tooltip title={`Org "${orgName}"`} placement="bottom">
<div styl={{cursor: "pointer"}} onClick={() => {
addFilter(orgId)
//setFilters(["Org "+orgName])
//setFilteredWorkflows(newWorkflows)
}}>
{image}
</div>
@@ -1360,20 +1255,6 @@ const Workflows = (props) => {
</Typography>
</span>
</Tooltip>
{/*
<Tooltip color="primary" title={`Actions: ${data.actions.length}`} placement="bottom">
<AppsIcon />
</Tooltip>
<Tooltip color="primary" title={`Webhooks: ${webhooks}`} placement="bottom">
<RestoreIcon />
</Tooltip>
: null}
{schedules > 0 ?
<Tooltip color="primary" title={`Schedules: ${schedules}`} placement="bottom">
<RestoreIcon />
</Tooltip>
: null}
*/}
</Grid>
<Grid item style={{justifyContent: "left", overflow: "hidden", marginTop: 5,}}>
{data.tags !== undefined ?
@@ -1404,23 +1285,13 @@ const Workflows = (props) => {
aria-label="more"
aria-controls="long-menu"
aria-haspopup="true"
style={{color: "white"}}
onClick={menuClick}
style={{padding:"0px",color:"white", color:"#979797"}}
style={{padding:"0px", color:"#979797"}}
>
<MoreVertIcon />
</IconButton>
{workflowMenuButtons}
</Grid>
{/*
<Grid>
<Link to={"/workflows/"+data.id}>
<Tooltip title="Edit workflow" placement="bottom">
<EditIcon style={{borderRadius: "4px", color: "#F85A3E", height: 20, width: 20, padding: 7, fontSize: "small"}} />
</Tooltip>
</Link>
</Grid>
*/}
</Grid>
: null}
</Paper>
@@ -1430,29 +1301,6 @@ const Workflows = (props) => {
const dividerColor = "rgb(225, 228, 232)"
const resultPaperAppStyle = {
minHeight: "100px",
minWidth: "100%",
overflow: "hidden",
maxWidth: "100%",
marginTop: "5px",
color: "white",
backgroundColor: surfaceColor,
cursor: "pointer",
display: "flex",
}
function replaceAll(string, search, replace) {
return string.split(search).join(replace);
}
const resultsLength = Object.getOwnPropertyNames(selectedExecution).length > 0 && selectedExecution.results !== null ? selectedExecution.results.length : 0
// Can create and set workflows
const setNewWorkflow = (name, description, tags, defaultReturnValue, editingWorkflow, redirect) => {
@@ -1480,11 +1328,9 @@ const Workflows = (props) => {
if (defaultReturnValue !== undefined) {
workflowdata["default_return_value"] = defaultReturnValue
//console.log("WORKFLOW: ", workflowdata)
}
//console.log(workflowdata)
//return
return fetch(globalUrl+"/api/v1/workflows"+extraData, {
method: method,
@@ -1535,7 +1381,6 @@ const Workflows = (props) => {
console.log("Importing!")
setImportLoading(true)
const file = event.target.value
if (event.target.files.length > 0) {
for (var key in event.target.files) {
const file = event.target.files[key]
@@ -1560,7 +1405,7 @@ const Workflows = (props) => {
}
// Initialize the workflow itself
const ret = setNewWorkflow(data.name, data.description, data.tags, data.default_return_value, {}, false)
setNewWorkflow(data.name, data.description, data.tags, data.default_return_value, {}, false)
.then((response) => {
if (response !== undefined) {
// SET THE FULL THING
@@ -1570,7 +1415,7 @@ const Workflows = (props) => {
data.is_valid = false
// Actually create it
const ret = setNewWorkflow(data.name, data.description, data.tags, data.default_return_value, data, false)
setNewWorkflow(data.name, data.description, data.tags, data.default_return_value, data, false)
.then((response) => {
if (response !== undefined) {
alert.success("Successfully imported "+data.name)
@@ -1593,26 +1438,17 @@ const Workflows = (props) => {
const getWorkflowMeta = (data) => {
let triggers = 0
let schedules = 0
let webhooks = 0
let subflows = 0
if (data.triggers !== undefined && data.triggers !== null && data.triggers.length > 0) {
triggers = data.triggers.length
for (let key in data.triggers) {
if (data.triggers[key].app_name === "Webhook") {
webhooks += 1
//webhookImg = data.triggers[key].large_image
} else if (data.triggers[key].app_name === "Schedule") {
schedules += 1
//scheduleImg = data.triggers[key].large_image
} else if (data.triggers[key].app_name === "Shuffle Workflow") {
if (data.triggers[key].app_name === "Shuffle Workflow") {
subflows += 1
}
}
}
return [triggers, schedules, webhooks, subflows]
return [triggers, subflows]
}
const WorkflowListView = () => {
@@ -1632,8 +1468,6 @@ const Workflows = (props) => {
}
var image = ""
var orgName = ""
var orgId = ""
if (userdata.orgs !== undefined) {
const foundOrg = userdata.orgs.find(org => org.id === data["org_id"])
if (foundOrg !== undefined && foundOrg !== null) {
@@ -1648,17 +1482,11 @@ const Workflows = (props) => {
//setFilteredWorkflows(newWorkflows)
}}/>
orgName = foundOrg.name
orgId = foundOrg.id
}
}
return (
<div styl={{cursor: "pointer"}} onClick={() => {
//addFilter(orgId)
//setFilters(["Org "+orgName])
//setFilteredWorkflows(newWorkflows)
}}>
<div styl={{cursor: "pointer"}} >
{image}
</div>
)
@@ -1676,21 +1504,13 @@ const Workflows = (props) => {
</Grid>
)
}},
{/* field: 'category', headerName: 'category', width: 140, renderCell: (params) => {
const data = params.row.record
const category = data.category === undefined ? "not defined" : data.category
return (
<Typography>
{category}
</Typography>
)
} */},
{ field: 'options', headerName: 'Options', width: 200, sortable: false,
disableClickEventBubbling: true,
renderCell: (params) => {
const data = params.row.record;
const actions = data.actions !== null ? data.actions.length : 0
let [triggers, schedules, webhooks, subflows] = getWorkflowMeta(data);
let [triggers, subflows] = getWorkflowMeta(data);
return (
<Grid item>
@@ -1889,10 +1709,9 @@ const Workflows = (props) => {
onDelete={(chip, index) => {
newWorkflowTags.splice(index, 1)
setNewWorkflowTags(newWorkflowTags)
setUpdate("delete "+chip)
}}
/>
{showExtraOptions ?
<TextField
onBlur={(event) => setDefaultReturnValue(event.target.value)}
InputProps={{
@@ -1908,7 +1727,7 @@ const Workflows = (props) => {
margin="dense"
fullWidth
/>
: null}
</DialogContent>
<DialogActions>
<Button style={{}} onClick={() => {
@@ -1967,11 +1786,7 @@ const Workflows = (props) => {
const workflowButtons =
<span>
{/*workflows.length > 0 ?
<Tooltip color="primary" title={"Create new workflow"} placement="top">
<Button color="primary" style={{}} variant="text" onClick={() => setModalOpen(true)}><AddIcon /></Button>
</Tooltip>
: null*/}
{view === "list" && (
<Tooltip color="primary" title={"Grid View"} placement="top">
<Button color="primary" variant="text" onClick={() => {
@@ -2056,7 +1871,6 @@ const Workflows = (props) => {
<h2>Workflows</h2>
</div>
</div>
{/*
<div style={flexContainerStyle}>
<div style={{...flexBoxStyle, ...activeWorkflowStyle}}>
@@ -2107,7 +1921,7 @@ const Workflows = (props) => {
<div style={{display: "flex", margin: "0px 0px 20px 0px"}}>
<div style={{flex: 1}}>
<Typography style={{marginTop: 7, marginBottom: "auto"}}>
<a rel="norefferer" target="_blank" href="https://shuffler.io/docs/workflows" target="_blank" style={{textDecoration: "none", color: "#f85a3e"}}>Learn more about Workflows</a>
<a rel="noopener noreferrer" href="https://shuffler.io/docs/workflows" target="_blank" style={{textDecoration: "none", color: "#f85a3e"}}>Learn more about Workflows</a>
</Typography>
</div>
<div style={{flex: 1, float: "right",}}>
@@ -2116,7 +1930,6 @@ const Workflows = (props) => {
InputProps={{
style:{
color: "white",
//backgroundColor: inputColor,
},
}}
placeholder="Add Filter"
@@ -2126,7 +1939,7 @@ const Workflows = (props) => {
onAdd={(chip) => {
addFilter(chip)
}}
onDelete={(chip, index) => {
onDelete={(_, index) => {
removeFilter(index)
}}
/>
@@ -2202,7 +2015,6 @@ const Workflows = (props) => {
}
alert.success("Getting specific workflows from your URL.")
var cors = "cors"
fetch(globalUrl+"/api/v1/workflows/download_remote", {
method: "POST",
mode: "cors",
@@ -2223,7 +2035,6 @@ const Workflows = (props) => {
return response.json()
})
.then((responseJson) => {
//console.log("DATA: ", responseJson)
if (!responseJson.success) {
if (responseJson.reason !== undefined) {
alert.error("Failed loading: "+responseJson.reason)
+1 -1
View File
@@ -11,7 +11,7 @@ RUN go get github.com/docker/docker/api/types && \
go get github.com/mackerelio/go-osstat/cpu && \
go get github.com/mackerelio/go-osstat/memory && \
go get github.com/satori/go.uuid && \
go get github.com/frikky/shuffle-shared
go get github.com/shuffle/shuffle-shared
RUN go build
RUN CGO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo -o orborus .
+1 -1
View File
@@ -1,5 +1,5 @@
NAME=shuffle-orborus
VERSION=0.9.28
VERSION=0.9.30
echo "Running docker build with $NAME:$VERSION"
#docker rmi frikky/shuffle:$NAME --force
+2 -3
View File
@@ -4,10 +4,9 @@ go 1.15
require (
github.com/containerd/containerd v1.5.7 // indirect
github.com/docker/docker v20.10.9+incompatible
github.com/docker/docker v20.10.10+incompatible
github.com/docker/go-connections v0.4.0 // indirect
github.com/mackerelio/go-osstat v0.2.1
github.com/satori/go.uuid v1.2.0
github.com/shuffle/shuffle-shared v0.1.19
go4.org v0.0.0-20201209231011-d4a079459e60 // indirect
github.com/shuffle/shuffle-shared v0.1.26
)
File diff suppressed because it is too large Load Diff
+178 -79
View File
@@ -12,6 +12,7 @@ import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"io/ioutil"
@@ -26,6 +27,7 @@ import (
"github.com/docker/docker/api/types"
"github.com/docker/docker/api/types/container"
"github.com/docker/docker/api/types/mount"
//"github.com/docker/docker/api/types/network"
"github.com/docker/docker/api/types/swarm"
//"github.com/docker/docker/api/types/filters"
dockerclient "github.com/docker/docker/client"
@@ -106,11 +108,13 @@ func getThisContainerId() {
out, err := exec.Command("bash", "-c", cmd).Output()
if err == nil {
containerId = strings.TrimSpace(string(out))
log.Printf("[DEBUG] Set containerId network to %s", containerId)
// cgroup error. Use fallback strategy below.
// https://github.com/moby/moby/issues/7015
//log.Printf("Checking if %s is in %s", ".scope", string(out))
if strings.Contains(string(out), ".scope") {
log.Printf("[DEBUG] ContainerId contains scope. setting to empty.")
containerId = ""
//docker-76c537e9a4b7c7233011f5d70e6b7f2d600b6413ac58a96519b8dca7a3f7117a.scope
}
@@ -132,49 +136,64 @@ func getThisContainerId() {
log.Printf(`[INFO] Started with containerId "%s"`, containerId)
}
// Deploys the internal worker whenever something happens
// https://docs.docker.com/engine/api/sdk/examples/
func deployWorker(image string, identifier string, env []string, executionRequest shuffle.ExecutionRequest) {
// Binds is the actual "-v" volume.
// Max 20% CPU every second
//CPUQuota: 25000,
//CPUPeriod: 100000,
//CPUShares: 256,
hostConfig := &container.HostConfig{
LogConfig: container.LogConfig{
Type: "json-file",
Config: map[string]string{},
},
Resources: container.Resources{},
Binds: []string{
"/var/run/docker.sock:/var/run/docker.sock:rw",
},
NetworkMode: container.NetworkMode(fmt.Sprintf("container:%s", containerId)),
}
if cleanupEnv == "true" {
hostConfig.AutoRemove = true
}
config := &container.Config{
Image: image,
Env: env,
}
//var swarmConfig = os.Getenv("SHUFFLE_SWARM_CONFIG")
parsedUuid := uuid.NewV4()
func deployServiceWorkers(image string) {
log.Printf("[DEBUG] Validating deployment of workers as services IF swarmConfig = run (value: %#v)", swarmConfig)
if swarmConfig == "run" {
// frikky@debian:~/git/shuffle/functions/onprem/worker$ docker service create --replicas 5 --name shuffle-workers --env SHUFFLE_SWARM_CONFIG=run --publish published=33333,target=33333 ghcr.io/frikky/shuffle-worker:nightly
networkName := "shuffle-executions"
ctx := context.Background()
//docker network create --driver=overlay workers
networkCreateOptions := types.NetworkCreate{
Driver: "overlay",
}
_, err := dockercli.NetworkCreate(
ctx,
networkName,
networkCreateOptions,
)
if err != nil {
if strings.Contains(fmt.Sprintf("%s", err), "already exists") {
} else {
log.Printf("[DEBUG] Failed to create network %s for workers: %s. This is not critical, and containers will still be added", networkName, err)
}
}
//serviceOptions := types.ServiceCreateOptions{}
//service, err := dockercli.ServiceCreate(
// context.Background(),
// serviceSpec,
// serviceOptions,
//)
log.Printf("[DEBUG] Deploying containers with swarm")
//containerName := fmt.Sprintf("shuffle-worker-%s", parsedUuid)
containerName := fmt.Sprintf("shuffle-workers")
replicas := uint64(2)
scaleReplicas := os.Getenv("SHUFFLE_SCALE_REPLICAS")
if len(scaleReplicas) > 0 {
log.Printf("[DEBUG] SHUFFLE_SCALE_REPLICAS set to value %#v. Trying to overwrite default (2/node)", scaleReplicas)
tmpInt, err := strconv.Atoi(scaleReplicas)
if err != nil {
log.Printf("[ERROR] %s is not a valid number for replication", scaleReplicas)
} else {
replicas = uint64(tmpInt)
}
}
innerContainerName := fmt.Sprintf("shuffle-workers")
log.Printf("[DEBUG] Deploying %d containers for worker with swarm to each node. Service name: %s. Image: %s", replicas, innerContainerName, image)
serviceSpec := swarm.ServiceSpec{
Annotations: swarm.Annotations{
Name: containerName,
Name: innerContainerName,
Labels: map[string]string{},
},
Networks: []swarm.NetworkAttachmentConfig{
swarm.NetworkAttachmentConfig{
Target: networkName,
},
},
EndpointSpec: &swarm.EndpointSpec{
Ports: []swarm.PortConfig{
swarm.PortConfig{
@@ -207,7 +226,7 @@ func deployWorker(image string, identifier string, env []string, executionReques
Condition: swarm.RestartPolicyConditionNone,
},
Placement: &swarm.Placement{
MaxReplicas: 1,
MaxReplicas: replicas,
},
},
}
@@ -216,27 +235,80 @@ func deployWorker(image string, identifier string, env []string, executionReques
serviceSpec.TaskTemplate.ContainerSpec.Env = append(serviceSpec.TaskTemplate.ContainerSpec.Env, fmt.Sprintf("DOCKER_API_VERSION=%s", dockerApiVersion))
}
if len(os.Getenv("SHUFFLE_SCALE_REPLICAS")) > 0 {
serviceSpec.TaskTemplate.ContainerSpec.Env = append(serviceSpec.TaskTemplate.ContainerSpec.Env, fmt.Sprintf("DOCKER_API_VERSION=%s", os.Getenv("SHUFFLE_SCALE_REPLICAS")))
}
serviceOptions := types.ServiceCreateOptions{}
service, err := dockercli.ServiceCreate(
context.Background(),
_, err = dockercli.ServiceCreate(
ctx,
serviceSpec,
serviceOptions,
)
if err == nil {
log.Printf("[DEBUG] Waiting 10 seconds for workers to come awake")
time.Sleep(time.Duration(10) * time.Second)
}
log.Printf("Servicecreate request: %#v %#v", service, err)
err = sendWorkerRequest(executionRequest)
if err != nil {
log.Printf("[ERROR] Failed worker request: %s", err)
log.Printf("[DEBUG] Successfully deployed workers with %d replicas", replicas)
//time.Sleep(time.Duration(10) * time.Second)
//log.Printf("[DEBUG] Servicecreate request: %#v %#v", service, err)
} else {
log.Printf("[DEBUG] Started worker from request: %s - %#v - %s", containerName, service, err)
if !strings.Contains(fmt.Sprintf("%s", err), "Already Exists") && !strings.Contains(fmt.Sprintf("%s", err), "is already in use by service") {
log.Printf("[ERROR] Failed making service: %s", err)
}
}
return
}
}
// Deploys the internal worker whenever something happens
// https://docs.docker.com/engine/api/sdk/examples/
func deployWorker(image string, identifier string, env []string, executionRequest shuffle.ExecutionRequest) error {
// Binds is the actual "-v" volume.
// Max 20% CPU every second
//CPUQuota: 25000,
//CPUPeriod: 100000,
//CPUShares: 256,
hostConfig := &container.HostConfig{
LogConfig: container.LogConfig{
Type: "json-file",
Config: map[string]string{},
},
Resources: container.Resources{},
Binds: []string{
"/var/run/docker.sock:/var/run/docker.sock:rw",
},
}
hostConfig.NetworkMode = container.NetworkMode(fmt.Sprintf("container:%s", containerId))
if cleanupEnv == "true" {
hostConfig.AutoRemove = true
}
config := &container.Config{
Image: image,
Env: env,
}
//var swarmConfig = os.Getenv("SHUFFLE_SWARM_CONFIG")
parsedUuid := uuid.NewV4()
if swarmConfig == "run" {
go func() {
err := sendWorkerRequest(executionRequest)
if err != nil {
log.Printf("[ERROR] Failed worker request for %s: %s", executionRequest.ExecutionId, err)
if strings.Contains(fmt.Sprintf("%s", err), "connection refused") || strings.Contains(fmt.Sprintf("%s", err), "EOF") {
workerImage := fmt.Sprintf("%s/%s/shuffle-worker:%s", baseimageregistry, baseimagename, workerVersion)
deployServiceWorkers(workerImage)
}
//return err
} else {
log.Printf("[DEBUG] Started worker from request with name: %s", executionRequest.ExecutionId)
}
}()
return nil
}
//log.Printf("[INFO] Identifier: %s", identifier)
@@ -264,19 +336,40 @@ func deployWorker(image string, identifier string, env []string, executionReques
if err != nil {
log.Printf("[ERROR] Container create error(2): %s", err)
return
return err
}
} else {
log.Printf("[ERROR] Container create error: %s", err)
return
return err
}
}
containerStartOptions := types.ContainerStartOptions{}
err = dockercli.ContainerStart(context.Background(), cont.ID, containerStartOptions)
if err != nil {
log.Printf("[ERROR] Failed to start container in environment %s: %s", environment, err)
return
log.Printf("[DEBUG] Failed initial container start. Running WITHOUT custom network. Err: %s", err)
// Trying to recreate and start WITHOUT network if it's possible. No extended checks. Old execution system (<0.9.30)
if strings.Contains(fmt.Sprintf("%s", err), "cannot join network") || strings.Contains(fmt.Sprintf("%s", err), "No such container") {
hostConfig.NetworkMode = ""
//container.NetworkMode(fmt.Sprintf("container:%s", containerId))
cont, err = dockercli.ContainerCreate(
context.Background(),
config,
hostConfig,
nil,
nil,
identifier+"-2",
)
err = dockercli.ContainerStart(context.Background(), cont.ID, containerStartOptions)
}
if err != nil {
log.Printf("[ERROR] Failed to start container in environment %s: %s", environment, err)
return err
} else {
log.Printf("[INFO] Container %s was created under environment %s", cont.ID, environment)
}
//stats, err := cli.ContainerInspect(context.Background(), containerName)
//if err != nil {
@@ -303,7 +396,7 @@ func deployWorker(image string, identifier string, env []string, executionReques
log.Printf("[INFO] Container %s was created under environment %s", cont.ID, environment)
}
return
return nil
}
func stopWorker(containername string) error {
@@ -336,6 +429,7 @@ func initializeImages() {
appSdkVersion = "0.8.97"
log.Printf("[WARNING] SHUFFLE_APP_SDK_VERSION not defined. Defaulting to %s", appSdkVersion)
}
if workerVersion == "" {
workerVersion = "nightly"
log.Printf("[WARNING] SHUFFLE_WORKER_VERSION not defined. Defaulting to %s", workerVersion)
@@ -486,6 +580,8 @@ func main() {
//workerImage := fmt.Sprintf("%s/%s:worker%s", baseimageregistry, baseimagename, baseimagetagsuffix)
workerImage := fmt.Sprintf("%s/%s/shuffle-worker:%s", baseimageregistry, baseimagename, workerVersion)
go deployServiceWorkers(workerImage)
log.Printf("[INFO] Finished configuring docker environment")
// FIXME - time limit
@@ -638,7 +734,6 @@ func main() {
continue
} else {
//log.Printf("[INFO] Adding to be ran %s", execution.ExecutionId)
executionIds = append(executionIds, execution.ExecutionId)
}
// Now, how do I execute this one?
@@ -665,11 +760,15 @@ func main() {
env = append(env, fmt.Sprintf("DOCKER_API_VERSION=%s", dockerApiVersion))
}
go deployWorker(workerImage, containerName, env, execution)
log.Printf("[INFO] ExecutionID %s was deployed and to be removed from queue.", execution.ExecutionId)
err = deployWorker(workerImage, containerName, env, execution)
zombiecounter += 1
toBeRemoved.Data = append(toBeRemoved.Data, execution)
if err == nil {
log.Printf("[INFO] ExecutionID %s was deployed and to be removed from queue.", execution.ExecutionId)
toBeRemoved.Data = append(toBeRemoved.Data, execution)
executionIds = append(executionIds, execution.ExecutionId)
} else {
log.Printf("[WARNING] Execution ID %s failed to deploy: %s", execution.ExecutionId, err)
}
}
// Removes handled workflows (worker is made)
@@ -731,6 +830,7 @@ func main() {
// Is this ok to do with Docker? idk :)
func getRunningWorkers(ctx context.Context, workerTimeout int) int {
//log.Printf("[DEBUG] Getting running workers with API version %s", dockerApiVersion)
containers, err := dockercli.ContainerList(ctx, types.ContainerListOptions{
All: true,
})
@@ -741,8 +841,8 @@ func getRunningWorkers(ctx context.Context, workerTimeout int) int {
newVersionSplit := strings.Split(fmt.Sprintf("%s", err), "version is")
if len(newVersionSplit) > 1 {
dockerApiVersion = strings.TrimSpace(newVersionSplit[1])
log.Printf("[INFO] Changed the API version to default to %s", dockerApiVersion)
//dockerApiVersion = strings.TrimSpace(newVersionSplit[1])
log.Printf("[DEBUG] WANT to change the API version to default to %s?", strings.TrimSpace(newVersionSplit[1]))
}
return maxConcurrency
@@ -790,6 +890,11 @@ func getRunningWorkers(ctx context.Context, workerTimeout int) int {
// Should it check what happened to the execution? idk
func zombiecheck(ctx context.Context, workerTimeout int) error {
executionIds = []string{}
if swarmConfig == "run" {
//log.Printf("[DEBUG] Skipping Zombie check due to new execution model (swarm)")
return nil
}
log.Println("[INFO] Looking for old containers (zombies)")
containers, err := dockercli.ContainerList(ctx, types.ContainerListOptions{
All: true,
@@ -876,20 +981,8 @@ func zombiecheck(ctx context.Context, workerTimeout int) error {
return nil
}
type ExecutionRequest struct {
ExecutionId string `json:"execution_id"`
Authorization string `json:"authorization"`
HTTPProxy string `json:"http_proxy"`
HTTPSProxy string `json:"https_proxy"`
BaseUrl string `json:"base_url"`
EnvironmentName string `json:"environment_name"`
Timezone string `json:"timezone"`
Cleanup string `json:"cleanup"`
ShufflePassProxyToApp string `json:"shuffle_pass_proxy_to_app"`
}
func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest) error {
parsedRequest := ExecutionRequest{
parsedRequest := shuffle.OrborusExecutionRequest{
ExecutionId: workflowExecution.ExecutionId,
Authorization: workflowExecution.Authorization,
BaseUrl: os.Getenv("BASE_URL"),
@@ -916,6 +1009,8 @@ func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest) error {
return err
}
//log.Printf("[DEBUG] Data: %s", string(data))
streamUrl := fmt.Sprintf("%s:33333/api/v1/execute", parsedBaseurl)
req, err := http.NewRequest(
"POST",
@@ -925,23 +1020,27 @@ func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest) error {
client := &http.Client{}
if err != nil {
log.Printf("[ERROR] Failed creating finishing request: %s", err)
log.Printf("[ERROR] Failed creating worker request: %s", err)
return err
}
newresp, err := client.Do(req)
if err != nil {
log.Printf("[ERROR] Error running finishing request: %s", err)
log.Printf("[ERROR] Error running worker request: %s", err)
return err
}
if newresp.StatusCode != 200 {
log.Printf("[ERROR] Error running request - status code is %d, not 200", newresp.StatusCode)
return errors.New(fmt.Sprintf("Bad statuscode: %d - expecting 200", newresp.StatusCode))
}
body, err := ioutil.ReadAll(newresp.Body)
if err != nil {
log.Printf("[ERROR] Failed reading body: %s", err)
log.Printf("[ERROR] Failed reading body in worker request: %s", err)
return err
} else {
log.Printf("[INFO] NEWRESP (from backend): %s", string(body))
}
log.Printf("[DEBUG] NEWRESP (from worker request %s): %s (Status: %d)", workflowExecution.ExecutionId, string(body), newresp.StatusCode)
return nil
}
+1 -1
View File
@@ -1,5 +1,5 @@
NAME=shuffle-worker
VERSION=0.9.29
VERSION=0.9.30
echo "Running docker build with $NAME:$VERSION"
#CGO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo -o worker.bin .
+3 -1
View File
@@ -1,6 +1,8 @@
module worker
go 1.15
//replace github.com/shuffle/shuffle-shared => ../../../../../git/shuffle-shared
require (
github.com/containerd/containerd v1.5.7 // indirect
@@ -8,6 +10,6 @@ require (
github.com/docker/go-connections v0.4.0 // indirect
github.com/gorilla/mux v1.8.0
github.com/patrickmn/go-cache v2.1.0+incompatible
github.com/shuffle/shuffle-shared v0.1.20
github.com/shuffle/shuffle-shared v0.1.27
go4.org v0.0.0-20201209231011-d4a079459e60 // indirect
)
File diff suppressed because it is too large Load Diff
+2
View File
@@ -0,0 +1,2 @@
#curl -XPOST "http://192.168.86.37:33335/api/v1/run" -H "Content-Type: application/json" -d '{"execution_id":"ec35241a-b713-45c5-bda2-823fa4ce21e6","authorization":"d044f927-ad19-4f0e-a01f-51365e46c4e9","http_proxy":"","https_proxy":"","base_url":"http:\/\/192.168.86.37:5001","url":"http:\/\/192.168.86.37:33333","environment_name":"Shuffle","timezone":"","cleanup":"false","shuffle_pass_proxy_to_app":"","action":{"app_name":"Shuffle Tools","app_version":"1.1.0","app_id":"080b539e-638e-4a2b-9202-07d6360c2ddf","errors":[],"id":"13d73f9c-82be-4104-8b1e-8651ef189412","is_valid":true,"isStartNode":true,"sharing":true,"label":"Shuffle Tools_1","public":true,"generated":false,"environment":"Shuffle","name":"repeat_back_to_me","parameters":[{"description":"The message to repeat","id":"572cfe8c-eddf-4b66-8e53-99aeecfbd838","name":"call","example":"REPEATING: Hello world","value":"hello world","multiline":true,"options":null,"action_field":"","variant":"STATIC_VALUE","required":true,"configuration":false,"tags":null,"schema":{"type":"string"},"skip_multicheck":false,"value_replace":null,"unique_toggled":false}],"execution_variable":{"description":"","id":"","name":"","value":""},"position":{"x":367.5,"y":398.25},"authentication_id":"","category":"Testing","reference_url":"","sub_action":false,"source_workflow":""},"workflow_execution":{"type":"workflow","status":"EXECUTING","start":"13d73f9c-82be-4104-8b1e-8651ef189412","execution_argument":"","execution_id":"ec35241a-b713-45c5-bda2-823fa4ce21e6","execution_org":"015c13d4-2e03-45fd-82a9-9b1f8095937a","started_at":1635630830,"completed_at":0,"workflow_id":"ca6d239b-6523-4b11-8c92-00ea4ab82b6d","last_node":"","authorization":"d044f927-ad19-4f0e-a01f-51365e46c4e9","result":"","project_id":"","locations":null,"workflow":{"actions":[{"app_name":"Shuffle Tools","app_version":"1.1.0","app_id":"080b539e-638e-4a2b-9202-07d6360c2ddf","errors":[],"id":"13d73f9c-82be-4104-8b1e-8651ef189412","is_valid":true,"isStartNode":true,"sharing":true,"label":"Shuffle Tools_1","public":true,"generated":false,"environm
curl -XPOST http://192.168.86.37:33335/api/v1/run -H "Content-Type: application/json" -d '{"execution_id":"ec35241a-b713-45c5-bda2-823fa4ce21e6","authorization":"d044f927-ad19-4f0e-a01f-51365e46c4e9","http_proxy":"","https_proxy":"","base_url":"http:\/\/192.168.86.37:5001","url":"http:\/\/192.168.86.37:33333","environment_name":"Shuffle","timezone":"","cleanup":"false","shuffle_pass_proxy_to_app":"","action":{"app_name":"Shuffle Tools","app_version":"1.1.0","app_id":"080b539e-638e-4a2b-9202-07d6360c2ddf","errors":[],"id":"13d73f9c-82be-4104-8b1e-8651ef189412","is_valid":true,"isStartNode":true,"sharing":true,"label":"Shuffle Tools_1","public":true,"generated":false,"environment":"Shuffle","name":"repeat_back_to_me","parameters":[{"description":"The message to repeat","id":"572cfe8c-eddf-4b66-8e53-99aeecfbd838","name":"call","example":"REPEATING: Hello world","value":"hello world","multiline":true,"options":null,"action_field":"","variant":"STATIC_VALUE","required":true,"configuration":false,"tags":null,"schema":{"type":"string"},"skip_multicheck":false,"value_replace":null,"unique_toggled":false}],"execution_variable":{"description":"","id":"","name":"","value":""},"position":{"x":367.5,"y":398.25},"authentication_id":"","category":"Testing","reference_url":"","sub_action":false,"source_workflow":""},"workflow_execution":{"type":"workflow","status":"EXECUTING","start":"13d73f9c-82be-4104-8b1e-8651ef189412","execution_argument":"","execution_id":"ec35241a-b713-45c5-bda2-823fa4ce21e6","execution_org":"015c13d4-2e03-45fd-82a9-9b1f8095937a","started_at":1635630830,"completed_at":0,"workflow_id":"ca6d239b-6523-4b11-8c92-00ea4ab82b6d","last_node":"","authorization":"d044f927-ad19-4f0e-a01f-51365e46c4e9","result":"","project_id":"","locations":null,"workflow":{"actions":[{"app_name":"Shuffle Tools","app_version":"1.1.0","app_id":"080b539e-638e-4a2b-9202-07d6360c2ddf","errors":[],"id":"13d73f9c-82be-4104-8b1e-8651ef189412","is_valid":true,"isStartNode":true,"sharing":true,"label":"Shuffle Tools_1","public":true,"generated":false,"environment":"Shuffle","name":"repeat_back_to_me","parameters":[{"description":"The message to repeat","id":"572cfe8c-eddf-4b66-8e53-99aeecfbd838","name":"call","example":"REPEATING: Hello world","value":"hello world","multiline":true,"options":null,"action_field":"","variant":"STATIC_VALUE","required":true,"configuration":false,"tags":null,"schema":{"type":"string"},"skip_multicheck":false,"value_replace":null,"unique_toggled":false}],"execution_variable":{"description":"","id":"","name":"","value":""},"position":{"x":367.5,"y":398.25},"authentication_id":"","category":"Testing","reference_url":"","sub_action":false,"source_workflow":""}],"branches":[],"visual_branches":[],"triggers":[],"schedules":[],"configuration":{"exit_on_error":false,"start_from_top":false,"skip_notifications":false},"created":1635611089,"edited":1635611106,"last_runtime":0,"id":"ca6d239b-6523-4b11-8c92-00ea4ab82b6d","is_valid":true,"name":"Tools testing","description":"","start":"13d73f9c-82be-4104-8b1e-8651ef189412","owner":"2b3a37f3-fdd2-4965-b78d-f5d2fe9a2572","sharing":"private","org":[{"name":"","id":"015c13d4-2e03-45fd-82a9-9b1f8095937a","users":null,"role":"","creator_org":"","image":""}],"execution_org":{"name":"","id":"015c13d4-2e03-45fd-82a9-9b1f8095937a","users":null,"role":"","creator_org":"","image":""},"org_id":"015c13d4-2e03-45fd-82a9-9b1f8095937a","workflow_variables":null,"execution_environment":"","previously_saved":true,"categories":{"siem":{"name":"","description":"","count":0},"communication":{"name":"","description":"","count":0},"assets":{"name":"","description":"","count":0},"cases":{"name":"","description":"","count":0},"network":{"name":"","description":"","count":0},"intel":{"name":"","description":"","count":1},"edr":{"name":"","description":"","count":0},"other":{"name":"","description":"","count":0}},"example_argument":"","public":false,"default_return_value":"","contact_info":{"name":"","url":""},"published_id":""},"results":[],"org_id":"","sub_execution_count":0,"execution_source":"default","execution_parent":"","execution_source_node":"","execution_source_auth":""}}'
+471 -60
View File
@@ -23,6 +23,7 @@ import (
"github.com/docker/docker/api/types/container"
//"github.com/docker/docker/api/types/filters"
"github.com/docker/docker/api/types/mount"
"github.com/docker/docker/api/types/swarm"
dockerclient "github.com/docker/docker/client"
//"github.com/go-git/go-billy/v5/memfs"
@@ -32,6 +33,10 @@ import (
"github.com/gorilla/mux"
"github.com/patrickmn/go-cache"
"github.com/satori/go.uuid"
// No necessary outside shared
"cloud.google.com/go/datastore"
"cloud.google.com/go/storage"
)
// This is getting out of hand :)
@@ -39,6 +44,7 @@ var environment = os.Getenv("ENVIRONMENT_NAME")
var baseUrl = os.Getenv("BASE_URL")
var appCallbackUrl = os.Getenv("BASE_URL")
var cleanupEnv = strings.ToLower(os.Getenv("CLEANUP"))
var dockerApiVersion = strings.ToLower(os.Getenv("DOCKER_API_VERSION"))
var timezone = os.Getenv("TZ")
var baseimagename = "frikky/shuffle"
var registryName = "registry.hub.docker.com"
@@ -48,20 +54,26 @@ var topClient *http.Client
var data string
var requestsSent = 0
/*
var environments []string
var parents map[string][]string
var children map[string][]string
var visited []string
var executed []string
var nextActions []string
var containerIds []string
var extra int
var startAction string
*/
var results []shuffle.ActionResult
var allLogs map[string]string
var containerIds []string
var executionRunning bool
// New Worker mappings
var portMappings map[string]int
var baseport = 33333
// removes every container except itself (worker)
func shutdown(workflowExecution shuffle.WorkflowExecution, nodeId string, reason string, handleResultSend bool) {
log.Printf("[INFO] Shutdown (%s) started with reason %#v. Result amount: %d. ResultsSent: %d, Send result: %#v", workflowExecution.Status, reason, len(workflowExecution.Results), requestsSent, handleResultSend)
@@ -131,11 +143,15 @@ func shutdown(workflowExecution shuffle.WorkflowExecution, nodeId string, reason
}
// FIXME: Add an API call to the backend
authorization := os.Getenv("AUTHORIZATION")
if len(authorization) > 0 {
req.Header.Add("Authorization", fmt.Sprintf("Bearer %s", authorization))
if os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" {
authorization := os.Getenv("AUTHORIZATION")
if len(authorization) > 0 {
req.Header.Add("Authorization", fmt.Sprintf("Bearer %s", authorization))
} else {
log.Printf("[ERROR] No authorization specified for abort")
}
} else {
log.Printf("[ERROR] No authorization specified for abort")
req.Header.Add("Authorization", fmt.Sprintf("Bearer %s", workflowExecution.Authorization))
}
req.Header.Add("Content-Type", "application/json")
@@ -175,28 +191,58 @@ func shutdown(workflowExecution shuffle.WorkflowExecution, nodeId string, reason
time.Sleep(time.Duration(sleepDuration) * time.Second)
os.Exit(3)
} else {
log.Printf("[DEBUG] Sending result and resetting values (K8s & Swarm).")
environments = []string{}
parents = map[string][]string{}
children = map[string][]string{}
visited = []string{}
executed = []string{}
nextActions = []string{}
containerIds = []string{}
extra = 0
startAction = ""
results = []shuffle.ActionResult{}
allLogs = map[string]string{}
log.Printf("\n\n[DEBUG] Sending result and resetting values (K8s & Swarm).\n\n")
//UpdateExecutionVariables(ctx, workflowExecution.ExecutionId, startAction, children, parents, visited, executed, nextActions, environments, extra)
/*
environments = []string{}
parents = map[string][]string{}
children = map[string][]string{}
visited = []string{}
executed = []string{}
nextActions = []string{}
containerIds = []string{}
extra = 0
startAction = ""
results = []shuffle.ActionResult{}
allLogs = map[string]string{}
*/
requestsSent = 0
executionRunning = false
}
//cacheKey := fmt.Sprintf("workflowexecution-%s", workflowExecution.ExecutionId)
}
// Deploys the internal worker whenever something happens
func deployApp(cli *dockerclient.Client, image string, identifier string, env []string, workflowExecution shuffle.WorkflowExecution, actionId string) error {
func deployApp(cli *dockerclient.Client, image string, identifier string, env []string, workflowExecution shuffle.WorkflowExecution, action shuffle.Action) error {
// form basic hostConfig
ctx := context.Background()
if os.Getenv("SHUFFLE_SWARM_CONFIG") == "run" {
//identifier := fmt.Sprintf("%s_%s_%s_%s", appname, appversion, action.ID, workflowExecution.ExecutionId)
appName := strings.Replace(identifier, fmt.Sprintf("_%s", action.ID), "", -1)
appName = strings.Replace(appName, fmt.Sprintf("_%s", workflowExecution.ExecutionId), "", -1)
appName = strings.ToLower(appName)
log.Printf("[INFO] New appname: %s, image: %s", appName, image)
exposedPort, err := findAppInfo(image, appName)
if err != nil {
log.Printf("[ERROR] Failed finding and creating port for %s: %s", appName, err)
return err
}
log.Printf("[DEBUG] Should run towards port %d for app %s", exposedPort, appName)
err = sendAppRequest(baseUrl, exposedPort, action, workflowExecution)
if err != nil {
log.Printf("[ERROR] Failed sending request to app %s on port %d: %s", appName, exposedPort, err)
return err
}
log.Printf("[DEBUG] Successfully ran request towards port %d for app %s", exposedPort, appName)
return nil
}
// Max 10% CPU every second
//CPUShares: 128,
//CPUQuota: 10000,
@@ -211,7 +257,7 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env []
if os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" {
hostConfig.NetworkMode = container.NetworkMode(fmt.Sprintf("container:worker-%s", workflowExecution.ExecutionId))
log.Printf("Environments: %#v", env)
//log.Printf("Environments: %#v", env)
}
// Removing because log extraction should happen first
@@ -245,7 +291,7 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env []
})
}
} else {
log.Printf("[WARNING] No mounted folders")
log.Printf("[WARNING] Not mounting folders")
}
config := &container.Config{
@@ -305,7 +351,7 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env []
stats, err := cli.ContainerInspect(ctx, cont.ID)
if err != nil {
log.Printf("[ERROR] Failed getting container stats")
log.Printf("[ERROR] Failed getting container stats for container %s: %s", cont.ID, err)
} else {
//log.Printf("[INFO] Info for container: %#v", stats)
//log.Printf("%#v", stats.Config)
@@ -328,7 +374,7 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env []
//allLogs[actionId] = logs
if stats.ContainerJSONBase.State.Status == "exited" && !strings.Contains(logs, "Normal execution.") {
log.Printf("[WARNING] BAD Execution Logs for %s: %s", actionId, logs)
log.Printf("[WARNING] BAD Execution Logs for %s: %s", action.ID, logs)
exit = true
}
}
@@ -530,7 +576,11 @@ func removeIndex(s []string, i int) []string {
}
func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
ctx := context.Background()
startAction, extra, children, parents, visited, executed, nextActions, environments := shuffle.GetExecutionVariables(ctx, workflowExecution.ExecutionId)
log.Printf("[INFO] Inside execution results with %d / %d results", len(workflowExecution.Results), len(workflowExecution.Workflow.Actions)+extra)
if len(startAction) == 0 {
startAction = workflowExecution.Start
if len(startAction) == 0 {
@@ -540,11 +590,11 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
}
//log.Printf("NEXTACTIONS: %s", nextActions)
queueNodes := []string{}
//if len(nextActions) == 0 {
// nextActions = append(nextActions, startAction)
//}
queueNodes := []string{}
if len(workflowExecution.Results) == 0 {
nextActions = []string{startAction}
} else {
@@ -602,7 +652,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
}
if len(appendActions) > 0 {
log.Printf("APPENDED NODES: %#v", appendActions)
//log.Printf("APPENDED NODES: %#v", appendActions)
nextActions = append(nextActions, appendActions...)
}
}
@@ -1013,7 +1063,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
// If cleanup is set, it should run for efficiency
pullOptions := types.ImagePullOptions{}
if cleanupEnv == "true" {
err = deployApp(dockercli, images[0], identifier, env, workflowExecution, action.ID)
err = deployApp(dockercli, images[0], identifier, env, workflowExecution, action)
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
if strings.Contains(err.Error(), "exited prematurely") {
log.Printf("[DEBUG] Shutting down (2)")
@@ -1025,8 +1075,8 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
executed := false
if err == nil {
log.Printf("[DEBUG] Downloaded image %s from backend (CLEANUP)", image)
//err = deployApp(dockercli, image, identifier, env, workflow, action.ID)
err = deployApp(dockercli, image, identifier, env, workflowExecution, action.ID)
//err = deployApp(dockercli, image, identifier, env, workflow, action)
err = deployApp(dockercli, image, identifier, env, workflowExecution, action)
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
if strings.Contains(err.Error(), "exited prematurely") {
log.Printf("[DEBUG] Shutting down (41)")
@@ -1040,7 +1090,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
if !executed {
image = images[2]
err = deployApp(dockercli, image, identifier, env, workflowExecution, action.ID)
err = deployApp(dockercli, image, identifier, env, workflowExecution, action)
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
if strings.Contains(err.Error(), "exited prematurely") {
log.Printf("[DEBUG] Shutting down (3)")
@@ -1078,7 +1128,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
log.Printf("[INFO] Successfully downloaded %s", image)
}
err = deployApp(dockercli, image, identifier, env, workflowExecution, action.ID)
err = deployApp(dockercli, image, identifier, env, workflowExecution, action)
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
log.Printf("[ERROR] Failed deploying image for the FOURTH time. Aborting if the image doesn't exist")
@@ -1101,7 +1151,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
}
} else {
err = deployApp(dockercli, images[0], identifier, env, workflowExecution, action.ID)
err = deployApp(dockercli, images[0], identifier, env, workflowExecution, action)
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
if strings.Contains(err.Error(), "exited prematurely") {
log.Printf("[DEBUG] Shutting down (9)")
@@ -1112,7 +1162,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
// Trying to replace with lowercase to deploy again. This seems to work with Dockerhub well.
// FIXME: Should try to remotely download directly if this persists.
image = images[1]
err = deployApp(dockercli, image, identifier, env, workflowExecution, action.ID)
err = deployApp(dockercli, image, identifier, env, workflowExecution, action)
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
if strings.Contains(err.Error(), "exited prematurely") {
log.Printf("[DEBUG] Shutting down (10)")
@@ -1125,8 +1175,8 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
executed := false
if err == nil {
log.Printf("[DEBUG] Downloaded image %s from backend (CLEANUP)", image)
//err = deployApp(dockercli, image, identifier, env, workflow, action.ID)
err = deployApp(dockercli, image, identifier, env, workflowExecution, action.ID)
//err = deployApp(dockercli, image, identifier, env, workflow, action)
err = deployApp(dockercli, image, identifier, env, workflowExecution, action)
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
if strings.Contains(err.Error(), "exited prematurely") {
log.Printf("[DEBUG] Shutting down (40)")
@@ -1140,7 +1190,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
if !executed {
image = images[2]
err = deployApp(dockercli, image, identifier, env, workflowExecution, action.ID)
err = deployApp(dockercli, image, identifier, env, workflowExecution, action)
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
if strings.Contains(err.Error(), "exited prematurely") {
log.Printf("[DEBUG] Shutting down (11)")
@@ -1177,7 +1227,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
}
}
err = deployApp(dockercli, image, identifier, env, workflowExecution, action.ID)
err = deployApp(dockercli, image, identifier, env, workflowExecution, action)
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
log.Printf("[ERROR] Failed deploying image for the FOURTH time. Aborting if the image doesn't exist")
if strings.Contains(err.Error(), "exited prematurely") {
@@ -1242,12 +1292,14 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
}
func executionInit(workflowExecution shuffle.WorkflowExecution) error {
parents = map[string][]string{}
children = map[string][]string{}
parents := map[string][]string{}
children := map[string][]string{}
nextActions := []string{}
extra := 0
results = workflowExecution.Results
startAction = workflowExecution.Start
startAction := workflowExecution.Start
log.Printf("[INFO] STARTACTION: %s", startAction)
if len(startAction) == 0 {
log.Printf("[INFO] Didn't find execution start action. Setting it to workflow start action.")
@@ -1342,7 +1394,7 @@ func executionInit(workflowExecution shuffle.WorkflowExecution) error {
pullOptions := types.ImagePullOptions{}
_ = pullOptions
for _, image := range onpremApps {
log.Printf("[INFO] Image: %s", image)
//log.Printf("[INFO] Image: %s", image)
// Kind of gambling that the image exists.
if strings.Contains(image, " ") {
image = strings.ReplaceAll(image, " ", "-")
@@ -1361,12 +1413,41 @@ func executionInit(workflowExecution shuffle.WorkflowExecution) error {
//log.Printf("Successfully downloaded and built %s", image)
}
ctx := context.Background()
visited := []string{}
executed := []string{}
environments := []string{}
for _, action := range workflowExecution.Workflow.Actions {
found := false
for _, environment := range environments {
if action.Environment == environment {
found = true
break
}
}
if !found {
environments = append(environments, action.Environment)
}
}
//var visited []string
//var executed []string
err := shuffle.UpdateExecutionVariables(ctx, workflowExecution.ExecutionId, startAction, children, parents, visited, executed, nextActions, environments, extra)
if err != nil {
log.Printf("\n\n[ERROR] Failed to update exec variables for execution %s: %s\n\n", workflowExecution.ExecutionId, err)
}
return nil
}
func handleDefaultExecution(client *http.Client, req *http.Request, workflowExecution shuffle.WorkflowExecution) error {
// if no onprem runs (shouldn't happen, but extra check), exit
// if there are some, load the images ASAP for the app
ctx := context.Background()
//startAction, extra, children, parents, visited, executed, nextActions, environments := shuffle.GetExecutionVariables(ctx, workflowExecution.ExecutionId)
startAction, extra, _, _, _, _, _, _ := shuffle.GetExecutionVariables(ctx, workflowExecution.ExecutionId)
err := executionInit(workflowExecution)
if err != nil {
@@ -1377,7 +1458,6 @@ func handleDefaultExecution(client *http.Client, req *http.Request, workflowExec
log.Printf("[DEBUG] DEFAULT EXECUTION Startaction: %s", startAction)
ctx := context.Background()
setWorkflowExecution(ctx, workflowExecution, false)
streamResultUrl := fmt.Sprintf("%s/api/v1/streams/results", baseUrl)
@@ -1616,6 +1696,7 @@ func runTestExecution(client *http.Client, workflowId, apikey string) (string, s
}
func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
//log.Printf("[DEBUG] Got stream workflow queue")
body, err := ioutil.ReadAll(request.Body)
if err != nil {
log.Println("(3) Failed reading body for workflowqueue")
@@ -1659,7 +1740,7 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
if workflowExecution.Status == "FINISHED" {
log.Printf("[DEBUG] Workflowexecution is already FINISHED. No further action can be taken")
resp.WriteHeader(401)
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Workflowexecution is already finished because it has status %s"}`, workflowExecution.LastNode, workflowExecution.Status)))
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Workflowexecution is already finished because it has status %s. Lastnode: %s"}`, workflowExecution.Status, workflowExecution.LastNode)))
return
}
@@ -1829,6 +1910,10 @@ func sendResult(workflowExecution shuffle.WorkflowExecution, data []byte) {
}
func validateFinished(workflowExecution shuffle.WorkflowExecution) {
ctx := context.Background()
//startAction, extra, children, parents, visited, executed, nextActions, environments := shuffle.GetExecutionVariables(ctx, workflowExecution.ExecutionId)
_, extra, _, _, _, _, _, environments := shuffle.GetExecutionVariables(ctx, workflowExecution.ExecutionId)
log.Printf("[INFO] VALIDATION. Status: %s, shuffle.Actions: %d, Extra: %d, Results: %d\n", workflowExecution.Status, len(workflowExecution.Workflow.Actions), extra, len(workflowExecution.Results))
//if len(workflowExecution.Results) == len(workflowExecution.Workflow.Actions)+extra {
@@ -1849,6 +1934,7 @@ func validateFinished(workflowExecution shuffle.WorkflowExecution) {
}
func handleGetStreamResults(resp http.ResponseWriter, request *http.Request) {
//log.Printf("[DEBUG] Got stream result")
body, err := ioutil.ReadAll(request.Body)
if err != nil {
log.Println("Failed reading body for stream result queue")
@@ -1963,13 +2049,13 @@ func webserverSetup(workflowExecution shuffle.WorkflowExecution) net.Listener {
return listener
}
log.Printf("OLD HOSTNAME: %s", appCallbackUrl)
log.Printf("[DEBUG] OLD HOSTNAME: %s", appCallbackUrl)
if os.Getenv("SHUFFLE_SWARM_CONFIG") == "run" {
log.Printf("\n\nStarting webserver on port 33333 with hostname: %s\n\n", hostname)
appCallbackUrl = fmt.Sprintf("http://%s:33333", hostname)
listener, err = net.Listen("tcp", ":33333")
log.Printf("\n\nStarting webserver on port %d with hostname: %s\n\n", baseport, hostname)
appCallbackUrl = fmt.Sprintf("http://%s:%d", hostname, baseport)
listener, err = net.Listen("tcp", fmt.Sprintf(":%d", baseport))
if err != nil {
log.Printf("[ERROR] Failed to assign port to 33333")
log.Printf("[ERROR] Failed to assign port to %d: %s", baseport, err)
return nil
}
@@ -1996,9 +2082,8 @@ func runWebserver(listener net.Listener) {
r.HandleFunc("/api/v1/execute", handleRunExecution).Methods("POST", "OPTIONS")
}
http.Handle("/", r)
//log.Fatal(http.ListenAndServe(port, nil))
http.Handle("/", r)
log.Fatal(http.Serve(listener, nil))
}
@@ -2057,7 +2142,7 @@ func downloadDockerImageBackend(client *http.Client, imageName string) error {
imageLoadResponse, err := dockercli.ImageLoad(context.Background(), tar, true)
if err != nil {
log.Printf("[ERROR] Error loading: %s", err)
log.Printf("[ERROR] Error loading images: %s", err)
return err
}
@@ -2077,11 +2162,316 @@ func downloadDockerImageBackend(client *http.Client, imageName string) error {
return nil
}
func deploySwarmService(dockercli *dockerclient.Client, name, image string, deployport int) error {
log.Printf("[DEBUG] Deploying service for %s to swarm on port %d", name, deployport)
//containerName := fmt.Sprintf("shuffle-worker-%s", parsedUuid)
if len(baseimagename) == 0 {
baseimagename = "frikky/shuffle"
//var baseimagename = "frikky/shuffle"
//var registryName = "registry.hub.docker.com"
}
//image := fmt.Sprintf("%s:%s", baseimagename, name)
log.Printf("[DEBUG] Deploying app with name %s with image %s", name, image)
containerName := fmt.Sprintf(strings.Replace(name, ".", "-", -1))
serviceSpec := swarm.ServiceSpec{
Annotations: swarm.Annotations{
Name: containerName,
Labels: map[string]string{},
},
Networks: []swarm.NetworkAttachmentConfig{
swarm.NetworkAttachmentConfig{
Target: "shuffle-executions",
},
},
EndpointSpec: &swarm.EndpointSpec{
Ports: []swarm.PortConfig{
swarm.PortConfig{
Protocol: swarm.PortConfigProtocolTCP,
PublishMode: swarm.PortConfigPublishModeIngress,
Name: "app-port",
PublishedPort: uint32(deployport),
TargetPort: uint32(deployport),
},
},
},
TaskTemplate: swarm.TaskSpec{
Resources: &swarm.ResourceRequirements{
Reservations: &swarm.Resources{},
},
ContainerSpec: &swarm.ContainerSpec{
Image: image,
Env: []string{
fmt.Sprintf("SHUFFLE_SWARM_CONFIG=%s", os.Getenv("SHUFFLE_SWARM_CONFIG")),
fmt.Sprintf("SHUFFLE_APP_EXPOSED_PORT=%d", deployport),
},
},
RestartPolicy: &swarm.RestartPolicy{
Condition: swarm.RestartPolicyConditionNone,
},
Placement: &swarm.Placement{
MaxReplicas: 5,
},
},
}
/*
Mounts: []mount.Mount{
mount.Mount{
Source: "/var/run/docker.sock",
Target: "/var/run/docker.sock",
Type: mount.TypeBind,
},
},
*/
if dockerApiVersion != "" {
serviceSpec.TaskTemplate.ContainerSpec.Env = append(serviceSpec.TaskTemplate.ContainerSpec.Env, fmt.Sprintf("DOCKER_API_VERSION=%s", dockerApiVersion))
}
serviceOptions := types.ServiceCreateOptions{}
service, err := dockercli.ServiceCreate(
context.Background(),
serviceSpec,
serviceOptions,
)
_ = service
if err != nil {
log.Printf("[DEBUG] Failed deploying %s with image %s: %s", name, image, err)
return err
}
log.Printf("[DEBUG] Successfully deployed service %s with image %s on port %d", name, image, deployport)
return nil
}
// Runs data discovery
func findAppInfo(image, name string) (int, error) {
dockercli, err := dockerclient.NewEnvClient()
if err != nil {
log.Printf("[ERROR] Unable to create docker client (2): %s", err)
return -1, err
}
highest := baseport
exposedPort := -1
// Exists as a "cache" layer
if portMappings != nil {
for key, value := range portMappings {
if value > highest {
highest = value
}
if key == name {
exposedPort = value
break
}
}
} else {
portMappings = make(map[string]int)
}
//Filters:
if exposedPort == -1 {
serviceListOptions := types.ServiceListOptions{}
services, err := dockercli.ServiceList(
context.Background(),
serviceListOptions,
)
// Basic self-correction
if err != nil {
log.Printf("[ERROR] Unable to list services: %s (may continue anyway?)", err)
if strings.Contains(fmt.Sprintf("%s", err), "is too new") {
// Static for some reason
defaultVersion := "1.40"
dockerApiVersion = defaultVersion
os.Setenv("DOCKER_API_VERSION", defaultVersion)
log.Printf("[DEBUG] Setting Docker API to %s default and retrying listing requests", defaultVersion)
} else {
return -1, err
}
services, err = dockercli.ServiceList(
context.Background(),
serviceListOptions,
)
if err != nil {
log.Printf("[ERROR] Unable to list services (2): %s", err)
return -1, err
}
}
for _, service := range services {
//log.Printf("[INFO] Service: %#v", service.Spec.Annotations.Name)
for _, endpoint := range service.Spec.EndpointSpec.Ports {
if strings.Contains(endpoint.Name, "port") {
portMappings[service.Spec.Annotations.Name] = int(endpoint.PublishedPort)
if int(endpoint.PublishedPort) > highest {
highest = int(endpoint.PublishedPort)
}
if service.Spec.Annotations.Name == name || service.Spec.Annotations.Name == strings.Replace(name, ".", "-", -1) {
exposedPort = int(endpoint.PublishedPort)
//break
}
}
}
//log.Printf("%s - %s", service.Spec.Annotations.Name, strings.Replace(name, ".", "-", -1))
if service.Spec.Annotations.Name != name && service.Spec.Annotations.Name != strings.Replace(name, ".", "-", -1) {
continue
}
// Break if it's the correct port, as it's the right service
if exposedPort >= 0 {
break
}
}
}
//log.Printf("[DEBUG] Portmappings: %#v", portMappings)
if exposedPort >= 0 {
log.Printf("[INFO] Found service %s on port %d - no need to deploy another", name, exposedPort)
} else {
// Increment by 1 for highest port
if highest <= baseport {
highest = baseport
}
highest += 1
err = deploySwarmService(dockercli, name, image, highest)
if err != nil {
log.Printf("[WARNING] NOT Found service: %s. error: %s", name, err)
return highest, err
} else {
log.Printf("[INFO] Deployed app with name %s", name)
}
exposedPort = highest
log.Printf("[DEBUG] Waiting 10 seconds before moving on to let app start")
time.Sleep(time.Duration(10) * time.Second)
}
return exposedPort, nil
}
func sendAppRequest(incomingUrl string, port int, action shuffle.Action, workflowExecution shuffle.WorkflowExecution) error {
parsedRequest := shuffle.OrborusExecutionRequest{
ExecutionId: workflowExecution.ExecutionId,
Authorization: workflowExecution.Authorization,
EnvironmentName: os.Getenv("ENVIRONMENT_NAME"),
Timezone: os.Getenv("TZ"),
Cleanup: os.Getenv("CLEANUP"),
HTTPProxy: os.Getenv("HTTP_PROXY"),
HTTPSProxy: os.Getenv("HTTPS_PROXY"),
ShufflePassProxyToApp: os.Getenv("SHUFFLE_PASS_APP_PROXY"),
BaseUrl: baseUrl,
Action: action,
FullExecution: workflowExecution,
}
//var baseUrl = os.Getenv("BASE_URL")
//var appCallbackUrl = os.Getenv("BASE_URL")
parsedBaseurl := incomingUrl
if strings.Count(baseUrl, ":") >= 2 {
baseUrlSplit := strings.Split(baseUrl, ":")
if len(baseUrlSplit) >= 3 {
parsedBaseurl = strings.Join(baseUrlSplit[0:2], ":")
//parsedRequest.BaseUrl = fmt.Sprintf("%s:33333", parsedBaseurl)
}
}
if len(parsedRequest.Url) == 0 {
// Fixed callback url to the worker itself
if strings.Count(parsedBaseurl, ":") >= 2 {
parsedRequest.Url = parsedBaseurl
} else {
// Callback to worker
parsedRequest.Url = fmt.Sprintf("%s:%d", parsedBaseurl, baseport)
}
log.Printf("[DEBUG] Should add a baseurl for the app to get back to: %s", parsedRequest.Url)
}
// FIXME: Swapping because this was confusing during dev
tmp := parsedRequest.Url
parsedRequest.Url = parsedRequest.BaseUrl
parsedRequest.BaseUrl = tmp
log.Printf("[DEBUG] Worker URL: %s, Backend URL: %s", parsedRequest.BaseUrl, parsedRequest.Url)
data, err := json.Marshal(parsedRequest)
if err != nil {
log.Printf("[ERROR] Failed marshalling worker request: %s", err)
return err
}
streamUrl := fmt.Sprintf("%s:%d/api/v1/run", parsedBaseurl, port)
req, err := http.NewRequest(
"POST",
streamUrl,
bytes.NewBuffer([]byte(data)),
)
client := &http.Client{}
if err != nil {
log.Printf("[ERROR] Failed creating finishing request: %s", err)
return err
}
newresp, err := client.Do(req)
if err != nil {
log.Printf("[ERROR] Error running finishing request: %s", err)
return err
}
body, err := ioutil.ReadAll(newresp.Body)
if err != nil {
log.Printf("[ERROR] Failed reading body: %s", err)
return err
} else {
log.Printf("[INFO] NEWRESP (from app): %s", string(body))
}
return nil
}
// Initial loop etc
func main() {
/*
appName := "shuffle-tools_1.1.0"
image := "frikky/shuffle:shuffle-tools_1.1.0"
exposedPort, err := findAppInfo(image, appName)
if err != nil {
log.Printf("[ERROR] Failed finding and creating port for %s: %s", appName, err)
os.Exit(3)
}
log.Printf("[DEBUG] Should run towards port %d for app %s", exposedPort, appName)
err = sendAppRequest(appCallbackUrl, exposedPort, shuffle.Action{}, shuffle.WorkflowExecution{})
if err != nil {
log.Printf("[ERROR] Failed sending request to app %s on port %d: %s", appName, exposedPort, err)
os.Exit(3)
}
*/
_, err := shuffle.RunInit(datastore.Client{}, storage.Client{}, "", "", false, "")
if err != nil {
log.Printf("[ERROR] Failed to run worker init: %s", err)
} else {
log.Printf("[DEBUG] Ran init for worker to set up cache system. Docker version: %s", dockerApiVersion)
}
log.Printf("[INFO] Setting up worker environment")
sleepTime := 5
client := &http.Client{
Transport: &http.Transport{
Proxy: nil,
@@ -2105,7 +2495,7 @@ func main() {
timezone = "Europe/Amsterdam"
}
log.Printf("[INFO] Running with timezone %s", timezone)
log.Printf("[INFO] Running with timezone %s and swarm config %#v", timezone, os.Getenv("SHUFFLE_SWARM_CONFIG"))
if os.Getenv("SHUFFLE_SWARM_CONFIG") == "run" {
workflowExecution := shuffle.WorkflowExecution{}
listener := webserverSetup(workflowExecution)
@@ -2121,6 +2511,8 @@ func main() {
//}()
runWebserver(listener)
log.Printf("[ERROR] Stopped listener - exiting.")
os.Exit(3)
}
//imageName := fmt.Sprintf("%s/%s:shuffle_openapi_1.0.0", registryName, baseimagename)
@@ -2174,6 +2566,7 @@ func main() {
topClient = client
firstRequest := true
environments := []string{}
for {
// Because of this, it always has updated data.
// Removed request requirement from app_sdk
@@ -2319,12 +2712,14 @@ func main() {
}
func handleRunExecution(resp http.ResponseWriter, request *http.Request) {
if executionRunning {
log.Println("[WARNING] An execution is already running on this worker")
resp.WriteHeader(500)
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "An execution is already running"}`)))
return
}
/*
if executionRunning {
log.Println("[WARNING] An execution is already running on this worker")
resp.WriteHeader(500)
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "An execution is already running"}`)))
return
}
*/
executionRunning = true
body, err := ioutil.ReadAll(request.Body)
@@ -2359,6 +2754,7 @@ func handleRunExecution(resp http.ResponseWriter, request *http.Request) {
}
//if strings.ToLower(os.Getenv("SHUFFLE_PASS_APP_PROXY")) == "true" {
// Is it ok if these are standard? Should they be update-able after launch? Hmm
if len(execRequest.HTTPProxy) > 0 {
log.Printf("[DEBUG] Sending proxy info to child process")
os.Setenv("SHUFFLE_PASS_APP_PROXY", execRequest.ShufflePassProxyToApp)
@@ -2444,11 +2840,23 @@ func handleRunExecution(resp http.ResponseWriter, request *http.Request) {
executionRunning = false
log.Printf("[INFO] Workflow %s is finished. Exiting worker.", workflowExecution.ExecutionId)
log.Printf("[DEBUG] Shutting down (20)")
resp.WriteHeader(401)
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Bad status %s"}`, workflowExecution.Status)))
resp.WriteHeader(200)
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Bad status for execution - already %s. Returning with 200 OK"}`, workflowExecution.Status)))
return
}
//ctx := context.Background()
//startAction, extra, children, parents, visited, executed, nextActions, environments := shuffle.GetExecutionVariables(ctx, workflowExecution.ExecutionId)
extra := 0
for _, trigger := range workflowExecution.Workflow.Triggers {
//log.Printf("Appname trigger (0): %s", trigger.AppName)
if trigger.AppName == "User Input" || trigger.AppName == "Shuffle Workflow" {
extra += 1
}
}
log.Printf("[INFO] Status: %s, Results: %d, actions: %d", workflowExecution.Status, len(workflowExecution.Results), len(workflowExecution.Workflow.Actions)+extra)
if workflowExecution.Status != "EXECUTING" {
executionRunning = false
@@ -2470,7 +2878,10 @@ func handleRunExecution(resp http.ResponseWriter, request *http.Request) {
if err != nil {
log.Printf("[INFO] Workflow setup failed: %s", workflowExecution.ExecutionId, err)
log.Printf("[DEBUG] Shutting down (30)")
shutdown(workflowExecution, "", "", true)
resp.WriteHeader(401)
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Error in execution init: %s"}`, err)))
return
//shutdown(workflowExecution, "", "", true)
}
handleExecutionResult(workflowExecution)