marker/shuffle-cracked
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fixed base of new execution model with worker, orborus etc

Browse Source
This commit is contained in:
frikky
2021-10-31 01:47:04 +02:00
parent 607c9adddb
commit 066329e3ae
14 changed files with 655 additions and 176 deletions
Download Patch File Download Diff File Expand all files Collapse all files
functions/onprem/worker/build.sh
+1 -1
View File
@@ -1,5 +1,5 @@
NAME=shuffle-worker
VERSION=0.9.29
VERSION=0.9.30
echo "Running docker build with $NAME:$VERSION"
#CGO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo -o worker.bin .
functions/onprem/worker/go.mod
+3 -1
View File
@@ -1,6 +1,8 @@
module worker
go 1.15
//replace github.com/shuffle/shuffle-shared => ../../../../../git/shuffle-shared
require (
github.com/containerd/containerd v1.5.7 // indirect
@@ -8,6 +10,6 @@ require (
github.com/docker/go-connections v0.4.0 // indirect
github.com/gorilla/mux v1.8.0
github.com/patrickmn/go-cache v2.1.0+incompatible
github.com/shuffle/shuffle-shared v0.1.20
github.com/shuffle/shuffle-shared v0.1.25
go4.org v0.0.0-20201209231011-d4a079459e60 // indirect
)
functions/onprem/worker/go.sum
+6
View File
@@ -573,6 +573,12 @@ github.com/shuffle/shuffle-shared v0.1.19 h1:bZmwdC3gKPFxtKoGEjHGY8C96+xB+cBdHaz
github.com/shuffle/shuffle-shared v0.1.19/go.mod h1:0QrK51T12CpCj/be8hXduj/RtDnoeaZ3rfogELZE2IU=
github.com/shuffle/shuffle-shared v0.1.20 h1:Wz3DZtFtsd/F3rQuZ4KpIURM2HT1jPwEF3eJg8MX9TE=
github.com/shuffle/shuffle-shared v0.1.20/go.mod h1:0QrK51T12CpCj/be8hXduj/RtDnoeaZ3rfogELZE2IU=
github.com/shuffle/shuffle-shared v0.1.21 h1:9QOCTwgmCG8JEh5yuvBZe1eUkD+iID/5iZZMgCI/50U=
github.com/shuffle/shuffle-shared v0.1.21/go.mod h1:0QrK51T12CpCj/be8hXduj/RtDnoeaZ3rfogELZE2IU=
github.com/shuffle/shuffle-shared v0.1.24 h1:5kJmwN175x8RiF1Ao2bD8WQyc+8ZuOHbrHeJ8H3nO2s=
github.com/shuffle/shuffle-shared v0.1.24/go.mod h1:0QrK51T12CpCj/be8hXduj/RtDnoeaZ3rfogELZE2IU=
github.com/shuffle/shuffle-shared v0.1.25 h1:+Wjgg5FptXdD6wbYgYQ7ALE+WyG97V4x/8VHfPnBMM0=
github.com/shuffle/shuffle-shared v0.1.25/go.mod h1:0QrK51T12CpCj/be8hXduj/RtDnoeaZ3rfogELZE2IU=
github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc=
github.com/sirupsen/logrus v1.0.4-0.20170822132746-89742aefa4b2/go.mod h1:pMByvHTf9Beacp5x1UXfOR9xyW/9antXMhjMPG0dEzc=
github.com/sirupsen/logrus v1.0.6/go.mod h1:pMByvHTf9Beacp5x1UXfOR9xyW/9antXMhjMPG0dEzc=
functions/onprem/worker/worker.go
+349 -32
View File
@@ -23,6 +23,7 @@ import (
"github.com/docker/docker/api/types/container"
//"github.com/docker/docker/api/types/filters"
"github.com/docker/docker/api/types/mount"
"github.com/docker/docker/api/types/swarm"
dockerclient "github.com/docker/docker/client"
//"github.com/go-git/go-billy/v5/memfs"
@@ -39,6 +40,7 @@ var environment = os.Getenv("ENVIRONMENT_NAME")
var baseUrl = os.Getenv("BASE_URL")
var appCallbackUrl = os.Getenv("BASE_URL")
var cleanupEnv = strings.ToLower(os.Getenv("CLEANUP"))
var dockerApiVersion = strings.ToLower(os.Getenv("DOCKER_API_VERSION"))
var timezone = os.Getenv("TZ")
var baseimagename = "frikky/shuffle"
var registryName = "registry.hub.docker.com"
@@ -62,6 +64,10 @@ var allLogs map[string]string
var executionRunning bool
// New Worker mappings
var portMappings map[string]int
var baseport = 33333
// removes every container except itself (worker)
func shutdown(workflowExecution shuffle.WorkflowExecution, nodeId string, reason string, handleResultSend bool) {
log.Printf("[INFO] Shutdown (%s) started with reason %#v. Result amount: %d. ResultsSent: %d, Send result: %#v", workflowExecution.Status, reason, len(workflowExecution.Results), requestsSent, handleResultSend)
@@ -175,7 +181,7 @@ func shutdown(workflowExecution shuffle.WorkflowExecution, nodeId string, reason
time.Sleep(time.Duration(sleepDuration) * time.Second)
os.Exit(3)
} else {
log.Printf("[DEBUG] Sending result and resetting values (K8s & Swarm).")
log.Printf("\n\n[DEBUG] Sending result and resetting values (K8s & Swarm).\n\n")
environments = []string{}
parents = map[string][]string{}
children = map[string][]string{}
@@ -187,16 +193,43 @@ func shutdown(workflowExecution shuffle.WorkflowExecution, nodeId string, reason
startAction = ""
results = []shuffle.ActionResult{}
allLogs = map[string]string{}
requestsSent = 0
executionRunning = false
}
//cacheKey := fmt.Sprintf("workflowexecution-%s", workflowExecution.ExecutionId)
}
// Deploys the internal worker whenever something happens
func deployApp(cli *dockerclient.Client, image string, identifier string, env []string, workflowExecution shuffle.WorkflowExecution, actionId string) error {
func deployApp(cli *dockerclient.Client, image string, identifier string, env []string, workflowExecution shuffle.WorkflowExecution, action shuffle.Action) error {
// form basic hostConfig
ctx := context.Background()
if os.Getenv("SHUFFLE_SWARM_CONFIG") == "run" {
//identifier := fmt.Sprintf("%s_%s_%s_%s", appname, appversion, action.ID, workflowExecution.ExecutionId)
appName := strings.Replace(identifier, fmt.Sprintf("_%s", action.ID), "", -1)
appName = strings.Replace(appName, fmt.Sprintf("_%s", workflowExecution.ExecutionId), "", -1)
appName = strings.ToLower(appName)
log.Printf("[INFO] New appname: %s, image: %s", appName, image)
exposedPort, err := findAppInfo(image, appName)
if err != nil {
log.Printf("[ERROR] Failed finding and creating port for %s: %s", appName, err)
return err
}
log.Printf("[DEBUG] Should run towards port %d for app %s", exposedPort, appName)
err = sendAppRequest(baseUrl, exposedPort, action, workflowExecution)
if err != nil {
log.Printf("[ERROR] Failed sending request to app %s on port %d: %s", appName, exposedPort, err)
return err
}
log.Printf("[DEBUG] Successfully ran request towards port %d for app %s", exposedPort, appName)
return nil
}
// Max 10% CPU every second
//CPUShares: 128,
//CPUQuota: 10000,
@@ -211,7 +244,7 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env []
if os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" {
hostConfig.NetworkMode = container.NetworkMode(fmt.Sprintf("container:worker-%s", workflowExecution.ExecutionId))
log.Printf("Environments: %#v", env)
//log.Printf("Environments: %#v", env)
}
// Removing because log extraction should happen first
@@ -245,7 +278,7 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env []
})
}
} else {
log.Printf("[WARNING] No mounted folders")
log.Printf("[WARNING] Not mounting folders")
}
config := &container.Config{
@@ -305,7 +338,7 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env []
stats, err := cli.ContainerInspect(ctx, cont.ID)
if err != nil {
log.Printf("[ERROR] Failed getting container stats")
log.Printf("[ERROR] Failed getting container stats for container %s: %s", cont.ID, err)
} else {
//log.Printf("[INFO] Info for container: %#v", stats)
//log.Printf("%#v", stats.Config)
@@ -328,7 +361,7 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env []
//allLogs[actionId] = logs
if stats.ContainerJSONBase.State.Status == "exited" && !strings.Contains(logs, "Normal execution.") {
log.Printf("[WARNING] BAD Execution Logs for %s: %s", actionId, logs)
log.Printf("[WARNING] BAD Execution Logs for %s: %s", action.ID, logs)
exit = true
}
}
@@ -540,11 +573,11 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
}
//log.Printf("NEXTACTIONS: %s", nextActions)
queueNodes := []string{}
//if len(nextActions) == 0 {
// nextActions = append(nextActions, startAction)
//}
queueNodes := []string{}
if len(workflowExecution.Results) == 0 {
nextActions = []string{startAction}
} else {
@@ -602,7 +635,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
}
if len(appendActions) > 0 {
log.Printf("APPENDED NODES: %#v", appendActions)
//log.Printf("APPENDED NODES: %#v", appendActions)
nextActions = append(nextActions, appendActions...)
}
}
@@ -1013,7 +1046,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
// If cleanup is set, it should run for efficiency
pullOptions := types.ImagePullOptions{}
if cleanupEnv == "true" {
err = deployApp(dockercli, images[0], identifier, env, workflowExecution, action.ID)
err = deployApp(dockercli, images[0], identifier, env, workflowExecution, action)
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
if strings.Contains(err.Error(), "exited prematurely") {
log.Printf("[DEBUG] Shutting down (2)")
@@ -1025,8 +1058,8 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
executed := false
if err == nil {
log.Printf("[DEBUG] Downloaded image %s from backend (CLEANUP)", image)
//err = deployApp(dockercli, image, identifier, env, workflow, action.ID)
err = deployApp(dockercli, image, identifier, env, workflowExecution, action.ID)
//err = deployApp(dockercli, image, identifier, env, workflow, action)
err = deployApp(dockercli, image, identifier, env, workflowExecution, action)
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
if strings.Contains(err.Error(), "exited prematurely") {
log.Printf("[DEBUG] Shutting down (41)")
@@ -1040,7 +1073,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
if !executed {
image = images[2]
err = deployApp(dockercli, image, identifier, env, workflowExecution, action.ID)
err = deployApp(dockercli, image, identifier, env, workflowExecution, action)
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
if strings.Contains(err.Error(), "exited prematurely") {
log.Printf("[DEBUG] Shutting down (3)")
@@ -1078,7 +1111,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
log.Printf("[INFO] Successfully downloaded %s", image)
}
err = deployApp(dockercli, image, identifier, env, workflowExecution, action.ID)
err = deployApp(dockercli, image, identifier, env, workflowExecution, action)
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
log.Printf("[ERROR] Failed deploying image for the FOURTH time. Aborting if the image doesn't exist")
@@ -1101,7 +1134,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
}
} else {
err = deployApp(dockercli, images[0], identifier, env, workflowExecution, action.ID)
err = deployApp(dockercli, images[0], identifier, env, workflowExecution, action)
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
if strings.Contains(err.Error(), "exited prematurely") {
log.Printf("[DEBUG] Shutting down (9)")
@@ -1112,7 +1145,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
// Trying to replace with lowercase to deploy again. This seems to work with Dockerhub well.
// FIXME: Should try to remotely download directly if this persists.
image = images[1]
err = deployApp(dockercli, image, identifier, env, workflowExecution, action.ID)
err = deployApp(dockercli, image, identifier, env, workflowExecution, action)
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
if strings.Contains(err.Error(), "exited prematurely") {
log.Printf("[DEBUG] Shutting down (10)")
@@ -1125,8 +1158,8 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
executed := false
if err == nil {
log.Printf("[DEBUG] Downloaded image %s from backend (CLEANUP)", image)
//err = deployApp(dockercli, image, identifier, env, workflow, action.ID)
err = deployApp(dockercli, image, identifier, env, workflowExecution, action.ID)
//err = deployApp(dockercli, image, identifier, env, workflow, action)
err = deployApp(dockercli, image, identifier, env, workflowExecution, action)
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
if strings.Contains(err.Error(), "exited prematurely") {
log.Printf("[DEBUG] Shutting down (40)")
@@ -1140,7 +1173,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
if !executed {
image = images[2]
err = deployApp(dockercli, image, identifier, env, workflowExecution, action.ID)
err = deployApp(dockercli, image, identifier, env, workflowExecution, action)
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
if strings.Contains(err.Error(), "exited prematurely") {
log.Printf("[DEBUG] Shutting down (11)")
@@ -1177,7 +1210,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
}
}
err = deployApp(dockercli, image, identifier, env, workflowExecution, action.ID)
err = deployApp(dockercli, image, identifier, env, workflowExecution, action)
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
log.Printf("[ERROR] Failed deploying image for the FOURTH time. Aborting if the image doesn't exist")
if strings.Contains(err.Error(), "exited prematurely") {
@@ -1616,6 +1649,7 @@ func runTestExecution(client *http.Client, workflowId, apikey string) (string, s
}
func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
log.Printf("[DEBUG] Got stream workflow queue")
body, err := ioutil.ReadAll(request.Body)
if err != nil {
log.Println("(3) Failed reading body for workflowqueue")
@@ -1849,6 +1883,7 @@ func validateFinished(workflowExecution shuffle.WorkflowExecution) {
}
func handleGetStreamResults(resp http.ResponseWriter, request *http.Request) {
log.Printf("[DEBUG] Got stream result")
body, err := ioutil.ReadAll(request.Body)
if err != nil {
log.Println("Failed reading body for stream result queue")
@@ -1965,11 +2000,11 @@ func webserverSetup(workflowExecution shuffle.WorkflowExecution) net.Listener {
log.Printf("OLD HOSTNAME: %s", appCallbackUrl)
if os.Getenv("SHUFFLE_SWARM_CONFIG") == "run" {
log.Printf("\n\nStarting webserver on port 33333 with hostname: %s\n\n", hostname)
appCallbackUrl = fmt.Sprintf("http://%s:33333", hostname)
listener, err = net.Listen("tcp", ":33333")
log.Printf("\n\nStarting webserver on port %d with hostname: %s\n\n", baseport, hostname)
appCallbackUrl = fmt.Sprintf("http://%s:%d", hostname, baseport)
listener, err = net.Listen("tcp", fmt.Sprintf(":%d", baseport))
if err != nil {
log.Printf("[ERROR] Failed to assign port to 33333")
log.Printf("[ERROR] Failed to assign port to %d: %s", baseport, err)
return nil
}
@@ -1996,9 +2031,8 @@ func runWebserver(listener net.Listener) {
r.HandleFunc("/api/v1/execute", handleRunExecution).Methods("POST", "OPTIONS")
}
http.Handle("/", r)
//log.Fatal(http.ListenAndServe(port, nil))
http.Handle("/", r)
log.Fatal(http.Serve(listener, nil))
}
@@ -2077,11 +2111,290 @@ func downloadDockerImageBackend(client *http.Client, imageName string) error {
return nil
}
func deploySwarmService(dockercli *dockerclient.Client, name, image string, deployport int) error {
log.Printf("[DEBUG] Deploying service for %s to swarm on port %d", name, deployport)
//containerName := fmt.Sprintf("shuffle-worker-%s", parsedUuid)
if len(baseimagename) == 0 {
baseimagename = "frikky/shuffle"
//var baseimagename = "frikky/shuffle"
//var registryName = "registry.hub.docker.com"
}
//image := fmt.Sprintf("%s:%s", baseimagename, name)
log.Printf("[DEBUG] Deploying app with name %s with image %s", name, image)
containerName := fmt.Sprintf(strings.Replace(name, ".", "-", -1))
serviceSpec := swarm.ServiceSpec{
Annotations: swarm.Annotations{
Name: containerName,
Labels: map[string]string{},
},
Networks: []swarm.NetworkAttachmentConfig{
swarm.NetworkAttachmentConfig{
Target: "shuffle-executions",
},
},
EndpointSpec: &swarm.EndpointSpec{
Ports: []swarm.PortConfig{
swarm.PortConfig{
Protocol: swarm.PortConfigProtocolTCP,
PublishMode: swarm.PortConfigPublishModeIngress,
Name: "app-port",
PublishedPort: uint32(deployport),
TargetPort: uint32(deployport),
},
},
},
TaskTemplate: swarm.TaskSpec{
Resources: &swarm.ResourceRequirements{
Reservations: &swarm.Resources{},
},
ContainerSpec: &swarm.ContainerSpec{
Image: image,
Env: []string{
fmt.Sprintf("SHUFFLE_SWARM_CONFIG=%s", os.Getenv("SHUFFLE_SWARM_CONFIG")),
fmt.Sprintf("SHUFFLE_APP_EXPOSED_PORT=%d", deployport),
},
},
RestartPolicy: &swarm.RestartPolicy{
Condition: swarm.RestartPolicyConditionNone,
},
Placement: &swarm.Placement{
MaxReplicas: 5,
},
},
}
/*
Mounts: []mount.Mount{
mount.Mount{
Source: "/var/run/docker.sock",
Target: "/var/run/docker.sock",
Type: mount.TypeBind,
},
},
*/
if dockerApiVersion != "" {
serviceSpec.TaskTemplate.ContainerSpec.Env = append(serviceSpec.TaskTemplate.ContainerSpec.Env, fmt.Sprintf("DOCKER_API_VERSION=%s", dockerApiVersion))
}
serviceOptions := types.ServiceCreateOptions{}
service, err := dockercli.ServiceCreate(
context.Background(),
serviceSpec,
serviceOptions,
)
_ = service
if err != nil {
log.Printf("[DEBUG] Failed deploying %s with image %s: %s", name, image, err)
return err
}
log.Printf("[DEBUG] Successfully deployed service %s with image %s on port %d", name, image, deployport)
return nil
}
// Runs data discovery
func findAppInfo(image, name string) (int, error) {
dockercli, err := dockerclient.NewEnvClient()
if err != nil {
log.Printf("[ERROR] Unable to create docker client (2): %s", err)
return -1, err
}
highest := baseport
exposedPort := -1
// Exists as a "cache" layer
if portMappings != nil {
for key, value := range portMappings {
if value > highest {
highest = value
}
if key == name {
exposedPort = value
break
}
}
} else {
portMappings = make(map[string]int)
}
//Filters:
if exposedPort == -1 {
serviceListOptions := types.ServiceListOptions{}
services, err := dockercli.ServiceList(
context.Background(),
serviceListOptions,
)
if err != nil {
log.Printf("[ERROR] Unable to list services: %s", err)
return -1, err
}
for _, service := range services {
//log.Printf("[INFO] Service: %#v", service.Spec.Annotations.Name)
for _, endpoint := range service.Spec.EndpointSpec.Ports {
if strings.Contains(endpoint.Name, "port") {
portMappings[service.Spec.Annotations.Name] = int(endpoint.PublishedPort)
if int(endpoint.PublishedPort) > highest {
highest = int(endpoint.PublishedPort)
}
if service.Spec.Annotations.Name == name || service.Spec.Annotations.Name == strings.Replace(name, ".", "-", -1) {
exposedPort = int(endpoint.PublishedPort)
//break
}
}
}
//log.Printf("%s - %s", service.Spec.Annotations.Name, strings.Replace(name, ".", "-", -1))
if service.Spec.Annotations.Name != name && service.Spec.Annotations.Name != strings.Replace(name, ".", "-", -1) {
continue
}
// Break if it's the correct port, as it's the right service
if exposedPort >= 0 {
break
}
}
}
//log.Printf("[DEBUG] Portmappings: %#v", portMappings)
if exposedPort >= 0 {
log.Printf("[INFO] Found service %s on port %d - no need to deploy another", name, exposedPort)
} else {
// Increment by 1 for highest port
if highest <= baseport {
highest = baseport
}
highest += 1
err = deploySwarmService(dockercli, name, image, highest)
if err != nil {
log.Printf("[WARNING] NOT Found service: %s. error: %s", name, err)
return highest, err
} else {
log.Printf("[INFO] Deployed app with name %s", name)
}
exposedPort = highest
log.Printf("[DEBUG] Waiting 10 seconds before moving on to let app start")
time.Sleep(time.Duration(10) * time.Second)
}
return exposedPort, nil
}
func sendAppRequest(incomingUrl string, port int, action shuffle.Action, workflowExecution shuffle.WorkflowExecution) error {
parsedRequest := shuffle.OrborusExecutionRequest{
ExecutionId: workflowExecution.ExecutionId,
Authorization: workflowExecution.Authorization,
EnvironmentName: os.Getenv("ENVIRONMENT_NAME"),
Timezone: os.Getenv("TZ"),
Cleanup: os.Getenv("CLEANUP"),
HTTPProxy: os.Getenv("HTTP_PROXY"),
HTTPSProxy: os.Getenv("HTTPS_PROXY"),
ShufflePassProxyToApp: os.Getenv("SHUFFLE_PASS_APP_PROXY"),
BaseUrl: baseUrl,
Action: action,
FullExecution: workflowExecution,
}
//var baseUrl = os.Getenv("BASE_URL")
//var appCallbackUrl = os.Getenv("BASE_URL")
parsedBaseurl := incomingUrl
if strings.Count(baseUrl, ":") >= 2 {
baseUrlSplit := strings.Split(baseUrl, ":")
if len(baseUrlSplit) >= 3 {
parsedBaseurl = strings.Join(baseUrlSplit[0:2], ":")
//parsedRequest.BaseUrl = fmt.Sprintf("%s:33333", parsedBaseurl)
}
}
if len(parsedRequest.Url) == 0 {
// Fixed callback url to the worker itself
if strings.Count(parsedBaseurl, ":") >= 2 {
parsedRequest.Url = parsedBaseurl
} else {
// Callback to worker
parsedRequest.Url = fmt.Sprintf("%s:%d", parsedBaseurl, baseport)
}
log.Printf("[DEBUG] Should add a baseurl for the app to get back to: %s", parsedRequest.Url)
}
// FIXME: Swapping because this was confusing during dev
tmp := parsedRequest.Url
parsedRequest.Url = parsedRequest.BaseUrl
parsedRequest.BaseUrl = tmp
log.Printf("[DEBUG] Worker URL: %s, Backend URL: %s", parsedRequest.BaseUrl, parsedRequest.Url)
data, err := json.Marshal(parsedRequest)
if err != nil {
log.Printf("[ERROR] Failed marshalling worker request: %s", err)
return err
}
streamUrl := fmt.Sprintf("%s:%d/api/v1/run", parsedBaseurl, port)
req, err := http.NewRequest(
"POST",
streamUrl,
bytes.NewBuffer([]byte(data)),
)
client := &http.Client{}
if err != nil {
log.Printf("[ERROR] Failed creating finishing request: %s", err)
return err
}
newresp, err := client.Do(req)
if err != nil {
log.Printf("[ERROR] Error running finishing request: %s", err)
return err
}
body, err := ioutil.ReadAll(newresp.Body)
if err != nil {
log.Printf("[ERROR] Failed reading body: %s", err)
return err
} else {
log.Printf("[INFO] NEWRESP (from app): %s", string(body))
}
return nil
}
// Initial loop etc
func main() {
/*
appName := "shuffle-tools_1.1.0"
image := "frikky/shuffle:shuffle-tools_1.1.0"
exposedPort, err := findAppInfo(image, appName)
if err != nil {
log.Printf("[ERROR] Failed finding and creating port for %s: %s", appName, err)
os.Exit(3)
}
log.Printf("[DEBUG] Should run towards port %d for app %s", exposedPort, appName)
err = sendAppRequest(appCallbackUrl, exposedPort, shuffle.Action{}, shuffle.WorkflowExecution{})
if err != nil {
log.Printf("[ERROR] Failed sending request to app %s on port %d: %s", appName, exposedPort, err)
os.Exit(3)
}
*/
log.Printf("[INFO] Setting up worker environment")
sleepTime := 5
client := &http.Client{
Transport: &http.Transport{
Proxy: nil,
@@ -2121,6 +2434,8 @@ func main() {
//}()
runWebserver(listener)
log.Printf("[ERROR] Stopped listener - exiting.")
os.Exit(3)
}
//imageName := fmt.Sprintf("%s/%s:shuffle_openapi_1.0.0", registryName, baseimagename)
@@ -2319,12 +2634,14 @@ func main() {
}
func handleRunExecution(resp http.ResponseWriter, request *http.Request) {
if executionRunning {
log.Println("[WARNING] An execution is already running on this worker")
resp.WriteHeader(500)
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "An execution is already running"}`)))
return
}
/*
if executionRunning {
log.Println("[WARNING] An execution is already running on this worker")
resp.WriteHeader(500)
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "An execution is already running"}`)))
return
}
*/
executionRunning = true
body, err := ioutil.ReadAll(request.Body)