diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index e489f0e6..4cb091f2 100755 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -1395,6 +1395,7 @@ func deployK8sWorker(image string, identifier string, env []string) error { return err } + svcAppProtocol := "http" service := &corev1.Service{ ObjectMeta: metav1.ObjectMeta{ Name: identifier, @@ -1404,9 +1405,10 @@ func deployK8sWorker(image string, identifier string, env []string) error { Selector: matchLabels, Ports: []corev1.ServicePort{ { - Protocol: "TCP", - Port: 33333, - TargetPort: intstr.FromInt(33333), + Protocol: "TCP", + AppProtocol: &svcAppProtocol, + Port: 33333, + TargetPort: intstr.FromInt(33333), }, }, Type: corev1.ServiceTypeClusterIP, diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index 752eca62..c4707c03 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" @@ -678,6 +678,7 @@ func deployk8sApp(image string, identifier string, env []string) error { return err } + svcAppProtocol := "http" service := &corev1.Service{ ObjectMeta: metav1.ObjectMeta{ Name: name, @@ -687,9 +688,10 @@ func deployk8sApp(image string, identifier string, env []string) error { Selector: matchLabels, Ports: []corev1.ServicePort{ { - Protocol: "TCP", - Port: 80, - TargetPort: intstr.FromInt(deployport), + Protocol: "TCP", + AppProtocol: &svcAppProtocol, + Port: 80, + TargetPort: intstr.FromInt(deployport), }, }, Type: corev1.ServiceTypeClusterIP, @@ -3233,8 +3235,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 +3247,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 +3257,6 @@ func deploySwarmService(dockercli *dockerclient.Client, name, image string, depl } } - if len(baseimagename) == 0 || baseimagename == "/" { baseimagename = "frikky/shuffle" //var baseimagename = "frikky/shuffle" @@ -3290,7 +3289,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 +3309,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 +3438,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 +3470,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 +3573,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 +3666,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 +3692,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 +3775,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 +3934,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 +3945,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 +4371,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 +4379,7 @@ func main() { debug = true log.Printf("[INFO] Disabled cleanup due to debug mode (DEBUG=true)") - cleanupEnv = "false" + cleanupEnv = "false" } /*** STARTREMOVE ***/