From f825306a85bcb9dc636e4890104f1563fca41f9c Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Fri, 26 Apr 2024 15:59:50 +0000 Subject: [PATCH 1/5] trying to improve tenzir deployement --- functions/onprem/orborus/orborus.go | 700 ++++++++++++++++++++++------ 1 file changed, 559 insertions(+), 141 deletions(-) diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index 74109e92..4a373ac0 100755 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -38,6 +38,7 @@ import ( "github.com/docker/docker/api/types/mount" "github.com/docker/docker/api/types/network" "github.com/docker/docker/api/types/swarm" + "github.com/docker/go-connections/nat" //"github.com/docker/docker/api/types/filters" dockerclient "github.com/docker/docker/client" @@ -1495,6 +1496,13 @@ func main() { initializeImages() + go func() { + if err := deployTenzirNode(); err != nil { + // Handle the error here + } + }() + + workerImage := fmt.Sprintf("%s/%s/shuffle-worker:%s", baseimageregistry, baseimagename, workerVersion) if len(newWorkerImage) > 0 { workerImage = newWorkerImage @@ -1666,14 +1674,14 @@ func main() { newrequests := []shuffle.ExecutionRequest{} for _, incRequest := range executionRequests.Data { // Looking for specific jobs - if incRequest.Type == "PIPELINE_CREATE" || incRequest.Type == "PIPELINE_UPDATE" || incRequest.Type == "PIPELINE_DELETE" { + if incRequest.Type == "PIPELINE_CREATE" || incRequest.Type == "PIPELINE_STOP" || incRequest.Type == "PIPELINE_DELETE" { err := handlePipeline(incRequest) if err != nil { log.Printf("[ERROR] Failed handling pipeline: %s", err) - } - - toBeRemoved.Data = append(toBeRemoved.Data, incRequest) + } else { + toBeRemoved.Data = append(toBeRemoved.Data, incRequest) + } } else if incRequest.Type == "DOCKER_IMAGE_DOWNLOAD" { log.Printf("[INFO] Should delete -> download new image %#v", incRequest.ExecutionArgument) @@ -1868,154 +1876,156 @@ func main() { } -func deployPipeline(image, identifier, command string) error { - if isKubernetes == "true" { - return errors.New("Kubernetes not implemented") - } +// 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{}, - } +// 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 - } +// hostConfig.NetworkMode = container.NetworkMode(fmt.Sprintf("container:%s", containerId)) +// if strings.ToLower(cleanupEnv) != "false" { +// hostConfig.AutoRemove = true +// } - envVariables := []string{ - } +// 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, - //}) +// // 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") +// // 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, - //} - } +// //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, - }, - } +// 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", - } +// // 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, - ) +// 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 - } - } +// 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) - } +// 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) - } +// 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) - } +// 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 - } +// 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) - } +// 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 +// } - // 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 @@ -2024,34 +2034,442 @@ func deployPipeline(image, identifier, command string) error { // Read from Cache and send it to a webhook // 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 { - if len(incRequest.ExecutionArgument) == 0 { + if incRequest.Type != "PIPELINE_STOP" && len(incRequest.ExecutionArgument) == 0 { log.Printf("[ERROR] No execution argument found for pipeline create. Skipping") - return errors.New("No execution argument found for pipeline create. Skipping") + return errors.New("no execution argument found for pipeline create. Skipping") } - image := "tenzir/tenzir:latest" + //image := "tenzir/tenzir:latest" 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 %#v. Name: %#v", incRequest.ExecutionArgument, identifier) - err := deployPipeline(image, identifier, command) + //err := deployPipeline(image, identifier, command) + pipelineId, err := createPipeline(command, identifier) if err != nil { log.Printf("[ERROR] Failed to deploy pipeline: %s", err) return err } else { - log.Printf("[INFO] Pipeline deployed successfully") + log.Printf("[INFO] Pipeline deployed successfully with Id: %s", pipelineId) + newErr := savePipelineData(pipelineId, identifier, "running") + if newErr != nil { + log.Printf("[DEBUG] failed to save the pipeline data: %s", newErr) + } else { + log.Printf("[INFO] succesfully saved the pipeline info ") + } } } else if incRequest.Type == "PIPELINE_DELETE" { log.Printf("[INFO] Should delete pipeline %#v", incRequest.ExecutionArgument) - } else if incRequest.Type == "PIPELINE_UPDATE" { - log.Printf("[INFO] Should update pipeline %#v", incRequest.ExecutionArgument) + 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) + return err + } + state, err := updatePipelineState(pipelineId, "stop") + if err != nil { + log.Printf("[ERROR] Failed to stop Pipeline: %s reason:%s ", pipelineId, err) + return err + } else { + log.Printf("[INFO] successfully stopped the Pipeline: %s", pipelineId) + } + err = savePipelineData(pipelineId, identifier, state) + if err != nil { + log.Printf("[DEBUG] failed to save the pipeline data: %s", err) + } else { + log.Printf("[INFO] succesfully saved the pipeline info ") + } + } else { log.Printf("[ERROR] Unknown type for pipeline: %s", incRequest.Type) - return errors.New("Unknown type for pipeline") + return errors.New("unknown type for pipeline") } + return nil +} + +func deployTenzirNode() error { + + if isKubernetes == "true" { + return errors.New("kubernetes not implemented") + } + + ctx := context.Background() + + imageName := "tenzir/tenzir" + containerName := "tenzir-node" + + healthconfig := &container.HealthConfig{ + Test: []string{"tenzir --connection-timeout=30s --connection-retry-delay=1s 'api /ping'"}, + Interval: 30 * time.Second, + Retries: 1, + } + + 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{}{}}, + Entrypoint: []string{containerName}, + } + + hostConfig := &container.HostConfig{ + PortBindings: nat.PortMap{ + "5160/tcp": []nat.PortBinding{{HostPort: "5160"}}, + }, + Mounts: []mount.Mount{ + { + Type: mount.TypeVolume, + Source: containerName, + Target: "/var/lib/tenzir/", + }, + }, + LogConfig: container.LogConfig{ + Type: "json-file", + Config: map[string]string{ + "max-size": "10m", + }, + }, + VolumeDriver: "local", + } + + // do we need to pull manually ?? + pullOptions := types.ImagePullOptions{} + out, err := dockercli.ImagePull(ctx, imageName, pullOptions) + if err != nil { + log.Printf("[ERROR] Failed to pull the tenzir image %s",err) + } + defer out.Close() + + containerStartOptions := container.StartOptions{} + _, err = dockercli.ContainerCreate(ctx, config, hostConfig, nil, nil, containerName) + if err != nil { + if strings.Contains(fmt.Sprintf("%s", err), "Conflict. The container name ") { + log.Printf("[DEBUG] Tenzir Node Container already exists, starting it") + + err = dockercli.ContainerStart(ctx, containerName, containerStartOptions) + if err != nil { + log.Printf("[ERROR] Failed to start existing Tenzir Node container: %v", err) + return err + } + + log.Printf("[INFO] Existing Tenzir Node container started successfully") + return nil + } + return fmt.Errorf("failed to create Tenzir Node container: %v", err) + } + + log.Printf("[INFO] New Tenzir Node container created successfully") + + err = dockercli.ContainerStart(ctx, containerName, containerStartOptions) + if err != nil { + log.Printf("[ERROR] Failed to start new Tenzir Node container: %v", err) + return err + } + + log.Printf("[INFO] New Tenzir Node container started successfully") + return nil +} + +func createPipeline(command, identifier string) (string, error) { + + toBeDeleted := false + pipelineId, err := searchPipeline(identifier) + + url := fmt.Sprintf("%s/api/v0/pipelines/create", tenzirUrl) + forwardMethod := "POST" + + if err != nil { + if strings.Contains(fmt.Sprintf("%s", err), "no existing pipeline found") { + log.Printf("[INFO] No existing pipeline found with name: %s. Creating a new one!", identifier) + } else { + log.Printf("[ERROR] Failed to search for existing pipeline but continuing anyway : %s", err) + } + } else { + log.Printf("[INFO] an existing pipeline found with ID: %s. it will be deleted", pipelineId) + toBeDeleted = true + } + + requestBody := map[string]interface{}{ + "definition": command, + "name": identifier, + "hidden": false, + "autostart": map[string]bool{ + "created": true, + "completed": false, + "failed": false, + }, + "autodelete": map[string]bool{ + "completed": false, + "failed": true, + "stopped": false, + }, + } + + requestBodyJSON, err := json.Marshal(requestBody) + if err != nil { + log.Printf("[ERROR] failed marshalling body: %s", err) + return "", err + } + + forwardData := bytes.NewBuffer(requestBodyJSON) + + 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("[DEBUG] status code is %d instead of 200", resp.StatusCode) + return "", fmt.Errorf("got the status code %d instead of 200", resp.StatusCode) + } + + type PipelineResponse struct { + ID string `json:"id"` + } + + var response PipelineResponse + if err := json.NewDecoder(resp.Body).Decode(&response); err != nil { + log.Printf("[ERROR] decoding response: %s", err) + return "", err + } + + if response.ID == "" { + log.Println("[DEBUG] ID not found or empty in response") + return "", errors.New("pipeline ID not found or empty in the response") + } + + id := response.ID + + if toBeDeleted { + go deletePipeline(pipelineId) + } + + return id, nil +} + +func updatePipelineState(pipelineId, action string) (string, error) { + + url := fmt.Sprintf("%s/api/v0/pipeline/update", tenzirUrl) + forwardMethod := "POST" + + requestBody := map[string]interface{}{ + "id": pipelineId, + "action": action, + "autostart": map[string]bool{ + "created": true, + "completed": false, + "failed": false, + }, + "autodelete": map[string]bool{ + "completed": false, + "failed": true, + "stopped": false, + }, + } + + requestBodyJSON, err := json.Marshal(requestBody) + if err != nil { + return "", err + } + forwardData := bytes.NewBuffer(requestBodyJSON) + + 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 != http.StatusOK { + return "", fmt.Errorf("got the status code %d instead of 200", resp.StatusCode) + } + + body, err := ioutil.ReadAll(resp.Body) + if err != nil { + return "", err + } + + var responseData struct { + Pipeline struct { + State string `json:"state"` + } `json:"pipeline"` + } + if err := json.Unmarshal(body, &responseData); err != nil { + return "", err + } + + return responseData.Pipeline.State, nil +} + +func deletePipeline(pipelineId string) error { + requestBody := map[string]string{ + "id": pipelineId, + } + + url := fmt.Sprintf("%s/api/v0/pipeline/delete", tenzirUrl) + forwardMethod := "POST" + + requestBodyJSON, err := json.Marshal(requestBody) + if err != nil { + log.Println("[ERROR] failed marshalling request body:", err) + return err + } + + forwardData := bytes.NewBuffer(requestBodyJSON) + + 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("[DEBUG] The deletion of pipeline with ID: %s is unsucessful as status code is NOT 200 !!!", pipelineId) + return fmt.Errorf("got the status code %d instead of 200", resp.StatusCode) + } + + log.Printf("[INFO] pipeline with ID: %s deleted successfully", pipelineId) + return nil +} + +func searchPipeline(identifier string) (string, error) { + + type pipelineInfo struct { + ID string `json:"id"` + Name string `json:"name"` + } + + var reqBody []byte + + url := fmt.Sprintf("%s/api/v0/pipeline/list", tenzirUrl) + + resp, err := http.Post(url, "application/json", bytes.NewBuffer(reqBody)) + if err != nil { + return "", err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return "", fmt.Errorf("got the status code %d instead of 200", resp.StatusCode) + } + + body, err := ioutil.ReadAll(resp.Body) + if err != nil { + return "", err + } + + var responseData struct { + Pipelines []pipelineInfo `json:"pipelines"` + } + if err := json.Unmarshal(body, &responseData); err != nil { + return "", err + } + + for _, pipeline := range responseData.Pipelines { + if pipeline.Name == identifier { + return pipeline.ID, nil + } + } + + return "", errors.New("no existing pipeline found with name") +} + +func savePipelineData(pipelineId, identifier, status string) error { + + url := fmt.Sprintf("%s/api/v1/triggers/pipeline/save", tenzirUrl) + + forwardMethod := "PUT" + + payload := map[string]interface{}{ + "pipeline_id": pipelineId, + "trigger_id": identifier, + "status": status, + } + + 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] Received non-successful HTTP status code: %d", resp.StatusCode) + return fmt.Errorf("unexpected HTTP status code: %d", resp.StatusCode) + } return nil } From c59f575711190a6a29096bc9eb064303aeb00e11 Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Fri, 26 Apr 2024 16:02:16 +0000 Subject: [PATCH 2/5] trying to improve deployment --- .env | 5 ++ functions/onprem/orborus/orborus.go | 84 +++++++++++++++++++++-------- 2 files changed, 66 insertions(+), 23 deletions(-) diff --git a/.env b/.env index 63ac9da3..b0c7ad25 100755 --- a/.env +++ b/.env @@ -100,4 +100,9 @@ SHUFFLE_OPENSEARCH_PROXY= SHUFFLE_OPENSEARCH_INDEX_PREFIX= SHUFFLE_OPENSEARCH_SKIPSSL_VERIFY=true +#Tenzir related +IS_TENZIR=false +SHUFFLE_TENZIR_URL=http://localhost:5160 + + DEBUG_MODE=false diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index 4a373ac0..1e19caba 100755 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -103,6 +103,8 @@ var swarmConfig = os.Getenv("SHUFFLE_SWARM_CONFIG") var swarmNetworkName = os.Getenv("SHUFFLE_SWARM_NETWORK_NAME") var orborusLabel = os.Getenv("SHUFFLE_ORBORUS_LABEL") var memcached = os.Getenv("SHUFFLE_MEMCACHED") +var isTenzir = os.Getenv("IS_TENZIR") +var tenzirUrl = os.Getenv("SHUFFLE_TENZIR_URL") var executionIds = []string{} var namespacemade = false // For K8s @@ -110,6 +112,7 @@ var namespacemade = false // For K8s var dockercli *dockerclient.Client var containerId string var executionCount = 0 +var isTenzirReady = false func init() { var err error @@ -1496,13 +1499,17 @@ func main() { initializeImages() - go func() { - if err := deployTenzirNode(); err != nil { - // Handle the error here - } - }() + if isTenzir == "true" { + func() { + if err := deployTenzirNode(); err != nil { + log.Printf("[ERROR] Failed to deploy the tenzir node, reason: %v", err) + } else { + log.Printf("[INFO] Tenzir node is deployed successfully and is available for requests!") + isTenzirReady = true + } + }() + } - workerImage := fmt.Sprintf("%s/%s/shuffle-worker:%s", baseimageregistry, baseimagename, workerVersion) if len(newWorkerImage) > 0 { workerImage = newWorkerImage @@ -1674,14 +1681,18 @@ func main() { newrequests := []shuffle.ExecutionRequest{} for _, incRequest := range executionRequests.Data { // Looking for specific jobs - if incRequest.Type == "PIPELINE_CREATE" || incRequest.Type == "PIPELINE_STOP" || incRequest.Type == "PIPELINE_DELETE" { - - err := handlePipeline(incRequest) - if err != nil { - log.Printf("[ERROR] Failed handling pipeline: %s", err) + if isTenzir == "true" && incRequest.Type == "PIPELINE_CREATE" || incRequest.Type == "PIPELINE_STOP" || incRequest.Type == "PIPELINE_DELETE" { + if isTenzirReady { + err := handlePipeline(incRequest) + if err != nil { + log.Printf("[ERROR] Failed handling pipeline: %s", err) + //update it to db ?? + } } else { - toBeRemoved.Data = append(toBeRemoved.Data, incRequest) - } + log.Printf("[WARNING] Couldnt Handle pipeline request as tenzir node is not ready") + } + toBeRemoved.Data = append(toBeRemoved.Data, incRequest) + } else if incRequest.Type == "DOCKER_IMAGE_DOWNLOAD" { log.Printf("[INFO] Should delete -> download new image %#v", incRequest.ExecutionArgument) @@ -2137,12 +2148,6 @@ func deployTenzirNode() error { Target: "/var/lib/tenzir/", }, }, - LogConfig: container.LogConfig{ - Type: "json-file", - Config: map[string]string{ - "max-size": "10m", - }, - }, VolumeDriver: "local", } @@ -2169,7 +2174,7 @@ func deployTenzirNode() error { log.Printf("[INFO] Existing Tenzir Node container started successfully") return nil } - return fmt.Errorf("failed to create Tenzir Node container: %v", err) + return err } log.Printf("[INFO] New Tenzir Node container created successfully") @@ -2181,15 +2186,46 @@ func deployTenzirNode() error { } log.Printf("[INFO] New Tenzir Node container started successfully") + + log.Printf("[INFO] Waiting for tenzir to become available ...") + err = checkTenzirNode() + if err != nil{ + return err + } + return nil } +func checkTenzirNode() error { + retries := 20 + retryInterval := 3 * time.Second + url := fmt.Sprintf("%s/ping",tenzirUrl) + forwardMethod := "POST" + + client := http.Client{} + req, err := http.NewRequest(forwardMethod, url, nil) + if err != nil { + log.Printf("[ERROR] Failed to create HTTP request: %s", err) + 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) + } + + return fmt.Errorf("tenzir node is not available") +} + func createPipeline(command, identifier string) (string, error) { toBeDeleted := false pipelineId, err := searchPipeline(identifier) - url := fmt.Sprintf("%s/api/v0/pipelines/create", tenzirUrl) + url := fmt.Sprintf("%s/api/v0/pipeline/create", tenzirUrl) forwardMethod := "POST" if err != nil { @@ -2217,6 +2253,7 @@ func createPipeline(command, identifier string) (string, error) { "failed": true, "stopped": false, }, + "retry_delay": "500.0ms", } requestBodyJSON, err := json.Marshal(requestBody) @@ -2429,13 +2466,14 @@ func searchPipeline(identifier string) (string, error) { func savePipelineData(pipelineId, identifier, status string) error { - url := fmt.Sprintf("%s/api/v1/triggers/pipeline/save", tenzirUrl) + url := fmt.Sprintf("%s/api/v1/triggers/pipeline/save", baseUrl) + identifierWithoutPrefix := strings.TrimPrefix(identifier, "shuffle-") forwardMethod := "PUT" payload := map[string]interface{}{ "pipeline_id": pipelineId, - "trigger_id": identifier, + "trigger_id": identifierWithoutPrefix, "status": status, } From 02fad7d797f510f2e8a71f587535fde818947f8c Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Sat, 27 Apr 2024 10:39:52 +0000 Subject: [PATCH 3/5] fixing things a little bit --- functions/onprem/orborus/orborus.go | 38 +++++++++++------------------ 1 file changed, 14 insertions(+), 24 deletions(-) diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index 1e19caba..cf921a32 100755 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -1500,7 +1500,7 @@ func main() { initializeImages() if isTenzir == "true" { - func() { + go func() { if err := deployTenzirNode(); err != nil { log.Printf("[ERROR] Failed to deploy the tenzir node, reason: %v", err) } else { @@ -1681,15 +1681,15 @@ func main() { newrequests := []shuffle.ExecutionRequest{} for _, incRequest := range executionRequests.Data { // Looking for specific jobs - if isTenzir == "true" && incRequest.Type == "PIPELINE_CREATE" || incRequest.Type == "PIPELINE_STOP" || incRequest.Type == "PIPELINE_DELETE" { - if isTenzirReady { + if incRequest.Type == "PIPELINE_CREATE" || incRequest.Type == "PIPELINE_STOP" || incRequest.Type == "PIPELINE_DELETE" { + if isTenzir == "true" && isTenzirReady { err := handlePipeline(incRequest) if err != nil { log.Printf("[ERROR] Failed handling pipeline: %s", err) //update it to db ?? } } else { - log.Printf("[WARNING] Couldnt Handle pipeline request as tenzir node is not ready") + log.Printf("[WARNING] Unable to Handle pipeline request as tenzir node is not ready") } toBeRemoved.Data = append(toBeRemoved.Data, incRequest) @@ -2120,7 +2120,7 @@ func deployTenzirNode() error { ctx := context.Background() - imageName := "tenzir/tenzir" + imageName := "tenzir/tenzir" containerName := "tenzir-node" healthconfig := &container.HealthConfig{ @@ -2131,7 +2131,7 @@ func deployTenzirNode() error { config := &container.Config{ Cmd: []string{"--commands=web server --mode=dev --bind=0.0.0.0"}, - Image: imageName, + Image: imageName, Healthcheck: healthconfig, ExposedPorts: nat.PortSet{"5160/tcp": struct{}{}}, Entrypoint: []string{containerName}, @@ -2155,7 +2155,7 @@ func deployTenzirNode() error { pullOptions := types.ImagePullOptions{} out, err := dockercli.ImagePull(ctx, imageName, pullOptions) if err != nil { - log.Printf("[ERROR] Failed to pull the tenzir image %s",err) + log.Printf("[ERROR] Failed to pull the tenzir image %s", err) } defer out.Close() @@ -2164,32 +2164,22 @@ func deployTenzirNode() error { if err != nil { if strings.Contains(fmt.Sprintf("%s", err), "Conflict. The container name ") { log.Printf("[DEBUG] Tenzir Node Container already exists, starting it") - - err = dockercli.ContainerStart(ctx, containerName, containerStartOptions) - if err != nil { - log.Printf("[ERROR] Failed to start existing Tenzir Node container: %v", err) - return err - } - - log.Printf("[INFO] Existing Tenzir Node container started successfully") - return nil + } else { + log.Printf("[ERROR] Failed to create Tenzir container: %s", err) + return err } - return err } - log.Printf("[INFO] New Tenzir Node container created successfully") - err = dockercli.ContainerStart(ctx, containerName, containerStartOptions) if err != nil { - log.Printf("[ERROR] Failed to start new Tenzir Node container: %v", err) + log.Printf("[ERROR] Failed to start Tenzir Node container: %v", err) return err } - - log.Printf("[INFO] New Tenzir Node container started successfully") + log.Printf("[INFO] Tenzir Node container started successfully") log.Printf("[INFO] Waiting for tenzir to become available ...") err = checkTenzirNode() - if err != nil{ + if err != nil { return err } @@ -2199,7 +2189,7 @@ func deployTenzirNode() error { func checkTenzirNode() error { retries := 20 retryInterval := 3 * time.Second - url := fmt.Sprintf("%s/ping",tenzirUrl) + url := fmt.Sprintf("%s/api/v0/ping",tenzirUrl) forwardMethod := "POST" client := http.Client{} From c37429d817bd1075d9b9fcec4c252bea8b89eb29 Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Sat, 27 Apr 2024 13:33:29 +0000 Subject: [PATCH 4/5] making ui not to allow more than 1 pipeline --- frontend/src/views/AngularWorkflow.jsx | 271 ++++++++++++++++--------- 1 file changed, 172 insertions(+), 99 deletions(-) diff --git a/frontend/src/views/AngularWorkflow.jsx b/frontend/src/views/AngularWorkflow.jsx index 09f00336..bf5b21e1 100755 --- a/frontend/src/views/AngularWorkflow.jsx +++ b/frontend/src/views/AngularWorkflow.jsx @@ -7178,47 +7178,73 @@ const AngularWorkflow = (defaultprops) => { toast("Error: name can't be empty"); return; } - - var mappedStartnode = "" - const alledges = cy.edges().jsons() + + var mappedStartnode = ""; + const alledges = cy.edges().jsons(); if (alledges !== undefined && alledges !== null && alledges.length > 0) { - for (let edgekey in alledges) { - const tmp = alledges[edgekey] - console.log("TMP: ", tmp, tmp.data.source) - if (tmp.data.source === trigger.id) { - mappedStartnode = tmp.data.target - break - } - } - } - - toast("Creating pipeline") - const data = usecase - const url = `${globalUrl}/api/v1/triggers/pipeline` - fetch(url, { - method: "POST", - headers: { - "Content-Type": "application/json", - Accept: "application/json", - }, - body: JSON.stringify(data), - credentials: "include", + for (let edgekey in alledges) { + const tmp = alledges[edgekey]; + console.log("TMP: ", tmp, tmp.data.source); + if (tmp.data.source === trigger.id) { + mappedStartnode = tmp.data.target; + break; + } } - ) + } + const data = usecase; + if (data.type === "create") toast("Creating pipeline"); + else toast("stopping pipeline"); + const url = `${globalUrl}/api/v1/triggers/pipeline`; + fetch(url, { + method: "POST", + headers: { + "Content-Type": "application/json", + Accept: "application/json", + }, + body: JSON.stringify(data), + credentials: "include", + }) .then((response) => { if (response.status !== 200) { console.log("Status not 200 for stream results :O!"); - } - + } + return response.json(); }) .then((responseJson) => { if (!responseJson.success) { toast("Failed to set pipeline: " + responseJson.reason); } else { - toast("Successfully created pipeline"); - workflow.triggers[triggerindex].status = "running"; - trigger.status = "running"; + if (data.type === "create") toast("Pipeline will be created!"); + else toast("Pipeline will be stopped!"); + if (!workflow.triggers[triggerindex].parameters) { + workflow.triggers[triggerindex].parameters = []; + } + + if (workflow.triggers[triggerindex].parameters.length > 0) { + workflow.triggers[triggerindex].parameters[0].name = data.name; + workflow.triggers[triggerindex].parameters[0].value = data.command; + + trigger.parameters[0].name = data.name; + trigger.parameters[0].value = data.command; + if (data.type === "stop") { + trigger.status = "stopped"; + workflow.triggers[triggerindex].status = "stopped"; + } else { + trigger.status = "running"; + workflow.triggers[triggerindex].status = "running"; + } + } else { + const newParameter = { + name: data.name, + value: data.command, + }; + trigger.parameters.push(newParameter); + if (data.type === "stop") trigger.status = "stopped"; + else trigger.status = "running"; + workflow.triggers[triggerindex] = trigger; + } + setSelectedTrigger(trigger); setWorkflow(workflow); console.log("Should set the status to running and save"); @@ -7227,9 +7253,9 @@ const AngularWorkflow = (defaultprops) => { }) .catch((error) => { //toast(error.toString()); - console.log("Get schedule error: ", error.toString()); + console.log("Get pipeline error: ", error.toString()); }); - } + }; const submitSchedule = (trigger, triggerindex) => { if (trigger.name.length <= 0) { @@ -13626,7 +13652,6 @@ const AngularWorkflow = (defaultprops) => { if (data.Name.toLowerCase() === "cloud") { return null } - return ( { Run HTTP Request */} +
{ + if (selectedTrigger.status == "running") { + return; // Do nothing if cursor is "not-allowed" + } + const pipelineConfig = { + name: selectedTrigger.label, + type: "create", + command: "load tcp://0.0.0.0:514 | read syslog | export", + environment: selectedTrigger.environment, + workflow_id: workflow.id, + trigger_id: selectedTrigger.id, + }; -
+ Start Syslog listener +
- }} - onClick={() => { - const pipelineConfig = { - "name": selectedTrigger.label, - "type": "create", - "command": "load tcp://0.0.0.0:514 | read syslog | export", - "environment": selectedTrigger.environment, - } +
{ + if (selectedTrigger.status == "running") { + return; + } + const pipelineConfig = { + name: selectedTrigger.label, + type: "create", + command: + "export --live | sigma /path/to/rules | to http://192.168.86.44:5002/api/v1/hooks/webhook_665ace5f-f27b-496a-a365-6e07eb61078c write lines", + environment: selectedTrigger.environment, + workflow_id: workflow.id, + trigger_id: selectedTrigger.id, + }; - submitPipeline(selectedTrigger, selectedTriggerIndex, pipelineConfig) - }} - > - Start Syslog listener -
+ submitPipeline(selectedTrigger, selectedTriggerIndex, pipelineConfig); + }} + > + Run Sigma Rulesearch +
-
{ + if (selectedTrigger.status == "running") { + return; + } + const pipelineConfig = { + name: selectedTrigger.label, + type: "create", + command: + "from kafka://1.2.3.4 --topic foo | to http://api.com X-Token:Secret", + environment: selectedTrigger.environment, + workflow_id: workflow.id, + trigger_id: selectedTrigger.id, + }; - }} - onClick={() => { - const pipelineConfig = { - "name": selectedTrigger.label, - "type": "create", - "command": "export --live | sigma /path/to/rules | to http://192.168.86.44:5002/api/v1/hooks/webhook_665ace5f-f27b-496a-a365-6e07eb61078c write lines", - "environment": selectedTrigger.environment, - } - - submitPipeline(selectedTrigger, selectedTriggerIndex, pipelineConfig) - }} - > - Run Sigma Rulesearch -
- -
{ - const pipelineConfig = { - "name": selectedTrigger.label, - "type": "create", - "command": "from kafka://1.2.3.4 --topic foo | to http://api.com X-Token:Secret", - "environment": selectedTrigger.environment, - } - - submitPipeline(selectedTrigger, selectedTriggerIndex, pipelineConfig) - }} - > - Follow Kafka Queue -
+ submitPipeline(selectedTrigger, selectedTriggerIndex, pipelineConfig); + }} + > + Follow Kafka Queue +
{ variant="contained" disabled={selectedTrigger.status === "running"} onClick={() => { - toast("Should start. But it doesn't") + toast("Select anyone of the parameters to start") }} color="primary" > Start +