Build new orborus \o/

functions/onprem/orborus/go.mod

@@ -10,7 +10,7 @@ require (
 	github.com/docker/docker v27.0.2+incompatible
 	github.com/docker/go-connections v0.5.0
 	github.com/satori/go.uuid v1.2.0
-	github.com/shuffle/shuffle-shared v0.6.82
+	github.com/shuffle/shuffle-shared v0.6.83
 	k8s.io/api v0.30.2
 	k8s.io/apimachinery v0.30.2
 )

functions/onprem/orborus/orborus.go

@@ -1,7 +1,7 @@
 package main
 
 /*
-	Orborus exists to listen for new workflow executions which are deployed as workers.
+	Orborus exists to listen for new jobs which are deployed as workers.
 */
 
 //  Potential issues:
@@ -2708,12 +2708,17 @@ func handlePipeline(incRequest shuffle.ExecutionRequest) error {
 }
 
 func deployTenzirNode() error {
+	if os.Getenv("SHUFFLE_SKIP_PIPELINES") == "true" {
+		return errors.New("Pipelines are disabled by user with SHUFFLE_SKIP_PIPELINES")
+	}
+
 	if isKubernetes == "true" {
 		return errors.New("Kubernetes not implemented for Tenzir node")
 	}
 
 	err := checkTenzirNode()
 	if err == nil {
+		log.Printf("[INFO] Tenzir Node is already running")
 		return nil
 	}
 
@@ -2851,16 +2856,18 @@ func createAndStartTenzirNode(ctx context.Context, containerName, imageName stri
 			tenzirStorageFolder = tenzirStorageFolder + "/"
 		}
 	} else {
-		tenzirStorageFolder = "/tmp/tenzir/"
+		tenzirStorageFolder = "/tmp/"
+		log.Printf("[DEBUG] Using folder %s for Tenzir storage. Change it using SHUFFLE_STORAGE_FOLDER", tenzirStorageFolder) 
 	}
 
 
 	if !anyFound {
-		log.Printf("[DEBUG] No Tenzir Plugin environment variables found.") 
+		//log.Printf("[DEBUG] No Tenzir Plugin environment variables found.") 
 	} else {
 		//log.Printf("[DEBUG] Attempting Tenzir connection with app.tenzir.com tenant '%s'", tenzirPluginsPlatform)
 	}
 
+
 	hostConfig := &container.HostConfig{
 		PortBindings: nat.PortMap{
 			"514/tcp":  []nat.PortBinding{{HostPort: "514"}},
@@ -2929,7 +2936,7 @@ func createAndStartTenzirNode(ctx context.Context, containerName, imageName stri
 	}
 
 	log.Printf("[INFO] Tenzir Node container started successfully. Waiting for it to become available..")
-	time.Sleep(10 * time.Second)
+	time.Sleep(20 * time.Second)
 	err = checkTenzirNode()
 	if err != nil {
 		log.Printf("[ERROR] Tenzir node is not available during deployment: %s", err)
@@ -2985,6 +2992,10 @@ func createNetworkIfNotExists(ctx context.Context, networkName, subnet, gateway
 }
 
 func checkTenzirNode() error {
+	if os.Getenv("SHUFFLE_SKIP_PIPELINES") == "true" {
+		return errors.New("Pipelines are disabled by user with SHUFFLE_SKIP_PIPELINES")
+	}
+
 	retries := 1
 
 	//retryInterval := 3 * time.Second
@@ -3519,9 +3530,13 @@ func sendPipelineHealthStatus() (shuffle.LakeConfig, error) {
 		pipelinePayload.Pipelines = pipelines
 	}
 
-	//err := deployTenzirNode() 
+	//err := checkTenzirNode()
-	err := checkTenzirNode()
+	err := deployTenzirNode() 
 	if err != nil {
+		if (!strings.Contains(err.Error(), "SHUFFLE_SKIP_PIPELINES")) {
+			log.Printf("[ERROR] Tenzir node connection problem: %s", err)
+		}	
+
 		return pipelinePayload, err
 	}
 

functions/onprem/worker/worker.go

@@ -4074,9 +4074,14 @@ func runWebserver(listener net.Listener) {
 	r := mux.NewRouter()
 	r.HandleFunc("/api/v1/streams", handleWorkflowQueue).Methods("POST", "OPTIONS")
 	r.HandleFunc("/api/v1/streams/results", handleGetStreamResults).Methods("POST", "OPTIONS")
+	r.HandleFunc("/api/v1/download", handleDownloadImage).Methods("POST", "OPTIONS")
+
+	// Synonyms. Require an execution ID + auth + shuffle backend
 	r.HandleFunc("/api/v1/execute", handleRunExecution).Methods("POST", "OPTIONS")
 	r.HandleFunc("/api/v1/run", handleRunExecution).Methods("POST", "OPTIONS")
-	r.HandleFunc("/api/v1/download", handleDownloadImage).Methods("POST", "OPTIONS")
+
+	// What would be require to run a workflow otherwise?
+	// Maybe directly /workflow/run
 
 	/*** STARTREMOVE ***/
 	if os.Getenv("SHUFFLE_SWARM_CONFIG") == "run" || os.Getenv("SHUFFLE_SWARM_CONFIG") == "swarm" {