Improve ECS categorization field mapping in kafka module (#16645) (#1…

…6742)

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

Closes #16167

(cherry picked from commit e7f9335)

CHANGELOG.next.asciidoc

@@ -237,6 +237,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - 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]
 - Improve ECS categorization field mappings in iis module. {issue}16165[16165] {pull}16618[16618]
+- Improve ECS categorization field mapping in kafka module. {issue}16167[16167] {pull}16645[16645]
 
 *Heartbeat*
 

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

@@ -0,0 +1,73 @@
+description: Pipeline for parsing Kafka log messages
+processors:
+- grok:
+    field: message
+    trace_match: true
+    patterns:
+    - (?m)%{TIMESTAMP_ISO8601:kafka.log.timestamp}. %{LOGLEVEL:log.level} +%{JAVALOGMESSAGE:message}
+      \(%{JAVACLASS:kafka.log.class}\)$[ \n]*(?'kafka.log.trace.full'.*)
+- grok:
+    field: message
+    pattern_definitions:
+      KAFKA_COMPONENT: '[^\]]*'
+    patterns:
+    - \[%{KAFKA_COMPONENT:kafka.log.component}\][,:.]? +%{JAVALOGMESSAGE:message}
+    on_failure:
+    - set:
+        field: kafka.log.component
+        value: unknown
+- grok:
+    field: kafka.log.trace.full
+    ignore_missing: true
+    patterns:
+    - '%{JAVACLASS:kafka.log.trace.class}:\s*%{JAVALOGMESSAGE:kafka.log.trace.message}'
+    on_failure:
+    - remove:
+        field: kafka.log.trace
+- remove:
+    field: kafka.log.trace.full
+    ignore_missing: true
+- rename:
+    field: '@timestamp'
+    target_field: event.created
+- date:
+    if: ctx.event.timezone == null
+    field: kafka.log.timestamp
+    target_field: '@timestamp'
+    formats:
+    - yyyy-MM-dd HH:mm:ss,SSS
+    on_failure:
+    - append:
+        field: error.message
+        value: '{{ _ingest.on_failure_message }}'
+- date:
+    if: ctx.event.timezone != null
+    field: kafka.log.timestamp
+    target_field: '@timestamp'
+    formats:
+    - yyyy-MM-dd HH:mm:ss,SSS
+    timezone: '{{ event.timezone }}'
+    on_failure:
+    - append:
+        field: error.message
+        value: '{{ _ingest.on_failure_message }}'
+- remove:
+    field: kafka.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";
+        }
+      }
+on_failure:
+- set:
+    field: error.log
+    value: '{{ _ingest.on_failure_message }}'

filebeat/module/kafka/log/manifest.yml

@@ -10,5 +10,5 @@ var:
       - "{{.kafka_home}}/logs/state-change.log*"
       - "{{.kafka_home}}/logs/kafka-*.log*"
 
-ingest_pipeline: ingest/pipeline.json
+ingest_pipeline: ingest/pipeline.yml
 input: config/log.yml