diff --git a/backend/app_sdk/app_base.py b/backend/app_sdk/app_base.py index ff341920..a5e52921 100644 --- a/backend/app_sdk/app_base.py +++ b/backend/app_sdk/app_base.py @@ -327,13 +327,36 @@ class AppBase: print("BEFORE VARIABLES!") if len(baseresult) == 0: - print("Variables: %s" % execution_data["workflow"]["workflow_variables"]) - for variable in execution_data["workflow"]["workflow_variables"]: - variablename = variable["name"].replace(" ", "_", -1).lower() + try: + print("WF Variables: %s" % execution_data["workflow"]["workflow_variables"]) + for variable in execution_data["workflow"]["workflow_variables"]: + variablename = variable["name"].replace(" ", "_", -1).lower() - if variablename.lower() == actionname_lower: - baseresult = variable["value"] - break + if variablename.lower() == actionname_lower: + baseresult = variable["value"] + break + except KeyError as e: + print("KeyError wf variables: %s" % e) + pass + except TypeError as e: + print("TypeError wf variables: %s" % e) + pass + + print("BEFORE EXECUTION VAR") + if len(baseresult) == 0: + try: + print("Execution Variables: %s" % execution_data["execution_variables"]) + for variable in execution_data["execution_variables"]: + variablename = variable["name"].replace(" ", "_", -1).lower() + if variablename.lower() == actionname_lower: + baseresult = variable["value"] + break + except KeyError as e: + print("KeyError exec variables: %s" % e) + pass + except TypeError as e: + print("TypeError exec variables: %s" % e) + pass except KeyError as error: print(f"KeyError in JSON: {error}") @@ -421,10 +444,33 @@ class AppBase: if parameter["variant"] == "WORKFLOW_VARIABLE": - for item in fullexecution["workflow"]["workflow_variables"]: - if parameter["action_field"] == item["name"]: - parameter["value"] = item["value"] - break + print("Handling workflow variable") + found = False + try: + for item in fullexecution["workflow"]["workflow_variables"]: + if parameter["action_field"] == item["name"]: + found = True + parameter["value"] = item["value"] + break + except KeyError as e: + print("KeyError WF variable 1: %s" % e) + pass + except TypeError as e: + print("TypeError WF variables 1: %s" % e) + pass + + if not found: + try: + for item in fullexecution["execution_variables"]: + if parameter["action_field"] == item["name"]: + parameter["value"] = item["value"] + break + except KeyError as e: + print("KeyError WF variable 2: %s" % e) + pass + except TypeError as e: + print("TypeError WF variables 2: %s" % e) + pass elif parameter["variant"] == "ACTION_RESULT": # FIXME - calculate value based on action_field and $if prominent diff --git a/backend/go-app/docker.go b/backend/go-app/docker.go index 4aaab56f..c42348bc 100644 --- a/backend/go-app/docker.go +++ b/backend/go-app/docker.go @@ -191,13 +191,14 @@ func buildImageMemory(fs billy.Filesystem, tags []string, dockerfileFolder strin ) //log.Printf("IMAGERESPONSE: %#v", imageBuildResponse.Body) + defer imageBuildResponse.Body.Close() + _, newerr := io.Copy(os.Stdout, imageBuildResponse.Body) + if newerr != nil { + log.Printf("Failed reading Docker build STDOUT: %s", newerr) + } + if err != nil { // Read the STDOUT from the build process - defer imageBuildResponse.Body.Close() - _, newerr := io.Copy(os.Stdout, imageBuildResponse.Body) - if newerr != nil { - log.Printf("Failed reading Docker build STDOUT: %s", newerr) - } return err } diff --git a/backend/go-app/main.go b/backend/go-app/main.go index d719b4c9..cf54d043 100644 --- a/backend/go-app/main.go +++ b/backend/go-app/main.go @@ -2148,7 +2148,7 @@ func getApikey(ctx context.Context, apikey string) (User, error) { var users []User _, err := dbclient.GetAll(ctx, q, &users) if err != nil { - log.Printf("Error getting users apikey: %s", err) + log.Printf("Error getting users apikey (getapikey): %s", err) return User{}, err } @@ -5969,7 +5969,7 @@ func runInit(ctx context.Context) { var users []User _, err = dbclient.GetAll(ctx, q, &users) if err != nil { - log.Printf("Error getting users apikey: %s", err) + log.Printf("Error getting users apikey (runinit): %s", err) } else { if len(users) == 0 { log.Printf("No active users found - setting ALL to active") @@ -6030,6 +6030,7 @@ func runInit(ctx context.Context) { } else { log.Printf("Setting up %d schedule(s)", len(schedules)) for _, schedule := range schedules { + //log.Printf("Schedule: %#v", schedule) job := func() { request := &http.Request{ Method: "POST", diff --git a/backend/go-app/walkoff.go b/backend/go-app/walkoff.go index 14fd6336..2499c225 100644 --- a/backend/go-app/walkoff.go +++ b/backend/go-app/walkoff.go @@ -138,8 +138,7 @@ type WorkflowAppAction struct { ID string `json:"id" datastore:"id"` Name string `json:"name" datastore:"name"` Value string `json:"value" datastore:"value"` - } `json:"execution_variables" datastore:"execution_variables"` - + } `json:"execution_variable" datastore:"execution_variables"` Returns struct { Description string `json:"description" datastore:"returns" yaml:"description,omitempty"` ID string `json:"id" datastore:"id" yaml:"id,omitempty"` @@ -149,41 +148,53 @@ type WorkflowAppAction struct { // FIXME: Generate a callback authentication ID? type WorkflowExecution struct { - Type string `json:"type" datastore:"type"` - Status string `json:"status" datastore:"status"` - Start string `json:"start" datastore:"start"` - ExecutionArgument string `json:"execution_argument" datastore:"execution_argument"` - ExecutionId string `json:"execution_id" datastore:"execution_id"` - WorkflowId string `json:"workflow_id" datastore:"workflow_id"` - LastNode string `json:"last_node" datastore:"last_node"` - Authorization string `json:"authorization" datastore:"authorization"` - Result string `json:"result" datastore:"result,noindex"` - StartedAt int64 `json:"started_at" datastore:"started_at"` - CompletedAt int64 `json:"completed_at" datastore:"completed_at"` - ProjectId string `json:"project_id" datastore:"project_id"` - Locations []string `json:"locations" datastore:"locations"` - Workflow Workflow `json:"workflow" datastore:"workflow,noindex"` - Results []ActionResult `json:"results" datastore:"results,noindex"` + Type string `json:"type" datastore:"type"` + Status string `json:"status" datastore:"status"` + Start string `json:"start" datastore:"start"` + ExecutionArgument string `json:"execution_argument" datastore:"execution_argument"` + ExecutionId string `json:"execution_id" datastore:"execution_id"` + WorkflowId string `json:"workflow_id" datastore:"workflow_id"` + LastNode string `json:"last_node" datastore:"last_node"` + Authorization string `json:"authorization" datastore:"authorization"` + Result string `json:"result" datastore:"result,noindex"` + StartedAt int64 `json:"started_at" datastore:"started_at"` + CompletedAt int64 `json:"completed_at" datastore:"completed_at"` + ProjectId string `json:"project_id" datastore:"project_id"` + Locations []string `json:"locations" datastore:"locations"` + Workflow Workflow `json:"workflow" datastore:"workflow,noindex"` + Results []ActionResult `json:"results" datastore:"results,noindex"` + ExecutionVariables []struct { + Description string `json:"description" datastore:"description"` + ID string `json:"id" datastore:"id"` + Name string `json:"name" datastore:"name"` + Value string `json:"value" datastore:"value"` + } `json:"execution_variables,omitempty" datastore:"execution_variables,omitempty"` } // Added environment for location to execute type Action struct { - AppName string `json:"app_name" datastore:"app_name"` - AppVersion string `json:"app_version" datastore:"app_version"` - AppID string `json:"app_id" datastore:"app_id"` - Errors []string `json:"errors" datastore:"errors"` - ID string `json:"id" datastore:"id"` - IsValid bool `json:"is_valid" datastore:"is_valid"` - IsStartNode bool `json:"isStartNode" datastore:"isStartNode"` - Sharing bool `json:"sharing" datastore:"sharing"` - PrivateID string `json:"private_id" datastore:"private_id"` - Label string `json:"label" datastore:"label"` - SmallImage string `json:"small_image" datastore:"small_image,noindex" required:false yaml:"small_image"` - LargeImage string `json:"large_image" datastore:"large_image,noindex" yaml:"large_image" required:false` - Environment string `json:"environment" datastore:"environment"` - Name string `json:"name" datastore:"name"` - Parameters []WorkflowAppActionParameter `json:"parameters" datastore: "parameters,noindex"` - Position struct { + AppName string `json:"app_name" datastore:"app_name"` + AppVersion string `json:"app_version" datastore:"app_version"` + AppID string `json:"app_id" datastore:"app_id"` + Errors []string `json:"errors" datastore:"errors"` + ID string `json:"id" datastore:"id"` + IsValid bool `json:"is_valid" datastore:"is_valid"` + IsStartNode bool `json:"isStartNode" datastore:"isStartNode"` + Sharing bool `json:"sharing" datastore:"sharing"` + PrivateID string `json:"private_id" datastore:"private_id"` + Label string `json:"label" datastore:"label"` + SmallImage string `json:"small_image" datastore:"small_image,noindex" required:false yaml:"small_image"` + LargeImage string `json:"large_image" datastore:"large_image,noindex" yaml:"large_image" required:false` + Environment string `json:"environment" datastore:"environment"` + Name string `json:"name" datastore:"name"` + Parameters []WorkflowAppActionParameter `json:"parameters" datastore: "parameters,noindex"` + ExecutionVariable struct { + Description string `json:"description" datastore:"description"` + ID string `json:"id" datastore:"id"` + Name string `json:"name" datastore:"name"` + Value string `json:"value" datastore:"value"` + } `json:"execution_variable,omitempty" datastore:"execution_variable,omitempty"` + Position struct { X float64 `json:"x" datastore:"x"` Y float64 `json:"y" datastore:"y"` } `json:"position"` @@ -265,7 +276,7 @@ type Workflow struct { ID string `json:"id" datastore:"id"` Name string `json:"name" datastore:"name"` Value string `json:"value" datastore:"value"` - } `json:"execution_variables" datastore:"execution_variables"` + } `json:"execution_variables,omitempty" datastore:"execution_variables,omitempty"` } type ActionResult struct { @@ -796,13 +807,20 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) { } if found { - // FIXME - this is broken, but why - //if workflowExecution.Results[outerindex].Status == actionResult.Status { - // log.Printf("Status of %s is already %s", actionResult.Action.ID, actionResult.Status) - // resp.WriteHeader(401) - // resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Status of %s is already %s"}`, actionResult.Action.ID, actionResult.Status))) - // return - //} + // If result exists and execution variable exists, update execution value + //log.Printf("Exec var backend: %s", workflowExecution.Results[outerindex].Action.ExecutionVariable.Name) + actionVarName := workflowExecution.Results[outerindex].Action.ExecutionVariable.Name + // Finds potential execution arguments + if len(actionVarName) > 0 { + log.Printf("EXECUTION VARIABLE LOCAL: %s", actionVarName) + for index, execvar := range workflowExecution.ExecutionVariables { + if execvar.Name == actionVarName { + // Sets the value for the variable + workflowExecution.ExecutionVariables[index].Value = actionResult.Result + break + } + } + } log.Printf("Updating %s in %s from %s to %s", actionResult.Action.ID, workflowExecution.ExecutionId, workflowExecution.Results[outerindex].Status, actionResult.Status) workflowExecution.Results[outerindex] = actionResult @@ -1013,6 +1031,7 @@ func setNewWorkflow(resp http.ResponseWriter, request *http.Request) { if len(newActions) == 0 { log.Printf("APPENDING NEW APP FOR NEW WORKFLOW") + // Adds the Testing app if it's a new workflow workflowapps, err := getAllWorkflowApps(ctx) if err == nil { for _, item := range workflowapps { @@ -1046,7 +1065,7 @@ func setNewWorkflow(resp http.ResponseWriter, request *http.Request) { } } } else { - log.Printf("Has actions already?") + log.Printf("Has %d actions already", len(newActions)) } workflow.Actions = newActions @@ -1412,6 +1431,8 @@ func saveWorkflow(resp http.ResponseWriter, request *http.Request) { "0ca8887e-b4af-4e3e-887c-87e9d3bc3d3e", } + log.Printf("%s Action execution var: %s", action.Label, action.ExecutionVariable.Name) + builtin := false for _, id := range reservedApps { if id == action.AppID { @@ -1935,6 +1956,8 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf // Status for the entire workflow. workflowExecution.Status = "EXECUTING" } + + workflowExecution.ExecutionVariables = workflow.ExecutionVariables // Local authorization for this single workflow used in workers. // FIXME: Used for cloud @@ -2095,7 +2118,6 @@ func executeWorkflow(resp http.ResponseWriter, request *http.Request) { } func stopSchedule(resp http.ResponseWriter, request *http.Request) { - log.Printf("Delete?") cors := handleCors(resp, request) if cors { return @@ -2262,20 +2284,19 @@ func stopScheduleGCP(resp http.ResponseWriter, request *http.Request) { func deleteSchedule(ctx context.Context, id string) error { log.Printf("Should stop schedule %s!", id) - //newscheduler "github.com/carlescere/scheduler" - log.Printf("Schedules: %#v", scheduledJobs) - if value, exists := scheduledJobs[id]; exists { - log.Printf("STOP THIS ONE: %s", value) - // Looks like this does the trick? Hurr - value.Lock() - err := DeleteKey(ctx, "schedules", id) - if err != nil { - log.Printf("Failed to delete schedule: %s", err) - return err - } + err := DeleteKey(ctx, "schedules", id) + if err != nil { + log.Printf("Failed to delete schedule: %s", err) + return err } else { - // FIXME - allow it to kind of stop anyway? - return errors.New("Can't find the schedule.") + if value, exists := scheduledJobs[id]; exists { + log.Printf("STOP THIS ONE: %s", value) + // Looks like this does the trick? Hurr + value.Lock() + } else { + // FIXME - allow it to kind of stop anyway? + return errors.New("Can't find the schedule.") + } } return nil @@ -2717,7 +2738,7 @@ func deleteUser(resp http.ResponseWriter, request *http.Request) { var users []User _, err := dbclient.GetAll(ctx, q, &users) if err != nil { - log.Printf("Error getting users apikey: %s", err) + log.Printf("Error getting users apikey (deleteuser): %s", err) resp.WriteHeader(401) resp.Write([]byte(`{"success": false, "reason": "Failed getting users for verification"}`)) return diff --git a/docker-compose.yml b/docker-compose.yml index ba3a495b..288063bb 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,7 +1,7 @@ version: '3' services: frontend: - build: ./frontend + #build: ./frontend image: frikky/shuffle:frontend container_name: shuffle-frontend hostname: shuffle-frontend @@ -14,7 +14,7 @@ services: depends_on: - backend backend: - build: ./backend + #build: ./backend image: frikky/shuffle:backend container_name: shuffle-backend hostname: ${BACKEND_HOSTNAME} diff --git a/frontend/src/Admin.js b/frontend/src/Admin.js index 8e25f77f..06f14f64 100644 --- a/frontend/src/Admin.js +++ b/frontend/src/Admin.js @@ -43,7 +43,7 @@ const Admin = (props) => { console.log("INPUT: ", data) // Just use this one? - const url = globalUrl+'/api/v1/workflows/'+data["workflow_id datastore:"]+"/schedule/"+data.id + const url = globalUrl+'/api/v1/workflows/'+data["workflow_id"]+"/schedule/"+data.id console.log("URL: ", url) fetch(url, { method: 'DELETE', @@ -54,6 +54,7 @@ const Admin = (props) => { }) .then(response => response.json().then(responseJson => { + console.log("RESP: ", responseJson) if (responseJson["success"] === false) { alert.error("Failed stopping schedule") } else { diff --git a/frontend/src/AngularWorkflow.js b/frontend/src/AngularWorkflow.js index 61c2428c..c7eb31a1 100644 --- a/frontend/src/AngularWorkflow.js +++ b/frontend/src/AngularWorkflow.js @@ -112,6 +112,7 @@ const AngularWorkflow = (props) => { const [appAuthentication, setAppAuthentication] = React.useState({}); const [variablesModalOpen, setVariablesModalOpen] = React.useState(false); + const [executionVariablesModalOpen, setExecutionVariablesModalOpen] = React.useState(false); const [authenticationModalOpen, setAuthenticationModalOpen] = React.useState(false); const [conditionsModalOpen, setConditionsModalOpen] = React.useState(false); const [newVariableName, setNewVariableName] = React.useState(""); @@ -791,7 +792,6 @@ const AngularWorkflow = (props) => { const onNodeSelect = (event) => { const data = event.target.data() - console.log("NODE: ", data) setLastSaved(false) //console.log(data) @@ -1412,22 +1412,6 @@ const AngularWorkflow = (props) => { const [open, setOpen] = React.useState(false); const [anchorEl, setAnchorEl] = React.useState(null); - if (workflow.workflow_variables === undefined || workflow.workflow_variables === null || workflow.workflow_variables.length === 0) { - return ( -
-
-
- Looks like you don't have any variables yet. -
-
- -
-
-
-
- ) - } - const menuClick = (event) => { setOpen(!open) setAnchorEl(event.currentTarget); @@ -1438,8 +1422,13 @@ const AngularWorkflow = (props) => { setWorkflow(workflow) } + const deleteExecutionVariable = (variableName) => { + workflow.execution_variables = workflow.execution_variables.filter(data => data.name !== variableName) + setWorkflow(workflow) + } + const variableScrollStyle = { - marginTop: "10px", + margin: 15, overflow: "scroll", height: "66vh", overflowX: "auto", @@ -1450,7 +1439,9 @@ const AngularWorkflow = (props) => { return (
- {workflow.workflow_variables.map(variable=> { + What are WORKFLOW variables? + {workflow.workflow_variables === null ? + null : workflow.workflow_variables.map(variable=> { return (
{ @@ -1508,10 +1499,69 @@ const AngularWorkflow = (props) => {
) })} +
+ +
+ + What are EXECUTION variables? + {workflow.execution_variables === null ? + null : workflow.execution_variables.map(variable=> { + return ( +
+ { + }}> +
+
+
{ + setNewVariableName(variable.name) + setExecutionVariablesModalOpen(true)}}> + Name: {variable.name} +
+
+ + + + { + setOpen(false) + setAnchorEl(null) + }} + > -
-
- + { + setOpen(false) + setNewVariableName(variable.name) + setExecutionVariablesModalOpen(true) + }} key={"Edit"}>{"Edit"} + { + deleteExecutionVariable(variable.name) + setOpen(false) + }} key={"Delete"}>{"Delete"} + +
+
+ +
+ ) + })} +
+ +
) @@ -1862,6 +1912,7 @@ const AngularWorkflow = (props) => { isStartNode: false, large_image: app.large_image, authentication: [], + execution_variable: undefined, } // const image = "url("+app.large_image+")" @@ -2361,7 +2412,7 @@ const AngularWorkflow = (props) => { } else if (data.variant === "WORKFLOW_VARIABLE") { varcolor = "#f85a3e" - if (workflow.workflow_variables === null || workflow.workflow_variables === undefined || workflow.workflow_variables.length === 0) { + if ((workflow.workflow_variables === null || workflow.workflow_variables === undefined || workflow.workflow_variables.length === 0) && (workflow.execution_variables === null || workflow.execution_variables === undefined || workflow.execution_variables.length === 0)) { setCurrentView(2) datafield =
@@ -2389,15 +2440,22 @@ const AngularWorkflow = (props) => { fullWidth value={selectedAction.parameters[count].action_field} onChange={(e) => { + console.log(e.target.value) changeActionParameterVariable(e.target.value, count) }} style={{backgroundColor: inputColor, color: "white", height: "50px"}} > - {workflow.workflow_variables.map(data => ( + {workflow.workflow_variables !== null ? workflow.workflow_variables.map(data => ( {data.name} - ))} + )) : null} + + {workflow.execution_variables !== null ? workflow.execution_variables.map(data => ( + + {data.name} + + )) : null} } @@ -2544,7 +2602,7 @@ const AngularWorkflow = (props) => { placeholder={selectedAction.label} onChange={selectedNameChange} /> - {environments !== undefined && environments.length > 1 ? + {environments !== undefined && environments !== null && environments.length > 1 ?
Environment
: null} + {workflow.execution_variables !== undefined && workflow.execution_variables !== null && workflow.execution_variables.length > 0 ? +
+ Execution variables + +
+ : null} {/*requiresAuthentication ?
+ + + {workflowExecutions.length > 0 ? + + + Values from last 3 executions + {workflowExecutions.slice(0,3).map((execution, index) => { + if (execution.execution_variables === undefined || execution.execution_variables === null || execution.execution_variables === 0) { + return null + } + + const variable = execution.execution_variables.find(data => data.name === newVariableName) + if (variable === undefined || variable.value === undefined) { + return null + } + + return ( +
+ {index+1}: {variable.value} +
+ ) + })} + + : null + } + + + : null const variablesModal = variablesModalOpen ? {
{newView} {variablesModal} + {executionVariableModal} {conditionsModal} {authenticationModal}
diff --git a/frontend/src/App.js b/frontend/src/App.js index af034001..891371d0 100644 --- a/frontend/src/App.js +++ b/frontend/src/App.js @@ -90,13 +90,11 @@ const App = (message, props) => { }) .then(response => response.json()) .then(responseJson => { - console.log(responseJson) if (responseJson.success === true) { setUserData(responseJson) setIsLoggedIn(true) // Updating cookie every request - console.log("COOKIES: ", cookies) for (var key in responseJson["cookies"]) { setCookie(responseJson["cookies"][key].key, responseJson["cookies"][key].value, {path: "/"}) } diff --git a/frontend/src/Apps.js b/frontend/src/Apps.js index b0f608c3..8bdb587f 100644 --- a/frontend/src/Apps.js +++ b/frontend/src/Apps.js @@ -905,6 +905,7 @@ const Apps = (props) => { : null + const circularLoader = validation ? : null const appsModalLoad = loadAppsModalOpen ? { : null const errorText = openApiError.length > 0 ?
Error: {openApiError}
: null - const circularLoader = validation ? : null const modalView = openApiModal ? workerTimeout { - zombiecheck() + go zombiecheck() zombiecounter = 0 } time.Sleep(time.Duration(sleepTime) * time.Second) @@ -257,7 +282,7 @@ func main() { log.Printf("Failed reading body: %s", err) zombiecounter += 1 if zombiecounter*sleepTime > workerTimeout { - zombiecheck() + go zombiecheck() zombiecounter = 0 } time.Sleep(time.Duration(sleepTime) * time.Second) @@ -271,7 +296,7 @@ func main() { sleepTime = 10 zombiecounter += 1 if zombiecounter*sleepTime > workerTimeout { - zombiecheck() + go zombiecheck() zombiecounter = 0 } time.Sleep(time.Duration(sleepTime) * time.Second) @@ -286,7 +311,7 @@ func main() { if len(executionRequests.Data) == 0 { zombiecounter += 1 if zombiecounter*sleepTime > workerTimeout { - zombiecheck() + go zombiecheck() zombiecounter = 0 } time.Sleep(time.Duration(sleepTime) * time.Second) @@ -322,34 +347,9 @@ func main() { env = append(env, fmt.Sprintf("DOCKER_API_VERSION=%s", dockerApiVersion)) } - err = deployWorker(dockercli, workerImage, containerName, env) - if err != nil { - stats, err := dockercli.ContainerInspect(context.Background(), containerName) - if err != nil { - log.Printf("Failed checking worker %s", execution.ExecutionId) - continue - } + go deployWorker(dockercli, workerImage, containerName, env) - containerStatus := stats.ContainerJSONBase.State.Status - if containerStatus != "running" { - log.Printf("Status of %s is %s. Should be running. Will reset", containerName, containerStatus) - err = stopWorker(containerName) - if err != nil { - log.Printf("Failed stopping worker %s", execution.ExecutionId) - continue - } - - err = deployWorker(dockercli, workerImage, containerName, env) - if err != nil { - log.Printf("Failed executing worker %s in state %s", execution.ExecutionId, containerStatus) - } - } else { - // Should basically never hit here rofl - log.Printf("ERROR: I HAVE NO IDEA WHAT WENT WRONG. CHECK %s", containerName) - } - } - - log.Printf("%s is deployed and to being removed from queue.", execution.ExecutionId) + log.Printf("%s is deployed and to be removed from queue.", execution.ExecutionId) zombiecounter += 1 toBeRemoved.Data = append(toBeRemoved.Data, execution) } @@ -427,34 +427,54 @@ func zombiecheck() error { All: true, }) + containerNames := map[string]string{} + stopContainers := []string{} removeContainers := []string{} for _, container := range containers { + + // Skip random containers. Only handle things related to Shuffle. + if !strings.Contains(container.Image, baseimagename) { + shuffleFound := false + for _, item := range container.Labels { + if item == "shuffle" { + shuffleFound = true + break + } + } + + // Check image name + if !shuffleFound { + continue + } + } + for _, name := range container.Names { // FIXME - add name_version_uid_uid regex check as well - if !strings.HasPrefix(name, "/worker") { + if strings.HasPrefix(name, "/shuffle") { continue } if container.State != "running" { removeContainers = append(removeContainers, container.ID) + containerNames[container.ID] = name } // stopcontainer & removecontainer currenttime := time.Now().Unix() + //log.Printf("Time: %d - %d", currenttime-container.Created, int64(workerTimeout)) if container.State == "running" && currenttime-container.Created > int64(workerTimeout) { stopContainers = append(stopContainers, container.ID) + containerNames[container.ID] = name } } } // FIXME - add killing of apps with same execution ID too for _, containername := range stopContainers { - if err := dockercli.ContainerStop(ctx, containername, nil); err != nil { - log.Printf("Unable to stop container: %s", err) - } else { - log.Printf("Stopped container %s", containername) - } + log.Printf("Stopping and removing container %s", containerNames[containername]) + go dockercli.ContainerStop(ctx, containername, nil) + removeContainers = append(removeContainers, containername) } removeOptions := types.ContainerRemoveOptions{ @@ -463,11 +483,7 @@ func zombiecheck() error { } for _, containername := range removeContainers { - if err := dockercli.ContainerRemove(ctx, containername, removeOptions); err != nil { - log.Printf("Unable to remove container: %s", err) - } else { - log.Printf("Removed container %s", containername) - } + go dockercli.ContainerRemove(ctx, containername, removeOptions) } return nil diff --git a/functions/onprem/worker/worker.bin b/functions/onprem/worker/worker.bin index 94382e0e..20c38881 100755 Binary files a/functions/onprem/worker/worker.bin and b/functions/onprem/worker/worker.bin differ