[Fleet] Vastly improve performance of Fleet final pipeline's date formatting logic for event.ingested (#167318)

## Summary

Vastly improve performance of the Fleet final pipeline's date processing
for the `event.ingested` field by using a `script` processor and
`DateTimeFormatter`. All credit to @joegallo for the performance
improvements.

Closes https://github.com/elastic/kibana/issues/157430

Our test coverage for the `event.ingested` field
[here](090569bb45/x-pack/test/fleet_api_integration/apis/epm/final_pipeline.ts (L109-L127))
should suffice for ensuring we're not breaking anything here.

## Screenshots

I fired up a local cluster and ran Fleet Server + a single Agent to
verify that `event.ingested` is set correctly, and that other pipeline
fields like `agent.id` are working as expected.


![image](684f6b5d-c053-4adb-93fa-96b85efaffd1)


![image](2c948e42-1a45-43b4-8927-807fd155e077)


![image](c802090a-b6aa-41fc-ac7b-d5efcf6e20cf)

---------

Co-authored-by: Kibana Machine <42973632+kibanamachine@users.noreply.github.com>

x-pack/plugins/fleet/server/constants/fleet_es_assets.ts

@@ -81,7 +81,7 @@ export const FLEET_COMPONENT_TEMPLATES = [
   },
 ];
 
-export const FLEET_FINAL_PIPELINE_VERSION = 3;
+export const FLEET_FINAL_PIPELINE_VERSION = 4;
 
 // If the content is updated you probably need to update the FLEET_FINAL_PIPELINE_VERSION too to allow upgrade of the pipeline
 export const FLEET_FINAL_PIPELINE_CONTENT = `---
@@ -92,21 +92,28 @@ _meta:
 description: >
   Final pipeline for processing all incoming Fleet Agent documents.
 processors:
-  - date:
+  - script:
       description: Add time when event was ingested (and remove sub-seconds to improve storage efficiency)
       tag: truncate-subseconds-event-ingested
-      field: _ingest.timestamp
-      target_field: event.ingested
-      formats:
-        - ISO8601
-      output_format: date_time_no_millis
       ignore_failure: true
+      source: |-
+        if (ctx?.event == null) {
+          ctx.event = [:];
+        }
+
+        ctx.event.ingested = metadata().now.withNano(0).format(DateTimeFormatter.ISO_OFFSET_DATE_TIME);
   - remove:
       description: Remove any pre-existing untrusted values.
       field:
         - event.agent_id_status
         - _security
       ignore_missing: true
+  - remove:
+      description: Remove event.original unless the preserve_original_event tag is set
+      field: event.original
+      if: "ctx?.tags == null || !(ctx.tags.contains('preserve_original_event'))"
+      ignore_failure: true
+      ignore_missing: true
   - set_security_user:
       field: _security
       properties:

x-pack/test/fleet_api_integration/apis/epm/final_pipeline.ts

@@ -92,7 +92,7 @@ export default function (providerContext: FtrProviderContext) {
       await supertest.post(`/api/fleet/setup`).set('kbn-xsrf', 'xxxx');
       const pipelineRes = await es.ingest.getPipeline({ id: FINAL_PIPELINE_ID });
       expect(pipelineRes).to.have.property(FINAL_PIPELINE_ID);
-      expect(pipelineRes[FINAL_PIPELINE_ID].version).to.be(3);
+      expect(pipelineRes[FINAL_PIPELINE_ID].version).to.be(4);
     });
 
     it('should correctly setup the final pipeline and apply to fleet managed index template', async () => {
@@ -149,6 +149,62 @@ export default function (providerContext: FtrProviderContext) {
       expect(event).to.have.property('ingested');
     });
 
+    it('removes event.original if preserve_original_event is not set', async () => {
+      const res = await es.index({
+        index: 'logs-log.log-test',
+        body: {
+          message: 'message-test-1',
+          event: {
+            original: {
+              foo: 'bar',
+            },
+          },
+          '@timestamp': '2023-01-01T09:00:00',
+          tags: [],
+          agent: {
+            id: 'agent1',
+          },
+        },
+      });
+
+      const doc: any = await es.get({
+        id: res._id,
+        index: res._index,
+      });
+
+      const event = doc._source.event;
+
+      expect(event.original).to.be(undefined);
+    });
+
+    it('preserves event.original if preserve_original_event is set', async () => {
+      const res = await es.index({
+        index: 'logs-log.log-test',
+        body: {
+          message: 'message-test-1',
+          event: {
+            original: {
+              foo: 'bar',
+            },
+          },
+          '@timestamp': '2023-01-01T09:00:00',
+          tags: ['preserve_original_event'],
+          agent: {
+            id: 'agent1',
+          },
+        },
+      });
+
+      const doc: any = await es.get({
+        id: res._id,
+        index: res._index,
+      });
+
+      const event = doc._source.event;
+
+      expect(event.original).to.eql({ foo: 'bar' });
+    });
+
     const scenarios = [
       {
         name: 'API key without metadata',