refactor(logparsingpipeline): unmarshal collector config once

User pipelines anchor on the first normalizer per its opamp.position; the prepended normalize operator is dropped when the processor already runs first.

pkg/flagger/registry.go

@@ -9,7 +9,6 @@ var (
 	FeaturePutMetersInZeus        = featuretypes.MustNewName("put_meters_in_zeus")
 	FeatureUseMeterReporter       = featuretypes.MustNewName("use_meter_reporter")
 	FeatureUseJSONBody            = featuretypes.MustNewName("use_json_body")
-	FeatureJSONBodyDualIngestion  = featuretypes.MustNewName("json_body_dual_ingestion")
 	FeatureEnableMetricsReduction = featuretypes.MustNewName("enable_metrics_reduction")
 	FeatureResolveSemconvFamilies = featuretypes.MustNewName("resolve_semconv_families")
 	FeatureUseTraceAttributesJSON = featuretypes.MustNewName("use_trace_attributes_json")
@@ -65,14 +64,6 @@ func MustNewRegistry() featuretypes.Registry {
 			DefaultVariant: featuretypes.MustNewName("disabled"),
 			Variants:       featuretypes.NewBooleanVariants(),
 		},
-		&featuretypes.Feature{
-			Name:           FeatureJSONBodyDualIngestion,
-			Kind:           featuretypes.KindBoolean,
-			Stage:          featuretypes.StageExperimental,
-			Description:    "Controls whether the collector's normalize operator keeps the original log body so it is ingested into both the legacy body and the JSON body columns",
-			DefaultVariant: featuretypes.MustNewName("disabled"),
-			Variants:       featuretypes.NewBooleanVariants(),
-		},
 		&featuretypes.Feature{
 			Name:           FeatureEnableMetricsReduction,
 			Kind:           featuretypes.KindBoolean,

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

@@ -2,6 +2,7 @@ package logparsingpipeline
 
 import (
 	"encoding/json"
+	"slices"
 	"strings"
 	"sync"
 
@@ -24,10 +25,13 @@ var (
 )
 
 const (
-	memoryLimiterProcessor       = "memory_limiter"
-	memoryLimiterProcessorPrefix = "memory_limiter/"
-	batchProcessor               = "batch"
-	batchProcessorPrefix         = "batch/"
+	memoryLimiterProcessor        = "memory_limiter"
+	memoryLimiterProcessorPrefix  = "memory_limiter/"
+	batchProcessor                = "batch"
+	batchProcessorPrefix          = "batch/"
+	normalizerProcessor           = "signozlogsnormalizer"
+	normalizerProcessorPrefix     = "signozlogsnormalizer/"
+	userPipelinesBeforeNormalizer = "before"
 )
 
 // check if the processors already exist
@@ -90,9 +94,14 @@ func getOtelPipelineFromConfig(config map[string]interface{}) (*otelPipeline, er
 //  2. signoz user-pipeline processors (in the order given by signozPipelineProcessorNames)
 //  3. custom processors (non-signoz, non-memory_limiter, non-batch processors from the current config)
 //  4. batch processors (any processor named "batch" or "batch/<id>") and anything after them
+//
+// If a signozlogsnormalizer processor is present, the order above is skipped: user-pipeline
+// processors go right after the first one, or right before it when its config sets
+// `opamp.position: before`, and the other processors keep their places.
 func buildCollectorPipelineProcessorsList(
 	currentCollectorProcessors []string,
 	signozPipelineProcessorNames []string,
+	processorConfigs map[string]interface{},
 ) ([]string, error) {
 	lockLogsPipelineSpec.Lock()
 	defer lockLogsPipelineSpec.Unlock()
@@ -103,6 +112,23 @@ func buildCollectorPipelineProcessorsList(
 		userPipelineSet[p] = struct{}{}
 	}
 
+	if normalizerIdx := slices.IndexFunc(currentCollectorProcessors, isNormalizerProcessor); normalizerIdx >= 0 {
+		before := opampPosition(processorConfigs, currentCollectorProcessors[normalizerIdx]) == userPipelinesBeforeNormalizer
+		result := make([]string, 0, len(currentCollectorProcessors)+len(signozPipelineProcessorNames))
+		for idx, p := range currentCollectorProcessors {
+			if idx == normalizerIdx && before {
+				result = append(result, signozPipelineProcessorNames...)
+			}
+			if _, inUserPipelines := userPipelineSet[p]; !inUserPipelines && !hasSignozPipelineProcessorPrefix(p) {
+				result = append(result, p)
+			}
+			if idx == normalizerIdx && !before {
+				result = append(result, signozPipelineProcessorNames...)
+			}
+		}
+		return result, nil
+	}
+
 	var memoryLimiters []string
 	var customProcessors []string
 	batchIdx := -1
@@ -135,13 +161,51 @@ func buildCollectorPipelineProcessorsList(
 	return result, nil
 }
 
-func GenerateCollectorConfigWithPipelines(config []byte, pipelines []pipelinetypes.GettablePipeline) ([]byte, error) {
+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]) != userPipelinesBeforeNormalizer
+}
+
+func opampPosition(processorConfigs map[string]interface{}, name string) string {
+	config, _ := processorConfigs[name].(map[string]interface{})
+	opamp, _ := config["opamp"].(map[string]interface{})
+	position, _ := opamp["position"].(string)
+	return position
+}
+
+func unmarshalCollectorConfig(config []byte) (map[string]interface{}, error) {
 	var collectorConf map[string]interface{}
-	err := yaml.Unmarshal([]byte(config), &collectorConf)
-	if err != nil {
+	if err := yaml.Unmarshal(config, &collectorConf); err != nil {
 		return nil, errors.WrapInvalidInputf(err, CodeCollectorConfigUnmarshalFailed, "could not unmarshal collector config")
 	}
+	return collectorConf, nil
+}
 
+func GenerateCollectorConfigWithPipelines(config []byte, pipelines []pipelinetypes.GettablePipeline) ([]byte, error) {
+	collectorConf, err := unmarshalCollectorConfig(config)
+	if err != nil {
+		return nil, err
+	}
+	return generateCollectorConfigWithPipelines(collectorConf, pipelines)
+}
+
+// generateCollectorConfigWithPipelines modifies collectorConf in place.
+func generateCollectorConfigWithPipelines(collectorConf map[string]any, pipelines []pipelinetypes.GettablePipeline) ([]byte, error) {
 	signozPipelineProcessors, signozPipelineProcNames, err := PreparePipelineProcessor(pipelines)
 	if err != nil {
 		return nil, err
@@ -180,7 +244,9 @@ func GenerateCollectorConfigWithPipelines(config []byte, pipelines []pipelinetyp
 		return nil, errors.NewInternalf(CodeCollectorConfigLogsPipelineNotFound, "logs pipeline doesn't exist")
 	}
 
-	updatedProcessorList, _ := buildCollectorPipelineProcessorsList(p.Pipelines.Logs.Processors, signozPipelineProcNames)
+	updatedProcessorList, _ := buildCollectorPipelineProcessorsList(
+		p.Pipelines.Logs.Processors, signozPipelineProcNames, collectorConf["processors"].(map[string]interface{}),
+	)
 	p.Pipelines.Logs.Processors = updatedProcessorList
 
 	// add the new processor to the data ( no checks required as the keys will exists)

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

@@ -5,11 +5,14 @@ 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"
 	"gopkg.in/yaml.v3"
 )
@@ -206,13 +209,74 @@ var BuildLogsPipelineTestData = []struct {
 func TestBuildLogsPipeline(t *testing.T) {
 	for _, test := range BuildLogsPipelineTestData {
 		Convey(test.Name, t, func() {
-			v, err := buildCollectorPipelineProcessorsList(test.fromCollector, test.userPipelines)
+			v, err := buildCollectorPipelineProcessorsList(test.fromCollector, test.userPipelines, nil)
 			So(err, ShouldBeNil)
 			So(v, ShouldResemble, test.finalOutput)
 		})
 	}
 }
 
+func TestBuildLogsPipelineWithNormalizer(t *testing.T) {
+	a := constants.LogsPPLPfx + "a"
+	b := constants.LogsPPLPfx + "b"
+	before := map[string]any{"opamp": map[string]any{"position": "before"}}
+	after := map[string]any{"opamp": map[string]any{"position": "after"}}
+
+	testCases := []struct {
+		name          string
+		fromCollector []string
+		userPipelines []string
+		configs       map[string]any
+		expected      []string
+	}{
+		{
+			name:          "ExistingAndStaleUserProcessorsReplaced",
+			fromCollector: []string{"memory_limiter", a, constants.LogsPPLPfx + "stale", constants.OldLogsPPLPfx + "old", "signozlogsnormalizer", "custom", "batch"},
+			userPipelines: []string{a, b},
+			expected:      []string{"memory_limiter", "signozlogsnormalizer", a, b, "custom", "batch"},
+		},
+		{
+			name:          "NamedNormalizerBefore_CustomProcessorsKeepPlace",
+			fromCollector: []string{"memory_limiter", "custom_a", "signozlogsnormalizer/x", a, "custom_b", "batch"},
+			userPipelines: []string{a},
+			configs:       map[string]any{"signozlogsnormalizer/x": before},
+			expected:      []string{"memory_limiter", "custom_a", a, "signozlogsnormalizer/x", "custom_b", "batch"},
+		},
+		{
+			name:          "FirstNormalizerBefore_SecondIgnored",
+			fromCollector: []string{"memory_limiter", "signozlogsnormalizer/first", "signozlogsnormalizer/second", "batch"},
+			userPipelines: []string{a},
+			configs:       map[string]any{"signozlogsnormalizer/first": before, "signozlogsnormalizer/second": after},
+			expected:      []string{"memory_limiter", a, "signozlogsnormalizer/first", "signozlogsnormalizer/second", "batch"},
+		},
+		{
+			name:          "FirstNormalizerAfter_SecondIgnored",
+			fromCollector: []string{"memory_limiter", "signozlogsnormalizer/first", "signozlogsnormalizer/second", "batch"},
+			userPipelines: []string{a},
+			configs:       map[string]any{"signozlogsnormalizer/first": after, "signozlogsnormalizer/second": before},
+			expected:      []string{"memory_limiter", "signozlogsnormalizer/first", a, "signozlogsnormalizer/second", "batch"},
+		},
+		{
+			name:          "NormalizerAfterBatch",
+			fromCollector: []string{"memory_limiter", "batch", "signozlogsnormalizer"},
+			userPipelines: []string{a},
+			expected:      []string{"memory_limiter", "batch", "signozlogsnormalizer", a},
+		},
+		{
+			name:          "NoUserPipelines_StaleRemoved",
+			fromCollector: []string{"memory_limiter", a, "signozlogsnormalizer", "batch"},
+			expected:      []string{"memory_limiter", "signozlogsnormalizer", "batch"},
+		},
+	}
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			processors, err := buildCollectorPipelineProcessorsList(testCase.fromCollector, testCase.userPipelines, testCase.configs)
+			require.NoError(t, err)
+			assert.Equal(t, testCase.expected, processors)
+		})
+	}
+}
+
 func TestPipelineAliasCollisionsDontResultInDuplicateCollectorProcessors(t *testing.T) {
 	require := require.New(t)
 
@@ -437,3 +501,61 @@ 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_OperatorDropped",
+			logs:     "[memory_limiter, signozlogsnormalizer, batch]",
+			expected: []any{"memory_limiter", "signozlogsnormalizer", user, "batch"},
+		},
+		{
+			name:       "NormalizerAfter_OperatorDropped",
+			processors: "signozlogsnormalizer:\n    opamp:\n      position: after\n",
+			logs:       "[memory_limiter, signozlogsnormalizer, batch]",
+			expected:   []any{"memory_limiter", "signozlogsnormalizer", user, "batch"},
+		},
+		{
+			name:       "NormalizerBefore_OperatorKept",
+			processors: "signozlogsnormalizer:\n    opamp:\n      position: before\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

@@ -220,26 +220,17 @@ func (ic *LogParsingPipelineController) ValidatePipelines(ctx context.Context,
 	return err
 }
 
-// withNormalizePipeline places normalize where the read path dictates. Ahead of user pipelines
-// when queries run on body_v2 (use_json_body), so operators see the body the explorer shows.
-// After them when dual ingestion alone writes body_v2, so operators keep seeing the raw body
-// users still query. Absent when neither flag is on.
-func (ic *LogParsingPipelineController) withNormalizePipeline(ctx context.Context, orgID valuer.UUID, pipelines []pipelinetypes.GettablePipeline) []pipelinetypes.GettablePipeline {
-	evalCtx := featuretypes.NewFlaggerEvaluationContext(orgID)
-	dualIngestion := ic.fl.BooleanOrEmpty(ctx, flagger.FeatureJSONBodyDualIngestion, evalCtx)
-	switch {
-	case ic.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, evalCtx):
-		return append([]pipelinetypes.GettablePipeline{getNormalizePipeline(dualIngestion)}, pipelines...)
-	case dualIngestion:
-		return append(slices.Clone(pipelines), getNormalizePipeline(true))
-	default:
+// 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...)
 }
 
-// stashOriginalBody makes normalize carry the pre-normalization body in an internal attribute
-// for the exporter to restore into the legacy body column.
-func getNormalizePipeline(stashOriginalBody bool) pipelinetypes.GettablePipeline {
+func getNormalizePipeline() pipelinetypes.GettablePipeline {
 	return pipelinetypes.GettablePipeline{
 		StoreablePipeline: pipelinetypes.StoreablePipeline{
 			Name:    "Default Pipeline - PreProcessing Body",
@@ -258,11 +249,10 @@ func getNormalizePipeline(stashOriginalBody bool) pipelinetypes.GettablePipeline
 		},
 		Config: []pipelinetypes.PipelineOperator{
 			{
-				ID:                    uuid.NewString(),
-				Type:                  "normalize",
-				Enabled:               true,
-				If:                    "body != nil",
-				JSONBodyDualIngestion: stashOriginalBody,
+				ID:      uuid.NewString(),
+				Type:    "normalize",
+				Enabled: true,
+				If:      "body != nil",
 			},
 		},
 	}
@@ -371,11 +361,8 @@ func (ic *LogParsingPipelineController) PreviewLogsPipelines(
 	}
 
 	// The collector gets the same pipeline prepended over opamp; see RecommendAgentConfig.
-	// Under dual ingestion alone it runs after user operators and only feeds body_v2, which
-	// the explorer does not show yet, so the preview leaves it out. The original-body stash
-	// is left off: the preview has no exporter to restore and strip it.
 	if ic.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(orgID)) {
-		pipelines = append([]pipelinetypes.GettablePipeline{getNormalizePipeline(false)}, pipelines...)
+		pipelines = append([]pipelinetypes.GettablePipeline{getNormalizePipeline()}, pipelines...)
 	}
 
 	result, collectorLogs, err := SimulatePipelinesProcessing(ctx, pipelines, request.Logs)
@@ -396,14 +383,14 @@ 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 or json_body_dual_ingestion is on) is placed
+// 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 or
-// json_body_dual_ingestion flags or updating operator IfExpressions), agents that already
+// 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.
 func (pc *LogParsingPipelineController) RecommendAgentConfig(
@@ -431,10 +418,15 @@ func (pc *LogParsingPipelineController) RecommendAgentConfig(
 		return nil, "", err
 	}
 
-	// normalize is only for sending to the collector, never persisted
-	enrichedPipelines = pc.withNormalizePipeline(ctx, orgId, enrichedPipelines)
+	collectorConf, err := unmarshalCollectorConfig(currentConfYaml)
+	if err != nil {
+		return nil, "", err
+	}
 
-	updatedConf, err := GenerateCollectorConfigWithPipelines(currentConfYaml, enrichedPipelines)
+	// normalize is only for sending to the collector, never persisted
+	enrichedPipelines = pc.withNormalizePipeline(ctx, orgId, enrichedPipelines, collectorNormalizesFirst(collectorConf))
+
+	updatedConf, err := generateCollectorConfigWithPipelines(collectorConf, enrichedPipelines)
 	if err != nil {
 		return nil, "", err
 	}

pkg/types/pipelinetypes/pipeline.go

@@ -122,9 +122,6 @@ type PipelineOperator struct {
 	EnablePaths        bool   `json:"enable_paths,omitempty" yaml:"enable_paths,omitempty"`
 	PathPrefix         string `json:"path_prefix,omitempty" yaml:"path_prefix,omitempty"`
 
-	// normalize fields, set by the server only
-	JSONBodyDualIngestion bool `json:"-" yaml:"json_body_dual_ingestion,omitempty"`
-
 	// Used in Severity Parsing and JSON Flattening mapping
 	Mapping map[string][]string `json:"mapping,omitempty" yaml:"mapping,omitempty"`
 	// severity parser fields