fix(logs pipelines): preview logs with the body the collector sees

The collector gets a normalize pipeline prepended ahead of user pipelines
when use_json_body is on, delivered over opamp, so a pipeline authored
against body.<field> behaved differently in preview than in production.
Preview evaluates the flag for the caller's org and prepends the same
pipeline.

Author: tushar-signoz

pkg/query-service/app/http_handler.go

@@ -3201,7 +3201,19 @@ func (aH *APIHandler) PreviewLogsPipelinesHandler(w http.ResponseWriter, r *http
 		return
 	}
 
-	resultLogs, err := aH.LogsParsingPipelineController.PreviewLogsPipelines(r.Context(), &req)
+	claims, errv2 := authtypes.ClaimsFromContext(r.Context())
+	if errv2 != nil {
+		render.Error(w, errv2)
+		return
+	}
+
+	orgID, errv2 := valuer.NewUUID(claims.OrgID)
+	if errv2 != nil {
+		render.Error(w, errv2)
+		return
+	}
+
+	resultLogs, err := aH.LogsParsingPipelineController.PreviewLogsPipelines(r.Context(), orgID, &req)
 	if err != nil {
 		render.Error(w, err)
 		return

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

@@ -342,13 +342,19 @@ type PipelinesPreviewResponse struct {
 
 func (ic *LogParsingPipelineController) PreviewLogsPipelines(
 	ctx context.Context,
+	orgID valuer.UUID,
 	request *PipelinesPreviewRequest,
 ) (*PipelinesPreviewResponse, error) {
 	pipelines, err := ic.enrichPipelinesFilters(ctx, request.Pipelines)
 	if err != nil {
 		return nil, err
 	}
 
+	// The collector gets the same pipeline prepended over opamp; see RecommendAgentConfig.
+	if ic.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(orgID)) {
+		pipelines = append([]pipelinetypes.GettablePipeline{ic.getNormalizePipeline()}, pipelines...)
+	}
+
 	result, collectorLogs, err := SimulatePipelinesProcessing(ctx, pipelines, request.Logs)
 	if err != nil {
 		return nil, err

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

@@ -6,51 +6,22 @@ import (
 	"testing"
 	"time"
 
+	"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
 	"github.com/SigNoz/signoz/pkg/query-service/model"
 	v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
 	"github.com/SigNoz/signoz/pkg/types/pipelinetypes"
+	"github.com/SigNoz/signoz/pkg/valuer"
 	"github.com/google/uuid"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/entry"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
 func TestPipelinePreview(t *testing.T) {
 	require := require.New(t)
 
 	testPipelines := []pipelinetypes.GettablePipeline{
-		{
-			StoreablePipeline: pipelinetypes.StoreablePipeline{
-				OrderID: 1,
-				Name:    "pipeline1",
-				Alias:   "pipeline1",
-				Enabled: true,
-			},
-			Filter: &v3.FilterSet{
-				Operator: "AND",
-				Items: []v3.FilterItem{
-					{
-						Key: v3.AttributeKey{
-							Key:      "method",
-							DataType: v3.AttributeKeyDataTypeString,
-							Type:     v3.AttributeKeyTypeTag,
-						},
-						Operator: "=",
-						Value:    "GET",
-					},
-				},
-			},
-			Config: []pipelinetypes.PipelineOperator{
-				{
-					OrderId: 1,
-					ID:      "add",
-					Type:    "add",
-					Field:   "attributes.test",
-					Value:   "val",
-					Enabled: true,
-					Name:    "test add",
-				},
-			},
-		},
+		makeTestAddAttributePipeline(),
 		{
 			StoreablePipeline: pipelinetypes.StoreablePipeline{
 				OrderID: 2,
@@ -148,6 +119,93 @@ func TestPipelinePreview(t *testing.T) {
 
 }
 
+func TestPipelinePreviewNormalizesBodyWithJSONBodyEnabled(t *testing.T) {
+	controller := &LogParsingPipelineController{fl: flaggertest.WithUseJSONBody(t, true)}
+
+	result, err := controller.PreviewLogsPipelines(
+		context.Background(),
+		valuer.GenerateUUID(),
+		&PipelinesPreviewRequest{
+			Pipelines: []pipelinetypes.GettablePipeline{makeTestAddAttributePipeline()},
+			Logs: []model.SignozLog{
+				makeTestSignozLog("test log body", map[string]interface{}{"method": "GET"}),
+				makeTestSignozLog(
+					`{"level":"error","msg":"json log body"}`,
+					map[string]interface{}{"method": "GET"},
+				),
+			},
+		},
+	)
+
+	require.NoError(t, err)
+	require.Len(t, result.OutputLogs, 2)
+
+	assert.Equal(t, `{"message":"test log body"}`, result.OutputLogs[0].Body)
+	assert.Equal(
+		t,
+		`{"level":"error","message":"json log body"}`,
+		result.OutputLogs[1].Body,
+	)
+	assert.Equal(t, "val", result.OutputLogs[0].Attributes_string["test"])
+}
+
+func TestPipelinePreviewKeepsBodyAsIsWithJSONBodyDisabled(t *testing.T) {
+	controller := &LogParsingPipelineController{fl: flaggertest.WithUseJSONBody(t, false)}
+
+	result, err := controller.PreviewLogsPipelines(
+		context.Background(),
+		valuer.GenerateUUID(),
+		&PipelinesPreviewRequest{
+			Pipelines: []pipelinetypes.GettablePipeline{makeTestAddAttributePipeline()},
+			Logs: []model.SignozLog{
+				makeTestSignozLog("test log body", map[string]interface{}{"method": "GET"}),
+			},
+		},
+	)
+
+	require.NoError(t, err)
+	require.Len(t, result.OutputLogs, 1)
+
+	assert.Equal(t, "test log body", result.OutputLogs[0].Body)
+	assert.Equal(t, "val", result.OutputLogs[0].Attributes_string["test"])
+}
+
+func makeTestAddAttributePipeline() pipelinetypes.GettablePipeline {
+	return pipelinetypes.GettablePipeline{
+		StoreablePipeline: pipelinetypes.StoreablePipeline{
+			OrderID: 1,
+			Name:    "pipeline1",
+			Alias:   "pipeline1",
+			Enabled: true,
+		},
+		Filter: &v3.FilterSet{
+			Operator: "AND",
+			Items: []v3.FilterItem{
+				{
+					Key: v3.AttributeKey{
+						Key:      "method",
+						DataType: v3.AttributeKeyDataTypeString,
+						Type:     v3.AttributeKeyTypeTag,
+					},
+					Operator: "=",
+					Value:    "GET",
+				},
+			},
+		},
+		Config: []pipelinetypes.PipelineOperator{
+			{
+				OrderId: 1,
+				ID:      "add",
+				Type:    "add",
+				Field:   "attributes.test",
+				Value:   "val",
+				Enabled: true,
+				Name:    "test add",
+			},
+		},
+	}
+}
+
 func TestGrokParsingProcessor(t *testing.T) {
 	require := require.New(t)