Prevent indexing of event.received field

apmpackage/apm/changelog.yml

@@ -12,6 +12,9 @@
     - description: Remove unused `processor.event` field from logs data streams
       type: enhancement
       link: https://github.com/elastic/apm-server/pull/11494
+    - description: Remove `event.received` fields from logs, metrics and traces data streams
+      type: enhancement
+      link: https://github.com/elastic/apm-server/pull/11602
 - version: "8.10.0"
   changes:
     - description: Add permissions to reroute to dedicated datasets for logs, metrics and traces

apmpackage/apm/data_stream/app_logs/elasticsearch/ingest_pipeline/default.yml

@@ -9,3 +9,5 @@ processors:
       name: observer_ids
   - pipeline:
       name: ecs_version
+  - pipeline:
+      name: remove_event_received

apmpackage/apm/data_stream/app_metrics/elasticsearch/ingest_pipeline/default.yml

@@ -15,3 +15,5 @@ processors:
       name: client_geoip
   - pipeline:
       name: set_metrics
+  - pipeline:
+      name: remove_event_received

apmpackage/apm/data_stream/error_logs/elasticsearch/ingest_pipeline/default.yml

@@ -20,3 +20,5 @@ processors:
       if: ctx.error?.log?.message != null
       field: error.grouping_name
       copy_from: error.log.message
+  - pipeline:
+      name: remove_event_received

apmpackage/apm/data_stream/internal_metrics/elasticsearch/ingest_pipeline/default.yml

@@ -22,3 +22,5 @@ processors:
   - remove:
       field: _dynamic_templates
       ignore_missing: true
+  - pipeline:
+      name: remove_event_received

apmpackage/apm/data_stream/rum_traces/elasticsearch/ingest_pipeline/default.yml

@@ -38,4 +38,5 @@ processors:
         - service.framework
       ignore_missing: true
       ignore_failure: true
-
+  - pipeline:
+      name: remove_event_received

apmpackage/apm/data_stream/traces/elasticsearch/ingest_pipeline/default.yml

@@ -35,3 +35,5 @@ processors:
       value: 0
       ignore_failure: true
   # end of event.success_count logic
+  - pipeline:
+      name: remove_event_received

apmpackage/cmd/genpackage/pipelines.go

@@ -36,14 +36,15 @@ import (
 //	  - ...
 func getCommonPipeline(name string, version *version.V) []map[string]interface{} {
 	commonPipelines := map[string][]map[string]interface{}{
-		"observer_version": getObserverVersionPipeline(version),
-		"observer_ids":     observerIDsPipeline,
-		"ecs_version":      ecsVersionPipeline,
-		"user_agent":       userAgentPipeline,
-		"process_ppid":     processPpidPipeline,
-		"client_geoip":     clientGeoIPPipeline,
-		"event_duration":   eventDurationPipeline,
-		"set_metrics":      setMetricsPipeline,
+		"observer_version":      getObserverVersionPipeline(version),
+		"observer_ids":          observerIDsPipeline,
+		"ecs_version":           ecsVersionPipeline,
+		"user_agent":            userAgentPipeline,
+		"process_ppid":          processPpidPipeline,
+		"client_geoip":          clientGeoIPPipeline,
+		"event_duration":        eventDurationPipeline,
+		"set_metrics":           setMetricsPipeline,
+		"remove_event_received": removeEventReceivedPipeline,
 	}
 	return commonPipelines[name]
 }
@@ -144,6 +145,15 @@ var clientGeoIPPipeline = []map[string]interface{}{{
 	},
 }}
 
+var removeEventReceivedPipeline = []map[string]interface{}{{
+	"remove": map[string]interface{}{
+		"field":          "event.received",
+		"ignore_missing": true,
+		"ignore_failure": true,
+		"description":    "The field is used for internal statistics but does not need to be indexed.",
+	},
+}}
+
 // This pipeline translates `event.duration` (defaulting to zero if not
 // found) to `transaction.duration.us` or `span.duration.us` depending on
 // the event type, and then removes `event.duration`. Older versions of

internal/beater/beater.go

@@ -509,10 +509,6 @@ func (s *Runner) Run(ctx context.Context) error {
 		modelpb.ProcessBatchFunc(rateLimitBatchProcessor),
 		modelpb.ProcessBatchFunc(authorizeEventIngestProcessor),
 
-		// Add a model processor that removes `event.received`, which is added by
-		// apm-data, but which we don't yet map.
-		modelpb.ProcessBatchFunc(removeEventReceivedBatchProcessor),
-
 		// Pre-process events before they are sent to the final processors for
 		// aggregation, sampling, and indexing.
 		modelprocessor.SetHostHostname{},

internal/beater/processors.go

@@ -85,17 +85,6 @@ func newObserverBatchProcessor() modelpb.ProcessBatchFunc {
 	}
 }
 
-// TODO remove this once we have added `event.received` to the
-// data stream mappings, in 8.10.
-func removeEventReceivedBatchProcessor(ctx context.Context, batch *modelpb.Batch) error {
-	for _, event := range *batch {
-		if event.Event != nil {
-			event.Event.Received = 0
-		}
-	}
-	return nil
-}
-
 func newDocappenderBatchProcessor(a *docappender.Appender) modelpb.ProcessBatchFunc {
 	var pool sync.Pool
 	pool.New = func() any {