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

backend/go-app/docker.go

@@ -349,7 +349,7 @@ func deleteJob(client *kubernetes.Clientset, jobName, namespace string) error {
 	})
 }
 
-func buildImage(tags []string, dockerfileFolder string) error {
+func buildImage(tags []string, dockerfileLocation string) error {
 
 	isKubernetes := false
 	if os.Getenv("IS_KUBERNETES") == "true" {
@@ -369,10 +369,8 @@ func buildImage(tags []string, dockerfileFolder string) error {
 
 		log.Printf("[INFO] registry name: %s", registryName)
 
-		contextDir := strings.Replace(dockerfileFolder, "Dockerfile", "", -1)
+		contextDir := filepath.Join("/app/", filepath.Dir(dockerfileLocation))
-		contextDir = "/app/" + contextDir
 		log.Print("contextDir: ", contextDir)
-		dockerFile := "./Dockerfile"
 
 		client, err := getK8sClient()
 		if err != nil {
@@ -407,7 +405,7 @@ func buildImage(tags []string, dockerfileFolder string) error {
 								Image: "gcr.io/kaniko-project/executor:latest",
 								Args: []string{
 									"--verbosity=debug",
-									"--dockerfile=" + dockerFile,
+									"--dockerfile=Dockerfile",
 									"--context=dir://" + contextDir,
 									"--skip-tls-verify",
 									"--destination=" + registryName + "/" + tags[1],
@@ -420,9 +418,7 @@ func buildImage(tags []string, dockerfileFolder string) error {
 								},
 							},
 						},
-						NodeSelector: map[string]string{
+						NodeName:      backendNodeName,
-							"node": backendNodeName,
-						},
 						RestartPolicy: corev1.RestartPolicyNever,
 						Volumes: []corev1.Volume{
 							{
@@ -486,7 +482,7 @@ func buildImage(tags []string, dockerfileFolder string) error {
 		}
 
 		log.Printf("[INFO] Docker Tags: %s", tags)
-		dockerfileSplit := strings.Split(dockerfileFolder, "/")
+		dockerfileSplit := strings.Split(dockerfileLocation, "/")
 
 		// Create a buffer
 		buf := new(bytes.Buffer)

backend/go-app/go.sum

@@ -334,10 +334,8 @@ github.com/sendgrid/sendgrid-go v3.14.0+incompatible/go.mod h1:QRQt+LX/NmgVEvmdR
 github.com/sergi/go-diff v1.0.0/go.mod h1:0CfEIISq7TuYL3j771MWULgwwjU+GofnZX9QAmXWZgo=
 github.com/sergi/go-diff v1.3.2-0.20230802210424-5b0b94c5c0d3 h1:n661drycOFuPLCN3Uc8sB6B/s6Z4t2xvBgU1htSHuq8=
 github.com/sergi/go-diff v1.3.2-0.20230802210424-5b0b94c5c0d3/go.mod h1:A0bzQcvG0E7Rwjx0REVgAGH58e96+X0MeOfepqsbeW4=
-github.com/shuffle/shuffle-shared v0.6.74 h1:os3BDSFZnl4U8ZgsTAY8IsTDADcMXhbc1rS9UMa0BIY=
+github.com/shuffle/shuffle-shared v0.6.77 h1:KKtM50xW2DLuRHINxhp3uXrNH0AhiwkeiiU93a8fB3A=
-github.com/shuffle/shuffle-shared v0.6.74/go.mod h1:RAJiSFjmuKmijKTbbEf9A6Ojb+3/te7g71lED7JjPus=
+github.com/shuffle/shuffle-shared v0.6.77/go.mod h1:RAJiSFjmuKmijKTbbEf9A6Ojb+3/te7g71lED7JjPus=
-github.com/shuffle/shuffle-shared v0.6.78 h1:INWlC0bzPqXLTGEGz2Id3pXXp4RPbUpWgqgbCBbGJ4A=
-github.com/shuffle/shuffle-shared v0.6.78/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=

backend/go-app/main.go

@@ -5112,7 +5112,8 @@ func initHandlers() {
 	r.HandleFunc("/api/v1/apps/{appId}", shuffle.UpdateWorkflowAppConfig).Methods("PATCH", "OPTIONS")
 	r.HandleFunc("/api/v1/apps/{appId}", shuffle.DeleteWorkflowApp).Methods("DELETE", "OPTIONS")
 	r.HandleFunc("/api/v1/apps/{appId}/config", shuffle.GetWorkflowAppConfig).Methods("GET", "OPTIONS")
-	r.HandleFunc("/api/v1/apps/run_hotload", handleAppHotloadRequest).Methods("GET", "OPTIONS")
+	r.HandleFunc("/api/v1/apps/run_hotload", handleAppHotloadRequest).Methods("GET", "POST", "OPTIONS")
+	r.HandleFunc("/api/v1/apps/{appName}/run_hotload", handleSingleAppHotloadRequest).Methods("POST", "OPTIONS")
 	r.HandleFunc("/api/v1/apps/get_existing", LoadSpecificApps).Methods("POST", "OPTIONS")
 	r.HandleFunc("/api/v1/apps/download_remote", LoadSpecificApps).Methods("POST", "OPTIONS")
 	r.HandleFunc("/api/v1/apps/validate", validateAppInput).Methods("POST", "OPTIONS")

backend/go-app/walkoff.go

@@ -291,34 +291,49 @@ func handleGetWorkflowqueue(resp http.ResponseWriter, request *http.Request) {
 	ctx := shuffle.GetContext(request)
 	env, err := shuffle.GetEnvironment(ctx, orgId, "")
 	timeNow := time.Now().Unix()
-	if err == nil && len(env.Id) > 0 && len(env.Name) > 0 {
+	if err == nil && len(env.Id) > 0 && len(env.Name) > 0 && request.Method == "POST" { 
 		// Updates every 60 seconds~
 		if time.Now().Unix() > env.Edited+60 {
 			env.RunningIp = shuffle.GetRequestIp(request)
+
+			// Orborus label = custom label for Orborus
 			if len(orborusLabel) > 0 {
 				env.RunningIp = orborusLabel
 			}
 
-			if request.Method == "POST" {
+			// Set the checkin cache
-				body, err := ioutil.ReadAll(request.Body)
-				if err == nil {
-					var envData shuffle.OrborusStats
-					err = json.Unmarshal(body, &envData)
-					if err == nil {
-						if envData.Swarm {
-							env.Licensed = true
-							env.RunType = "docker"
-						}
 
-						if envData.Kubernetes {
+
-							env.RunType = "k8s"
+			body, err := ioutil.ReadAll(request.Body)
-						}
+			if err == nil {
+				var envData shuffle.OrborusStats
+				err = json.Unmarshal(body, &envData)
+				if err == nil {
+					envData.RunningIp = env.RunningIp
+
+					marshalled, err := json.Marshal(envData)
+					if err == nil {
+						cacheKey := fmt.Sprintf("queueconfig-%s-%s", env.Name, env.OrgId)
+						go shuffle.SetCache(context.Background(), cacheKey, marshalled, 2)
 					}
+
+
+
+					if envData.Swarm {
+						env.Licensed = true
+						env.RunType = "docker"
+					} 
+
+					if envData.Kubernetes {
+						env.RunType = "k8s"
+					}
+
+					envData.DataLake = env.DataLake
 				}
 			}
 
 			env.Checkin = timeNow
-			err = shuffle.SetEnvironment(ctx, env)
+			err = shuffle.SetEnvironment(ctx, &env)
 			if err != nil {
 				log.Printf("[ERROR] Failed updating environment: %s", err)
 			}
@@ -2802,6 +2817,70 @@ func loadSpecificWorkflows(resp http.ResponseWriter, request *http.Request) {
 	resp.Write([]byte(fmt.Sprintf(`{"success": true}`)))
 }
 
+func handleSingleAppHotloadRequest(resp http.ResponseWriter, request *http.Request) {
+	cors := shuffle.HandleCors(resp, request)
+	if cors {
+		return
+	}
+	ctx := context.Background()
+	cacheKey := fmt.Sprintf("workflowapps-sorted-1000")
+	shuffle.DeleteCache(ctx, cacheKey)
+	cacheKey = fmt.Sprintf("workflowapps-sorted-500")
+	shuffle.DeleteCache(ctx, cacheKey)
+	cacheKey = fmt.Sprintf("workflowapps-sorted-0")
+	shuffle.DeleteCache(ctx, cacheKey)
+	// Just need to be logged in
+	// FIXME - should have some permissions?
+	user, err := shuffle.HandleApiAuthentication(resp, request)
+	if err != nil {
+		log.Printf("Api authentication failed in app hotload: %s", err)
+		resp.WriteHeader(401)
+		resp.Write([]byte(`{"success": false}`))
+		return
+	}
+	if user.Role != "admin" {
+		resp.WriteHeader(401)
+		resp.Write([]byte(`{"success": false, "reason": "Must be admin to hotload apps"}`))
+		return
+	}
+	location := os.Getenv("SHUFFLE_APP_HOTLOAD_FOLDER")
+	if len(location) == 0 {
+		resp.WriteHeader(500)
+		resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "SHUFFLE_APP_HOTLOAD_FOLDER not specified in .env"}`)))
+		return
+	}
+	requestUrlFields := strings.Split(request.URL.String(), "/")
+	var appName string
+	if requestUrlFields[1] == "api" {
+		if len(requestUrlFields) <= 4 {
+			resp.WriteHeader(401)
+			resp.Write([]byte(`{"success": false}`))
+			return
+		}
+		appName = requestUrlFields[4]
+		if strings.Contains(appName, "?") {
+			appName = strings.Split(appName, "?")[0]
+		}
+	}
+	location = location + "/" + appName
+	log.Printf("[INFO] Starting hotloading from %s", location)
+	err = handleAppHotload(ctx, location, true)
+	if err != nil {
+		log.Printf("[WARNING] Failed app hotload: %s", err)
+		resp.WriteHeader(500)
+		resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "%s"}`, err)))
+		return
+	}
+	cacheKey = fmt.Sprintf("workflowapps-sorted-100")
+	shuffle.DeleteCache(ctx, cacheKey)
+	cacheKey = fmt.Sprintf("workflowapps-sorted-500")
+	shuffle.DeleteCache(ctx, cacheKey)
+	cacheKey = fmt.Sprintf("workflowapps-sorted-1000")
+	shuffle.DeleteCache(ctx, cacheKey)
+	resp.WriteHeader(200)
+	resp.Write([]byte(fmt.Sprintf(`{"success": true}`)))
+}
+
 func handleAppHotloadRequest(resp http.ResponseWriter, request *http.Request) {
 	cors := shuffle.HandleCors(resp, request)
 	if cors {

functions/onprem/orborus/go.mod

@@ -4,7 +4,7 @@ go 1.22.0
 
 toolchain go1.22.2
 
-//replace github.com/shuffle/shuffle-shared => ../../../../shuffle-shared
+replace github.com/shuffle/shuffle-shared => ../../../../shuffle-shared
 
 require (
 	github.com/docker/docker v27.0.2+incompatible

functions/onprem/orborus/orborus.go

@@ -30,8 +30,8 @@ import (
 	"strings"
 	"sync"
 	"time"
-	"math/rand"
 
+	"math/rand"
 	//"os/signal"
 	//"syscall"
 
@@ -103,12 +103,14 @@ var swarmNetworkName = os.Getenv("SHUFFLE_SWARM_NETWORK_NAME")
 var orborusLabel = os.Getenv("SHUFFLE_ORBORUS_LABEL")
 var memcached = os.Getenv("SHUFFLE_MEMCACHED")
 
-// For it to download from Sigma? 
+// For it to download from Sigma?
-var apiKey = os.Getenv("AUTH_FOR_ORBORUS") 
+var apiKey = os.Getenv("AUTH_FOR_ORBORUS")
 var pipelineUrl = os.Getenv("SHUFFLE_PIPELINE_URL")
 
 var executionIds = []string{}
+var pipelines = []shuffle.PipelineInfoMini{}
 var namespacemade = false // For K8s
+var skipPipelineMount = false 
 
 var dockercli *dockerclient.Client
 var containerId string
@@ -725,7 +727,6 @@ func buildEnvVars(envMap map[string]string) []corev1.EnvVar {
 
 func handleBackendImageDownload(ctx context.Context, images string) error {
 
-
 	// Replicate images with lowercase, as the name may be wrong
 	// Most of the time lowercase is correct. Swapping to have that first
 	originalImages := images
@@ -756,7 +757,7 @@ func handleBackendImageDownload(ctx context.Context, images string) error {
 		resp, err := dockercli.ImageRemove(ctx, image, removeOptions)
 		if err != nil {
 			log.Printf("[ERROR] Failed removing image: %s. Resp: %#v", err, resp)
-		
+
 			// Goroutining images that don't already exist, as they are most likely not the correct one
 			go shuffle.DownloadDockerImageBackend(&http.Client{Timeout: imagedownloadTimeout}, image)
 		} else {
@@ -806,10 +807,10 @@ func handleBackendImageDownload(ctx context.Context, images string) error {
 					//docker service update --image username/imagename:latest servicename --force
 					serviceUpdateOptions := types.ServiceUpdateOptions{}
 					resp, err := dockercli.ServiceUpdate(
-						ctx, 
+						ctx,
-						service.ID, 
+						service.ID,
-						service.Version, 
+						service.Version,
-						service.Spec, 
+						service.Spec,
 						serviceUpdateOptions,
 					)
 
@@ -824,7 +825,7 @@ func handleBackendImageDownload(ctx context.Context, images string) error {
 					}
 				}
 			}
-		
+
 		}
 
 	}
@@ -1522,7 +1523,6 @@ func checkSwarmService(ctx context.Context) {
 	log.Printf("[DEBUG] Swarm info: %s\n\n", ret)
 }
 
-
 func getContainerResourceUsage(ctx context.Context, cli *dockerclient.Client, containerID string) (float64, float64, error) {
 	// Get container stats
 	stats, err := cli.ContainerStats(ctx, containerID, false)
@@ -1916,6 +1916,7 @@ func main() {
 		}
 
 		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)
 	}
 
 	// FIXME - during init, BUILD and/or LOAD worker and app_sdk
@@ -2019,8 +2020,6 @@ func main() {
 	log.Printf("[INFO] Waiting for executions at %s with Environment %#v", fullUrl, environment)
 	hasStarted := false
 	for {
-		_ = sendTenzirHealthStatus()
-
 		if req.Method == "POST" {
 			// Should find data to send (memory etc.)
 
@@ -2030,6 +2029,14 @@ func main() {
 
 			// Marshal and set body
 			orborusStats := getOrborusStats(ctx)
+			pipelinePayload, pipelineerr := sendPipelineHealthStatus()
+
+			if pipelineerr != nil {
+				// Too verbose to be enabled.
+				//log.Printf("[ERROR] Failed sending pipeline health status: %s", pipelineerr)
+			}
+
+			orborusStats.DataLake = pipelinePayload
 			jsonData, err := json.Marshal(orborusStats)
 			if err == nil {
 				req.Body = ioutil.NopCloser(bytes.NewBuffer(jsonData))
@@ -2115,23 +2122,52 @@ func main() {
 		var toBeRemoved shuffle.ExecutionRequestWrapper
 		if len(executionRequests.Data) > 0 {
 			newrequests := []shuffle.ExecutionRequest{}
+
+			// Deduplicating in case same job shows up multiple times
+			// This is specifically to handle data pipelines better
+			deduplicatedJobs := []shuffle.ExecutionRequest{}
 			for _, incRequest := range executionRequests.Data {
+				if !strings.Contains(incRequest.Type, "DOCKER") && !strings.Contains(incRequest.Type, "PIPELINE") && !strings.Contains(incRequest.Type, "SIGMA") && !strings.Contains(incRequest.Type, "TENZIR") {
+					deduplicatedJobs = append(deduplicatedJobs, incRequest)
+					continue
+				}
+
+				found := false
+				for _, dedupRequest := range deduplicatedJobs {
+					if incRequest.ExecutionArgument == dedupRequest.ExecutionArgument && incRequest.Type == dedupRequest.Type {
+						found = true
+						break
+					}
+				}
+
+				if found {
+					toBeRemoved.Data = append(toBeRemoved.Data, incRequest)
+					continue
+				}
+
+				deduplicatedJobs = append(deduplicatedJobs, incRequest)
+			}
+
+			executionRequests.Data = deduplicatedJobs
+			for _, incRequest := range executionRequests.Data {
+
 				// Looking for specific jobs
 				if incRequest.Type == "PIPELINE_CREATE" || incRequest.Type == "PIPELINE_START" || incRequest.Type == "PIPELINE_STOP" || incRequest.Type == "PIPELINE_DELETE" {
 
 					err := handlePipeline(incRequest)
 					if err != nil {
-						log.Printf("[ERROR] Failed handling pipeline: %s", err)
+
-					} else {
+						log.Printf("[ERROR] Failed handling pipeline (%s %s): %s. Deleting job anyway.", incRequest.Type, incRequest.ExecutionSource, err)
-						toBeRemoved.Data = append(toBeRemoved.Data, incRequest)
 					}
+			
+					toBeRemoved.Data = append(toBeRemoved.Data, incRequest)
 				} else if incRequest.Type == "DOCKER_IMAGE_DOWNLOAD" {
 					log.Printf("[INFO] Should delete -> download new images: %#v", incRequest.ExecutionArgument)
 
 					if len(incRequest.ExecutionArgument) > 0 {
 						// FIXME: Wait X seconds before running this as the image build may not be done yet. This is shitty, but may be ok to do in Orborus. Easy fix for the future: Just let it run through jobs 5-10 times before actually picking it up
 
-						// Run after 25 seconds in the goroutine instead 
+						// Run after 25 seconds in the goroutine instead
 						go handleBackendImageDownload(ctx, incRequest.ExecutionArgument)
 					} else {
 						log.Printf("[ERROR] No image name provided for download. Removing job from queue.")
@@ -2143,7 +2179,7 @@ func main() {
 
 					err := deployTenzirNode()
 					if err != nil {
-						log.Printf("[ERROR] Failed to deploy CATEGORY UPDATE, reason: %s", err)
+						log.Printf("[ERROR] Failed to run CATEGORY UPDATE, reason: %s", err)
 					} else {
 						continue
 					}
@@ -2159,7 +2195,7 @@ func main() {
 					fileName := incRequest.ExecutionArgument
 					err := deployTenzirNode()
 					if err != nil {
-						log.Printf("[ERROR] Failed to deploy DISABLE SIGMA FILE, reason: %s", err)
+						log.Printf("[ERROR] Failed to run DISABLE SIGMA FILE, reason: %s", err)
 					} else {
 						continue
 					}
@@ -2175,11 +2211,11 @@ func main() {
 					fileName := incRequest.ExecutionArgument
 					err := deployTenzirNode()
 					if err != nil {
-						log.Printf("[ERROR] Failed to deploy ENABLE SIGMA FILE, reason: %s", err)
+						log.Printf("[ERROR] Failed to run ENABLE SIGMA FILE, reason: %s", err)
 					} else {
 						continue
 					}
-					
+
 					err = enableRule(fileName)
 					if err != nil {
 						log.Printf("[ERROR] Failed to disable the sigma file %s, reason: %s", fileName, err)
@@ -2190,7 +2226,7 @@ func main() {
 				} else if incRequest.Type == "DISABLE_SIGMA_FOLDER" {
 					err := deployTenzirNode()
 					if err != nil {
-						log.Printf("[ERROR] Failed to deploy DISABLE SIGMA FOLDER, reason: %s", err)
+						log.Printf("[ERROR] Failed to run DISABLE SIGMA FOLDER, reason: %s", err)
 					}
 
 					err = removeAllFiles()
@@ -2201,9 +2237,28 @@ func main() {
 					}
 				} else if incRequest.Type == "START_TENZIR" {
 					log.Printf("[INFO] Got job to start tenzir")
+
 					err := deployTenzirNode()
 					if err != nil {
-						log.Printf("[ERROR] Failed to deploy the pipeline, reason: %s", err)
+						if strings.Contains(fmt.Sprintf("%s", err), "node available") {
+							toBeRemoved.Data = append(toBeRemoved.Data, incRequest)
+						} else {
+							log.Printf("[ERROR] Failed to start tenzir, reason: %s", err)
+							err = shuffle.CreateOrgNotification(
+								ctx,
+								fmt.Sprintf("Failed to start Tenzir: %s", err),
+								fmt.Sprintf("Tenzir failed to start due to: %s", err),
+								fmt.Sprintf("/detections/Sigma"),
+								org,
+								true,
+							)
+
+							if err != nil {
+								log.Printf("[ERROR] Failed to send notification: %s", err)
+								return
+							}
+						}
+
 					} else {
 						toBeRemoved.Data = append(toBeRemoved.Data, incRequest)
 					}
@@ -2557,6 +2612,8 @@ func main() {
 // docker run tenzir/tenzir:latest 'from http://192.168.86.44:5002/api/v1/orgs/7e9b9007-5df2-4b47-bca5-c4d267ef2943/cache/CIDR%20ranges?type=text&authorization=cec9d01f-09b2-4419-8a0a-76c6046e3fef read lines | to http://192.168.86.44:5002/api/v1/hooks/webhook_665ace5f-f27b-496a-a365-6e07eb61078c write lines'
 func handlePipeline(incRequest shuffle.ExecutionRequest) error {
 
+	log.Printf("[INFO] Pipeline: %s to %s", incRequest.Type, incRequest.ExecutionSource)
+
 	err := deployTenzirNode()
 	if err != nil {
 		log.Printf("[ERROR] Failed to deploy the pipeline, reason: %s", err)
@@ -2569,10 +2626,13 @@ func handlePipeline(incRequest shuffle.ExecutionRequest) error {
 
 		return errors.New("no execution argument found for pipeline create. Skipping")
 	}
-	//image := "tenzir/tenzir:latest"
-	identifier := fmt.Sprintf("shuffle-%s", strings.ToLower(strings.ReplaceAll(incRequest.ExecutionSource, " ", "-")))
-	command := incRequest.ExecutionArgument
 
+	identifier := strings.ToLower(strings.ReplaceAll(incRequest.ExecutionSource, " ", "-"))
+	if !strings.HasPrefix(strings.ToLower(incRequest.ExecutionSource), "shuffle") {
+		identifier = fmt.Sprintf("shuffle-%s", strings.ToLower(strings.ReplaceAll(incRequest.ExecutionSource, " ", "-")))
+	}
+
+	command := incRequest.ExecutionArgument
 	if incRequest.Type == "PIPELINE_CREATE" {
 		log.Printf("[INFO] Should delete -> recreate new pipeline with id %#v", identifier)
 		//err := deployPipeline(image, identifier, command)
@@ -2581,23 +2641,28 @@ func handlePipeline(incRequest shuffle.ExecutionRequest) error {
 			log.Printf("[ERROR] Failed to create pipeline: %s", err)
 			return err
 		}
-	} else if incRequest.Type == "PIPELINE_DELETE" {
+	} else if incRequest.Type == "PIPELINE_DELETE" || incRequest.Type == "PIPELINE_STOP" { {
 		log.Printf("[INFO] Should delete pipeline %#v", identifier)
 		pipelineId, err := searchPipeline(identifier)
 		if err != nil {
 			log.Printf("[ERROR] Failed searching for Pipeline with name %s reason:%s ", identifier, err)
 			return err
 		}
+
 		err = deletePipeline(pipelineId)
 		if err != nil {
 			log.Printf("[ERROR] Failed Deleting Pipeline %s", err)
 			return err
 		}
+	}
+		
+	/*
 	} else if incRequest.Type == "PIPELINE_STOP" {
 		log.Printf("[INFO] Should stop the pipeline %#v", identifier)
 		pipelineId, err := searchPipeline(identifier)
 		if err != nil {
 			log.Printf("[ERROR] Failed searching for Pipeline with name %s reason:%s ", identifier, err)
+			toBeRemoved.Data = append(toBeRemoved.Data, incRequest)
 			return err
 		}
 		_, err = updatePipelineState(command, pipelineId, "stop")
@@ -2605,18 +2670,20 @@ func handlePipeline(incRequest shuffle.ExecutionRequest) error {
 			log.Printf("[ERROR] Failed to stop Pipeline: %s reason:%s ", pipelineId, err)
 			return err
 		} else {
-			log.Printf("[INFO] successfully stopped the Pipeline: %s", pipelineId)
+			log.Printf("[INFO] Successfully stopped the Pipeline: %s", pipelineId)
 		}
+		*/
 
 	} else if incRequest.Type == "PIPELINE_START" {
 		log.Printf("[INFO] Should start the pipeline %#v", identifier)
 		pipelineId, err := searchPipeline(identifier)
 		if err != nil {
 			if err.Error() == "no existing pipeline found with name" {
-				log.Printf("[WARNING] no pipeline found for %s, creating a new one", identifier)
+				log.Printf("[WARNING] No pipeline found for '%s', creating a new one", identifier)
 				_, CreateErr := createPipeline(command, identifier)
 				return CreateErr
 			}
+
 			log.Printf("[ERROR] Failed searching for Pipeline with name %s reason:%s ", identifier, err)
 			return err
 		}
@@ -2625,7 +2692,7 @@ func handlePipeline(incRequest shuffle.ExecutionRequest) error {
 			log.Printf("[ERROR] Failed to start Pipeline: %s reason:%s ", pipelineId, err)
 			return err
 		} else {
-			log.Printf("[INFO] successfully started the Pipeline: %s", pipelineId)
+			log.Printf("[INFO] Successfully started the Pipeline: %s", pipelineId)
 		}
 
 	} else {
@@ -2638,7 +2705,12 @@ func handlePipeline(incRequest shuffle.ExecutionRequest) error {
 
 func deployTenzirNode() error {
 	if isKubernetes == "true" {
-		return errors.New("kubernetes not implemented")
+		return errors.New("Kubernetes not implemented for Tenzir node")
+	}
+
+	err := checkTenzirNode()
+	if err == nil {
+		return nil
 	}
 
 	ctx := context.Background()
@@ -2647,8 +2719,7 @@ func deployTenzirNode() error {
 	imageName := "tenzir/tenzir:latest"
 	containerName := "tenzir-node"
 	containerStartOptions := container.StartOptions{}
-
+	_, err = shuffle.GetCache(ctx, cacheKey)
-	_, err := shuffle.GetCache(ctx, cacheKey)
 	if err == nil {
 		return nil
 	}
@@ -2693,13 +2764,14 @@ func deployTenzirNode() error {
 		}
 	} else {
 		if !containerInfo.State.Running {
-			log.Printf("[DEBUG] Tenzir Node exists but is not running")
+			log.Printf("[DEBUG] Tenzir Node exists but is not running. Restarting it.")
 			err := dockercli.ContainerStart(ctx, containerName, containerStartOptions)
 			if err != nil {
 				log.Printf("[ERROR] Failed to start Tenzir Node container: %v", err)
 				return err
 			}
 
+			time.Sleep(10 * time.Second)
 			log.Printf("[INFO] Waiting for Tenzir to become available ...")
 			err = checkTenzirNode()
 			if err != nil {
@@ -2733,26 +2805,89 @@ func createAndStartTenzirNode(ctx context.Context, containerName, imageName stri
 		Retries:  1,
 	}
 
+	// Ensure restart policy is there
 	config := &container.Config{
 		Cmd:          []string{"--commands=web server --mode=dev --bind=0.0.0.0"},
 		Image:        imageName,
 		Healthcheck:  healthconfig,
-		ExposedPorts: nat.PortSet{"5160/tcp": struct{}{}},
+		ExposedPorts: nat.PortSet{
+			"5160/tcp": struct{}{},
+			"514/udp": struct{}{},
+			"514/tcp": struct{}{},
+		},
 		Entrypoint:   []string{containerName},
+		Env:		  []string{},
+	}
+
+	tenzirApikey := os.Getenv("TENZIR_PLUGINS__PLATFORM__API_KEY")
+	tenzirControlEndpoint := os.Getenv("TENZIR_PLUGINS__PLATFORM__CONTROL_ENDPOINT")
+	tenzirPluginsPlatform := os.Getenv("TENZIR_PLUGINS__PLATFORM__TENANT_ID")
+
+	anyFound := false
+	if len(tenzirApikey) > 0 {
+		config.Env = append(config.Env, fmt.Sprintf("TENZIR_PLUGINS__PLATFORM__API_KEY=%s", tenzirApikey))
+		anyFound = true 
+	}
+
+	if len(tenzirControlEndpoint) > 0 {
+		config.Env = append(config.Env, fmt.Sprintf("TENZIR_PLUGINS__PLATFORM__CONTROL_ENDPOINT=%s", tenzirControlEndpoint))
+		anyFound = true 
+	}
+
+	if len(tenzirPluginsPlatform) > 0 {
+		config.Env = append(config.Env, fmt.Sprintf("TENZIR_PLUGINS__PLATFORM__TENANT_ID=%s", tenzirPluginsPlatform))
+		anyFound = true 
+	}
+
+	tenzirStorageFolder := os.Getenv("SHUFFLE_STORAGE_FOLDER")
+	if len(tenzirStorageFolder) > 0 {
+		tenzirStorageFolder = tenzirStorageFolder 
+
+		if !strings.HasSuffix(tenzirStorageFolder, "/") {
+			tenzirStorageFolder = tenzirStorageFolder + "/"
+		}
+	} else {
+		tenzirStorageFolder = "/tmp/tenzir/"
+	}
+
+
+	if !anyFound {
+		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"}},
+			"514/udp":  []nat.PortBinding{{HostPort: "514"}},
 			"5160/tcp": []nat.PortBinding{{HostPort: "5160"}},
 		},
 		Mounts: []mount.Mount{
 			{
-				Type:   mount.TypeVolume,
+				Type:   "bind",
-				Source: containerName,
+				Source: tenzirStorageFolder,
 				Target: "/var/lib/tenzir/",
 			},
+			{
+				Type:   "bind",
+				Source: tenzirStorageFolder,
+				Target: "/var/log/tenzir/",
+			},
+			{
+				Type:   "bind",
+				Source: tenzirStorageFolder,
+				Target: "/var/cache/tenzir/",
+			},
 		},
 		VolumeDriver: "local",
+		RestartPolicy: container.RestartPolicy{
+			Name: "always",
+		},
+	}
+
+	if skipPipelineMount {
+		hostConfig.Mounts = []mount.Mount{}
 	}
 
 	networkingConfig := &network.NetworkingConfig{
@@ -2767,6 +2902,13 @@ func createAndStartTenzirNode(ctx context.Context, containerName, imageName stri
 
 	_, 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)
+			skipPipelineMount = true
+		} else {
+			log.Printf("[ERROR] Failed to create Tenzir Node container: %v", err)
+		}
+
 		return err
 	}
 
@@ -2775,14 +2917,23 @@ func createAndStartTenzirNode(ctx context.Context, containerName, imageName stri
 		log.Printf("[ERROR] Failed to start Tenzir Node container: %v", err)
 		return err
 	}
-	log.Printf("[INFO] Tenzir Node container started successfully")
 
-	log.Printf("[INFO] Waiting for Tenzir to become available ...")
+	log.Printf("[INFO] Tenzir Node container started successfully. Waiting for it to become available..")
+	time.Sleep(10 * time.Second)
 	err = checkTenzirNode()
 	if err != nil {
+		log.Printf("[ERROR] Tenzir node is not available during deployment: %s", err)
 		return err
 	}
-	log.Printf("[INFO] Successfully deployed Tenzir Node!")
+
+	log.Printf("[INFO] Successfully deployed Tenzir Node! Setting up default syslog listener on UDP 514")
+
+	command := "from udp://0.0.0.0:514 read syslog | import"
+	_, err = createPipeline(command, "default-syslog-514")
+	if err != nil {
+		log.Printf("[ERROR] Failed to create default syslog pipeline: %s", err)
+		return nil
+	}
 
 	return nil
 }
@@ -2853,22 +3004,31 @@ func checkTenzirNode() error {
 
 func createPipeline(command, identifier string) (string, error) {
 
-	toBeDeleted := false
+	//toBeDeleted := false
-	pipelineId, err := searchPipeline(identifier)
+	/*
+		// Pre-checked. No point here
+		pipelineId, err := searchPipeline(identifier)
+		if err != nil {
+			return "", err
+		}
+	*/
 
 	url := fmt.Sprintf("%s/api/v0/pipeline/create", pipelineUrl)
 	forwardMethod := "POST"
 
-	if err != nil {
+	/*
-		if strings.Contains(fmt.Sprintf("%s", err), "no existing pipeline found") {
+		if err != nil {
-			log.Printf("[INFO] No existing pipeline found with id: %s. Creating a new one!", identifier)
+			if strings.Contains(fmt.Sprintf("%s", err), "no existing pipeline found") {
+				log.Printf("[INFO] No existing pipeline found with id: %s. Creating a new one!", identifier)
+			} else {
+				log.Printf("[ERROR] Failed to search for existing pipeline but continuing anyway : %s", err)
+			}
 		} else {
-			log.Printf("[ERROR] Failed to search for existing pipeline but continuing anyway : %s", err)
+			log.Printf("[INFO] an existing pipeline found with ID: %s. it will be deleted", pipelineId)
+			toBeDeleted = true
 		}
-	} else {
+	*/
-		log.Printf("[INFO] an existing pipeline found with ID: %s. it will be deleted", pipelineId)
+
-		toBeDeleted = true
-	}
 	// if strings.Contains(command, "shuffler.io") {
 
 	// } else {
@@ -2957,9 +3117,9 @@ func createPipeline(command, identifier string) (string, error) {
 
 	id := response.ID
 
-	if toBeDeleted {
+	//if toBeDeleted {
-		go deletePipeline(pipelineId)
+	//	go deletePipeline(pipelineId)
-	}
+	//}
 
 	return id, nil
 }
@@ -3071,44 +3231,48 @@ func deletePipeline(pipelineId string) error {
 		return fmt.Errorf("got the status code %d instead of 200", resp.StatusCode)
 	}
 
-	log.Printf("[INFO] pipeline with ID: %s deleted successfully", pipelineId)
+	log.Printf("[INFO] Pipeline with ID: %s deleted successfully", pipelineId)
+
+	pipelines = []shuffle.PipelineInfoMini{}
 	return nil
 }
 
-func searchPipeline(identifier string) (string, error) {
+// Lists the pipelines from the API exactly as they are. Definition is set up in Shuffle structs
-
+func listPipelines() ([]shuffle.PipelineInfo, error) {
-	type pipelineInfo struct {
+	responseData := shuffle.PipelineInfoWrapper{}
-		ID   string `json:"id"`
-		Name string `json:"name"`
-	}
 
 	var reqBody []byte
-
 	url := fmt.Sprintf("%s/api/v0/pipeline/list", pipelineUrl)
-
 	resp, err := http.Post(url, "application/json", bytes.NewBuffer(reqBody))
-	if err != nil {
-		return "", err
-	}
-	defer resp.Body.Close()
 
+	if err != nil {
+		return responseData.Pipelines, err
+	}
+
+	defer resp.Body.Close()
 	if resp.StatusCode != http.StatusOK {
-		return "", fmt.Errorf("got the status code %d instead of 200", resp.StatusCode)
+		return responseData.Pipelines, fmt.Errorf("Got the status code %d instead of 200 from Pipeline node", resp.StatusCode)
 	}
 
 	body, err := ioutil.ReadAll(resp.Body)
 	if err != nil {
-		return "", err
+		return responseData.Pipelines, err
 	}
 
-	var responseData struct {
-		Pipelines []pipelineInfo `json:"pipelines"`
-	}
 	if err := json.Unmarshal(body, &responseData); err != nil {
+		return responseData.Pipelines, err
+	}
+
+	return responseData.Pipelines, nil
+}
+
+func searchPipeline(identifier string) (string, error) {
+	allPipelines, err := listPipelines()
+	if err != nil {
 		return "", err
 	}
 
-	for _, pipeline := range responseData.Pipelines {
+	for _, pipeline := range allPipelines {
 		if pipeline.Name == identifier {
 			return pipeline.ID, nil
 		}
@@ -3317,72 +3481,45 @@ func removePath(containerName, path string) error {
 	return nil
 }
 
-func sendTenzirHealthStatus() error {
+func sendPipelineHealthStatus() (shuffle.LakeConfig, error) {
-	// Check one in every 10 times only
+	pipelinePayload := shuffle.LakeConfig{
-	randint := rand.Intn(10)
+		Enabled:   false,
-	_ = randint
+		Pipelines: []shuffle.PipelineInfoMini{},
-	//if randint != 0 {
+	}
-	//	return nil
-	//}
 
-	var status string
+	// To not spam down the list API too much
-	url := fmt.Sprintf("%s/api/v1/detections/siem/health", baseUrl)
+	randint := rand.Intn(5)
-	err := checkTenzirNode()
+	if len(pipelines) == 0 || randint == 0 {
-	if err != nil {
+		pipelineDef, err := listPipelines()
-		return err
+
+		if err == nil {
+			for _, pipeline := range pipelineDef {
+				pipelinePayload.Pipelines = append(pipelinePayload.Pipelines, shuffle.PipelineInfoMini{
+					ID:         pipeline.ID,
+					Name:       pipeline.Name,
+					Definition: pipeline.Definition,
+					TotalRuns:  pipeline.TotalRuns,
+					CreatedAt:  pipeline.CreatedAt,
+				})
+			}
+
+			pipelines = pipelinePayload.Pipelines
+		}
 	} else {
-		status = "active"
+		pipelinePayload.Pipelines = pipelines
 	}
 
-	//log.Printf("[DEBUG] Sending Tenzir health update to backend url '%s'", baseUrl)
+	//url := fmt.Sprintf("%s/api/v1/detections/siem/health", baseUrl)
-	forwardMethod := "POST"
+	//err := checkTenzirNode()
-	payload := map[string]interface{}{
+	err := deployTenzirNode() 
-		"status": status,
-		"environment": environment,
-		"authorization": "",
-		"org_id": "",
-
-	}
-
-	if len(auth) > 0 {
-		payload["authorization"] = auth
-	}
-
-	if len(org) > 0 {
-		payload["org_id"] = org
-	}
-
-	payloadBytes, err := json.Marshal(payload)
 	if err != nil {
-		log.Printf("[ERROR] Failed to marshal payload: %s", err)
+		return pipelinePayload, err
-		return err
-	}
-	forwardData := bytes.NewBuffer(payloadBytes)
-	req, err := http.NewRequest(
-		forwardMethod,
-		url,
-		forwardData,
-	)
-	if err != nil {
-		log.Printf("[ERROR] Failed to create HTTP request: %s", err)
-		return err
 	}
 
-	req.Header.Set("Content-Type", "application/json")
+	pipelinePayload.Enabled = true
-	client := &http.Client{Timeout: 10 * time.Second}
-	resp, err := client.Do(req)
-	if err != nil {
-		log.Printf("[ERROR] Failed to send HTTP request: %s", err)
-		return err
-	}
 
-	defer resp.Body.Close()
+	// No direct sending.
-	if resp.StatusCode != 200 {
+	return pipelinePayload, nil
-		//log.Printf("[ERROR] Pipeline: status for URL %s: %d", url, resp.StatusCode)
-		return fmt.Errorf("unexpected HTTP status code: %d", resp.StatusCode)
-	}
-
-	return nil
 }
 
 func disableRule(fileName string) error {
@@ -3438,7 +3575,7 @@ func enableRule(fileName string) error {
 		return fmt.Errorf("error moving file: %v", err)
 	}
 
-	fmt.Printf("File %s moved to %s successfully.\n", fileName, destDir)
+	fmt.Printf("[DEBUG] File %s moved to %s successfully.\n", fileName, destDir)
 	return nil
 }
 
@@ -3575,7 +3712,7 @@ func zombiecheck(ctx context.Context, workerTimeout int) error {
 
 			// Check image name
 			if !shuffleFound {
-				log.Printf("[WARNING] Zombie container skip: %#v, %s", container.Labels, container.Image)
+				//log.Printf("[WARNING] Zombie container skip: %#v, %s", container.Labels, container.Image)
 				continue
 			}
 			//} else {
@@ -3675,7 +3812,7 @@ func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest, image string,
 
 	if strings.Contains(streamUrl, "shuffler.io") || strings.Contains(streamUrl, "localhost") || strings.Contains(streamUrl, "127.0.0.1") || strings.Contains(streamUrl, "shuffle-backend") {
 
-		// Specific to debugging 
+		// Specific to debugging
 		if len(workerServerUrl) == 0 {
 			log.Printf("[INFO] Using default worker server url as previous is invalid: %s", streamUrl)
 		}