Add event.ingested to all Filebeat modules (#20386) (#20483)

The event.ingested field defines time at which the event was ingested to Elasticsearch
and it added by the Ingest Node pipeline. This field is important when trying to build
alerts for activities that may have been reported long after they occurred (@timestamp is
much older than event.ingested). This might happen if an agent was offline for a period
of time or the processing was delayed.

This adds a test to ensure all modules create event.ingested.

Use Filebeat read time instead of ingest time as event.created in Zeek.

Closes #20073

(cherry picked from commit 829c3b7)

CHANGELOG.next.asciidoc

@@ -592,6 +592,7 @@ field. You can revert this change by configuring tags for the module and omittin
 - Add experimental dataset zscaler/zia for Zscaler Internet Access logs {pull}19713[19713]
 - Add event.ingested for CrowdStrike module {pull}20138[20138]
 - Add support for additional fields and FirewallMatchEvent type events in CrowdStrike module {pull}20138[20138]
+- Add event.ingested to all Filebeat modules. {pull}20386[20386]
 - Add event.ingested for Suricata module {pull}20220[20220]
 
 *Heartbeat*

filebeat/module/apache/access/ingest/pipeline.yml

@@ -1,6 +1,9 @@
 description: "Pipeline for parsing Apache HTTP Server access logs. Requires the geoip and user_agent plugins."
 
 processors:
+- set:
+    field: event.ingested
+    value: '{{_ingest.timestamp}}'
 - grok:
     field: message
     patterns:

filebeat/module/apache/error/ingest/pipeline.yml

@@ -1,5 +1,8 @@
 description: Pipeline for parsing apache error logs
 processors:
+- set:
+    field: event.ingested
+    value: '{{_ingest.timestamp}}'
 - grok:
     field: message
     patterns:

filebeat/module/auditd/log/ingest/pipeline.yml

@@ -1,6 +1,9 @@
 ---
 description: Pipeline for parsing Linux auditd logs
 processors:
+- set:
+    field: event.ingested
+    value: '{{_ingest.timestamp}}'
 - grok:
     field: message
     pattern_definitions:

filebeat/module/elasticsearch/audit/ingest/pipeline.yml

@@ -1,5 +1,8 @@
 description: Pipeline for parsing elasticsearch audit logs
 processors:
+- set:
+    field: event.ingested
+    value: '{{_ingest.timestamp}}'
 - rename:
     field: '@timestamp'
     target_field: event.created

filebeat/module/elasticsearch/deprecation/ingest/pipeline.yml

@@ -1,5 +1,8 @@
 description: Pipeline for parsing elasticsearch deprecation logs
 processors:
+- set:
+    field: event.ingested
+    value: '{{_ingest.timestamp}}'
 - rename:
     field: '@timestamp'
     target_field: event.created

filebeat/module/elasticsearch/gc/ingest/pipeline.yml

@@ -1,5 +1,8 @@
 description: Pipeline for parsing Elasticsearch JVM garbage collection logs
 processors:
+- set:
+    field: event.ingested
+    value: '{{_ingest.timestamp}}'
 - grok:
     field: message
     patterns:

filebeat/module/elasticsearch/server/ingest/pipeline.yml

@@ -1,5 +1,8 @@
 description: Pipeline for parsing elasticsearch server logs
 processors:
+- set:
+    field: event.ingested
+    value: '{{_ingest.timestamp}}'
 - rename:
     field: '@timestamp'
     target_field: event.created

filebeat/module/elasticsearch/slowlog/ingest/pipeline.yml

@@ -1,5 +1,8 @@
 description: Pipeline for parsing elasticsearch slow logs.
 processors:
+- set:
+    field: event.ingested
+    value: '{{_ingest.timestamp}}'
 - rename:
     field: '@timestamp'
     target_field: event.created

filebeat/module/haproxy/log/ingest/pipeline.yml

@@ -1,6 +1,9 @@
 description: Pipeline for parsing HAProxy http, tcp and default logs. Requires the
   geoip plugin.
 processors:
+- set:
+    field: event.ingested
+    value: '{{_ingest.timestamp}}'
 - grok:
     field: message
     patterns:

filebeat/module/icinga/debug/ingest/pipeline.yml

@@ -1,5 +1,8 @@
 description: Pipeline for parsing icinga debug logs
 processors:
+- set:
+    field: event.ingested
+    value: '{{_ingest.timestamp}}'
 - grok:
     field: message
     patterns:

filebeat/module/icinga/main/ingest/pipeline.yml

@@ -1,5 +1,8 @@
 description: Pipeline for parsing icinga main logs
 processors:
+- set:
+    field: event.ingested
+    value: '{{_ingest.timestamp}}'
 - grok:
     field: message
     patterns:

filebeat/module/icinga/startup/ingest/pipeline.yml

@@ -1,5 +1,8 @@
 description: Pipeline for parsing icinga startup logs
 processors:
+- set:
+    field: event.ingested
+    value: '{{_ingest.timestamp}}'
 - grok:
     field: message
     patterns:

filebeat/module/iis/access/ingest/pipeline.yml

@@ -1,6 +1,9 @@
 description: Pipeline for parsing IIS access logs. Requires the geoip and user_agent
   plugins.
 processors:
+- set:
+    field: event.ingested
+    value: '{{_ingest.timestamp}}'
 - grok:
     field: message
     patterns:

filebeat/module/iis/error/ingest/pipeline.yml

@@ -1,5 +1,8 @@
 description: Pipeline for parsing IIS error logs. Requires the geoip plugin.
 processors:
+- set:
+    field: event.ingested
+    value: '{{_ingest.timestamp}}'
 - grok:
     field: message
     patterns:

filebeat/module/kafka/log/ingest/pipeline.yml

@@ -1,5 +1,8 @@
 description: Pipeline for parsing Kafka log messages
 processors:
+- set:
+    field: event.ingested
+    value: '{{_ingest.timestamp}}'
 - grok:
     field: message
     trace_match: true

filebeat/module/kibana/log/ingest/pipeline.yml

@@ -4,6 +4,9 @@ on_failure:
     field: error.message
     value: '{{ _ingest.on_failure_message }}'
 processors:
+- set:
+    field: event.ingested
+    value: '{{_ingest.timestamp}}'
 - rename:
     field: '@timestamp'
     target_field: event.created

filebeat/module/logstash/log/ingest/pipeline.yml

@@ -1,5 +1,8 @@
 description: Pipeline for parsing logstash node logs
 processors:
+- set:
+    field: event.ingested
+    value: '{{_ingest.timestamp}}'
 - rename:
     field: '@timestamp'
     target_field: event.created

filebeat/module/logstash/slowlog/ingest/pipeline.yml

@@ -1,5 +1,8 @@
 description: Pipeline for parsing logstash slow logs
 processors:
+- set:
+    field: event.ingested
+    value: '{{_ingest.timestamp}}'
 - rename:
     field: '@timestamp'
     target_field: event.created

filebeat/module/mongodb/log/ingest/pipeline.yml

@@ -1,5 +1,8 @@
 description: Pipeline for parsing MongoDB logs
 processors:
+- set:
+    field: event.ingested
+    value: '{{_ingest.timestamp}}'
 - grok:
     field: message
     patterns:

filebeat/module/mysql/error/ingest/pipeline.yml

@@ -1,5 +1,8 @@
 description: Pipeline for parsing MySQL error logs
 processors:
+- set:
+    field: event.ingested
+    value: '{{_ingest.timestamp}}'
 - grok:
     field: message
     patterns:

filebeat/module/mysql/slowlog/ingest/pipeline.json

@@ -1,6 +1,11 @@
 {
   "description": "Pipeline for parsing MySQL slow logs.",
   "processors": [{
+    "set": {
+      "field": "event.ingested",
+      "value": "{{_ingest.timestamp}}"
+    }
+  }, {
     "grok": {
       "field": "message",
       "patterns":[

filebeat/module/nats/log/ingest/pipeline.yml

@@ -1,5 +1,8 @@
 description: Pipeline for parsing nats log logs
 processors:
+- set:
+    field: event.ingested
+    value: '{{_ingest.timestamp}}'
 - grok:
     field: message
     patterns:

filebeat/module/nginx/access/ingest/pipeline.yml

@@ -1,6 +1,9 @@
 description: Pipeline for parsing Nginx access logs. Requires the geoip and user_agent
   plugins.
 processors:
+- set:
+    field: event.ingested
+    value: '{{_ingest.timestamp}}'
 - grok:
     field: message
     patterns:
@@ -145,7 +148,7 @@ processors:
 - set:
     field: event.outcome
     value: failure
-    if: "ctx?.http?.response?.status_code != null && ctx.http.response.status_code >= 400"      
+    if: "ctx?.http?.response?.status_code != null && ctx.http.response.status_code >= 400"
 - append:
     field: related.ip
     value: "{{source.ip}}"

filebeat/module/nginx/error/ingest/pipeline.yml

@@ -1,5 +1,8 @@
 description: Pipeline for parsing the Nginx error logs
 processors:
+- set:
+    field: event.ingested
+    value: '{{_ingest.timestamp}}'
 - grok:
     field: message
     patterns:

filebeat/module/nginx/ingress_controller/ingest/pipeline.yml

@@ -1,6 +1,9 @@
 description: Pipeline for parsing Nginx ingress controller access logs. Requires the
   geoip and user_agent plugins.
 processors:
+- set:
+    field: event.ingested
+    value: '{{_ingest.timestamp}}'
 - grok:
     field: message
     patterns:

filebeat/module/osquery/result/ingest/pipeline.json

@@ -2,6 +2,11 @@
   "description": "Pipeline for parsing osquery result logs",
   "processors": [
     {
+      "set":{
+        "field": "event.ingested",
+        "value": "{{_ingest.timestamp}}"
+      }
+    }, {
       "rename": {
         "field": "@timestamp",
         "target_field": "event.created"

filebeat/module/postgresql/log/ingest/pipeline.yml

@@ -1,5 +1,8 @@
 description: Pipeline for parsing PostgreSQL logs.
 processors:
+- set:
+    field: event.ingested
+    value: '{{_ingest.timestamp}}'
 - grok:
     field: message
     ignore_missing: true

filebeat/module/redis/log/ingest/pipeline.yml

@@ -1,5 +1,8 @@
 description: Pipeline for parsing redis logs
 processors:
+- set:
+    field: event.ingested
+    value: '{{_ingest.timestamp}}'
 - grok:
     field: message
     patterns:

filebeat/module/santa/log/ingest/pipeline.yml

@@ -1,5 +1,8 @@
 description: Pipeline for parsing Google Santa logs.
 processors:
+- set:
+    field: event.ingested
+    value: '{{_ingest.timestamp}}'
 - grok:
     field: message
     patterns:

filebeat/module/system/auth/ingest/pipeline.yml

@@ -1,5 +1,8 @@
 description: Pipeline for parsing system authorisation/secure logs
 processors:
+- set:
+    field: event.ingested
+    value: '{{_ingest.timestamp}}'
 - grok:
     field: message
     ignore_missing: true

filebeat/module/system/syslog/ingest/pipeline.yml

@@ -1,5 +1,8 @@
 description: Pipeline for parsing Syslog messages.
 processors:
+- set:
+    field: event.ingested
+    value: '{{_ingest.timestamp}}'
 - grok:
     field: message
     patterns:

filebeat/module/traefik/access/ingest/pipeline.yml

@@ -1,6 +1,9 @@
 description: Pipeline for parsing Traefik access logs. Requires the geoip and user_agent
   plugins.
 processors:
+- set:
+    field: event.ingested
+    value: '{{_ingest.timestamp}}'
 - dissect:
     field: message
     pattern: '%{source.address} %{traefik.access.user_identifier} %{user.name} [%{traefik.access.time}]

filebeat/tests/system/test_modules.py

@@ -169,6 +169,10 @@ def run_on_file(self, module, fileset, test_file, cfgfile):
             assert obj["event"]["module"] == module, "expected event.module={} but got {}".format(
                 module, obj["event"]["module"])
 
+            # All modules must include a set processor that adds the time that
+            # the event was ingested to Elasticsearch
+            assert "ingested" in obj["event"], "missing event.ingested timestamp"
+
             assert "error" not in obj, "not error expected but got: {}.\n The related error message is: {}".format(
                 obj, obj["error"].get("message"))
 

x-pack/filebeat/module/activemq/audit/ingest/pipeline.yml

@@ -1,6 +1,9 @@
 ---
 description: Pipeline for parsing ActiveMQ audit logs.
 processors:
+  - set:
+      field: event.ingested
+      value: '{{_ingest.timestamp}}'
   - grok:
       field: message
       pattern_definitions:

x-pack/filebeat/module/activemq/log/ingest/pipeline.yml

@@ -1,6 +1,9 @@
 ---
 description: Pipeline for parsing ActiveMQ logs.
 processors:
+  - set:
+      field: event.ingested
+      value: '{{_ingest.timestamp}}'
   - grok:
       field: message
       pattern_definitions:

x-pack/filebeat/module/aws/cloudtrail/ingest/pipeline.yml

@@ -1,6 +1,9 @@
 ---
 description: Pipeline for AWS CloudTrail Logs
 processors:
+  - set:
+      field: event.ingested
+      value: '{{_ingest.timestamp}}'
   - rename:
       field: "message"
       target_field: "event.original"
@@ -614,7 +617,7 @@ processors:
         if (ctx.event.action == 'ConsoleLogin' && ctx?.aws?.cloudtrail?.flattened?.response_elements.ConsoleLogin != null) {
             ctx.event.outcome = Processors.lowercase(ctx.aws.cloudtrail.flattened.response_elements.ConsoleLogin);
         }
-        
+
         def hm = new HashMap(params.get(ctx.event.action));
         hm.forEach((k, v) -> ctx.event[k] = v);
 

x-pack/filebeat/module/aws/cloudwatch/ingest/pipeline.yml

@@ -1,6 +1,9 @@
 description: "Pipeline for CloudWatch logs"
 
 processors:
+  - set:
+      field: event.ingested
+      value: '{{_ingest.timestamp}}'
   - grok:
       field: message
       patterns:

x-pack/filebeat/module/aws/ec2/ingest/pipeline.yml

@@ -1,6 +1,9 @@
 description: "Pipeline for EC2 logs in CloudWatch"
 
 processors:
+  - set:
+      field: event.ingested
+      value: '{{_ingest.timestamp}}'
   - grok:
       field: message
       patterns:

x-pack/filebeat/module/aws/elb/ingest/pipeline.yml

@@ -1,6 +1,9 @@
 description: "Pipeline for ELB logs"
 
 processors:
+  - set:
+      field: event.ingested
+      value: '{{_ingest.timestamp}}'
   - grok:
       field: message
       # Classic ELB patterns documented in https://docs.aws.amazon.com/elasticloadbalancing/latest/classic/access-log-collection.html