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

feat(k8s): specify app protocol for worker and app services

Browse Source 
Signed-off-by: Pascal Sthamer <10992664+P4sca1@users.noreply.github.com>
This commit is contained in:
Pascal Sthamer
2025-10-15 10:43:45 +02:00
parent 39c7b1e7d0
commit 68efb7111a
2 changed files with 27 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
functions/onprem/orborus/orborus.go
+5 -3
View File
@@ -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,
functions/onprem/worker/worker.go
+22 -28
View File
@@ -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 ***/