From 580f8e403faf44a84950b4d5a5bb97c0d5f0bc4a Mon Sep 17 00:00:00 2001 From: frikky Date: Sun, 31 Oct 2021 22:16:59 +0100 Subject: [PATCH] Added major fixes for new execution system with backwards compatibility and auto-runs --- backend/app_sdk/app_base.py | 42 +++++--- backend/go-app/go.mod | 4 +- docker-compose.yml | 3 +- functions/onprem/orborus/go.mod | 2 +- functions/onprem/orborus/go.sum | 2 + functions/onprem/orborus/orborus.go | 98 ++++++++++++----- functions/onprem/worker/go.mod | 2 +- functions/onprem/worker/go.sum | 2 + functions/onprem/worker/run.sh | 2 + functions/onprem/worker/worker.go | 160 ++++++++++++++++++++++------ 10 files changed, 238 insertions(+), 79 deletions(-) create mode 100644 functions/onprem/worker/run.sh diff --git a/backend/app_sdk/app_base.py b/backend/app_sdk/app_base.py index 5b080805..7e238e76 100644 --- a/backend/app_sdk/app_base.py +++ b/backend/app_sdk/app_base.py @@ -91,7 +91,7 @@ class AppBase: #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 @@ -855,7 +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" @@ -2015,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"] @@ -2129,7 +2130,8 @@ class AppBase: 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 @@ -2561,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 = "" @@ -2636,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 @@ -2847,7 +2851,8 @@ class AppBase: flask_app = Flask(__name__) @flask_app.route("/api/v1/run", methods=["POST"]) - async def execute(): + #async def execute(): + def execute(): if request.method == "POST": #print(request.get_json(force=True)) #print("DATA: ", request.data) @@ -2860,12 +2865,12 @@ class AppBase: "reason": f"Invalid Action data {e}", } - logger.info(f"[DEBUG] Datatype: {type(requestdata)}: {requestdata}") + #logger.info(f"[DEBUG] Datatype: {type(requestdata)}: {requestdata}") # Remaking class for each request - app = cls(redis=None, logger=logger, console_logger=logger) #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) @@ -2899,7 +2904,8 @@ class AppBase: logger.info("Failed parsing base url") #await - await app.execute_action(app.action) + app.execute_action(app.action) + logger.info("\n\n[DEBUG] Done awaiting app action running\n\n") except Exception as e: return { "success": False, @@ -2908,6 +2914,7 @@ class AppBase: return { "success": True, + "reason": "App successfully finished", } else: return { @@ -2916,7 +2923,15 @@ class AppBase: } logger.info(f"[DEBUG] Serving on port {port}") - serve(flask_app, host="0.0.0.0", 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: # Has to start like this due to imports in other apps @@ -2950,7 +2965,8 @@ class AppBase: else: self.logger.info("ACTION TYPE (unhandled): %s" % type(action)) - await app.execute_action(app.action) + #await app.execute_action(app.action) + app.execute_action(app.action) #app.run(host="0.0.0.0", port=33334) diff --git a/backend/go-app/go.mod b/backend/go-app/go.mod index 81f28753..e76a9b50 100644 --- a/backend/go-app/go.mod +++ b/backend/go-app/go.mod @@ -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.21 + 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 diff --git a/docker-compose.yml b/docker-compose.yml index 4205bb18..494635d1 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -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} diff --git a/functions/onprem/orborus/go.mod b/functions/onprem/orborus/go.mod index 9b4f9942..287818a4 100644 --- a/functions/onprem/orborus/go.mod +++ b/functions/onprem/orborus/go.mod @@ -8,5 +8,5 @@ require ( 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.24 + github.com/shuffle/shuffle-shared v0.1.26 ) diff --git a/functions/onprem/orborus/go.sum b/functions/onprem/orborus/go.sum index 78bc5e0e..1b7d1985 100644 --- a/functions/onprem/orborus/go.sum +++ b/functions/onprem/orborus/go.sum @@ -567,6 +567,8 @@ github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdh github.com/seccomp/libseccomp-golang v0.9.1/go.mod h1:GbW5+tmTXfcxTToHLXlScSlAvWlF4P2Ca7zGrPiEpWo= 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.26 h1:NlUlUrA/wNLwo49Id5SKhBlLhuizjy/T2rqCvaxqBuk= +github.com/shuffle/shuffle-shared v0.1.26/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/orborus/orborus.go b/functions/onprem/orborus/orborus.go index 0790de85..677931e8 100644 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -65,7 +65,6 @@ 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 @@ -84,7 +83,6 @@ func init() { // form id of current running container func getThisContainerId() { fCol := "" - dockerized = true // some adjusting based on current running mode switch runningMode { @@ -102,7 +100,6 @@ func getThisContainerId() { default: fCol = "3" // for backward-compatibility with production - dockerized = false log.Printf("[WARNING] RUNNING_MODE not set - defaulting to Docker (NOT Kubernetes).") } @@ -111,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 } @@ -138,7 +137,7 @@ func getThisContainerId() { } func deployServiceWorkers(image string) { - log.Printf("[DEBUG] Validating deployment of workers as services IF swarmConfig = run") + 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" @@ -168,10 +167,22 @@ func deployServiceWorkers(image string) { // serviceOptions, //) - log.Printf("[DEBUG] Deploying containers for worker with swarm") //containerName := fmt.Sprintf("shuffle-worker-%s", parsedUuid) - innerContainerName := 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{ @@ -224,6 +235,10 @@ func deployServiceWorkers(image string) { 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{} _, err = dockercli.ServiceCreate( ctx, @@ -236,7 +251,7 @@ func deployServiceWorkers(image string) { //time.Sleep(time.Duration(10) * time.Second) //log.Printf("[DEBUG] Servicecreate request: %#v %#v", service, err) } else { - if !strings.Contains(fmt.Sprintf("%s", err), "Already Exists") { + 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) } } @@ -264,9 +279,7 @@ func deployWorker(image string, identifier string, env []string, executionReques }, } - if len(containerId) > 0 && dockerized == true { - hostConfig.NetworkMode = container.NetworkMode(fmt.Sprintf("container:%s", containerId)) - } + hostConfig.NetworkMode = container.NetworkMode(fmt.Sprintf("container:%s", containerId)) if cleanupEnv == "true" { hostConfig.AutoRemove = true @@ -280,18 +293,20 @@ func deployWorker(image string, identifier string, env []string, executionReques //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) + 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") { - workerImage := fmt.Sprintf("%s/%s/shuffle-worker:%s", baseimageregistry, baseimagename, workerVersion) - go deployServiceWorkers(workerImage) + 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 err - } else { - log.Printf("[DEBUG] Started worker from request with name: %s", executionRequest.ExecutionId) - } + }() return nil } @@ -332,8 +347,29 @@ func deployWorker(image string, identifier string, env []string, executionReques 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 err + 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 { @@ -393,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) @@ -793,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, }) @@ -803,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 @@ -852,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, @@ -966,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", @@ -992,11 +1037,10 @@ func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest) error { 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 } diff --git a/functions/onprem/worker/go.mod b/functions/onprem/worker/go.mod index 3ddcbb2d..f7bdf6c6 100644 --- a/functions/onprem/worker/go.mod +++ b/functions/onprem/worker/go.mod @@ -10,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.25 + github.com/shuffle/shuffle-shared v0.1.27 go4.org v0.0.0-20201209231011-d4a079459e60 // indirect ) diff --git a/functions/onprem/worker/go.sum b/functions/onprem/worker/go.sum index 97f693db..32e87a78 100644 --- a/functions/onprem/worker/go.sum +++ b/functions/onprem/worker/go.sum @@ -579,6 +579,8 @@ github.com/shuffle/shuffle-shared v0.1.24 h1:5kJmwN175x8RiF1Ao2bD8WQyc+8ZuOHbrHe 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/shuffle/shuffle-shared v0.1.27 h1:dFISdLvQF0cpAuFzNgVk85Jy5uTc+QVkK1n2LjA8r9c= +github.com/shuffle/shuffle-shared v0.1.27/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/run.sh b/functions/onprem/worker/run.sh new file mode 100644 index 00000000..969ea433 --- /dev/null +++ b/functions/onprem/worker/run.sh @@ -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":""}}' diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index 03aa9ff4..cebab1b5 100644 --- a/functions/onprem/worker/worker.go +++ b/functions/onprem/worker/worker.go @@ -33,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 :) @@ -50,17 +54,19 @@ 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 @@ -137,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") @@ -182,19 +192,22 @@ func shutdown(workflowExecution shuffle.WorkflowExecution, nodeId string, reason os.Exit(3) } else { log.Printf("\n\n[DEBUG] Sending result and resetting values (K8s & Swarm).\n\n") - 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 + //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) @@ -563,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 { @@ -1275,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.") @@ -1375,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, " ", "-") @@ -1394,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 { @@ -1410,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) @@ -1649,7 +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") + //log.Printf("[DEBUG] Got stream workflow queue") body, err := ioutil.ReadAll(request.Body) if err != nil { log.Println("(3) Failed reading body for workflowqueue") @@ -1693,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 } @@ -1863,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 { @@ -1883,7 +1934,7 @@ func validateFinished(workflowExecution shuffle.WorkflowExecution) { } func handleGetStreamResults(resp http.ResponseWriter, request *http.Request) { - log.Printf("[DEBUG] Got stream result") + //log.Printf("[DEBUG] Got stream result") body, err := ioutil.ReadAll(request.Body) if err != nil { log.Println("Failed reading body for stream result queue") @@ -1998,7 +2049,7 @@ 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 %d with hostname: %s\n\n", baseport, hostname) appCallbackUrl = fmt.Sprintf("http://%s:%d", hostname, baseport) @@ -2091,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 } @@ -2232,9 +2283,28 @@ func findAppInfo(image, name string) (int, error) { serviceListOptions, ) + // Basic self-correction if err != nil { - log.Printf("[ERROR] Unable to list services: %s", err) - return -1, err + 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 { @@ -2393,6 +2463,13 @@ func main() { } */ + _, 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{ @@ -2418,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) @@ -2489,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 @@ -2676,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) @@ -2761,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 @@ -2787,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)