[Filebeat] Improve ECS categorization field mappings in redis module

CHANGELOG.next.asciidoc

@@ -268,6 +268,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Enhance `elasticsearch/slowlog` fileset to handle ECS-compatible logs emitted by Elasticsearch. {issue}17715[17715] {pull}17729[17729]
 - Improve ECS categorization field mappings in misp module. {issue}16026[16026] {pull}17344[17344]
 - Added Unix stream socket support as an input source and a syslog input source. {pull}17492[17492]
+- Improve ECS categorization field mappings in redis module. {issue}16179[16179] {pull}17918[17918]
 
 *Heartbeat*
 

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

@@ -0,0 +1,84 @@
+description: Pipeline for parsing redis logs
+processors:
+- grok:
+    field: message
+    patterns:
+    - (%{POSINT:process.pid:long}:%{CHAR:redis.log.role} )?(%{REDISTIMESTAMP1:redis.log.timestamp}||%{REDISTIMESTAMP2:redis.log.timestamp})
+      %{REDISLEVEL:log.level} %{GREEDYDATA:message}
+    - '%{POSINT:process.pid:long}:signal-handler \(%{POSINT:redis.log.timestamp}\)
+      %{GREEDYDATA:message}'
+    pattern_definitions:
+      CHAR: '[a-zA-Z]'
+      REDISLEVEL: '[.\-*#]'
+      REDISTIMESTAMP1: '%{MONTHDAY} %{MONTH} %{TIME}'
+      REDISTIMESTAMP2: '%{MONTHDAY} %{MONTH} %{YEAR} %{TIME}'
+- script:
+    lang: painless
+    source: >-
+      if (ctx.log.level == params.dot) {
+        ctx.log.level = params.debug;
+      } else if (ctx.log.level == params.dash) {
+        ctx.log.level = params.verbose;
+      } else if (ctx.log.level == params.asterisk) {
+        ctx.log.level = params.notice;
+      } else if (ctx.log.level == params.hash) {
+        ctx.log.level = params.warning;
+      }
+    params:
+      dot: .
+      debug: debug
+      dash: '-'
+      verbose: verbose
+      asterisk: '*'
+      notice: notice
+      hash: '#'
+      warning: warning
+- script:
+    lang: painless
+    source: >-
+      if (ctx.redis.log.role == params.master_abbrev) {
+        ctx.redis.log.role = params.master;
+      } else if (ctx.redis.log.role == params.slave_abbrev) {
+        ctx.redis.log.role = params.slave;
+      } else if (ctx.redis.log.role == params.child_abbrev) {
+        ctx.redis.log.role = params.child;
+      } else if (ctx.redis.log.role == params.sentinel_abbrev) {
+        ctx.redis.log.role = params.sentinel;
+      }
+    params:
+      master_abbrev: M
+      master: master
+      slave_abbrev: S
+      slave: slave
+      child_abbrev: C
+      child: child
+      sentinel_abbrev: X
+      sentinel: sentinel
+- rename:
+    field: '@timestamp'
+    target_field: event.created
+- date:
+    field: redis.log.timestamp
+    target_field: '@timestamp'
+    formats:
+    - dd MMM yyyy H:m:s.SSS
+    - dd MMM H:m:s.SSS
+    - dd MMM H:m:s
+    - UNIX
+    ignore_failure: true
+- remove:
+    field: redis.log.timestamp
+    ignore_failure: true
+- set:
+    field: event.kind
+    value: event
+- append:
+    field: event.category
+    value: database
+- append:
+    field: event.type
+    value: info
+on_failure:
+- set:
+    field: error.message
+    value: '{{ _ingest.on_failure_message }}'

filebeat/module/redis/log/manifest.yml

@@ -10,5 +10,5 @@ var:
     os.windows:
       - "c:/program files/Redis/logs/redis.log*"
 
-ingest_pipeline: ingest/pipeline.json
+ingest_pipeline: ingest/pipeline.yml
 input: config/log.yml

filebeat/module/redis/log/test/redis-5.0.3.log-expected.json

@@ -1,7 +1,14 @@
 [
     {
+        "event.category": [
+            "database"
+        ],
         "event.dataset": "redis.log",
+        "event.kind": "event",
         "event.module": "redis",
+        "event.type": [
+            "info"
+        ],
         "fileset.name": "log",
         "input.type": "log",
         "log.level": "notice",