Frikky
5659 lines
184 KiB
Go
5659 lines
184 KiB
Go
package main
|
|
|
|
import (
|
|
"github.com/shuffle/shuffle-shared"
|
|
singul "github.com/shuffle/singul/pkg"
|
|
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"io/ioutil"
|
|
"log"
|
|
"net"
|
|
"net/http"
|
|
"net/http/pprof"
|
|
"net/url"
|
|
"os"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/docker/docker/api/types"
|
|
"github.com/docker/docker/api/types/container"
|
|
"github.com/docker/docker/api/types/filters"
|
|
dockerimage "github.com/docker/docker/api/types/image"
|
|
"github.com/docker/docker/api/types/mount"
|
|
"github.com/docker/docker/api/types/network"
|
|
dockerclient "github.com/docker/docker/client"
|
|
|
|
// This is for automatic removal of certain code :)
|
|
/*** STARTREMOVE ***/
|
|
"math/rand"
|
|
|
|
"github.com/docker/docker/api/types/swarm"
|
|
uuid "github.com/satori/go.uuid"
|
|
|
|
/*** ENDREMOVE ***/
|
|
|
|
"github.com/gorilla/mux"
|
|
|
|
//k8s deps
|
|
appsv1 "k8s.io/api/apps/v1"
|
|
corev1 "k8s.io/api/core/v1"
|
|
"k8s.io/apimachinery/pkg/api/resource"
|
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
|
"k8s.io/apimachinery/pkg/util/intstr"
|
|
"k8s.io/client-go/kubernetes"
|
|
)
|
|
|
|
// This is getting out of hand :)
|
|
var timezone = os.Getenv("TZ")
|
|
var baseUrl = os.Getenv("BASE_URL")
|
|
var appCallbackUrl = os.Getenv("BASE_URL")
|
|
var isKubernetes = os.Getenv("IS_KUBERNETES")
|
|
var environment = os.Getenv("ENVIRONMENT_NAME")
|
|
var logsDisabled = os.Getenv("SHUFFLE_LOGS_DISABLED")
|
|
var cleanupEnv = strings.ToLower(os.Getenv("CLEANUP"))
|
|
var swarmNetworkName = os.Getenv("SHUFFLE_SWARM_NETWORK_NAME")
|
|
var dockerApiVersion = strings.ToLower(os.Getenv("DOCKER_API_VERSION"))
|
|
var shutdownDisabled = strings.ToLower(os.Getenv("SHUFFLE_WORKER_SHUTDOWN_DISABLED"))
|
|
|
|
// Kubernetes settings
|
|
var appServiceAccountName = os.Getenv("SHUFFLE_APP_SERVICE_ACCOUNT_NAME")
|
|
var appPodSecurityContext = os.Getenv("SHUFFLE_APP_POD_SECURITY_CONTEXT")
|
|
var appContainerSecurityContext = os.Getenv("SHUFFLE_APP_CONTAINER_SECURITY_CONTEXT")
|
|
var kubernetesNamespace = os.Getenv("KUBERNETES_NAMESPACE")
|
|
|
|
var executionCount int64
|
|
|
|
var baseimagename = os.Getenv("SHUFFLE_BASE_IMAGE_NAME")
|
|
|
|
// var baseimagename = "registry.hub.docker.com/frikky/shuffle"
|
|
var registryName = "registry.hub.docker.com"
|
|
var sleepTime = 2
|
|
var topClient *http.Client
|
|
var data string
|
|
var requestsSent = 0
|
|
var appsInitialized = false
|
|
|
|
var hostname string
|
|
var maxReplicas = uint64(12)
|
|
var debug bool
|
|
|
|
/*
|
|
var environments []string
|
|
var parents map[string][]string
|
|
var children map[string][]string
|
|
var visited []string
|
|
var executed []string
|
|
var nextActions []string
|
|
var extra int
|
|
var startAction string
|
|
*/
|
|
//var results []shuffle.ActionResult
|
|
//var allLogs map[string]string
|
|
//var containerIds []string
|
|
var downloadedImages []string
|
|
|
|
type ImageDownloadBody struct {
|
|
Image string `json:"image"`
|
|
}
|
|
|
|
type ImageRequest struct {
|
|
Image string `json:"image"`
|
|
}
|
|
|
|
var finishedExecutions []string
|
|
var imagesDistributed []string
|
|
var imagedownloadTimeout = time.Second * 300
|
|
|
|
var subflowPollBackoff sync.Map
|
|
|
|
var window = shuffle.NewTimeWindow(10 * time.Second)
|
|
|
|
func restoreActionConfig(ctx context.Context, executionID string, action *shuffle.Action, workflowExecution *shuffle.WorkflowExecution) {
|
|
if len(executionID) == 0 || action == nil {
|
|
return
|
|
}
|
|
|
|
cacheKey := fmt.Sprintf("workflowexecution_%s_action_snapshot", executionID)
|
|
cacheDataRaw, err := shuffle.GetCache(ctx, cacheKey)
|
|
if err != nil {
|
|
return
|
|
}
|
|
|
|
originalActions := []shuffle.Action{}
|
|
cacheData := []byte(cacheDataRaw.([]uint8))
|
|
if err := json.Unmarshal(cacheData, &originalActions); err != nil {
|
|
return
|
|
}
|
|
|
|
original := shuffle.Action{}
|
|
for _, candidate := range originalActions {
|
|
if candidate.ID == action.ID {
|
|
original = candidate
|
|
break
|
|
}
|
|
}
|
|
|
|
if len(original.ID) == 0 {
|
|
return
|
|
}
|
|
|
|
for index, param := range action.Parameters {
|
|
if !param.Configuration || len(param.Value) > 0 {
|
|
continue
|
|
}
|
|
|
|
for _, originalParam := range original.Parameters {
|
|
if !strings.EqualFold(strings.TrimSpace(param.Name), strings.TrimSpace(originalParam.Name)) {
|
|
continue
|
|
}
|
|
|
|
if len(originalParam.Value) > 0 {
|
|
action.Parameters[index].Value = originalParam.Value
|
|
}
|
|
|
|
break
|
|
}
|
|
}
|
|
|
|
if workflowExecution == nil {
|
|
return
|
|
}
|
|
|
|
for actionIndex, workflowAction := range workflowExecution.Workflow.Actions {
|
|
if workflowAction.ID != action.ID {
|
|
continue
|
|
}
|
|
|
|
for parameterIndex, workflowParam := range workflowAction.Parameters {
|
|
if !workflowParam.Configuration || len(workflowParam.Value) > 0 {
|
|
continue
|
|
}
|
|
|
|
for _, originalParam := range original.Parameters {
|
|
if !strings.EqualFold(strings.TrimSpace(workflowParam.Name), strings.TrimSpace(originalParam.Name)) {
|
|
continue
|
|
}
|
|
|
|
if len(originalParam.Value) > 0 {
|
|
workflowExecution.Workflow.Actions[actionIndex].Parameters[parameterIndex].Value = originalParam.Value
|
|
}
|
|
|
|
break
|
|
}
|
|
}
|
|
|
|
break
|
|
}
|
|
}
|
|
|
|
func getAppProxyValue(primaryKey, fallbackKey string) string {
|
|
value := strings.TrimSpace(os.Getenv(primaryKey))
|
|
if len(value) > 0 {
|
|
return value
|
|
}
|
|
|
|
return strings.TrimSpace(os.Getenv(fallbackKey))
|
|
}
|
|
|
|
func appendAppProxyEnv(env []string) []string {
|
|
httpProxy := getAppProxyValue("HTTP_PROXY", "SHUFFLE_APP_HTTP_PROXY")
|
|
httpsProxy := getAppProxyValue("HTTPS_PROXY", "SHUFFLE_APP_HTTPS_PROXY")
|
|
noProxy := getAppProxyValue("NO_PROXY", "SHUFFLE_APP_NO_PROXY")
|
|
noProxyLower := getAppProxyValue("no_proxy", "SHUFFLE_APP_no_proxy")
|
|
|
|
if len(httpProxy) > 0 {
|
|
env = append(env, fmt.Sprintf("HTTP_PROXY=%s", httpProxy))
|
|
}
|
|
|
|
if len(httpsProxy) > 0 {
|
|
env = append(env, fmt.Sprintf("HTTPS_PROXY=%s", httpsProxy))
|
|
}
|
|
|
|
if len(noProxy) > 0 {
|
|
env = append(env, fmt.Sprintf("NO_PROXY=%s", noProxy))
|
|
}
|
|
|
|
if len(noProxyLower) > 0 {
|
|
env = append(env, fmt.Sprintf("no_proxy=%s", noProxyLower))
|
|
}
|
|
|
|
return env
|
|
}
|
|
|
|
// Images to be autodeployed in the latest version of Shuffle.
|
|
var autoDeploy = map[string]string{
|
|
"http:1.4.0": "frikky/shuffle:http_1.4.0",
|
|
"shuffle-tools:1.2.0": "frikky/shuffle:shuffle-tools_1.2.0",
|
|
"shuffle-subflow:1.1.0": "frikky/shuffle:shuffle-subflow_1.1.0",
|
|
"shuffle-ai:1.1.0": "frikky/shuffle:shuffle-ai_1.1.0",
|
|
// "shuffle-tools-fork:1.0.0": "frikky/shuffle:shuffle-tools-fork_1.0.0",
|
|
}
|
|
|
|
// New Worker mappings
|
|
// visited, appendActions, nextActions, notFound, queueNodes, toRemove, executed, env
|
|
var portMappings map[string]int
|
|
var baseport = 33333
|
|
|
|
type UserInputSubflow struct {
|
|
Argument string `json:"execution_argument"`
|
|
ContinueUrl string `json:"continue_url"`
|
|
CancelUrl string `json:"cancel_url"`
|
|
}
|
|
|
|
// Not using shuffle.SetWorkflowExecution as we only want to use cache in reality
|
|
func setWorkflowExecution(ctx context.Context, workflowExecution shuffle.WorkflowExecution, dbSave bool) error {
|
|
if len(workflowExecution.ExecutionId) == 0 {
|
|
log.Printf("[DEBUG] Workflowexecution executionId can't be empty.")
|
|
return errors.New("ExecutionId can't be empty.")
|
|
}
|
|
|
|
if len(workflowExecution.Workflow.Actions) > 0 {
|
|
snapshotKey := fmt.Sprintf("workflowexecution_%s_action_snapshot", workflowExecution.ExecutionId)
|
|
if _, err := shuffle.GetCache(ctx, snapshotKey); err != nil {
|
|
if payload, marshalErr := json.Marshal(workflowExecution.Workflow.Actions); marshalErr == nil {
|
|
_ = shuffle.SetCache(ctx, snapshotKey, payload, 3600)
|
|
}
|
|
}
|
|
}
|
|
|
|
//log.Printf("[DEBUG][%s] Setting with %d results (pre)", workflowExecution.ExecutionId, len(workflowExecution.Results))
|
|
workflowExecution, _ = shuffle.Fixexecution(ctx, workflowExecution)
|
|
cacheKey := fmt.Sprintf("workflowexecution_%s", workflowExecution.ExecutionId)
|
|
|
|
execData, err := json.Marshal(workflowExecution)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed marshalling execution during set: %s", err)
|
|
return err
|
|
}
|
|
|
|
err = shuffle.SetCache(ctx, cacheKey, execData, 30)
|
|
if err != nil {
|
|
log.Printf("[ERROR][%s] Failed adding to cache during setexecution", workflowExecution.ExecutionId)
|
|
return err
|
|
}
|
|
|
|
handleExecutionResult(workflowExecution)
|
|
validated := shuffle.ValidateFinished(ctx, -1, workflowExecution)
|
|
if validated {
|
|
shutdownData, err := json.Marshal(workflowExecution)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed marshalling shutdowndata during set: %s", err)
|
|
}
|
|
|
|
log.Printf("[DEBUG][%s] Sending result (set). Status: %s, Actions: %d, Results: %d", workflowExecution.ExecutionId, workflowExecution.Status, len(workflowExecution.Workflow.Actions), len(workflowExecution.Results))
|
|
sendResult(workflowExecution, shutdownData)
|
|
return nil
|
|
}
|
|
|
|
// FIXME: Should this shutdown OR send the result?
|
|
// The worker may not be running the backend hmm
|
|
if dbSave {
|
|
if workflowExecution.ExecutionSource == "default" {
|
|
log.Printf("[DEBUG][%s] Shutting down (25)", workflowExecution.ExecutionId)
|
|
shutdown(workflowExecution, "", "", true)
|
|
//return
|
|
} else {
|
|
log.Printf("[DEBUG][%s] NOT shutting down with dbSave (%s). Instead sending result to backend and start polling until subflow is updated", workflowExecution.ExecutionId, workflowExecution.ExecutionSource)
|
|
|
|
shutdownData, err := json.Marshal(workflowExecution)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed marshalling shutdowndata during dbSave handler: %s", err)
|
|
}
|
|
|
|
sendResult(workflowExecution, shutdownData)
|
|
|
|
// Poll for 1 minute max if there is a "wait for results" subflow
|
|
subflowId := ""
|
|
for _, result := range workflowExecution.Results {
|
|
if result.Status == "WAITING" {
|
|
//log.Printf("[DEBUG][%s] Found waiting result", workflowExecution.ExecutionId)
|
|
subflowId = result.Action.ID
|
|
}
|
|
}
|
|
|
|
if len(subflowId) == 0 {
|
|
log.Printf("[DEBUG][%s] No waiting result found. Not polling", workflowExecution.ExecutionId)
|
|
|
|
for _, action := range workflowExecution.Workflow.Actions {
|
|
if action.AppName == "User Input" || action.AppName == "Shuffle Workflow" || action.AppName == "shuffle-subflow" {
|
|
workflowExecution.Workflow.Triggers = append(workflowExecution.Workflow.Triggers, shuffle.Trigger{
|
|
AppName: action.AppName,
|
|
Parameters: action.Parameters,
|
|
ID: action.ID,
|
|
})
|
|
}
|
|
}
|
|
|
|
for _, trigger := range workflowExecution.Workflow.Triggers {
|
|
//log.Printf("[DEBUG] Found trigger %s", trigger.AppName)
|
|
if trigger.AppName != "User Input" && trigger.AppName != "Shuffle Workflow" && trigger.AppName != "shuffle-subflow" {
|
|
continue
|
|
}
|
|
|
|
// check if it has wait for results in params
|
|
wait := false
|
|
for _, param := range trigger.Parameters {
|
|
//log.Printf("[DEBUG] Found param %s with value %s", param.Name, param.Value)
|
|
if param.Name == "check_result" && strings.ToLower(param.Value) == "true" {
|
|
//log.Printf("[DEBUG][%s] Found check result param!", workflowExecution.ExecutionId)
|
|
wait = true
|
|
break
|
|
}
|
|
}
|
|
|
|
if wait {
|
|
// Check if it has a result or not
|
|
found := false
|
|
for _, result := range workflowExecution.Results {
|
|
//log.Printf("[DEBUG][%s] Found result %s", workflowExecution.ExecutionId, result.Action.ID)
|
|
if result.Action.ID == trigger.ID && result.Status != "SUCCESS" && result.Status != "FAILURE" {
|
|
//log.Printf("[DEBUG][%s] Found subflow result that is not handled. Waiting for results", workflowExecution.ExecutionId)
|
|
|
|
subflowId = result.Action.ID
|
|
found = true
|
|
break
|
|
}
|
|
}
|
|
|
|
if !found {
|
|
log.Printf("[DEBUG][%s] No result found for subflow. Setting subflowId to %s", workflowExecution.ExecutionId, trigger.ID)
|
|
subflowId = trigger.ID
|
|
}
|
|
}
|
|
|
|
if len(subflowId) > 0 {
|
|
break
|
|
}
|
|
}
|
|
}
|
|
|
|
if len(subflowId) > 0 {
|
|
// Under rerun period timeout
|
|
timeComparison := 120
|
|
log.Printf("[DEBUG][%s] Starting polling for %d seconds to see if new subflow updates are found on the backend that are not handled. Subflow ID: %s", workflowExecution.ExecutionId, timeComparison, subflowId)
|
|
timestart := time.Now()
|
|
streamResultUrl := fmt.Sprintf("%s/api/v1/streams/results", baseUrl)
|
|
for {
|
|
err = handleSubflowPoller(ctx, workflowExecution, streamResultUrl, subflowId)
|
|
if err == nil {
|
|
log.Printf("[DEBUG] Subflow is finished and we are breaking the thingy")
|
|
|
|
if os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "swarm" && workflowExecution.ExecutionSource != "default" {
|
|
log.Printf("[DEBUG] Force shutdown of worker due to optimized run with webserver. Expecting reruns to take care of this")
|
|
os.Exit(0)
|
|
}
|
|
|
|
break
|
|
}
|
|
|
|
timepassed := time.Since(timestart)
|
|
if timepassed.Seconds() > float64(timeComparison) {
|
|
log.Printf("[DEBUG][%s] Max poll time reached to look for updates. Stopping poll. This poll is here to send personal results back to itself to be handled, then to stop this thread.", workflowExecution.ExecutionId)
|
|
break
|
|
}
|
|
|
|
}
|
|
} else {
|
|
log.Printf("[DEBUG][%s] No need to poll for results. Not polling", workflowExecution.ExecutionId)
|
|
}
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// removes every container except itself (worker)
|
|
func shutdown(workflowExecution shuffle.WorkflowExecution, nodeId string, reason string, handleResultSend bool) {
|
|
log.Printf("[DEBUG][%s] Shutdown (%s) started with reason %#v. Result amount: %d. ResultsSent: %d, Send result: %#v, Parent: %#v", workflowExecution.ExecutionId, workflowExecution.Status, reason, len(workflowExecution.Results), requestsSent, handleResultSend, workflowExecution.ExecutionParent)
|
|
|
|
// This is an escape hatch for development only
|
|
// Typically meant to be used when you aren't sure how to make a workflow run in bad scenarios, and want to rapidly debug it.
|
|
if shutdownDisabled == "true" {
|
|
log.Printf("[ERROR] Shutdown disabled: NOT shutting down. This should ONLY be used for development & debugging.")
|
|
os.Exit(3)
|
|
}
|
|
|
|
sleepDuration := 1
|
|
if handleResultSend && requestsSent < 2 {
|
|
shutdownData, err := json.Marshal(workflowExecution)
|
|
if err == nil {
|
|
sendResult(workflowExecution, shutdownData)
|
|
//log.Printf("[WARNING][%s] Sent shutdown update with %d results and result value %s", workflowExecution.ExecutionId, len(workflowExecution.Results), reason)
|
|
} else {
|
|
log.Printf("[WARNING][%s] Failed to send update: %s", workflowExecution.ExecutionId, err)
|
|
}
|
|
|
|
time.Sleep(time.Duration(sleepDuration) * time.Second)
|
|
}
|
|
|
|
// Might not be necessary because of cleanupEnv hostconfig autoremoval
|
|
if strings.ToLower(cleanupEnv) == "true" && (os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "swarm") {
|
|
/*
|
|
ctx := context.Background()
|
|
dockercli, err := dockerclient.NewEnvClient()
|
|
if err == nil {
|
|
log.Printf("[INFO] Cleaning up %d containers", len(containerIds))
|
|
removeOptions := types.ContainerRemoveOptions{
|
|
RemoveVolumes: true,
|
|
Force: true,
|
|
}
|
|
|
|
for _, containername := range containerIds {
|
|
log.Printf("[INFO] Should stop and and remove container %s (deprecated)", containername)
|
|
//dockercli.ContainerStop(ctx, containername, nil)
|
|
//dockercli.ContainerRemove(ctx, containername, removeOptions)
|
|
//removeContainers = append(removeContainers, containername)
|
|
}
|
|
}
|
|
*/
|
|
} else {
|
|
|
|
/*** STARTREMOVE ***/
|
|
if os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "swarm" {
|
|
log.Printf("[DEBUG][%s] NOT cleaning up containers. IDS: %d, CLEANUP env: %s", workflowExecution.ExecutionId, 0, cleanupEnv)
|
|
}
|
|
/*** ENDREMOVE ***/
|
|
}
|
|
|
|
if len(reason) > 0 && len(nodeId) > 0 {
|
|
//log.Printf("[INFO] Running abort of workflow because it should be finished")
|
|
|
|
abortUrl := fmt.Sprintf("%s/api/v1/workflows/%s/executions/%s/abort", baseUrl, workflowExecution.Workflow.ID, workflowExecution.ExecutionId)
|
|
path := fmt.Sprintf("?reason=%s", url.QueryEscape(reason))
|
|
if len(nodeId) > 0 {
|
|
path += fmt.Sprintf("&node=%s", url.QueryEscape(nodeId))
|
|
}
|
|
if len(environment) > 0 {
|
|
path += fmt.Sprintf("&env=%s", url.QueryEscape(environment))
|
|
}
|
|
|
|
abortUrl += path
|
|
log.Printf("[DEBUG][%s] Abort URL: %s", workflowExecution.ExecutionId, abortUrl)
|
|
|
|
req, err := http.NewRequest(
|
|
"GET",
|
|
abortUrl,
|
|
nil,
|
|
)
|
|
|
|
if err != nil {
|
|
log.Printf("[WARNING][%s] Failed building request: %s", workflowExecution.ExecutionId, err)
|
|
}
|
|
|
|
// FIXME: Add an API call to the backend
|
|
if os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "swarm" {
|
|
authorization := os.Getenv("AUTHORIZATION")
|
|
if len(authorization) > 0 {
|
|
req.Header.Add("Authorization", fmt.Sprintf("Bearer %s", authorization))
|
|
} else {
|
|
log.Printf("[ERROR][%s] No authorization specified for abort", workflowExecution.ExecutionId)
|
|
}
|
|
} else {
|
|
req.Header.Add("Authorization", fmt.Sprintf("Bearer %s", workflowExecution.Authorization))
|
|
}
|
|
|
|
req.Header.Add("Content-Type", "application/json")
|
|
|
|
//log.Printf("[DEBUG][%s] All App Logs: %#v", workflowExecution.ExecutionId, allLogs)
|
|
client := shuffle.GetExternalClient(abortUrl)
|
|
newresp, err := client.Do(req)
|
|
if err != nil {
|
|
log.Printf("[WARNING][%s] Failed abort request: %s", workflowExecution.ExecutionId, err)
|
|
} else {
|
|
defer newresp.Body.Close()
|
|
}
|
|
} else {
|
|
//log.Printf("[INFO][%s] NOT running abort during shutdown.", workflowExecution.ExecutionId)
|
|
}
|
|
|
|
log.Printf("[DEBUG][%s] Finished shutdown (after %d seconds). ", workflowExecution.ExecutionId, sleepDuration)
|
|
//Finished shutdown (after %d seconds). ", sleepDuration)
|
|
|
|
// Allows everything to finish in subprocesses (apps)
|
|
if os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "swarm" && isKubernetes != "true" {
|
|
time.Sleep(time.Duration(sleepDuration) * time.Second)
|
|
os.Exit(3)
|
|
} else {
|
|
log.Printf("[DEBUG][%s] Sending result and resetting values (K8s & Swarm).", workflowExecution.ExecutionId)
|
|
}
|
|
}
|
|
|
|
// ** STARTREMOVE ***/
|
|
func deployk8sApp(image string, identifier string, env []string) error {
|
|
if len(os.Getenv("KUBERNETES_NAMESPACE")) > 0 {
|
|
kubernetesNamespace = os.Getenv("KUBERNETES_NAMESPACE")
|
|
} else {
|
|
kubernetesNamespace = "default"
|
|
}
|
|
|
|
ctx := context.Background()
|
|
|
|
log.Printf("[DEBUG] Deploying k8s app with identifier %s to namespace %s", identifier, kubernetesNamespace)
|
|
deployport, err := strconv.Atoi(os.Getenv("SHUFFLE_APP_EXPOSED_PORT"))
|
|
if err != nil {
|
|
deployport = 80
|
|
}
|
|
|
|
envMap := make(map[string]string)
|
|
for _, envStr := range env {
|
|
parts := strings.SplitN(envStr, "=", 2)
|
|
if len(parts) == 2 {
|
|
envMap[parts[0]] = parts[1]
|
|
}
|
|
}
|
|
|
|
// add to env
|
|
envMap["SHUFFLE_APP_EXPOSED_PORT"] = strconv.Itoa(deployport)
|
|
envMap["SHUFFLE_SWARM_CONFIG"] = os.Getenv("SHUFFLE_SWARM_CONFIG")
|
|
envMap["BASE_URL"] = "http://shuffle-workers:33333"
|
|
|
|
if len(os.Getenv("SHUFFLE_LOGS_DISABLED")) > 0 {
|
|
envMap["SHUFFLE_LOGS_DISABLED"] = os.Getenv("SHUFFLE_LOGS_DISABLED")
|
|
}
|
|
|
|
clientset, _, err := shuffle.GetKubernetesClient()
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed getting kubernetes: %s", err)
|
|
return err
|
|
}
|
|
|
|
// str := strings.ToLower(identifier)
|
|
// strSplit := strings.Split(str, "_")
|
|
// value := strSplit[0]
|
|
// value = strings.ReplaceAll(value, "_", "-")
|
|
value := identifier
|
|
|
|
baseDeployMode := false
|
|
|
|
// check if autoDeploy contains a value
|
|
// that is equal to the image being deployed.
|
|
for _, value := range autoDeploy {
|
|
if value == image {
|
|
baseDeployMode = true
|
|
}
|
|
}
|
|
|
|
autoDeployOverride := os.Getenv("SHUFFLE_USE_GHCR_OVERRIDE_FOR_AUTODEPLOY") == "true"
|
|
|
|
localRegistry := ""
|
|
|
|
// Checking if app is generated or not
|
|
if !(baseDeployMode && autoDeployOverride) {
|
|
localRegistry = os.Getenv("REGISTRY_URL")
|
|
} else {
|
|
log.Printf("[DEBUG] Detected baseDeploy image (%s) and ghcr override. Resorting to using ghcr instead of registry", image)
|
|
}
|
|
|
|
/*
|
|
appDetails := strings.Split(image, ":")[1]
|
|
appDetailsSplit := strings.Split(appDetails, "_")
|
|
appName := strings.Join(appDetailsSplit[:len(appDetailsSplit)-1], "_")
|
|
appVersion := appDetailsSplit[len(appDetailsSplit)-1]
|
|
for _, app := range workflowExecution.Workflow.Actions {
|
|
// log.Printf("[DEBUG] App: %s, Version: %s", appName, appVersion)
|
|
// log.Printf("[DEBUG] Checking app %s with version %s", app.AppName, app.AppVersion)
|
|
if app.AppName == appName && app.AppVersion == appVersion {
|
|
if app.Generated == true {
|
|
log.Printf("[DEBUG] Generated app, setting local registry")
|
|
image = fmt.Sprintf("%s/%s", localRegistry, image)
|
|
break
|
|
} else {
|
|
log.Printf("[DEBUG] Not generated app, setting shuffle registry")
|
|
}
|
|
}
|
|
}
|
|
*/
|
|
|
|
if (len(localRegistry) == 0 && len(os.Getenv("SHUFFLE_BASE_IMAGE_REGISTRY")) > 0) && !(baseDeployMode && autoDeployOverride) {
|
|
localRegistry = os.Getenv("SHUFFLE_BASE_IMAGE_REGISTRY")
|
|
}
|
|
|
|
if (len(localRegistry) > 0 && strings.Count(image, "/") <= 2) && !(baseDeployMode && autoDeployOverride) {
|
|
log.Printf("[DEBUG] Using REGISTRY_URL %s", localRegistry)
|
|
image = fmt.Sprintf("%s/%s", localRegistry, image)
|
|
} else {
|
|
if strings.Count(image, "/") <= 2 && !strings.HasPrefix(image, "frikky/shuffle:") {
|
|
image = fmt.Sprintf("frikky/shuffle:%s", image)
|
|
}
|
|
}
|
|
|
|
log.Printf("[DEBUG] Got kubernetes with namespace %#v to run image '%s'", kubernetesNamespace, image)
|
|
|
|
//fix naming convention
|
|
// podUuid := uuid.NewV4().String()
|
|
// name := fmt.Sprintf("%s-%s", value, podUuid)
|
|
// replace identifier "_" with "-"
|
|
name := strings.ReplaceAll(identifier, "_", "-")
|
|
|
|
labels := map[string]string{
|
|
// Well-known Kubernetes labels
|
|
"app.kubernetes.io/name": "shuffle-app",
|
|
"app.kubernetes.io/instance": name,
|
|
"app.kubernetes.io/part-of": "shuffle",
|
|
"app.kubernetes.io/managed-by": "shuffle-worker",
|
|
// Keep legacy labels for backward compatibility
|
|
"app": name,
|
|
// TODO: Add Shuffle specific labels
|
|
// "app.shuffler.io/name": "APP_NAME",
|
|
// "app.shuffler.io/version": "APP_VERSION",
|
|
}
|
|
|
|
matchLabels := map[string]string{
|
|
"app.kubernetes.io/name": "shuffle-app",
|
|
"app.kubernetes.io/instance": name,
|
|
}
|
|
|
|
// Parse security contexts from env
|
|
var podSecurityContext *corev1.PodSecurityContext
|
|
var containerSecurityContext *corev1.SecurityContext
|
|
|
|
if len(appPodSecurityContext) > 0 {
|
|
podSecurityContext = &corev1.PodSecurityContext{}
|
|
err = json.Unmarshal([]byte(appPodSecurityContext), podSecurityContext)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed to unmarshal app pod security context: %v", err)
|
|
return fmt.Errorf("failed to unmarshal app pod security context: %v", err)
|
|
}
|
|
}
|
|
|
|
if len(appContainerSecurityContext) > 0 {
|
|
containerSecurityContext = &corev1.SecurityContext{}
|
|
err = json.Unmarshal([]byte(appContainerSecurityContext), containerSecurityContext)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed to unmarshal app container security context: %v", err)
|
|
return fmt.Errorf("failed to unmarshal app container security context: %v", err)
|
|
}
|
|
}
|
|
|
|
// pod := &corev1.Pod{
|
|
// ObjectMeta: metav1.ObjectMeta{
|
|
// Name: podName,
|
|
// Labels: map[string]string{
|
|
// "app": podName,
|
|
// // "executionId": workflowExecution.ExecutionId,
|
|
// },
|
|
// },
|
|
// Spec: corev1.PodSpec{
|
|
// RestartPolicy: "Never", // As a crash is not useful in this context
|
|
// // DNSPolicy: "Default",
|
|
// DNSPolicy: corev1.DNSClusterFirst,
|
|
// // NodeName: "worker1"
|
|
// Containers: []corev1.Container{
|
|
// {
|
|
// Name: value,
|
|
// Image: image,
|
|
// Env: buildEnvVars(envMap),
|
|
|
|
// // Pull if not available
|
|
// ImagePullPolicy: corev1.PullIfNotPresent,
|
|
// },
|
|
// },
|
|
// },
|
|
// }
|
|
|
|
// createdPod, err := clientset.CoreV1().Pods(kubernetesNamespace).Create(context.Background(), pod, metav1.CreateOptions{})
|
|
// if err != nil {
|
|
// log.Printf("[ERROR] Failed creating pod: %v", err)
|
|
// // os.Exit(1)
|
|
// } else {
|
|
// log.Printf("[DEBUG] Created pod %#v in namespace %#v", createdPod.Name, kubernetesNamespace)
|
|
// }
|
|
|
|
// service := &corev1.Service{
|
|
// ObjectMeta: metav1.ObjectMeta{
|
|
// Name: identifier,
|
|
// },
|
|
// Spec: corev1.ServiceSpec{
|
|
// Selector: map[string]string{
|
|
// "app": podName,
|
|
// },
|
|
// Ports: []corev1.ServicePort{
|
|
// {
|
|
// Protocol: "TCP",
|
|
// Port: 80,
|
|
// TargetPort: intstr.FromInt(80),
|
|
// },
|
|
// },
|
|
// Type: corev1.ServiceTypeNodePort,
|
|
// },
|
|
// }
|
|
|
|
// _, err = clientset.CoreV1().Services(kubernetesNamespace).Create(context.TODO(), service, metav1.CreateOptions{})
|
|
// if err != nil {
|
|
// log.Printf("[ERROR] Failed creating service: %v", err)
|
|
// return err
|
|
// }
|
|
|
|
// use deployment instead of pod
|
|
// then expose a service similarly.
|
|
// number of replicas can be set to os.Getenv("SHUFFLE_APP_REPLICAS")
|
|
replicaNumber := 1
|
|
replicaNumberStr := os.Getenv("SHUFFLE_APP_REPLICAS")
|
|
if len(replicaNumberStr) > 0 {
|
|
tmpInt, err := strconv.Atoi(replicaNumberStr)
|
|
if err != nil {
|
|
log.Printf("[ERROR] %s is not a valid number for replication", replicaNumberStr)
|
|
} else {
|
|
replicaNumber = tmpInt
|
|
|
|
}
|
|
}
|
|
|
|
existing, err := clientset.AppsV1().Deployments(kubernetesNamespace).List(
|
|
ctx,
|
|
metav1.ListOptions{
|
|
LabelSelector: fmt.Sprintf("app: %s", name),
|
|
},
|
|
)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed listing existing deployments: %v", err)
|
|
}
|
|
|
|
if len(existing.Items) > 0 {
|
|
log.Printf("[INFO] Found existing deployments, skipping creation")
|
|
return nil
|
|
}
|
|
|
|
replicaNumberInt32 := int32(replicaNumber)
|
|
// apps do not need access the k8s api.
|
|
automountServiceAccountToken := false
|
|
|
|
deployment := &appsv1.Deployment{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: name,
|
|
Labels: labels,
|
|
},
|
|
Spec: appsv1.DeploymentSpec{
|
|
Replicas: &replicaNumberInt32,
|
|
Selector: &metav1.LabelSelector{
|
|
MatchLabels: matchLabels,
|
|
},
|
|
Template: corev1.PodTemplateSpec{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Labels: labels,
|
|
},
|
|
Spec: corev1.PodSpec{
|
|
Containers: []corev1.Container{
|
|
{
|
|
Name: value,
|
|
Image: image,
|
|
Env: buildEnvVars(envMap),
|
|
Ports: []corev1.ContainerPort{
|
|
{
|
|
Protocol: "TCP",
|
|
ContainerPort: int32(deployport),
|
|
},
|
|
},
|
|
SecurityContext: containerSecurityContext,
|
|
Resources: buildResourcesFromEnv(),
|
|
},
|
|
},
|
|
DNSPolicy: corev1.DNSClusterFirst,
|
|
ServiceAccountName: appServiceAccountName,
|
|
AutomountServiceAccountToken: &automountServiceAccountToken,
|
|
SecurityContext: podSecurityContext,
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
if os.Getenv("SHUFFLE_APP_MOUNT_TMP_VOLUME") == "true" {
|
|
deployment.Spec.Template.Spec.Volumes = append(
|
|
deployment.Spec.Template.Spec.Volumes,
|
|
corev1.Volume{
|
|
Name: "tmp",
|
|
VolumeSource: corev1.VolumeSource{
|
|
EmptyDir: &corev1.EmptyDirVolumeSource{},
|
|
},
|
|
},
|
|
)
|
|
|
|
deployment.Spec.Template.Spec.Containers[0].VolumeMounts = append(
|
|
deployment.Spec.Template.Spec.Containers[0].VolumeMounts,
|
|
corev1.VolumeMount{
|
|
Name: "tmp",
|
|
ReadOnly: false,
|
|
MountPath: "/tmp",
|
|
},
|
|
)
|
|
}
|
|
|
|
if len(os.Getenv("REGISTRY_URL")) > 0 && len(os.Getenv("SHUFFLE_BASE_IMAGE_NAME")) > 0 {
|
|
log.Printf("[INFO] Setting image pull policy to Always as private registry is used.")
|
|
//containerAttachment.ImagePullPolicy = corev1.PullAlways
|
|
deployment.Spec.Template.Spec.Containers[0].ImagePullPolicy = corev1.PullAlways
|
|
} else {
|
|
deployment.Spec.Template.Spec.Containers[0].ImagePullPolicy = corev1.PullIfNotPresent
|
|
}
|
|
|
|
_, err = clientset.AppsV1().Deployments(kubernetesNamespace).Create(context.Background(), deployment, metav1.CreateOptions{})
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed creating deployment: %v", err)
|
|
return err
|
|
}
|
|
|
|
svcAppProtocol := "http"
|
|
service := &corev1.Service{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: name,
|
|
Labels: labels,
|
|
},
|
|
Spec: corev1.ServiceSpec{
|
|
Selector: matchLabels,
|
|
Ports: []corev1.ServicePort{
|
|
{
|
|
Protocol: "TCP",
|
|
AppProtocol: &svcAppProtocol,
|
|
Port: 80,
|
|
TargetPort: intstr.FromInt(deployport),
|
|
},
|
|
},
|
|
Type: corev1.ServiceTypeClusterIP,
|
|
},
|
|
}
|
|
|
|
_, err = clientset.CoreV1().Services(kubernetesNamespace).Create(context.TODO(), service, metav1.CreateOptions{})
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed creating service: %v", err)
|
|
return err
|
|
}
|
|
|
|
// Giving the service time to start before we contineu anything
|
|
log.Printf("[DEBUG] Waiting 20 seconds before moving on to let app '%s' start properly. Service: %s (k8s)", name, image)
|
|
time.Sleep(20 * time.Second)
|
|
|
|
return nil
|
|
}
|
|
|
|
//** ENDREMOVE ***/
|
|
|
|
// Deploys the internal worker whenever something happens
|
|
func deployApp(cli *dockerclient.Client, image string, identifier string, env []string, workflowExecution shuffle.WorkflowExecution, action shuffle.Action) error {
|
|
// if isKubernetes == "true" {
|
|
// if len(os.Getenv("KUBERNETES_NAMESPACE")) > 0 {
|
|
// kubernetesNamespace = os.Getenv("KUBERNETES_NAMESPACE")
|
|
// } else {
|
|
// kubernetesNamespace = "default"
|
|
// }
|
|
|
|
// envMap := make(map[string]string)
|
|
// for _, envStr := range env {
|
|
// parts := strings.SplitN(envStr, "=", 2)
|
|
// if len(parts) == 2 {
|
|
// envMap[parts[0]] = parts[1]
|
|
// }
|
|
// }
|
|
|
|
// clientset, _, err := shuffle.GetKubernetesClient()
|
|
// if err != nil {
|
|
// log.Printf("[ERROR] Failed getting kubernetes: %s", err)
|
|
// return err
|
|
// }
|
|
|
|
// str := strings.ToLower(identifier)
|
|
// strSplit := strings.Split(str, "_")
|
|
// value := strSplit[0]
|
|
// value = strings.ReplaceAll(value, "_", "-")
|
|
|
|
// // Checking if app is generated or not
|
|
// localRegistry := os.Getenv("REGISTRY_URL")
|
|
// /*
|
|
// appDetails := strings.Split(image, ":")[1]
|
|
// appDetailsSplit := strings.Split(appDetails, "_")
|
|
// appName := strings.Join(appDetailsSplit[:len(appDetailsSplit)-1], "_")
|
|
// appVersion := appDetailsSplit[len(appDetailsSplit)-1]
|
|
// for _, app := range workflowExecution.Workflow.Actions {
|
|
// // log.Printf("[DEBUG] App: %s, Version: %s", appName, appVersion)
|
|
// // log.Printf("[DEBUG] Checking app %s with version %s", app.AppName, app.AppVersion)
|
|
// if app.AppName == appName && app.AppVersion == appVersion {
|
|
// if app.Generated == true {
|
|
// log.Printf("[DEBUG] Generated app, setting local registry")
|
|
// image = fmt.Sprintf("%s/%s", localRegistry, image)
|
|
// break
|
|
// } else {
|
|
// log.Printf("[DEBUG] Not generated app, setting shuffle registry")
|
|
// }
|
|
// }
|
|
// }
|
|
// */
|
|
|
|
// if len(localRegistry) == 0 && len(os.Getenv("SHUFFLE_BASE_IMAGE_REGISTRY")) > 0 {
|
|
// localRegistry = os.Getenv("SHUFFLE_BASE_IMAGE_REGISTRY")
|
|
// }
|
|
|
|
// if len(localRegistry) > 0 && strings.Count(image, "/") <= 2 {
|
|
// log.Printf("[DEBUG] Using REGISTRY_URL %s", localRegistry)
|
|
// image = fmt.Sprintf("%s/%s", localRegistry, image)
|
|
// } else {
|
|
// if strings.Count(image, "/") <= 2 {
|
|
// image = fmt.Sprintf("frikky/shuffle:%s", image)
|
|
// }
|
|
// }
|
|
|
|
// log.Printf("[DEBUG] Got kubernetes with namespace %#v to run image '%s'", kubernetesNamespace, image)
|
|
|
|
// //fix naming convention
|
|
// podUuid := uuid.NewV4().String()
|
|
// podName := fmt.Sprintf("%s-%s", value, podUuid)
|
|
|
|
// pod := &corev1.Pod{
|
|
// ObjectMeta: metav1.ObjectMeta{
|
|
// Name: podName,
|
|
// Labels: map[string]string{
|
|
// "app": "shuffle-app",
|
|
// "executionId": workflowExecution.ExecutionId,
|
|
// },
|
|
// },
|
|
// Spec: corev1.PodSpec{
|
|
// RestartPolicy: "Never", // As a crash is not useful in this context
|
|
// // DNSPolicy: "Default",
|
|
// DNSPolicy: corev1.DNSClusterFirst,
|
|
// // NodeName: "worker1"
|
|
// Containers: []corev1.Container{
|
|
// {
|
|
// Name: value,
|
|
// Image: image,
|
|
// Env: buildEnvVars(envMap),
|
|
|
|
// // Pull if not available
|
|
// ImagePullPolicy: corev1.PullIfNotPresent,
|
|
// },
|
|
// },
|
|
// },
|
|
// }
|
|
|
|
// createdPod, err := clientset.CoreV1().Pods(kubernetesNamespace).Create(context.Background(), pod, metav1.CreateOptions{})
|
|
// if err != nil {
|
|
// log.Printf("[ERROR] Failed creating pod: %v", err)
|
|
// // os.Exit(1)
|
|
// } else {
|
|
// log.Printf("[DEBUG] Created pod %#v in namespace %#v", createdPod.Name, kubernetesNamespace)
|
|
// }
|
|
|
|
// return nil
|
|
// }
|
|
|
|
// form basic hostConfig
|
|
ctx := context.Background()
|
|
|
|
// Check action if subflow
|
|
// Check if url is default (shuffle-backend)
|
|
// If it doesn't exist, add it
|
|
|
|
// FIXME: This does NOT replace it in all cases as the data
|
|
// is not saved in the database as the correct param.
|
|
if action.AppName == "shuffle-subflow" {
|
|
// Automatic replacement of URL
|
|
for paramIndex, param := range action.Parameters {
|
|
if param.Name != "backend_url" {
|
|
continue
|
|
}
|
|
|
|
if !strings.Contains(param.Value, "shuffle-backend") {
|
|
continue
|
|
}
|
|
|
|
// Automatic replacement as this is default
|
|
if len(os.Getenv("BASE_URL")) > 0 {
|
|
action.Parameters[paramIndex].Value = os.Getenv("BASE_URL")
|
|
log.Printf("[DEBUG][%s] Replaced backend_url with base_url %s", workflowExecution.ExecutionId, os.Getenv("BASE_URL"))
|
|
}
|
|
|
|
if len(os.Getenv("SHUFFLE_CLOUDRUN_URL")) > 0 {
|
|
action.Parameters[paramIndex].Value = os.Getenv("SHUFFLE_CLOUDRUN_URL")
|
|
log.Printf("[DEBUG][%s] Replaced backend_url with cloudrun %s", workflowExecution.ExecutionId, os.Getenv("SHUFFLE_CLOUDRUN_URL"))
|
|
}
|
|
}
|
|
}
|
|
|
|
/*** STARTREMOVE ***/
|
|
if os.Getenv("SHUFFLE_SWARM_CONFIG") == "run" || os.Getenv("SHUFFLE_SWARM_CONFIG") == "swarm" {
|
|
|
|
appName := strings.Replace(identifier, fmt.Sprintf("_%s", action.ID), "", -1)
|
|
appName = strings.Replace(appName, fmt.Sprintf("_%s", workflowExecution.ExecutionId), "", -1)
|
|
appName = strings.ToLower(appName)
|
|
//log.Printf("[INFO][%s] New appname: %s, image: %s", workflowExecution.ExecutionId, appName, image)
|
|
|
|
if !shuffle.ArrayContains(downloadedImages, image) && isKubernetes != "true" {
|
|
log.Printf("[DEBUG] Downloading image %s from backend as it's first iteration for this image on the worker. Timeout: 60", image)
|
|
// FIXME: Not caring if it's ok or not. Just continuing
|
|
// This is working as intended, just designed to download an updated
|
|
// image on every Orborus/new worker restart.
|
|
|
|
// Running as coroutine for eventual completeness
|
|
// FIXME: With goroutines it got too much trouble of deploying with an older version
|
|
// Allowing slow startups, as long as it's eventually fast, and uses the same registry as on host.
|
|
err := shuffle.DownloadDockerImageBackend(&http.Client{Timeout: imagedownloadTimeout}, image)
|
|
if err == nil {
|
|
downloadedImages = append(downloadedImages, image)
|
|
}
|
|
}
|
|
|
|
var exposedPort int
|
|
var err error
|
|
|
|
if isKubernetes != "true" {
|
|
exposedPort, err = findAppInfo(image, appName, false)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed finding and creating port for %s: %s", appName, err)
|
|
return err
|
|
}
|
|
} else {
|
|
// ** STARTREMOVE ***/
|
|
exposedPort = 80
|
|
//deployport, err := strconv.Atoi(os.Getenv("SHUFFLE_APP_EXPOSED_PORT"))
|
|
//if err == nil {
|
|
// exposedPort = deployport
|
|
//}
|
|
|
|
err = findAppInfoKubernetes(image, appName, env)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed finding and creating port for %s: %s", appName, err)
|
|
return err
|
|
}
|
|
// ** ENDREMOVE ***/
|
|
}
|
|
|
|
/*
|
|
// Makes it not run at all.
|
|
cacheData := []byte("1")
|
|
newExecId := fmt.Sprintf("%s_%s", workflowExecution.ExecutionId, action.ID)
|
|
err = shuffle.SetCache(ctx, newExecId, cacheData, 30)
|
|
if err != nil {
|
|
log.Printf("[WARNING] (1) Failed setting cache for action %s: %s", newExecId, err)
|
|
} else {
|
|
log.Printf("[DEBUG][%s] (1) Adding %s to cache (%#v)", workflowExecution.ExecutionId, newExecId, action.Name)
|
|
}
|
|
*/
|
|
|
|
log.Printf("[DEBUG][%s] Should run towards port %d for app %s. DELAY: %d", workflowExecution.ExecutionId, exposedPort, appName, action.ExecutionDelay)
|
|
ctx := context.Background()
|
|
if action.ExecutionDelay > 0 {
|
|
//log.Printf("[DEBUG] Running app %s with delay of %d", action.Name, action.ExecutionDelay)
|
|
waitTime := time.Duration(action.ExecutionDelay) * time.Second
|
|
|
|
time.AfterFunc(waitTime, func() {
|
|
err = sendAppRequest(ctx, baseUrl, appName, exposedPort, &action, &workflowExecution, image, 0)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed sending SCHEDULED request to app %s on port %d: %s", appName, exposedPort, err)
|
|
}
|
|
})
|
|
|
|
} else {
|
|
rand.Seed(time.Now().UnixNano())
|
|
waitTime := time.Duration(rand.Intn(500)) * time.Millisecond
|
|
|
|
// Added a random delay + context timeout to ensure that the function returns, and only once
|
|
time.AfterFunc(waitTime, func() {
|
|
ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
|
|
defer cancel() // Cancel the context to release resources even if not used
|
|
|
|
go sendAppRequest(ctx, baseUrl, appName, exposedPort, &action, &workflowExecution, image, 0)
|
|
})
|
|
}
|
|
|
|
return nil
|
|
}
|
|
/*** ENDREMOVE ***/
|
|
|
|
// Max 10% CPU every second
|
|
//CPUShares: 128,
|
|
//CPUQuota: 10000,
|
|
//CPUPeriod: 100000,
|
|
hostConfig := &container.HostConfig{
|
|
LogConfig: container.LogConfig{
|
|
Type: "json-file",
|
|
Config: map[string]string{
|
|
"max-size": "10m",
|
|
},
|
|
},
|
|
Resources: container.Resources{},
|
|
}
|
|
|
|
if os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "swarm" {
|
|
hostConfig.NetworkMode = container.NetworkMode(fmt.Sprintf("container:worker-%s", workflowExecution.ExecutionId))
|
|
//log.Printf("Environments: %#v", env)
|
|
}
|
|
|
|
// Removing because log extraction should happen first
|
|
if strings.ToLower(cleanupEnv) == "true" {
|
|
hostConfig.AutoRemove = true
|
|
}
|
|
|
|
// Get environment for certificates
|
|
volumeBinds := []string{}
|
|
volumeBindString := os.Getenv("SHUFFLE_VOLUME_BINDS")
|
|
if len(volumeBindString) > 0 {
|
|
volumeBindSplit := strings.Split(volumeBindString, ",")
|
|
for _, volumeBind := range volumeBindSplit {
|
|
if volumeBind == "srcfolder=dstfolder" || volumeBind == "srcfolder:dstfolder" || volumeBind == "/srcfolder:/dstfolder" {
|
|
log.Printf("[DEBUG] Volume bind '%s' is invalid.", volumeBind)
|
|
continue
|
|
}
|
|
|
|
if !strings.HasPrefix(volumeBind, "/") {
|
|
log.Printf("[ERROR] Volume bind '%s' is invalid. Use absolute paths.", volumeBind)
|
|
continue
|
|
}
|
|
|
|
if !strings.Contains(volumeBind, ":") {
|
|
log.Printf("[ERROR] Volume bind '%s' is invalid. Use absolute paths with colon inbetween them (/srcpath:dstpath/", volumeBind)
|
|
continue
|
|
}
|
|
|
|
volumeBinds = append(volumeBinds, volumeBind)
|
|
}
|
|
}
|
|
|
|
// Add more volume binds if possible
|
|
if len(volumeBinds) > 0 {
|
|
|
|
// Only use mounts, not direct binds
|
|
hostConfig.Binds = []string{}
|
|
hostConfig.Mounts = []mount.Mount{}
|
|
for _, bind := range volumeBinds {
|
|
if !strings.Contains(bind, ":") || strings.Contains(bind, "..") || strings.HasPrefix(bind, "~") {
|
|
log.Printf("[ERROR] Volume bind '%s' is invalid. Use absolute paths.", bind)
|
|
continue
|
|
}
|
|
|
|
log.Printf("[DEBUG] Appending bind %s to App container", bind)
|
|
bindSplit := strings.Split(bind, ":")
|
|
sourceFolder := bindSplit[0]
|
|
destinationFolder := bindSplit[1]
|
|
|
|
readOnly := false
|
|
if len(bindSplit) > 2 {
|
|
mode := bindSplit[2]
|
|
if mode == "ro" {
|
|
readOnly = true
|
|
}
|
|
}
|
|
|
|
builtMount := mount.Mount{
|
|
Type: mount.TypeBind,
|
|
Source: sourceFolder,
|
|
Target: destinationFolder,
|
|
ReadOnly: readOnly,
|
|
}
|
|
|
|
hostConfig.Mounts = append(hostConfig.Mounts, builtMount)
|
|
}
|
|
}
|
|
|
|
config := &container.Config{
|
|
Image: image,
|
|
Env: env,
|
|
}
|
|
|
|
//log.Printf("[DEBUG] Deploying image with env: %#v", env)
|
|
|
|
// Checking as late as possible, just in case.
|
|
newExecId := fmt.Sprintf("%s_%s", workflowExecution.ExecutionId, action.ID)
|
|
_, err := shuffle.GetCache(ctx, newExecId)
|
|
if err == nil {
|
|
log.Printf("[DEBUG][%s] Result for action %s already found - returning", newExecId, action.ID)
|
|
return nil
|
|
}
|
|
|
|
cacheData := []byte("1")
|
|
err = shuffle.SetCache(ctx, newExecId, cacheData, 30)
|
|
if err != nil {
|
|
//log.Printf("[WARNING][%s] Failed setting cache for action: %s", newExecId, err)
|
|
} else {
|
|
//log.Printf("[DEBUG][%s] Adding to cache. Name: %s", workflowExecution.ExecutionId, action.Name)
|
|
}
|
|
|
|
if action.ExecutionDelay > 0 {
|
|
log.Printf("[DEBUG][%s] Running app '%s' with label '%s' in docker with delay of %d", workflowExecution.ExecutionId, action.AppName, action.Label, action.ExecutionDelay)
|
|
waitTime := time.Duration(action.ExecutionDelay) * time.Second
|
|
|
|
time.AfterFunc(waitTime, func() {
|
|
DeployContainer(ctx, cli, config, hostConfig, identifier, workflowExecution, newExecId)
|
|
})
|
|
} else {
|
|
log.Printf("[DEBUG][%s] Running app %s in docker NORMALLY as there is no delay set with identifier %s", workflowExecution.ExecutionId, action.Name, identifier)
|
|
returnvalue := DeployContainer(ctx, cli, config, hostConfig, identifier, workflowExecution, newExecId)
|
|
//log.Printf("[DEBUG][%s] Normal deploy ret: %s", workflowExecution.ExecutionId, returnvalue)
|
|
return returnvalue
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func cleanupKubernetesExecution(clientset *kubernetes.Clientset, workflowExecution shuffle.WorkflowExecution, namespace string) error {
|
|
// workerName := fmt.Sprintf("worker-%s", workflowExecution.ExecutionId)
|
|
// FIXME: The executionId label is currently not set
|
|
labelSelector := fmt.Sprintf("app.kubernetes.io/name=shuffle-app,executionId=%s", workflowExecution.ExecutionId)
|
|
|
|
podList, err := clientset.CoreV1().Pods(namespace).List(context.TODO(), metav1.ListOptions{
|
|
LabelSelector: labelSelector,
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("[ERROR] Failed to list apps with label selector %s: %#vv", labelSelector, err)
|
|
}
|
|
|
|
for _, pod := range podList.Items {
|
|
err := clientset.CoreV1().Pods(namespace).Delete(context.TODO(), pod.Name, metav1.DeleteOptions{})
|
|
if err != nil {
|
|
return fmt.Errorf("failed to delete app %s: %v", pod.Name, err)
|
|
}
|
|
log.Printf("App %s in namespace %s deleted.", pod.Name, namespace)
|
|
}
|
|
|
|
// podErr := clientset.CoreV1().Pods(namespace).Delete(context.TODO(), workerName, metav1.DeleteOptions{})
|
|
// if podErr != nil {
|
|
// return fmt.Errorf("[ERROR] failed to delete the worker %s in namespace %s: %v", workerName, namespace, podErr)
|
|
// }
|
|
// log.Printf("[DEBUG] %s in namespace %s deleted.", workerName, namespace)
|
|
return nil
|
|
}
|
|
|
|
func DeployContainer(ctx context.Context, cli *dockerclient.Client, config *container.Config, hostConfig *container.HostConfig, identifier string, workflowExecution shuffle.WorkflowExecution, actionExecId string) error {
|
|
cont, err := cli.ContainerCreate(
|
|
ctx,
|
|
config,
|
|
hostConfig,
|
|
nil,
|
|
nil,
|
|
identifier,
|
|
)
|
|
|
|
//log.Printf("[DEBUG] config set: %#v", config)
|
|
|
|
if err != nil {
|
|
//log.Printf("[ERROR] Failed creating container: %s", err)
|
|
if !strings.Contains(err.Error(), "Conflict. The container name") {
|
|
log.Printf("[ERROR] Container CREATE error (1): %s", err)
|
|
|
|
cacheErr := shuffle.DeleteCache(ctx, actionExecId)
|
|
if cacheErr != nil {
|
|
log.Printf("[ERROR] FAILURE Deleting cache for %s: %s", actionExecId, cacheErr)
|
|
}
|
|
|
|
return err
|
|
} else {
|
|
parsedUuid := uuid.NewV4()
|
|
identifier = fmt.Sprintf("%s-%s", identifier, parsedUuid)
|
|
//hostConfig.NetworkMode = container.NetworkMode(fmt.Sprintf("container:worker-%s", workflowExecution.ExecutionId))
|
|
|
|
log.Printf("[DEBUG] 2 - Identifier: %s", identifier)
|
|
cont, err = cli.ContainerCreate(
|
|
context.Background(),
|
|
config,
|
|
hostConfig,
|
|
nil,
|
|
nil,
|
|
identifier,
|
|
)
|
|
|
|
if err != nil {
|
|
log.Printf("[ERROR] Container create error (2): %s", err)
|
|
|
|
cacheErr := shuffle.DeleteCache(ctx, actionExecId)
|
|
if cacheErr != nil {
|
|
log.Printf("[ERROR] FAILURE Deleting cache for %s: %s", actionExecId, cacheErr)
|
|
}
|
|
|
|
return err
|
|
}
|
|
|
|
//log.Printf("[DEBUG] Made new container ID
|
|
}
|
|
}
|
|
|
|
err = cli.ContainerStart(ctx, cont.ID, container.StartOptions{})
|
|
if err != nil {
|
|
if strings.Contains(fmt.Sprintf("%s", err), "cannot join network") || strings.Contains(fmt.Sprintf("%s", err), "No such container") {
|
|
// Remove the "CREATED" one from the previous if possible
|
|
go cli.ContainerRemove(ctx, cont.ID, container.RemoveOptions{})
|
|
|
|
log.Printf("[WARNING] Failed deploying App on first attempt: %s. Removing some HostConfig configs.", err)
|
|
parsedUuid := uuid.NewV4()
|
|
identifier = fmt.Sprintf("%s-%s-nonetwork", identifier, parsedUuid)
|
|
|
|
hostConfig.NetworkMode = container.NetworkMode("")
|
|
hostConfig.LogConfig = container.LogConfig{
|
|
Type: "json-file",
|
|
Config: map[string]string{
|
|
"max-size": "10m",
|
|
},
|
|
}
|
|
hostConfig.Resources = container.Resources{}
|
|
|
|
cont, err = cli.ContainerCreate(
|
|
context.Background(),
|
|
config,
|
|
hostConfig,
|
|
nil,
|
|
nil,
|
|
identifier,
|
|
)
|
|
|
|
if err != nil {
|
|
log.Printf("[ERROR] Container create error (3): %s", err)
|
|
|
|
cacheErr := shuffle.DeleteCache(ctx, actionExecId)
|
|
if cacheErr != nil {
|
|
log.Printf("[ERROR] FAILURE Deleting cache for %s: %s", actionExecId, cacheErr)
|
|
}
|
|
|
|
return err
|
|
}
|
|
|
|
//log.Printf("[DEBUG] Running secondary check without network with worker")
|
|
err = cli.ContainerStart(ctx, cont.ID, container.StartOptions{})
|
|
}
|
|
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed to start container (2) in runtime location %s: %s", environment, err)
|
|
|
|
cacheErr := shuffle.DeleteCache(ctx, actionExecId)
|
|
if cacheErr != nil {
|
|
log.Printf("[ERROR] FAILURE Deleting cache for %s: %s", actionExecId, cacheErr)
|
|
}
|
|
|
|
//shutdown(workflowExecution, workflowExecution.Workflow.ID, true)
|
|
return err
|
|
}
|
|
}
|
|
|
|
log.Printf("[DEBUG][%s] Container %s was created for %s", workflowExecution.ExecutionId, cont.ID, identifier)
|
|
|
|
// Waiting to see if it exits.. Stupid, but stable(r)
|
|
if workflowExecution.ExecutionSource != "default" {
|
|
log.Printf("[INFO][%s] Handling NON-default execution source %s - NOT waiting or validating!", workflowExecution.ExecutionId, workflowExecution.ExecutionSource)
|
|
} else if workflowExecution.ExecutionSource == "default" {
|
|
log.Printf("[INFO][%s] Handling DEFAULT execution source %s - SKIPPING wait anyway due to exited issues!", workflowExecution.ExecutionId, workflowExecution.ExecutionSource)
|
|
}
|
|
|
|
//log.Printf("[DEBUG] Deployed container ID %s", cont.ID)
|
|
//containerIds = append(containerIds, cont.ID)
|
|
|
|
return nil
|
|
}
|
|
|
|
func removeContainer(containername string) error {
|
|
ctx := context.Background()
|
|
|
|
// cli, err := dockerclient.NewEnvClient()
|
|
cli, _, err := shuffle.GetDockerClient()
|
|
if err != nil {
|
|
log.Printf("[DEBUG] Unable to create docker client: %s", err)
|
|
return err
|
|
}
|
|
|
|
defer cli.Close()
|
|
|
|
// FIXME - ucnomment
|
|
// containers, err := cli.ContainerList(ctx, types.ContainerListOptions{
|
|
// All: true,
|
|
// })
|
|
|
|
_ = ctx
|
|
_ = cli
|
|
//if err := cli.ContainerStop(ctx, containername, nil); err != nil {
|
|
// log.Printf("Unable to stop container %s - running removal anyway, just in case: %s", containername, err)
|
|
//}
|
|
|
|
removeOptions := container.RemoveOptions{
|
|
RemoveVolumes: true,
|
|
Force: true,
|
|
}
|
|
|
|
// FIXME - remove comments etc
|
|
_ = removeOptions
|
|
//if err := cli.ContainerRemove(ctx, containername, removeOptions); err != nil {
|
|
// log.Printf("Unable to remove container: %s", err)
|
|
//}
|
|
|
|
return nil
|
|
}
|
|
|
|
func runFilter(workflowExecution shuffle.WorkflowExecution, action shuffle.Action) {
|
|
// 1. Get the parameter $.#.id
|
|
if action.Label == "filter_cases" && len(action.Parameters) > 0 {
|
|
if action.Parameters[0].Variant == "ACTION_RESULT" {
|
|
param := action.Parameters[0]
|
|
value := param.Value
|
|
_ = value
|
|
|
|
// Loop cases.. Hmm, that's tricky
|
|
}
|
|
} else {
|
|
log.Printf("No handler for filter %s with %d params", action.Label, len(action.Parameters))
|
|
}
|
|
}
|
|
|
|
func removeIndex(s []string, i int) []string {
|
|
s[len(s)-1], s[i] = s[i], s[len(s)-1]
|
|
return s[:len(s)-1]
|
|
}
|
|
|
|
func countHealthyServiceTasks(dockercli *dockerclient.Client, serviceID string) (int, error) {
|
|
ctx := context.Background()
|
|
filters := filters.NewArgs()
|
|
filters.Add("service", serviceID)
|
|
filters.Add("desired-state", "running")
|
|
|
|
tasks, err := dockercli.TaskList(ctx, types.TaskListOptions{Filters: filters})
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
|
|
count := 0
|
|
for _, task := range tasks {
|
|
if task.Status.State == swarm.TaskStateRunning {
|
|
count += 1
|
|
}
|
|
}
|
|
|
|
return count, nil
|
|
}
|
|
|
|
func getWorkerURLs() ([]string, error) {
|
|
workerUrls := []string{}
|
|
|
|
if isKubernetes == "true" {
|
|
workerUrls = append(workerUrls, "http://shuffle-workers:33333")
|
|
// workerUrls = append(workerUrls, "http://192.168.29.16:33333")
|
|
|
|
// get service "shuffle-workers" "Endpoints"
|
|
// serviceName := "shuffle-workers"
|
|
// clientset, _, err := shuffle.GetKubernetesClient()
|
|
// if err != nil {
|
|
// log.Println("[ERROR] Failed to get Kubernetes client:", err)
|
|
// return workerUrls, err
|
|
// }
|
|
|
|
// services, err := clientset.CoreV1().Services("default").List(context.Background(), metav1.ListOptions{})
|
|
// if err != nil {
|
|
// log.Println("[ERROR] Failed to list services:", err)
|
|
// return workerUrls, err
|
|
// }
|
|
|
|
// for _, service := range services.Items {
|
|
// if service.Name == serviceName {
|
|
// endpoints, err := clientset.CoreV1().Endpoints("default").Get(context.Background(), serviceName, metav1.GetOptions{})
|
|
// if err != nil {
|
|
// log.Println("[ERROR] Failed to get endpoints for service:", err)
|
|
// return workerUrls, err
|
|
// }
|
|
|
|
// for _, subset := range endpoints.Subsets {
|
|
// for _, address := range subset.Addresses {
|
|
// for _, port := range subset.Ports {
|
|
// url := fmt.Sprintf("http://%s:%d", address.IP, port.Port)
|
|
// workerUrls = append(workerUrls, url)
|
|
// }
|
|
// }
|
|
// }
|
|
// }
|
|
// }
|
|
//log.Printf("[DEBUG] Worker URLs for k8s: %#v", workerUrls)
|
|
|
|
return workerUrls, nil
|
|
}
|
|
|
|
// Create a new Docker client
|
|
// cli, err := dockerclient.NewEnvClient()
|
|
cli, _, err := shuffle.GetDockerClient()
|
|
if err != nil {
|
|
log.Println("[ERROR] Failed to create Docker client:", err)
|
|
return workerUrls, err
|
|
}
|
|
|
|
defer cli.Close()
|
|
|
|
// Specify the name of the service for which you want to list tasks
|
|
serviceName := "shuffle-workers"
|
|
|
|
// Get the list of tasks for the service
|
|
tasks, err := cli.TaskList(context.Background(), types.TaskListOptions{
|
|
Filters: filters.NewArgs(filters.Arg("service", serviceName)),
|
|
})
|
|
|
|
if err != nil {
|
|
log.Println("[ERROR] Failed to list tasks for service:", err)
|
|
return workerUrls, err
|
|
}
|
|
|
|
// Print task information
|
|
for _, task := range tasks {
|
|
url := fmt.Sprintf("http://%s.%d.%s:33333", serviceName, task.Slot, task.ID)
|
|
workerUrls = append(workerUrls, url)
|
|
}
|
|
|
|
return workerUrls, nil
|
|
}
|
|
|
|
func askOtherWorkersToDownloadImage(image string) {
|
|
// Why wouldn't it happen on swarm? Hmm
|
|
if os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "swarm" {
|
|
return
|
|
}
|
|
|
|
// Check environment SHUFFLE_AUTO_IMAGE_DOWNLOAD
|
|
if os.Getenv("SHUFFLE_AUTO_IMAGE_DOWNLOAD") == "false" {
|
|
log.Printf("[DEBUG] SHUFFLE_AUTO_IMAGE_DOWNLOAD is false. NOT distributing images %s", image)
|
|
return
|
|
}
|
|
|
|
if shuffle.ArrayContains(imagesDistributed, image) {
|
|
return
|
|
}
|
|
|
|
urls, err := getWorkerURLs()
|
|
if err != nil {
|
|
log.Printf("[ERROR] Error in listing worker urls: %s", err)
|
|
return
|
|
}
|
|
|
|
if len(urls) < 2 {
|
|
return
|
|
}
|
|
|
|
httpClient := &http.Client{}
|
|
distributed := false
|
|
for _, url := range urls {
|
|
//log.Printf("[DEBUG] Trying to speak to: %s", url)
|
|
imagesRequest := ImageRequest{
|
|
Image: image,
|
|
}
|
|
|
|
url = fmt.Sprintf("%s/api/v1/download", url)
|
|
|
|
//log.Printf("[INFO] Making a request to %s to download images", url)
|
|
imageJSON, err := json.Marshal(imagesRequest)
|
|
req, err := http.NewRequest(
|
|
"POST",
|
|
url,
|
|
bytes.NewBuffer(imageJSON),
|
|
)
|
|
|
|
if err != nil {
|
|
log.Printf("[ERROR] Error in making request to %s : %s", url, err)
|
|
continue
|
|
}
|
|
|
|
resp, err := httpClient.Do(req)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Error in making request to %s : %s", url, err)
|
|
continue
|
|
}
|
|
|
|
defer resp.Body.Close()
|
|
respBody, err := ioutil.ReadAll(resp.Body)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Error in reading response body : %s", err)
|
|
continue
|
|
}
|
|
|
|
log.Printf("[INFO] Response body when tried sending images for nodes to download: %s", respBody)
|
|
distributed = true
|
|
}
|
|
|
|
if distributed {
|
|
imagesDistributed = append(imagesDistributed, image)
|
|
}
|
|
}
|
|
|
|
func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
|
|
ctx := context.Background()
|
|
|
|
workflowExecution, relevantActions := shuffle.DecideExecution(ctx, workflowExecution, environment)
|
|
if workflowExecution.Status == "FINISHED" || workflowExecution.Status == "FAILURE" || workflowExecution.Status == "ABORTED" {
|
|
log.Printf("[DEBUG][%s] Shutting down because status is %s", workflowExecution.ExecutionId, workflowExecution.Status)
|
|
shutdown(workflowExecution, "", "Workflow run is already finished", true)
|
|
return
|
|
}
|
|
|
|
startAction, extra, children, parents, visited, executed, nextActions, environments := shuffle.GetExecutionVariables(ctx, workflowExecution.ExecutionId)
|
|
|
|
var dockercli *dockerclient.Client
|
|
var err error
|
|
|
|
if isKubernetes != "true" {
|
|
// dockercli, err = dockerclient.NewEnvClient()
|
|
dockercli, _, err = shuffle.GetDockerClient()
|
|
if err != nil {
|
|
log.Printf("[ERROR] Unable to create docker client (3): %s", err)
|
|
return
|
|
}
|
|
defer dockercli.Close()
|
|
}
|
|
|
|
for _, action := range relevantActions {
|
|
appname := action.AppName
|
|
appversion := action.AppVersion
|
|
appname = strings.Replace(appname, ".", "-", -1)
|
|
appversion = strings.Replace(appversion, ".", "-", -1)
|
|
|
|
action, _ = singul.HandleSingulStartnode(workflowExecution, action, []string{})
|
|
|
|
parsedAppname := strings.Replace(strings.ToLower(action.AppName), " ", "-", -1)
|
|
// if strings.ToLower(parsedAppname) == "singul" {
|
|
// parsedAppname = "shuffle-ai"
|
|
// appversion = "1.1.0"
|
|
// appname = "shuffle-ai"
|
|
// }
|
|
|
|
if parsedAppname == "ai-agent" {
|
|
parsedAppname = "shuffle-ai"
|
|
appversion = "1.1.0"
|
|
appname = "shuffle-ai"
|
|
|
|
action.AppVersion = "1.1.0"
|
|
action.AppName = "shuffle-ai"
|
|
action.Name = "run_agent"
|
|
|
|
inputParamValue := ""
|
|
allowedActions := ""
|
|
for _, param := range action.Parameters {
|
|
if strings.ToLower(param.Name) == "input" {
|
|
inputParamValue = param.Value
|
|
} else if strings.ToLower(param.Name) == "action" {
|
|
param.Value = strings.ReplaceAll("Nothing,", " ", "")
|
|
param.Value = strings.ReplaceAll("Nothing", " ", "")
|
|
allowedActions = param.Value
|
|
}
|
|
}
|
|
|
|
// Rewriting them
|
|
action.Parameters = []shuffle.WorkflowAppActionParameter{
|
|
shuffle.WorkflowAppActionParameter{
|
|
Name: "input_data",
|
|
Value: inputParamValue,
|
|
},
|
|
shuffle.WorkflowAppActionParameter{
|
|
Name: "actions",
|
|
Value: allowedActions,
|
|
},
|
|
}
|
|
}
|
|
|
|
imageName := fmt.Sprintf("%s:%s_%s", baseimagename, parsedAppname, action.AppVersion)
|
|
if strings.Contains(imageName, " ") {
|
|
imageName = strings.ReplaceAll(imageName, " ", "-")
|
|
}
|
|
|
|
// Kubernetes specific.
|
|
// Should it be though?
|
|
if isKubernetes == "true" {
|
|
// Map it to:
|
|
// <registry>/baseimagename/<appname>:<appversion>
|
|
localRegistry := os.Getenv("REGISTRY_URL")
|
|
if len(localRegistry) > 0 && len(baseimagename) > 0 {
|
|
|
|
newImageName := fmt.Sprintf("%s/%s/%s:%s", localRegistry, baseimagename, parsedAppname, action.AppVersion)
|
|
|
|
log.Printf("[INFO] Remapping image name %s to %s due to registry+image name existing on k8s", imageName, newImageName)
|
|
|
|
imageName = newImageName
|
|
|
|
}
|
|
}
|
|
|
|
askOtherWorkersToDownloadImage(imageName)
|
|
|
|
// Added UUID to identifier just in case
|
|
//identifier := fmt.Sprintf("%s_%s_%s_%s_%s", appname, appversion, action.ID, workflowExecution.ExecutionId, uuid.NewV4())
|
|
identifier := fmt.Sprintf("%s_%s_%s_%s", appname, appversion, action.ID, workflowExecution.ExecutionId)
|
|
if strings.Contains(identifier, " ") {
|
|
identifier = strings.ReplaceAll(identifier, " ", "-")
|
|
}
|
|
|
|
//if arrayContains(executed, action.ID) || arrayContains(visited, action.ID) {
|
|
// log.Printf("[WARNING] Action %s is already executed")
|
|
// continue
|
|
//}
|
|
//visited = append(visited, action.ID)
|
|
//executed = append(executed, action.ID)
|
|
|
|
// FIXME - check whether it's running locally yet too
|
|
// take care of auto clean up later on for k8s
|
|
if isKubernetes != "true" {
|
|
stats, err := dockercli.ContainerInspect(context.Background(), identifier)
|
|
if err != nil || stats.ContainerJSONBase.State.Status != "running" {
|
|
// REMOVE
|
|
if err == nil {
|
|
log.Printf("[DEBUG][%s] Docker Container Status: %s, should kill: %s", workflowExecution.ExecutionId, stats.ContainerJSONBase.State.Status, identifier)
|
|
err = removeContainer(identifier)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Error killing container: %s", err)
|
|
}
|
|
} else {
|
|
//log.Printf("WHAT TO DO HERE?: %s", err)
|
|
}
|
|
} else if stats.ContainerJSONBase.State.Status == "running" {
|
|
//log.Printf("
|
|
continue
|
|
}
|
|
}
|
|
|
|
if len(action.Parameters) == 0 {
|
|
action.Parameters = []shuffle.WorkflowAppActionParameter{}
|
|
}
|
|
|
|
if len(action.Errors) == 0 {
|
|
action.Errors = []string{}
|
|
}
|
|
|
|
// marshal action and put it in there rofl
|
|
//log.Printf("[INFO][%s] Time to execute %s (%s) with app %s:%s, function %s, env %s with %d parameters.", workflowExecution.ExecutionId, action.ID, action.Label, action.AppName, action.AppVersion, action.Name, action.Environment, len(action.Parameters))
|
|
|
|
log.Printf("[DEBUG][%s] Action: Send, Label: '%s', Action: '%s', Run status: %s, Extra=", workflowExecution.ExecutionId, action.Label, action.AppName, workflowExecution.Status)
|
|
|
|
actionData, err := json.Marshal(action)
|
|
if err != nil {
|
|
log.Printf("[WARNING] Failed unmarshalling action: %s", err)
|
|
continue
|
|
}
|
|
|
|
if action.AppID == "0ca8887e-b4af-4e3e-887c-87e9d3bc3d3e" {
|
|
log.Printf("[DEBUG] Should run filter: %#v", action)
|
|
runFilter(workflowExecution, action)
|
|
continue
|
|
}
|
|
|
|
executionData, err := json.Marshal(workflowExecution)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed marshalling executiondata: %s", err)
|
|
executionData = []byte("")
|
|
}
|
|
|
|
// Sending full execution so that it won't have to load in every app
|
|
// This might be an issue if they can read environments, but that's alright
|
|
// if everything is generated during execution
|
|
//log.Printf("[DEBUG][%s] Deployed with CALLBACK_URL %s and BASE_URL %s", workflowExecution.ExecutionId, appCallbackUrl, baseUrl)
|
|
env := []string{
|
|
fmt.Sprintf("EXECUTIONID=%s", workflowExecution.ExecutionId),
|
|
fmt.Sprintf("AUTHORIZATION=%s", workflowExecution.Authorization),
|
|
fmt.Sprintf("CALLBACK_URL=%s", baseUrl),
|
|
fmt.Sprintf("BASE_URL=%s", appCallbackUrl),
|
|
fmt.Sprintf("TZ=%s", timezone),
|
|
fmt.Sprintf("SHUFFLE_LOGS_DISABLED=%s", logsDisabled),
|
|
}
|
|
|
|
if len(actionData) >= 100000 {
|
|
log.Printf("[WARNING] Omitting some data from action execution. Length: %d. Fix in SDK!", len(actionData))
|
|
newParams := []shuffle.WorkflowAppActionParameter{}
|
|
for _, param := range action.Parameters {
|
|
paramData, err := json.Marshal(param)
|
|
if err != nil {
|
|
log.Printf("[WARNING] Failed to marshal param %s: %s", param.Name, err)
|
|
newParams = append(newParams, param)
|
|
continue
|
|
}
|
|
|
|
if len(paramData) >= 50000 {
|
|
log.Printf("[WARNING] Removing a lot of data from param %s with length %d", param.Name, len(paramData))
|
|
param.Value = "SHUFFLE_AUTO_REMOVED"
|
|
}
|
|
|
|
newParams = append(newParams, param)
|
|
}
|
|
|
|
action.Parameters = newParams
|
|
actionData, err = json.Marshal(action)
|
|
if err == nil {
|
|
log.Printf("[DEBUG] Ran data replace on action %s. new length: %d", action.Name, len(actionData))
|
|
} else {
|
|
log.Printf("[WARNING] Failed to marshal new actionData: %s", err)
|
|
|
|
}
|
|
} else {
|
|
//log.Printf("[DEBUG] Actiondata is NOT 100000 in length. Adding as normal.")
|
|
}
|
|
|
|
actionEnv := fmt.Sprintf("ACTION=%s", string(actionData))
|
|
env = append(env, actionEnv)
|
|
|
|
if strings.ToLower(os.Getenv("SHUFFLE_PASS_APP_PROXY")) == "true" {
|
|
//log.Printf("APPENDING PROXY TO THE APP!")
|
|
env = appendAppProxyEnv(env)
|
|
}
|
|
|
|
overrideHttpProxy := os.Getenv("SHUFFLE_INTERNAL_HTTP_PROXY")
|
|
overrideHttpsProxy := os.Getenv("SHUFFLE_INTERNAL_HTTPS_PROXY")
|
|
if overrideHttpProxy != "" {
|
|
env = append(env, fmt.Sprintf("SHUFFLE_INTERNAL_HTTP_PROXY=%s", overrideHttpProxy))
|
|
}
|
|
|
|
if overrideHttpsProxy != "" {
|
|
env = append(env, fmt.Sprintf("SHUFFLE_INTERNAL_HTTPS_PROXY=%s", overrideHttpsProxy))
|
|
}
|
|
|
|
proxyConfigOverride := os.Getenv("SHUFFLE_APP_PROXY_CONFIG_OVERRIDE")
|
|
if proxyConfigOverride != "" {
|
|
env = append(env, fmt.Sprintf("SHUFFLE_APP_PROXY_CONFIG_OVERRIDE=%s", proxyConfigOverride))
|
|
}
|
|
|
|
if len(os.Getenv("SHUFFLE_APP_SDK_TIMEOUT")) > 0 {
|
|
env = append(env, fmt.Sprintf("SHUFFLE_APP_SDK_TIMEOUT=%s", os.Getenv("SHUFFLE_APP_SDK_TIMEOUT")))
|
|
}
|
|
|
|
// FIXME: Ensure to NEVER do this anymore
|
|
// This potentially breaks too much stuff. Better to have the app poll the data.
|
|
_ = executionData
|
|
/*
|
|
maxSize := 32700 - len(string(actionData)) - 2000
|
|
if len(executionData) < maxSize {
|
|
log.Printf("[INFO] ADDING FULL_EXECUTION because size is smaller than %d", maxSize)
|
|
env = append(env, fmt.Sprintf("FULL_EXECUTION=%s", string(executionData)))
|
|
} else {
|
|
log.Printf("[WARNING] Skipping FULL_EXECUTION because size is larger than %d", maxSize)
|
|
}
|
|
*/
|
|
|
|
// Uses a few ways of getting / checking if an app is available
|
|
// 1. Try original with lowercase
|
|
// 2. Go to original (no spaces)
|
|
// 3. Add remote repo location
|
|
images := []string{
|
|
imageName,
|
|
fmt.Sprintf("%s/%s:%s_%s", registryName, baseimagename, parsedAppname, action.AppVersion),
|
|
fmt.Sprintf("%s:%s_%s", baseimagename, parsedAppname, action.AppVersion),
|
|
}
|
|
|
|
// This is the weirdest shit ever looking back at
|
|
// Needs optimization lol
|
|
|
|
pullOptions := dockerimage.PullOptions{}
|
|
if strings.ToLower(cleanupEnv) == "true" {
|
|
err = deployApp(dockercli, images[0], identifier, env, workflowExecution, action)
|
|
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
|
|
if strings.Contains(err.Error(), "exited prematurely") {
|
|
log.Printf("[DEBUG] Shutting down (2)")
|
|
shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true)
|
|
return
|
|
}
|
|
|
|
err := shuffle.DownloadDockerImageBackend(&http.Client{Timeout: imagedownloadTimeout}, imageName)
|
|
executed := false
|
|
if err == nil {
|
|
log.Printf("[DEBUG] Downloaded image %s from backend (CLEANUP)", imageName)
|
|
downloadedImages = append(downloadedImages, imageName)
|
|
//err = deployApp(dockercli, image, identifier, env, workflow, action)
|
|
err = deployApp(dockercli, imageName, identifier, env, workflowExecution, action)
|
|
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
|
|
if strings.Contains(err.Error(), "exited prematurely") {
|
|
log.Printf("[DEBUG] Shutting down (41)")
|
|
shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true)
|
|
return
|
|
}
|
|
} else {
|
|
executed = true
|
|
}
|
|
}
|
|
|
|
if !executed {
|
|
imageName = images[2]
|
|
err = deployApp(dockercli, imageName, identifier, env, workflowExecution, action)
|
|
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
|
|
if strings.Contains(err.Error(), "exited prematurely") {
|
|
log.Printf("[DEBUG] Shutting down (3)")
|
|
shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true)
|
|
return
|
|
}
|
|
|
|
//log.Printf("[WARNING] Failed CLEANUP execution. Downloading image %s remotely.", image)
|
|
|
|
log.Printf("[WARNING] Failed to download image %s (CLEANUP): %s", imageName, err)
|
|
|
|
reader, err := dockercli.ImagePull(context.Background(), imageName, pullOptions)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed getting %s. Couldn't be find locally, AND is missing.", imageName)
|
|
log.Printf("[DEBUG] Shutting down (4)")
|
|
shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true)
|
|
return
|
|
} else {
|
|
defer reader.Close()
|
|
baseTag := strings.Split(imageName, ":")
|
|
if len(baseTag) > 1 {
|
|
tag := baseTag[1]
|
|
log.Printf("[DEBUG] Creating tag copies of registry downloaded containers from tag %s", tag)
|
|
|
|
// Remapping
|
|
ctx := context.Background()
|
|
dockercli.ImageTag(ctx, imageName, fmt.Sprintf("frikky/shuffle:%s", tag))
|
|
dockercli.ImageTag(ctx, imageName, fmt.Sprintf("registry.hub.docker.com/frikky/shuffle:%s", tag))
|
|
}
|
|
}
|
|
|
|
buildBuf := new(strings.Builder)
|
|
_, err = io.Copy(buildBuf, reader)
|
|
if err != nil && !strings.Contains(fmt.Sprintf("%s", err.Error()), "Conflict. The container name") {
|
|
log.Printf("[ERROR] Error in IO copy: %s", err)
|
|
log.Printf("[DEBUG] Shutting down (5)")
|
|
shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true)
|
|
return
|
|
} else {
|
|
if strings.Contains(buildBuf.String(), "errorDetail") {
|
|
log.Printf("[ERROR] Docker build:%sERROR ABOVE: Trying to pull tags from: %s", buildBuf.String(), imageName)
|
|
log.Printf("[DEBUG] Shutting down (6)")
|
|
shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true)
|
|
return
|
|
}
|
|
|
|
log.Printf("[INFO] Successfully downloaded %s", imageName)
|
|
}
|
|
|
|
err = deployApp(dockercli, imageName, identifier, env, workflowExecution, action)
|
|
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
|
|
|
|
log.Printf("[ERROR] Failed deploying image for the FOURTH time. Aborting if the image doesn't exist")
|
|
if strings.Contains(err.Error(), "exited prematurely") {
|
|
log.Printf("[DEBUG] Shutting down (7)")
|
|
shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true)
|
|
return
|
|
}
|
|
|
|
if strings.Contains(err.Error(), "No such image") {
|
|
//log.Printf("[WARNING] Failed deploying %s from image %s: %s", identifier, image, err)
|
|
log.Printf("[ERROR] Image doesn't exist. Shutting down")
|
|
log.Printf("[DEBUG] Shutting down (8)")
|
|
shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true)
|
|
return
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
} else {
|
|
|
|
err = deployApp(dockercli, images[0], identifier, env, workflowExecution, action)
|
|
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
|
|
log.Printf("[DEBUG] Failed deploying app? %s", err)
|
|
if strings.Contains(err.Error(), "exited prematurely") {
|
|
log.Printf("[DEBUG] Shutting down (9)")
|
|
shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true)
|
|
return
|
|
}
|
|
|
|
// Trying to replace with lowercase to deploy again. This seems to work with Dockerhub well.
|
|
// FIXME: Should try to remotely download directly if this persists.
|
|
imageName = images[1]
|
|
err = deployApp(dockercli, imageName, identifier, env, workflowExecution, action)
|
|
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
|
|
if strings.Contains(err.Error(), "exited prematurely") {
|
|
log.Printf("[DEBUG] Shutting down (10)")
|
|
shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true)
|
|
return
|
|
}
|
|
|
|
log.Printf("[DEBUG][%s] Failed deploy. Downloading image %s: %s", workflowExecution.ExecutionId, imageName, err)
|
|
err := shuffle.DownloadDockerImageBackend(&http.Client{Timeout: imagedownloadTimeout}, imageName)
|
|
|
|
executed := false
|
|
if err == nil {
|
|
log.Printf("[DEBUG] Downloaded image %s from backend (CLEANUP)", imageName)
|
|
downloadedImages = append(downloadedImages, imageName)
|
|
//err = deployApp(dockercli, image, identifier, env, workflow, action)
|
|
err = deployApp(dockercli, imageName, identifier, env, workflowExecution, action)
|
|
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
|
|
log.Printf("[ERROR] Err: %s", err)
|
|
if strings.Contains(err.Error(), "exited prematurely") {
|
|
log.Printf("[DEBUG] Shutting down (40)")
|
|
shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true)
|
|
return
|
|
}
|
|
} else {
|
|
executed = true
|
|
}
|
|
}
|
|
|
|
if !executed {
|
|
imageName = images[2]
|
|
err = deployApp(dockercli, imageName, identifier, env, workflowExecution, action)
|
|
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
|
|
log.Printf("[ERROR] Err: %s", err)
|
|
if strings.Contains(err.Error(), "exited prematurely") {
|
|
log.Printf("[DEBUG] Shutting down (11)")
|
|
shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true)
|
|
return
|
|
}
|
|
|
|
log.Printf("[WARNING] Failed deploying image THREE TIMES. Attempting to download %s as last resort from backend and dockerhub: %s", imageName, err)
|
|
|
|
if isKubernetes == "true" {
|
|
log.Printf("[ERROR] Image %s doesn't exist. Returning error for now")
|
|
return
|
|
}
|
|
|
|
reader, err := dockercli.ImagePull(context.Background(), imageName, pullOptions)
|
|
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
|
|
log.Printf("[ERROR] Failed getting %s. The couldn't be find locally, AND is missing.", imageName)
|
|
log.Printf("[DEBUG] Shutting down (12)")
|
|
shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true)
|
|
return
|
|
} else {
|
|
defer reader.Close()
|
|
baseTag := strings.Split(imageName, ":")
|
|
if len(baseTag) > 1 {
|
|
tag := baseTag[1]
|
|
log.Printf("[DEBUG] Creating tag copies of registry downloaded containers from tag %s", tag)
|
|
|
|
// Remapping
|
|
ctx := context.Background()
|
|
dockercli.ImageTag(ctx, imageName, fmt.Sprintf("frikky/shuffle:%s", tag))
|
|
dockercli.ImageTag(ctx, imageName, fmt.Sprintf("registry.hub.docker.com/frikky/shuffle:%s", tag))
|
|
}
|
|
}
|
|
|
|
buildBuf := new(strings.Builder)
|
|
_, err = io.Copy(buildBuf, reader)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Error in IO copy: %s", err)
|
|
log.Printf("[DEBUG] Shutting down (13)")
|
|
shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true)
|
|
return
|
|
} else {
|
|
if strings.Contains(buildBuf.String(), "errorDetail") {
|
|
log.Printf("[ERROR] Docker build:%sERROR ABOVE: Trying to pull tags from: %s", buildBuf.String(), imageName)
|
|
log.Printf("[DEBUG] Shutting down (14)")
|
|
shutdown(workflowExecution, action.ID, fmt.Sprintf("Error deploying container: %s", buildBuf.String()), true)
|
|
return
|
|
}
|
|
|
|
log.Printf("[INFO] Successfully downloaded %s", imageName)
|
|
}
|
|
}
|
|
|
|
err = deployApp(dockercli, imageName, identifier, env, workflowExecution, action)
|
|
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
|
|
log.Printf("[ERROR] Failed deploying image for the FOURTH time. Aborting if the image doesn't exist")
|
|
if strings.Contains(err.Error(), "exited prematurely") {
|
|
log.Printf("[DEBUG] Shutting down (15)")
|
|
shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true)
|
|
return
|
|
}
|
|
|
|
if strings.Contains(err.Error(), "No such image") {
|
|
//log.Printf("[WARNING] Failed deploying %s from image %s: %s", identifier, image, err)
|
|
log.Printf("[ERROR] Image doesn't exist. Shutting down")
|
|
log.Printf("[DEBUG] Shutting down (16)")
|
|
shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true)
|
|
return
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
//log.Printf("[INFO][%s] Adding visited (3): %s (%s). Actions: %d, Results: %d", workflowExecution.ExecutionId, action.Label, action.ID, len(workflowExecution.Workflow.Actions), len(workflowExecution.Results))
|
|
|
|
visited = append(visited, action.ID)
|
|
executed = append(executed, action.ID)
|
|
|
|
// If children of action.ID are NOT in executed:
|
|
// Remove them from visited.
|
|
//log.Printf("EXECUTED: %#v", executed)
|
|
}
|
|
|
|
//log.Printf(nextAction)
|
|
//log.Printf(startAction, children[startAction])
|
|
|
|
// FIXME - new request here
|
|
// FIXME - clean up stopped (remove) containers with this execution id
|
|
err = shuffle.UpdateExecutionVariables(ctx, workflowExecution.ExecutionId, startAction, children, parents, visited, executed, nextActions, environments, extra)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed to update exec variables for execution %s: %s (2)", workflowExecution.ExecutionId, err)
|
|
}
|
|
|
|
if len(workflowExecution.Results) == len(workflowExecution.Workflow.Actions)+extra {
|
|
shutdownCheck := true
|
|
for _, result := range workflowExecution.Results {
|
|
if result.Status == "EXECUTING" || result.Status == "WAITING" {
|
|
// Cleaning up executing stuff
|
|
shutdownCheck = false
|
|
// USED TO BE CONTAINER REMOVAL
|
|
// FIXME - send POST request to kill the container
|
|
//log.Printf("Should remove (POST request) stopped containers")
|
|
//ret = requests.post("%s%s" % (self.url, stream_path), headers=headers, json=action_result)
|
|
}
|
|
}
|
|
|
|
if shutdownCheck {
|
|
log.Printf("[INFO][%s] BREAKING BECAUSE RESULTS IS SAME LENGTH AS ACTIONS. SHOULD CHECK ALL RESULTS FOR WHETHER THEY'RE DONE", workflowExecution.ExecutionId)
|
|
validated := shuffle.ValidateFinished(ctx, -1, workflowExecution)
|
|
if validated {
|
|
shutdownData, err := json.Marshal(workflowExecution)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed marshalling shutdowndata during set: %s", err)
|
|
}
|
|
|
|
sendResult(workflowExecution, shutdownData)
|
|
}
|
|
|
|
log.Printf("[DEBUG][%s] Shutting down (17)", workflowExecution.ExecutionId)
|
|
if isKubernetes == "true" {
|
|
// log.Printf("workflow execution: %#v", workflowExecution)
|
|
clientset, _, err := shuffle.GetKubernetesClient()
|
|
if err != nil {
|
|
log.Println("[ERROR] Error getting kubernetes client (1):", err)
|
|
os.Exit(1)
|
|
}
|
|
cleanupKubernetesExecution(clientset, workflowExecution, kubernetesNamespace)
|
|
} else {
|
|
shutdown(workflowExecution, "", "", true)
|
|
}
|
|
return
|
|
}
|
|
}
|
|
|
|
time.Sleep(time.Duration(sleepTime) * time.Second)
|
|
return
|
|
}
|
|
|
|
func executionInit(workflowExecution shuffle.WorkflowExecution) error {
|
|
ctx := context.Background()
|
|
|
|
parents := map[string][]string{}
|
|
children := map[string][]string{}
|
|
nextActions := []string{}
|
|
extra := 0
|
|
|
|
startAction := workflowExecution.Start
|
|
//log.Printf("[INFO][%s] STARTACTION: %s", workflowExecution.ExecutionId, startAction)
|
|
if len(startAction) == 0 {
|
|
log.Printf("[INFO][%s] Didn't find execution start action. Setting it to workflow start action.", workflowExecution.ExecutionId)
|
|
startAction = workflowExecution.Workflow.Start
|
|
}
|
|
|
|
// Setting up extra counter
|
|
for _, trigger := range workflowExecution.Workflow.Triggers {
|
|
//log.Printf("[DEBUG] Appname trigger (0): %s", trigger.AppName)
|
|
if trigger.AppName == "User Input" || trigger.AppName == "Shuffle Workflow" {
|
|
extra += 1
|
|
}
|
|
}
|
|
|
|
// Validates RERUN of single actions
|
|
// Identified by:
|
|
// 1. Predefined result from previous exec
|
|
// 2. Only ONE action
|
|
// 3. Every predefined result having result.Action.Category == "rerun"
|
|
/*
|
|
if len(workflowExecution.Workflow.Actions) == 1 && len(workflowExecution.Results) > 0 {
|
|
finished := shuffle.ValidateFinished(ctx, extra, workflowExecution)
|
|
if finished {
|
|
return nil
|
|
}
|
|
}
|
|
*/
|
|
|
|
nextActions = append(nextActions, startAction)
|
|
for _, branch := range workflowExecution.Workflow.Branches {
|
|
// Check what the parent is first. If it's trigger - skip
|
|
sourceFound := false
|
|
destinationFound := false
|
|
for _, action := range workflowExecution.Workflow.Actions {
|
|
if action.ID == branch.SourceID {
|
|
sourceFound = true
|
|
}
|
|
|
|
if action.ID == branch.DestinationID {
|
|
destinationFound = true
|
|
}
|
|
}
|
|
|
|
for _, trigger := range workflowExecution.Workflow.Triggers {
|
|
//log.Printf("Appname trigger (0): %s (%s)", trigger.AppName, trigger.ID)
|
|
if trigger.AppName == "User Input" || trigger.AppName == "Shuffle Workflow" {
|
|
if trigger.ID == branch.SourceID {
|
|
sourceFound = true
|
|
} else if trigger.ID == branch.DestinationID {
|
|
destinationFound = true
|
|
}
|
|
}
|
|
}
|
|
|
|
if sourceFound {
|
|
parents[branch.DestinationID] = append(parents[branch.DestinationID], branch.SourceID)
|
|
} else {
|
|
log.Printf("[DEBUG] Parent ID %s was not found in actions! Skipping parent. (TRIGGER?)", branch.SourceID)
|
|
}
|
|
|
|
if destinationFound {
|
|
children[branch.SourceID] = append(children[branch.SourceID], branch.DestinationID)
|
|
} else {
|
|
log.Printf("[DEBUG] Child ID %s was not found in actions! Skipping child. (TRIGGER?)", branch.SourceID)
|
|
}
|
|
}
|
|
|
|
log.Printf("[INFO][%s] shuffle.Actions: %d + Special shuffle.Triggers: %d", workflowExecution.ExecutionId, len(workflowExecution.Workflow.Actions), extra)
|
|
onpremApps := []string{}
|
|
toExecuteOnprem := []string{}
|
|
for _, action := range workflowExecution.Workflow.Actions {
|
|
if strings.ToLower(action.Environment) != strings.ToLower(environment) {
|
|
continue
|
|
}
|
|
|
|
toExecuteOnprem = append(toExecuteOnprem, action.ID)
|
|
actionName := fmt.Sprintf("%s:%s_%s", baseimagename, action.AppName, action.AppVersion)
|
|
found := false
|
|
for _, app := range onpremApps {
|
|
if actionName == app {
|
|
found = true
|
|
}
|
|
}
|
|
|
|
if !found {
|
|
onpremApps = append(onpremApps, actionName)
|
|
}
|
|
}
|
|
|
|
if len(onpremApps) == 0 {
|
|
//return errors.New(fmt.Sprintf("No apps to handle onprem (%s)", environment))
|
|
log.Printf("[INFO][%s] No apps to handle onprem (%s). Returning 200 OK anyway", workflowExecution.ExecutionId, environment)
|
|
return nil
|
|
}
|
|
|
|
pullOptions := dockerimage.PullOptions{}
|
|
_ = pullOptions
|
|
for _, image := range onpremApps {
|
|
//log.Printf("[INFO] Image: %s", image)
|
|
// Kind of gambling that the image exists.
|
|
if strings.Contains(image, " ") {
|
|
image = strings.ReplaceAll(image, " ", "-")
|
|
}
|
|
|
|
// FIXME: Reimplement for speed later
|
|
// Skip to make it faster
|
|
//reader, err := dockercli.ImagePull(context.Background(), image, pullOptions)
|
|
//if err != nil {
|
|
// log.Printf("Failed getting %s. The app is missing or some other issue", image)
|
|
// shutdown(workflowExecution)
|
|
//}
|
|
|
|
////io.Copy(os.Stdout, reader)
|
|
//_ = reader
|
|
//log.Printf("Successfully downloaded and built %s", image)
|
|
}
|
|
|
|
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)
|
|
}
|
|
}
|
|
|
|
err := shuffle.UpdateExecutionVariables(ctx, workflowExecution.ExecutionId, startAction, children, parents, visited, executed, nextActions, environments, extra)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed to update exec variables for execution %s: %s", workflowExecution.ExecutionId, err)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func handleSubflowPoller(ctx context.Context, workflowExecution shuffle.WorkflowExecution, streamResultUrl, subflowId string) error {
|
|
// FIXME: If MEMCACHE is enabled, check in this order:
|
|
|
|
extra := 0
|
|
for _, trigger := range workflowExecution.Workflow.Triggers {
|
|
if trigger.AppName == "User Input" || trigger.AppName == "Shuffle Workflow" {
|
|
extra += 1
|
|
}
|
|
}
|
|
|
|
key := fmt.Sprintf("%s:%s", workflowExecution.ExecutionId, subflowId)
|
|
cacheKey := fmt.Sprintf("workflowexecution_%s", workflowExecution.ExecutionId)
|
|
usedCache := false
|
|
if cacheData, err := shuffle.GetCache(ctx, cacheKey); err == nil {
|
|
cachedBytes, ok := cacheData.([]uint8)
|
|
if ok {
|
|
cacheWorkflow := shuffle.WorkflowExecution{}
|
|
if jsonErr := json.Unmarshal([]byte(cachedBytes), &cacheWorkflow); jsonErr == nil {
|
|
workflowExecution = cacheWorkflow
|
|
usedCache = true
|
|
log.Printf("[DEBUG][%s] Using cached workflow execution for subflow poll", workflowExecution.ExecutionId)
|
|
|
|
if workflowExecution.Status == "FINISHED" || workflowExecution.Status == "SUCCESS" {
|
|
log.Printf("[INFO][%s] Workflow execution is finished (cache). Exiting worker.", workflowExecution.ExecutionId)
|
|
log.Printf("[DEBUG] Shutting down (20)")
|
|
resetSubflowPollDelay(key)
|
|
if isKubernetes == "true" {
|
|
clientset, _, err := shuffle.GetKubernetesClient()
|
|
if err != nil {
|
|
log.Println("[ERROR] Error getting kubernetes client (2):", err)
|
|
os.Exit(1)
|
|
}
|
|
|
|
cleanupKubernetesExecution(clientset, workflowExecution, kubernetesNamespace)
|
|
} else {
|
|
shutdown(workflowExecution, "", "", true)
|
|
}
|
|
}
|
|
|
|
for _, result := range workflowExecution.Results {
|
|
if result.Action.ID != subflowId {
|
|
continue
|
|
}
|
|
|
|
if result.Status == "SUCCESS" || result.Status == "FINISHED" || result.Status == "FAILURE" || result.Status == "ABORTED" {
|
|
resetSubflowPollDelay(key)
|
|
setWorkflowExecution(ctx, workflowExecution, false)
|
|
return nil
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
if usedCache {
|
|
delay := nextSubflowPollDelay(key)
|
|
attempt := getSubflowPollAttempt(key)
|
|
log.Printf("[DEBUG][%s] Subflow poll backoff attempt %d for %s (cache hit), sleeping %s", workflowExecution.ExecutionId, attempt, subflowId, delay)
|
|
time.Sleep(delay)
|
|
return errors.New("Subflow status not found yet (cache)")
|
|
}
|
|
|
|
if len(data) == 0 {
|
|
log.Printf("[WARNING] Stream result missing execution ID and authorization; injecting them from workflow execution")
|
|
data = fmt.Sprintf(`{"execution_id": "%s", "authorization": "%s"}`, workflowExecution.ExecutionId, workflowExecution.Authorization)
|
|
}
|
|
|
|
req, err := http.NewRequest(
|
|
"POST",
|
|
streamResultUrl,
|
|
bytes.NewBuffer([]byte(data)),
|
|
)
|
|
|
|
client := shuffle.GetExternalClient(streamResultUrl)
|
|
newresp, err := client.Do(req)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed making request (1): %s", err)
|
|
time.Sleep(nextSubflowPollDelay(key))
|
|
return err
|
|
}
|
|
|
|
defer newresp.Body.Close()
|
|
body, err := ioutil.ReadAll(newresp.Body)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed reading body (1): %s", err)
|
|
time.Sleep(nextSubflowPollDelay(key))
|
|
return err
|
|
}
|
|
|
|
if newresp.StatusCode != 200 {
|
|
log.Printf("[ERROR] Bad statuscode: %d, %s", newresp.StatusCode, string(body))
|
|
|
|
if strings.Contains(string(body), "Workflowexecution is already finished") {
|
|
log.Printf("[DEBUG] Shutting down (19)")
|
|
shutdown(workflowExecution, "", "", true)
|
|
}
|
|
|
|
time.Sleep(nextSubflowPollDelay(key))
|
|
return errors.New(fmt.Sprintf("Bad statuscode: %d", newresp.StatusCode))
|
|
}
|
|
|
|
err = json.Unmarshal(body, &workflowExecution)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed workflowExecution unmarshal: %s", err)
|
|
time.Sleep(nextSubflowPollDelay(key))
|
|
return err
|
|
}
|
|
|
|
if workflowExecution.Status == "FINISHED" || workflowExecution.Status == "SUCCESS" {
|
|
log.Printf("[INFO][%s] Workflow execution is finished. Exiting worker.", workflowExecution.ExecutionId)
|
|
log.Printf("[DEBUG] Shutting down (20)")
|
|
resetSubflowPollDelay(key)
|
|
if isKubernetes == "true" {
|
|
// log.Printf("workflow execution: %#v", workflowExecution)
|
|
clientset, _, err := shuffle.GetKubernetesClient()
|
|
if err != nil {
|
|
log.Println("[ERROR] Error getting kubernetes client (2):", err)
|
|
os.Exit(1)
|
|
}
|
|
|
|
cleanupKubernetesExecution(clientset, workflowExecution, kubernetesNamespace)
|
|
} else {
|
|
shutdown(workflowExecution, "", "", true)
|
|
}
|
|
}
|
|
|
|
hasUserinput := false
|
|
foundSubflow := false
|
|
for _, result := range workflowExecution.Results {
|
|
if result.Action.ID != subflowId {
|
|
continue
|
|
}
|
|
|
|
foundSubflow = true
|
|
|
|
if result.Action.AppName == "User Input" {
|
|
hasUserinput = true
|
|
}
|
|
|
|
log.Printf("[DEBUG][%s] Found subflow to handle: %s (%s)", workflowExecution.ExecutionId, result.Action.AppName, result.Status)
|
|
if result.Status == "SUCCESS" || result.Status == "FINISHED" || result.Status == "FAILURE" || result.Status == "ABORTED" {
|
|
// Check for results
|
|
resetSubflowPollDelay(key)
|
|
setWorkflowExecution(ctx, workflowExecution, false)
|
|
return nil
|
|
}
|
|
}
|
|
|
|
if workflowExecution.Status == "WAITING" && !foundSubflow && workflowExecution.ExecutionSource != "default" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "swarm" {
|
|
log.Printf("[INFO][%s] Workflow execution is waiting without subflow progress. Exiting worker, as backend will restart it.", workflowExecution.ExecutionId)
|
|
resetSubflowPollDelay(key)
|
|
shutdown(workflowExecution, "", "", true)
|
|
return nil
|
|
}
|
|
|
|
log.Printf("[INFO][%s] (2) Status: %s, Results: %d, actions: %d. Userinput: %#v", workflowExecution.ExecutionId, workflowExecution.Status, len(workflowExecution.Results), len(workflowExecution.Workflow.Actions)+extra, hasUserinput)
|
|
delay := nextSubflowPollDelay(key)
|
|
attempt := getSubflowPollAttempt(key)
|
|
log.Printf("[DEBUG][%s] Subflow poll backoff attempt %d for %s, sleeping %s", workflowExecution.ExecutionId, attempt, subflowId, delay)
|
|
time.Sleep(delay)
|
|
return errors.New("Subflow status not found yet")
|
|
}
|
|
|
|
func nextSubflowPollDelay(key string) time.Duration {
|
|
baseDelay := 250 * time.Millisecond
|
|
maxDelay := 5 * time.Second
|
|
|
|
current := 0
|
|
if raw, ok := subflowPollBackoff.Load(key); ok {
|
|
if value, ok := raw.(int); ok {
|
|
current = value
|
|
}
|
|
}
|
|
|
|
if current < 8 {
|
|
current += 1
|
|
}
|
|
|
|
subflowPollBackoff.Store(key, current)
|
|
|
|
delay := baseDelay * time.Duration(1<<uint(current-1))
|
|
if delay > maxDelay {
|
|
delay = maxDelay
|
|
}
|
|
|
|
return delay
|
|
}
|
|
|
|
func resetSubflowPollDelay(key string) {
|
|
subflowPollBackoff.Delete(key)
|
|
}
|
|
|
|
func getSubflowPollAttempt(key string) int {
|
|
if raw, ok := subflowPollBackoff.Load(key); ok {
|
|
if value, ok := raw.(int); ok {
|
|
return value
|
|
}
|
|
}
|
|
|
|
return 0
|
|
}
|
|
|
|
func handleDefaultExecutionWrapper(ctx context.Context, workflowExecution shuffle.WorkflowExecution, streamResultUrl string, extra int) error {
|
|
if extra == -1 {
|
|
extra = 0
|
|
for _, trigger := range workflowExecution.Workflow.Triggers {
|
|
if trigger.AppName == "User Input" || trigger.AppName == "Shuffle Workflow" {
|
|
extra += 1
|
|
}
|
|
}
|
|
}
|
|
|
|
req, err := http.NewRequest(
|
|
"POST",
|
|
streamResultUrl,
|
|
bytes.NewBuffer([]byte(data)),
|
|
)
|
|
|
|
newresp, err := topClient.Do(req)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed making request (1): %s", err)
|
|
time.Sleep(time.Duration(sleepTime) * time.Second)
|
|
return err
|
|
}
|
|
|
|
defer newresp.Body.Close()
|
|
body, err := ioutil.ReadAll(newresp.Body)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed reading body (1): %s", err)
|
|
time.Sleep(time.Duration(sleepTime) * time.Second)
|
|
return err
|
|
}
|
|
|
|
if newresp.StatusCode != 200 {
|
|
log.Printf("[ERROR] Bad statuscode: %d, %s", newresp.StatusCode, string(body))
|
|
|
|
if strings.Contains(string(body), "Workflowexecution is already finished") {
|
|
log.Printf("[DEBUG] Shutting down (19)")
|
|
shutdown(workflowExecution, "", "", true)
|
|
}
|
|
|
|
time.Sleep(time.Duration(sleepTime) * time.Second)
|
|
return errors.New(fmt.Sprintf("Bad statuscode: %d", newresp.StatusCode))
|
|
}
|
|
|
|
err = json.Unmarshal(body, &workflowExecution)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed workflowExecution unmarshal: %s", err)
|
|
time.Sleep(time.Duration(sleepTime) * time.Second)
|
|
return err
|
|
}
|
|
|
|
if workflowExecution.Status == "FINISHED" || workflowExecution.Status == "SUCCESS" {
|
|
log.Printf("[INFO][%s] Workflow execution is finished. Exiting worker.", workflowExecution.ExecutionId)
|
|
log.Printf("[DEBUG] Shutting down (20)")
|
|
if isKubernetes == "true" {
|
|
// log.Printf("workflow execution: %#v", workflowExecution)
|
|
clientset, _, err := shuffle.GetKubernetesClient()
|
|
if err != nil {
|
|
log.Println("[ERROR] Error getting kubernetes client (2):", err)
|
|
os.Exit(1)
|
|
}
|
|
cleanupKubernetesExecution(clientset, workflowExecution, kubernetesNamespace)
|
|
} else {
|
|
shutdown(workflowExecution, "", "", true)
|
|
}
|
|
}
|
|
|
|
log.Printf("[INFO][%s] (3) Status: %s, Results: %d, actions: %d", workflowExecution.ExecutionId, workflowExecution.Status, len(workflowExecution.Results), len(workflowExecution.Workflow.Actions)+extra)
|
|
if workflowExecution.Status != "EXECUTING" {
|
|
log.Printf("[WARNING][%s] Exiting as worker execution has status %s!", workflowExecution.ExecutionId, workflowExecution.Status)
|
|
log.Printf("[DEBUG] Shutting down (21)")
|
|
if isKubernetes == "true" {
|
|
// log.Printf("workflow execution: %#v", workflowExecution)
|
|
clientset, _, err := shuffle.GetKubernetesClient()
|
|
if err != nil {
|
|
log.Println("[ERROR] Error getting kubernetes client (3):", err)
|
|
os.Exit(1)
|
|
}
|
|
|
|
cleanupKubernetesExecution(clientset, workflowExecution, kubernetesNamespace)
|
|
} else {
|
|
shutdown(workflowExecution, "", "", true)
|
|
}
|
|
}
|
|
|
|
setWorkflowExecution(ctx, workflowExecution, false)
|
|
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 {
|
|
log.Printf("[INFO] Workflow setup failed for %s: %s", workflowExecution.ExecutionId, err)
|
|
log.Printf("[DEBUG] Shutting down (18)")
|
|
shutdown(workflowExecution, "", "", true)
|
|
}
|
|
|
|
log.Printf("[DEBUG] DEFAULT EXECUTION Startaction: %s", startAction)
|
|
|
|
setWorkflowExecution(ctx, workflowExecution, false)
|
|
|
|
streamResultUrl := fmt.Sprintf("%s/api/v1/streams/results", baseUrl)
|
|
for {
|
|
err = handleDefaultExecutionWrapper(ctx, workflowExecution, streamResultUrl, extra)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed handling default execution: %s", err)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func arrayContains(visited []string, id string) bool {
|
|
found := false
|
|
for _, item := range visited {
|
|
if item == id {
|
|
found = true
|
|
break
|
|
}
|
|
}
|
|
|
|
return found
|
|
}
|
|
|
|
func isTerminalResultStatus(status string) bool {
|
|
return status == "SUCCESS" || status == "FINISHED" || status == "FAILURE" || status == "ABORTED"
|
|
}
|
|
|
|
func hasWaitForResultParameter(params []shuffle.WorkflowAppActionParameter) bool {
|
|
for _, param := range params {
|
|
if param.Name == "check_result" && strings.ToLower(param.Value) == "true" {
|
|
return true
|
|
}
|
|
}
|
|
|
|
return false
|
|
}
|
|
|
|
func getSubflowBarrierActionIDs(workflowExecution shuffle.WorkflowExecution) map[string]struct{} {
|
|
actionIDs := map[string]struct{}{}
|
|
|
|
for _, action := range workflowExecution.Workflow.Actions {
|
|
if action.AppName != "shuffle-subflow" && action.AppName != "shuffle-subflow-v2" && action.AppName != "Shuffle Workflow" {
|
|
continue
|
|
}
|
|
|
|
if !hasWaitForResultParameter(action.Parameters) {
|
|
continue
|
|
}
|
|
|
|
actionIDs[action.ID] = struct{}{}
|
|
}
|
|
|
|
for _, trigger := range workflowExecution.Workflow.Triggers {
|
|
if trigger.AppName != "shuffle-subflow" && trigger.AppName != "shuffle-subflow-v2" && trigger.AppName != "Shuffle Workflow" {
|
|
continue
|
|
}
|
|
|
|
if !hasWaitForResultParameter(trigger.Parameters) {
|
|
continue
|
|
}
|
|
|
|
actionIDs[trigger.ID] = struct{}{}
|
|
}
|
|
|
|
return actionIDs
|
|
}
|
|
|
|
func getSubflowBarrierProgress(workflowExecution shuffle.WorkflowExecution) (int, int, int) {
|
|
barrierActions := getSubflowBarrierActionIDs(workflowExecution)
|
|
expected := len(barrierActions)
|
|
if expected == 0 {
|
|
return 0, 0, 0
|
|
}
|
|
|
|
terminalByAction := map[string]string{}
|
|
for _, result := range workflowExecution.Results {
|
|
if _, ok := barrierActions[result.Action.ID]; !ok {
|
|
continue
|
|
}
|
|
|
|
if !isTerminalResultStatus(result.Status) {
|
|
continue
|
|
}
|
|
|
|
terminalByAction[result.Action.ID] = result.Status
|
|
}
|
|
|
|
completed := len(terminalByAction)
|
|
failed := 0
|
|
for _, status := range terminalByAction {
|
|
if status == "FAILURE" || status == "ABORTED" {
|
|
failed += 1
|
|
}
|
|
}
|
|
|
|
return expected, completed, failed
|
|
}
|
|
|
|
func enforceSubflowBarrier(workflowExecution *shuffle.WorkflowExecution) (int, int, int) {
|
|
if workflowExecution == nil {
|
|
return 0, 0, 0
|
|
}
|
|
|
|
if workflowExecution.Status == "FINISHED" || workflowExecution.Status == "FAILURE" || workflowExecution.Status == "ABORTED" {
|
|
return 0, 0, 0
|
|
}
|
|
|
|
expected, completed, failed := getSubflowBarrierProgress(*workflowExecution)
|
|
if expected == 0 {
|
|
return 0, 0, 0
|
|
}
|
|
|
|
if completed < expected {
|
|
workflowExecution.Status = "WAITING"
|
|
return expected, completed, failed
|
|
}
|
|
|
|
if workflowExecution.Status == "WAITING" {
|
|
if failed > 0 {
|
|
workflowExecution.Status = "FAILURE"
|
|
workflowExecution.Result = fmt.Sprintf("Subflow barrier failed: %d of %d subflows failed or aborted", failed, expected)
|
|
} else {
|
|
workflowExecution.Status = "EXECUTING"
|
|
}
|
|
}
|
|
|
|
return expected, completed, failed
|
|
}
|
|
|
|
func getResult(workflowExecution shuffle.WorkflowExecution, id string) shuffle.ActionResult {
|
|
for _, actionResult := range workflowExecution.Results {
|
|
if actionResult.Action.ID == id {
|
|
return actionResult
|
|
}
|
|
}
|
|
|
|
return shuffle.ActionResult{}
|
|
}
|
|
|
|
func getAction(workflowExecution shuffle.WorkflowExecution, id, environment string) shuffle.Action {
|
|
for _, action := range workflowExecution.Workflow.Actions {
|
|
if action.ID == id {
|
|
return action
|
|
}
|
|
}
|
|
|
|
for _, trigger := range workflowExecution.Workflow.Triggers {
|
|
if trigger.ID == id {
|
|
return shuffle.Action{
|
|
ID: trigger.ID,
|
|
AppName: trigger.AppName,
|
|
Name: trigger.AppName,
|
|
Environment: environment,
|
|
Label: trigger.Label,
|
|
}
|
|
log.Printf("FOUND TRIGGER: %#v!", trigger)
|
|
}
|
|
}
|
|
|
|
return shuffle.Action{}
|
|
}
|
|
|
|
func runSkipAction(client *http.Client, action shuffle.Action, workflowId, workflowExecutionId, authorization string, configuration string) error {
|
|
timeNow := time.Now().Unix()
|
|
result := shuffle.ActionResult{
|
|
Action: action,
|
|
ExecutionId: workflowExecutionId,
|
|
Authorization: authorization,
|
|
Result: configuration,
|
|
StartedAt: timeNow,
|
|
CompletedAt: 0,
|
|
Status: "SUCCESS",
|
|
}
|
|
|
|
resultData, err := json.Marshal(result)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
streamUrl := fmt.Sprintf("%s/api/v1/streams", baseUrl)
|
|
req, err := http.NewRequest(
|
|
"POST",
|
|
streamUrl,
|
|
bytes.NewBuffer([]byte(resultData)),
|
|
)
|
|
|
|
if err != nil {
|
|
log.Printf("[WARNING] Error building skip request (0): %s", err)
|
|
return err
|
|
}
|
|
|
|
newresp, err := topClient.Do(req)
|
|
if err != nil {
|
|
log.Printf("[WARNING] Error running skip request (0): %s", err)
|
|
return err
|
|
}
|
|
|
|
defer newresp.Body.Close()
|
|
body, err := ioutil.ReadAll(newresp.Body)
|
|
if err != nil {
|
|
log.Printf("[WARNING] Failed reading body when skipping (0): %s", err)
|
|
return err
|
|
}
|
|
|
|
log.Printf("[INFO] Skip Action Body: %s", string(body))
|
|
return nil
|
|
}
|
|
|
|
func runTestExecution(client *http.Client, workflowId, apikey string) (string, string) {
|
|
executeUrl := fmt.Sprintf("%s/api/v1/workflows/%s/execute", baseUrl, workflowId)
|
|
req, err := http.NewRequest(
|
|
"GET",
|
|
executeUrl,
|
|
nil,
|
|
)
|
|
|
|
if err != nil {
|
|
log.Printf("Error building test request: %s", err)
|
|
return "", ""
|
|
}
|
|
|
|
req.Header.Add("Authorization", fmt.Sprintf("Bearer %s", apikey))
|
|
newresp, err := topClient.Do(req)
|
|
if err != nil {
|
|
log.Printf("[WARNING] Error running test request (3): %s", err)
|
|
return "", ""
|
|
}
|
|
|
|
defer newresp.Body.Close()
|
|
body, err := ioutil.ReadAll(newresp.Body)
|
|
if err != nil {
|
|
log.Printf("[WARNING] Failed reading body: %s", err)
|
|
return "", ""
|
|
}
|
|
|
|
log.Printf("[INFO] Test Body: %s", string(body))
|
|
var workflowExecution shuffle.WorkflowExecution
|
|
err = json.Unmarshal(body, &workflowExecution)
|
|
if err != nil {
|
|
log.Printf("Failed workflowExecution unmarshal: %s", err)
|
|
return "", ""
|
|
}
|
|
|
|
return workflowExecution.Authorization, workflowExecution.ExecutionId
|
|
}
|
|
|
|
func isRunningInCluster() bool {
|
|
_, existsHost := os.LookupEnv("KUBERNETES_SERVICE_HOST")
|
|
_, existsPort := os.LookupEnv("KUBERNETES_SERVICE_PORT")
|
|
return existsHost && existsPort
|
|
}
|
|
|
|
func buildEnvVars(envMap map[string]string) []corev1.EnvVar {
|
|
var envVars []corev1.EnvVar
|
|
for key, value := range envMap {
|
|
envVars = append(envVars, corev1.EnvVar{Name: key, Value: value})
|
|
}
|
|
return envVars
|
|
}
|
|
|
|
func buildResourcesFromEnv() corev1.ResourceRequirements {
|
|
requests := corev1.ResourceList{}
|
|
limits := corev1.ResourceList{}
|
|
|
|
type item struct {
|
|
env string
|
|
resourceName corev1.ResourceName
|
|
resourceList corev1.ResourceList
|
|
}
|
|
|
|
items := []item{
|
|
// kubernetes requests
|
|
{env: "SHUFFLE_APP_CPU_REQUEST", resourceName: corev1.ResourceCPU, resourceList: requests},
|
|
{env: "SHUFFLE_APP_MEMORY_REQUEST", resourceName: corev1.ResourceMemory, resourceList: requests},
|
|
{env: "SHUFFLE_APP_EPHEMERAL_STORAGE_REQUEST", resourceName: corev1.ResourceEphemeralStorage, resourceList: requests},
|
|
// kubernetes limits
|
|
{env: "SHUFFLE_APP_CPU_LIMIT", resourceName: corev1.ResourceCPU, resourceList: limits},
|
|
{env: "SHUFFLE_APP_MEMORY_LIMIT", resourceName: corev1.ResourceMemory, resourceList: limits},
|
|
{env: "SHUFFLE_APP_EPHEMERAL_STORAGE_LIMIT", resourceName: corev1.ResourceEphemeralStorage, resourceList: limits},
|
|
}
|
|
|
|
for _, it := range items {
|
|
if value := strings.TrimSpace(os.Getenv(it.env)); value != "" {
|
|
if quantity, err := resource.ParseQuantity(value); err == nil {
|
|
it.resourceList[it.resourceName] = quantity
|
|
} else {
|
|
log.Printf("[WARNING] Cannot parse %s=%q as resource quantity: %v", it.env, value, err)
|
|
}
|
|
}
|
|
}
|
|
|
|
rr := corev1.ResourceRequirements{}
|
|
if len(requests) > 0 {
|
|
rr.Requests = requests
|
|
}
|
|
if len(limits) > 0 {
|
|
rr.Limits = limits
|
|
}
|
|
|
|
return rr
|
|
}
|
|
|
|
func getWorkerBackendExecution(auth string, executionId string) (*shuffle.WorkflowExecution, error) {
|
|
backendUrl := os.Getenv("BASE_URL")
|
|
if len(backendUrl) == 0 {
|
|
backendUrl = "http://shuffle-backend:5001"
|
|
}
|
|
|
|
var workflowExecution *shuffle.WorkflowExecution
|
|
|
|
streamResultUrl := fmt.Sprintf("%s/api/v1/streams/results", backendUrl)
|
|
topClient := shuffle.GetExternalClient(backendUrl)
|
|
requestData := shuffle.ActionResult{
|
|
Authorization: auth,
|
|
ExecutionId: executionId,
|
|
}
|
|
|
|
data, err := json.Marshal(requestData)
|
|
if err != nil {
|
|
return workflowExecution, err
|
|
}
|
|
|
|
req, err := http.NewRequest(
|
|
"POST",
|
|
streamResultUrl,
|
|
bytes.NewBuffer([]byte(data)),
|
|
)
|
|
|
|
newresp, err := topClient.Do(req)
|
|
if err != nil {
|
|
return workflowExecution, err
|
|
}
|
|
|
|
defer newresp.Body.Close()
|
|
if newresp.StatusCode != 200 {
|
|
return workflowExecution, errors.New(fmt.Sprintf("Got bad status code from backend %d", newresp.StatusCode))
|
|
}
|
|
|
|
body, err := ioutil.ReadAll(newresp.Body)
|
|
if err != nil {
|
|
return workflowExecution, err
|
|
}
|
|
|
|
err = json.Unmarshal(body, &workflowExecution)
|
|
if err != nil {
|
|
return workflowExecution, err
|
|
}
|
|
|
|
if debug {
|
|
log.Printf("[INFO] Here is the result we got back from backend: %s", workflowExecution.Results)
|
|
}
|
|
|
|
//setWorkflowExecution(context.Background(), *workflowExecution, false)
|
|
return workflowExecution, nil
|
|
}
|
|
|
|
func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
|
|
if request.Body == nil {
|
|
resp.WriteHeader(http.StatusBadRequest)
|
|
return
|
|
}
|
|
|
|
defer request.Body.Close()
|
|
body, err := ioutil.ReadAll(request.Body)
|
|
if err != nil {
|
|
log.Printf("[WARNING] (3) Failed reading body for workflowqueue")
|
|
resp.WriteHeader(401)
|
|
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "%s"}`, err)))
|
|
return
|
|
}
|
|
|
|
var actionResult shuffle.ActionResult
|
|
err = json.Unmarshal(body, &actionResult)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed shuffle.ActionResult unmarshaling (2): %s", err)
|
|
//resp.WriteHeader(401)
|
|
//resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "%s"}`, err)))
|
|
//return
|
|
}
|
|
|
|
if len(actionResult.ExecutionId) == 0 {
|
|
log.Printf("[ERROR] No workflow execution id in action result. Data: %s", string(body))
|
|
resp.WriteHeader(400)
|
|
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "No workflow execution id in action result"}`)))
|
|
return
|
|
}
|
|
|
|
// 1. Get the shuffle.WorkflowExecution(ExecutionId) from the database
|
|
// 2. if shuffle.ActionResult.Authentication != shuffle.WorkflowExecution.Authentication -> exit
|
|
// 3. Add to and update actionResult in workflowExecution
|
|
// 4. Push to db
|
|
// IF FAIL: Set executionstatus: abort or cancel
|
|
ctx := context.Background()
|
|
if actionResult.ExecutionId == "TBD" {
|
|
return
|
|
}
|
|
|
|
workflowExecution, err := shuffle.GetWorkflowExecution(ctx, actionResult.ExecutionId)
|
|
if err != nil {
|
|
log.Printf("[WARNING][%s] Failed to find execution in cache requesting backend (1): %s", actionResult.ExecutionId, err)
|
|
workflowExecution, err = getWorkerBackendExecution(actionResult.Authorization, actionResult.ExecutionId)
|
|
if err != nil {
|
|
log.Printf("[ERROR][%s] Failed getting execution (workflowqueue) %s: %s", actionResult.ExecutionId, actionResult.ExecutionId, err)
|
|
resp.WriteHeader(500)
|
|
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed getting execution ID %s because it doesn't exist locally."}`, actionResult.ExecutionId)))
|
|
return
|
|
}
|
|
}
|
|
|
|
if workflowExecution.Authorization != actionResult.Authorization {
|
|
log.Printf("[ERROR][%s] Bad authorization key when updating node (workflowQueue). Want: %s, Have: %s", actionResult.ExecutionId, workflowExecution.Authorization, actionResult.Authorization)
|
|
resp.WriteHeader(403)
|
|
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Bad authorization key"}`)))
|
|
return
|
|
}
|
|
|
|
if workflowExecution.Status == "FINISHED" {
|
|
log.Printf("[DEBUG][%s] Workflowexecution is already FINISHED. No further action can be taken", workflowExecution.ExecutionId)
|
|
resp.WriteHeader(200)
|
|
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Workflowexecution is already finished because it has status %s. Lastnode: %s"}`, workflowExecution.Status, workflowExecution.LastNode)))
|
|
return
|
|
}
|
|
|
|
if workflowExecution.Status == "ABORTED" || workflowExecution.Status == "FAILURE" {
|
|
log.Printf("[WARNING][%s] Workflowexecution already has status %s. No further action can be taken", workflowExecution.ExecutionId, workflowExecution.Status)
|
|
resp.WriteHeader(200)
|
|
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Workflowexecution is aborted because of %s with result %s and status %s"}`, workflowExecution.LastNode, workflowExecution.Result, workflowExecution.Status)))
|
|
return
|
|
}
|
|
|
|
retries := 0
|
|
retry, retriesok := request.URL.Query()["retries"]
|
|
if retriesok && len(retry) > 0 {
|
|
val, err := strconv.Atoi(retry[0])
|
|
if err == nil {
|
|
retries = val
|
|
}
|
|
}
|
|
|
|
// Not doing environment as we don't want to hook a worker to specific env. It should just not handle cloud actions
|
|
// limiting a worker to an env will not allow us to run multiple orborus in the same server?
|
|
if strings.EqualFold(actionResult.Action.Environment, "cloud") {
|
|
log.Printf("[WARNING] Got an action for %s environment forwarding it to the backend", actionResult.Action.Environment)
|
|
|
|
streamUrl := fmt.Sprintf("%s/api/v1/streams", baseUrl)
|
|
req, err := http.NewRequest(
|
|
"POST",
|
|
streamUrl,
|
|
bytes.NewBuffer([]byte(body)),
|
|
)
|
|
|
|
if err != nil {
|
|
log.Printf("[ERROR] Error building subflow (%s) request: %s", workflowExecution.ExecutionId, err)
|
|
return
|
|
}
|
|
|
|
newresp, err := topClient.Do(req)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Error running subflow (%s) request: %s", workflowExecution.ExecutionId, err)
|
|
return
|
|
}
|
|
|
|
defer newresp.Body.Close()
|
|
if newresp.StatusCode != 200 {
|
|
body, err := ioutil.ReadAll(newresp.Body)
|
|
if err != nil {
|
|
log.Printf("[INFO][%s] Failed reading body after subflow request: %s", workflowExecution.ExecutionId, err)
|
|
return
|
|
} else {
|
|
log.Printf("[ERROR][%s] Failed forwarding subflow request of length %d\n: %s", workflowExecution.ExecutionId, len(actionResult.Result), string(body))
|
|
}
|
|
}
|
|
|
|
return
|
|
}
|
|
|
|
log.Printf("[DEBUG][%s] Action: Received, Label: '%s', Action: '%s', Status: %s, Run status: %s, Extra=Retry:%d", workflowExecution.ExecutionId, actionResult.Action.Label, actionResult.Action.AppName, actionResult.Status, workflowExecution.Status, retries)
|
|
|
|
// results = append(results, actionResult)
|
|
// log.Printf("[INFO][%s] Time to execute %s (%s) with app %s:%s, function %s, env %s with %d parameters.", workflowExecution.ExecutionId, action.ID, action.Label, action.AppName, action.AppVersion, action.Name, action.Environment, len(action.Parameters))
|
|
// log.Printf("[DEBUG][%s] In workflowQueue with transaction", workflowExecution.ExecutionId)
|
|
runWorkflowExecutionTransaction(ctx, 0, workflowExecution.ExecutionId, actionResult, resp)
|
|
}
|
|
|
|
// 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("[WARNING][%s] Failed to find execution in cache requesting backend (2): %s", actionResult.ExecutionId, err)
|
|
workflowExecution, err = getWorkerBackendExecution(actionResult.Authorization, actionResult.ExecutionId)
|
|
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
|
|
|
|
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 {
|
|
resp.WriteHeader(200)
|
|
resp.Write([]byte(fmt.Sprintf(`{"success": true, "reason": "Execution is not executing, but %s"}`, workflowExecution.Status)))
|
|
}
|
|
|
|
log.Printf("[DEBUG][%s] Shutting down (35)", workflowExecution.ExecutionId)
|
|
|
|
// Force sending result
|
|
shutdownData, err := json.Marshal(workflowExecution)
|
|
if err != nil {
|
|
log.Printf("[ERROR][%s] Failed marshalling execution (35): %s", workflowExecution.ExecutionId, err)
|
|
}
|
|
|
|
sendResult(*workflowExecution, shutdownData)
|
|
shutdown(*workflowExecution, "", "", false)
|
|
return
|
|
}
|
|
|
|
/*** STARTREMOVE ***/
|
|
if workflowExecution.Status == "WAITING" && (os.Getenv("SHUFFLE_SWARM_CONFIG") == "run" || os.Getenv("SHUFFLE_SWARM_CONFIG") == "swarm") {
|
|
log.Printf("[INFO][%s] Workflow execution is waiting while in swarm. Sending info to backend to ensure execution stops.", workflowExecution.ExecutionId)
|
|
|
|
shutdownData, err := json.Marshal(workflowExecution)
|
|
if err != nil {
|
|
log.Printf("[ERROR][%s] Failed marshalling execution (36) - not sending backend WAITING: %s", workflowExecution.ExecutionId, err)
|
|
} else {
|
|
sendResult(*workflowExecution, shutdownData)
|
|
shutdown(*workflowExecution, "", "", false)
|
|
}
|
|
}
|
|
/*** ENDREMOVE ***/
|
|
} 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
|
|
}
|
|
|
|
log.Printf("[DEBUG] Rerunning transaction? %s", err)
|
|
if strings.Contains(fmt.Sprintf("%s", err), "Rerun this transaction") {
|
|
workflowExecution, err := shuffle.GetWorkflowExecution(ctx, workflowExecutionId)
|
|
if err != nil {
|
|
log.Printf("[WARNING][%s] Failed to find execution in cache requesting backend (3): %s", actionResult.ExecutionId, err)
|
|
workflowExecution, err = getWorkerBackendExecution(actionResult.Authorization, actionResult.ExecutionId)
|
|
if err != nil {
|
|
log.Printf("[ERROR][%s] Failed getting execution cache (2): %s", workflowExecution.ExecutionId, err)
|
|
resp.WriteHeader(400)
|
|
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed getting execution (2)"}`)))
|
|
return
|
|
}
|
|
}
|
|
|
|
resultLength = len(workflowExecution.Results)
|
|
setExecution = true
|
|
|
|
workflowExecution, dbSave, err = shuffle.ParsedExecutionResult(ctx, *workflowExecution, actionResult, false, 0)
|
|
if err != nil {
|
|
log.Printf("[ERROR][%s] Failed execution of parsedexecution (2): %s", workflowExecution.ExecutionId, err)
|
|
resp.WriteHeader(401)
|
|
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed getting execution (2)"}`)))
|
|
return
|
|
} else {
|
|
log.Printf("[DEBUG][%s] Successfully got ParsedExecution with %d results!", workflowExecution.ExecutionId, len(workflowExecution.Results))
|
|
}
|
|
} else {
|
|
log.Printf("[ERROR][%s] Failed execution of parsedexecution: %s", workflowExecution.ExecutionId, err)
|
|
resp.WriteHeader(401)
|
|
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
|
|
|
|
//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 {
|
|
//parsedValue := value.(*shuffle.WorkflowExecution)
|
|
|
|
parsedValue := &shuffle.WorkflowExecution{}
|
|
cacheData := []byte(cache.([]uint8))
|
|
err = json.Unmarshal(cacheData, &workflowExecution)
|
|
if err != nil {
|
|
log.Printf("[ERROR][%s] Failed unmarshalling workflowexecution: %s", workflowExecution.ExecutionId, err)
|
|
}
|
|
|
|
if len(parsedValue.Results) > 0 && len(parsedValue.Results) != resultLength {
|
|
setExecution = false
|
|
if attempts > 5 {
|
|
}
|
|
|
|
attempts += 1
|
|
log.Printf("[DEBUG][%s] Rerunning transaction as results has changed. %d vs %d", workflowExecution.ExecutionId, len(parsedValue.Results), resultLength)
|
|
/*
|
|
if len(workflowExecution.Results) <= len(workflowExecution.Workflow.Actions) {
|
|
log.Printf("[DEBUG][%s] Rerunning transaction as results has changed. %d vs %d", workflowExecution.ExecutionId, len(workflowExecution.Results), len(workflowExecution.Workflow.Actions))
|
|
runWorkflowExecutionTransaction(ctx, attempts, workflowExecutionId, actionResult, resp)
|
|
return
|
|
}
|
|
*/
|
|
}
|
|
}
|
|
|
|
if setExecution || workflowExecution.Status == "FINISHED" || workflowExecution.Status == "ABORTED" || workflowExecution.Status == "FAILURE" {
|
|
if debug {
|
|
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)
|
|
if err != nil {
|
|
log.Printf("[ERROR][%s] Failed setting execution: %s", workflowExecution.ExecutionId, err)
|
|
|
|
resp.WriteHeader(400)
|
|
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed setting workflowexecution actionresult: %s"}`, err)))
|
|
return
|
|
}
|
|
|
|
/*** STARTREMOVE ***/
|
|
if os.Getenv("SHUFFLE_SWARM_CONFIG") == "run" || os.Getenv("SHUFFLE_SWARM_CONFIG") == "swarm" {
|
|
finished := shuffle.ValidateFinished(ctx, -1, *workflowExecution)
|
|
if !finished {
|
|
if debug {
|
|
log.Printf("[DEBUG][%s] Handling next node since it's not finished!", workflowExecution.ExecutionId)
|
|
}
|
|
|
|
handleExecutionResult(*workflowExecution)
|
|
} else {
|
|
shutdownData, err := json.Marshal(workflowExecution)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed marshalling shutdowndata during set: %s", err)
|
|
}
|
|
|
|
sendResult(*workflowExecution, shutdownData)
|
|
}
|
|
}
|
|
/*** ENDREMOVE ***/
|
|
} 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)
|
|
//}
|
|
|
|
resp.WriteHeader(200)
|
|
resp.Write([]byte(fmt.Sprintf(`{"success": true}`)))
|
|
}
|
|
|
|
func sendSelfRequest(actionResult shuffle.ActionResult) {
|
|
|
|
/*** STARTREMOVE ***/
|
|
if os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "swarm" {
|
|
log.Printf("[INFO][%s] Not sending self request info since source is default (not swarm)", actionResult.ExecutionId)
|
|
return
|
|
}
|
|
/*** ENDREMOVE ***/
|
|
|
|
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" {
|
|
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)
|
|
|
|
// 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(
|
|
"POST",
|
|
streamUrl,
|
|
bytes.NewBuffer([]byte(data)),
|
|
)
|
|
|
|
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 {
|
|
log.Printf("[ERROR][%s] Error running finishing request (2): %s", actionResult.ExecutionId, err)
|
|
return
|
|
}
|
|
|
|
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 {
|
|
log.Printf("[ERROR][%s] Failed reading body: %s", actionResult.ExecutionId, err)
|
|
} else {
|
|
log.Printf("[DEBUG][%s] Sent update to backend - 2: %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" {
|
|
//log.Printf("[INFO][%s] Not sending backend info since source is default (not swarm)", workflowExecution.ExecutionId)
|
|
//return
|
|
} else {
|
|
}
|
|
|
|
// 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 {
|
|
log.Printf("[DEBUG][%s] Removing old execution from finishedExecutions: %s", workflowExecution.ExecutionId, finishedExecutions[0])
|
|
finishedExecutions = finishedExecutions[99:]
|
|
}
|
|
|
|
finishedExecutions = append(finishedExecutions, workflowExecution.ExecutionId)
|
|
*/
|
|
|
|
streamUrl := fmt.Sprintf("%s/api/v1/streams", baseUrl)
|
|
req, err := http.NewRequest(
|
|
"POST",
|
|
streamUrl,
|
|
bytes.NewBuffer([]byte(data)),
|
|
)
|
|
|
|
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 {
|
|
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 {
|
|
body, err := ioutil.ReadAll(newresp.Body)
|
|
//log.Printf("[INFO] BACKEND STATUS: %d", newresp.StatusCode)
|
|
if err != nil {
|
|
log.Printf("[ERROR][%s] Failed reading body: %s", workflowExecution.ExecutionId, err)
|
|
} else {
|
|
log.Printf("[DEBUG][%s] Sent request to backend: %s", workflowExecution.ExecutionId, string(body))
|
|
}
|
|
}
|
|
}
|
|
|
|
func validateFinished(workflowExecution shuffle.WorkflowExecution) bool {
|
|
ctx := context.Background()
|
|
|
|
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 {
|
|
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)
|
|
|
|
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" {
|
|
for _, result := range workflowExecution.Results {
|
|
if result.Status == "EXECUTING" || result.Status == "WAITING" {
|
|
log.Printf("[WARNING] NOT returning full result, as a result may be unfinished: %s (%s) - %s", result.Action.Label, result.Action.ID, result.Status)
|
|
return false
|
|
}
|
|
}
|
|
}
|
|
|
|
/*** STARTREMOVE ***/
|
|
if os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "swarm" {
|
|
requestsSent += 1
|
|
}
|
|
/*** ENDREMOVE ***/
|
|
|
|
log.Printf("[DEBUG][%s] Should send full result to %s", workflowExecution.ExecutionId, baseUrl)
|
|
|
|
//data = fmt.Sprintf(`{"execution_id": "%s", "authorization": "%s"}`, executionId, authorization)
|
|
shutdownData, err := json.Marshal(workflowExecution)
|
|
if err != nil {
|
|
log.Printf("[ERROR][%s] Shutting down (32): Failed to unmarshal data for backend: %s", workflowExecution.ExecutionId, err)
|
|
shutdown(workflowExecution, "", "", true)
|
|
}
|
|
|
|
cacheKey := fmt.Sprintf("workflowexecution_%s", workflowExecution.ExecutionId)
|
|
if len(workflowExecution.Authorization) > 0 {
|
|
err = shuffle.SetCache(ctx, cacheKey, shutdownData, 31)
|
|
if err != nil {
|
|
log.Printf("[ERROR][%s] Failed adding to cache during ValidateFinished", workflowExecution)
|
|
}
|
|
}
|
|
|
|
shuffle.RunCacheCleanup(ctx, workflowExecution)
|
|
sendResult(workflowExecution, shutdownData)
|
|
return true
|
|
}
|
|
|
|
return false
|
|
}
|
|
|
|
func handleGetStreamResults(resp http.ResponseWriter, request *http.Request) {
|
|
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 {
|
|
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 {
|
|
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 {
|
|
log.Printf("[WARNING][%s] Failed to find execution in cache requesting backend (4): %s", actionResult.ExecutionId, err)
|
|
workflowExecution, err = getWorkerBackendExecution(actionResult.Authorization, actionResult.ExecutionId)
|
|
if err != nil {
|
|
log.Printf("[ERROR] 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 {
|
|
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 {
|
|
resp.WriteHeader(500)
|
|
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed unpacking workflow execution"}`)))
|
|
return
|
|
}
|
|
|
|
resp.WriteHeader(200)
|
|
resp.Write(newjson)
|
|
|
|
}
|
|
|
|
// GetLocalIP returns the non loopback local IP of the host
|
|
func getLocalIP() string {
|
|
|
|
/*** STARTREMOVE ***/
|
|
if os.Getenv("IS_KUBERNETES") == "true" {
|
|
return "shuffle-workers"
|
|
}
|
|
|
|
if os.Getenv("SHUFFLE_SWARM_CONFIG") == "run" || os.Getenv("SHUFFLE_SWARM_CONFIG") == "swarm" {
|
|
name, err := os.Hostname()
|
|
if err != nil {
|
|
log.Printf("[ERROR] Couldn't find hostname of worker: %s", err)
|
|
os.Exit(3)
|
|
}
|
|
|
|
log.Printf("[DEBUG] Found hostname %s since worker is running with \"run\" command", name)
|
|
return name
|
|
|
|
/**
|
|
Everything below was a test to see if we needed to match directly to a network interface. May require docker network API.
|
|
**/
|
|
|
|
log.Printf("[DEBUG] Looking for IP for the external docker-network %s", swarmNetworkName)
|
|
// Different process to ensure we find the right IP.
|
|
// Necessary due to Ingress being added to docker ser
|
|
ifaces, err := net.Interfaces()
|
|
if err != nil {
|
|
log.Printf("[ERROR] FATAL: networks the container is listening in %s: %s", swarmNetworkName, err)
|
|
os.Exit(3)
|
|
}
|
|
|
|
foundIP := ""
|
|
for _, i := range ifaces {
|
|
log.Printf("NETWORK: %s", i.Name)
|
|
//If i.Name != swarmNetworkName {
|
|
// continue
|
|
//}
|
|
|
|
addrs, err := i.Addrs()
|
|
if err != nil {
|
|
log.Printf("[ERROR] FATAL: Failed getting address for listener in network %s: %s", swarmNetworkName, err)
|
|
continue
|
|
}
|
|
|
|
for _, addr := range addrs {
|
|
var ip net.IP
|
|
switch v := addr.(type) {
|
|
case *net.IPNet:
|
|
ip = v.IP
|
|
case *net.IPAddr:
|
|
ip = v.IP
|
|
}
|
|
|
|
log.Printf("%s: IP: %#v", i.Name, ip)
|
|
|
|
// FIXME: Allow for IPv6 too!
|
|
//if strings.Count(ip.String(), ".") == 3 {
|
|
// foundIP = ip.String()
|
|
// break
|
|
//}
|
|
// process IP address
|
|
}
|
|
}
|
|
|
|
if len(foundIP) == 0 {
|
|
log.Printf("[ERROR] FATAL: No valid IP found for network %s. Defaulting to base IP", swarmNetworkName)
|
|
} else {
|
|
return foundIP
|
|
}
|
|
}
|
|
/*** ENDREMOVE ***/
|
|
|
|
addrs, err := net.InterfaceAddrs()
|
|
if err != nil {
|
|
return ""
|
|
}
|
|
|
|
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 ""
|
|
}
|
|
|
|
func getAvailablePort() (net.Listener, error) {
|
|
listener, err := net.Listen("tcp", ":0")
|
|
if err != nil {
|
|
log.Printf("[WARNING] Failed to assign port by default. Defaulting to 5001")
|
|
return nil, err
|
|
}
|
|
|
|
return listener, nil
|
|
//return fmt.Sprintf(":%d", port)
|
|
}
|
|
|
|
func webserverSetup(workflowExecution shuffle.WorkflowExecution) net.Listener {
|
|
hostname = getLocalIP()
|
|
if isKubernetes == "true" {
|
|
os.Setenv("WORKER_HOSTNAME", "shuffle-workers")
|
|
} else {
|
|
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 {
|
|
log.Printf("[ERROR] Failed to create init listener: %s", err)
|
|
return listener
|
|
}
|
|
|
|
log.Printf("[DEBUG] OLD HOSTNAME: %s", appCallbackUrl)
|
|
|
|
/*** STARTREMOVE ***/
|
|
if os.Getenv("SHUFFLE_SWARM_CONFIG") == "run" || os.Getenv("SHUFFLE_SWARM_CONFIG") == "swarm" {
|
|
log.Printf("[DEBUG] Starting webserver (1) on port %d with hostname: %s", baseport, hostname)
|
|
|
|
os.Setenv("WORKER_PORT", fmt.Sprintf("%d", baseport))
|
|
workerServerUrl := os.Getenv("SHUFFLE_WORKER_SERVER_URL")
|
|
if len(workerServerUrl) > 0 {
|
|
appCallbackUrl = workerServerUrl
|
|
log.Printf("[DEBUG] Using SHUFFLE_WORKER_SERVER_URL for callback: %s", appCallbackUrl)
|
|
} else if os.Getenv("IS_KUBERNETES") == "true" {
|
|
appCallbackUrl = fmt.Sprintf("http://%s:%d", "shuffle-workers", baseport)
|
|
log.Printf("[DEBUG] NEW WORKER APP: %s", appCallbackUrl)
|
|
hostname = "shuffle-workers"
|
|
} else {
|
|
appCallbackUrl = fmt.Sprintf("http://%s:%d", hostname, baseport)
|
|
}
|
|
|
|
listener, err = net.Listen("tcp", fmt.Sprintf(":%d", baseport))
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed to assign port to %d: %s", baseport, err)
|
|
return nil
|
|
}
|
|
|
|
return listener
|
|
}
|
|
/*** ENDREMOVE ***/
|
|
|
|
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 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 {
|
|
//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 {
|
|
maxNodes, err := strconv.ParseInt(maxNodesString, 10, 64)
|
|
if err != nil {
|
|
return nodeCount, err
|
|
}
|
|
|
|
if nodeCount > maxNodes {
|
|
nodeCount = maxNodes
|
|
}
|
|
}
|
|
|
|
return nodeCount, nil
|
|
|
|
/*
|
|
containers, err := dockercli.ContainerList(ctx, types.ContainerListOptions{
|
|
All: true,
|
|
})
|
|
*/
|
|
}
|
|
|
|
/*** STARTREMOVE ***/
|
|
func deploySwarmService(dockercli *dockerclient.Client, name, image string, deployport int, inputReplicas int64, retry bool) error {
|
|
log.Printf("[DEBUG] Deploying service for %s to swarm on port %d", name, deployport)
|
|
//containerName := fmt.Sprintf("shuffle-worker-%s", parsedUuid)
|
|
|
|
// Check if the image exists or not - just in case
|
|
_, _, err := dockercli.ImageInspectWithRaw(context.Background(), image)
|
|
if err != nil {
|
|
log.Printf("[INFO] Image %s not found locally. Pulling from registry...", image)
|
|
|
|
localRegistry := os.Getenv("REGISTRY_URL")
|
|
if !strings.HasPrefix(image, localRegistry) && len(localRegistry) > 0 {
|
|
image = fmt.Sprintf("%s/%s", localRegistry, image)
|
|
log.Printf("[DEBUG] Changed image to %s", image)
|
|
}
|
|
|
|
_, err := dockercli.ImagePull(
|
|
context.Background(),
|
|
image,
|
|
dockerimage.PullOptions{},
|
|
)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed pulling image %s: %s", image, err)
|
|
return err
|
|
}
|
|
}
|
|
|
|
if len(baseimagename) == 0 || baseimagename == "/" {
|
|
baseimagename = "frikky/shuffle"
|
|
//var baseimagename = "frikky/shuffle"
|
|
//var registryName = "registry.hub.docker.com"
|
|
}
|
|
|
|
//image := fmt.Sprintf("%s:%s", baseimagename, name)
|
|
networkName := "shuffle-executions"
|
|
if len(swarmNetworkName) > 0 {
|
|
networkName = swarmNetworkName
|
|
}
|
|
|
|
// Apps used a lot should have 2 replicas (default)
|
|
// New default to 3 (as the chance of queues piling up is lower)
|
|
replicas := uint64(3)
|
|
|
|
// Sent from Orborus
|
|
// Should be equal to
|
|
scaleReplicas := os.Getenv("SHUFFLE_APP_REPLICAS")
|
|
if len(scaleReplicas) > 0 {
|
|
tmpInt, err := strconv.Atoi(scaleReplicas)
|
|
if err != nil {
|
|
log.Printf("[ERROR] %s is not a valid number for replication", scaleReplicas)
|
|
} else {
|
|
replicas = uint64(tmpInt)
|
|
}
|
|
|
|
log.Printf("[DEBUG] SHUFFLE_APP_REPLICAS set to value %#v. Trying to overwrite default (%d/node)", scaleReplicas, replicas)
|
|
}
|
|
|
|
// Max scale as well
|
|
nodeCount := uint64(1)
|
|
if inputReplicas > 0 && inputReplicas < 100 {
|
|
if replicas != uint64(inputReplicas) {
|
|
log.Printf("[DEBUG] Overwriting replicas to %d/node as inputReplicas is set to %d", inputReplicas, inputReplicas)
|
|
}
|
|
|
|
replicas = uint64(inputReplicas)
|
|
} else {
|
|
cnt, err := findActiveSwarmNodes(dockercli)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Unable to find active swarm nodes: %s", err)
|
|
}
|
|
|
|
if cnt > 0 {
|
|
nodeCount = uint64(cnt)
|
|
}
|
|
|
|
// FIXME: From September 2025 - This is set back to 1, as this doesn't really reflect how scale works at all. It is just confusing, and makes number larger/smaller "arbitrarily" instead of using default docker scale
|
|
nodeCount = 1
|
|
}
|
|
|
|
replicatedJobs := uint64(replicas * nodeCount)
|
|
log.Printf("[DEBUG] Deploying app with name %s with image %s", name, image)
|
|
|
|
containerName := fmt.Sprintf(strings.Replace(name, ".", "-", -1))
|
|
serviceSpec := swarm.ServiceSpec{
|
|
Annotations: swarm.Annotations{
|
|
Name: containerName,
|
|
Labels: map[string]string{},
|
|
},
|
|
Mode: swarm.ServiceMode{
|
|
Replicated: &swarm.ReplicatedService{
|
|
// Max replicas total (?)
|
|
Replicas: &replicatedJobs,
|
|
},
|
|
},
|
|
Networks: []swarm.NetworkAttachmentConfig{
|
|
swarm.NetworkAttachmentConfig{
|
|
Target: networkName,
|
|
},
|
|
},
|
|
EndpointSpec: &swarm.EndpointSpec{
|
|
Ports: []swarm.PortConfig{
|
|
swarm.PortConfig{
|
|
Protocol: swarm.PortConfigProtocolTCP,
|
|
PublishMode: swarm.PortConfigPublishModeIngress,
|
|
Name: "app-port",
|
|
PublishedPort: uint32(deployport),
|
|
TargetPort: uint32(deployport),
|
|
},
|
|
},
|
|
},
|
|
TaskTemplate: swarm.TaskSpec{
|
|
Resources: &swarm.ResourceRequirements{
|
|
Reservations: &swarm.Resources{},
|
|
},
|
|
LogDriver: &swarm.Driver{
|
|
Name: "json-file",
|
|
Options: map[string]string{
|
|
"max-size": "10m",
|
|
},
|
|
},
|
|
ContainerSpec: &swarm.ContainerSpec{
|
|
Image: image,
|
|
Env: []string{
|
|
fmt.Sprintf("SHUFFLE_APP_EXPOSED_PORT=%d", deployport),
|
|
fmt.Sprintf("SHUFFLE_SWARM_CONFIG=%s", os.Getenv("SHUFFLE_SWARM_CONFIG")),
|
|
fmt.Sprintf("SHUFFLE_LOGS_DISABLED=%s", logsDisabled),
|
|
fmt.Sprintf("CALLBACK_URL=%s", baseUrl),
|
|
fmt.Sprintf("BASE_URL=%s", appCallbackUrl),
|
|
},
|
|
Hosts: []string{
|
|
containerName,
|
|
},
|
|
},
|
|
RestartPolicy: &swarm.RestartPolicy{
|
|
Condition: swarm.RestartPolicyConditionAny,
|
|
},
|
|
Placement: &swarm.Placement{
|
|
Constraints: []string{},
|
|
},
|
|
},
|
|
}
|
|
|
|
if len(os.Getenv("SHUFFLE_SWARM_OTHER_NETWORK")) > 0 {
|
|
serviceSpec.Networks = append(serviceSpec.Networks, swarm.NetworkAttachmentConfig{
|
|
Target: "shuffle_shuffle",
|
|
})
|
|
}
|
|
|
|
if strings.ToLower(os.Getenv("SHUFFLE_PASS_APP_PROXY")) == "true" {
|
|
serviceSpec.TaskTemplate.ContainerSpec.Env = appendAppProxyEnv(serviceSpec.TaskTemplate.ContainerSpec.Env)
|
|
}
|
|
|
|
overrideHttpProxy := os.Getenv("SHUFFLE_INTERNAL_HTTP_PROXY")
|
|
overrideHttpsProxy := os.Getenv("SHUFFLE_INTERNAL_HTTPS_PROXY")
|
|
if overrideHttpProxy != "" {
|
|
serviceSpec.TaskTemplate.ContainerSpec.Env = append(serviceSpec.TaskTemplate.ContainerSpec.Env, fmt.Sprintf("SHUFFLE_INTERNAL_HTTP_PROXY=%s", overrideHttpProxy))
|
|
|
|
}
|
|
|
|
if overrideHttpsProxy != "" {
|
|
serviceSpec.TaskTemplate.ContainerSpec.Env = append(serviceSpec.TaskTemplate.ContainerSpec.Env, fmt.Sprintf("SHUFFLE_INTERNAL_HTTPS_PROXY=%s", overrideHttpsProxy))
|
|
}
|
|
|
|
proxyConfigOverride := os.Getenv("SHUFFLE_APP_PROXY_CONFIG_OVERRIDE")
|
|
if proxyConfigOverride != "" {
|
|
serviceSpec.TaskTemplate.ContainerSpec.Env = append(serviceSpec.TaskTemplate.ContainerSpec.Env, fmt.Sprintf("SHUFFLE_APP_PROXY_CONFIG_OVERRIDE=%s", proxyConfigOverride))
|
|
}
|
|
|
|
/*
|
|
Mounts: []mount.Mount{
|
|
mount.Mount{
|
|
Source: "/var/run/docker.sock",
|
|
Target: "/var/run/docker.sock",
|
|
Type: mount.TypeBind,
|
|
},
|
|
},
|
|
*/
|
|
|
|
if dockerApiVersion != "" {
|
|
serviceSpec.TaskTemplate.ContainerSpec.Env = append(serviceSpec.TaskTemplate.ContainerSpec.Env, fmt.Sprintf("DOCKER_API_VERSION=%s", dockerApiVersion))
|
|
}
|
|
|
|
if len(os.Getenv("SHUFFLE_APP_SDK_TIMEOUT")) > 0 {
|
|
serviceSpec.TaskTemplate.ContainerSpec.Env = append(serviceSpec.TaskTemplate.ContainerSpec.Env, fmt.Sprintf("SHUFFLE_APP_SDK_TIMEOUT=%s", os.Getenv("SHUFFLE_APP_SDK_TIMEOUT")))
|
|
}
|
|
|
|
// Required for certain apps
|
|
if timezone == "" {
|
|
timezone = "Europe/Amsterdam"
|
|
}
|
|
|
|
serviceSpec.TaskTemplate.ContainerSpec.Env = append(serviceSpec.TaskTemplate.ContainerSpec.Env, fmt.Sprintf("TZ=%s", timezone))
|
|
|
|
serviceOptions := types.ServiceCreateOptions{}
|
|
service, err := dockercli.ServiceCreate(
|
|
context.Background(),
|
|
serviceSpec,
|
|
serviceOptions,
|
|
)
|
|
_ = service
|
|
|
|
if err != nil {
|
|
if strings.Contains(fmt.Sprintf("%s", err), "network") && strings.Contains(fmt.Sprintf("%s", err), "not found") {
|
|
log.Printf("[DEBUG] Network %s not found. Trying to initialize it.", networkName)
|
|
networkErr := initSwarmNetwork()
|
|
if networkErr != nil {
|
|
log.Printf("[ERROR] Failed initializing swarm network: %s", err)
|
|
//return err
|
|
}
|
|
|
|
// Retry deploying the service (once)
|
|
if !retry {
|
|
return deploySwarmService(dockercli, name, image, deployport, -1, true)
|
|
}
|
|
}
|
|
|
|
// For port mapping.
|
|
if strings.Contains(fmt.Sprintf("%s", err), "InvalidArgument") && strings.Contains(fmt.Sprintf("%s", err), "is already in use") {
|
|
//log.Printf("\n\n[WARNING] Port %d is already allocated. Trying to deploy on next port.\n\n", deployport)
|
|
|
|
// Random sleep 1-4 seconds
|
|
time.Sleep(time.Duration(rand.Intn(4)+1) * time.Second)
|
|
|
|
return deploySwarmService(dockercli, name, image, deployport+1, -1, retry)
|
|
}
|
|
|
|
log.Printf("[DEBUG] Failed deploying %s with image %s: %s", name, image, err)
|
|
return err
|
|
} else {
|
|
// wait for service to be ready
|
|
time.Sleep(time.Duration(rand.Intn(4)+1) * time.Second)
|
|
//log.Printf("[DEBUG] Servicecreate request: %#v %#v", service, err)
|
|
// patch service network
|
|
// this is an edgecase that we noticed on docker version 29
|
|
// and API version 1.44
|
|
|
|
// get networkID of swarmNetworkName
|
|
networkID := ""
|
|
|
|
ctx := context.Background()
|
|
|
|
// find network ID
|
|
networks, err := dockercli.NetworkList(ctx, network.ListOptions{})
|
|
if err == nil {
|
|
for _, net := range networks {
|
|
if net.Name == networkName {
|
|
if net.Scope == "swarm" {
|
|
log.Printf("[DEBUG] Found swarm-scoped network: %s (%s)", networkName, net.ID)
|
|
networkID = net.ID
|
|
} else {
|
|
log.Printf("[WARNING] Network %s exists but is not swarm scoped (scope=%s)", networkName, net.Scope)
|
|
}
|
|
|
|
break
|
|
}
|
|
}
|
|
}
|
|
|
|
if networkID == "" {
|
|
log.Printf("[ERROR] Network %s not found", networkName)
|
|
networkID = networkName
|
|
}
|
|
|
|
services, serr := dockercli.ServiceList(ctx, types.ServiceListOptions{})
|
|
if serr == nil {
|
|
for _, svc := range services {
|
|
if svc.ID != service.ID {
|
|
continue
|
|
}
|
|
|
|
// Check if the network is already in there or not
|
|
foundNetwork := false
|
|
for _, netAttach := range svc.Spec.TaskTemplate.Networks {
|
|
if netAttach.Target == networkID {
|
|
foundNetwork = true
|
|
break
|
|
}
|
|
}
|
|
|
|
if foundNetwork {
|
|
log.Printf("[DEBUG] Service %s (%s) already attached to network %s, skipping patch", service.ID, svc.ID, networkID)
|
|
continue
|
|
}
|
|
|
|
log.Printf("[DEBUG] Found service %s (%s) — patching network attach", service.ID, svc.ID)
|
|
|
|
spec := svc.Spec
|
|
spec.TaskTemplate.Networks = append(spec.TaskTemplate.Networks, swarm.NetworkAttachmentConfig{
|
|
Target: networkID,
|
|
})
|
|
|
|
_, uerr := dockercli.ServiceUpdate(ctx, svc.ID, svc.Version, spec, types.ServiceUpdateOptions{})
|
|
if uerr != nil {
|
|
log.Printf("[WARNING] Failed to patch service %s with network %s: %v", service.ID, networkID, uerr)
|
|
} else {
|
|
log.Printf("[INFO] Successfully attached network %s to service %s", networkID, service.ID)
|
|
}
|
|
break
|
|
}
|
|
} else {
|
|
log.Printf("[WARNING] Failed to list services for patching network attach: %v", serr)
|
|
}
|
|
}
|
|
|
|
log.Printf("[DEBUG] Successfully deployed service %s with image %s on port %d", name, image, deployport)
|
|
|
|
return nil
|
|
}
|
|
|
|
/*** ENDREMOVE ***/
|
|
|
|
func findAppInfo(image, name string, redeploy bool) (int, error) {
|
|
|
|
// Sleep between 0 and 1.5 second - ensures deployments have a higher
|
|
// chance of being successful
|
|
time.Sleep(time.Duration(rand.Intn(1500)) * time.Millisecond)
|
|
|
|
highest := baseport
|
|
exposedPort := -1
|
|
|
|
// Exists as a "cache" layer
|
|
if portMappings != nil {
|
|
for key, value := range portMappings {
|
|
if value > highest {
|
|
highest = value
|
|
}
|
|
|
|
if key == name {
|
|
exposedPort = value
|
|
break
|
|
}
|
|
}
|
|
} else {
|
|
portMappings = make(map[string]int)
|
|
}
|
|
|
|
//Filters:
|
|
if exposedPort == -1 || redeploy {
|
|
// dockercli, err := dockerclient.NewEnvClient()
|
|
dockercli, _, err := shuffle.GetDockerClient()
|
|
if err != nil {
|
|
log.Printf("[ERROR] Unable to create docker client (2): %s", err)
|
|
return -1, err
|
|
}
|
|
|
|
serviceListOptions := types.ServiceListOptions{}
|
|
services, err := dockercli.ServiceList(
|
|
context.Background(),
|
|
serviceListOptions,
|
|
)
|
|
|
|
// Basic self-correction
|
|
if err != nil {
|
|
log.Printf("[ERROR] Unable to list services: %s (may continue anyway?)", err)
|
|
if strings.Contains(fmt.Sprintf("%s", err), "is too new") {
|
|
// Static for some reason
|
|
defaultVersion := "1.40"
|
|
dockerApiVersion = defaultVersion
|
|
os.Setenv("DOCKER_API_VERSION", defaultVersion)
|
|
log.Printf("[DEBUG] Setting Docker API to %s default and retrying listing requests", defaultVersion)
|
|
} else {
|
|
return -1, err
|
|
}
|
|
|
|
services, err = dockercli.ServiceList(
|
|
context.Background(),
|
|
serviceListOptions,
|
|
)
|
|
|
|
if err != nil {
|
|
log.Printf("[ERROR] Unable to list services (2): %s", err)
|
|
return -1, err
|
|
}
|
|
}
|
|
|
|
for _, service := range services {
|
|
//log.Printf("[INFO] Service: %#v. Ports: %#v", service.Spec.Annotations.Name, service.Spec.EndpointSpec)
|
|
|
|
for _, endpoint := range service.Spec.EndpointSpec.Ports {
|
|
if !strings.Contains(endpoint.Name, "port") {
|
|
continue
|
|
}
|
|
|
|
// This seems to have concurrency issues
|
|
portMappings[service.Spec.Annotations.Name] = int(endpoint.PublishedPort)
|
|
if int(endpoint.PublishedPort) > highest {
|
|
highest = int(endpoint.PublishedPort)
|
|
}
|
|
|
|
if service.Spec.Annotations.Name == name || service.Spec.Annotations.Name == strings.Replace(name, ".", "-", -1) {
|
|
exposedPort = int(endpoint.PublishedPort)
|
|
//break
|
|
}
|
|
}
|
|
|
|
if service.Spec.Annotations.Name != name && service.Spec.Annotations.Name != strings.Replace(name, ".", "-", -1) {
|
|
continue
|
|
}
|
|
|
|
if redeploy {
|
|
healthyTasks, taskErr := countHealthyServiceTasks(dockercli, service.ID)
|
|
if taskErr != nil {
|
|
log.Printf("[WARNING] Failed checking tasks for service %s: %s", name, taskErr)
|
|
}
|
|
|
|
if healthyTasks > 0 {
|
|
log.Printf("[INFO] Skipping redeploy of app %s since %d task(s) are running", name, healthyTasks)
|
|
break
|
|
}
|
|
|
|
log.Printf("[INFO] Found to redeploy! Service: %s with image %s on port %d", name, image, exposedPort)
|
|
// Remove the service and redeploy it.
|
|
// There are cases where the service doesn't update properly
|
|
// Check when the last update happened. If it was within the last few minutes, skip
|
|
updateAgeSeconds := int(time.Since(service.UpdatedAt).Seconds())
|
|
if updateAgeSeconds > 300 {
|
|
|
|
log.Printf("[INFO] Attempting redeploy of app %s with image %s since last update was %d seconds ago.", name, image, updateAgeSeconds)
|
|
|
|
err = dockercli.ServiceRemove(
|
|
context.Background(),
|
|
service.ID,
|
|
)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed auto-removing service %s: %s", name, err)
|
|
} else {
|
|
log.Printf("[INFO] Auto-removed service %s successfully (rebuild due to redeploy).", name)
|
|
|
|
// Sleep between 8 and 12 seconds
|
|
time.Sleep(time.Duration(rand.Intn(4)+8) * time.Second)
|
|
replicas := service.Spec.Mode.Replicated.Replicas
|
|
err = deploySwarmService(
|
|
dockercli,
|
|
name,
|
|
image,
|
|
exposedPort,
|
|
int64(*replicas),
|
|
false,
|
|
)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed re-deploying service %s: %s", name, err)
|
|
} else {
|
|
time.Sleep(10 * time.Second)
|
|
}
|
|
}
|
|
} else {
|
|
log.Printf("[INFO] Skipping redeploy of app %s since last update was %d seconds ago.", name, updateAgeSeconds)
|
|
}
|
|
}
|
|
|
|
// Break if it's the correct port, as it's the right service
|
|
if exposedPort >= 0 {
|
|
break
|
|
}
|
|
}
|
|
}
|
|
|
|
//log.Printf("[DEBUG] Portmappings: %#v", portMappings)
|
|
|
|
if exposedPort >= 0 {
|
|
//log.Printf("[INFO] Found service %s on port %d - no need to deploy another", name, exposedPort)
|
|
} else {
|
|
// dockercli, err := dockerclient.NewEnvClient()
|
|
dockercli, _, err := shuffle.GetDockerClient()
|
|
if err != nil {
|
|
log.Printf("[ERROR] Unable to create docker client (2): %s", err)
|
|
return -1, err
|
|
}
|
|
|
|
// Increment by 1 for highest port
|
|
if highest <= baseport {
|
|
highest = baseport
|
|
}
|
|
|
|
highest += 1
|
|
err = deploySwarmService(dockercli, name, image, highest, -1, false)
|
|
if err != nil {
|
|
log.Printf("[WARNING] NOT Found service: %s. error: %s", name, err)
|
|
return highest, err
|
|
} else {
|
|
log.Printf("[DEBUG] Waiting 20 seconds before moving on to let app '%s' start properly. Service: %s (swarm)", name, image)
|
|
time.Sleep(time.Duration(20) * time.Second)
|
|
}
|
|
|
|
exposedPort = highest
|
|
//return exposedPort, errors.New("Deployed app %s")
|
|
}
|
|
|
|
return exposedPort, nil
|
|
}
|
|
|
|
// Runs data discovery
|
|
/*** STARTREMOVE ***/
|
|
|
|
func findAppInfoKubernetes(image, name string, env []string) error {
|
|
clientset, _, err := shuffle.GetKubernetesClient()
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed getting kubernetes: %s", err)
|
|
return err
|
|
}
|
|
|
|
// Check if it exists as a pod
|
|
namespace := "default"
|
|
if len(kubernetesNamespace) > 0 {
|
|
namespace = kubernetesNamespace
|
|
}
|
|
|
|
// check deployments
|
|
deployments, err := clientset.AppsV1().Deployments(namespace).List(context.Background(), metav1.ListOptions{})
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed listing deployments: %s", err)
|
|
return err
|
|
}
|
|
|
|
name = strings.Replace(name, "_", "-", -1)
|
|
|
|
// check if it exists as a pod
|
|
// for _, pod := range pods.Items {
|
|
// if pod.Name == name {
|
|
// log.Printf("[INFO] Found pod %s - no need to deploy another", name)
|
|
// return nil
|
|
// }
|
|
// }
|
|
|
|
for _, deployment := range deployments.Items {
|
|
if deployment.Name == name {
|
|
if debug {
|
|
log.Printf("[DEBUG] Found deployment %s - no need to deploy another", name)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
}
|
|
|
|
err = deployk8sApp(image, name, env)
|
|
return err
|
|
}
|
|
|
|
// Backups in case networks are removed
|
|
func initSwarmNetwork() error {
|
|
ctx := context.Background()
|
|
// dockercli, err := dockerclient.NewEnvClient()
|
|
dockercli, _, err := shuffle.GetDockerClient()
|
|
if err != nil {
|
|
log.Printf("[ERROR] Unable to create docker client (2): %s", err)
|
|
return err
|
|
}
|
|
|
|
// Create the network options with the specified MTU
|
|
options := make(map[string]string)
|
|
mtu := 1500
|
|
options["com.docker.network.driver.mtu"] = fmt.Sprintf("%d", mtu)
|
|
|
|
ingressOptions := network.CreateOptions{
|
|
Driver: "overlay",
|
|
Attachable: false,
|
|
Ingress: true,
|
|
IPAM: &network.IPAM{
|
|
Driver: "default",
|
|
Config: []network.IPAMConfig{
|
|
network.IPAMConfig{
|
|
Subnet: "10.225.225.0/24",
|
|
Gateway: "10.225.225.1",
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
_, err = dockercli.NetworkCreate(
|
|
ctx,
|
|
"ingress",
|
|
ingressOptions,
|
|
)
|
|
|
|
if err != nil {
|
|
log.Printf("[WARNING] Ingress network may already exist: %s", err)
|
|
}
|
|
|
|
//docker network create --driver=overlay workers
|
|
// Specific subnet?
|
|
networkName := "shuffle_swarm_executions"
|
|
if len(swarmNetworkName) > 0 {
|
|
networkName = swarmNetworkName
|
|
}
|
|
|
|
networkCreateOptions := network.CreateOptions{
|
|
Driver: "overlay",
|
|
Options: options,
|
|
Attachable: true,
|
|
Ingress: false,
|
|
IPAM: &network.IPAM{
|
|
Driver: "default",
|
|
Config: []network.IPAMConfig{
|
|
network.IPAMConfig{
|
|
Subnet: "10.224.224.0/24",
|
|
Gateway: "10.224.224.1",
|
|
},
|
|
},
|
|
},
|
|
}
|
|
_, err = dockercli.NetworkCreate(
|
|
ctx,
|
|
networkName,
|
|
networkCreateOptions,
|
|
)
|
|
|
|
if err != nil {
|
|
log.Printf("[WARNING] Swarm Executions network may already exist: %s", err)
|
|
}
|
|
|
|
networkName = "shuffle-executions"
|
|
networkCreateOptions = network.CreateOptions{
|
|
Driver: "overlay",
|
|
Options: options,
|
|
Attachable: true,
|
|
Ingress: false,
|
|
IPAM: &network.IPAM{
|
|
Driver: "default",
|
|
Config: []network.IPAMConfig{
|
|
network.IPAMConfig{
|
|
Subnet: "10.223.223.0/24",
|
|
Gateway: "10.223.223.1",
|
|
},
|
|
},
|
|
},
|
|
}
|
|
_, err = dockercli.NetworkCreate(
|
|
ctx,
|
|
networkName,
|
|
networkCreateOptions,
|
|
)
|
|
|
|
if err != nil {
|
|
log.Printf("[WARNING] Swarm Executions network may already exist: %s", err)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
/*** ENDREMOVE ***/
|
|
|
|
func sendAppRequest(ctx context.Context, incomingUrl, appName string, port int, action *shuffle.Action, workflowExecution *shuffle.WorkflowExecution, image string, attempts int64) error {
|
|
parsedRequest := shuffle.OrborusExecutionRequest{
|
|
Cleanup: cleanupEnv,
|
|
ExecutionId: workflowExecution.ExecutionId,
|
|
Authorization: workflowExecution.Authorization,
|
|
EnvironmentName: os.Getenv("ENVIRONMENT_NAME"),
|
|
Timezone: os.Getenv("TZ"),
|
|
HTTPProxy: os.Getenv("HTTP_PROXY"),
|
|
HTTPSProxy: os.Getenv("HTTPS_PROXY"),
|
|
ShufflePassProxyToApp: os.Getenv("SHUFFLE_PASS_APP_PROXY"),
|
|
Url: baseUrl,
|
|
BaseUrl: baseUrl,
|
|
Action: *action,
|
|
FullExecution: *workflowExecution,
|
|
}
|
|
// Sometimes makes it have the wrong data due to timing
|
|
|
|
// Specific for subflow to ensure worker matches the backend correctly
|
|
|
|
parsedBaseurl := incomingUrl
|
|
if strings.Count(baseUrl, ":") >= 2 {
|
|
baseUrlSplit := strings.Split(baseUrl, ":")
|
|
if len(baseUrlSplit) >= 3 {
|
|
parsedBaseurl = strings.Join(baseUrlSplit[0:2], ":")
|
|
//parsedRequest.BaseUrl = fmt.Sprintf("%s:33333", parsedBaseurl)
|
|
}
|
|
}
|
|
|
|
if len(parsedRequest.Url) == 0 {
|
|
// Fixed callback url to the worker itself
|
|
if strings.Count(parsedBaseurl, ":") >= 2 {
|
|
parsedRequest.Url = parsedBaseurl
|
|
} else {
|
|
// Callback to worker
|
|
parsedRequest.Url = fmt.Sprintf("%s:%d", parsedBaseurl, baseport)
|
|
|
|
//parsedRequest.Url
|
|
}
|
|
|
|
//log.Printf("[DEBUG][%s] Should add a baseurl for the app to get back to: %s", workflowExecution.ExecutionId, parsedRequest.Url)
|
|
}
|
|
|
|
// Swapping because this was confusing during dev
|
|
// No real reason, just variable names
|
|
tmp := parsedRequest.Url
|
|
parsedRequest.Url = parsedRequest.BaseUrl
|
|
parsedRequest.BaseUrl = tmp
|
|
|
|
callbackUrl := os.Getenv("SHUFFLE_WORKER_SERVER_URL")
|
|
if len(callbackUrl) > 0 {
|
|
parsedRequest.BaseUrl = callbackUrl
|
|
if parsedRequest.Action.AppName == "shuffle-subflow" || parsedRequest.Action.AppName == "shuffle-subflow-v2" || parsedRequest.Action.AppName == "User Input" {
|
|
parsedRequest.BaseUrl = fmt.Sprintf("http://%s:%d", hostname, baseport)
|
|
//parsedRequest.Url = parsedRequest.BaseUrl
|
|
}
|
|
} else if len(hostname) > 0 {
|
|
// Run with proper hostname, but set to shuffle-worker to avoid specific host target.
|
|
// This means running with VIP instead.
|
|
parsedRequest.BaseUrl = fmt.Sprintf("http://%s:%d", hostname, baseport)
|
|
//parsedRequest.BaseUrl = fmt.Sprintf("http://shuffle-workers:%d", baseport)
|
|
//log.Printf("[DEBUG][%s] Changing hostname to local hostname in Docker network for WORKER URL: %s", workflowExecution.ExecutionId, parsedRequest.BaseUrl)
|
|
|
|
if parsedRequest.Action.AppName == "shuffle-subflow" || parsedRequest.Action.AppName == "shuffle-subflow-v2" || parsedRequest.Action.AppName == "User Input" {
|
|
parsedRequest.BaseUrl = fmt.Sprintf("http://%s:%d", hostname, baseport)
|
|
//parsedRequest.Url = parsedRequest.BaseUrl
|
|
}
|
|
}
|
|
|
|
// Making sure to get the LATEST execution data
|
|
// This is due to cache timing issues
|
|
exec, err := shuffle.GetWorkflowExecution(ctx, workflowExecution.ExecutionId)
|
|
if err == nil && len(exec.ExecutionId) > 0 {
|
|
parsedRequest.FullExecution = *exec
|
|
}
|
|
|
|
restoreActionConfig(ctx, workflowExecution.ExecutionId, action, &parsedRequest.FullExecution)
|
|
parsedRequest.Action = *action
|
|
|
|
data, err := json.Marshal(parsedRequest)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed marshalling worker request: %s", err)
|
|
return err
|
|
}
|
|
|
|
if isKubernetes == "true" {
|
|
appName = strings.Replace(appName, "_", "-", -1)
|
|
}
|
|
|
|
// Shitty hardcoded fix for now
|
|
if strings.Contains(appName, "1.0.0") {
|
|
appName = strings.Replace(appName, "1.0.0", "1-0-0", 1)
|
|
} else if strings.Contains(appName, "1.1.0") {
|
|
appName = strings.Replace(appName, "1.1.0", "1-1-0", 1)
|
|
} else if strings.Contains(appName, "1.2.0") {
|
|
appName = strings.Replace(appName, "1.2.0", "1-2-0", 1)
|
|
} else if strings.Contains(appName, "1.4.0") {
|
|
appName = strings.Replace(appName, "1.4.0", "1-4-0", 1)
|
|
} else if strings.Contains(appName, "2.0.0") {
|
|
appName = strings.Replace(appName, "2.0.0", "2-0-0", 1)
|
|
}
|
|
|
|
streamUrl := fmt.Sprintf("http://%s:%d/api/v1/run", appName, port)
|
|
// log.Printf("[DEBUG][%s] Worker URL: %s, Backend URL: %s, Target App: %s", workflowExecution.ExecutionId, parsedRequest.BaseUrl, parsedRequest.Url, streamUrl)
|
|
req, err := http.NewRequest(
|
|
"POST",
|
|
streamUrl,
|
|
bytes.NewBuffer([]byte(data)),
|
|
)
|
|
|
|
// Checking as LATE as possible, ensuring we don't rerun what's already ran
|
|
// ctx = context.Background()
|
|
|
|
// Sleep between 0 and 250 ms for randomness so no same worker check at same time (same as cloud)
|
|
rand.Seed(time.Now().UnixNano())
|
|
randMs := rand.Intn(250)
|
|
time.Sleep(time.Duration(randMs) * time.Millisecond)
|
|
|
|
newExecId := fmt.Sprintf("%s_%s", workflowExecution.ExecutionId, action.ID)
|
|
_, err = shuffle.GetCache(ctx, newExecId)
|
|
if err == nil {
|
|
log.Printf("[DEBUG] Result for %s already found (PRE REQUEST) - returning", newExecId)
|
|
return nil
|
|
}
|
|
|
|
cacheData := []byte("1")
|
|
err = shuffle.SetCache(ctx, newExecId, cacheData, 30)
|
|
if err != nil {
|
|
log.Printf("[WARNING] Failed setting cache for action %s: %s", newExecId, err)
|
|
} else {
|
|
//log.Printf("[DEBUG][%s] Adding %s to cache (%#v)", workflowExecution.ExecutionId, newExecId, action.Name)
|
|
}
|
|
|
|
client := shuffle.GetExternalClient(streamUrl)
|
|
customTimeout := os.Getenv("SHUFFLE_APP_REQUEST_TIMEOUT")
|
|
if len(customTimeout) > 0 {
|
|
// convert to int
|
|
timeoutInt, err := strconv.Atoi(customTimeout)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed converting SHUFFLE_APP_REQUEST_TIMEOUT to int: %s", err)
|
|
} else {
|
|
log.Printf("[DEBUG] Setting client timeout to %d seconds for app request", timeoutInt)
|
|
client.Timeout = time.Duration(timeoutInt) * time.Second
|
|
}
|
|
} else if len(os.Getenv("SHUFFLE_APP_SDK_TIMEOUT")) > 0 {
|
|
timeoutInt, err := strconv.Atoi(os.Getenv("SHUFFLE_APP_SDK_TIMEOUT"))
|
|
if err == nil && timeoutInt > 0 {
|
|
log.Printf("[DEBUG] Using SHUFFLE_APP_SDK_TIMEOUT=%d seconds for app request", timeoutInt)
|
|
client.Timeout = time.Duration(timeoutInt) * time.Second
|
|
}
|
|
}
|
|
|
|
// Content type required
|
|
req.Header.Set("Content-Type", "application/json")
|
|
|
|
newresp, err := client.Do(req)
|
|
if err != nil {
|
|
// Another timeout issue here somewhere
|
|
// context deadline
|
|
if strings.Contains(fmt.Sprintf("%s", err), "context deadline exceeded") || strings.Contains(fmt.Sprintf("%s", err), "Client.Timeout exceeded") {
|
|
return nil
|
|
}
|
|
|
|
if strings.Contains(fmt.Sprintf("%s", err), "timeout awaiting response") {
|
|
return nil
|
|
}
|
|
|
|
newerr := fmt.Sprintf("%s", err)
|
|
if strings.Contains(newerr, "connection refused") || strings.Contains(newerr, "no such host") {
|
|
newerr = fmt.Sprintf("Failed connecting to app %s. Is the Docker image available?", appName)
|
|
} else {
|
|
// escape quotes and newlines
|
|
newerr = strings.ReplaceAll(strings.ReplaceAll(newerr, "\"", "\\\""), "\n", "\\n")
|
|
}
|
|
|
|
if strings.Contains(fmt.Sprintf("%s", err), "no such host") {
|
|
log.Printf("[ERROR] Should be removing references to location for app '%s' as to be rediscovered. URL: %s. Error: %s", action.AppName, streamUrl, err)
|
|
|
|
//for k, v := range portMappings {
|
|
// if strings.Contains(strings.ToLower(strings.ReplaceAll(action.AppName, " ", "_"))) {
|
|
// }
|
|
//}
|
|
|
|
//var portMappings map[string]int
|
|
}
|
|
|
|
// Try redeployment
|
|
attempts += 1
|
|
if attempts < 2 {
|
|
// Check the service and fix it.
|
|
if isKubernetes == "true" {
|
|
log.Printf("[WARNING] App Redeployment in K8s isn't fully supported yet, but should be done for app %s with image %s.", appName, image)
|
|
} else {
|
|
_, err = findAppInfo(image, appName, true)
|
|
if err != nil {
|
|
log.Printf("[ERROR][%s] Error re-deploying app %s: %s", workflowExecution.ExecutionId, appName, err)
|
|
}
|
|
|
|
return sendAppRequest(ctx, incomingUrl, appName, port, action, workflowExecution, image, attempts)
|
|
}
|
|
}
|
|
|
|
log.Printf("[ERROR][%s] Error running app run request: %s", workflowExecution.ExecutionId, err)
|
|
actionResult := shuffle.ActionResult{
|
|
Action: *action,
|
|
ExecutionId: workflowExecution.ExecutionId,
|
|
Authorization: workflowExecution.Authorization,
|
|
Result: fmt.Sprintf(`{"success": false, "attempts": %d, "reason": "Failed to connect to app %s in swarm. Try the action again, restart Orborus if this is recurring, or contact support@shuffler.io.", "details": "%s"}`, attempts, streamUrl, newerr),
|
|
StartedAt: int64(time.Now().Unix()),
|
|
CompletedAt: int64(time.Now().Unix()),
|
|
Status: "FAILURE",
|
|
}
|
|
|
|
// If this happens - send failure signal to stop the workflow?
|
|
sendSelfRequest(actionResult)
|
|
return err
|
|
}
|
|
|
|
defer newresp.Body.Close()
|
|
body, err := ioutil.ReadAll(newresp.Body)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed reading app request body body: %s", err)
|
|
return err
|
|
} else {
|
|
if debug {
|
|
log.Printf("[DEBUG][%s] NEWRESP (from app): %s", workflowExecution.ExecutionId, string(body))
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// Function to auto-deploy certain apps if "run" is set
|
|
// Has some issues with loading when running multiple workers and such.
|
|
func baseDeploy() {
|
|
|
|
var cli *dockerclient.Client
|
|
//var err error
|
|
|
|
if isKubernetes != "true" {
|
|
// cli, err := dockerclient.NewEnvClient()
|
|
cli, _, err := shuffle.GetDockerClient()
|
|
if err != nil {
|
|
log.Printf("[ERROR] Unable to create docker client (3): %s", err)
|
|
return
|
|
}
|
|
defer cli.Close()
|
|
}
|
|
|
|
for key, value := range autoDeploy {
|
|
newNameSplit := strings.Split(key, ":")
|
|
|
|
action := shuffle.Action{
|
|
AppName: newNameSplit[0],
|
|
AppVersion: newNameSplit[1],
|
|
ID: "TBD",
|
|
}
|
|
|
|
workflowExecution := shuffle.WorkflowExecution{
|
|
ExecutionId: "TBD",
|
|
}
|
|
|
|
appname := action.AppName
|
|
appversion := action.AppVersion
|
|
appname = strings.Replace(appname, ".", "-", -1)
|
|
appversion = strings.Replace(appversion, ".", "-", -1)
|
|
|
|
env := []string{
|
|
fmt.Sprintf("EXECUTIONID=%s", workflowExecution.ExecutionId),
|
|
fmt.Sprintf("AUTHORIZATION=%s", workflowExecution.Authorization),
|
|
fmt.Sprintf("CALLBACK_URL=%s", baseUrl),
|
|
fmt.Sprintf("BASE_URL=%s", appCallbackUrl),
|
|
fmt.Sprintf("TZ=%s", timezone),
|
|
fmt.Sprintf("SHUFFLE_LOGS_DISABLED=%s", logsDisabled),
|
|
}
|
|
|
|
if key == "shuffle-tools-fork:1.0.0" {
|
|
env = append(env, fmt.Sprintf("SHUFFLE_ALLOW_PACKAGE_INSTALL=%s", "true"))
|
|
}
|
|
|
|
if strings.ToLower(os.Getenv("SHUFFLE_PASS_APP_PROXY")) == "true" {
|
|
//log.Printf("APPENDING PROXY TO THE APP!")
|
|
env = appendAppProxyEnv(env)
|
|
}
|
|
|
|
if len(os.Getenv("SHUFFLE_APP_SDK_TIMEOUT")) > 0 {
|
|
log.Printf("[DEBUG] Setting SHUFFLE_APP_SDK_TIMEOUT to %s", os.Getenv("SHUFFLE_APP_SDK_TIMEOUT"))
|
|
env = append(env, fmt.Sprintf("SHUFFLE_APP_SDK_TIMEOUT=%s", os.Getenv("SHUFFLE_APP_SDK_TIMEOUT")))
|
|
}
|
|
|
|
identifier := fmt.Sprintf("%s_%s", appname, appversion)
|
|
//identifier := fmt.Sprintf("%s_%s_%s_%s", appname, appversion, action.ID, workflowExecution.ExecutionId)
|
|
//if strings.Contains(identifier, " ") {
|
|
// identifier = strings.ReplaceAll(identifier, " ", "-")
|
|
//}
|
|
|
|
//deployApp(cli, value, identifier, env, workflowExecution, action)
|
|
log.Printf("[DEBUG] Deploying app with identifier %s to ensure basic apps are available from the get-go", identifier)
|
|
|
|
//findAppInfo("frikky/shuffle:http_1.4.0", "http_1-4-0", true)
|
|
// go findAppInfo(value, identifier, false)
|
|
|
|
// findAppInfo leads to the following error:
|
|
// Unable to list services: Cannot connect to the Docker daemon at unix:///var/run/docker.sock. Is the docker daemon running? (may continue anyway?)
|
|
// Replaced with deployApp again. See https://github.com/Shuffle/Shuffle/issues/1817.
|
|
go deployApp(cli, value, identifier, env, workflowExecution, action)
|
|
//err := deployApp(cli, value, identifier, env, workflowExecution, action)
|
|
//if err != nil {
|
|
// log.Printf("[DEBUG] Failed deploying app %s: %s", value, err)
|
|
//}
|
|
}
|
|
|
|
appsInitialized = true
|
|
}
|
|
|
|
func getStreamResultsWrapper(client *http.Client, req *http.Request, workflowExecution shuffle.WorkflowExecution, firstRequest bool, environments []string) ([]string, error) {
|
|
// Because of this, it always has updated data.
|
|
// Removed request requirement from app_sdk
|
|
newresp, err := topClient.Do(req)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed request: %s", err)
|
|
time.Sleep(time.Duration(sleepTime) * time.Second)
|
|
return environments, err
|
|
}
|
|
|
|
defer newresp.Body.Close()
|
|
body, err := ioutil.ReadAll(newresp.Body)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed reading body: %s", err)
|
|
time.Sleep(time.Duration(sleepTime) * time.Second)
|
|
return environments, err
|
|
}
|
|
|
|
if newresp.StatusCode != 200 {
|
|
log.Printf("[ERROR] StatusCode (1): %d - %s", newresp.StatusCode, string(body))
|
|
time.Sleep(time.Duration(sleepTime) * time.Second)
|
|
return environments, errors.New(fmt.Sprintf("Bad status code from backend: %d", newresp.StatusCode))
|
|
}
|
|
|
|
err = json.Unmarshal(body, &workflowExecution)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed workflowExecution unmarshal: %s", err)
|
|
time.Sleep(time.Duration(sleepTime) * time.Second)
|
|
return environments, err
|
|
}
|
|
|
|
if firstRequest {
|
|
firstRequest = false
|
|
|
|
ctx := context.Background()
|
|
cacheKey := fmt.Sprintf("workflowexecution_%s", workflowExecution.ExecutionId)
|
|
execData, err := json.Marshal(workflowExecution)
|
|
if err != nil {
|
|
log.Printf("[ERROR][%s] Failed marshalling execution during set (3): %s", workflowExecution.ExecutionId, err)
|
|
} else {
|
|
err = shuffle.SetCache(ctx, cacheKey, execData, 30)
|
|
if err != nil {
|
|
log.Printf("[ERROR][%s] Failed adding to cache during setexecution (3): %s", workflowExecution.ExecutionId, err)
|
|
}
|
|
}
|
|
|
|
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)
|
|
}
|
|
}
|
|
|
|
// Checks if a subflow is child of the startnode, as sub-subflows aren't working properly yet
|
|
childNodes := shuffle.FindChildNodes(workflowExecution.Workflow, workflowExecution.Start, []string{}, []string{})
|
|
//log.Printf("[DEBUG] Looking for subflow in %#v to check execution pattern as child of %s", childNodes, workflowExecution.Start)
|
|
subflowFound := false
|
|
for _, childNode := range childNodes {
|
|
for _, trigger := range workflowExecution.Workflow.Triggers {
|
|
if trigger.ID != childNode {
|
|
continue
|
|
}
|
|
|
|
if trigger.AppName == "Shuffle Workflow" {
|
|
subflowFound = true
|
|
break
|
|
}
|
|
}
|
|
|
|
if subflowFound {
|
|
break
|
|
}
|
|
}
|
|
|
|
log.Printf("[DEBUG] Environments: %s. Source: %s. 1 env = webserver, 0 or >1 = default. Subflow exists: %#v", environments, workflowExecution.ExecutionSource, subflowFound)
|
|
if len(environments) == 1 && workflowExecution.ExecutionSource != "default" && !subflowFound {
|
|
log.Printf("[DEBUG] Running OPTIMIZED execution (not manual)")
|
|
os.Setenv("SHUFFLE_OPTIMIZED", "true")
|
|
listener := webserverSetup(workflowExecution)
|
|
err := executionInit(workflowExecution)
|
|
if err != nil {
|
|
log.Printf("[DEBUG] Workflow setup failed: %s", workflowExecution.ExecutionId, err)
|
|
log.Printf("[DEBUG] Shutting down (30)")
|
|
shutdown(workflowExecution, "", "", true)
|
|
}
|
|
|
|
go func() {
|
|
time.Sleep(time.Duration(1))
|
|
handleExecutionResult(workflowExecution)
|
|
}()
|
|
|
|
log.Printf("[DEBUG] Running with port %#v", os.Getenv("WORKER_PORT"))
|
|
|
|
runWebserver(listener)
|
|
|
|
// Set environment variable
|
|
|
|
//log.Printf("Before wait")
|
|
//wg := sync.WaitGroup{}
|
|
//wg.Add(1)
|
|
//wg.Wait()
|
|
} else {
|
|
log.Printf("[DEBUG] Running NON-OPTIMIZED execution for type %s with %d environment(s). This only happens when ran manually OR when running with subflows. Status: %s", workflowExecution.ExecutionSource, len(environments), workflowExecution.Status)
|
|
err := executionInit(workflowExecution)
|
|
if err != nil {
|
|
log.Printf("[DEBUG] Workflow setup failed: %s", workflowExecution.ExecutionId, err)
|
|
shutdown(workflowExecution, "", "", true)
|
|
}
|
|
|
|
// Trying to make worker into microservice~ :)
|
|
}
|
|
}
|
|
|
|
if workflowExecution.Status == "FINISHED" || workflowExecution.Status == "SUCCESS" {
|
|
log.Printf("[DEBUG] Workflow %s is finished. Exiting worker.", workflowExecution.ExecutionId)
|
|
log.Printf("[DEBUG] Shutting down (31)")
|
|
shutdown(workflowExecution, "", "", true)
|
|
}
|
|
|
|
if workflowExecution.Status == "EXECUTING" || workflowExecution.Status == "RUNNING" {
|
|
//log.Printf("Status: %s", workflowExecution.Status)
|
|
err = handleDefaultExecution(client, req, workflowExecution)
|
|
if err != nil {
|
|
log.Printf("[DEBUG] Workflow %s is finished: %s", workflowExecution.ExecutionId, err)
|
|
log.Printf("[DEBUG] Shutting down (32)")
|
|
shutdown(workflowExecution, "", "", true)
|
|
}
|
|
} else {
|
|
log.Printf("[DEBUG] Workflow %s has status %s. Exiting worker (if WAITING, rerun will happen).", workflowExecution.ExecutionId, workflowExecution.Status)
|
|
log.Printf("[DEBUG] Shutting down (33)")
|
|
shutdown(workflowExecution, workflowExecution.Workflow.ID, "", true)
|
|
}
|
|
|
|
time.Sleep(time.Duration(sleepTime) * time.Second)
|
|
return environments, nil
|
|
}
|
|
|
|
func checkStandaloneRun() {
|
|
// Check if the required argc/argv is set
|
|
//log.Printf("ARGS: %#v", os.Args)
|
|
if len(os.Args) < 4 {
|
|
if debug {
|
|
log.Printf("[DEBUG] You can run the worker in standalone mode with: go run worker.go standalone <executionid> <authorization> <optional:url>")
|
|
}
|
|
|
|
return
|
|
}
|
|
|
|
if os.Args[1] != "standalone" {
|
|
log.Printf("[ERROR] First argument should be 'standalone' to run worker standalone")
|
|
return
|
|
}
|
|
|
|
if os.Args[2] == "" || len(os.Args[2]) != 36 {
|
|
log.Printf("[ERROR] Second argument should be the execution ID, with next being authorization")
|
|
return
|
|
}
|
|
|
|
if os.Args[3] == "" || len(os.Args[3]) < 10 {
|
|
log.Printf("[ERROR] Third argument should be the authorization key")
|
|
return
|
|
}
|
|
|
|
backendUrl := "https://shuffler.io"
|
|
if len(os.Args) > 4 {
|
|
backendUrl = os.Args[4]
|
|
}
|
|
|
|
if !strings.Contains(backendUrl, "http") {
|
|
log.Printf("[ERROR] Backend URL should start with http:// or https://")
|
|
return
|
|
|
|
}
|
|
|
|
// Format:
|
|
// go run worker.go standalone <executionid> <authorization> <optional:url>
|
|
executionId := os.Args[2]
|
|
authorization := os.Args[3]
|
|
|
|
os.Setenv("EXECUTIONID", executionId)
|
|
os.Setenv("AUTHORIZATION", authorization)
|
|
os.Setenv("BASE_URL", backendUrl)
|
|
|
|
os.Setenv("STANDALONE_EXECUTION", "true")
|
|
|
|
log.Printf("\n\n\n[DEBUG] Running worker in standalone mode with execution ID %s and authorization %s. Backend URL (default): %s\nThis means we will first RESET the execution results, then rerun it\n\n", executionId, authorization, backendUrl)
|
|
|
|
// 1. Reset the execution after getting it
|
|
data = fmt.Sprintf(`{"execution_id": "%s", "authorization": "%s"}`, executionId, authorization)
|
|
streamResultUrl := fmt.Sprintf("%s/api/v1/streams/results", backendUrl)
|
|
client := shuffle.GetExternalClient(streamResultUrl)
|
|
req, err := http.NewRequest(
|
|
"POST",
|
|
streamResultUrl,
|
|
bytes.NewBuffer([]byte(data)),
|
|
)
|
|
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed making request builder for backend: %s", err)
|
|
os.Exit(1)
|
|
}
|
|
|
|
// Read the data and unmarshal it
|
|
newresp, err := client.Do(req)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed standalone request in setup: %s", err)
|
|
os.Exit(1)
|
|
}
|
|
|
|
defer newresp.Body.Close()
|
|
body, err := ioutil.ReadAll(newresp.Body)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed reading body: %s", err)
|
|
os.Exit(1)
|
|
}
|
|
|
|
if newresp.StatusCode != 200 {
|
|
log.Printf("[ERROR] Failed resetting execution: %s. Body: %s", newresp.Status, string(body))
|
|
os.Exit(1)
|
|
}
|
|
|
|
// Map to shuffle.Workflowexecution struct
|
|
workflowExecution := shuffle.WorkflowExecution{}
|
|
err = json.Unmarshal(body, &workflowExecution)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed unmarshalling body: %s", err)
|
|
os.Exit(1)
|
|
}
|
|
|
|
log.Printf("[DEBUG][%s] Got %d results with status %s. Running full reset IF status is not executing.", workflowExecution.ExecutionId, len(workflowExecution.Results), workflowExecution.Status)
|
|
|
|
// Just continue as per usual?
|
|
//if workflowExecution.Status == "EXECUTING" {
|
|
// return
|
|
//}
|
|
|
|
workflowExecution.Status = "EXECUTING"
|
|
|
|
newResults := []shuffle.ActionResult{}
|
|
for _, result := range workflowExecution.Results {
|
|
if result.Status == "SKIPPED" {
|
|
newResults = append(newResults, result)
|
|
continue
|
|
}
|
|
|
|
// This is to handle reruns of SINGLE actions
|
|
if result.Action.Category == "rerun" {
|
|
newResults = append(newResults, result)
|
|
continue
|
|
}
|
|
|
|
// Anything else here.
|
|
}
|
|
|
|
workflowExecution.Results = newResults
|
|
workflowExecution.Status = "EXECUTING"
|
|
workflowExecution.CompletedAt = 0
|
|
|
|
marshalledResult, err := json.Marshal(workflowExecution)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed marshalling body: %s", err)
|
|
os.Exit(1)
|
|
}
|
|
|
|
// Send a /api/v1/streams result back
|
|
// 1. Reset the execution after getting it
|
|
streamUrl := fmt.Sprintf("%s/api/v1/streams?reset=true", backendUrl)
|
|
req, err = http.NewRequest(
|
|
"POST",
|
|
streamUrl,
|
|
bytes.NewBuffer([]byte(marshalledResult)),
|
|
)
|
|
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed making request builder (2) for backend: %s", err)
|
|
os.Exit(1)
|
|
}
|
|
|
|
// Read the data and unmarshal it
|
|
newresp, err = client.Do(req)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed standalone request in setup: %s", err)
|
|
os.Exit(1)
|
|
}
|
|
|
|
defer newresp.Body.Close()
|
|
body, err = ioutil.ReadAll(newresp.Body)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed reading body (2): %s", err)
|
|
os.Exit(1)
|
|
}
|
|
|
|
if newresp.StatusCode != 200 {
|
|
log.Printf("[ERROR] Failed resetting execution (2): %s. Body: %s", newresp.Status, string(body))
|
|
os.Exit(1)
|
|
}
|
|
|
|
log.Printf("\n\n\n[DEBUG] Finished resetting execution %s. Body: %s. Starting execution.\n\n\n", newresp.Status, string(body))
|
|
|
|
}
|
|
|
|
// Initial loop etc
|
|
func main() {
|
|
// Testing swarm auto-replacements. This also tests ports
|
|
// in rapid succession
|
|
|
|
checkStandaloneRun()
|
|
if os.Getenv("DEBUG") == "true" {
|
|
debug = true
|
|
|
|
log.Printf("[INFO] Disabled cleanup due to debug mode (DEBUG=true)")
|
|
cleanupEnv = "false"
|
|
}
|
|
|
|
/*** STARTREMOVE ***/
|
|
if os.Getenv("SHUFFLE_SWARM_CONFIG") == "run" || os.Getenv("SHUFFLE_SWARM_CONFIG") == "swarm" {
|
|
logsDisabled = "true"
|
|
os.Setenv("SHUFFLE_LOGS_DISABLED", "true")
|
|
}
|
|
/*** ENDREMOVE ***/
|
|
|
|
// Elasticsearch necessary to ensure we'ren ot running with Datastore configurations for minimal/maximal data sizes
|
|
// Recursive import kind of :)
|
|
_, err := shuffle.RunInit(*shuffle.GetDatastore(), *shuffle.GetStorage(), "", "worker", true, "elasticsearch", false, 0)
|
|
if err != nil {
|
|
if !strings.Contains(fmt.Sprintf("%s", err), "no such host") {
|
|
log.Printf("[ERROR] Failed to run worker init: %s", err)
|
|
}
|
|
} else {
|
|
if isKubernetes != "true" {
|
|
log.Printf("[DEBUG] Ran init for worker to set up cache system. Docker version: %s", dockerApiVersion)
|
|
} else {
|
|
log.Printf("[DEBUG] Ran init for worker to set up cache system on Kubernetes")
|
|
}
|
|
}
|
|
|
|
//log.Printf("[INFO] Setting up worker environment")
|
|
sleepTime = 5
|
|
client := shuffle.GetExternalClient(baseUrl)
|
|
|
|
if timezone == "" {
|
|
timezone = "Europe/Amsterdam"
|
|
}
|
|
|
|
if baseimagename == "" {
|
|
log.Printf("[DEBUG] Setting baseimagename to frikky/shuffle as it's empty (docker.io)")
|
|
baseimagename = "frikky/shuffle" // Dockerhub
|
|
//baseimagename = "shuffle" // Github (ghcr.io)
|
|
}
|
|
|
|
topClient = client
|
|
swarmConfig := os.Getenv("SHUFFLE_SWARM_CONFIG")
|
|
log.Printf("[INFO] Running with timezone %s and swarm config %#v", timezone, swarmConfig)
|
|
|
|
/*** STARTREMOVE ***/
|
|
if swarmConfig == "run" || swarmConfig == "swarm" {
|
|
// Forcing download just in case on the first iteration.
|
|
log.Printf("[INFO] Running in swarm mode - forcing download of apps")
|
|
workflowExecution := shuffle.WorkflowExecution{}
|
|
|
|
go baseDeploy()
|
|
|
|
listener := webserverSetup(workflowExecution)
|
|
runWebserver(listener)
|
|
|
|
// Should never get down here
|
|
log.Printf("[ERROR] Stopped listener %#v - exiting.", listener)
|
|
os.Exit(3)
|
|
}
|
|
/*** ENDREMOVE ***/
|
|
|
|
authorization := ""
|
|
executionId := ""
|
|
|
|
// INFO: Allows you to run a test execution
|
|
testing := os.Getenv("WORKER_TESTING_WORKFLOW")
|
|
shuffle_apikey := os.Getenv("WORKER_TESTING_APIKEY")
|
|
if len(testing) > 0 && len(shuffle_apikey) > 0 {
|
|
// Execute a workflow and use that info
|
|
log.Printf("[WARNING] Running test environment for worker by executing workflow %s. PS: This may NOT reach the worker in real time, but rather be deployed as a docker container (bad). Instead use AUTHORIZATION and EXECUTIONID for direct testing", testing)
|
|
authorization, executionId = runTestExecution(client, testing, shuffle_apikey)
|
|
|
|
} else {
|
|
authorization = os.Getenv("AUTHORIZATION")
|
|
executionId = os.Getenv("EXECUTIONID")
|
|
log.Printf("[INFO] Running normal execution with auth %s and ID %s", authorization, executionId)
|
|
}
|
|
|
|
workflowExecution := shuffle.WorkflowExecution{
|
|
ExecutionId: executionId,
|
|
}
|
|
if len(authorization) == 0 {
|
|
log.Printf("[INFO] No AUTHORIZATION key set in env")
|
|
log.Printf("[DEBUG] Shutting down (27)")
|
|
shutdown(workflowExecution, "", "", false)
|
|
}
|
|
|
|
if len(executionId) == 0 {
|
|
log.Printf("[INFO] No EXECUTIONID key set in env")
|
|
log.Printf("[DEBUG] Shutting down (28)")
|
|
shutdown(workflowExecution, "", "", false)
|
|
}
|
|
|
|
data = fmt.Sprintf(`{"execution_id": "%s", "authorization": "%s"}`, executionId, authorization)
|
|
streamResultUrl := fmt.Sprintf("%s/api/v1/streams/results", baseUrl)
|
|
req, err := http.NewRequest(
|
|
"POST",
|
|
streamResultUrl,
|
|
bytes.NewBuffer([]byte(data)),
|
|
)
|
|
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed making request builder for backend")
|
|
log.Printf("[DEBUG] Shutting down (29)")
|
|
shutdown(workflowExecution, "", "", true)
|
|
}
|
|
|
|
firstRequest := true
|
|
environments := []string{}
|
|
for {
|
|
environments, err = getStreamResultsWrapper(client, req, workflowExecution, firstRequest, environments)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed getting stream results: %s", err)
|
|
}
|
|
}
|
|
}
|
|
|
|
func checkUnfinished(execRequest shuffle.OrborusExecutionRequest) {
|
|
// Meant as a function that periodically checks whether previous executions have finished or not.
|
|
// Should probably be based on executedIds and finishedIds
|
|
// Schedule a check in the future instead?
|
|
|
|
ctx := context.Background()
|
|
exec, err := shuffle.GetWorkflowExecution(ctx, execRequest.ExecutionId)
|
|
log.Printf("[DEBUG][%s] Rechecking execution and it's status to send to backend IF the status is EXECUTING (%s - %d/%d finished)", execRequest.ExecutionId, exec.Status, len(exec.Results), len(exec.Workflow.Actions))
|
|
|
|
// FIXMe: Does this create issue with infinite loops?
|
|
// Usually caused by issue during startup
|
|
if exec.Status == "" {
|
|
//handleRunExecution(resp, request)
|
|
return
|
|
}
|
|
|
|
if exec.Status != "EXECUTING" {
|
|
return
|
|
}
|
|
|
|
log.Printf("[DEBUG][%s] Should send full result for execution to backend as it has %d results. Status: %s", execRequest.ExecutionId, len(exec.Results), exec.Status)
|
|
data, err := json.Marshal(exec)
|
|
if err != nil {
|
|
return
|
|
}
|
|
|
|
sendResult(*exec, data)
|
|
}
|
|
|
|
func processRunExecution(execRequest shuffle.OrborusExecutionRequest) {
|
|
// Checks if a workflow is done 30 seconds later, and sends info to backend no matter what
|
|
go func() {
|
|
time.Sleep(time.Duration(30) * time.Second)
|
|
checkUnfinished(execRequest)
|
|
}()
|
|
window.AddEvent(time.Now())
|
|
|
|
ctx := context.Background()
|
|
|
|
if len(execRequest.HTTPProxy) > 0 {
|
|
log.Printf("[DEBUG] Sending proxy info to child process")
|
|
os.Setenv("SHUFFLE_PASS_APP_PROXY", execRequest.ShufflePassProxyToApp)
|
|
}
|
|
if len(execRequest.HTTPProxy) > 0 {
|
|
log.Printf("[DEBUG] Running with default HTTP proxy %s", execRequest.HTTPProxy)
|
|
os.Setenv("HTTP_PROXY", execRequest.HTTPProxy)
|
|
}
|
|
if len(execRequest.HTTPSProxy) > 0 {
|
|
log.Printf("[DEBUG] Running with default HTTPS proxy %s", execRequest.HTTPSProxy)
|
|
os.Setenv("HTTPS_PROXY", execRequest.HTTPSProxy)
|
|
}
|
|
if len(execRequest.EnvironmentName) > 0 {
|
|
os.Setenv("ENVIRONMENT_NAME", execRequest.EnvironmentName)
|
|
environment = execRequest.EnvironmentName
|
|
}
|
|
if len(execRequest.Timezone) > 0 {
|
|
os.Setenv("TZ", execRequest.Timezone)
|
|
timezone = execRequest.Timezone
|
|
}
|
|
if len(execRequest.Cleanup) > 0 {
|
|
os.Setenv("CLEANUP", execRequest.Cleanup)
|
|
cleanupEnv = execRequest.Cleanup
|
|
}
|
|
if len(execRequest.BaseUrl) > 0 {
|
|
os.Setenv("BASE_URL", execRequest.BaseUrl)
|
|
baseUrl = execRequest.BaseUrl
|
|
}
|
|
|
|
if len(execRequest.Authorization) > 0 && len(os.Getenv("AUTHORIZATION")) == 0 {
|
|
os.Setenv("AUTHORIZATION", execRequest.Authorization)
|
|
}
|
|
|
|
var workflowExecution shuffle.WorkflowExecution
|
|
streamResultUrl := fmt.Sprintf("%s/api/v1/streams/results", baseUrl)
|
|
req, err := http.NewRequest(
|
|
"POST",
|
|
streamResultUrl,
|
|
bytes.NewBuffer([]byte(fmt.Sprintf(`{"execution_id": "%s", "authorization": "%s"}`, execRequest.ExecutionId, execRequest.Authorization))),
|
|
)
|
|
|
|
if err != nil {
|
|
log.Printf("[ERROR][%s] Failed to create stream results request: %s", execRequest.ExecutionId, err)
|
|
return
|
|
}
|
|
|
|
client := shuffle.GetExternalClient(streamResultUrl)
|
|
newresp, err := client.Do(req)
|
|
if err != nil {
|
|
log.Printf("[ERROR][%s] Failed making stream results request: %s", execRequest.ExecutionId, err)
|
|
return
|
|
}
|
|
|
|
defer newresp.Body.Close()
|
|
body, err := ioutil.ReadAll(newresp.Body)
|
|
if err != nil {
|
|
log.Printf("[ERROR][%s] Failed reading stream results response body: %s", execRequest.ExecutionId, err)
|
|
return
|
|
}
|
|
|
|
if newresp.StatusCode != 200 {
|
|
log.Printf("[ERROR][%s] Bad statuscode from stream results: %d, %s", execRequest.ExecutionId, newresp.StatusCode, string(body))
|
|
return
|
|
}
|
|
|
|
err = json.Unmarshal(body, &workflowExecution)
|
|
if err != nil {
|
|
log.Printf("[ERROR][%s] Failed workflowExecution unmarshal: %s", execRequest.ExecutionId, err)
|
|
return
|
|
}
|
|
|
|
err = setWorkflowExecution(ctx, workflowExecution, true)
|
|
if err != nil {
|
|
log.Printf("[ERROR][%s] Failed initializing execution saving: %s", workflowExecution.ExecutionId, err)
|
|
}
|
|
|
|
if workflowExecution.Status == "FINISHED" || workflowExecution.Status == "SUCCESS" {
|
|
log.Printf("[DEBUG] Workflow %s is finished before dispatch. Exiting async run setup.", workflowExecution.ExecutionId)
|
|
return
|
|
}
|
|
|
|
extra := 0
|
|
for _, trigger := range workflowExecution.Workflow.Triggers {
|
|
if trigger.AppName == "User Input" || trigger.AppName == "Shuffle Workflow" {
|
|
extra += 1
|
|
}
|
|
}
|
|
|
|
log.Printf("[INFO][%s] (1) Status: %s, Results: %d, actions: %d", workflowExecution.ExecutionId, workflowExecution.Status, len(workflowExecution.Results), len(workflowExecution.Workflow.Actions)+extra)
|
|
|
|
if workflowExecution.Status != "EXECUTING" {
|
|
log.Printf("[WARNING][%s] Exiting async run as execution status is %s", workflowExecution.ExecutionId, workflowExecution.Status)
|
|
return
|
|
}
|
|
|
|
cacheKey := fmt.Sprintf("workflowexecution_%s", workflowExecution.ExecutionId)
|
|
execData, err := json.Marshal(workflowExecution)
|
|
if err != nil {
|
|
log.Printf("[ERROR][%s] Failed marshalling execution during set (3): %s", workflowExecution.ExecutionId, err)
|
|
} else {
|
|
err = shuffle.SetCache(ctx, cacheKey, execData, 31)
|
|
if err != nil {
|
|
log.Printf("[ERROR][%s] Failed adding to cache during setexecution (3): %s", workflowExecution.ExecutionId, err)
|
|
}
|
|
}
|
|
|
|
err = executionInit(workflowExecution)
|
|
if err != nil {
|
|
log.Printf("[DEBUG][%s] Shutting down (30) - Workflow setup failed: %s", workflowExecution.ExecutionId, err)
|
|
return
|
|
}
|
|
|
|
handleExecutionResult(workflowExecution)
|
|
}
|
|
|
|
func handleRunExecution(resp http.ResponseWriter, request *http.Request) {
|
|
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(400)
|
|
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "%s"}`, err)))
|
|
return
|
|
}
|
|
|
|
//log.Printf("[DEBUG] In run execution with body length %d", len(body))
|
|
var execRequest shuffle.OrborusExecutionRequest
|
|
err = json.Unmarshal(body, &execRequest)
|
|
if err != nil {
|
|
log.Printf("[WARNING] Failed shuffle.WorkflowExecution unmarshaling: %s", err)
|
|
resp.WriteHeader(400)
|
|
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "%s"}`, err)))
|
|
return
|
|
}
|
|
|
|
resp.WriteHeader(http.StatusAccepted)
|
|
resp.Write([]byte(`{"success": true, "accepted": true}`))
|
|
|
|
go processRunExecution(execRequest)
|
|
}
|
|
|
|
func handleDownloadImage(resp http.ResponseWriter, request *http.Request) {
|
|
// Read the request body
|
|
defer request.Body.Close()
|
|
bodyBytes, err := ioutil.ReadAll(request.Body)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed reading body for stream result queue. Error: %s", err)
|
|
resp.WriteHeader(401)
|
|
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "%s"}`, err)))
|
|
return
|
|
}
|
|
|
|
// get images from request
|
|
imageBody := &ImageDownloadBody{}
|
|
err = json.Unmarshal(bodyBytes, imageBody)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Error in unmarshalling body: %s", err)
|
|
resp.WriteHeader(401)
|
|
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "%s"}`, err)))
|
|
return
|
|
}
|
|
|
|
// client, err := dockerclient.NewEnvClient()
|
|
client, _, err := shuffle.GetDockerClient()
|
|
if err != nil {
|
|
log.Printf("[ERROR] Unable to create docker client (4): %s", err)
|
|
resp.WriteHeader(401)
|
|
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "%s"}`, err)))
|
|
return
|
|
}
|
|
|
|
defer client.Close()
|
|
|
|
// check if images are already downloaded
|
|
// Retrieve a list of Docker images
|
|
listOptions := dockerimage.ListOptions{}
|
|
images, err := client.ImageList(context.Background(), listOptions)
|
|
if err != nil {
|
|
log.Printf("[ERROR] listing images: %s", err)
|
|
resp.WriteHeader(401)
|
|
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "%s"}`, err)))
|
|
return
|
|
}
|
|
|
|
for _, img := range images {
|
|
for _, tag := range img.RepoTags {
|
|
splitTag := strings.Split(tag, ":")
|
|
baseTag := tag
|
|
if len(splitTag) > 1 {
|
|
baseTag = splitTag[1]
|
|
}
|
|
|
|
var possibleNames []string
|
|
possibleNames = append(possibleNames, fmt.Sprintf("frikky/shuffle:%s", baseTag))
|
|
possibleNames = append(possibleNames, fmt.Sprintf("registry.hub.docker.com/frikky/shuffle:%s", baseTag))
|
|
|
|
if arrayContains(possibleNames, imageBody.Image) {
|
|
log.Printf("[DEBUG] Image %s already downloaded that has been requested to download", imageBody.Image)
|
|
resp.WriteHeader(200)
|
|
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "image already present"}`)))
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
log.Printf("[INFO] Downloading image %s", imageBody.Image)
|
|
err = shuffle.DownloadDockerImageBackend(&http.Client{Timeout: imagedownloadTimeout}, imageBody.Image)
|
|
if err == nil {
|
|
downloadedImages = append(downloadedImages, imageBody.Image)
|
|
}
|
|
|
|
// return success
|
|
resp.WriteHeader(200)
|
|
resp.Write([]byte(fmt.Sprintf(`{"success": true, "status": "starting download"}`)))
|
|
}
|
|
|
|
func runWebserver(listener net.Listener) {
|
|
r := mux.NewRouter()
|
|
r.HandleFunc("/api/v1/streams", handleWorkflowQueue).Methods("POST", "OPTIONS")
|
|
r.HandleFunc("/api/v1/streams/results", handleGetStreamResults).Methods("POST", "OPTIONS")
|
|
r.HandleFunc("/api/v1/download", handleDownloadImage).Methods("POST", "OPTIONS")
|
|
|
|
// Synonyms. Require an execution ID + auth + shuffle backend
|
|
r.HandleFunc("/api/v1/execute", handleRunExecution).Methods("POST", "OPTIONS")
|
|
r.HandleFunc("/api/v1/run", handleRunExecution).Methods("POST", "OPTIONS")
|
|
|
|
// What would be require to run a workflow otherwise?
|
|
// Maybe directly /workflow/run
|
|
|
|
/*** STARTREMOVE ***/
|
|
if os.Getenv("SHUFFLE_SWARM_CONFIG") == "run" || os.Getenv("SHUFFLE_SWARM_CONFIG") == "swarm" {
|
|
log.Printf("[DEBUG] Running webserver config for SWARM and K8s")
|
|
}
|
|
/*** ENDREMOVE ***/
|
|
// var dockercli *dockerclient.Client
|
|
// ctx := context.Background()
|
|
scaleReplicas := os.Getenv("SHUFFLE_APP_REPLICAS")
|
|
if len(scaleReplicas) > 0 {
|
|
tmpInt, err := strconv.Atoi(scaleReplicas)
|
|
if err != nil {
|
|
log.Printf("[ERROR] %s is not a valid number for replication", scaleReplicas)
|
|
} else {
|
|
maxReplicas = uint64(tmpInt)
|
|
_ = tmpInt
|
|
}
|
|
|
|
log.Printf("[DEBUG] SHUFFLE_APP_REPLICAS set to value %#v. Trying to overwrite default (%d/node)", scaleReplicas, maxReplicas)
|
|
}
|
|
|
|
maxExecutionsPerMinute := 10
|
|
if os.Getenv("SHUFFLE_APP_EXECUTIONS_PER_MINUTE") != "" {
|
|
tmpInt, err := strconv.Atoi(os.Getenv("SHUFFLE_APP_EXECUTIONS_PER_MINUTE"))
|
|
if err != nil {
|
|
log.Printf("[ERROR] %s is not a valid number for executions per minute", os.Getenv("SHUFFLE_APP_EXECUTIONS_PER_MINUTE"))
|
|
} else {
|
|
maxExecutionsPerMinute = tmpInt
|
|
}
|
|
|
|
log.Printf("[DEBUG] SHUFFLE_APP_EXECUTIONS_PER_MINUTE set to value %s. Trying to overwrite default (%d)", os.Getenv("SHUFFLE_APP_EXECUTIONS_PER_MINUTE"), maxExecutionsPerMinute)
|
|
}
|
|
|
|
if strings.ToLower(os.Getenv("SHUFFLE_SWARM_CONFIG")) == "run" || strings.ToLower(os.Getenv("SHUFFLE_APP_REPLICAS")) == "" {
|
|
// go AutoScaleApps(ctx, dockercli, maxExecutionsPerMinute)
|
|
}
|
|
if strings.ToLower(os.Getenv("SHUFFLE_DEBUG_MEMORY")) == "true" || strings.ToLower(os.Getenv("DEBUG_MEMORY")) == "true" {
|
|
r.HandleFunc("/debug/pprof/", pprof.Index)
|
|
r.HandleFunc("/debug/pprof/heap", pprof.Handler("heap").ServeHTTP)
|
|
r.HandleFunc("/debug/pprof/profile", pprof.Profile)
|
|
r.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
|
|
r.HandleFunc("/debug/pprof/trace", pprof.Trace)
|
|
}
|
|
|
|
//log.Fatal(http.ListenAndServe(port, nil))
|
|
//srv := http.Server{
|
|
// Addr: ":8888",
|
|
// WriteTimeout: 1 * time.Second,
|
|
// Handler: http.HandlerFunc(slowHandler),
|
|
//}
|
|
|
|
//log.Fatal(http.Serve(listener, nil))
|
|
|
|
log.Printf("[DEBUG] NEW webserver setup. Port: %s", listener.Addr().String())
|
|
|
|
http.Handle("/", r)
|
|
srv := http.Server{
|
|
Handler: r,
|
|
ReadTimeout: 60 * time.Second,
|
|
ReadHeaderTimeout: 60 * time.Second,
|
|
IdleTimeout: 60 * time.Second,
|
|
WriteTimeout: 60 * time.Second,
|
|
}
|
|
|
|
err := srv.Serve(listener)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Serve issue in worker: %#v", err)
|
|
}
|
|
}
|
|
|
|
// 0x0elliot:
|
|
// IF we had to rewrite this, we will focus on ONLY auto scale for apps.
|
|
// i recommend we target executions/minute (?) as a metric.
|
|
// edge-case: subflows are helped with when worker replicas are higher.
|
|
// i kind of never want to scale down. at least, not now.
|
|
// also, algorithm is very broken. executions/worker
|
|
func AutoScaleApps(ctx context.Context, client *dockerclient.Client, maxExecutionsPerMinute int) {
|
|
ticker := time.NewTicker(1 * time.Second)
|
|
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
|
|
case <-ticker.C:
|
|
count := window.CountEvents(time.Now())
|
|
j := numberOfApps(ctx, client)
|
|
workers := numberOfWorkers(ctx, client)
|
|
execPerMin := maxExecutionsPerMinute / workers
|
|
if count >= execPerMin {
|
|
log.Printf("[DEBUG] Too many executions per minute (%d). Scaling down to %d", count, execPerMin)
|
|
scaleApps(ctx, client, uint64(j+1))
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func scaleApps(ctx context.Context, client *dockerclient.Client, replicas uint64) error {
|
|
// client, err := dockerclient.NewEnvClient()
|
|
client, _, err := shuffle.GetDockerClient()
|
|
if err != nil {
|
|
log.Printf("[ERROR] Unable to create docker client (scaleApps): %s", err)
|
|
return err
|
|
}
|
|
|
|
services, err := client.ServiceList(ctx, types.ServiceListOptions{})
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed to find services in the swarm: %s", err)
|
|
}
|
|
|
|
networkId, err := getNetworkId(ctx, client)
|
|
if err != nil {
|
|
log.Printf("[ERROR] Failed to get network Id in the swarm service: %s", err)
|
|
}
|
|
|
|
workers := numberOfWorkers(ctx, client)
|
|
if replicas > uint64(workers) {
|
|
return nil
|
|
}
|
|
|
|
for _, service := range services {
|
|
if service.Spec.Name == "shuffle-workers" {
|
|
continue
|
|
}
|
|
|
|
inNetwork := false
|
|
for _, vip := range service.Endpoint.VirtualIPs {
|
|
if vip.NetworkID == networkId {
|
|
inNetwork = true
|
|
break
|
|
}
|
|
}
|
|
if !inNetwork {
|
|
continue // skip services not in the target network
|
|
}
|
|
|
|
if service.Spec.Mode.Replicated == nil {
|
|
return errors.New("Service is not replicated")
|
|
}
|
|
|
|
if *service.Spec.Mode.Replicated.Replicas >= replicas {
|
|
continue
|
|
}
|
|
|
|
service.Spec.Mode.Replicated.Replicas = &replicas
|
|
_, err = client.ServiceUpdate(ctx, service.ID, service.Version, service.Spec, types.ServiceUpdateOptions{})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
}
|
|
|
|
log.Printf("[DEBUG] Scaled all services to %d replicas", replicas)
|
|
return nil
|
|
}
|
|
|
|
func getNetworkId(ctx context.Context, dockercli *dockerclient.Client) (string, error) {
|
|
networkFilter := filters.NewArgs()
|
|
networkFilter.Add("name", swarmNetworkName)
|
|
|
|
listOptions := network.ListOptions{
|
|
Filters: networkFilter,
|
|
}
|
|
networks, err := dockercli.NetworkList(ctx, listOptions)
|
|
|
|
if err != nil || len(networks) == 0 {
|
|
return "", err
|
|
}
|
|
networkId := networks[0].ID
|
|
|
|
return networkId, nil
|
|
}
|
|
|
|
func numberOfApps(ctx context.Context, dockercli *dockerclient.Client) int {
|
|
// swarmNetworkName
|
|
|
|
var err error
|
|
if swarmNetworkName == "" {
|
|
swarmNetworkName = "shuffle_swarm_executions"
|
|
}
|
|
|
|
if dockercli == nil {
|
|
// dockercli, err = dockerclient.NewEnvClient()
|
|
dockercli, _, err = shuffle.GetDockerClient()
|
|
if err != nil {
|
|
log.Printf("[ERROR] Unable to create docker client (5): %s", err)
|
|
return 0
|
|
}
|
|
}
|
|
|
|
networkFilter := filters.NewArgs()
|
|
networkFilter.Add("name", swarmNetworkName)
|
|
|
|
listOptions := network.ListOptions{
|
|
Filters: networkFilter,
|
|
}
|
|
networks, err := dockercli.NetworkList(ctx, listOptions)
|
|
|
|
if err != nil || len(networks) == 0 {
|
|
return 0
|
|
}
|
|
|
|
networkId, err := getNetworkId(ctx, dockercli)
|
|
if err != nil {
|
|
log.Printf("[WARNING] Failed to get networkID is worker running in swarm: %s", err)
|
|
return 0
|
|
}
|
|
|
|
services, err := dockercli.ServiceList(ctx, types.ServiceListOptions{})
|
|
if err != nil {
|
|
log.Printf("[WARNING] Can't found any services. %s", err)
|
|
return 0
|
|
}
|
|
|
|
runningReplicas := 0
|
|
|
|
for _, service := range services {
|
|
if service.Spec.Name == "shuffle-workers" {
|
|
continue
|
|
}
|
|
|
|
inNetwork := false
|
|
for _, vip := range service.Endpoint.VirtualIPs {
|
|
if vip.NetworkID == networkId {
|
|
inNetwork = true
|
|
break
|
|
}
|
|
}
|
|
if !inNetwork {
|
|
continue // skip services not in the target network
|
|
}
|
|
|
|
filterArgs := filters.NewArgs()
|
|
filterArgs.Add("service", service.Spec.Name)
|
|
filterArgs.Add("desired-state", "running")
|
|
|
|
task, err := dockercli.TaskList(ctx, types.TaskListOptions{
|
|
Filters: filterArgs,
|
|
})
|
|
if err != nil {
|
|
log.Printf("[WARNING] Failed to get the list of running services %s: %s", service.Spec.Name, err)
|
|
continue
|
|
}
|
|
|
|
runningReplicas = len(task)
|
|
break
|
|
}
|
|
|
|
return runningReplicas
|
|
}
|
|
|
|
func IsServiceRunning(ctx context.Context, cli *dockerclient.Client) bool {
|
|
serviceName := "shuffle-tools_1-2-0"
|
|
filterArgs := filters.NewArgs()
|
|
filterArgs.Add("name", serviceName)
|
|
|
|
services, err := cli.ServiceList(ctx, types.ServiceListOptions{Filters: filterArgs})
|
|
|
|
if err != nil {
|
|
log.Printf("[ERROR] Couldn't find %s service running got error: %s", serviceName, err)
|
|
return false
|
|
}
|
|
if len(services) > 0 {
|
|
return true
|
|
}
|
|
|
|
return false
|
|
}
|
|
|
|
func numberOfWorkers(ctx context.Context, cli *dockerclient.Client) int {
|
|
// cli, err := dockerclient.NewEnvClient()
|
|
cli, _, err := shuffle.GetDockerClient()
|
|
if err != nil {
|
|
log.Printf("[ERROR] Unable to create docker client (5): %s", err)
|
|
return 0
|
|
}
|
|
|
|
service, _, err := cli.ServiceInspectWithRaw(ctx, "shuffle-workers", types.ServiceInspectOptions{})
|
|
if err != nil {
|
|
return 0
|
|
}
|
|
|
|
if service.Spec.Mode.Replicated == nil {
|
|
return 0
|
|
}
|
|
|
|
replics := *service.Spec.Mode.Replicated.Replicas
|
|
return int(replics)
|
|
}
|