diff --git a/functions/onprem/worker/build.sh b/functions/onprem/worker/build.sh index 2fac6a31..25fc3778 100644 --- a/functions/onprem/worker/build.sh +++ b/functions/onprem/worker/build.sh @@ -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 . diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index 70706443..c32c26cf 100644 --- a/functions/onprem/worker/worker.go +++ b/functions/onprem/worker/worker.go @@ -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")