From a3d43702d5a1884b3129cb9075a5559135231b28 Mon Sep 17 00:00:00 2001 From: satti-hari-krishna-reddy Date: Sun, 19 May 2024 23:07:48 +0530 Subject: [PATCH 1/7] enhancing the ui for kafka pipeline --- frontend/src/views/AngularWorkflow.jsx | 722 ++++++++++++++++--------- 1 file changed, 469 insertions(+), 253 deletions(-) diff --git a/frontend/src/views/AngularWorkflow.jsx b/frontend/src/views/AngularWorkflow.jsx index 3f25507b..f825f18f 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 { @@ -250,7 +251,6 @@ export const triggers = [ id: "", }, ]; - // Adds specific text to items // https://stackoverflow.com/questions/19014250/rerender-view-on-browser-resize-with-react @@ -551,6 +551,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, @@ -1276,6 +1280,50 @@ 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 + }); + } + + if (bootstrapServers) { + trigger.parameters.push({ + name: "bootstrap_servers", + value: bootstrapServers + }); + } + + 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 @@ -7231,47 +7279,55 @@ 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!"); + + 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"); @@ -7279,11 +7335,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"); @@ -7761,7 +7816,7 @@ const AngularWorkflow = (defaultprops) => { if (trigger.trigger_type === "PIPELINE") { if (userdata.support !== true) { - return null + // return null } } @@ -7826,26 +7881,31 @@ const AngularWorkflow = (defaultprops) => { > {imageline} {isMobile ? null : - + + -

+

{trigger.name}

- + {trigger.description} @@ -8784,11 +8844,11 @@ const AngularWorkflow = (defaultprops) => { } if (app.trigger_type === "PIPELINE" && userdata.support !== true) { - return null + //return null } if (app.id === "integration" && userdata.support !== true) { - return null + //return null } var extraMessage = "" @@ -13692,226 +13752,241 @@ const AngularWorkflow = (defaultprops) => { return null } - 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 :
@@ -14048,7 +14123,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="" @@ -18749,6 +18824,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) { @@ -19565,6 +19780,7 @@ const AngularWorkflow = (defaultprops) => { {codePopoutModal} {workflowRevisions} {authenticationModal} + {tenzirConfigModal} {/*editWorkflowModal*/} {executionArgumentModal} {configureWorkflowModal} From b10831970099af7464c38e67f6127f6584d2babe Mon Sep 17 00:00:00 2001 From: satti-hari-krishna-reddy Date: Tue, 21 May 2024 12:53:36 +0530 Subject: [PATCH 2/7] making start and stop buttons to work in PipelineSidebar --- frontend/src/views/AngularWorkflow.jsx | 103 +++++++++++++++++++++---- 1 file changed, 88 insertions(+), 15 deletions(-) diff --git a/frontend/src/views/AngularWorkflow.jsx b/frontend/src/views/AngularWorkflow.jsx index f825f18f..0aa2f38d 100755 --- a/frontend/src/views/AngularWorkflow.jsx +++ b/frontend/src/views/AngularWorkflow.jsx @@ -1298,6 +1298,9 @@ const AngularWorkflow = (defaultprops) => { name: "topic", value: topic }); + } else { + toast("please enter the topic name"); + return; } if (bootstrapServers) { @@ -1305,6 +1308,9 @@ const AngularWorkflow = (defaultprops) => { name: "bootstrap_servers", value: bootstrapServers }); + } else { + toast("please enter bootstrap server details"); + return; } if (groupId) { @@ -6895,6 +6901,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); } } @@ -7293,8 +7313,14 @@ const AngularWorkflow = (defaultprops) => { } } const data = usecase; - if (data.type === "create") toast("Creating pipeline"); - else toast("stopping pipeline"); + data.start_node = mappedStartnode + + if (data.type === "create") { + toast("Creating pipeline"); + } else if (data.type === "stop") { + toast("stopping pipeline"); + } + const url = `${globalUrl}/api/v1/triggers/pipeline`; fetch(url, { method: "POST", @@ -7316,8 +7342,14 @@ const AngularWorkflow = (defaultprops) => { if (!responseJson.success) { toast("Failed to set pipeline: " + responseJson.reason); } else { - if (data.type === "create") toast("Pipeline will be created!"); - else toast("Pipeline will be stopped!"); + 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, @@ -13947,24 +13979,46 @@ const AngularWorkflow = (defaultprops) => { disabled={selectedTrigger.status === "running"} onClick={() => { - 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; + const topic = (selectedTrigger?.parameters?.find(param => param.name === "topic")?.value) || '' + const bootstrapServers = (selectedTrigger?.parameters?.find(param => param.name === "bootstrap_servers")?.value) || '' + const groupId = (selectedTrigger?.parameters?.find(param => param.name === "group_id")?.value) || '' + const autoOffsetReset = (selectedTrigger?.parameters?.find(param => param.name === "auto_offset_reset")?.value) || '' let command = "from kafka" - if(topic) command = `${command} --topic ${topic}` - if(bootstrapServers) command = `${command} --set bootstrap.servers=${bootstrapServers}` - if(groupId) command = `${command},group.id=${groupId}` - if(autoOffsetReset) command = `${command},auto.offset.reset=${autoOffsetReset}` + if(topic) { + command = `${command} -t ${topic}` + } else { + toast("please enter the topic name") + return; + } + if(bootstrapServers) { + command = `${command} -e -o stored -X bootstrap.servers=${bootstrapServers}` + } else { + toast("please enter the bootstrap servers details") + return; + } + + if(groupId) { + command = `${command},group.id=${groupId}` + } else { + command = `${command},group.id=${selectedTrigger.id}` + } + if(autoOffsetReset) { + command = `${command},auto.offset.reset=${autoOffsetReset}` + } else { + command = `${command},auto.offset.reset=earliest` + } + command = `${command},client.id=${selectedTrigger.id},enable.auto.commit=true,auto.commit.interval.ms=1` + command = `${command} read json | to ${globalUrl}/api/v1/pipelines/pipeline_${selectedTrigger.id}` const pipelineConfig = { command: command, name: selectedTrigger.label, - type: "start", + type: "create", environment: selectedTrigger.environment, workflow_id: workflow.id, trigger_id: selectedTrigger.id, + start_node: "", }; submitPipeline(selectedTrigger, selectedTriggerIndex, pipelineConfig); }} @@ -13977,7 +14031,15 @@ const AngularWorkflow = (defaultprops) => { variant="contained" disabled={selectedTrigger.status !== "running"} onClick={() => { - toast("Should stop trigger"); + const pipelineConfig = { + name: selectedTrigger.label, + type: "stop", + environment: selectedTrigger.environment, + workflow_id: workflow.id, + trigger_id: selectedTrigger.id, + start_node: "", + }; + submitPipeline(selectedTrigger, selectedTriggerIndex, pipelineConfig); }} color="primary" > @@ -15902,7 +15964,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 && From 646d69fcdcb6c2187f279a7192194d8b89c55894 Mon Sep 17 00:00:00 2001 From: satti-hari-krishna-reddy Date: Tue, 21 May 2024 12:59:12 +0530 Subject: [PATCH 3/7] made pipeline an actual trigger --- backend/go-app/main.go | 141 ++++++++++++++++++++++++++++++++++++++++- 1 file changed, 140 insertions(+), 1 deletion(-) diff --git a/backend/go-app/main.go b/backend/go-app/main.go index 9db78ed0..a08a27e1 100755 --- a/backend/go-app/main.go +++ b/backend/go-app/main.go @@ -1973,6 +1973,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 { @@ -4909,7 +5047,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") From ac66b9b623a6ee396254e6dbce355c2f9f77d7ad Mon Sep 17 00:00:00 2001 From: satti-hari-krishna-reddy Date: Tue, 21 May 2024 12:59:55 +0530 Subject: [PATCH 4/7] removed localhost from the url --- .env | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 From 1f09cce1ddfd2b8be3486bee8133e92d8c79671e Mon Sep 17 00:00:00 2001 From: satti-hari-krishna-reddy Date: Tue, 21 May 2024 14:06:32 +0530 Subject: [PATCH 5/7] reverting styling changes made to TriggersView --- frontend/src/views/AngularWorkflow.jsx | 35 +++++++++++--------------- 1 file changed, 15 insertions(+), 20 deletions(-) diff --git a/frontend/src/views/AngularWorkflow.jsx b/frontend/src/views/AngularWorkflow.jsx index 0aa2f38d..83970ed6 100755 --- a/frontend/src/views/AngularWorkflow.jsx +++ b/frontend/src/views/AngularWorkflow.jsx @@ -7848,7 +7848,7 @@ const AngularWorkflow = (defaultprops) => { if (trigger.trigger_type === "PIPELINE") { if (userdata.support !== true) { - // return null + return null } } @@ -7913,31 +7913,26 @@ const AngularWorkflow = (defaultprops) => { >
{imageline} {isMobile ? null : - - + -

+

{trigger.name}

- + {trigger.description} @@ -8876,11 +8871,11 @@ const AngularWorkflow = (defaultprops) => { } if (app.trigger_type === "PIPELINE" && userdata.support !== true) { - //return null + return null } if (app.id === "integration" && userdata.support !== true) { - //return null + return null } var extraMessage = "" @@ -13784,7 +13779,7 @@ const AngularWorkflow = (defaultprops) => { return null } - const PipelineSidebar = Object.getOwnPropertyNames(selectedTrigger).length === 0 || workflow.triggers[selectedTriggerIndex] === undefined && selectedTrigger.trigger_type !== "SCHEDULE" ? null : + 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} From abef955448609aa7b5944095fc56d1b8b8c46eb0 Mon Sep 17 00:00:00 2001 From: satti-hari-krishna-reddy Date: Tue, 21 May 2024 14:09:02 +0530 Subject: [PATCH 6/7] added image pull checking and removed few log messages --- functions/onprem/orborus/orborus.go | 211 ++++++++++++++++------------ 1 file changed, 125 insertions(+), 86 deletions(-) diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index 6a8e260e..dc9e554c 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 } @@ -2265,18 +2287,35 @@ func createPipeline(command, identifier string) (string, error) { toBeDeleted = true } + if strings.Contains(command, "kafka") { + 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, "name": identifier, "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 +2387,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 +2529,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 { From 2efa051f68d0d3c62bb941b8292e3429c379f5a5 Mon Sep 17 00:00:00 2001 From: satti-hari-krishna-reddy Date: Tue, 21 May 2024 15:31:21 +0530 Subject: [PATCH 7/7] added a conditional check to not replace the base url if it shuffler.io --- functions/onprem/orborus/orborus.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index dc9e554c..cdface09 100755 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -2286,8 +2286,9 @@ 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") { - if strings.Contains(command, "kafka") { + } else { var scheme string if strings.Contains(command, "http://") { scheme = "http://"