Merge branch '2.0.0' of github.com:Shuffle/Shuffle into 2.0.0

backend/go-app/docker.go

@@ -211,6 +211,7 @@ func fixTags(tags []string) []string {
 func buildImageMemory(fs billy.Filesystem, tags []string, dockerfileFolder string, downloadIfFail bool) error {
 	ctx := context.Background()
 	client, err := client.NewEnvClient()
+        defer client.Close()
 	if err != nil {
 		log.Printf("Unable to create docker client: %s", err)
 		return err
@@ -476,6 +477,7 @@ func buildImage(tags []string, dockerfileLocation string) error {
 
 		ctx := context.Background()
 		client, err := client.NewEnvClient()
+                defer client.Close()
 		if err != nil {
 			log.Printf("Unable to create docker client: %s", err)
 			return err
@@ -832,14 +834,23 @@ func handleRemoteDownloadApp(resp http.ResponseWriter, ctx context.Context, user
 		type tmpapp struct {
 			Success bool   `json:"success"`
 			OpenAPI string `json:"openapi"`
+			App 	string `json:"app"`
 		}
 
 		app := tmpapp{}
 		err := json.Unmarshal(respBody, &app)
 		if err != nil || app.Success == false || len(app.OpenAPI) == 0 {
 			log.Printf("[ERROR] Failed app unmarshal during auto-download. Success: %#v. Applength: %d: %s", app.Success, len(app.OpenAPI), err)
+
 			resp.WriteHeader(401)
+			if len(app.App) > 0 {
+				resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Not an OpenAPI app, but a Python app. Please download the app using the Remote Download system: https://shuffler.io/docs/apps#importing-remote-apps"}`)))
+			} else {
 				resp.Write([]byte(`{"success": false, "reason": "App doesn't exist"}`))
+			}
+
+			resp.Write([]byte(`{"success": false, "reason": "App doesn't exist"}`))
+
 			return
 		}
 

backend/go-app/go.mod

@@ -2,7 +2,7 @@ module shuffle
 
 go 1.22.0
 
-//replace github.com/shuffle/shuffle-shared => ../../../shuffle-shared
+replace github.com/shuffle/shuffle-shared => ../../../shuffle-shared
 
 toolchain go1.22.2
 

backend/go-app/walkoff.go

@@ -333,7 +333,7 @@ func handleGetWorkflowqueue(resp http.ResponseWriter, request *http.Request) {
 			}
 
 			env.Checkin = timeNow
-			err = shuffle.SetEnvironment(ctx, &env)
+			err = shuffle.SetEnvironment(ctx, env)
 			if err != nil {
 				log.Printf("[ERROR] Failed updating environment: %s", err)
 			}
@@ -668,7 +668,8 @@ func handleGetStreamResults(resp http.ResponseWriter, request *http.Request) {
 			Sharing: 		workflowExecution.Workflow.Sharing,
 			Description:    workflowExecution.Workflow.Description,
 			InputQuestions: workflowExecution.Workflow.InputQuestions,
-			InputMarkdown:  workflowExecution.Workflow.InputMarkdown,
+
+			FormControl:	workflowExecution.Workflow.FormControl,
 		}
 
 		workflowExecution.Results = []shuffle.ActionResult{}

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.79
+	github.com/shuffle/shuffle-shared v0.6.83
 	k8s.io/api v0.30.2
 	k8s.io/apimachinery v0.30.2
 )

functions/onprem/orborus/go.sum

@@ -301,6 +301,10 @@ github.com/sendgrid/sendgrid-go v3.14.0+incompatible h1:KDSasSTktAqMJCYClHVE94Fc
 github.com/sendgrid/sendgrid-go v3.14.0+incompatible/go.mod h1:QRQt+LX/NmgVEvmdRw0VT/QgUn499+iza2FnDca9fg8=
 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/shuffle/shuffle-shared v0.6.79 h1:MIy5kcShHYN05ov/50YJ+la1C2v1rL8IENapOvX9I8U=
 github.com/shuffle/shuffle-shared v0.6.79/go.mod h1:RAJiSFjmuKmijKTbbEf9A6Ojb+3/te7g71lED7JjPus=
 github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=

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:
@@ -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
@@ -324,7 +325,12 @@ func cleanupExistingNodes(ctx context.Context) error {
 
 func deployServiceWorkers(image string) {
 	log.Printf("[DEBUG] Validating deployment of workers as services IF swarmConfig = run (value: %#v)", swarmConfig)
-	if swarmConfig == "run" || swarmConfig == "swarm" {
+	if swarmConfig != "run" && swarmConfig != "swarm" {
+		log.Printf("[DEBUG] Skipping deployment of workers as services as swarmConfig is not set to run or swarm. Value: %#v", swarmConfig)
+		return
+	}
+
+
 	ctx := context.Background()
 	// Looks for and cleans up all existing items in swarm we can't re-use (Shuffle only)
 
@@ -684,7 +690,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)
@@ -710,8 +717,6 @@ func deployServiceWorkers(image string) {
 			}
 		}
 	}
-
-	}
 }
 
 // Deploys the worker with the current available environments
@@ -1908,7 +1913,7 @@ func main() {
 		pipelineUrl = "http://localhost:5160"
 
 		// Find the IP in baseUrl. Base format is http://<ip>:<port>
-		if baseUrl != "" {
+		if baseUrl != "" && !strings.Contains(baseUrl, "shuffle") && !strings.Contains(baseUrl, "localhost") && !strings.Contains(baseUrl, "run.app") {
 			urlSplit := strings.Split(baseUrl, "://")
 			if len(urlSplit) > 1 {
 				// Find the IP
@@ -1919,6 +1924,10 @@ func main() {
 			}
 		}
 
+		if len(containerId) > 0 {
+			pipelineUrl = "http://tenzir-node:5160"
+		} 
+
 		log.Printf("[WARNING] SHUFFLE_PIPELINE_URL not set, falling back to default URL: %s. If BASE_URL is set, we use the external IP for that", pipelineUrl)
 		os.Setenv("SHUFFLE_PIPELINE_URL", pipelineUrl)
 	}
@@ -2461,153 +2470,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
@@ -2708,12 +2570,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
 	}
 
@@ -2745,7 +2612,7 @@ func deployTenzirNode() error {
 			// Check if image exists
 			_, _, err := dockercli.ImageInspectWithRaw(ctx, imageName)
 			if dockerclient.IsErrNotFound(err) {
-				log.Printf("[DEBUG] Pulling image %s", imageName)
+				log.Printf("[DEBUG] Pulling image %s. This may take a while.", imageName)
 				pullOptions := image.PullOptions{}
 				out, err := dockercli.ImagePull(ctx, imageName, pullOptions)
 				if err != nil {
@@ -2811,6 +2678,7 @@ func createAndStartTenzirNode(ctx context.Context, containerName, imageName stri
 
 	// Ensure restart policy is there
 	config := &container.Config{
+		Hostname:     containerName,
 		Cmd:          []string{"--commands=web server --mode=dev --bind=0.0.0.0"},
 		Image:        imageName,
 		Healthcheck:  healthconfig,
@@ -2851,16 +2719,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"}},
@@ -2904,10 +2774,14 @@ func createAndStartTenzirNode(ctx context.Context, containerName, imageName stri
 		},
 	}
 
+	if isKubernetes != "true" {
+		hostConfig.NetworkMode = container.NetworkMode(fmt.Sprintf("container:%s", containerId))
+	}
+
 	_, err := dockercli.ContainerCreate(ctx, config, hostConfig, networkingConfig, nil, containerName)
 	if err != nil {
 		if strings.Contains(err.Error(), "path does not exist") {
-			log.Printf("[ERROR] Not using permanent pipeline storage as storage folder /opt/tenzir/ does not exist. If you want permanent storage, create the /opt/tendir/ folder then restart Orborus. Raw: %s", err)
+			log.Printf("[ERROR] Not using permanent pipeline storage as storage folder %s does not exist. If you want permanent storage, create the %s folder then restart Orborus (1). Raw: %s", tenzirStorageFolder, tenzirStorageFolder, err)
 			skipPipelineMount = true
 		} else {
 			log.Printf("[ERROR] Failed to create Tenzir Node container: %v", err)
@@ -2918,12 +2792,18 @@ func createAndStartTenzirNode(ctx context.Context, containerName, imageName stri
 
 	err = dockercli.ContainerStart(ctx, containerName, containerStartOptions)
 	if err != nil {
-		log.Printf("[ERROR] Failed to start Tenzir Node container: %v", err)
+		if strings.Contains(err.Error(), "path does not exist") {
+			log.Printf("[ERROR] Not using permanent pipeline storage as storage folder %s does not exist. If you want permanent storage, create the %s folder then restart Orborus (2). Raw: %s", tenzirStorageFolder, tenzirStorageFolder, err)
+			skipPipelineMount = true
+		} else {
+			log.Printf("[ERROR] Failed to START Tenzir Node container: %v", err)
+		}
+
 		return err
 	}
 
 	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)
@@ -2979,9 +2859,10 @@ func createNetworkIfNotExists(ctx context.Context, networkName, subnet, gateway
 }
 
 func checkTenzirNode() error {
-	retries := 1
+	if os.Getenv("SHUFFLE_SKIP_PIPELINES") == "true" {
+		return errors.New("Pipelines are disabled by user with SHUFFLE_SKIP_PIPELINES")
+	}
 
-	//retryInterval := 3 * time.Second
 	url := fmt.Sprintf("%s/api/v0/ping", pipelineUrl)
 	forwardMethod := "POST"
 
@@ -2994,16 +2875,14 @@ func checkTenzirNode() error {
 		return err
 	}
 
-	for i := 0; i < retries; i++ {
 	resp, err := client.Do(req)
 	if err == nil && resp.StatusCode == http.StatusOK {
 		return nil
 	}
 
-		//time.Sleep(retryInterval)
+	log.Printf("[DEBUG] Failed to verify Tenzir node on %s: %s", url, err)
-	}
 
-	return fmt.Errorf("tenzir node is not available")
+	return fmt.Errorf("Tenzir node is not available")
 }
 
 func createPipeline(command, identifier string) (string, error) {
@@ -3513,10 +3392,20 @@ func sendPipelineHealthStatus() (shuffle.LakeConfig, error) {
 		pipelinePayload.Pipelines = pipelines
 	}
 
-	//url := fmt.Sprintf("%s/api/v1/detections/siem/health", baseUrl)
+	if tenzirDisabled {
-	//err := checkTenzirNode()
+		return pipelinePayload, nil
+	}
+
 	err := deployTenzirNode() 
 	if err != nil {
+		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
 	}
 

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" {