diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index e489f0e6..02f91d0e 100755 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -1146,6 +1146,10 @@ func deployK8sWorker(image string, identifier string, env []string) error { env = append(env, fmt.Sprintf("SHUFFLE_APP_CONTAINER_SECURITY_CONTEXT=%s", appContainerSecurityContext)) } + if len(os.Getenv("SHUFFLE_APP_MOUNT_TMP_VOLUME")) > 0 { + env = append(env, fmt.Sprintf("SHUFFLE_APP_MOUNT_TMP_VOLUME=%s", os.Getenv("SHUFFLE_APP_MOUNT_TMP_VOLUME"))) + } + if len(os.Getenv("SHUFFLE_LOGS_DISABLED")) > 0 { env = append(env, fmt.Sprintf("SHUFFLE_LOGS_DISABLED=%s", os.Getenv("SHUFFLE_LOGS_DISABLED"))) } @@ -2125,7 +2129,7 @@ func main() { if len(os.Getenv("SHUFFLE_SKIP_PIPELINES")) == 0 { os.Setenv("SHUFFLE_SKIP_PIPELINES", "false") os.Setenv("SHUFFLE_PIPELINE_ENABLED", "true") - } + } if os.Getenv("SHUFFLE_SKIP_PIPELINES") != "true" && os.Getenv("SHUFFLE_PIPELINE_ENABLED") != "false" { // Run in 15 seconds in a goroutine @@ -3049,8 +3053,8 @@ func createAndStartTenzirNode(ctx context.Context, containerName, imageName stri Healthcheck: healthconfig, ExposedPorts: nat.PortSet{ "5160/tcp": struct{}{}, - "1514/udp": struct{}{}, - "1514/tcp": struct{}{}, + "1514/udp": struct{}{}, + "1514/tcp": struct{}{}, }, Entrypoint: []string{containerName}, Env: []string{}, @@ -3096,8 +3100,8 @@ func createAndStartTenzirNode(ctx context.Context, containerName, imageName stri hostConfig := &container.HostConfig{ PortBindings: nat.PortMap{ - "1514/tcp": []nat.PortBinding{{HostPort: "1514"}}, - "1514/udp": []nat.PortBinding{{HostPort: "1514"}}, + "1514/tcp": []nat.PortBinding{{HostPort: "1514"}}, + "1514/udp": []nat.PortBinding{{HostPort: "1514"}}, "5160/tcp": []nat.PortBinding{{HostPort: "5160"}}, }, Mounts: []mount.Mount{ @@ -3107,16 +3111,16 @@ func createAndStartTenzirNode(ctx context.Context, containerName, imageName stri Target: "/tmp", }, /* - { - Type: "bind", - Source: tenzirStorageFolder, - Target: "/var/log/tenzir/", - }, - { - Type: "bind", - Source: tenzirStorageFolder, - Target: "/var/cache/tenzir/", - }, + { + Type: "bind", + Source: tenzirStorageFolder, + Target: "/var/log/tenzir/", + }, + { + Type: "bind", + Source: tenzirStorageFolder, + Target: "/var/cache/tenzir/", + }, */ }, VolumeDriver: "local", @@ -3334,7 +3338,7 @@ func createPipeline(command, identifier string) (string, error) { "name": identifier, "hidden": false, "retry_delay": "500.0ms", - "unstoppable": true, + "unstoppable": true, } requestBodyJSON, err := json.Marshal(requestBody) @@ -3406,20 +3410,20 @@ func updatePipelineState(command, pipelineId, action string) (string, error) { url := fmt.Sprintf("%s/api/v0/pipeline/update", pipelineUrl) forwardMethod := "POST" requestBody := map[string]interface{}{ - "id": pipelineId, - "action": action, + "id": pipelineId, + "action": action, /* - "autostart": map[string]bool{ - "created": true, - "completed": false, - "failed": false, - }, - "autodelete": map[string]bool{ - "completed": false, - "failed": false, - "stopped": false, - }, + "autostart": map[string]bool{ + "created": true, + "completed": false, + "failed": false, + }, + "autodelete": map[string]bool{ + "completed": false, + "failed": false, + "stopped": false, + }, */ } diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index 752eca62..e8969424 100644 --- a/functions/onprem/worker/worker.go +++ b/functions/onprem/worker/worker.go @@ -2,7 +2,7 @@ package main import ( "github.com/shuffle/shuffle-shared" - "github.com/shuffle/singul/pkg" + singul "github.com/shuffle/singul/pkg" "bytes" "context" @@ -664,6 +664,27 @@ func deployk8sApp(image string, identifier string, env []string) error { }, } + 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 @@ -3233,8 +3254,6 @@ func deploySwarmService(dockercli *dockerclient.Client, name, image string, depl 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 { @@ -3247,8 +3266,8 @@ func deploySwarmService(dockercli *dockerclient.Client, name, image string, depl } _, err := dockercli.ImagePull( - context.Background(), - image, + context.Background(), + image, dockerimage.PullOptions{}, ) if err != nil { @@ -3257,7 +3276,6 @@ func deploySwarmService(dockercli *dockerclient.Client, name, image string, depl } } - if len(baseimagename) == 0 || baseimagename == "/" { baseimagename = "frikky/shuffle" //var baseimagename = "frikky/shuffle" @@ -3290,7 +3308,7 @@ func deploySwarmService(dockercli *dockerclient.Client, name, image string, depl // Max scale as well nodeCount := uint64(1) - if inputReplicas > 0 && inputReplicas < 100 { + 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) } @@ -3310,7 +3328,6 @@ func deploySwarmService(dockercli *dockerclient.Client, name, image string, depl nodeCount = 1 } - replicatedJobs := uint64(replicas * nodeCount) log.Printf("[DEBUG] Deploying app with name %s with image %s", name, image) @@ -3440,12 +3457,12 @@ func deploySwarmService(dockercli *dockerclient.Client, name, image string, depl } // Retry deploying the service (once) - if !retry { + if !retry { return deploySwarmService(dockercli, name, image, deployport, -1, true) } } - // For port mapping. + // 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) @@ -3472,7 +3489,6 @@ func findAppInfo(image, name string, redeploy bool) (int, error) { // chance of being successful time.Sleep(time.Duration(rand.Intn(1500)) * time.Millisecond) - highest := baseport exposedPort := -1 @@ -3576,10 +3592,10 @@ func findAppInfo(image, name string, redeploy bool) (int, error) { time.Sleep(time.Duration(rand.Intn(4)+8) * time.Second) replicas := service.Spec.Mode.Replicated.Replicas err = deploySwarmService( - dockercli, - name, - image, - exposedPort, + dockercli, + name, + image, + exposedPort, int64(*replicas), false, ) @@ -3669,7 +3685,7 @@ func findAppInfoKubernetes(image, name string, env []string) error { for _, deployment := range deployments.Items { if deployment.Name == name { - if debug { + if debug { log.Printf("[DEBUG] Found deployment %s - no need to deploy another", name) } @@ -3695,7 +3711,6 @@ func initSwarmNetwork() error { mtu := 1500 options["com.docker.network.driver.mtu"] = fmt.Sprintf("%d", mtu) - ingressOptions := network.CreateOptions{ Driver: "overlay", Attachable: false, @@ -3779,11 +3794,9 @@ func initSwarmNetwork() error { log.Printf("[WARNING] Swarm Executions network may already exist: %s", err) } - return nil + return nil } - - /*** ENDREMOVE ***/ func sendAppRequest(ctx context.Context, incomingUrl, appName string, port int, action *shuffle.Action, workflowExecution *shuffle.WorkflowExecution, image string, attempts int64) error { @@ -3940,9 +3953,9 @@ func sendAppRequest(ctx context.Context, incomingUrl, appName string, port int, } // Try redeployment - attempts += 1 + attempts += 1 if attempts < 2 { - // Check the service and fix it. + // 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 { @@ -3951,7 +3964,7 @@ func sendAppRequest(ctx context.Context, incomingUrl, appName string, port int, log.Printf("[ERROR][%s] Error re-deploying app %s: %s", workflowExecution.ExecutionId, appName, err) } - return sendAppRequest(ctx, incomingUrl, appName, port, action, workflowExecution, image, attempts) + return sendAppRequest(ctx, incomingUrl, appName, port, action, workflowExecution, image, attempts) } } @@ -4377,7 +4390,7 @@ func checkStandaloneRun() { // Initial loop etc func main() { - // Testing swarm auto-replacements. This also tests ports + // Testing swarm auto-replacements. This also tests ports // in rapid succession checkStandaloneRun() @@ -4385,7 +4398,7 @@ func main() { debug = true log.Printf("[INFO] Disabled cleanup due to debug mode (DEBUG=true)") - cleanupEnv = "false" + cleanupEnv = "false" } /*** STARTREMOVE ***/ @@ -4835,7 +4848,7 @@ func runWebserver(listener net.Listener) { 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") { + 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)