From c154fa98d0fda9143848c0048fe2bb8d1a30b15e Mon Sep 17 00:00:00 2001 From: Pascal Sthamer <10992664+P4sca1@users.noreply.github.com> Date: Wed, 23 Oct 2024 12:20:28 +0200 Subject: [PATCH 1/9] go mod tidy --- backend/go-app/go.sum | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/backend/go-app/go.sum b/backend/go-app/go.sum index 3a682615..28b022f4 100644 --- a/backend/go-app/go.sum +++ b/backend/go-app/go.sum @@ -334,8 +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.74/go.mod h1:RAJiSFjmuKmijKTbbEf9A6Ojb+3/te7g71lED7JjPus= +github.com/shuffle/shuffle-shared v0.6.77 h1:KKtM50xW2DLuRHINxhp3uXrNH0AhiwkeiiU93a8fB3A= +github.com/shuffle/shuffle-shared v0.6.77/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= From 7b5fe4a84c39ee2bcb51ef51bacf469d6e2978be Mon Sep 17 00:00:00 2001 From: Pascal Sthamer <10992664+P4sca1@users.noreply.github.com> Date: Wed, 23 Oct 2024 12:21:05 +0200 Subject: [PATCH 2/9] Use NodeName instead of NodeSelector --- backend/go-app/docker.go | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/backend/go-app/docker.go b/backend/go-app/docker.go index 03312201..ad016745 100755 --- a/backend/go-app/docker.go +++ b/backend/go-app/docker.go @@ -420,9 +420,7 @@ func buildImage(tags []string, dockerfileFolder string) error { }, }, }, - NodeSelector: map[string]string{ - "node": backendNodeName, - }, + NodeName: backendNodeName, RestartPolicy: corev1.RestartPolicyNever, Volumes: []corev1.Volume{ { From 90b62a5e9297672b46935439debf984c38f3247c Mon Sep 17 00:00:00 2001 From: Pascal Sthamer <10992664+P4sca1@users.noreply.github.com> Date: Wed, 23 Oct 2024 12:22:32 +0200 Subject: [PATCH 3/9] Fix --dockerfile flag --- backend/go-app/docker.go | 1 - 1 file changed, 1 deletion(-) diff --git a/backend/go-app/docker.go b/backend/go-app/docker.go index ad016745..d44e5f82 100755 --- a/backend/go-app/docker.go +++ b/backend/go-app/docker.go @@ -372,7 +372,6 @@ func buildImage(tags []string, dockerfileFolder string) error { contextDir := strings.Replace(dockerfileFolder, "Dockerfile", "", -1) contextDir = "/app/" + contextDir log.Print("contextDir: ", contextDir) - dockerFile := "./Dockerfile" client, err := getK8sClient() if err != nil { From 840288c226cabb69db3e595cff9c531a842e4e98 Mon Sep 17 00:00:00 2001 From: Pascal Sthamer <10992664+P4sca1@users.noreply.github.com> Date: Wed, 23 Oct 2024 12:23:02 +0200 Subject: [PATCH 4/9] Use filepath module for constructing contextDir --- backend/go-app/docker.go | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/backend/go-app/docker.go b/backend/go-app/docker.go index d44e5f82..a8a34072 100755 --- a/backend/go-app/docker.go +++ b/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,8 +369,7 @@ func buildImage(tags []string, dockerfileFolder string) error { log.Printf("[INFO] registry name: %s", registryName) - contextDir := strings.Replace(dockerfileFolder, "Dockerfile", "", -1) - contextDir = "/app/" + contextDir + contextDir := filepath.Join("/app/", filepath.Dir(dockerfileLocation)) log.Print("contextDir: ", contextDir) client, err := getK8sClient() @@ -406,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], @@ -483,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) From d8c5fa220f5ce9c3bc2e3ba33bfc6aaf48b06b14 Mon Sep 17 00:00:00 2001 From: LuisThuillier Date: Wed, 30 Oct 2024 18:37:59 +0100 Subject: [PATCH 5/9] Adding single app hotload & fix POST as an option to both hotloads --- backend/go-app/main.go | 3 +- backend/go-app/walkoff.go | 64 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 66 insertions(+), 1 deletion(-) diff --git a/backend/go-app/main.go b/backend/go-app/main.go index 7ce1309c..26de3dc9 100755 --- a/backend/go-app/main.go +++ b/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("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") diff --git a/backend/go-app/walkoff.go b/backend/go-app/walkoff.go index 7b8008ce..0f3c7b68 100755 --- a/backend/go-app/walkoff.go +++ b/backend/go-app/walkoff.go @@ -2802,6 +2802,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 { From b9ca7b0f9a6f5b3700f840f44d971c145df18161 Mon Sep 17 00:00:00 2001 From: Frikky Date: Thu, 31 Oct 2024 01:08:33 +0100 Subject: [PATCH 6/9] Massive pipeline overhaul to allow for more control and visibility into what is happenign in Orborus --- backend/go-app/walkoff.go | 45 ++-- functions/onprem/orborus/go.mod | 2 +- functions/onprem/orborus/orborus.go | 314 ++++++++++++++++------------ 3 files changed, 217 insertions(+), 144 deletions(-) diff --git a/backend/go-app/walkoff.go b/backend/go-app/walkoff.go index 7b8008ce..922bdced 100755 --- a/backend/go-app/walkoff.go +++ b/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" { - 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" - } + // Set the checkin cache - 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) } diff --git a/functions/onprem/orborus/go.mod b/functions/onprem/orborus/go.mod index 74471ede..06ae0659 100644 --- a/functions/onprem/orborus/go.mod +++ b/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 diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index 3246fb4a..dfa645b7 100755 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -30,8 +30,8 @@ import ( "strings" "sync" "time" - "math/rand" + "math/rand" //"os/signal" //"syscall" @@ -103,11 +103,12 @@ 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? -var apiKey = os.Getenv("AUTH_FOR_ORBORUS") +// For it to download from Sigma? +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 dockercli *dockerclient.Client @@ -725,7 +726,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 +756,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 +806,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, - service.ID, - service.Version, - service.Spec, + ctx, + service.ID, + service.Version, + service.Spec, serviceUpdateOptions, ) @@ -824,7 +824,7 @@ func handleBackendImageDownload(ctx context.Context, images string) error { } } } - + } } @@ -1522,7 +1522,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) @@ -2019,8 +2018,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 +2027,13 @@ 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 +2119,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 { - toBeRemoved.Data = append(toBeRemoved.Data, incRequest) + + log.Printf("[ERROR] Failed handling pipeline (%s %s): %s. Deleting job anyway.", incRequest.Type, incRequest.ExecutionSource, err) } + + 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 +2176,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 +2192,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 +2208,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 +2223,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 +2234,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 +2609,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 +2623,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 +2638,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 +2667,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 +2689,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 { @@ -2743,6 +2807,7 @@ func createAndStartTenzirNode(ctx context.Context, containerName, imageName stri hostConfig := &container.HostConfig{ PortBindings: nat.PortMap{ + "514/udp": []nat.PortBinding{{HostPort: "514"}}, "5160/tcp": []nat.PortBinding{{HostPort: "5160"}}, }, Mounts: []mount.Mount{ @@ -2775,14 +2840,22 @@ 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 { 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 +2926,31 @@ func checkTenzirNode() error { func createPipeline(command, identifier string) (string, error) { - toBeDeleted := false - pipelineId, err := searchPipeline(identifier) + //toBeDeleted := false + /* + // 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") { - log.Printf("[INFO] No existing pipeline found with id: %s. Creating a new one!", identifier) + /* + if err != nil { + 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 +3039,9 @@ func createPipeline(command, identifier string) (string, error) { id := response.ID - if toBeDeleted { - go deletePipeline(pipelineId) - } + //if toBeDeleted { + // go deletePipeline(pipelineId) + //} return id, nil } @@ -3071,44 +3153,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) { - - type pipelineInfo struct { - ID string `json:"id"` - Name string `json:"name"` - } +// Lists the pipelines from the API exactly as they are. Definition is set up in Shuffle structs +func listPipelines() ([]shuffle.PipelineInfo, error) { + responseData := shuffle.PipelineInfoWrapper{} 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 +3403,44 @@ func removePath(containerName, path string) error { return nil } -func sendTenzirHealthStatus() error { - // Check one in every 10 times only - randint := rand.Intn(10) - _ = randint - //if randint != 0 { - // return nil - //} +func sendPipelineHealthStatus() (shuffle.LakeConfig, error) { + pipelinePayload := shuffle.LakeConfig{ + Enabled: false, + Pipelines: []shuffle.PipelineInfoMini{}, + } - var status string - url := fmt.Sprintf("%s/api/v1/detections/siem/health", baseUrl) + // To not spam down the list API too much + randint := rand.Intn(5) + if len(pipelines) == 0 || randint == 0 { + pipelineDef, err := listPipelines() + + 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 { + pipelinePayload.Pipelines = pipelines + } + + //url := fmt.Sprintf("%s/api/v1/detections/siem/health", baseUrl) err := checkTenzirNode() if err != nil { - return err - } else { - status = "active" + return pipelinePayload, err } - //log.Printf("[DEBUG] Sending Tenzir health update to backend url '%s'", baseUrl) - forwardMethod := "POST" - payload := map[string]interface{}{ - "status": status, - "environment": environment, - "authorization": "", - "org_id": "", + pipelinePayload.Enabled = true - } - - 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 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") - 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() - if resp.StatusCode != 200 { - //log.Printf("[ERROR] Pipeline: status for URL %s: %d", url, resp.StatusCode) - return fmt.Errorf("unexpected HTTP status code: %d", resp.StatusCode) - } - - return nil + // No direct sending. + return pipelinePayload, nil } func disableRule(fileName string) error { @@ -3438,7 +3496,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 +3633,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 +3733,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) } From 8e8114d1cece453c7b5113132e7dc09d39e7e00a Mon Sep 17 00:00:00 2001 From: Frikky Date: Thu, 31 Oct 2024 14:38:28 +0100 Subject: [PATCH 7/9] Started cleaning up library stuff --- functions/onprem/orborus/orborus.go | 54 +++++++++++++++++++++++++---- 1 file changed, 48 insertions(+), 6 deletions(-) diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index dfa645b7..2a501c36 100755 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -1915,6 +1915,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 @@ -2028,6 +2029,7 @@ 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) @@ -2702,7 +2704,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() @@ -2711,8 +2718,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 } @@ -2757,13 +2763,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 { @@ -2797,16 +2804,47 @@ 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 + } + + if !anyFound { + log.Printf("[DEBUG] No Tenzir Plugin environment variables found.") } hostConfig := &container.HostConfig{ PortBindings: nat.PortMap{ + "514/tcp": []nat.PortBinding{{HostPort: "514"}}, "514/udp": []nat.PortBinding{{HostPort: "514"}}, "5160/tcp": []nat.PortBinding{{HostPort: "5160"}}, }, @@ -2818,6 +2856,9 @@ func createAndStartTenzirNode(ctx context.Context, containerName, imageName stri }, }, VolumeDriver: "local", + RestartPolicy: container.RestartPolicy{ + Name: "always", + }, } networkingConfig := &network.NetworkingConfig{ @@ -3432,7 +3473,8 @@ func sendPipelineHealthStatus() (shuffle.LakeConfig, error) { } //url := fmt.Sprintf("%s/api/v1/detections/siem/health", baseUrl) - err := checkTenzirNode() + //err := checkTenzirNode() + err := deployTenzirNode() if err != nil { return pipelinePayload, err } From c16168bef460dc3f2a958bc0c86e2a441da7ef28 Mon Sep 17 00:00:00 2001 From: Frikky Date: Thu, 31 Oct 2024 15:10:58 +0100 Subject: [PATCH 8/9] Orborus autodeploy pipeline --- functions/onprem/orborus/orborus.go | 41 +++++++++++++++++++++++++++-- 1 file changed, 39 insertions(+), 2 deletions(-) diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index 2a501c36..857e216d 100755 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -110,6 +110,7 @@ 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 @@ -2838,8 +2839,22 @@ func createAndStartTenzirNode(ctx context.Context, containerName, imageName stri 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{ @@ -2850,10 +2865,20 @@ func createAndStartTenzirNode(ctx context.Context, containerName, imageName stri }, Mounts: []mount.Mount{ { - Type: mount.TypeVolume, - Source: containerName, + Type: "bind", + 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{ @@ -2861,6 +2886,10 @@ func createAndStartTenzirNode(ctx context.Context, containerName, imageName stri }, } + if skipPipelineMount { + hostConfig.Mounts = []mount.Mount{} + } + networkingConfig := &network.NetworkingConfig{ EndpointsConfig: map[string]*network.EndpointSettings{ "tenzir-network": { @@ -2873,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 } @@ -2886,6 +2922,7 @@ func createAndStartTenzirNode(ctx context.Context, containerName, imageName stri time.Sleep(10 * time.Second) err = checkTenzirNode() if err != nil { + log.Printf("[ERROR] Tenzir node is not available during deployment: %s", err) return err } From 1bebc7e0544e315f5169a5aa93c4f2f03b4b5f07 Mon Sep 17 00:00:00 2001 From: LuisThuillier Date: Fri, 1 Nov 2024 08:06:25 +0100 Subject: [PATCH 9/9] Add GET and POST to not break break backwards compatibility. --- backend/go-app/main.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/backend/go-app/main.go b/backend/go-app/main.go index 26de3dc9..6b3ba779 100755 --- a/backend/go-app/main.go +++ b/backend/go-app/main.go @@ -5112,7 +5112,7 @@ 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("POST", "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")