fix: update integration tests

* fix(dashboard): count panel stats from the v2 spec

The panel counters walked a top-level `widgets` array and returned early
when the key was missing. A v2 dashboard stores only `metadata` and
`spec`, with panels as a map under `spec.panels`, so every v2 row hit
that early return and all `dashboard.panels.*` stats stayed at zero.
`dashboard.count` was unaffected — it is a row count.

Read the v2 spec instead: count the panels under `spec.panels` and take
each panel's signal from its query envelope, reusing the typed read path
and `QueryEnvelope.GetSignal`. Signal-less queries (promql, clickhouse
sql, formulas) count towards the panel total only. v1 rows are no longer
parsed for panel stats and contribute to `dashboard.count` alone.

* test(dashboard): drop the constant name arg from the stats query helper

statsBuilderQuery only ever received "A", which go-lint flags via unparam.
A panel holds a single query, so the name never mattered to the assertions;
the composite test still names its sub-queries through statsBuilderQuerySpec.

* refactor(dashboard): move v2 panel stats to a perses_ file

All v2 code lives in perses_-prefixed files until the v1 code goes away.
Pure move of the stats block out of dashboard.go, tests alongside it.

* fix(dashboard): count create-v2 stats off the postable spec

CreateV2 already holds the postable dashboard, so decoding the storable
back into a v2 dashboard just to count its panels was a needless type
conversion on the create path.

Split the panel walk into addPanelStats over a DashboardSpec, and add
NewStatsFromPostableDashboardV2 for the create path. The storable variant
keeps its signature for the periodic collectors, which only have rows.

pkg/modules/dashboard/impldashboard/v2_module.go

@@ -20,7 +20,6 @@ func (m *module) CreateV2(ctx context.Context, orgID valuer.UUID, createdBy stri
 	}
 
 	dashboard := postable.NewDashboardV2(orgID, createdBy, source)
-	var storableDashboard *dashboardtypes.StorableDashboard
 
 	err := m.store.RunInTx(ctx, func(ctx context.Context) error {
 		resolvedTags, err := m.tagModule.SyncTags(ctx, orgID, coretypes.KindDashboard, dashboard.ID, postable.Tags)
@@ -33,14 +32,13 @@ func (m *module) CreateV2(ctx context.Context, orgID valuer.UUID, createdBy stri
 		if err != nil {
 			return err
 		}
-		storableDashboard = storable
 		return m.store.Create(ctx, storable)
 	})
 	if err != nil {
 		return nil, err
 	}
 
-	m.analytics.TrackUser(ctx, orgID.String(), creator.String(), "Dashboard Created", dashboardtypes.NewStatsFromStorableDashboards([]*dashboardtypes.StorableDashboard{storableDashboard}))
+	m.analytics.TrackUser(ctx, orgID.String(), creator.String(), "Dashboard Created", dashboardtypes.NewStatsFromPostableDashboardV2(postable))
 	return dashboard, nil
 }
 

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) {

pkg/types/dashboardtypes/dashboard.go

@@ -176,69 +176,6 @@ func NewGettableDashboardFromDashboard(dashboard *Dashboard) (*GettableDashboard
 	}, nil
 }
 
-func NewStatsFromStorableDashboards(dashboards []*StorableDashboard) map[string]any {
-	stats := make(map[string]any)
-	stats["dashboard.panels.count"] = int64(0)
-	stats["dashboard.panels.traces.count"] = int64(0)
-	stats["dashboard.panels.metrics.count"] = int64(0)
-	stats["dashboard.panels.logs.count"] = int64(0)
-	for _, dashboard := range dashboards {
-		addStatsFromStorableDashboard(dashboard, stats)
-	}
-
-	stats["dashboard.count"] = int64(len(dashboards))
-	return stats
-}
-
-func addStatsFromStorableDashboard(dashboard *StorableDashboard, stats map[string]any) {
-	if dashboard.Data == nil {
-		return
-	}
-
-	if dashboard.Data["widgets"] == nil {
-		return
-	}
-
-	widgets, ok := dashboard.Data["widgets"]
-	if !ok {
-		return
-	}
-
-	data, ok := widgets.([]interface{})
-	if !ok {
-		return
-	}
-
-	for _, widget := range data {
-		sData, ok := widget.(map[string]interface{})
-		if ok && sData["query"] != nil {
-			stats["dashboard.panels.count"] = stats["dashboard.panels.count"].(int64) + 1
-			query, ok := sData["query"].(map[string]interface{})
-			if ok && query["queryType"] == "builder" && query["builder"] != nil {
-				builderData, ok := query["builder"].(map[string]interface{})
-				if ok && builderData["queryData"] != nil {
-					builderQueryData, ok := builderData["queryData"].([]interface{})
-					if ok {
-						for _, queryData := range builderQueryData {
-							data, ok := queryData.(map[string]interface{})
-							if ok {
-								switch data["dataSource"] {
-								case "traces":
-									stats["dashboard.panels.traces.count"] = stats["dashboard.panels.traces.count"].(int64) + 1
-								case "metrics":
-									stats["dashboard.panels.metrics.count"] = stats["dashboard.panels.metrics.count"].(int64) + 1
-								case "logs":
-									stats["dashboard.panels.logs.count"] = stats["dashboard.panels.logs.count"].(int64) + 1
-								}
-							}
-						}
-					}
-				}
-			}
-		}
-	}
-}
-
 func (storableDashboardData *StorableDashboardData) GetWidgetIds() []string {
 	data := *storableDashboardData
 	widgetIds := []string{}

pkg/types/dashboardtypes/perses_dashboard_stats.go

@@ -0,0 +1,90 @@
+package dashboardtypes
+
+import (
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+)
+
+const (
+	statKeyDashboardCount    = "dashboard.count"
+	statKeyPanelCount        = "dashboard.panels.count"
+	statKeyPanelTracesCount  = "dashboard.panels.traces.count"
+	statKeyPanelMetricsCount = "dashboard.panels.metrics.count"
+	statKeyPanelLogsCount    = "dashboard.panels.logs.count"
+)
+
+// panelSignalStatKeys maps a builder query's signal to the stat it contributes
+// to. Signal-less queries (promql, clickhouse sql, formulas) count towards the
+// panel total only.
+var panelSignalStatKeys = map[telemetrytypes.Signal]string{
+	telemetrytypes.SignalTraces:  statKeyPanelTracesCount,
+	telemetrytypes.SignalMetrics: statKeyPanelMetricsCount,
+	telemetrytypes.SignalLogs:    statKeyPanelLogsCount,
+}
+
+// NewStatsFromStorableDashboards reports the stats of stored dashboards. Rows that
+// do not decode as v2 contribute to dashboard.count only.
+func NewStatsFromStorableDashboards(dashboards []*StorableDashboard) map[string]any {
+	stats := newPanelStats()
+
+	for _, dashboard := range dashboards {
+		if dashboard == nil {
+			continue
+		}
+		dashboardV2, err := dashboard.ToDashboardV2(nil)
+		if err != nil {
+			continue
+		}
+		addPanelStats(&dashboardV2.Spec, stats)
+	}
+
+	stats[statKeyDashboardCount] = int64(len(dashboards))
+	return stats
+}
+
+// NewStatsFromPostableDashboardV2 reports the stats of a dashboard as it is
+// created, straight off the postable spec — the create path has no reason to make
+// a storable round-trip just to be counted.
+func NewStatsFromPostableDashboardV2(postable PostableDashboardV2) map[string]any {
+	stats := newPanelStats()
+	addPanelStats(&postable.Spec, stats)
+
+	stats[statKeyDashboardCount] = int64(1)
+	return stats
+}
+
+func newPanelStats() map[string]any {
+	return map[string]any{
+		statKeyPanelCount:        int64(0),
+		statKeyPanelTracesCount:  int64(0),
+		statKeyPanelMetricsCount: int64(0),
+		statKeyPanelLogsCount:    int64(0),
+	}
+}
+
+// addPanelStats counts the panels of a v2 spec, and each panel's queries against
+// the signal they read.
+func addPanelStats(spec *DashboardSpec, stats map[string]any) {
+	for _, panel := range spec.Panels {
+		if panel == nil {
+			continue
+		}
+		incrementStat(stats, statKeyPanelCount)
+
+		for _, query := range panel.Spec.Queries {
+			composite, err := query.Spec.Plugin.buildV5CompositeQueryFromPlugin()
+			if err != nil {
+				continue
+			}
+			for _, envelope := range composite.Queries {
+				if key, ok := panelSignalStatKeys[envelope.GetSignal()]; ok {
+					incrementStat(stats, key)
+				}
+			}
+		}
+	}
+}
+
+func incrementStat(stats map[string]any, key string) {
+	count, _ := stats[key].(int64)
+	stats[key] = count + 1
+}

pkg/types/dashboardtypes/perses_dashboard_stats_test.go

@@ -0,0 +1,220 @@
+package dashboardtypes
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/SigNoz/signoz/pkg/types"
+	"github.com/SigNoz/signoz/pkg/valuer"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func statsSpecJSON(panelsJSON string) string {
+	return `{
+		"display": {"name": "Stats Dashboard"},
+		"variables": [],
+		"panels": {` + panelsJSON + `},
+		"layouts": [],
+		"links": []
+	}`
+}
+
+// newStatsStorableV2 builds a stored v2 row from a panels JSON fragment, going
+// through the untyped data blob the way a row read off the DB does.
+func newStatsStorableV2(t *testing.T, panelsJSON string) *StorableDashboard {
+	t.Helper()
+
+	raw := `{
+		"metadata": {"schemaVersion": "` + SchemaVersion + `"},
+		"spec": ` + statsSpecJSON(panelsJSON) + `
+	}`
+
+	var data StorableDashboardData
+	require.NoError(t, json.Unmarshal([]byte(raw), &data))
+
+	return &StorableDashboard{
+		Identifiable: types.Identifiable{ID: valuer.GenerateUUID()},
+		OrgID:        valuer.GenerateUUID(),
+		Source:       SourceUser,
+		Name:         "stats-dashboard",
+		Data:         data,
+	}
+}
+
+func newStatsPostableV2(t *testing.T, panelsJSON string) PostableDashboardV2 {
+	t.Helper()
+
+	var spec DashboardSpec
+	require.NoError(t, json.Unmarshal([]byte(statsSpecJSON(panelsJSON)), &spec))
+
+	return PostableDashboardV2{
+		DashboardV2MetadataBase: DashboardV2MetadataBase{SchemaVersion: SchemaVersion},
+		Name:                    "stats-dashboard",
+		Spec:                    spec,
+	}
+}
+
+func statsPanel(queriesJSON string) string {
+	return `{
+		"kind": "Panel",
+		"spec": {
+			"links": [],
+			"plugin": {"kind": "signoz/TimeSeriesPanel", "spec": {}},
+			"queries": [` + queriesJSON + `]
+		}
+	}`
+}
+
+// A panel holds a single query, so its name never matters to the assertions.
+func statsBuilderQuery(signal string) string {
+	return `{
+		"kind": "time_series",
+		"spec": {"plugin": {"kind": "signoz/BuilderQuery", "spec": ` + statsBuilderQuerySpec("A", signal) + `}}
+	}`
+}
+
+func statsBuilderQuerySpec(name, signal string) string {
+	aggregations := `[{"expression": "count()"}]`
+	if signal == "metrics" {
+		aggregations = `[{"metricName": "m", "timeAggregation": "rate", "spaceAggregation": "sum"}]`
+	}
+	return `{"name": "` + name + `", "signal": "` + signal + `", "aggregations": ` + aggregations + `}`
+}
+
+func TestNewStatsFromStorableDashboardsCountsV2Panels(t *testing.T) {
+	dashboard := newStatsStorableV2(t, `
+		"p1": `+statsPanel(statsBuilderQuery("logs"))+`,
+		"p2": `+statsPanel(statsBuilderQuery("metrics"))+`,
+		"p3": `+statsPanel(statsBuilderQuery("traces"))+`
+	`)
+
+	stats := NewStatsFromStorableDashboards([]*StorableDashboard{dashboard})
+
+	assert.Equal(t, int64(1), stats[statKeyDashboardCount])
+	assert.Equal(t, int64(3), stats[statKeyPanelCount])
+	assert.Equal(t, int64(1), stats[statKeyPanelLogsCount])
+	assert.Equal(t, int64(1), stats[statKeyPanelMetricsCount])
+	assert.Equal(t, int64(1), stats[statKeyPanelTracesCount])
+}
+
+// A panel carries exactly one query envelope, so multi-signal panels arrive as a
+// composite: the panel counts once and every builder sub-query counts its signal.
+func TestNewStatsFromStorableDashboardsCountsCompositeSubQueries(t *testing.T) {
+	composite := `{
+		"kind": "time_series",
+		"spec": {"plugin": {"kind": "signoz/CompositeQuery", "spec": {"queries": [
+			{"type": "builder_query", "spec": ` + statsBuilderQuerySpec("A", "traces") + `},
+			{"type": "builder_query", "spec": ` + statsBuilderQuerySpec("B", "logs") + `}
+		]}}}
+	}`
+	dashboard := newStatsStorableV2(t, `"p1": `+statsPanel(composite))
+
+	stats := NewStatsFromStorableDashboards([]*StorableDashboard{dashboard})
+
+	assert.Equal(t, int64(1), stats[statKeyPanelCount])
+	assert.Equal(t, int64(1), stats[statKeyPanelTracesCount])
+	assert.Equal(t, int64(1), stats[statKeyPanelLogsCount])
+}
+
+// promql and clickhouse queries carry no signal, so they land in the panel total
+// and nowhere else.
+func TestNewStatsFromStorableDashboardsIgnoresSignallessQueries(t *testing.T) {
+	promql := `{
+		"kind": "time_series",
+		"spec": {"plugin": {"kind": "signoz/PromQLQuery", "spec": {"name": "A", "query": "up"}}}
+	}`
+	dashboard := newStatsStorableV2(t, `"p1": `+statsPanel(promql))
+
+	stats := NewStatsFromStorableDashboards([]*StorableDashboard{dashboard})
+
+	assert.Equal(t, int64(1), stats[statKeyPanelCount])
+	assert.Equal(t, int64(0), stats[statKeyPanelTracesCount])
+	assert.Equal(t, int64(0), stats[statKeyPanelMetricsCount])
+	assert.Equal(t, int64(0), stats[statKeyPanelLogsCount])
+}
+
+func TestNewStatsFromStorableDashboardsAggregatesAcrossDashboards(t *testing.T) {
+	first := newStatsStorableV2(t, `"p1": `+statsPanel(statsBuilderQuery("logs")))
+	second := newStatsStorableV2(t, `
+		"p1": `+statsPanel(statsBuilderQuery("logs"))+`,
+		"p2": `+statsPanel(statsBuilderQuery("traces"))+`
+	`)
+
+	stats := NewStatsFromStorableDashboards([]*StorableDashboard{first, second})
+
+	assert.Equal(t, int64(2), stats[statKeyDashboardCount])
+	assert.Equal(t, int64(3), stats[statKeyPanelCount])
+	assert.Equal(t, int64(2), stats[statKeyPanelLogsCount])
+	assert.Equal(t, int64(1), stats[statKeyPanelTracesCount])
+}
+
+// v1 rows are counted as dashboards but contribute no panel stats — the counters
+// read the v2 spec only.
+func TestNewStatsFromStorableDashboardsSkipsNonV2Rows(t *testing.T) {
+	v1 := &StorableDashboard{
+		Identifiable: types.Identifiable{ID: valuer.GenerateUUID()},
+		OrgID:        valuer.GenerateUUID(),
+		Source:       SourceUser,
+		Name:         "legacy-dashboard",
+		Data: StorableDashboardData{
+			"title":   "Legacy Title",
+			"version": "v5",
+			"widgets": []any{
+				map[string]any{"query": map[string]any{
+					"queryType": "builder",
+					"builder": map[string]any{
+						"queryData": []any{map[string]any{"dataSource": "logs"}},
+					},
+				}},
+			},
+		},
+	}
+	empty := &StorableDashboard{
+		Identifiable: types.Identifiable{ID: valuer.GenerateUUID()},
+		OrgID:        valuer.GenerateUUID(),
+		Source:       SourceUser,
+		Name:         "bare",
+	}
+
+	stats := NewStatsFromStorableDashboards([]*StorableDashboard{v1, empty})
+
+	assert.Equal(t, int64(2), stats[statKeyDashboardCount])
+	assert.Equal(t, int64(0), stats[statKeyPanelCount])
+	assert.Equal(t, int64(0), stats[statKeyPanelLogsCount])
+}
+
+// The create path counts off the postable spec, so it never round-trips through a
+// storable to be counted.
+func TestNewStatsFromPostableDashboardV2(t *testing.T) {
+	postable := newStatsPostableV2(t, `
+		"p1": `+statsPanel(statsBuilderQuery("logs"))+`,
+		"p2": `+statsPanel(statsBuilderQuery("traces"))+`
+	`)
+
+	stats := NewStatsFromPostableDashboardV2(postable)
+
+	assert.Equal(t, int64(1), stats[statKeyDashboardCount])
+	assert.Equal(t, int64(2), stats[statKeyPanelCount])
+	assert.Equal(t, int64(1), stats[statKeyPanelLogsCount])
+	assert.Equal(t, int64(1), stats[statKeyPanelTracesCount])
+	assert.Equal(t, int64(0), stats[statKeyPanelMetricsCount])
+}
+
+func TestNewStatsFromPostableDashboardV2WithNoPanels(t *testing.T) {
+	stats := NewStatsFromPostableDashboardV2(newStatsPostableV2(t, ``))
+
+	assert.Equal(t, int64(1), stats[statKeyDashboardCount])
+	assert.Equal(t, int64(0), stats[statKeyPanelCount])
+	assert.Equal(t, int64(0), stats[statKeyPanelLogsCount])
+}
+
+func TestNewStatsFromStorableDashboardsWithNoDashboards(t *testing.T) {
+	stats := NewStatsFromStorableDashboards(nil)
+
+	assert.Equal(t, int64(0), stats[statKeyDashboardCount])
+	assert.Equal(t, int64(0), stats[statKeyPanelCount])
+	assert.Equal(t, int64(0), stats[statKeyPanelTracesCount])
+	assert.Equal(t, int64(0), stats[statKeyPanelMetricsCount])
+	assert.Equal(t, int64(0), stats[statKeyPanelLogsCount])
+}

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(