diff --git a/.env b/.env index 61885b3b..64e325bc 100755 --- a/.env +++ b/.env @@ -101,7 +101,7 @@ SHUFFLE_OPENSEARCH_INDEX_PREFIX= SHUFFLE_OPENSEARCH_SKIPSSL_VERIFY=true #Tenzir related -SHUFFLE_TENZIR_URL=http://localhost:5160 +SHUFFLE_TENZIR_URL= DEBUG_MODE=false diff --git a/backend/go-app/main.go b/backend/go-app/main.go index 0547b7b6..54f0874c 100755 --- a/backend/go-app/main.go +++ b/backend/go-app/main.go @@ -1975,6 +1975,144 @@ func handleWebhookCallback(resp http.ResponseWriter, request *http.Request) { } +func handlePipelineCallback(resp http.ResponseWriter, request *http.Request) { + + if request.Method != "POST" { + request.Method = "POST" + } + + if request.Body == nil { + stringReader := strings.NewReader("") + request.Body = ioutil.NopCloser(stringReader) + } + + path := strings.Split(request.URL.String(), "/") + if len(path) < 4 { + resp.WriteHeader(403) + resp.Write([]byte(`{"success": false}`)) + return + } + + ctx := context.Background() + location := strings.Split(request.URL.String(), "/") + + var pipelineId string + + if location[1] == "api" { + if len(location) <= 4 { + log.Printf("[INFO] Couldn't handle location. Too short in pipeline: %d", len(location)) + resp.WriteHeader(401) + resp.Write([]byte(`{"success": false}`)) + return + } + + pipelineId = location[4] + } + + userAgent := request.Header.Get("User-Agent") + if strings.Contains(strings.ToLower(userAgent), "microsoftpreview") || strings.Contains(strings.ToLower(userAgent), "googlebot") { + log.Printf("[AUDIT] Blocking googlebot and microsoftbot for pielines. UA: '%s'", userAgent) + resp.WriteHeader(400) + resp.Write([]byte(`{"success": false, "reason": "Google/Microsoft preview bots not allowed. Please change the useragent."}`)) + return + } + + if len(pipelineId) != 45 { + log.Printf("[INFO] Couldn't handle pipeline. Too short in pipeline: %d", len(pipelineId)) + resp.WriteHeader(401) + resp.Write([]byte(`{"success": false, "reason": "pipeline ID not valid"}`)) + return + } + + pipelineId = pipelineId[9:] + + pipeline, err := shuffle.GetPipeline(ctx, pipelineId) + if err != nil { + log.Printf("[WARNING] Failed getting pipeline %s (callback): %s", pipelineId, err) + resp.WriteHeader(401) + resp.Write([]byte(`{"success": false}`)) + return + } + + if pipeline.Status != "running" { + log.Printf("[WARNING] Not running %s because pipeline status is %s", pipeline.TriggerId, pipeline.Status) + resp.WriteHeader(401) + resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "The pipeline isn't running"}`))) + return + } + + if pipeline.WorkflowId == "" { + log.Printf("[DEBUG] Not running because pipeline isn't connected to any workflows") + resp.WriteHeader(401) + resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "No workflows are defined"}`))) + return + } + + body, err := ioutil.ReadAll(request.Body) + if err != nil { + log.Printf("[DEBUG] Body data error: %s", err) + resp.WriteHeader(401) + resp.Write([]byte(`{"success": false}`)) + return + } + + parsedBody := shuffle.GetExecutionbody(body) + newBody := shuffle.ExecutionStruct{ + Start: pipeline.StartNode, + ExecutionSource: "pipeline", + ExecutionArgument: parsedBody, + } + + workflow, err := shuffle.GetWorkflow(ctx, pipeline.WorkflowId) + if err == nil { + for _, branch := range workflow.Branches { + if branch.SourceID == pipeline.TriggerId { + log.Printf("[DEBUG] Found ID %s for pipeline", pipeline.TriggerId) + if branch.DestinationID != pipeline.StartNode { + newBody.Start = branch.DestinationID + break + } + } + } + } + + b, err := json.Marshal(newBody) + if err != nil { + log.Printf("[ERROR] Failed newBody marshaling for pipeline: %s", err) + resp.WriteHeader(500) + resp.Write([]byte(`{"success": false}`)) + return + } + + log.Printf("[INFO] Running pipeline for workflow %s with startnode %s", pipeline.WorkflowId, pipeline.StartNode) + + newWorkflow := shuffle.Workflow{ + ID: "", + } + + if len(pipeline.StartNode) == 0 { + log.Printf("[WARNING] No start node for pipeline %s - running with workflow default.", pipeline.TriggerId) + + } + + newRequest := &http.Request{ + URL: &url.URL{}, + Method: "POST", + Body: ioutil.NopCloser(bytes.NewReader(b)), + } + + workflowExecution, executionResp, err := handleExecution(pipeline.WorkflowId, newWorkflow, newRequest, pipeline.OrgId) + + if err == nil { + resp.WriteHeader(200) + resp.Write([]byte(fmt.Sprintf(`{"success": true, "execution_id": "%s"}`, workflowExecution.ExecutionId))) + return + } + + resp.WriteHeader(500) + resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "%s"}`, executionResp))) +} + func executeCloudAction(action shuffle.CloudSyncJob, apikey string) error { data, err := json.Marshal(action) if err != nil { @@ -4911,7 +5049,8 @@ func initHandlers() { r.HandleFunc("/api/v1/triggers/gmail/register", shuffle.HandleNewGmailRegister).Methods("GET", "OPTIONS") r.HandleFunc("/api/v1/triggers/gmail/getFolders", shuffle.HandleGetGmailFolders).Methods("GET", "OPTIONS") r.HandleFunc("/api/v1/triggers/pipeline", shuffle.HandleNewPipelineRegister).Methods("POST", "OPTIONS") - r.HandleFunc("/api/v1/triggers/pipeline/save", shuffle.HandleSavePipelineInfo).Methods("PUT", "OPTIONS") + //r.HandleFunc("/api/v1/triggers/pipeline/save", shuffle.HandleSavePipelineInfo).Methods("PUT", "OPTIONS") + r.HandleFunc("/api/v1/pipelines/{key}", handlePipelineCallback).Methods("POST", "GET", "PATCH", "PUT", "DELETE", "OPTIONS") r.HandleFunc("/api/v1/triggers", shuffle.HandleGetTriggers).Methods("GET", "OPTIONS") //r.HandleFunc("/api/v1/triggers/gmail/routing", handleGmailRouting).Methods("POST", "OPTIONS") diff --git a/frontend/src/views/AngularWorkflow.jsx b/frontend/src/views/AngularWorkflow.jsx index 10d76509..de8c0bae 100755 --- a/frontend/src/views/AngularWorkflow.jsx +++ b/frontend/src/views/AngularWorkflow.jsx @@ -68,7 +68,8 @@ import { ListItemAvatar, Badge, AvatarGroup, - Autocomplete, + Autocomplete, + Radio, } from "@mui/material"; import { @@ -220,7 +221,6 @@ export const triggers = [ id: "", }, ]; - // Adds specific text to items // https://stackoverflow.com/questions/19014250/rerender-view-on-browser-resize-with-react @@ -521,6 +521,10 @@ const AngularWorkflow = (defaultprops) => { const [highlightedApp, setHighlightedApp] = React.useState("") const [listCache, setListCache] = React.useState([]); + + const [selectedOption, setSelectedOption] = React.useState(""); + const [tenzirConfigModalOpen, setTenzirConfigModalOpen] = React.useState(false); + const [suggestionBox, setSuggestionBox] = React.useState({ "position": { "top": 500, @@ -1342,6 +1346,56 @@ const AngularWorkflow = (defaultprops) => { }); }; + const handleKafkaSubmit = (trigger) => { + if (trigger.trigger_type !== "PIPELINE") { + toast("Unable to save the configuration"); + return; + } + + trigger.parameters = []; + + const topic = document.getElementById('topic')?.value; + const bootstrapServers = document.getElementById('bootstrap_servers')?.value; + const groupId = document.getElementById('group_id')?.value; + const autoOffsetReset = document.getElementById('auto_offset_reset')?.value; + + if(topic) { + trigger.parameters.push({ + name: "topic", + value: topic + }); + } else { + toast("please enter the topic name"); + return; + } + + if (bootstrapServers) { + trigger.parameters.push({ + name: "bootstrap_servers", + value: bootstrapServers + }); + } else { + toast("please enter bootstrap server details"); + return; + } + + if (groupId) { + trigger.parameters.push({ + name: "group_id", + value: groupId + }); + } + + if (autoOffsetReset) { + trigger.parameters.push({ + name: "auto_offset_reset", + value: autoOffsetReset + }); + } + + setTenzirConfigModalOpen(false); + }; + const handleColoring = (actionId, status, label) => { if (cy === undefined) { return @@ -6963,6 +7017,20 @@ const AngularWorkflow = (defaultprops) => { } else if (selectedNode.data().trigger_type === "EMAIL") { setSelectedTrigger(selectedNode.data()); stopMailSub(selectedTrigger, triggerindex); + } else if (selectedNode.data().trigger_type === "PIPELINE") { + setSelectedTrigger(selectedNode.data()); + + const pipelineConfig = { + command: "", + name: selectedNode.data().label, + type: "delete", + environment: selectedNode.data().environment, + workflow_id: workflow.id, + trigger_id: selectedNode.data().id, + start_node: "", + }; + + submitPipeline(selectedNode.data(), triggerindex, pipelineConfig); } } @@ -7348,47 +7416,67 @@ 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 - } - } + 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; + data.start_node = mappedStartnode + + if (data.type === "create") { + toast("Creating pipeline"); + } else if (data.type === "stop") { + toast("stopping pipeline"); } - toast("Creating pipeline") - const data = usecase - const url = `${globalUrl}/api/v1/triggers/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", - } - ) + 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 if (data.type === "stop") { + toast("Pipeline will be stopped!"); + } else { + toast("Pipeline deleted!") + return + } + + trigger.parameters.push({ + name: data.name, + value: data.command, + }); + + 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"); @@ -7396,11 +7484,10 @@ 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) { toast("Error: name can't be empty"); @@ -13825,225 +13912,270 @@ const AngularWorkflow = (defaultprops) => { } const PipelineSidebar = Object.getOwnPropertyNames(selectedTrigger).length === 0 || workflow.triggers[selectedTriggerIndex] === undefined && selectedTrigger.trigger_type !== "SCHEDULE" ? null : !userdata.support === true ? null : -
-

- {selectedTrigger.app_name}: {selectedTrigger.status} -

- - What are pipelines? - - -
Name
- - -
- Environment - { + selectedTrigger.environment = e.target.value; + setSelectedTrigger(selectedTrigger); - return ( - - {data.Name} - - ); - })} - -
- -
-
- Parameters - - {/* -
{ - const pipelineConfig = { - "name": "HTTP Testing", - "type": "create", - "command": "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", - "environment": selectedTrigger.environment, - } - - submitPipeline(selectedTrigger, selectedTriggerIndex, pipelineConfig) - }} - > - Run HTTP Request -
- */} - -
{ - const pipelineConfig = { - "name": selectedTrigger.label, - "type": "create", - "command": "load tcp://0.0.0.0:514 | read syslog | export", - "environment": selectedTrigger.environment, - } - - submitPipeline(selectedTrigger, selectedTriggerIndex, pipelineConfig) - }} - > - Start Syslog listener -
- -
{ - 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 -
- -
-
+ +
+
+ Parameters +
{ - toast("Should start. But it doesn't") + // setSelectedOption("Syslog listener") + // setTenzirConfigModalOpen(true); + }} + style={{ + border: "1px solid rgba(255,255,255,0.3)", + borderRadius: theme.palette.borderRadius, + padding: 10, + cursor: "not-allowed", + marginTop: 5, + display: "flex", + alignItems: "center", }} - color="primary" > - Start - -
+ +
{ - toast("Should stop triggert") + // setSelectedOption("Sigma Rulesearch") + // setTenzirConfigModalOpen(true); + }} + style={{ + border: "1px solid rgba(255,255,255,0.3)", + borderRadius: theme.palette.borderRadius, + padding: 10, + cursor: "not-allowed", + marginTop: 5, + display: "flex", + alignItems: "center", }} - color="primary" > - Stop - + setSelectedOption("Sigma Rulesearch")} + value={"Sigma Rulesearch"} + name="option" + disabled={true} + /> + } + label="Run Sigma Rulesearch" + /> +
+ +
{ + if(selectedTrigger.status === "running"){ + toast("please stop the trigger to edit the configuration"); + return; + } else { + setSelectedOption("Kafka Queue"); + setTenzirConfigModalOpen(true); + }}} + style={{ + border: "1px solid rgba(255,255,255,0.3)", + borderRadius: theme.palette.borderRadius, + padding: 10, + cursor: "pointer", + marginTop: 5, + display: "flex", + alignItems: "center", + }} + > + setSelectedOption("Kafka Queue")} + value={"Kafka Queue"} + name="option" + /> + } + label="Follow Kafka Queue" + /> +
+ +
+ + +
-
const ScheduleSidebar = Object.getOwnPropertyNames(selectedTrigger).length === 0 || workflow.triggers[selectedTriggerIndex] === undefined && selectedTrigger.trigger_type !== "SCHEDULE" ? null :
@@ -14180,7 +14312,7 @@ const AngularWorkflow = (defaultprops) => { selectedTrigger.status === "running" } defaultValue={ - selectedTrigger.parameters === undefined ? "" : selectedTrigger.parameters[0].value + selectedTrigger.parameters === undefined ? "" : selectedTrigger.parameters[0]?.value } color="primary" placeholder="" @@ -15997,7 +16129,18 @@ const AngularWorkflow = (defaultprops) => { style={{paddingTop: 8, paddingLeft: 4, height: 25, width: 25, }} /> ); - } + } else if (execution.execution_source === "pipeline") { + return ( + {"pipeline"} trigger.trigger_type === "PIPELINE") + .large_image + } + style={{ width: size, height: size }} + /> + ); + } if ( execution.execution_parent !== null && @@ -18978,6 +19121,146 @@ const AngularWorkflow = (defaultprops) => { ) : null; + const tenzirConfigModal = tenzirConfigModalOpen ? ( + +
+ +
Configuration options for {selectedOption}
+ + + {selectedOption === "Kafka Queue" && ( + <> + Topic + param.name === "topic")?.value) || ''} + /> + bootstrap.servers + param.name === "bootstrap_servers")?.value) || ''} + /> + group.id + param.name === "group_id")?.value) || ''} + /> + auto.offest.reset + param.name === "auto_offset_reset")?.value) || ''} + /> + + )} + + + + + +
+ + { + setTenzirConfigModalOpen(false); + }} + > + + + + ) : null; + // Should get AI autocompletes const aiSubmit = (value, setResponseMsg, setSuggestionLoading, inputAction) => { if (setResponseMsg !== undefined) { @@ -19794,6 +20077,7 @@ const AngularWorkflow = (defaultprops) => { {codePopoutModal} {workflowRevisions} {authenticationModal} + {tenzirConfigModal} {/*editWorkflowModal*/} {executionArgumentModal} {configureWorkflowModal} diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index 6a8e260e..cdface09 100755 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -1667,7 +1667,7 @@ 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" { + if incRequest.Type == "PIPELINE_CREATE" || incRequest.Type == "PIPELINE_START" || incRequest.Type == "PIPELINE_STOP" || incRequest.Type == "PIPELINE_DELETE" { err := handlePipeline(incRequest) if err != nil { @@ -2027,40 +2027,38 @@ func main() { // 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 tenzirUrl == "" { + tenzirUrl = "http://localhost:5160" + log.Printf("[WARNING] SHUFFLE_TENZIR_URL not set, falling back to default URL: %s",tenzirUrl) + } + err := deployTenzirNode() if err != nil{ log.Printf("[ERROR] failed to deploy the pipeline, reason: %s", err) + return err } - // no need of execution arguments for state updates - if incRequest.Type != "PIPELINE_STOP" && len(incRequest.ExecutionArgument) == 0 { + // no need of execution arguments for STOP and DELETE + if (incRequest.Type != "PIPELINE_STOP" && incRequest.Type != "PIPELINE_DELETE") && 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") } - //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) + log.Printf("[INFO] Should delete -> recreate new pipeline with id %#v", identifier) //err := deployPipeline(image, identifier, command) - pipelineId, err := createPipeline(command, identifier) + _, err := createPipeline(command, identifier) if err != nil { log.Printf("[ERROR] Failed to create pipeline: %s", err) return err - } else { - log.Printf("[INFO] Pipeline created 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) + 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) @@ -2070,6 +2068,8 @@ func handlePipeline(incRequest shuffle.ExecutionRequest) error { if err != nil { log.Printf("[ERROR] Failed Deleting Pipeline %s", err) return err + } else { + log.Printf("[INFO] successfully deleted the Pipeline: %s", pipelineId) } } else if incRequest.Type == "PIPELINE_STOP" { log.Printf("[INFO] Should stop the pipeline %#v", identifier) @@ -2078,18 +2078,32 @@ func handlePipeline(incRequest shuffle.ExecutionRequest) error { log.Printf("[ERROR] Failed searching for Pipeline with name %s reason:%s ", identifier, err) return err } - state, err := updatePipelineState(pipelineId, "stop") + _, 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) + + } 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) + _, CreateErr := createPipeline(command, identifier) + return CreateErr + } + log.Printf("[ERROR] Failed searching for Pipeline with name %s reason:%s ", identifier, err) + return err + } + _, err = updatePipelineState(pipelineId, "start") if err != nil { - log.Printf("[DEBUG] failed to save the pipeline data: %s", err) + log.Printf("[ERROR] Failed to start Pipeline: %s reason:%s ", pipelineId, err) + return err } else { - log.Printf("[INFO] succesfully saved the pipeline info ") + log.Printf("[INFO] successfully started the Pipeline: %s", pipelineId) } } else { @@ -2106,7 +2120,7 @@ func deployTenzirNode() error { } ctx := context.Background() - cacheKey := "tenzir-key" + cacheKey := "tenzir-key" imageName := "tenzir/tenzir:latest" containerName := "tenzir-node" @@ -2114,19 +2128,29 @@ func deployTenzirNode() error { _, err := shuffle.GetCache(ctx, cacheKey) if err == nil { - return nil - } + return nil + } containerInfo, err := dockercli.ContainerInspect(ctx, containerName) if err != nil { if dockerclient.IsErrNotFound(err) { - pullOptions := types.ImagePullOptions{} - out, err := dockercli.ImagePull(ctx, imageName, pullOptions) - if err != nil { - log.Printf("[ERROR] Failed to pull the Tenzir image: %s", err) + + // Check if image exists + _, _, err := dockercli.ImageInspectWithRaw(ctx, imageName) + if dockerclient.IsErrNotFound(err) { + log.Printf("[DEBUG] pulling image %s", imageName) + pullOptions := types.ImagePullOptions{} + out, err := dockercli.ImagePull(ctx, imageName, pullOptions) + if err != nil { + log.Printf("[ERROR] Failed to pull the Tenzir image: %s", err) + return err + } + defer out.Close() + + io.Copy(io.Discard, out) + } else if err != nil { return err } - defer out.Close() err = createAndStartTenzirNode(ctx, containerName, imageName, containerStartOptions) if err != nil { @@ -2137,37 +2161,35 @@ func deployTenzirNode() error { } } else { if !containerInfo.State.Running { - log.Printf("[DEBUG] Tenzir Node exists but is not running, starting it") + log.Printf("[DEBUG] Tenzir Node exists but is not running") err := dockercli.ContainerStart(ctx, containerName, containerStartOptions) if err != nil { 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 ...") err = checkTenzirNode() if err != nil { return err } - log.Printf("[INFO] Successfully deployed Tenzir Node!") } } - tenzirStatus := struct { - ContainerStatus string `json:"container_status"` - }{ - ContainerStatus: "running", - } + tenzirStatus := struct { + ContainerStatus string `json:"container_status"` + }{ + ContainerStatus: "running", + } - cacheData, err := json.Marshal(tenzirStatus) - if err != nil { - log.Printf("[WARNING] Failed marshalling execution: %s", err) - } - err = shuffle.SetCache(ctx, cacheKey, cacheData, 1) - if err != nil { + cacheData, err := json.Marshal(tenzirStatus) + if err != nil { + log.Printf("[WARNING] Failed marshalling execution: %s", err) + } + err = shuffle.SetCache(ctx, cacheKey, cacheData, 1) + if err != nil { log.Printf("[WARNING] Failed updating cache for tenzir: %s", err) - } + } return nil } @@ -2264,6 +2286,24 @@ func createPipeline(command, identifier string) (string, error) { log.Printf("[INFO] an existing pipeline found with ID: %s. it will be deleted", pipelineId) toBeDeleted = true } + if strings.Contains(command, "shuffler.io") { + + } else { + var scheme string + if strings.Contains(command, "http://") { + scheme = "http://" + } else if strings.Contains(command, "https://") { + scheme = "https://" + } + + startIndex := strings.Index(command, scheme) + if startIndex != -1 { + endIndex := startIndex + len(scheme) + endIndex += strings.Index(command[endIndex:], "/") + + command = command[:startIndex] + baseUrl + command[endIndex:] + } + } requestBody := map[string]interface{}{ "definition": command, @@ -2271,12 +2311,12 @@ func createPipeline(command, identifier string) (string, error) { "hidden": false, "autostart": map[string]bool{ "created": true, - "completed": false, - "failed": false, + "completed": true, + "failed": true, }, "autodelete": map[string]bool{ "completed": false, - "failed": true, + "failed": false, "stopped": false, }, "retry_delay": "500.0ms", @@ -2348,12 +2388,12 @@ func updatePipelineState(pipelineId, action string) (string, error) { "action": action, "autostart": map[string]bool{ "created": true, - "completed": false, - "failed": false, + "completed": true, + "failed": true, }, "autodelete": map[string]bool{ "completed": false, - "failed": true, + "failed": false, "stopped": false, }, } @@ -2490,53 +2530,53 @@ func searchPipeline(identifier string) (string, error) { return "", errors.New("no existing pipeline found with name") } -func savePipelineData(pipelineId, identifier, status string) error { +// func savePipelineData(pipelineId, identifier, status string) error { - url := fmt.Sprintf("%s/api/v1/triggers/pipeline/save", baseUrl) - identifierWithoutPrefix := strings.TrimPrefix(identifier, "shuffle-") +// url := fmt.Sprintf("%s/api/v1/triggers/pipeline/save", baseUrl) +// identifierWithoutPrefix := strings.TrimPrefix(identifier, "shuffle-") - forwardMethod := "PUT" +// forwardMethod := "PUT" - payload := map[string]interface{}{ - "pipeline_id": pipelineId, - "trigger_id": identifierWithoutPrefix, - "status": status, - } +// payload := map[string]interface{}{ +// "pipeline_id": pipelineId, +// "trigger_id": identifierWithoutPrefix, +// "status": status, +// } - payloadBytes, err := json.Marshal(payload) - if err != nil { - log.Printf("[ERROR] Failed to marshal payload: %s", err) - return err - } +// payloadBytes, err := json.Marshal(payload) +// if err != nil { +// log.Printf("[ERROR] Failed to marshal payload: %s", err) +// return err +// } - forwardData := bytes.NewBuffer(payloadBytes) +// 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") +// 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() +// 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) - } +// 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 -} +// return nil +// } // Is this ok to do with Docker? idk :) func getRunningWorkers(ctx context.Context, workerTimeout int) int {