fix: update integration tests

pkg/query-service/app/logparsingpipeline/pipelineBuilder.go

@@ -239,16 +239,9 @@ func processJSONParser(parent *pipelinetypes.PipelineOperator) ([]pipelinetypes.
 		return nil, errors.NewInternalf(CodeInvalidOperatorType, "operator type received %s", parent.Type)
 	}
 
-	parseFromNotNilCheck, err := fieldNotNilCheck(parent.ParseFrom)
-	if err != nil {
-		return nil, errors.WrapInvalidInputf(err, CodeFieldNilCheckType,
-			"couldn't generate nil check for parseFrom of json parser op %s: %s", parent.Name, err,
-		)
-	}
-	parent.If = fmt.Sprintf(
-		`%s && ((type(%s) == "string" && isJSON(%s) && type(fromJSON(unquote(%s))) == "map" ) || type(%s) == "map")`,
-		parseFromNotNilCheck, parent.ParseFrom, parent.ParseFrom, parent.ParseFrom, parent.ParseFrom,
-	)
+	// on_error: send_quiet replaces the expensive isJSON `if` check;
+	// parse failures pass the record through unchanged without noisy logs.
+	parent.OnError = signozstanzahelper.SendOnErrorQuiet
 	if parent.EnableFlattening {
 		parent.MaxFlatteningDepth = constants.MaxJSONFlatteningDepth
 	}
@@ -298,7 +291,7 @@ func processJSONParser(parent *pipelinetypes.PipelineOperator) ([]pipelinetypes.
 	}
 
 	// JSONMapping: host
-	err = generateMoveOperators(mapping[pipelinetypes.Host], `resource["host.name"]`)
+	err := generateMoveOperators(mapping[pipelinetypes.Host], `resource["host.name"]`)
 	if err != nil {
 		return nil, err
 	}

pkg/query-service/app/logparsingpipeline/pipelineBuilder_test.go

@@ -324,6 +324,17 @@ func TestNoCollectorErrorsFromProcessorsForMismatchedLogs(t *testing.T) {
 			makeTestLog("mismatching log", map[string]string{
 				"test_json": "bad json",
 			}),
+		}, {
+			"json parser should quietly ignore log with non JSON body",
+			pipelinetypes.PipelineOperator{
+				ID:        "json",
+				Type:      "json_parser",
+				Enabled:   true,
+				Name:      "json parser",
+				ParseFrom: "body",
+				ParseTo:   "attributes",
+			},
+			makeTestLog("plain text log", map[string]string{}),
 		}, {
 			"move parser should ignore non matching logs",
 			pipelinetypes.PipelineOperator{
@@ -894,8 +905,8 @@ func TestProcessJSONParser_WithFlatteningAndMapping(t *testing.T) {
 	require.Equal(t, 1, parentOp.MaxFlatteningDepth)
 	require.Nil(t, parentOp.Mapping) // Mapping should be removed
 	require.Nil(t, parent.Mapping)   // Mapping should be removed
-	require.Contains(t, parentOp.If, `isJSON(body)`)
-	require.Contains(t, parentOp.If, `type(body)`)
+	require.Empty(t, parentOp.If)
+	require.Equal(t, signozstanzahelper.SendOnErrorQuiet, parentOp.OnError)
 
 	require.Equal(t, 1+totalOps, len(ops))
 
@@ -951,7 +962,8 @@ func TestProcessJSONParser_WithoutMapping(t *testing.T) {
 	require.True(t, op.EnableFlattening)
 	require.True(t, op.EnablePaths)
 	require.Equal(t, "parsed", op.PathPrefix)
-	require.Contains(t, op.If, `isJSON(body)`)
+	require.Empty(t, op.If)
+	require.Equal(t, signozstanzahelper.SendOnErrorQuiet, op.OnError)
 }
 
 func TestProcessJSONParser_Simple(t *testing.T) {
@@ -975,7 +987,8 @@ func TestProcessJSONParser_Simple(t *testing.T) {
 	require.False(t, op.EnableFlattening)
 	require.False(t, op.EnablePaths)
 	require.Equal(t, "", op.PathPrefix)
-	require.Contains(t, op.If, `isJSON(body)`)
+	require.Empty(t, op.If)
+	require.Equal(t, signozstanzahelper.SendOnErrorQuiet, op.OnError)
 }
 
 func TestProcessJSONParser_InvalidType(t *testing.T) {

tests/integration/tests/logspipelines/01_logs_pipelines.py

@@ -359,15 +359,28 @@ def test_preview_logs_pipelines_success(
 ) -> None:
     """
     Setup:
-    Create a preview request with a pipeline and sample logs.
+    Preview a json_parser pipeline with one JSON log and one plain-text log.
 
     Tests:
-    1. Send preview request with valid pipeline configuration
-    2. Verify the preview processes logs correctly
-    3. Verify the response contains processed logs
+    1. JSON body gets parsed into attributes
+    2. Non-JSON body passes through unchanged instead of being dropped
     """
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
+    empty_log_fields = {
+        "id": "",
+        "trace_id": "",
+        "span_id": "",
+        "trace_flags": 0,
+        "severity_text": "",
+        "severity_number": 0,
+        "attributes_string": {},
+        "attributes_int": {},
+        "attributes_float": {},
+        "attributes_bool": {},
+        "resources_string": {},
+    }
+
     preview_payload = {
         "pipelines": [
             {
@@ -396,29 +409,25 @@ def test_preview_logs_pipelines_success(
                     {
                         "type": "json_parser",
                         "id": "json-parser-preview",
+                        "orderId": 1,
+                        "enabled": True,
                         "parse_from": "body",
                         "parse_to": "attributes",
-                        "on_error": "send",
                     }
                 ],
             }
         ],
         "logs": [
             {
-                "body": '{"level": "info", "message": "Test log message", "timestamp": "2024-01-01T00:00:00Z"}',
+                "body": '{"level": "info", "message": "json log"}',
                 "timestamp": 1704067200000000000,  # nanoseconds, not milliseconds
-                "id": "",
-                "trace_id": "",
-                "span_id": "",
-                "trace_flags": 0,
-                "severity_text": "",
-                "severity_number": 0,
-                "attributes_string": {},
-                "attributes_int": {},
-                "attributes_float": {},
-                "attributes_bool": {},
-                "resources_string": {"service.name": "test-service"},
-            }
+                **empty_log_fields,
+            },
+            {
+                "body": "plain text log that is not json",
+                "timestamp": 1704067201000000000,
+                **empty_log_fields,
+            },
         ],
     }
 
@@ -435,13 +444,16 @@ def test_preview_logs_pipelines_success(
     assert response.status_code == HTTPStatus.OK
     response_data = response.json()
     assert response_data["status"] == "success"
-    assert "data" in response_data
-    assert "logs" in response_data["data"]
-    assert len(response_data["data"]["logs"]) == 1
+    logs = response_data["data"]["logs"]
+    assert len(logs) == 2
 
-    # Verify the log was processed
-    processed_log = response_data["data"]["logs"][0]
-    assert "attributes_string" in processed_log or "attributes" in processed_log
+    json_log = next(log for log in logs if log["body"].startswith("{"))
+    assert json_log["attributes_string"]["level"] == "info"
+    assert json_log["attributes_string"]["message"] == "json log"
+
+    plain_log = next(log for log in logs if not log["body"].startswith("{"))
+    assert plain_log["body"] == "plain text log that is not json"
+    assert plain_log["attributes_string"] == {}
 
 
 def test_create_multiple_pipelines_success(