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

Deploying with new Worker build

Browse Source
This commit is contained in:
frikky
2022-11-14 14:51:06 +01:00
parent 20462417a1
commit 45ed3ea774
2 changed files with 1 additions and 432 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.71
VERSION=1.1.0
echo "Running docker build with $NAME:$VERSION"
#CGO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo -o worker.bin .
functions/onprem/worker/worker.go
-431
View File
@@ -16,7 +16,6 @@ import (
"net/http"
"net/url"
"os"
"strconv"
"strings"
"time"
@@ -24,7 +23,6 @@ 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"
@@ -2375,419 +2373,6 @@ 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)
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.
func baseDeploy() {
//return
@@ -2852,22 +2437,6 @@ func baseDeploy() {
// 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)
}
*/
// 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")