refactor(logparsingpipeline): remove normalize operator

signozlogsnormalizer in the collector normalizes log bodies, so the server
no longer injects the stanza normalize operator under use_json_body.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

ee/query-service/app/server.go

@@ -56,7 +56,6 @@ func NewServer(config signoz.Config, signoz *signoz.SigNoz) (*Server, error) {
 		signoz.SQLStore,
 		integrationsController.GetPipelinesForInstalledIntegrations,
 		reader,
-		signoz.Flagger,
 	)
 	if err != nil {
 		return nil, err

pkg/query-service/app/http_handler.go

@@ -2688,19 +2688,7 @@ func (aH *APIHandler) PreviewLogsPipelinesHandler(w http.ResponseWriter, r *http
 		return
 	}
 
-	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)
+	resultLogs, err := aH.LogsParsingPipelineController.PreviewLogsPipelines(r.Context(), &req)
 	if err != nil {
 		render.Error(w, err)
 		return

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

@@ -165,22 +165,6 @@ func isNormalizerProcessor(name string) bool {
 	return name == normalizerProcessor || strings.HasPrefix(name, normalizerProcessorPrefix)
 }
 
-// collectorNormalizesFirst reports whether the logs pipeline of the collector config runs
-// signozlogsnormalizer ahead of user pipelines, judged by the first one as in buildCollectorPipelineProcessorsList.
-// Malformed configs report false; generateCollectorConfigWithPipelines surfaces their errors.
-func collectorNormalizesFirst(collectorConf map[string]interface{}) bool {
-	p, err := getOtelPipelineFromConfig(collectorConf)
-	if err != nil || p.Pipelines.Logs == nil {
-		return false
-	}
-	normalizerIdx := slices.IndexFunc(p.Pipelines.Logs.Processors, isNormalizerProcessor)
-	if normalizerIdx < 0 {
-		return false
-	}
-	processorConfigs, _ := collectorConf["processors"].(map[string]interface{})
-	return opampPosition(processorConfigs, p.Pipelines.Logs.Processors[normalizerIdx]) == normalizerBeforeUserPipelines
-}
-
 func opampPosition(processorConfigs map[string]interface{}, name string) string {
 	config, _ := processorConfigs[name].(map[string]interface{})
 	opamp, _ := config["opamp"].(map[string]interface{})

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

@@ -5,12 +5,10 @@ import (
 	"fmt"
 	"testing"
 
-	"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
 	"github.com/SigNoz/signoz/pkg/query-service/constants"
 	"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/smartystreets/goconvey/convey"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -501,61 +499,3 @@ func TestPipeCharInAliasDoesntBreakCollectorConfig(t *testing.T) {
 		}, result[0].Attributes_string,
 	)
 }
-
-func TestRecommendAgentConfigWithNormalizerProcessorAndJSONBody(t *testing.T) {
-	userPipeline := makeTestAddAttributePipeline()
-	controller := &LogParsingPipelineController{
-		fl: flaggertest.WithUseJSONBody(t, true),
-		GetIntegrationPipelines: func(context.Context, string) ([]pipelinetypes.GettablePipeline, error) {
-			return []pipelinetypes.GettablePipeline{userPipeline}, nil
-		},
-	}
-	normalizeOperator := CollectorConfProcessorName(getNormalizePipeline())
-	user := CollectorConfProcessorName(userPipeline)
-
-	testCases := []struct {
-		name       string
-		processors string
-		logs       string
-		expected   []any
-	}{
-		{
-			name:     "NoNormalizer_OperatorPrepended",
-			logs:     "[memory_limiter, batch]",
-			expected: []any{"memory_limiter", normalizeOperator, user, "batch"},
-		},
-		{
-			name:     "NormalizerDefault_OperatorKept",
-			logs:     "[memory_limiter, signozlogsnormalizer, batch]",
-			expected: []any{"memory_limiter", normalizeOperator, user, "signozlogsnormalizer", "batch"},
-		},
-		{
-			name:       "NormalizerBefore_OperatorDropped",
-			processors: "signozlogsnormalizer:\n    opamp:\n      position: before\n",
-			logs:       "[memory_limiter, signozlogsnormalizer, batch]",
-			expected:   []any{"memory_limiter", "signozlogsnormalizer", user, "batch"},
-		},
-		{
-			name:       "NormalizerAfter_OperatorKept",
-			processors: "signozlogsnormalizer:\n    opamp:\n      position: after\n",
-			logs:       "[memory_limiter, signozlogsnormalizer, batch]",
-			expected:   []any{"memory_limiter", normalizeOperator, user, "signozlogsnormalizer", "batch"},
-		},
-	}
-	for _, testCase := range testCases {
-		t.Run(testCase.name, func(t *testing.T) {
-			conf := fmt.Sprintf(
-				"processors:\n  memory_limiter: {}\n  batch: {}\n  %s\nservice:\n  pipelines:\n    logs:\n      receivers: [otlp]\n      processors: %s\n      exporters: [clickhouselogsexporter]\n",
-				testCase.processors, testCase.logs,
-			)
-
-			recommended, _, err := controller.RecommendAgentConfig(valuer.GenerateUUID(), []byte(conf), nil)
-			require.NoError(t, err)
-
-			var recommendedConf map[string]any
-			require.NoError(t, yaml.Unmarshal(recommended, &recommendedConf))
-			logs := recommendedConf["service"].(map[string]any)["pipelines"].(map[string]any)["logs"].(map[string]any)
-			assert.Equal(t, testCase.expected, logs["processors"])
-		})
-	}
-}

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

@@ -11,7 +11,6 @@ import (
 	"github.com/google/uuid"
 
 	"github.com/SigNoz/signoz/pkg/errors"
-	"github.com/SigNoz/signoz/pkg/flagger"
 	"github.com/SigNoz/signoz/pkg/query-service/agentConf"
 	"github.com/SigNoz/signoz/pkg/query-service/constants"
 	"github.com/SigNoz/signoz/pkg/query-service/interfaces"
@@ -20,7 +19,6 @@ import (
 	"github.com/SigNoz/signoz/pkg/query-service/utils"
 	"github.com/SigNoz/signoz/pkg/sqlstore"
 	"github.com/SigNoz/signoz/pkg/types"
-	"github.com/SigNoz/signoz/pkg/types/featuretypes"
 	"github.com/SigNoz/signoz/pkg/types/opamptypes"
 	"github.com/SigNoz/signoz/pkg/types/pipelinetypes"
 	"github.com/SigNoz/signoz/pkg/valuer"
@@ -39,21 +37,18 @@ type LogParsingPipelineController struct {
 	GetIntegrationPipelines func(context.Context, string) ([]pipelinetypes.GettablePipeline, error)
 	// TODO(Piyush): remove with qbv5 migration
 	reader interfaces.Reader
-	fl     flagger.Flagger
 }
 
 func NewLogParsingPipelinesController(
 	sqlStore sqlstore.SQLStore,
 	getIntegrationPipelines func(context.Context, string) ([]pipelinetypes.GettablePipeline, error),
 	reader interfaces.Reader,
-	fl flagger.Flagger,
 ) (*LogParsingPipelineController, error) {
 	repo := NewRepo(sqlStore)
 	return &LogParsingPipelineController{
 		Repo:                    repo,
 		GetIntegrationPipelines: getIntegrationPipelines,
 		reader:                  reader,
-		fl:                      fl,
 	}, nil
 }
 
@@ -220,44 +215,6 @@ func (ic *LogParsingPipelineController) ValidatePipelines(ctx context.Context,
 	return err
 }
 
-// withNormalizePipeline places normalize ahead of user pipelines when queries run on body_v2
-// (use_json_body), so operators see the body the explorer shows. Absent when the flag is off
-// or collectorNormalizesFirst.
-func (ic *LogParsingPipelineController) withNormalizePipeline(ctx context.Context, orgID valuer.UUID, pipelines []pipelinetypes.GettablePipeline, collectorNormalizesFirst bool) []pipelinetypes.GettablePipeline {
-	if collectorNormalizesFirst || !ic.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(orgID)) {
-		return pipelines
-	}
-	return append([]pipelinetypes.GettablePipeline{getNormalizePipeline()}, pipelines...)
-}
-
-func getNormalizePipeline() pipelinetypes.GettablePipeline {
-	return pipelinetypes.GettablePipeline{
-		StoreablePipeline: pipelinetypes.StoreablePipeline{
-			Name:    "Default Pipeline - PreProcessing Body",
-			Alias:   "NormalizeBodyDefault",
-			Enabled: true,
-		},
-		Filter: &v3.FilterSet{
-			Items: []v3.FilterItem{
-				{
-					Key: v3.AttributeKey{
-						Key: "body",
-					},
-					Operator: v3.FilterOperatorExists,
-				},
-			},
-		},
-		Config: []pipelinetypes.PipelineOperator{
-			{
-				ID:      uuid.NewString(),
-				Type:    "normalize",
-				Enabled: true,
-				If:      "body != nil",
-			},
-		},
-	}
-}
-
 // Returns effective list of pipelines including user created
 // pipelines and pipelines for installed integrations
 func (ic *LogParsingPipelineController) getEffectivePipelinesByVersion(
@@ -352,7 +309,6 @@ type PipelinesPreviewResponse struct {
 
 func (ic *LogParsingPipelineController) PreviewLogsPipelines(
 	ctx context.Context,
-	orgID valuer.UUID,
 	request *PipelinesPreviewRequest,
 ) (*PipelinesPreviewResponse, error) {
 	pipelines, err := ic.enrichPipelinesFilters(ctx, request.Pipelines)
@@ -360,11 +316,6 @@ func (ic *LogParsingPipelineController) PreviewLogsPipelines(
 		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{getNormalizePipeline()}, pipelines...)
-	}
-
 	result, collectorLogs, err := SimulatePipelinesProcessing(ctx, pipelines, request.Logs)
 	if err != nil {
 		return nil, err
@@ -383,16 +334,13 @@ func (pc *LogParsingPipelineController) AgentFeatureType() agentConf.AgentFeatur
 
 // Implements agentConf.AgentFeature interface.
 // RecommendAgentConfig generates the collector config to be sent to agents.
-// The normalize pipeline (when use_json_body is on) is placed
-// here, after rawPipelineData is serialized. So it is only present in the config sent to
-// the collector and never persisted to the database as part of the user's pipeline list.
 //
 // NOTE: The configId sent to agents is derived from the pipeline version number
 // (e.g. "LogPipelines:5"), not the YAML content. If server-side logic changes
-// the generated YAML without bumping the version (e.g. toggling the use_json_body
-// flag or updating operator IfExpressions), agents that already
-// applied that version will not re-apply the new config. In such cases, users must save a
-// new pipeline version via the API to force agents to pick up the change.
+// the generated YAML without bumping the version (e.g. updating operator IfExpressions),
+// agents that already applied that version will not re-apply the new config. In such
+// cases, users must save a new pipeline version via the API to force agents to pick up
+// the change.
 func (pc *LogParsingPipelineController) RecommendAgentConfig(
 	orgId valuer.UUID,
 	currentConfYaml []byte,
@@ -418,15 +366,7 @@ func (pc *LogParsingPipelineController) RecommendAgentConfig(
 		return nil, "", err
 	}
 
-	collectorConf, err := unmarshalCollectorConfig(currentConfYaml)
-	if err != nil {
-		return nil, "", err
-	}
-
-	// normalize is only for sending to the collector, never persisted
-	enrichedPipelines = pc.withNormalizePipeline(ctx, orgId, enrichedPipelines, collectorNormalizesFirst(collectorConf))
-
-	updatedConf, err := generateCollectorConfigWithPipelines(collectorConf, enrichedPipelines)
+	updatedConf, err := GenerateCollectorConfigWithPipelines(currentConfYaml, enrichedPipelines)
 	if err != nil {
 		return nil, "", err
 	}

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

@@ -6,11 +6,9 @@ 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"
@@ -119,42 +117,11 @@ func TestPipelinePreview(t *testing.T) {
 
 }
 
-func TestPipelinePreviewNormalizesBodyWithJSONBodyEnabled(t *testing.T) {
-	controller := &LogParsingPipelineController{fl: flaggertest.WithUseJSONBody(t, true)}
+func TestPipelinePreviewKeepsBodyAsIs(t *testing.T) {
+	controller := &LogParsingPipelineController{}
 
 	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{

pkg/query-service/app/server.go

@@ -46,7 +46,6 @@ func NewServer(config signoz.Config, signoz *signoz.SigNoz) (*Server, error) {
 		signoz.SQLStore,
 		integrationsController.GetPipelinesForInstalledIntegrations,
 		reader,
-		signoz.Flagger,
 	)
 	if err != nil {
 		return nil, err

tests/fixtures/logs.py

@@ -174,7 +174,7 @@ class Logs(ABC):
             self.body_v2 = body_v2
         else:
             # Try to parse body as JSON; if successful use it directly,
-            # otherwise wrap as {"message": body} matching the normalize operator behavior.
+            # otherwise wrap as {"message": body} matching signozlogsnormalizer.
             try:
                 json.loads(body)
                 self.body_v2 = body