From 066329e3ae29368898072ea69a199985114243ac Mon Sep 17 00:00:00 2001 From: frikky Date: Sun, 31 Oct 2021 01:47:04 +0200 Subject: [PATCH] Fixed base of new execution model with worker, orborus etc --- backend/app_sdk/app_base.py | 201 +++++++++---- backend/app_sdk/build.sh | 2 +- backend/app_sdk/requirements.txt | 3 + backend/go-app/go.mod | 2 +- frontend/src/views/AngularWorkflow.jsx | 3 +- functions/onprem/orborus/Dockerfile | 2 +- functions/onprem/orborus/build.sh | 2 +- functions/onprem/orborus/go.mod | 5 +- functions/onprem/orborus/go.sum | 17 +- functions/onprem/orborus/orborus.go | 201 ++++++++----- functions/onprem/worker/build.sh | 2 +- functions/onprem/worker/go.mod | 4 +- functions/onprem/worker/go.sum | 6 + functions/onprem/worker/worker.go | 381 ++++++++++++++++++++++--- 14 files changed, 655 insertions(+), 176 deletions(-) diff --git a/backend/app_sdk/app_base.py b/backend/app_sdk/app_base.py index d8e04d6e..5b080805 100644 --- a/backend/app_sdk/app_base.py +++ b/backend/app_sdk/app_base.py @@ -4,16 +4,18 @@ import sys import re import time import json +import liquid import logging -import requests -import urllib.parse -import http.client import urllib3 import hashlib -from liquid import Liquid -import liquid import zipfile +import requests +import http.client +import urllib.parse from io import BytesIO +from liquid import Liquid + +runtime = os.getenv("SHUFFLE_SWARM_CONFIG", "") class AppBase: __version__ = None @@ -27,7 +29,7 @@ class AppBase: # apikey is for the user / org # authorization is for the specific workflow - self.url = os.getenv("CALLBACK_URL", "https://shuffler.io") + self.url = os.getenv("CALLBACK_URL", "https://shuffler.io") self.base_url = os.getenv("BASE_URL", "https://shuffler.io") self.action = os.getenv("ACTION", "") self.original_action = os.getenv("ACTION", "") @@ -51,8 +53,10 @@ class AppBase: try: self.action = json.loads(self.action) self.original_action = json.loads(self.action) - except: - self.logger.info("[WARNING] Failed parsing action as JSON") + except Exception as e: + self.logger.info(f"[WARNING] Failed parsing action as JSON (init): {e}. NOT important if running apps with webserver") + + #print(f"ACTION: {self.action}") if len(self.base_url) == 0: self.base_url = self.url @@ -80,7 +84,7 @@ 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}") @@ -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}") @@ -850,7 +856,6 @@ class AppBase: return file_ids async def execute_action(self, action): - # !!! Let this line stay - its used for some horrible codegeneration / stitching !!! # #STARTCOPY stream_path = "/api/v1/streams" @@ -880,19 +885,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 +923,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 +942,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 +951,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 +1651,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 @@ -2105,7 +2115,7 @@ class AppBase: if " " in actionname: actionname.replace(" ", "_", -1) - + #print(action) #if action.generated: # actionname = actionname.lower() @@ -2113,7 +2123,7 @@ 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): @@ -2460,7 +2470,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 @@ -2821,34 +2831,129 @@ 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(): + 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 + app = cls(redis=None, logger=logger, console_logger=logger) + #print(f"APP: {app}") + + 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 + await app.execute_action(app.action) + except Exception as e: + return { + "success": False, + "reason": f"Problem in execution {e}", + } + + return { + "success": True, + } + 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) + ####################### 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.run(host="0.0.0.0", port=33334) + +if __name__ == "__main__": + import asyncio + asyncio.run(AppBase.run(), debug=True) diff --git a/backend/app_sdk/build.sh b/backend/app_sdk/build.sh index 847c0185..c42e273d 100644 --- a/backend/app_sdk/build.sh +++ b/backend/app_sdk/build.sh @@ -3,7 +3,7 @@ ### DEFAULT NAME=shuffle-app_sdk -VERSION=0.9.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 diff --git a/backend/app_sdk/requirements.txt b/backend/app_sdk/requirements.txt index d561950b..3cba831d 100644 --- a/backend/app_sdk/requirements.txt +++ b/backend/app_sdk/requirements.txt @@ -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 diff --git a/backend/go-app/go.mod b/backend/go-app/go.mod index c175d626..81f28753 100644 --- a/backend/go-app/go.mod +++ b/backend/go-app/go.mod @@ -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.21 go4.org v0.0.0-20201209231011-d4a079459e60 // indirect golang.org/x/crypto v0.0.0-20210921155107-089bfa567519 google.golang.org/api v0.58.0 diff --git a/frontend/src/views/AngularWorkflow.jsx b/frontend/src/views/AngularWorkflow.jsx index 8b27b6ab..90c2274c 100644 --- a/frontend/src/views/AngularWorkflow.jsx +++ b/frontend/src/views/AngularWorkflow.jsx @@ -1956,6 +1956,7 @@ const AngularWorkflow = (props) => { } workflow.start = parentNode.data('id') + setLastSaved(true) parentNode.data('isStartNode', true) } @@ -8133,7 +8134,7 @@ const AngularWorkflow = (props) => { } const executionModal = - 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}}}> + 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 ?
diff --git a/functions/onprem/orborus/Dockerfile b/functions/onprem/orborus/Dockerfile index b7160a20..00cd6b55 100644 --- a/functions/onprem/orborus/Dockerfile +++ b/functions/onprem/orborus/Dockerfile @@ -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 . diff --git a/functions/onprem/orborus/build.sh b/functions/onprem/orborus/build.sh index 30cdf0c5..570cced1 100644 --- a/functions/onprem/orborus/build.sh +++ b/functions/onprem/orborus/build.sh @@ -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 diff --git a/functions/onprem/orborus/go.mod b/functions/onprem/orborus/go.mod index 1f8bc78d..9b4f9942 100644 --- a/functions/onprem/orborus/go.mod +++ b/functions/onprem/orborus/go.mod @@ -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.24 ) diff --git a/functions/onprem/orborus/go.sum b/functions/onprem/orborus/go.sum index 01532c6e..78bc5e0e 100644 --- a/functions/onprem/orborus/go.sum +++ b/functions/onprem/orborus/go.sum @@ -63,7 +63,6 @@ github.com/Microsoft/go-winio v0.4.16-0.20201130162521-d1ffc52c7331/go.mod h1:XB github.com/Microsoft/go-winio v0.4.16/go.mod h1:XB6nPKklQyQ7GC9LdcBEcBl8PF76WugXOPRXwdLnMv0= github.com/Microsoft/go-winio v0.4.17-0.20210211115548-6eac466e5fa3/go.mod h1:JPGBdM1cNvN/6ISo+n8V5iA4v8pBzdOpzfwIujj1a84= github.com/Microsoft/go-winio v0.4.17-0.20210324224401-5516f17a5958/go.mod h1:JPGBdM1cNvN/6ISo+n8V5iA4v8pBzdOpzfwIujj1a84= -github.com/Microsoft/go-winio v0.4.17 h1:iT12IBVClFevaf8PuVyi3UmZOVh4OqnaLxDTW2O6j3w= github.com/Microsoft/go-winio v0.4.17/go.mod h1:JPGBdM1cNvN/6ISo+n8V5iA4v8pBzdOpzfwIujj1a84= github.com/Microsoft/hcsshim v0.8.6/go.mod h1:Op3hHsoHPAvb6lceZHDtd9OkTew38wNoXnJs8iY7rUg= github.com/Microsoft/hcsshim v0.8.7-0.20190325164909-8abdbb8205e4/go.mod h1:Op3hHsoHPAvb6lceZHDtd9OkTew38wNoXnJs8iY7rUg= @@ -242,8 +241,8 @@ github.com/docker/distribution v2.7.1-0.20190205005809-0d3efadf0154+incompatible github.com/docker/distribution v2.7.1+incompatible h1:a5mlkVzth6W5A4fOsS3D2EO5BUmsJpcB+cRlLU7cSug= github.com/docker/distribution v2.7.1+incompatible/go.mod h1:J2gT2udsDAN96Uj4KfcMRqY0/ypR+oyYUYmja8H+y+w= github.com/docker/docker v1.13.1 h1:IkZjBSIc8hBjLpqeAbeE5mca5mNgeatLHBy3GO78BWo= -github.com/docker/docker v20.10.9+incompatible h1:JlsVnETOjM2RLQa0Cc1XCIspUdXW3Zenq9P54uXBm6k= -github.com/docker/docker v20.10.9+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk= +github.com/docker/docker v20.10.10+incompatible h1:GKkP0T7U4ks6X3lmmHKC2QDprnpRJor2Z5a8m62R9ZM= +github.com/docker/docker v20.10.10+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk= github.com/docker/go-connections v0.4.0 h1:El9xVISelRB7BuFusrZozjnkIM5YnzCViNKohAFqRJQ= github.com/docker/go-connections v0.4.0/go.mod h1:Gbd7IOopHjR8Iph03tsViu4nIes5XhDvyHbTtUxmeec= github.com/docker/go-events v0.0.0-20170721190031-9461782956ad/go.mod h1:Uw6UezgYA44ePAFQYUehOuCzmy5zmg/+nl2ZfMWGkpA= @@ -424,7 +423,6 @@ github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCV github.com/json-iterator/go v1.1.7/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4= github.com/json-iterator/go v1.1.10/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4= github.com/jstemmer/go-junit-report v0.0.0-20190106144839-af01ea7f8024/go.mod h1:6v2b51hI/fHJwM22ozAgKL4VKDeJcHhJFhtBdhmNjmU= -github.com/jstemmer/go-junit-report v0.9.1 h1:6QPYqodiu3GuPL+7mfx+NwDdp2eTkp9IfEUpgAwUN0o= github.com/jstemmer/go-junit-report v0.9.1/go.mod h1:Brl9GWCQeLvo8nXZwPNNblvFj/XSXhF0NWZEnDohbsk= github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU= github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w= @@ -563,13 +561,12 @@ github.com/prometheus/tsdb v0.7.1/go.mod h1:qhTCs0VvXwvX/y3TZrWD7rabWM+ijKTux40T github.com/rogpeppe/fastuuid v0.0.0-20150106093220-6724a57986af/go.mod h1:XWv6SoW27p1b0cqNHllgS5HIMJraePCO15w5zCzIWYg= github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4= github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= -github.com/rwcarlsen/goexif v0.0.0-20190401172101-9e8deecbddbd/go.mod h1:hPqNNc0+uJM6H+SuU8sEs5K5IQeKccPqeSjfgcKGgPk= github.com/safchain/ethtool v0.0.0-20190326074333-42ed695e3de8/go.mod h1:Z0q5wiBQGYcxhMZ6gUqHn6pYNLypFAvaL3UvgZLR0U4= github.com/satori/go.uuid v1.2.0 h1:0uYX9dsZ2yD7q2RtLRtPSdGDWzjeM3TbMJP9utgA0ww= github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0= github.com/seccomp/libseccomp-golang v0.9.1/go.mod h1:GbW5+tmTXfcxTToHLXlScSlAvWlF4P2Ca7zGrPiEpWo= -github.com/shuffle/shuffle-shared v0.1.19 h1:bZmwdC3gKPFxtKoGEjHGY8C96+xB+cBdHaz6glP8SwA= -github.com/shuffle/shuffle-shared v0.1.19/go.mod h1:0QrK51T12CpCj/be8hXduj/RtDnoeaZ3rfogELZE2IU= +github.com/shuffle/shuffle-shared v0.1.24 h1:5kJmwN175x8RiF1Ao2bD8WQyc+8ZuOHbrHeJ8H3nO2s= +github.com/shuffle/shuffle-shared v0.1.24/go.mod h1:0QrK51T12CpCj/be8hXduj/RtDnoeaZ3rfogELZE2IU= github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc= github.com/sirupsen/logrus v1.0.4-0.20170822132746-89742aefa4b2/go.mod h1:pMByvHTf9Beacp5x1UXfOR9xyW/9antXMhjMPG0dEzc= github.com/sirupsen/logrus v1.0.6/go.mod h1:pMByvHTf9Beacp5x1UXfOR9xyW/9antXMhjMPG0dEzc= @@ -656,8 +653,6 @@ go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE= go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE= go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0= go.uber.org/zap v1.10.0/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q= -go4.org v0.0.0-20201209231011-d4a079459e60 h1:iqAGo78tVOJXELHQFRjR6TMwItrvXH4hrGJ32I/NFF8= -go4.org v0.0.0-20201209231011-d4a079459e60/go.mod h1:CIiUVy99QCPfoE13bO4EZaz5GZMZXMSBGhxRdsvzbkg= golang.org/x/crypto v0.0.0-20171113213409-9f005a07e0d3/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4= golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4= golang.org/x/crypto v0.0.0-20181009213950-7c1a557ab941/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4= @@ -695,7 +690,6 @@ golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHl golang.org/x/lint v0.0.0-20191125180803-fdd1cda4f05f/go.mod h1:5qLYkcX4OjUUV8bRuDixDT3tpyyb+LUpUlRWLxfhWrs= golang.org/x/lint v0.0.0-20200130185559-910be7a94367/go.mod h1:3xt1FjdF8hUf6vQPIChWIBhFzV8gjjsPE/fR3IyQdNY= golang.org/x/lint v0.0.0-20200302205851-738671d3881b/go.mod h1:3xt1FjdF8hUf6vQPIChWIBhFzV8gjjsPE/fR3IyQdNY= -golang.org/x/lint v0.0.0-20201208152925-83fdc39ff7b5 h1:2M3HP5CCK1Si9FQhwnzYhXdG6DXeebvUHFpre8QvbyI= golang.org/x/lint v0.0.0-20201208152925-83fdc39ff7b5/go.mod h1:3xt1FjdF8hUf6vQPIChWIBhFzV8gjjsPE/fR3IyQdNY= golang.org/x/mobile v0.0.0-20190312151609-d3739f865fa6/go.mod h1:z+o9i4GpDbdi3rU15maQ/Ox0txvL9dWGYEHz965HBQE= golang.org/x/mobile v0.0.0-20190719004257-d2bd2a29d028/go.mod h1:E/iHnbuqvinMTCcRqshq8CkpyQDoeVncDDYHnLhea+o= @@ -706,7 +700,6 @@ golang.org/x/mod v0.1.1-0.20191107180719-034126e5016b/go.mod h1:QqPTAvyqsEbceGzB golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.4.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= -golang.org/x/mod v0.4.1 h1:Kvvh58BN8Y9/lBi7hTekvtMpm07eUZ0ck5pRHpsMWrY= golang.org/x/mod v0.4.1/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= @@ -911,12 +904,10 @@ golang.org/x/tools v0.0.0-20201110124207-079ba7bd75cd/go.mod h1:emZCQorbCU4vsT4f golang.org/x/tools v0.0.0-20201201161351-ac6f37ff4c2a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= golang.org/x/tools v0.0.0-20210108195828-e2f9c7f1fc8e/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= -golang.org/x/tools v0.0.0-20210114065538-d78b04bdf963 h1:K+NlvTLy0oONtRtkl1jRD9xIhnItbG2PiE7YOdjPb+k= golang.org/x/tools v0.0.0-20210114065538-d78b04bdf963/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= -golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 h1:go1bK/D/BFZV2I8cIQd1NKEZ+0owSTG1fDTci4IqFcE= golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= google.golang.org/api v0.0.0-20160322025152-9bf6e6e569ff/go.mod h1:4mhQ8q/RsB7i+udVvVy5NUi08OU8ZlA0gRVgrF7VFY0= google.golang.org/api v0.4.0/go.mod h1:8k5glujaEP+g9n7WNsDg8QP6cUVNI86fCNMcbazEtwE= diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index 24ffb94d..0790de85 100644 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -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" @@ -63,6 +65,7 @@ var timezone = os.Getenv("TZ") var containerName = os.Getenv("ORBORUS_CONTAINER_NAME") var swarmConfig = os.Getenv("SHUFFLE_SWARM_CONFIG") var executionIds = []string{} +var dockerized bool var dockercli *dockerclient.Client var containerId string @@ -81,6 +84,7 @@ func init() { // form id of current running container func getThisContainerId() { fCol := "" + dockerized = true // some adjusting based on current running mode switch runningMode { @@ -98,6 +102,7 @@ func getThisContainerId() { default: fCol = "3" // for backward-compatibility with production + dockerized = false log.Printf("[WARNING] RUNNING_MODE not set - defaulting to Docker (NOT Kubernetes).") } @@ -132,49 +137,52 @@ 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") 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() - log.Printf("[DEBUG] Deploying containers with swarm") + //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 for worker with swarm") //containerName := fmt.Sprintf("shuffle-worker-%s", parsedUuid) - containerName := fmt.Sprintf("shuffle-workers") + innerContainerName := fmt.Sprintf("shuffle-workers") + replicas := uint64(2) + 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 +215,7 @@ func deployWorker(image string, identifier string, env []string, executionReques Condition: swarm.RestartPolicyConditionNone, }, Placement: &swarm.Placement{ - MaxReplicas: 1, + MaxReplicas: replicas, }, }, } @@ -217,26 +225,75 @@ func deployWorker(image string, identifier string, env []string, executionReques } 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") { + 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", + }, + } + + if len(containerId) > 0 && dockerized == true { + 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" { + 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") { + workerImage := fmt.Sprintf("%s/%s/shuffle-worker:%s", baseimageregistry, baseimagename, workerVersion) + go 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,11 +321,11 @@ 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 } } @@ -276,7 +333,7 @@ func deployWorker(image string, identifier string, env []string, executionReques 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 + return err //stats, err := cli.ContainerInspect(context.Background(), containerName) //if err != nil { @@ -303,7 +360,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 { @@ -486,6 +543,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 +697,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 +723,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) @@ -876,20 +938,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"), @@ -925,16 +975,21 @@ 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) diff --git a/functions/onprem/worker/build.sh b/functions/onprem/worker/build.sh index 11350ed7..a78c2607 100644 --- a/functions/onprem/worker/build.sh +++ b/functions/onprem/worker/build.sh @@ -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 . diff --git a/functions/onprem/worker/go.mod b/functions/onprem/worker/go.mod index 197e1404..3ddcbb2d 100644 --- a/functions/onprem/worker/go.mod +++ b/functions/onprem/worker/go.mod @@ -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.25 go4.org v0.0.0-20201209231011-d4a079459e60 // indirect ) diff --git a/functions/onprem/worker/go.sum b/functions/onprem/worker/go.sum index 21cea8f2..97f693db 100644 --- a/functions/onprem/worker/go.sum +++ b/functions/onprem/worker/go.sum @@ -573,6 +573,12 @@ github.com/shuffle/shuffle-shared v0.1.19 h1:bZmwdC3gKPFxtKoGEjHGY8C96+xB+cBdHaz github.com/shuffle/shuffle-shared v0.1.19/go.mod h1:0QrK51T12CpCj/be8hXduj/RtDnoeaZ3rfogELZE2IU= github.com/shuffle/shuffle-shared v0.1.20 h1:Wz3DZtFtsd/F3rQuZ4KpIURM2HT1jPwEF3eJg8MX9TE= github.com/shuffle/shuffle-shared v0.1.20/go.mod h1:0QrK51T12CpCj/be8hXduj/RtDnoeaZ3rfogELZE2IU= +github.com/shuffle/shuffle-shared v0.1.21 h1:9QOCTwgmCG8JEh5yuvBZe1eUkD+iID/5iZZMgCI/50U= +github.com/shuffle/shuffle-shared v0.1.21/go.mod h1:0QrK51T12CpCj/be8hXduj/RtDnoeaZ3rfogELZE2IU= +github.com/shuffle/shuffle-shared v0.1.24 h1:5kJmwN175x8RiF1Ao2bD8WQyc+8ZuOHbrHeJ8H3nO2s= +github.com/shuffle/shuffle-shared v0.1.24/go.mod h1:0QrK51T12CpCj/be8hXduj/RtDnoeaZ3rfogELZE2IU= +github.com/shuffle/shuffle-shared v0.1.25 h1:+Wjgg5FptXdD6wbYgYQ7ALE+WyG97V4x/8VHfPnBMM0= +github.com/shuffle/shuffle-shared v0.1.25/go.mod h1:0QrK51T12CpCj/be8hXduj/RtDnoeaZ3rfogELZE2IU= github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc= github.com/sirupsen/logrus v1.0.4-0.20170822132746-89742aefa4b2/go.mod h1:pMByvHTf9Beacp5x1UXfOR9xyW/9antXMhjMPG0dEzc= github.com/sirupsen/logrus v1.0.6/go.mod h1:pMByvHTf9Beacp5x1UXfOR9xyW/9antXMhjMPG0dEzc= diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index f02be254..03aa9ff4 100644 --- a/functions/onprem/worker/worker.go +++ b/functions/onprem/worker/worker.go @@ -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" @@ -39,6 +40,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" @@ -62,6 +64,10 @@ var allLogs map[string]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) @@ -175,7 +181,7 @@ 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).") + log.Printf("\n\n[DEBUG] Sending result and resetting values (K8s & Swarm).\n\n") environments = []string{} parents = map[string][]string{} children = map[string][]string{} @@ -187,16 +193,43 @@ func shutdown(workflowExecution shuffle.WorkflowExecution, nodeId string, reason 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 +244,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 +278,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 +338,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 +361,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 } } @@ -540,11 +573,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 +635,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 +1046,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 +1058,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 +1073,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 +1111,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 +1134,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 +1145,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 +1158,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 +1173,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 +1210,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") { @@ -1616,6 +1649,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") @@ -1849,6 +1883,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") @@ -1965,11 +2000,11 @@ func webserverSetup(workflowExecution shuffle.WorkflowExecution) net.Listener { log.Printf("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 +2031,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)) } @@ -2077,11 +2111,290 @@ 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, + ) + + if err != nil { + log.Printf("[ERROR] Unable to list services: %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) + } + */ + log.Printf("[INFO] Setting up worker environment") sleepTime := 5 - client := &http.Client{ Transport: &http.Transport{ Proxy: nil, @@ -2121,6 +2434,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) @@ -2319,12 +2634,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)