skip log & metric enrichment (#908)

Author: rubvs

processor/elasticapmprocessor/README.md

@@ -22,3 +22,13 @@ The processor enriches traces, metrics, and logs with elastic specific requireme
 processors:
   elasticapm:
 ```
+
+### Conditionally enrich logs and metrics only for ECS mapping mode
+
+When `skip_enrichment` is set to `true`, logs and metrics are only enriched when the `x-elastic-mapping-mode` metadata is set to `ecs`. Traces are always enriched regardless of this setting. This defaults to `false` for backwards compatibility (always enrich).
+
+```yaml
+processors:
+  elasticapm:
+    skip_enrichment: true
+```

processor/elasticapmprocessor/config.go

@@ -21,4 +21,10 @@ import "github.com/elastic/opentelemetry-lib/enrichments/config"
 
 type Config struct {
 	config.Config `mapstructure:",squash"`
+
+	// SkipEnrichment controls whether enrichment should be skipped for logs and metrics
+	// when the mapping mode is not "ecs". When true, logs and metrics are only enriched when
+	// the x-elastic-mapping-mode metadata is set to "ecs". Traces are always enriched regardless
+	// of this setting. Defaults to false for backwards compatibility (always enrich).
+	SkipEnrichment bool `mapstructure:"skip_enrichment"`
 }

processor/elasticapmprocessor/processor.go

@@ -92,16 +92,18 @@ type LogProcessor struct {
 	component.StartFunc
 	component.ShutdownFunc
 
-	next     consumer.Logs
-	enricher *enrichments.Enricher
-	logger   *zap.Logger
+	next           consumer.Logs
+	enricher       *enrichments.Enricher
+	logger         *zap.Logger
+	skipEnrichment bool
 }
 
 func newLogProcessor(cfg *Config, next consumer.Logs, logger *zap.Logger) *LogProcessor {
 	return &LogProcessor{
-		next:     next,
-		logger:   logger,
-		enricher: enrichments.NewEnricher(cfg.Config),
+		next:           next,
+		logger:         logger,
+		enricher:       enrichments.NewEnricher(cfg.Config),
+		skipEnrichment: cfg.SkipEnrichment,
 	}
 }
 
@@ -113,16 +115,18 @@ type MetricProcessor struct {
 	component.StartFunc
 	component.ShutdownFunc
 
-	next     consumer.Metrics
-	enricher *enrichments.Enricher
-	logger   *zap.Logger
+	next           consumer.Metrics
+	enricher       *enrichments.Enricher
+	logger         *zap.Logger
+	skipEnrichment bool
 }
 
 func newMetricProcessor(cfg *Config, next consumer.Metrics, logger *zap.Logger) *MetricProcessor {
 	return &MetricProcessor{
-		next:     next,
-		logger:   logger,
-		enricher: enrichments.NewEnricher(cfg.Config),
+		next:           next,
+		logger:         logger,
+		enricher:       enrichments.NewEnricher(cfg.Config),
+		skipEnrichment: cfg.SkipEnrichment,
 	}
 }
 
@@ -131,7 +135,8 @@ func (p *MetricProcessor) Capabilities() consumer.Capabilities {
 }
 
 func (p *MetricProcessor) ConsumeMetrics(ctx context.Context, md pmetric.Metrics) error {
-	if isECS(ctx) {
+	ecsMode := isECS(ctx)
+	if ecsMode {
 		resourceMetrics := md.ResourceMetrics()
 		for i := 0; i < resourceMetrics.Len(); i++ {
 			resourceMetric := resourceMetrics.At(i)
@@ -141,12 +146,17 @@ func (p *MetricProcessor) ConsumeMetrics(ctx context.Context, md pmetric.Metrics
 			p.enricher.Config.Resource.DeploymentEnvironment.Enabled = false
 		}
 	}
-	p.enricher.EnrichMetrics(md)
+	// When skipEnrichment is true, only enrich when mapping mode is ecs
+	// When skipEnrichment is false (default), always enrich (backwards compatible)
+	if !p.skipEnrichment || ecsMode {
+		p.enricher.EnrichMetrics(md)
+	}
 	return p.next.ConsumeMetrics(ctx, md)
 }
 
 func (p *LogProcessor) ConsumeLogs(ctx context.Context, ld plog.Logs) error {
-	if isECS(ctx) {
+	ecsMode := isECS(ctx)
+	if ecsMode {
 		resourceLogs := ld.ResourceLogs()
 		for i := 0; i < resourceLogs.Len(); i++ {
 			resourceLog := resourceLogs.At(i)
@@ -157,6 +167,10 @@ func (p *LogProcessor) ConsumeLogs(ctx context.Context, ld plog.Logs) error {
 			p.enricher.Config.Resource.DeploymentEnvironment.Enabled = false
 		}
 	}
-	p.enricher.EnrichLogs(ld)
+	// When skipEnrichment is true, only enrich when mapping mode is ecs
+	// When skipEnrichment is false (default), always enrich (backwards compatible)
+	if !p.skipEnrichment || ecsMode {
+		p.enricher.EnrichLogs(ld)
+	}
 	return p.next.ConsumeLogs(ctx, ld)
 }

processor/elasticapmprocessor/processor_test.go

@@ -24,6 +24,8 @@ import (
 	"testing"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest/plogtest"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest/pmetrictest"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest/ptracetest"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -118,3 +120,137 @@ func TestProcessorECS(t *testing.T) {
 	}
 	assert.NoError(t, ptracetest.CompareTraces(expectedTraces, actual))
 }
+
+// TestSkipEnrichmentLogs tests that logs are only enriched when skipEnrichment is false or when mapping mode is ecs
+func TestSkipEnrichmentLogs(t *testing.T) {
+	testCases := []struct {
+		name           string
+		skipEnrichment bool
+		mappingMode    string
+	}{
+		{
+			name:           "logs_false",
+			skipEnrichment: false,
+			mappingMode:    "",
+		},
+		{
+			name:           "logs_false_ecs",
+			skipEnrichment: false,
+			mappingMode:    "ecs",
+		},
+		{
+			name:           "logs_true_ecs",
+			skipEnrichment: true,
+			mappingMode:    "ecs",
+		},
+		{
+			name:           "logs_true_no_ecs",
+			skipEnrichment: true,
+			mappingMode:    "",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := context.Background()
+			if tc.mappingMode != "" {
+				ctx = client.NewContext(ctx, client.Info{
+					Metadata: client.NewMetadata(map[string][]string{"x-elastic-mapping-mode": {tc.mappingMode}}),
+				})
+			}
+
+			factory := NewFactory()
+			settings := processortest.NewNopSettings(metadata.Type)
+			settings.TelemetrySettings.Logger = zaptest.NewLogger(t, zaptest.Level(zapcore.DebugLevel))
+			next := &consumertest.LogsSink{}
+
+			cfg := createDefaultConfig().(*Config)
+			cfg.SkipEnrichment = tc.skipEnrichment
+
+			lp, err := factory.CreateLogs(ctx, settings, cfg, next)
+			require.NoError(t, err)
+
+			dir := filepath.Join("testdata", "skip_enrichment")
+			inputLogs, err := golden.ReadLogs(filepath.Join(dir, tc.name+"_input.yaml"))
+			require.NoError(t, err)
+
+			outputFile := filepath.Join(dir, tc.name+"_output.yaml")
+			require.NoError(t, lp.ConsumeLogs(ctx, inputLogs))
+			actual := next.AllLogs()[0]
+			if *update {
+				err := golden.WriteLogs(t, outputFile, actual)
+				assert.NoError(t, err)
+			}
+			expectedLogs, err := golden.ReadLogs(outputFile)
+			require.NoError(t, err)
+			assert.NoError(t, plogtest.CompareLogs(expectedLogs, actual))
+		})
+	}
+}
+
+// TestSkipEnrichmentMetrics tests that metrics are only enriched when skipEnrichment is false or when mapping mode is ecs
+func TestSkipEnrichmentMetrics(t *testing.T) {
+	testCases := []struct {
+		name           string
+		skipEnrichment bool
+		mappingMode    string
+	}{
+		{
+			name:           "metrics_false",
+			skipEnrichment: false,
+			mappingMode:    "",
+		},
+		{
+			name:           "metrics_false_ecs",
+			skipEnrichment: false,
+			mappingMode:    "ecs",
+		},
+		{
+			name:           "metrics_true_ecs",
+			skipEnrichment: true,
+			mappingMode:    "ecs",
+		},
+		{
+			name:           "metrics_true_no_ecs",
+			skipEnrichment: true,
+			mappingMode:    "",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := context.Background()
+			if tc.mappingMode != "" {
+				ctx = client.NewContext(ctx, client.Info{
+					Metadata: client.NewMetadata(map[string][]string{"x-elastic-mapping-mode": {tc.mappingMode}}),
+				})
+			}
+
+			factory := NewFactory()
+			settings := processortest.NewNopSettings(metadata.Type)
+			settings.TelemetrySettings.Logger = zaptest.NewLogger(t, zaptest.Level(zapcore.DebugLevel))
+			next := &consumertest.MetricsSink{}
+
+			cfg := createDefaultConfig().(*Config)
+			cfg.SkipEnrichment = tc.skipEnrichment
+
+			mp, err := factory.CreateMetrics(ctx, settings, cfg, next)
+			require.NoError(t, err)
+
+			dir := filepath.Join("testdata", "skip_enrichment")
+			inputMetrics, err := golden.ReadMetrics(filepath.Join(dir, tc.name+"_input.yaml"))
+			require.NoError(t, err)
+
+			outputFile := filepath.Join(dir, tc.name+"_output.yaml")
+			require.NoError(t, mp.ConsumeMetrics(ctx, inputMetrics))
+			actual := next.AllMetrics()[0]
+			if *update {
+				err := golden.WriteMetrics(t, outputFile, actual)
+				assert.NoError(t, err)
+			}
+			expectedMetrics, err := golden.ReadMetrics(outputFile)
+			require.NoError(t, err)
+			assert.NoError(t, pmetrictest.CompareMetrics(expectedMetrics, actual, pmetrictest.IgnoreMetricsOrder(), pmetrictest.IgnoreResourceMetricsOrder()))
+		})
+	}
+}

processor/elasticapmprocessor/testdata/skip_enrichment/logs_false_ecs_input.yaml

@@ -0,0 +1,12 @@
+resourceLogs:
+  - resource:
+      attributes:
+        - key: service.name
+          value:
+            stringValue: test-service
+    scopeLogs:
+      - logRecords:
+          - body:
+              stringValue: test log message
+            timeUnixNano: "0"
+        scope: {}

processor/elasticapmprocessor/testdata/skip_enrichment/logs_false_ecs_output.yaml

@@ -0,0 +1,23 @@
+resourceLogs:
+  - resource:
+      attributes:
+        - key: service.name
+          value:
+            stringValue: test-service
+        - key: data_stream.type
+          value:
+            stringValue: logs
+        - key: data_stream.dataset
+          value:
+            stringValue: apm
+        - key: data_stream.namespace
+          value:
+            stringValue: default
+        - key: agent.name
+          value:
+            stringValue: otlp
+    scopeLogs:
+      - logRecords:
+          - body:
+              stringValue: test log message
+        scope: {}

processor/elasticapmprocessor/testdata/skip_enrichment/logs_false_input.yaml

@@ -0,0 +1,12 @@
+resourceLogs:
+  - resource:
+      attributes:
+        - key: service.name
+          value:
+            stringValue: test-service
+    scopeLogs:
+      - logRecords:
+          - body:
+              stringValue: test log message
+            timeUnixNano: "0"
+        scope: {}

processor/elasticapmprocessor/testdata/skip_enrichment/logs_false_output.yaml

@@ -0,0 +1,17 @@
+resourceLogs:
+  - resource:
+      attributes:
+        - key: service.name
+          value:
+            stringValue: test-service
+        - key: agent.name
+          value:
+            stringValue: otlp
+        - key: agent.version
+          value:
+            stringValue: unknown
+    scopeLogs:
+      - logRecords:
+          - body:
+              stringValue: test log message
+        scope: {}

processor/elasticapmprocessor/testdata/skip_enrichment/logs_true_ecs_input.yaml

@@ -0,0 +1,12 @@
+resourceLogs:
+  - resource:
+      attributes:
+        - key: service.name
+          value:
+            stringValue: test-service
+    scopeLogs:
+      - logRecords:
+          - body:
+              stringValue: test log message
+            timeUnixNano: "0"
+        scope: {}

processor/elasticapmprocessor/testdata/skip_enrichment/logs_true_ecs_output.yaml

@@ -0,0 +1,23 @@
+resourceLogs:
+  - resource:
+      attributes:
+        - key: service.name
+          value:
+            stringValue: test-service
+        - key: data_stream.type
+          value:
+            stringValue: logs
+        - key: data_stream.dataset
+          value:
+            stringValue: apm
+        - key: data_stream.namespace
+          value:
+            stringValue: default
+        - key: agent.name
+          value:
+            stringValue: otlp
+    scopeLogs:
+      - logRecords:
+          - body:
+              stringValue: test log message
+        scope: {}