Fixed worker in swarm using user input
This commit is contained in:
Frikky
@@ -6,7 +6,7 @@ require (
|
||||
github.com/docker/docker v26.1.0+incompatible
|
||||
github.com/gorilla/mux v1.8.1
|
||||
github.com/satori/go.uuid v1.2.0
|
||||
github.com/shuffle/shuffle-shared v0.6.16
|
||||
github.com/shuffle/shuffle-shared v0.6.24
|
||||
k8s.io/api v0.30.0
|
||||
k8s.io/apimachinery v0.30.0
|
||||
k8s.io/client-go v0.30.0
|
||||
|
||||
+214
-213
@@ -2041,20 +2041,20 @@ runWorkflowExecutionTransaction(ctx, 0, workflowExecution.ExecutionId, actionRes
|
||||
|
||||
// Will make sure transactions are always ran for an execution. This is recursive if it fails. Allowed to fail up to 5 times
|
||||
func runWorkflowExecutionTransaction(ctx context.Context, attempts int64, workflowExecutionId string, actionResult shuffle.ActionResult, resp http.ResponseWriter) {
|
||||
//log.Printf("[DEBUG][%s] IN WORKFLOWEXECUTION SUB!", actionResult.ExecutionId)
|
||||
workflowExecution, err := shuffle.GetWorkflowExecution(ctx, workflowExecutionId)
|
||||
if err != nil {
|
||||
//log.Printf("[DEBUG][%s] IN WORKFLOWEXECUTION SUB!", actionResult.ExecutionId)
|
||||
workflowExecution, err := shuffle.GetWorkflowExecution(ctx, workflowExecutionId)
|
||||
if err != nil {
|
||||
log.Printf("[ERROR] Failed getting execution cache: %s", err)
|
||||
resp.WriteHeader(400)
|
||||
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed getting execution"}`)))
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
resultLength := len(workflowExecution.Results)
|
||||
setExecution := true
|
||||
resultLength := len(workflowExecution.Results)
|
||||
setExecution := true
|
||||
|
||||
workflowExecution, dbSave, err := shuffle.ParsedExecutionResult(ctx, *workflowExecution, actionResult, true, 0)
|
||||
if err == nil {
|
||||
workflowExecution, dbSave, err := shuffle.ParsedExecutionResult(ctx, *workflowExecution, actionResult, true, 0)
|
||||
if err == nil {
|
||||
if workflowExecution.Status != "EXECUTING" && workflowExecution.Status != "WAITING" {
|
||||
log.Printf("[WARNING][%s] Execution is not executing, but %s. Stopping Transaction update.", workflowExecution.ExecutionId, workflowExecution.Status)
|
||||
if resp != nil {
|
||||
@@ -2075,7 +2075,8 @@ if err == nil {
|
||||
shutdown(*workflowExecution, "", "", false)
|
||||
return
|
||||
}
|
||||
} else {
|
||||
|
||||
} else {
|
||||
if strings.Contains(strings.ToLower(fmt.Sprintf("%s", err)), "already been ran") || strings.Contains(strings.ToLower(fmt.Sprintf("%s", err)), "already finished") {
|
||||
log.Printf("[ERROR][%s] Skipping rerun of action result as it's already been ran: %s", workflowExecution.ExecutionId)
|
||||
return
|
||||
@@ -2109,20 +2110,20 @@ if err == nil {
|
||||
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed getting execution"}`)))
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
//log.Printf(`[DEBUG][%s] Got result %s from %s. Execution status: %s. Save: %#v. Parent: %#v`, actionResult.ExecutionId, actionResult.Status, actionResult.Action.ID, workflowExecution.Status, dbSave, workflowExecution.ExecutionParent)
|
||||
//dbSave := false
|
||||
//log.Printf(`[DEBUG][%s] Got result %s from %s. Execution status: %s. Save: %#v. Parent: %#v`, actionResult.ExecutionId, actionResult.Status, actionResult.Action.ID, workflowExecution.Status, dbSave, workflowExecution.ExecutionParent)
|
||||
//dbSave := false
|
||||
|
||||
//if len(results) != len(workflowExecution.Results) {
|
||||
// log.Printf("[DEBUG][%s] There may have been an issue in transaction queue. Result lengths: %d vs %d. Should check which exists the base results, but not in entire execution, then append.", workflowExecution.ExecutionId, len(results), len(workflowExecution.Results))
|
||||
//}
|
||||
//if len(results) != len(workflowExecution.Results) {
|
||||
// log.Printf("[DEBUG][%s] There may have been an issue in transaction queue. Result lengths: %d vs %d. Should check which exists the base results, but not in entire execution, then append.", workflowExecution.ExecutionId, len(results), len(workflowExecution.Results))
|
||||
//}
|
||||
|
||||
// Validating that action results hasn't changed
|
||||
// Handled using cachhing, so actually pretty fast
|
||||
cacheKey := fmt.Sprintf("workflowexecution_%s", workflowExecution.ExecutionId)
|
||||
cache, err := shuffle.GetCache(ctx, cacheKey)
|
||||
if err == nil {
|
||||
// Validating that action results hasn't changed
|
||||
// Handled using cachhing, so actually pretty fast
|
||||
cacheKey := fmt.Sprintf("workflowexecution_%s", workflowExecution.ExecutionId)
|
||||
cache, err := shuffle.GetCache(ctx, cacheKey)
|
||||
if err == nil {
|
||||
//parsedValue := value.(*shuffle.WorkflowExecution)
|
||||
|
||||
parsedValue := &shuffle.WorkflowExecution{}
|
||||
@@ -2147,9 +2148,9 @@ if err == nil {
|
||||
}
|
||||
*/
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if setExecution || workflowExecution.Status == "FINISHED" || workflowExecution.Status == "ABORTED" || workflowExecution.Status == "FAILURE" {
|
||||
if setExecution || workflowExecution.Status == "FINISHED" || workflowExecution.Status == "ABORTED" || workflowExecution.Status == "FAILURE" {
|
||||
log.Printf("[DEBUG][%s] Running setexec with status %s and %d/%d results", workflowExecution.ExecutionId, workflowExecution.Status, len(workflowExecution.Results), len(workflowExecution.Workflow.Actions))
|
||||
//result(s)", workflowExecution.ExecutionId, workflowExecution.Status, len(workflowExecution.Results))
|
||||
err = setWorkflowExecution(ctx, *workflowExecution, dbSave)
|
||||
@@ -2159,65 +2160,65 @@ if setExecution || workflowExecution.Status == "FINISHED" || workflowExecution.S
|
||||
return
|
||||
}
|
||||
|
||||
} else {
|
||||
} else {
|
||||
log.Printf("[INFO][%s] Skipping setexec with status %s", workflowExecution.ExecutionId, workflowExecution.Status)
|
||||
|
||||
// Just in case. Should MAYBE validate finishing another time as well.
|
||||
// This fixes issues with e.g. shuffle.Action -> shuffle.Trigger -> shuffle.Action.
|
||||
handleExecutionResult(*workflowExecution)
|
||||
}
|
||||
}
|
||||
|
||||
//if newExecutions && len(nextActions) > 0 {
|
||||
// log.Printf("[DEBUG][%s] New execution: %#v. NextActions: %#v", newExecutions, nextActions)
|
||||
// //handleExecutionResult(*workflowExecution)
|
||||
//}
|
||||
//if newExecutions && len(nextActions) > 0 {
|
||||
// log.Printf("[DEBUG][%s] New execution: %#v. NextActions: %#v", newExecutions, nextActions)
|
||||
// //handleExecutionResult(*workflowExecution)
|
||||
//}
|
||||
|
||||
resp.WriteHeader(200)
|
||||
resp.Write([]byte(fmt.Sprintf(`{"success": true}`)))
|
||||
resp.WriteHeader(200)
|
||||
resp.Write([]byte(fmt.Sprintf(`{"success": true}`)))
|
||||
}
|
||||
|
||||
func sendSelfRequest(actionResult shuffle.ActionResult) {
|
||||
|
||||
|
||||
data, err := json.Marshal(actionResult)
|
||||
if err != nil {
|
||||
data, err := json.Marshal(actionResult)
|
||||
if err != nil {
|
||||
log.Printf("[ERROR][%s] Shutting down (24): Failed to unmarshal data for backend: %s", actionResult.ExecutionId, err)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
if actionResult.ExecutionId == "TBD" {
|
||||
if actionResult.ExecutionId == "TBD" {
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
log.Printf("[DEBUG][%s] Sending FAILURE to self to stop the workflow execution. Action: %s (%s), app %s:%s", actionResult.ExecutionId, actionResult.Action.Label, actionResult.Action.ID, actionResult.Action.AppName, actionResult.Action.AppVersion)
|
||||
log.Printf("[DEBUG][%s] Sending FAILURE to self to stop the workflow execution. Action: %s (%s), app %s:%s", actionResult.ExecutionId, actionResult.Action.Label, actionResult.Action.ID, actionResult.Action.AppName, actionResult.Action.AppVersion)
|
||||
|
||||
// Literally sending to same worker to run it as a new request
|
||||
streamUrl := fmt.Sprintf("http://localhost:33333/api/v1/streams")
|
||||
hostenv := os.Getenv("WORKER_HOSTNAME")
|
||||
if len(hostenv) > 0 {
|
||||
// Literally sending to same worker to run it as a new request
|
||||
streamUrl := fmt.Sprintf("http://localhost:33333/api/v1/streams")
|
||||
hostenv := os.Getenv("WORKER_HOSTNAME")
|
||||
if len(hostenv) > 0 {
|
||||
streamUrl = fmt.Sprintf("http://%s:33333/api/v1/streams", hostenv)
|
||||
}
|
||||
}
|
||||
|
||||
req, err := http.NewRequest(
|
||||
req, err := http.NewRequest(
|
||||
"POST",
|
||||
streamUrl,
|
||||
bytes.NewBuffer([]byte(data)),
|
||||
)
|
||||
)
|
||||
|
||||
if err != nil {
|
||||
if err != nil {
|
||||
log.Printf("[ERROR][%s] Failed creating self request (1): %s", actionResult.ExecutionId, err)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
client := shuffle.GetExternalClient(streamUrl)
|
||||
newresp, err := client.Do(req)
|
||||
if err != nil {
|
||||
client := shuffle.GetExternalClient(streamUrl)
|
||||
newresp, err := client.Do(req)
|
||||
if err != nil {
|
||||
log.Printf("[ERROR][%s] Error running finishing request (2): %s", actionResult.ExecutionId, err)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
defer newresp.Body.Close()
|
||||
if newresp.Body != nil {
|
||||
defer newresp.Body.Close()
|
||||
if newresp.Body != nil {
|
||||
body, err := ioutil.ReadAll(newresp.Body)
|
||||
//log.Printf("[INFO] BACKEND STATUS: %d", newresp.StatusCode)
|
||||
if err != nil {
|
||||
@@ -2225,59 +2226,59 @@ if newresp.Body != nil {
|
||||
} else {
|
||||
log.Printf("[DEBUG][%s] NEWRESP (from backend): %s", actionResult.ExecutionId, string(body))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func sendResult(workflowExecution shuffle.WorkflowExecution, data []byte) {
|
||||
if workflowExecution.ExecutionSource == "default" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "swarm" {
|
||||
if workflowExecution.ExecutionSource == "default" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "swarm" {
|
||||
//log.Printf("[INFO][%s] Not sending backend info since source is default (not swarm)", workflowExecution.ExecutionId)
|
||||
//return
|
||||
} else {
|
||||
}
|
||||
} else {
|
||||
}
|
||||
|
||||
// Basically to reduce backend strain
|
||||
/*
|
||||
if shuffle.ArrayContains(finishedExecutions, workflowExecution.ExecutionId) {
|
||||
// Basically to reduce backend strain
|
||||
/*
|
||||
if shuffle.ArrayContains(finishedExecutions, workflowExecution.ExecutionId) {
|
||||
log.Printf("[INFO][%s] NOT sending backend info since it's already been sent before.", workflowExecution.ExecutionId)
|
||||
return
|
||||
}
|
||||
*/
|
||||
}
|
||||
*/
|
||||
|
||||
// Take it down again
|
||||
/*
|
||||
if len(finishedExecutions) > 100 {
|
||||
// Take it down again
|
||||
/*
|
||||
if len(finishedExecutions) > 100 {
|
||||
log.Printf("[DEBUG][%s] Removing old execution from finishedExecutions: %s", workflowExecution.ExecutionId, finishedExecutions[0])
|
||||
finishedExecutions = finishedExecutions[99:]
|
||||
}
|
||||
}
|
||||
|
||||
finishedExecutions = append(finishedExecutions, workflowExecution.ExecutionId)
|
||||
*/
|
||||
finishedExecutions = append(finishedExecutions, workflowExecution.ExecutionId)
|
||||
*/
|
||||
|
||||
streamUrl := fmt.Sprintf("%s/api/v1/streams", baseUrl)
|
||||
req, err := http.NewRequest(
|
||||
streamUrl := fmt.Sprintf("%s/api/v1/streams", baseUrl)
|
||||
req, err := http.NewRequest(
|
||||
"POST",
|
||||
streamUrl,
|
||||
bytes.NewBuffer([]byte(data)),
|
||||
)
|
||||
)
|
||||
|
||||
if err != nil {
|
||||
if err != nil {
|
||||
log.Printf("[ERROR][%s] Failed creating finishing request: %s", workflowExecution.ExecutionId, err)
|
||||
log.Printf("[DEBUG][%s] Shutting down (22)", workflowExecution.ExecutionId)
|
||||
shutdown(workflowExecution, "", "", false)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
client := shuffle.GetExternalClient(streamUrl)
|
||||
newresp, err := client.Do(req)
|
||||
if err != nil {
|
||||
client := shuffle.GetExternalClient(streamUrl)
|
||||
newresp, err := client.Do(req)
|
||||
if err != nil {
|
||||
log.Printf("[ERROR][%s] Error running finishing request (1): %s", workflowExecution.ExecutionId, err)
|
||||
log.Printf("[DEBUG][%s] Shutting down (23)", workflowExecution.ExecutionId)
|
||||
shutdown(workflowExecution, "", "", false)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
defer newresp.Body.Close()
|
||||
if newresp.Body != nil {
|
||||
defer newresp.Body.Close()
|
||||
if newresp.Body != nil {
|
||||
body, err := ioutil.ReadAll(newresp.Body)
|
||||
//log.Printf("[INFO] BACKEND STATUS: %d", newresp.StatusCode)
|
||||
if err != nil {
|
||||
@@ -2285,27 +2286,27 @@ if newresp.Body != nil {
|
||||
} else {
|
||||
log.Printf("[DEBUG][%s] NEWRESP (from backend): %s", workflowExecution.ExecutionId, string(body))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func validateFinished(workflowExecution shuffle.WorkflowExecution) bool {
|
||||
ctx := context.Background()
|
||||
func validateFinished(workflowExecution shuffle.WorkflowExecution) bool {
|
||||
ctx := context.Background()
|
||||
|
||||
newexec, err := shuffle.GetWorkflowExecution(ctx, workflowExecution.ExecutionId)
|
||||
if err != nil {
|
||||
newexec, err := shuffle.GetWorkflowExecution(ctx, workflowExecution.ExecutionId)
|
||||
if err != nil {
|
||||
log.Printf("[ERROR][%s] Failed getting workflow execution: %s", workflowExecution.ExecutionId, err)
|
||||
return false
|
||||
} else {
|
||||
} else {
|
||||
workflowExecution = *newexec
|
||||
}
|
||||
}
|
||||
|
||||
//startAction, extra, children, parents, visited, executed, nextActions, environments := shuffle.GetExecutionVariables(ctx, workflowExecution.ExecutionId)
|
||||
workflowExecution, _ = shuffle.Fixexecution(ctx, workflowExecution)
|
||||
_, extra, _, _, _, _, _, environments := shuffle.GetExecutionVariables(ctx, workflowExecution.ExecutionId)
|
||||
//startAction, extra, children, parents, visited, executed, nextActions, environments := shuffle.GetExecutionVariables(ctx, workflowExecution.ExecutionId)
|
||||
workflowExecution, _ = shuffle.Fixexecution(ctx, workflowExecution)
|
||||
_, extra, _, _, _, _, _, environments := shuffle.GetExecutionVariables(ctx, workflowExecution.ExecutionId)
|
||||
|
||||
log.Printf("[INFO][%s] VALIDATION. Status: %s, shuffle.Actions: %d, Extra: %d, Results: %d. Parent: %#v", workflowExecution.ExecutionId, workflowExecution.Status, len(workflowExecution.Workflow.Actions), extra, len(workflowExecution.Results), workflowExecution.ExecutionParent)
|
||||
log.Printf("[INFO][%s] VALIDATION. Status: %s, shuffle.Actions: %d, Extra: %d, Results: %d. Parent: %#v", workflowExecution.ExecutionId, workflowExecution.Status, len(workflowExecution.Workflow.Actions), extra, len(workflowExecution.Results), workflowExecution.ExecutionParent)
|
||||
|
||||
if workflowExecution.Status == "FINISHED" || workflowExecution.Status == "ABORTED" || (len(environments) == 1 && requestsSent == 0 && len(workflowExecution.Results) >= 1 && os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "swarm") || (len(workflowExecution.Results) >= len(workflowExecution.Workflow.Actions)+extra && len(workflowExecution.Workflow.Actions) > 0) {
|
||||
if workflowExecution.Status == "FINISHED" || workflowExecution.Status == "ABORTED" || (len(environments) == 1 && requestsSent == 0 && len(workflowExecution.Results) >= 1 && os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "swarm") || (len(workflowExecution.Results) >= len(workflowExecution.Workflow.Actions)+extra && len(workflowExecution.Workflow.Actions) > 0) {
|
||||
|
||||
if workflowExecution.Status == "FINISHED" {
|
||||
for _, result := range workflowExecution.Results {
|
||||
@@ -2337,63 +2338,63 @@ if workflowExecution.Status == "FINISHED" || workflowExecution.Status == "ABORT
|
||||
shuffle.RunCacheCleanup(ctx, workflowExecution)
|
||||
sendResult(workflowExecution, shutdownData)
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
return false
|
||||
return false
|
||||
}
|
||||
|
||||
func handleGetStreamResults(resp http.ResponseWriter, request *http.Request) {
|
||||
defer request.Body.Close()
|
||||
body, err := ioutil.ReadAll(request.Body)
|
||||
if err != nil {
|
||||
defer request.Body.Close()
|
||||
body, err := ioutil.ReadAll(request.Body)
|
||||
if err != nil {
|
||||
log.Printf("[WARNING] Failed reading body for stream result queue")
|
||||
resp.WriteHeader(500)
|
||||
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "%s"}`, err)))
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
var actionResult shuffle.ActionResult
|
||||
err = json.Unmarshal(body, &actionResult)
|
||||
if err != nil {
|
||||
var actionResult shuffle.ActionResult
|
||||
err = json.Unmarshal(body, &actionResult)
|
||||
if err != nil {
|
||||
log.Printf("[WARNING] Failed shuffle.ActionResult unmarshaling: %s", err)
|
||||
//resp.WriteHeader(400)
|
||||
//resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "%s"}`, err)))
|
||||
//return
|
||||
}
|
||||
}
|
||||
|
||||
if len(actionResult.ExecutionId) == 0 {
|
||||
if len(actionResult.ExecutionId) == 0 {
|
||||
log.Printf("[WARNING] No workflow execution id in action result (2). Data: %s", string(body))
|
||||
resp.WriteHeader(400)
|
||||
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "No workflow execution id in action result"}`)))
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
workflowExecution, err := shuffle.GetWorkflowExecution(ctx, actionResult.ExecutionId)
|
||||
if err != nil {
|
||||
ctx := context.Background()
|
||||
workflowExecution, err := shuffle.GetWorkflowExecution(ctx, actionResult.ExecutionId)
|
||||
if err != nil {
|
||||
log.Printf("[INFO] Failed getting execution (streamresult) %s: %s", actionResult.ExecutionId, err)
|
||||
resp.WriteHeader(400)
|
||||
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Bad authorization key or execution_id might not exist."}`)))
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// Authorization is done here
|
||||
if workflowExecution.Authorization != actionResult.Authorization {
|
||||
// Authorization is done here
|
||||
if workflowExecution.Authorization != actionResult.Authorization {
|
||||
log.Printf("[ERROR] Bad authorization key when getting stream results from cache %s.", actionResult.ExecutionId)
|
||||
resp.WriteHeader(401)
|
||||
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Bad authorization key or execution_id might not exist."}`)))
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
newjson, err := json.Marshal(workflowExecution)
|
||||
if err != nil {
|
||||
newjson, err := json.Marshal(workflowExecution)
|
||||
if err != nil {
|
||||
resp.WriteHeader(500)
|
||||
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed unpacking workflow execution"}`)))
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
resp.WriteHeader(200)
|
||||
resp.Write(newjson)
|
||||
resp.WriteHeader(200)
|
||||
resp.Write(newjson)
|
||||
|
||||
}
|
||||
|
||||
@@ -2401,152 +2402,152 @@ resp.Write(newjson)
|
||||
func getLocalIP() string {
|
||||
|
||||
|
||||
addrs, err := net.InterfaceAddrs()
|
||||
if err != nil {
|
||||
addrs, err := net.InterfaceAddrs()
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
}
|
||||
|
||||
for _, address := range addrs {
|
||||
for _, address := range addrs {
|
||||
// check the address type and if it is not a loopback the display it
|
||||
if ipnet, ok := address.(*net.IPNet); ok && !ipnet.IP.IsLoopback() {
|
||||
if ipnet.IP.To4() != nil {
|
||||
return ipnet.IP.String()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return ""
|
||||
return ""
|
||||
}
|
||||
|
||||
func getAvailablePort() (net.Listener, error) {
|
||||
listener, err := net.Listen("tcp", ":0")
|
||||
if err != nil {
|
||||
listener, err := net.Listen("tcp", ":0")
|
||||
if err != nil {
|
||||
log.Printf("[WARNING] Failed to assign port by default. Defaulting to 5001")
|
||||
//return ":5001"
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
//defer listener.Close()
|
||||
//defer listener.Close()
|
||||
|
||||
return listener, nil
|
||||
//return fmt.Sprintf(":%d", port)
|
||||
return listener, nil
|
||||
//return fmt.Sprintf(":%d", port)
|
||||
}
|
||||
|
||||
func webserverSetup(workflowExecution shuffle.WorkflowExecution) net.Listener {
|
||||
hostname = getLocalIP()
|
||||
os.Setenv("WORKER_HOSTNAME", hostname)
|
||||
hostname = getLocalIP()
|
||||
os.Setenv("WORKER_HOSTNAME", hostname)
|
||||
|
||||
// FIXME: This MAY not work because of speed between first
|
||||
// container being launched and port being assigned to webserver
|
||||
listener, err := getAvailablePort()
|
||||
if err != nil {
|
||||
// FIXME: This MAY not work because of speed between first
|
||||
// container being launched and port being assigned to webserver
|
||||
listener, err := getAvailablePort()
|
||||
if err != nil {
|
||||
log.Printf("[ERROR] Failed to create init listener: %s", err)
|
||||
return listener
|
||||
}
|
||||
|
||||
log.Printf("[DEBUG] OLD HOSTNAME: %s", appCallbackUrl)
|
||||
|
||||
|
||||
port := listener.Addr().(*net.TCPAddr).Port
|
||||
// Set the port environment variable
|
||||
os.Setenv("WORKER_PORT", fmt.Sprintf("%d", port))
|
||||
|
||||
log.Printf("[DEBUG] Starting webserver (2) on port %d with hostname: %s", port, hostname)
|
||||
appCallbackUrl = fmt.Sprintf("http://%s:%d", hostname, port)
|
||||
|
||||
log.Printf("[INFO] NEW WORKER HOSTNAME: %s", appCallbackUrl)
|
||||
return listener
|
||||
}
|
||||
|
||||
log.Printf("[DEBUG] OLD HOSTNAME: %s", appCallbackUrl)
|
||||
|
||||
|
||||
port := listener.Addr().(*net.TCPAddr).Port
|
||||
// Set the port environment variable
|
||||
os.Setenv("WORKER_PORT", fmt.Sprintf("%d", port))
|
||||
|
||||
log.Printf("[DEBUG] Starting webserver (2) on port %d with hostname: %s", port, hostname)
|
||||
appCallbackUrl = fmt.Sprintf("http://%s:%d", hostname, port)
|
||||
|
||||
log.Printf("[INFO] NEW WORKER HOSTNAME: %s", appCallbackUrl)
|
||||
return listener
|
||||
}
|
||||
|
||||
func downloadDockerImageBackend(client *http.Client, imageName string) error {
|
||||
// Check environment SHUFFLE_AUTO_IMAGE_DOWNLOAD
|
||||
if os.Getenv("SHUFFLE_AUTO_IMAGE_DOWNLOAD") == "false" {
|
||||
// Check environment SHUFFLE_AUTO_IMAGE_DOWNLOAD
|
||||
if os.Getenv("SHUFFLE_AUTO_IMAGE_DOWNLOAD") == "false" {
|
||||
//log.Printf("[DEBUG] SHUFFLE_AUTO_IMAGE_DOWNLOAD is false. Not downloading image %s", imageName)
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
if arrayContains(downloadedImages, imageName) {
|
||||
if arrayContains(downloadedImages, imageName) {
|
||||
log.Printf("[DEBUG] Image %s already downloaded", imageName)
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
log.Printf("[DEBUG] Trying to download image %s from backend %s as it doesn't exist. All images: %#v", imageName, baseUrl, downloadedImages)
|
||||
log.Printf("[DEBUG] Trying to download image %s from backend %s as it doesn't exist. All images: %#v", imageName, baseUrl, downloadedImages)
|
||||
|
||||
downloadedImages = append(downloadedImages, imageName)
|
||||
downloadedImages = append(downloadedImages, imageName)
|
||||
|
||||
data := fmt.Sprintf(`{"name": "%s"}`, imageName)
|
||||
dockerImgUrl := fmt.Sprintf("%s/api/v1/get_docker_image", baseUrl)
|
||||
data := fmt.Sprintf(`{"name": "%s"}`, imageName)
|
||||
dockerImgUrl := fmt.Sprintf("%s/api/v1/get_docker_image", baseUrl)
|
||||
|
||||
req, err := http.NewRequest(
|
||||
req, err := http.NewRequest(
|
||||
"POST",
|
||||
dockerImgUrl,
|
||||
bytes.NewBuffer([]byte(data)),
|
||||
)
|
||||
)
|
||||
|
||||
authorization := os.Getenv("AUTHORIZATION")
|
||||
if len(authorization) > 0 {
|
||||
authorization := os.Getenv("AUTHORIZATION")
|
||||
if len(authorization) > 0 {
|
||||
req.Header.Add("Authorization", fmt.Sprintf("Bearer %s", authorization))
|
||||
} else {
|
||||
} else {
|
||||
log.Printf("[WARNING] No auth found - running backend download without it.")
|
||||
//return
|
||||
}
|
||||
}
|
||||
|
||||
newresp, err := topClient.Do(req)
|
||||
if err != nil {
|
||||
newresp, err := topClient.Do(req)
|
||||
if err != nil {
|
||||
log.Printf("[ERROR] Failed download request for %s: %s", imageName, err)
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
defer newresp.Body.Close()
|
||||
if newresp.StatusCode != 200 {
|
||||
defer newresp.Body.Close()
|
||||
if newresp.StatusCode != 200 {
|
||||
log.Printf("[ERROR] Docker download for image %s (backend) StatusCode (1): %d", imageName, newresp.StatusCode)
|
||||
return errors.New(fmt.Sprintf("Failed to get image - status code %d", newresp.StatusCode))
|
||||
}
|
||||
}
|
||||
|
||||
newImageName := strings.Replace(imageName, "/", "_", -1)
|
||||
newFileName := newImageName + ".tar"
|
||||
newImageName := strings.Replace(imageName, "/", "_", -1)
|
||||
newFileName := newImageName + ".tar"
|
||||
|
||||
tar, err := os.Create(newFileName)
|
||||
if err != nil {
|
||||
tar, err := os.Create(newFileName)
|
||||
if err != nil {
|
||||
log.Printf("[WARNING] Failed creating file: %s", err)
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
defer tar.Close()
|
||||
_, err = io.Copy(tar, newresp.Body)
|
||||
if err != nil {
|
||||
defer tar.Close()
|
||||
_, err = io.Copy(tar, newresp.Body)
|
||||
if err != nil {
|
||||
log.Printf("[WARNING] Failed response body copying: %s", err)
|
||||
return err
|
||||
}
|
||||
tar.Seek(0, 0)
|
||||
}
|
||||
tar.Seek(0, 0)
|
||||
|
||||
dockercli, err := dockerclient.NewEnvClient()
|
||||
if err != nil {
|
||||
dockercli, err := dockerclient.NewEnvClient()
|
||||
if err != nil {
|
||||
log.Printf("[ERROR] Unable to create docker client (3): %s", err)
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
defer dockercli.Close()
|
||||
defer dockercli.Close()
|
||||
|
||||
imageLoadResponse, err := dockercli.ImageLoad(context.Background(), tar, true)
|
||||
if err != nil {
|
||||
imageLoadResponse, err := dockercli.ImageLoad(context.Background(), tar, true)
|
||||
if err != nil {
|
||||
log.Printf("[ERROR] Error loading images: %s", err)
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
defer imageLoadResponse.Body.Close()
|
||||
body, err := ioutil.ReadAll(imageLoadResponse.Body)
|
||||
if err != nil {
|
||||
defer imageLoadResponse.Body.Close()
|
||||
body, err := ioutil.ReadAll(imageLoadResponse.Body)
|
||||
if err != nil {
|
||||
log.Printf("[ERROR] Error reading: %s", err)
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
if strings.Contains(string(body), "no such file") {
|
||||
if strings.Contains(string(body), "no such file") {
|
||||
return errors.New(string(body))
|
||||
}
|
||||
}
|
||||
|
||||
baseTag := strings.Split(imageName, ":")
|
||||
if len(baseTag) > 1 {
|
||||
baseTag := strings.Split(imageName, ":")
|
||||
if len(baseTag) > 1 {
|
||||
tag := baseTag[1]
|
||||
log.Printf("[DEBUG] Creating tag copies of downloaded containers from tag %s", tag)
|
||||
|
||||
@@ -2558,33 +2559,33 @@ if len(baseTag) > 1 {
|
||||
downloadedImages = append(downloadedImages, fmt.Sprintf("frikky/shuffle:%s", tag))
|
||||
downloadedImages = append(downloadedImages, fmt.Sprintf("registry.hub.docker.com/frikky/shuffle:%s", tag))
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
os.Remove(newFileName)
|
||||
os.Remove(newFileName)
|
||||
|
||||
log.Printf("[INFO] Successfully loaded image %s: %s", imageName, string(body))
|
||||
return nil
|
||||
}
|
||||
log.Printf("[INFO] Successfully loaded image %s: %s", imageName, string(body))
|
||||
return nil
|
||||
}
|
||||
|
||||
func findActiveSwarmNodes(dockercli *dockerclient.Client) (int64, error) {
|
||||
ctx := context.Background()
|
||||
nodes, err := dockercli.NodeList(ctx, types.NodeListOptions{})
|
||||
if err != nil {
|
||||
func findActiveSwarmNodes(dockercli *dockerclient.Client) (int64, error) {
|
||||
ctx := context.Background()
|
||||
nodes, err := dockercli.NodeList(ctx, types.NodeListOptions{})
|
||||
if err != nil {
|
||||
return 1, err
|
||||
}
|
||||
}
|
||||
|
||||
nodeCount := int64(0)
|
||||
for _, node := range nodes {
|
||||
nodeCount := int64(0)
|
||||
for _, node := range nodes {
|
||||
//log.Printf("ID: %s - %#v", node.ID, node.Status.State)
|
||||
if node.Status.State == "ready" {
|
||||
nodeCount += 1
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Check for SHUFFLE_MAX_NODES
|
||||
maxNodesString := os.Getenv("SHUFFLE_MAX_SWARM_NODES")
|
||||
// Make it into a number and check if it's lower than nodeCount
|
||||
if len(maxNodesString) > 0 {
|
||||
// Check for SHUFFLE_MAX_NODES
|
||||
maxNodesString := os.Getenv("SHUFFLE_MAX_SWARM_NODES")
|
||||
// Make it into a number and check if it's lower than nodeCount
|
||||
if len(maxNodesString) > 0 {
|
||||
maxNodes, err := strconv.ParseInt(maxNodesString, 10, 64)
|
||||
if err != nil {
|
||||
return nodeCount, err
|
||||
@@ -2593,15 +2594,15 @@ if len(maxNodesString) > 0 {
|
||||
if nodeCount > maxNodes {
|
||||
nodeCount = maxNodes
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return nodeCount, nil
|
||||
return nodeCount, nil
|
||||
|
||||
/*
|
||||
/*
|
||||
containers, err := dockercli.ContainerList(ctx, types.ContainerListOptions{
|
||||
All: true,
|
||||
})
|
||||
*/
|
||||
*/
|
||||
}
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user