apm-data: add `event.ingested` field to APM documents (#117970)

Restore a field originally populated by Fleet final ingest pipeline
in all APM documents.

---------

Signed-off-by: inge4pres <francesco.gualazzi@elastic.co>

x-pack/plugin/apm-data/src/main/resources/index-templates/traces-apm.sampled@template.yaml

@@ -30,7 +30,4 @@ template:
       data_stream.dataset:
         type: constant_keyword
         value: apm.sampled
-      event.ingested:
-        type: date
-        format: date_time_no_millis
-        script: "emit(System.currentTimeMillis())"
+

x-pack/plugin/apm-data/src/main/resources/ingest-pipelines/apm@pipeline.yaml

@@ -24,3 +24,13 @@ processors:
     target_field: user_agent
     ignore_failure: true
     ignore_missing: true
+- script:
+    description: Add time when event was ingested (and remove sub-seconds to improve storage efficiency)
+    tag: truncate-subseconds-event-ingested
+    ignore_failure: true
+    source: |-
+      if (ctx?.event == null) {
+        ctx.event = [:];
+      }
+
+      ctx.event.ingested = metadata().now.withNano(0).format(DateTimeFormatter.ISO_OFFSET_DATE_TIME);

x-pack/plugin/apm-data/src/yamlRestTest/resources/rest-api-spec/test/20_error_logs.yml

@@ -26,3 +26,29 @@ setup:
   - length: { hits.hits: 2 }
   - match: { hits.hits.0.fields: { "log.level": ["error"] } }
   - match: { hits.hits.1.fields: { "log.level": ["warn"] } }
+---
+"Test logs-apm.error-* setting event.ingested via ingest pipeline":
+  - requires:
+      test_runner_features: [is_after]
+  - do:
+      bulk:
+        index: logs-apm.error-event-ingested-testing
+        refresh: true
+        body:
+          # No event.ingested field populated, it gets added
+          - create: {}
+          - '{"@timestamp": "2017-06-22", "log": {"level": "error"}, "error": {"log": {"message": "loglevel"}, "exception": [{"message": "exception_used"}]}}'
+
+          # event.ingested populated, it is overwritten with current timestamp
+          - create: {}
+          - '{"@timestamp": "2017-06-22", "log": {"level": "warn"}, "error": {"log": {"message": "loglevel"}, "exception": [{"message": "exception_used"}]}, "event": {"ingested": "2017-06-22T12:34:56.789123Z"}}'
+
+  - is_false: errors
+
+  - do:
+      search:
+        index: logs-apm.error-event-ingested-testing
+        body:
+          fields: ["event.ingested"]
+  - length: { hits.hits: 2 }
+  - is_after: { hits.hits.1.fields.event\.ingested.0: "2017-06-22T12:34:56.789Z" }

x-pack/plugin/apm-data/src/yamlRestTest/resources/rest-api-spec/test/20_metrics_ingest.yml

@@ -0,0 +1,32 @@
+---
+setup:
+  - do:
+      cluster.health:
+        wait_for_events: languid
+
+---
+"Test metrics-apm.app-* setting event.ingested via ingest pipeline":
+  - requires:
+      test_runner_features: [is_after]
+  - do:
+      bulk:
+        index: metrics-apm.app.ingest
+        refresh: true
+        body:
+          # No event.ingested field populated, it gets added
+          - create: {}
+          - '{"@timestamp": "2017-06-22"}'
+
+          # event.ingested populated, it is overwritten with current timestamp
+          - create: {}
+          - '{"@timestamp": "2017-06-22", "event": {"ingested": "2017-06-22T00:00:00.789123Z"}}'
+
+  - is_false: errors
+
+  - do:
+      search:
+        index: metrics-apm.app.ingest
+        body:
+          fields: ["event.ingested"]
+  - length: { hits.hits: 2 }
+  - is_after: { hits.hits.1.fields.event\.ingested.0: "2017-06-22T00:00:00.000Z" }

x-pack/plugin/apm-data/src/yamlRestTest/resources/rest-api-spec/test/20_traces_ingest.yml

@@ -97,3 +97,30 @@ setup:
   - match: { hits.hits.1.fields: null }
   - match: { hits.hits.2.fields: {"event.success_count": [1]} }
   - match: { hits.hits.3.fields: {"event.success_count": [0]} }
+---
+"Test traces-apm-* setting event.ingested via ingest pipeline":
+  - requires:
+      test_runner_features: [is_after]
+  - do:
+      bulk:
+        index: traces-apm-testing
+        refresh: true
+        body:
+          # No event.ingested field populated, it gets added
+          - create: {}
+          - '{"@timestamp": "2017-06-22"}'
+
+          # event.ingested populated, it is preserved
+          - create: {}
+          - '{"@timestamp": "2017-06-22", "event": {"ingested": "2017-06-22T00:00:00.123456Z"}}'
+
+  - is_false: errors
+
+  - do:
+      search:
+        index: traces-apm-testing
+        body:
+          fields: ["event.ingested"]
+  - length: { hits.hits: 2 }
+  - is_after: { hits.hits.1.fields.event\.ingested.0: "2017-06-22T00:00:00.000Z" }
+