Merge branch 'launch' of https://github.com/frikky/Shuffle into launch
This commit is contained in:
@@ -2,16 +2,21 @@
|
|||||||
|
|
||||||
### DEFAULT
|
### DEFAULT
|
||||||
NAME=shuffle-app_sdk
|
NAME=shuffle-app_sdk
|
||||||
VERSION=1.0.9
|
VERSION=1.1.0
|
||||||
|
|
||||||
docker rmi docker.pkg.github.com/frikky/shuffle/$NAME:$VERSION --force
|
docker rmi docker.pkg.github.com/frikky/shuffle/$NAME:$VERSION --force
|
||||||
docker build . -f Dockerfile -t frikky/shuffle:app_sdk -t frikky/$NAME:$VERSION -t docker.pkg.github.com/frikky/shuffle/$NAME:$VERSION -t ghcr.io/frikky/$NAME:$VERSION -t ghcr.io/frikky/$NAME:nightly
|
docker build . -f Dockerfile -t frikky/shuffle:app_sdk -t frikky/$NAME:$VERSION -t docker.pkg.github.com/frikky/shuffle/$NAME:$VERSION -t ghcr.io/frikky/$NAME:$VERSION -t ghcr.io/frikky/$NAME:nightly -t shuffle/shuffle:app_sdk -t shuffle/$NAME:$VERSION -t docker.pkg.github.com/shuffle/shuffle/$NAME:$VERSION -t ghcr.io/shuffle/$NAME:$VERSION -t ghcr.io/shuffle/$NAME:nightly
|
||||||
|
|
||||||
docker push frikky/shuffle:app_sdk
|
docker push frikky/shuffle:app_sdk
|
||||||
docker push ghcr.io/frikky/$NAME:$VERSION
|
docker push ghcr.io/frikky/$NAME:$VERSION
|
||||||
docker push ghcr.io/frikky/$NAME:nightly
|
docker push ghcr.io/frikky/$NAME:nightly
|
||||||
docker push ghcr.io/frikky/$NAME:latest
|
docker push ghcr.io/frikky/$NAME:latest
|
||||||
|
|
||||||
|
docker push shuffle/shuffle:app_sdk
|
||||||
|
docker push ghcr.io/shuffle/$NAME:$VERSION
|
||||||
|
docker push ghcr.io/shuffle/$NAME:nightly
|
||||||
|
docker push ghcr.io/shuffle/$NAME:latest
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -58,6 +58,7 @@ var workerTimeoutEnv = os.Getenv("SHUFFLE_ORBORUS_EXECUTION_TIMEOUT")
|
|||||||
var concurrencyEnv = os.Getenv("SHUFFLE_ORBORUS_EXECUTION_CONCURRENCY")
|
var concurrencyEnv = os.Getenv("SHUFFLE_ORBORUS_EXECUTION_CONCURRENCY")
|
||||||
var appSdkVersion = os.Getenv("SHUFFLE_APP_SDK_VERSION")
|
var appSdkVersion = os.Getenv("SHUFFLE_APP_SDK_VERSION")
|
||||||
var workerVersion = os.Getenv("SHUFFLE_WORKER_VERSION")
|
var workerVersion = os.Getenv("SHUFFLE_WORKER_VERSION")
|
||||||
|
var newWorkerImage = os.Getenv("SHUFFLE_WORKER_IMAGE")
|
||||||
|
|
||||||
//var baseimagename = "docker.pkg.github.com/frikky/shuffle"
|
//var baseimagename = "docker.pkg.github.com/frikky/shuffle"
|
||||||
//var baseimagename = "ghcr.io/frikky"
|
//var baseimagename = "ghcr.io/frikky"
|
||||||
@@ -689,12 +690,12 @@ func initializeImages() {
|
|||||||
ctx := context.Background()
|
ctx := context.Background()
|
||||||
|
|
||||||
if appSdkVersion == "" {
|
if appSdkVersion == "" {
|
||||||
appSdkVersion = "1.0.0"
|
appSdkVersion = "1.1.0"
|
||||||
log.Printf("[WARNING] SHUFFLE_APP_SDK_VERSION not defined. Defaulting to %s", appSdkVersion)
|
log.Printf("[WARNING] SHUFFLE_APP_SDK_VERSION not defined. Defaulting to %s", appSdkVersion)
|
||||||
}
|
}
|
||||||
|
|
||||||
if workerVersion == "" {
|
if workerVersion == "" {
|
||||||
workerVersion = "nightly"
|
workerVersion = "1.1.0"
|
||||||
log.Printf("[WARNING] SHUFFLE_WORKER_VERSION not defined. Defaulting to %s", workerVersion)
|
log.Printf("[WARNING] SHUFFLE_WORKER_VERSION not defined. Defaulting to %s", workerVersion)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -704,24 +705,23 @@ func initializeImages() {
|
|||||||
log.Printf("[DEBUG] Setting baseimageregistry")
|
log.Printf("[DEBUG] Setting baseimageregistry")
|
||||||
}
|
}
|
||||||
if baseimagename == "" {
|
if baseimagename == "" {
|
||||||
baseimagename = "shuffle/shuffle"
|
baseimagename = "shuffle/shuffle" // Dockerhub
|
||||||
baseimagename = "shuffle"
|
baseimagename = "shuffle" // Github
|
||||||
log.Printf("[DEBUG] Setting baseimagename")
|
log.Printf("[DEBUG] Setting baseimagename")
|
||||||
}
|
}
|
||||||
|
|
||||||
log.Printf("[DEBUG] Setting swarm config to %#v. Default is empty.", swarmConfig)
|
log.Printf("[DEBUG] Setting swarm config to %#v. Default is empty.", swarmConfig)
|
||||||
|
|
||||||
|
newWorker := fmt.Sprintf("%s/%s/shuffle-worker:%s", baseimageregistry, baseimagename, workerVersion)
|
||||||
|
if len(newWorkerImage) > 0 {
|
||||||
|
newWorker = newWorkerImage
|
||||||
|
}
|
||||||
|
|
||||||
// check whether they are the same first
|
// check whether they are the same first
|
||||||
images := []string{
|
images := []string{
|
||||||
fmt.Sprintf("shuffle/shuffle:app_sdk"),
|
fmt.Sprintf("shuffle/shuffle:app_sdk"),
|
||||||
fmt.Sprintf("%s/%s/shuffle-app_sdk:%s", baseimageregistry, baseimagename, appSdkVersion),
|
fmt.Sprintf("%s/%s/shuffle-app_sdk:%s", baseimageregistry, baseimagename, appSdkVersion),
|
||||||
fmt.Sprintf("%s/%s/shuffle-worker:%s", baseimageregistry, baseimagename, workerVersion),
|
newWorker,
|
||||||
// fmt.Sprintf("docker.io/%s:app_sdk", baseimagename),
|
|
||||||
// fmt.Sprintf("docker.io/%s:worker", baseimagename),
|
|
||||||
|
|
||||||
//fmt.Sprintf("%s/worker:%s", baseimagename, workerVersion),
|
|
||||||
//fmt.Sprintf("%s/app_sdk:%s", baseimagename, appSdkVersion),
|
|
||||||
//fmt.Sprintf("frikky/shuffle:app_sdk"),
|
|
||||||
}
|
}
|
||||||
|
|
||||||
pullOptions := types.ImagePullOptions{}
|
pullOptions := types.ImagePullOptions{}
|
||||||
@@ -919,6 +919,10 @@ func main() {
|
|||||||
initializeImages()
|
initializeImages()
|
||||||
|
|
||||||
workerImage := fmt.Sprintf("%s/%s/shuffle-worker:%s", baseimageregistry, baseimagename, workerVersion)
|
workerImage := fmt.Sprintf("%s/%s/shuffle-worker:%s", baseimageregistry, baseimagename, workerVersion)
|
||||||
|
if len(newWorkerImage) > 0 {
|
||||||
|
workerImage = newWorkerImage
|
||||||
|
}
|
||||||
|
|
||||||
if swarmConfig == "run" || swarmConfig == "swarm" {
|
if swarmConfig == "run" || swarmConfig == "swarm" {
|
||||||
checkSwarmService(ctx)
|
checkSwarmService(ctx)
|
||||||
|
|
||||||
@@ -986,9 +990,9 @@ func main() {
|
|||||||
hasStarted := false
|
hasStarted := false
|
||||||
for {
|
for {
|
||||||
//go getStats()
|
//go getStats()
|
||||||
log.Printf("[DEBUG] Prerequest - queue")
|
//log.Printf("[DEBUG] Prerequest - queue")
|
||||||
newresp, err := client.Do(req)
|
newresp, err := client.Do(req)
|
||||||
log.Printf("[DEBUG] Postrequest - queue")
|
//log.Printf("[DEBUG] Postrequest - queue")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Printf("[WARNING] Failed making request to %s: %s", fullUrl, err)
|
log.Printf("[WARNING] Failed making request to %s: %s", fullUrl, err)
|
||||||
|
|
||||||
@@ -1390,7 +1394,6 @@ func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest) error {
|
|||||||
|
|
||||||
//log.Printf("[DEBUG] Data: %s", string(data))
|
//log.Printf("[DEBUG] Data: %s", string(data))
|
||||||
|
|
||||||
//streamUrl := fmt.Sprintf("http://shuffle-workers:33333/api/v1/execute", parsedBaseurl)
|
|
||||||
streamUrl := fmt.Sprintf("http://shuffle-workers:33333/api/v1/execute")
|
streamUrl := fmt.Sprintf("http://shuffle-workers:33333/api/v1/execute")
|
||||||
if containerId == "" || containerId == "shuffle-orborus" {
|
if containerId == "" || containerId == "shuffle-orborus" {
|
||||||
streamUrl = fmt.Sprintf("%s:33333/api/v1/execute", parsedBaseurl)
|
streamUrl = fmt.Sprintf("%s:33333/api/v1/execute", parsedBaseurl)
|
||||||
@@ -1407,6 +1410,10 @@ func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest) error {
|
|||||||
log.Printf("[ERROR] Failed creating worker request: %s", err)
|
log.Printf("[ERROR] Failed creating worker request: %s", err)
|
||||||
if strings.Contains(fmt.Sprintf("%s", err), "connection refused") || strings.Contains(fmt.Sprintf("%s", err), "EOF") {
|
if strings.Contains(fmt.Sprintf("%s", err), "connection refused") || strings.Contains(fmt.Sprintf("%s", err), "EOF") {
|
||||||
workerImage := fmt.Sprintf("%s/%s/shuffle-worker:%s", baseimageregistry, baseimagename, workerVersion)
|
workerImage := fmt.Sprintf("%s/%s/shuffle-worker:%s", baseimageregistry, baseimagename, workerVersion)
|
||||||
|
|
||||||
|
if len(newWorkerImage) > 0 {
|
||||||
|
workerImage = newWorkerImage
|
||||||
|
}
|
||||||
deployServiceWorkers(workerImage)
|
deployServiceWorkers(workerImage)
|
||||||
|
|
||||||
time.Sleep(time.Duration(10) * time.Second)
|
time.Sleep(time.Duration(10) * time.Second)
|
||||||
@@ -1421,6 +1428,11 @@ func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest) error {
|
|||||||
log.Printf("[ERROR] Error running worker request to %s (1): %s", streamUrl, err)
|
log.Printf("[ERROR] Error running worker request to %s (1): %s", streamUrl, err)
|
||||||
if strings.Contains(fmt.Sprintf("%s", err), "connection refused") || strings.Contains(fmt.Sprintf("%s", err), "EOF") {
|
if strings.Contains(fmt.Sprintf("%s", err), "connection refused") || strings.Contains(fmt.Sprintf("%s", err), "EOF") {
|
||||||
workerImage := fmt.Sprintf("%s/%s/shuffle-worker:%s", baseimageregistry, baseimagename, workerVersion)
|
workerImage := fmt.Sprintf("%s/%s/shuffle-worker:%s", baseimageregistry, baseimagename, workerVersion)
|
||||||
|
|
||||||
|
if len(newWorkerImage) > 0 {
|
||||||
|
workerImage = newWorkerImage
|
||||||
|
}
|
||||||
|
|
||||||
deployServiceWorkers(workerImage)
|
deployServiceWorkers(workerImage)
|
||||||
|
|
||||||
time.Sleep(time.Duration(10) * time.Second)
|
time.Sleep(time.Duration(10) * time.Second)
|
||||||
@@ -1444,12 +1456,6 @@ func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest) error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
//workerImage := fmt.Sprintf("%s/%s/shuffle-worker:%s", baseimageregistry, baseimagename, workerVersion)
|
|
||||||
//deployServiceWorkers(workerImage)
|
|
||||||
|
|
||||||
//time.Sleep(time.Duration(10) * time.Second)
|
|
||||||
//err = sendWorkerRequest(executionRequest)
|
|
||||||
|
|
||||||
return errors.New(fmt.Sprintf("Bad statuscode from worker: %d - expecting 200", newresp.StatusCode))
|
return errors.New(fmt.Sprintf("Bad statuscode from worker: %d - expecting 200", newresp.StatusCode))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
NAME=shuffle-worker
|
NAME=shuffle-worker
|
||||||
VERSION=0.9.71
|
VERSION=1.1.0
|
||||||
|
|
||||||
echo "Running docker build with $NAME:$VERSION"
|
echo "Running docker build with $NAME:$VERSION"
|
||||||
#CGO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo -o worker.bin .
|
#CGO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo -o worker.bin .
|
||||||
|
|||||||
@@ -16,7 +16,6 @@ import (
|
|||||||
"net/http"
|
"net/http"
|
||||||
"net/url"
|
"net/url"
|
||||||
"os"
|
"os"
|
||||||
"strconv"
|
|
||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -24,7 +23,6 @@ import (
|
|||||||
"github.com/docker/docker/api/types/container"
|
"github.com/docker/docker/api/types/container"
|
||||||
//"github.com/docker/docker/api/types/filters"
|
//"github.com/docker/docker/api/types/filters"
|
||||||
"github.com/docker/docker/api/types/mount"
|
"github.com/docker/docker/api/types/mount"
|
||||||
"github.com/docker/docker/api/types/swarm"
|
|
||||||
dockerclient "github.com/docker/docker/client"
|
dockerclient "github.com/docker/docker/client"
|
||||||
//"github.com/go-git/go-billy/v5/memfs"
|
//"github.com/go-git/go-billy/v5/memfs"
|
||||||
|
|
||||||
@@ -2375,419 +2373,6 @@ func downloadDockerImageBackend(client *http.Client, imageName string) error {
|
|||||||
return nil
|
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)
|
|
||||||
networkName := "shuffle-executions"
|
|
||||||
if len(swarmNetworkName) > 0 {
|
|
||||||
networkName = swarmNetworkName
|
|
||||||
}
|
|
||||||
|
|
||||||
replicatedJobs := uint64(1)
|
|
||||||
|
|
||||||
// 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 {
|
|
||||||
replicatedJobs = uint64(tmpInt)
|
|
||||||
}
|
|
||||||
|
|
||||||
log.Printf("[DEBUG] SHUFFLE_APP_REPLICAS set to value %#v. Trying to overwrite default (%d/node)", scaleReplicas, replicatedJobs)
|
|
||||||
}
|
|
||||||
|
|
||||||
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 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", os.Getenv("SHUFFLE_LOGS_DISABLED")),
|
|
||||||
},
|
|
||||||
Hosts: []string{
|
|
||||||
containerName,
|
|
||||||
},
|
|
||||||
},
|
|
||||||
RestartPolicy: &swarm.RestartPolicy{
|
|
||||||
Condition: swarm.RestartPolicyConditionNone,
|
|
||||||
},
|
|
||||||
Placement: &swarm.Placement{
|
|
||||||
// Max per node
|
|
||||||
MaxReplicas: 1,
|
|
||||||
},
|
|
||||||
},
|
|
||||||
}
|
|
||||||
|
|
||||||
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 = append(serviceSpec.TaskTemplate.ContainerSpec.Env, fmt.Sprintf("HTTP_PROXY=%s", os.Getenv("HTTP_PROXY")))
|
|
||||||
serviceSpec.TaskTemplate.ContainerSpec.Env = append(serviceSpec.TaskTemplate.ContainerSpec.Env, fmt.Sprintf("HTTPS_PROXY=%s", os.Getenv("HTTPS_PROXY")))
|
|
||||||
serviceSpec.TaskTemplate.ContainerSpec.Env = append(serviceSpec.TaskTemplate.ContainerSpec.Env, fmt.Sprintf("NO_PROXY=%s", os.Getenv("NO_PROXY")))
|
|
||||||
}
|
|
||||||
|
|
||||||
/*
|
|
||||||
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))
|
|
||||||
}
|
|
||||||
|
|
||||||
// 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 {
|
|
||||||
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,
|
|
||||||
)
|
|
||||||
|
|
||||||
// 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", 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
|
|
||||||
|
|
||||||
if appsInitialized {
|
|
||||||
log.Printf("[DEBUG] Waiting 30 seconds before moving on to let app start")
|
|
||||||
time.Sleep(time.Duration(30) * time.Second)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
return exposedPort, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func sendAppRequest(incomingUrl, appName 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)
|
|
||||||
|
|
||||||
//parsedRequest.Url
|
|
||||||
}
|
|
||||||
|
|
||||||
//log.Printf("[DEBUG][%s] Should add a baseurl for the app to get back to: %s", workflowExecution.ExecutionId, parsedRequest.Url)
|
|
||||||
}
|
|
||||||
|
|
||||||
// FIXME: Swapping because this was confusing during dev
|
|
||||||
tmp := parsedRequest.Url
|
|
||||||
parsedRequest.Url = parsedRequest.BaseUrl
|
|
||||||
parsedRequest.BaseUrl = tmp
|
|
||||||
|
|
||||||
//http://3e05d1e7d7a0:33333,
|
|
||||||
|
|
||||||
// Run with proper hostname, but set to shuffle-worker to avoid specific host target.
|
|
||||||
// This means running with VIP instead.
|
|
||||||
if len(hostname) > 0 {
|
|
||||||
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)
|
|
||||||
}
|
|
||||||
|
|
||||||
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)
|
|
||||||
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)),
|
|
||||||
)
|
|
||||||
|
|
||||||
client := &http.Client{}
|
|
||||||
if err != nil {
|
|
||||||
log.Printf("[ERROR] Failed creating app run request: %s", err)
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
// Checking as LATE as possible, ensuring we don't rerun what's already ran
|
|
||||||
ctx := context.Background()
|
|
||||||
newExecId := fmt.Sprintf("%s_%s", workflowExecution.ExecutionId, action.ID)
|
|
||||||
_, err = shuffle.GetCache(ctx, newExecId)
|
|
||||||
if err == nil {
|
|
||||||
log.Printf("\n\n[DEBUG] Result for %s already found (PRE REQUEST) - returning\n\n", newExecId)
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
cacheData := []byte("1")
|
|
||||||
err = shuffle.SetCache(ctx, newExecId, cacheData)
|
|
||||||
if err != nil {
|
|
||||||
log.Printf("[WARNING] Failed setting cache for action %s: %s", newExecId, err)
|
|
||||||
} else {
|
|
||||||
log.Printf("[DEBUG] Adding %s to cache (%s)", newExecId, action.Name)
|
|
||||||
}
|
|
||||||
|
|
||||||
// FIXME:
|
|
||||||
|
|
||||||
newresp, err := client.Do(req)
|
|
||||||
if err != nil {
|
|
||||||
if strings.Contains(fmt.Sprintf("%s", err), "timeout awaiting response") {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
log.Printf("[ERROR] Error running app run request: %s", err)
|
|
||||||
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
body, err := ioutil.ReadAll(newresp.Body)
|
|
||||||
if err != nil {
|
|
||||||
log.Printf("[ERROR] Failed reading app request body body: %s", err)
|
|
||||||
return err
|
|
||||||
} else {
|
|
||||||
log.Printf("[INFO][%s] NEWRESP (from app): %s", workflowExecution.ExecutionId, string(body))
|
|
||||||
}
|
|
||||||
|
|
||||||
// FIXME: Remove
|
|
||||||
/*
|
|
||||||
if len(hostname) > 0 {
|
|
||||||
//streamUrl := fmt.Sprintf("%s:%d/api/v1/run", parsedBaseurl, port)
|
|
||||||
streamUrl := fmt.Sprintf("http://%s:%d/api/v1/run", appName, port)
|
|
||||||
log.Printf("\n\n[DEBUG] Trying execution towards %s", streamUrl)
|
|
||||||
req, err := http.NewRequest(
|
|
||||||
"POST",
|
|
||||||
streamUrl,
|
|
||||||
bytes.NewBuffer([]byte(data)),
|
|
||||||
)
|
|
||||||
|
|
||||||
client := &http.Client{}
|
|
||||||
if err != nil {
|
|
||||||
log.Printf("[ERROR] Failed creating app run request: %s", err)
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
newresp, err := client.Do(req)
|
|
||||||
if err != nil {
|
|
||||||
log.Printf("[ERROR] Error running app run 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
|
|
||||||
}
|
|
||||||
|
|
||||||
// Has some issues with loading when running multiple workers and such.
|
// Has some issues with loading when running multiple workers and such.
|
||||||
func baseDeploy() {
|
func baseDeploy() {
|
||||||
//return
|
//return
|
||||||
@@ -2852,22 +2437,6 @@ func baseDeploy() {
|
|||||||
|
|
||||||
// Initial loop etc
|
// Initial loop etc
|
||||||
func main() {
|
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)
|
|
||||||
}
|
|
||||||
*/
|
|
||||||
|
|
||||||
// Elasticsearch necessary to ensure we'ren ot running with Datastore configurations for minimal/maximal data sizes
|
// Elasticsearch necessary to ensure we'ren ot running with Datastore configurations for minimal/maximal data sizes
|
||||||
_, err := shuffle.RunInit(datastore.Client{}, storage.Client{}, "", "", true, "elasticsearch")
|
_, err := shuffle.RunInit(datastore.Client{}, storage.Client{}, "", "", true, "elasticsearch")
|
||||||
|
|||||||
Reference in New Issue
Block a user