Minor orborus fixes to map tenzir node into orborus container networking if possible

functions/onprem/orborus/go.sum

@@ -303,6 +303,8 @@ github.com/sergi/go-diff v1.1.0 h1:we8PVUC3FE2uYfodKH/nBHMSetSfHDR6scGdBi+erh0=
 github.com/sergi/go-diff v1.1.0/go.mod h1:STckp+ISIX8hZLjrqAeVduY0gWCT9IjLuqbuNXdaHfM=
 github.com/shuffle/shuffle-shared v0.6.74 h1:os3BDSFZnl4U8ZgsTAY8IsTDADcMXhbc1rS9UMa0BIY=
 github.com/shuffle/shuffle-shared v0.6.74/go.mod h1:RAJiSFjmuKmijKTbbEf9A6Ojb+3/te7g71lED7JjPus=
+github.com/shuffle/shuffle-shared v0.6.83 h1:gceT91WtFqh3h9juzTipDhWpxZLfrdtbcsnK+XNj57g=
+github.com/shuffle/shuffle-shared v0.6.83/go.mod h1:RAJiSFjmuKmijKTbbEf9A6Ojb+3/te7g71lED7JjPus=
 github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
 github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
 github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=

functions/onprem/orborus/orborus.go

@@ -111,6 +111,7 @@ var executionIds = []string{}
 var pipelines = []shuffle.PipelineInfoMini{}
 var namespacemade = false // For K8s
 var skipPipelineMount = false 
+var tenzirDisabled = false 
 
 var dockercli *dockerclient.Client
 var containerId string
@@ -684,7 +685,8 @@ func deployServiceWorkers(image string) {
 			serviceOptions,
 		)
 
-		//dockercli.ServiceUpdate(
+		// Force deploy if it's not disabled
+		deployTenzirNode()
 
 		if err == nil {
 			log.Printf("[DEBUG] Successfully deployed workers with %d replica(s) on %d node(s)", replicas, cnt)
@@ -2461,153 +2463,6 @@ func main() {
 	}
 }
 
-// func deployPipeline(image, identifier, command string) error {
-// 	if isKubernetes == "true" {
-// 		return errors.New("Kubernetes not implemented")
-// 	}
-
-// 	ctx := context.Background()
-// 	hostConfig := &container.HostConfig{
-// 		LogConfig: container.LogConfig{
-// 			Type: "json-file",
-// 			Config: map[string]string{
-// 				"max-size": "10m",
-// 			},
-// 		},
-// 		Resources: container.Resources{},
-// 	}
-
-// 	hostConfig.NetworkMode = container.NetworkMode(fmt.Sprintf("container:%s", containerId))
-// 	if strings.ToLower(cleanupEnv) != "false" {
-// 		hostConfig.AutoRemove = true
-// 	}
-
-// 	envVariables := []string{
-// 	}
-
-// 	// Add volume binds for storage
-// 	// Want read/write with full access for the container
-// 	//sourceFolder := "/Users/frikky/git/shuffle/shuffle-database"
-// 	//destinationFolder := "/tmp/storage"
-// 	//hostConfig.Mounts = append(hostConfig.Mounts, mount.Mount{
-// 	//	Type:   mount.TypeBind,
-// 	//	Source: sourceFolder,
-// 	//	Target: destinationFolder,
-// 	//})
-
-// 	// FIXME: Is using sigma "automatically" here good?
-// 	// Or is it better to run it as a separate workflow?
-// 	if strings.Contains(command, "sigma") {
-// 		log.Printf("[DEBUG] Should LOAD sigma from backend in realtime and dump it in a folder inside the container")
-
-// 		//sourceFolder := "/tmp/tenzir/sigma"
-// 		//sigmaFolder := "/tmp/tenzir/sigma"
-// 		//hostConfig.Mounts = append(hostConfig.Mounts, mount.Mount{
-// 		//	Type:   mount.TypeBind,
-// 		//	Source: sigmaFolder,
-// 		//	Target: sigmaFolder,
-// 		//}
-// 	}
-
-// 	config := &container.Config{
-// 		Image: image,
-// 		Env:   envVariables,
-// 		Cmd:   []string{
-// 			command,
-// 		},
-// 	}
-
-// 	// Add label to container in case of zombies
-// 	config.Labels = map[string]string{
-// 		"name":   identifier,
-// 		"shuffle": "shuffle",
-// 	}
-
-// 	cont, err := dockercli.ContainerCreate(
-// 		ctx,
-// 		config,
-// 		hostConfig,
-// 		nil,
-// 		nil,
-// 		identifier,
-// 	)
-
-// 	if err != nil {
-// 		if strings.Contains(fmt.Sprintf("%s", err), "Conflict. The container name ") {
-// 			log.Printf("[DEBUG] Pipeline Container %s already exists, removing it", identifier)
-// 		} else {
-// 			log.Printf("[ERROR] Failed to create pipeline container %s: %s", identifier, err)
-// 			return err
-// 		}
-// 	}
-
-// 	containerStartOptions := container.StartOptions{}
-// 	err = dockercli.ContainerStart(
-// 		ctx,
-// 		cont.ID,
-// 		containerStartOptions,
-// 	)
-// 	if err != nil {
-// 		if strings.Contains(fmt.Sprintf("%s", err), "cannot join network") || strings.Contains(fmt.Sprintf("%s", err), "No such container") {
-// 			hostConfig.NetworkMode = ""
-// 			cont, err = dockercli.ContainerCreate(
-// 				ctx,
-// 				config,
-// 				hostConfig,
-// 				nil,
-// 				nil,
-// 				identifier+"-2",
-// 			)
-// 			if err != nil {
-// 				log.Printf("[ERROR] Failed to CREATE pipeline container (2): %s", err)
-// 			}
-
-// 			err = dockercli.ContainerStart(
-// 				ctx,
-// 				cont.ID,
-// 				containerStartOptions,
-// 			)
-// 			if err != nil {
-// 				log.Printf("[ERROR] Failed to start pipeline container (2): %s", err)
-// 				return err
-// 			}
-// 		} else {
-// 			log.Printf("[ERROR] Failed initial pipeline container start. Quitting as this is NOT a simple network issue. Err: %s", err)
-// 		}
-
-// 		if err != nil {
-// 			log.Printf("[ERROR] Failed to start pipeline container in environment %s: %s", environment, err)
-// 			return err
-// 		} else {
-// 			log.Printf("[INFO] Pipeline Container created (1). Environment %s: docker logs %s", environment, cont.ID)
-// 		}
-
-// 		stats, err := dockercli.ContainerInspect(ctx, cont.ID)
-// 		if err != nil {
-// 			log.Printf("[ERROR] Failed checking pipeline with containername '%s'", cont.ID)
-// 			return nil
-// 		}
-
-// 		containerStatus := stats.ContainerJSONBase.State.Status
-// 		log.Printf("[DEBUG] Status of pipeline '%s' is %s. Should be running. Will reset", containerName, containerStatus)
-// 	}
-
-// 	// Wait for the container to finish
-// 	/*
-// 	statusCh, errCh := dockercli.ContainerWait(ctx, cont.ID, container.WaitConditionNotRunning)
-// 	select {
-// 	case err := <-errCh:
-// 		if err != nil {
-// 			log.Printf("[ERROR] Failed to wait for container: %s", err)
-// 		}
-// 	case <-statusCh:
-// 		log.Printf("[INFO] Container finished")
-// 	}
-// 	*/
-
-// 	return nil
-// }
-
 // Tenzir command samples
 // docker pull ghcr.io/dominiklohmann/tenzir-arm64:latest
 // docker tag ghcr.io/dominiklohmann/tenzir-arm64:latest tenzir/tenzir:latest
@@ -2911,6 +2766,13 @@ func createAndStartTenzirNode(ctx context.Context, containerName, imageName stri
 		},
 	}
 
+	if isKubernetes != "true" {
+		hostConfig.NetworkMode = container.NetworkMode(fmt.Sprintf("container:%s", containerId))
+		if strings.ToLower(cleanupEnv) != "false" {
+			hostConfig.AutoRemove = true
+		}
+	}
+
 	_, err := dockercli.ContainerCreate(ctx, config, hostConfig, networkingConfig, nil, containerName)
 	if err != nil {
 		if strings.Contains(err.Error(), "path does not exist") {
@@ -3530,12 +3392,19 @@ func sendPipelineHealthStatus() (shuffle.LakeConfig, error) {
 		pipelinePayload.Pipelines = pipelines
 	}
 
-	//err := checkTenzirNode()
+	if tenzirDisabled {
+		return pipelinePayload, nil
+	}
+
 	err := deployTenzirNode() 
 	if err != nil {
-		if (!strings.Contains(err.Error(), "SHUFFLE_SKIP_PIPELINES")) {
+		if (!strings.Contains(err.Error(), "SHUFFLE_SKIP_PIPELINES") && !strings.Contains(err.Error(), "Kubernetes not implemented for Tenzir node")) && !strings.Contains(err.Error(), "Tenzir Node is already running") && !strings.Contains(err.Error(), "docker daemon") {
+
 			log.Printf("[ERROR] Tenzir node connection problem: %s", err)
-		}	
+		} else {
+			tenzirDisabled = true
+			log.Printf("[ERROR] Disabling pipelines: %s. You will need to restart the Orborus to fix this.", err)
+		}
 
 		return pipelinePayload, err
 	}