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 }}'
