marker/shuffle-cracked
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #1402 from satti-hari-krishna-reddy/main

Browse Source 
fixing few things to make pipelines show up in the triggers ui
This commit is contained in:
Aditya
2024-05-22 22:06:15 +05:30
committed by GitHub
parent 44e91b0cd3 baf85e0cef
commit 3958d57a98
3 changed files with 33 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frontend/src/views/Admin.jsx
+9 -6
View File
@@ -805,7 +805,7 @@ If you're interested, please let me know a time that works for you, or set up a
.then((responseJson) => { .then((responseJson) => {
setWebHooks(responseJson.webhooks || []); // Handling the case where the result is null or undefined setWebHooks(responseJson.webhooks || []); // Handling the case where the result is null or undefined
setAllSchedules(responseJson.schedules || []); setAllSchedules(responseJson.schedules || []);
// setPipelines(responseJson.pipelines || []); setPipelines(responseJson.pipelines || []);
}) })
.catch((error) => { .catch((error) => {
// toast(error.toString()); // toast(error.toString());
@@ -990,11 +990,13 @@ If you're interested, please let me know a time that works for you, or set up a
} }
const data = { const data = {
command: pipeline.command,
name: pipeline.name, name: pipeline.name,
type: state, type: state,
environment: pipeline.environment, environment: pipeline.environment,
workflow_id: pipeline.workflow_id, workflow_id: pipeline.workflow_id,
trigger_id: pipeline.trigger_id, trigger_id: pipeline.trigger_id,
start_node: pipeline.start_node,
}; };
if (state === "start") toast("starting the pipeline"); if (state === "start") toast("starting the pipeline");
@@ -1025,6 +1027,7 @@ If you're interested, please let me know a time that works for you, or set up a
if (state === "start") toast("Successfully created pipeline"); if (state === "start") toast("Successfully created pipeline");
else toast("Sucessfully stopped the pipeline"); else toast("Sucessfully stopped the pipeline");
} }
setTimeout(handleGetAllTriggers, 1000);
}) })
.catch((error) => { .catch((error) => {
//toast(error.toString()); //toast(error.toString());
@@ -4938,7 +4941,7 @@ If you're interested, please let me know a time that works for you, or set up a
<ListItemText> <ListItemText>
<Button <Button
style={{ marginLeft: "18%" }} style={{ marginLeft: "140px" }}
variant={ variant={
webhook.status === "running" ? "contained" : "outlined" webhook.status === "running" ? "contained" : "outlined"
} }
@@ -4960,10 +4963,10 @@ If you're interested, please let me know a time that works for you, or set up a
</List> </List>
)} )}
{/* <div style={{ marginTop: 20, marginBottom: 20 }}> <div style={{ marginTop: 20, marginBottom: 20 }}>
<h2 style={{ display: "inline" }}>Tenzir Pipelines</h2> <h2 style={{ display: "inline" }}>Tenzir Pipelines</h2>
<span style={{ marginLeft: 25 }}> <span style={{ marginLeft: 25 }}>
Controls a pipeline to run things.{" "} Controls the Tenzir pipeline operations.{" "}
<a <a
target="_blank" target="_blank"
rel="noopener noreferrer" rel="noopener noreferrer"
@@ -5010,7 +5013,7 @@ If you're interested, please let me know a time that works for you, or set up a
primary="Workflow" primary="Workflow"
style={{ maxWidth: 315, minWidth: 315 }} style={{ maxWidth: 315, minWidth: 315 }}
/> />
<ListItemText primary="Actions" /> <ListItemText primary="Actions" style={{ marginLeft: '120px' }} />
</ListItem> </ListItem>
{pipelines.map((pipeline, index) => { {pipelines.map((pipeline, index) => {
var bgColor = "#27292d"; var bgColor = "#27292d";
@@ -5063,7 +5066,7 @@ If you're interested, please let me know a time that works for you, or set up a
); );
})} })}
</List> </List>
)}*/} )}
</div> </div>
) : null; ) : null;
frontend/src/views/AngularWorkflow.jsx
+22 -30
View File
@@ -1362,7 +1362,7 @@ const AngularWorkflow = (defaultprops) => {
const topic = document.getElementById('topic')?.value; const topic = document.getElementById('topic')?.value;
const bootstrapServers = document.getElementById('bootstrap_servers')?.value; const bootstrapServers = document.getElementById('bootstrap_servers')?.value;
const groupId = document.getElementById('group_id')?.value; const groupId = document.getElementById('group_id')?.value;
const autoOffsetReset = document.getElementById('auto_offset_reset')?.value; //const autoOffsetReset = document.getElementById('auto_offset_reset')?.value;
if(topic) { if(topic) {
trigger.parameters.push({ trigger.parameters.push({
@@ -1391,12 +1391,12 @@ const AngularWorkflow = (defaultprops) => {
}); });
} }
if (autoOffsetReset) { // if (autoOffsetReset) {
trigger.parameters.push({ // trigger.parameters.push({
name: "auto_offset_reset", // name: "auto_offset_reset",
value: autoOffsetReset // value: autoOffsetReset
}); // });
} // }
setTenzirConfigModalOpen(false); setTenzirConfigModalOpen(false);
}; };
@@ -7488,12 +7488,6 @@ const AngularWorkflow = (defaultprops) => {
const data = usecase; const data = usecase;
data.start_node = mappedStartnode 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`; const url = `${globalUrl}/api/v1/triggers/pipeline`;
fetch(url, { fetch(url, {
method: "POST", method: "POST",
@@ -7512,12 +7506,8 @@ const AngularWorkflow = (defaultprops) => {
return response.json(); return response.json();
}) })
.then((responseJson) => { .then((responseJson) => {
if (!responseJson.success) { if (!responseJson.success && data.type !== "delete") {
if (responseJson.reason !== undefined) { toast("Failed to set pipeline: " + responseJson.reason);
toast("Failed to set pipeline: " + responseJson.reason);
} else {
toast.error("Failed to stop pipeline")
}
} else { } else {
if (data.type === "create") { if (data.type === "create") {
toast("Pipeline will be created!"); toast("Pipeline will be created!");
@@ -8026,7 +8016,7 @@ const AngularWorkflow = (defaultprops) => {
/* /*
if (trigger.trigger_type === "PIPELINE") { if (trigger.trigger_type === "PIPELINE") {
if (userdata.support !== true) { if (userdata.support !== true) {
return null return null
} }
} }
*/ */
@@ -14381,7 +14371,7 @@ const AngularWorkflow = (defaultprops) => {
const topic = (selectedTrigger?.parameters?.find(param => param.name === "topic")?.value) || '' const topic = (selectedTrigger?.parameters?.find(param => param.name === "topic")?.value) || ''
const bootstrapServers = (selectedTrigger?.parameters?.find(param => param.name === "bootstrap_servers")?.value) || '' const bootstrapServers = (selectedTrigger?.parameters?.find(param => param.name === "bootstrap_servers")?.value) || ''
const groupId = (selectedTrigger?.parameters?.find(param => param.name === "group_id")?.value) || '' const groupId = (selectedTrigger?.parameters?.find(param => param.name === "group_id")?.value) || ''
const autoOffsetReset = (selectedTrigger?.parameters?.find(param => param.name === "auto_offset_reset")?.value) || '' // const autoOffsetReset = (selectedTrigger?.parameters?.find(param => param.name === "auto_offset_reset")?.value) || ''
let command = "from kafka" let command = "from kafka"
if(topic) { if(topic) {
@@ -14402,11 +14392,13 @@ const AngularWorkflow = (defaultprops) => {
} else { } else {
command = `${command},group.id=${selectedTrigger.id}` command = `${command},group.id=${selectedTrigger.id}`
} }
if(autoOffsetReset) { // if(autoOffsetReset) {
command = `${command},auto.offset.reset=${autoOffsetReset}` // command = `${command},auto.offset.reset=${autoOffsetReset}`
} else { // } else {
command = `${command},auto.offset.reset=earliest` // command = `${command},auto.offset.reset=earliest`
}
// }
command = `${command},auto.offset.reset=earliest`
command = `${command},client.id=${selectedTrigger.id},enable.auto.commit=true,auto.commit.interval.ms=1` 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}` command = `${command} read json | to ${globalUrl}/api/v1/pipelines/pipeline_${selectedTrigger.id}`
@@ -19462,8 +19454,8 @@ const AngularWorkflow = (defaultprops) => {
pointerEvents: "auto", pointerEvents: "auto",
color: "white", color: "white",
minWidth: 600, minWidth: 600,
minHeight: 500, minHeight: 450,
maxHeight: 500, maxHeight: 450,
padding: 15, padding: 15,
overflow: "hidden", overflow: "hidden",
zIndex: 10012, zIndex: 10012,
@@ -19532,7 +19524,7 @@ const AngularWorkflow = (defaultprops) => {
placeholder={"tenzir"} placeholder={"tenzir"}
defaultValue={(selectedTrigger?.parameters?.find(param => param.name === "group_id")?.value) || ''} defaultValue={(selectedTrigger?.parameters?.find(param => param.name === "group_id")?.value) || ''}
/> />
<b>auto.offest.reset</b> {/* <b>auto.offest.reset</b>
<TextField <TextField
id="auto_offset_reset" id="auto_offset_reset"
style={{ style={{
@@ -19546,7 +19538,7 @@ const AngularWorkflow = (defaultprops) => {
color="primary" color="primary"
placeholder={"earliest"} placeholder={"earliest"}
defaultValue={(selectedTrigger?.parameters?.find(param => param.name === "auto_offset_reset")?.value) || ''} defaultValue={(selectedTrigger?.parameters?.find(param => param.name === "auto_offset_reset")?.value) || ''}
/> /> */}
</> </>
)} )}
</DialogContent> </DialogContent>
functions/onprem/orborus/orborus.go
+2 -5
View File
@@ -2091,9 +2091,7 @@ func handlePipeline(incRequest shuffle.ExecutionRequest) error {
if err != nil { if err != nil {
log.Printf("[ERROR] Failed Deleting Pipeline %s", err) log.Printf("[ERROR] Failed Deleting Pipeline %s", err)
return err return err
} else { }
log.Printf("[INFO] successfully deleted the Pipeline: %s", pipelineId)
}
} else if incRequest.Type == "PIPELINE_STOP" { } else if incRequest.Type == "PIPELINE_STOP" {
log.Printf("[INFO] Should stop the pipeline %#v", identifier) log.Printf("[INFO] Should stop the pipeline %#v", identifier)
pipelineId, err := searchPipeline(identifier) pipelineId, err := searchPipeline(identifier)
@@ -2301,7 +2299,7 @@ func createPipeline(command, identifier string) (string, error) {
if err != nil { if err != nil {
if strings.Contains(fmt.Sprintf("%s", err), "no existing pipeline found") { 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) log.Printf("[INFO] No existing pipeline found with id: %s. Creating a new one!", identifier)
} else { } else {
log.Printf("[ERROR] Failed to search for existing pipeline but continuing anyway : %s", err) log.Printf("[ERROR] Failed to search for existing pipeline but continuing anyway : %s", err)
} }
@@ -2327,7 +2325,6 @@ func createPipeline(command, identifier string) (string, error) {
command = command[:startIndex] + baseUrl + command[endIndex:] command = command[:startIndex] + baseUrl + command[endIndex:]
} }
} }
requestBody := map[string]interface{}{ requestBody := map[string]interface{}{
"definition": command, "definition": command,
"name": identifier, "name": identifier,