made the pipeline to parse json logs

backend/go-app/main.go

@@ -2065,8 +2065,8 @@ func handlePipelineCallback(resp http.ResponseWriter, request *http.Request) {
 		resp.Write([]byte(`{"success": false}`))
 		return
 	}
-// fix this
+
-	parsedBody, err := string(jsonList)
+	parsedBody, err := json.Marshal(jsonList)
 	if err != nil {
 		log.Printf("[ERROR] Failed to marshal jsonList: %s", err)
 		resp.WriteHeader(500)
@@ -2077,7 +2077,7 @@ func handlePipelineCallback(resp http.ResponseWriter, request *http.Request) {
 	newBody := shuffle.ExecutionStruct{
 		Start:             pipeline.StartNode,
 		ExecutionSource:   "pipeline",
-		ExecutionArgument: parsedBody,
+		ExecutionArgument: string(parsedBody),
 	}
 
 	workflow, err := shuffle.GetWorkflow(ctx, pipeline.WorkflowId)
@@ -2131,41 +2131,24 @@ func handlePipelineCallback(resp http.ResponseWriter, request *http.Request) {
 
 func parseConcatenatedJSONLogs(logs string) ([]map[string]interface{}, error) {
     var jsonList []map[string]interface{}
-	var currentObject []rune
+    decoder := json.NewDecoder(strings.NewReader(logs))
-	var depth int
 
-	for _, char := range logs {
+    for decoder.More() {
-		if char == '{' {
-			depth++
-		}
-		if char == '}' {
-			depth--
-		}
-
-		currentObject = append(currentObject, char)
-
-		// When depth is 0, it means we have a complete JSON object but will this work ??
-		if depth == 0 && len(currentObject) > 0 {
         var jsonObject map[string]interface{}
-			err := json.Unmarshal([]byte(string(currentObject)), &jsonObject)
+        if err := decoder.Decode(&jsonObject); err != nil {
-			if err != nil {
+            log.Printf("[WARNING] JSON decoding error: %s. Skipping this object.", err)
-				log.Printf("[WARNING] JSON unmarshal error: %s. Skipping this object.", err)
+            continue
-			} else {
+        }
         jsonList = append(jsonList, jsonObject)
     }
-			currentObject = nil
-		}
-	}
 
-	currentObject = []rune(strings.TrimSpace(string(currentObject)))
+    if err := decoder.Decode(&struct{}{}); err != io.EOF {
-	if len(currentObject) > 0 {
+        return nil, fmt.Errorf("error after decoding all JSON objects: %v", err)
-		log.Printf("[WARNING] Incomplete JSON object found: %s. Skipping this object.", string(currentObject))
     }
 
     return jsonList, nil
 }
 
-
 func executeCloudAction(action shuffle.CloudSyncJob, apikey string) error {
 	data, err := json.Marshal(action)
 	if err != nil {

functions/onprem/orborus/orborus.go

@@ -2462,7 +2462,7 @@ func handlePipeline(incRequest shuffle.ExecutionRequest) error {
 			log.Printf("[ERROR] Failed searching for Pipeline with name %s reason:%s ", identifier, err)
 			return err
 		}
-		_, err = updatePipelineState(pipelineId, "stop")
+		_, err = updatePipelineState(command, pipelineId, "stop")
 		if err != nil {
 			log.Printf("[ERROR] Failed to stop Pipeline: %s reason:%s ", pipelineId, err)
 			return err
@@ -2482,7 +2482,7 @@ func handlePipeline(incRequest shuffle.ExecutionRequest) error {
 			log.Printf("[ERROR] Failed searching for Pipeline with name %s reason:%s ", identifier, err)
 			return err
 		}
-		_, err = updatePipelineState(pipelineId, "start")
+		_, err = updatePipelineState(command, pipelineId, "start")
 		if err != nil {
 			log.Printf("[ERROR] Failed to start Pipeline: %s reason:%s ", pipelineId, err)
 			return err
@@ -2689,7 +2689,11 @@ func createPipeline(command, identifier string) (string, error) {
 	// 	}
 	// }
 
-	command = "from file /var/lib/tenzir/sysmon_logs.ndjson read json | sigma /var/lib/tenzir/rule.yaml | to https://shuffler.io/api/v1/hooks/webhook_d295c43a-e322-4afc-9a59-af167ae7c190"
+	//command = "from file /var/lib/tenzir/sysmon_logs.ndjson read json | sigma /var/lib/tenzir/rule.yaml"
+	//command = "from file /var/lib/tenzir/sysmon_logs.ndjson read json | import"
+	//command = "export | to https://expert-acorn-v6vg4j4j5w7q2wg6g-5001.app.github.dev/api/v1/hooks/webhook_623eab3f-0af4-4d40-abb9-699d9a493411"
+   log.Printf("[HARI] this is the command %s", command)
+
 	requestBody := map[string]interface{}{
 		"definition": command,
 		"name":       identifier,
@@ -2697,7 +2701,7 @@ func createPipeline(command, identifier string) (string, error) {
 		"autostart": map[string]bool{
 			"created":   true,
 			"completed": false,
-			"failed":    true,
+			"failed":    false,
 		},
 		"autodelete": map[string]bool{
 			"completed": false,
@@ -2763,13 +2767,14 @@ func createPipeline(command, identifier string) (string, error) {
 	return id, nil
 }
 
-func updatePipelineState(pipelineId, action string) (string, error) {
+func updatePipelineState(command, pipelineId, action string) (string, error) {
 
 	url := fmt.Sprintf("%s/api/v0/pipeline/update", tenzirUrl)
 	forwardMethod := "POST"
 
 	requestBody := map[string]interface{}{
 		"id":     pipelineId,
+		"definition": command,
 		"action": action,
 		"autostart": map[string]bool{
 			"created":   true,