diff --git a/functions/onprem/orborus/go.mod b/functions/onprem/orborus/go.mod index 5b25037b..0c7b28df 100644 --- a/functions/onprem/orborus/go.mod +++ b/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 ) diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index 8cf1d331..fb5847cc 100755 --- a/functions/onprem/orborus/orborus.go +++ b/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 } diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index a05de7fc..c166e7ae 100644 --- a/functions/onprem/worker/worker.go +++ b/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" {