Add config to enable the host.ip enrichment

Author: isaacaflores2

processor/elasticapmprocessor/config.go

@@ -27,4 +27,9 @@ type Config struct {
 	// 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"`
+
+	// HostIPEnabled controls whether the `host.ip` resource attribute should be set using client info address.
+	// When true, the processor will set the `host.ip` attribute from the client address when
+	// the mapping mode is "ecs". Defaults to true.
+	HostIPEnabled bool `mapstructure:"host_ip_enabled"`
 }

processor/elasticapmprocessor/factory.go

@@ -43,7 +43,9 @@ func NewFactory() processor.Factory {
 
 func createDefaultConfig() component.Config {
 	return &Config{
-		Config: config.Enabled(),
+		Config:         config.Enabled(),
+		SkipEnrichment: false,
+		HostIPEnabled:  true,
 	}
 }
 

processor/elasticapmprocessor/processor.go

@@ -45,13 +45,15 @@ type TraceProcessor struct {
 	next     consumer.Traces
 	enricher *enrichments.Enricher
 	logger   *zap.Logger
+	cfg      *Config
 }
 
 func NewTraceProcessor(cfg *Config, next consumer.Traces, logger *zap.Logger) *TraceProcessor {
 	return &TraceProcessor{
 		next:     next,
 		logger:   logger,
 		enricher: enrichments.NewEnricher(cfg.Config),
+		cfg:      cfg,
 	}
 }
 
@@ -66,7 +68,9 @@ func (p *TraceProcessor) ConsumeTraces(ctx context.Context, td ptrace.Traces) er
 			resourceSpan := resourceSpans.At(i)
 			resource := resourceSpan.Resource()
 			ecs.TranslateResourceMetadata(resource)
-			ecs.SetHostIP(ctx, resource.Attributes())
+			if p.cfg.HostIPEnabled {
+				ecs.SetHostIP(ctx, resource.Attributes())
+			}
 			routing.EncodeDataStream(resource, "traces")
 			p.enricher.Config.Resource.DeploymentEnvironment.Enabled = false
 		}
@@ -94,18 +98,18 @@ type LogProcessor struct {
 	component.StartFunc
 	component.ShutdownFunc
 
-	next           consumer.Logs
-	enricher       *enrichments.Enricher
-	logger         *zap.Logger
-	skipEnrichment bool
+	next     consumer.Logs
+	enricher *enrichments.Enricher
+	logger   *zap.Logger
+	cfg      *Config
 }
 
 func newLogProcessor(cfg *Config, next consumer.Logs, logger *zap.Logger) *LogProcessor {
 	return &LogProcessor{
-		next:           next,
-		logger:         logger,
-		enricher:       enrichments.NewEnricher(cfg.Config),
-		skipEnrichment: cfg.SkipEnrichment,
+		next:     next,
+		logger:   logger,
+		enricher: enrichments.NewEnricher(cfg.Config),
+		cfg:      cfg,
 	}
 }
 
@@ -117,18 +121,18 @@ type MetricProcessor struct {
 	component.StartFunc
 	component.ShutdownFunc
 
-	next           consumer.Metrics
-	enricher       *enrichments.Enricher
-	logger         *zap.Logger
-	skipEnrichment bool
+	next     consumer.Metrics
+	enricher *enrichments.Enricher
+	logger   *zap.Logger
+	cfg      *Config
 }
 
 func newMetricProcessor(cfg *Config, next consumer.Metrics, logger *zap.Logger) *MetricProcessor {
 	return &MetricProcessor{
-		next:           next,
-		logger:         logger,
-		enricher:       enrichments.NewEnricher(cfg.Config),
-		skipEnrichment: cfg.SkipEnrichment,
+		next:     next,
+		logger:   logger,
+		enricher: enrichments.NewEnricher(cfg.Config),
+		cfg:      cfg,
 	}
 }
 
@@ -144,14 +148,16 @@ func (p *MetricProcessor) ConsumeMetrics(ctx context.Context, md pmetric.Metrics
 			resourceMetric := resourceMetrics.At(i)
 			resource := resourceMetric.Resource()
 			ecs.TranslateResourceMetadata(resource)
-			ecs.SetHostIP(ctx, resource.Attributes())
+			if p.cfg.HostIPEnabled {
+				ecs.SetHostIP(ctx, resource.Attributes())
+			}
 			routing.EncodeDataStream(resource, "metrics")
 			p.enricher.Config.Resource.DeploymentEnvironment.Enabled = false
 		}
 	}
 	// When skipEnrichment is true, only enrich when mapping mode is ecs
 	// When skipEnrichment is false (default), always enrich (backwards compatible)
-	if !p.skipEnrichment || ecsMode {
+	if !p.cfg.SkipEnrichment || ecsMode {
 		p.enricher.EnrichMetrics(md)
 	}
 	return p.next.ConsumeMetrics(ctx, md)
@@ -165,15 +171,17 @@ func (p *LogProcessor) ConsumeLogs(ctx context.Context, ld plog.Logs) error {
 			resourceLog := resourceLogs.At(i)
 			resource := resourceLog.Resource()
 			ecs.TranslateResourceMetadata(resource)
-			ecs.SetHostIP(ctx, resource.Attributes())
+			if p.cfg.HostIPEnabled {
+				ecs.SetHostIP(ctx, resource.Attributes())
+			}
 			routing.EncodeDataStream(resource, "logs")
 			p.enricher.Config.Resource.AgentVersion.Enabled = false
 			p.enricher.Config.Resource.DeploymentEnvironment.Enabled = false
 		}
 	}
 	// When skipEnrichment is true, only enrich when mapping mode is ecs
 	// When skipEnrichment is false (default), always enrich (backwards compatible)
-	if !p.skipEnrichment || ecsMode {
+	if !p.cfg.SkipEnrichment || ecsMode {
 		p.enricher.EnrichLogs(ld)
 	}
 	return p.next.ConsumeLogs(ctx, ld)