Improve ECS categorization field mappings in logstash module (#16668) (…

…#16875)

- event.kind
- event.type
- convert pipeline to yaml

Closes #16169

(cherry picked from commit 8e8da46)

CHANGELOG.next.asciidoc

@@ -234,6 +234,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Add `cloudfoundry` input to send events from Cloud Foundry. {pull}16586[16586]
 - Improve ECS field mappings in haproxy module. {issue}16162[16162] {pull}16529[16529]
 - Allow users to override pipeline ID in fileset input config. {issue}9531[9531] {pull}16561[16561]
+- Improve ECS categorization field mappings in logstash module. {issue}16169[16169] {pull}16668[16668]
 
 *Heartbeat*
 

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

@@ -0,0 +1,50 @@
+description: Pipeline for parsing logstash logs
+on_failure:
+- set:
+    field: error.message
+    value: '{{ _ingest.on_failure_message }}'
+processors:
+- json:
+    field: message
+    target_field: logstash.log
+- rename:
+    field: '@timestamp'
+    target_field: event.created
+- convert:
+    field: logstash.log.timeMillis
+    type: string
+- date:
+    field: logstash.log.timeMillis
+    formats:
+    - UNIX_MS
+    target_field: '@timestamp'
+- rename:
+    field: logstash.log.loggerName
+    target_field: logstash.log.module
+- remove:
+    field:
+    - message
+    - logstash.log.timeMillis
+- rename:
+    field: logstash.log.logEvent.message
+    target_field: message
+- rename:
+    field: logstash.log.logEvent
+    target_field: logstash.log.log_event
+- rename:
+    field: logstash.log.level
+    target_field: log.level
+- set:
+    field: event.kind
+    value: event
+- script:
+    lang: painless
+    source: >-
+      def errorLevels = ["ERROR", "FATAL"];
+      if (ctx?.log?.level != null) {
+        if (errorLevels.contains(ctx.log.level)) {
+          ctx.event.type = "error";
+        } else {
+          ctx.event.type = "info";
+        }
+      }

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

@@ -0,0 +1,59 @@
+description: Pipeline for parsing logstash logs in the plain format
+on_failure:
+- set:
+    field: error.message
+    value: '{{ _ingest.on_failure_message }}'
+processors:
+- grok:
+    field: message
+    pattern_definitions:
+      LOGSTASH_CLASS_MODULE: '[\w\.]+'
+      LOGSTASH_LOGLEVEL: INFO|ERROR|DEBUG|FATAL|WARN|TRACE
+      GREEDYMULTILINE: |-
+        (.|
+        )*
+    patterns:
+    - \[%{TIMESTAMP_ISO8601:logstash.log.timestamp}\]\[%{LOGSTASH_LOGLEVEL:log.level}\s?\]\[%{LOGSTASH_CLASS_MODULE:logstash.log.module}\s*\]\[%{WORD:logstash.log.pipeline_id}\]
+      %{GREEDYMULTILINE:message}
+    - \[%{TIMESTAMP_ISO8601:logstash.log.timestamp}\]\[%{LOGSTASH_LOGLEVEL:log.level}\s?\]\[%{LOGSTASH_CLASS_MODULE:logstash.log.module}\s*\]
+      %{GREEDYMULTILINE:message}
+- rename:
+    field: '@timestamp'
+    target_field: event.created
+- date:
+    if: ctx.event.timezone == null
+    field: logstash.log.timestamp
+    target_field: '@timestamp'
+    formats:
+    - yyyy-MM-dd'T'HH:mm:ss,SSS
+    on_failure:
+    - append:
+        field: error.message
+        value: '{{ _ingest.on_failure_message }}'
+- date:
+    if: ctx.event.timezone != null
+    field: logstash.log.timestamp
+    target_field: '@timestamp'
+    formats:
+    - yyyy-MM-dd'T'HH:mm:ss,SSS
+    timezone: '{{ event.timezone }}'
+    on_failure:
+    - append:
+        field: error.message
+        value: '{{ _ingest.on_failure_message }}'
+- remove:
+    field: logstash.log.timestamp
+- set:
+    field: event.kind
+    value: event
+- script:
+    lang: painless
+    source: >-
+      def errorLevels = ["ERROR", "FATAL"];
+      if (ctx?.log?.level != null) {
+        if (errorLevels.contains(ctx.log.level)) {
+          ctx.event.type = "error";
+        } else {
+          ctx.event.type = "info";
+        }
+      }

filebeat/module/logstash/log/manifest.yml

@@ -9,5 +9,5 @@ var:
     os.windows:
       - c:/programdata/logstash/logs/logstash-{{.format}}*.log
 
-ingest_pipeline: ingest/pipeline-{{.format}}.json
+ingest_pipeline: ingest/pipeline-{{.format}}.yml
 input: config/log.yml

filebeat/module/logstash/log/test/logstash-json.log-expected.json

@@ -2,7 +2,9 @@
     {
         "@timestamp": "2019-01-07T21:25:21.871Z",
         "event.dataset": "logstash.log",
+        "event.kind": "event",
         "event.module": "logstash",
+        "event.type": "info",
         "fileset.name": "log",
         "input.type": "log",
         "log.level": "INFO",
@@ -28,7 +30,9 @@
     {
         "@timestamp": "2019-01-07T21:25:22.538Z",
         "event.dataset": "logstash.log",
+        "event.kind": "event",
         "event.module": "logstash",
+        "event.type": "info",
         "fileset.name": "log",
         "input.type": "log",
         "log.level": "INFO",
@@ -43,7 +47,9 @@
     {
         "@timestamp": "2019-01-07T21:25:22.594Z",
         "event.dataset": "logstash.log",
+        "event.kind": "event",
         "event.module": "logstash",
+        "event.type": "info",
         "fileset.name": "log",
         "input.type": "log",
         "log.level": "INFO",

filebeat/module/logstash/log/test/logstash-plain-7.4.log-expected.json

@@ -2,8 +2,10 @@
     {
         "@timestamp": "2019-11-20T19:04:48.468-02:00",
         "event.dataset": "logstash.log",
+        "event.kind": "event",
         "event.module": "logstash",
         "event.timezone": "-02:00",
+        "event.type": "info",
         "fileset.name": "log",
         "input.type": "log",
         "log.level": "WARN",

filebeat/module/logstash/log/test/logstash-plain.log-expected.json

@@ -2,8 +2,10 @@
     {
         "@timestamp": "2017-10-23T14:20:12.046-02:00",
         "event.dataset": "logstash.log",
+        "event.kind": "event",
         "event.module": "logstash",
         "event.timezone": "-02:00",
+        "event.type": "info",
         "fileset.name": "log",
         "input.type": "log",
         "log.level": "INFO",
@@ -15,8 +17,10 @@
     {
         "@timestamp": "2017-11-20T03:55:00.318-02:00",
         "event.dataset": "logstash.log",
+        "event.kind": "event",
         "event.module": "logstash",
         "event.timezone": "-02:00",
+        "event.type": "info",
         "fileset.name": "log",
         "input.type": "log",
         "log.flags": [