diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index e489f0e6..c56c0ecf 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"))) } diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index 752eca62..48ab7aa3 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 ***/