crowbar/crowbar-openstack

input {
    kafka {
        zk_connect => "<%= @zookeeper_hosts %>"
        topic_id => "log"
        group_id => "transformer-logstash-consumer"
        consumer_threads => "1"
        fetch_message_max_bytes => "1048576"
    }
}

filter {
    ruby {
      code => "event['message_tmp'] = event['log']['message'][0..49]"
    }
    grok {
        match => {
            "[message_tmp]" => "(?i)(?<log_level>AUDIT|CRITICAL|DEBUG|INFO|TRACE|ERR(OR)?|WARN(ING)?)|\"level\":\s?(?<log_level>\d{2})"
        }
    }
    if ! [log_level] {
        grok {
            match => {
                "[log][message]" => "(?i)(?<log_level>AUDIT|CRITICAL|DEBUG|INFO|TRACE|ERR(OR)?|WARN(ING)?)|\"level\":\s?(?<log_level>\d{2})"
            }
        }
    }
    ruby {
        init => "
            LOG_LEVELS_MAP = {
              # SYSLOG
              'warn' => :Warning,
              'err' => :Error,
              # Bunyan errcodes
              '10' => :Trace,
              '20' => :Debug,
              '30' => :Info,
              '40' => :Warning,
              '50' => :Error,
              '60' => :Fatal
            }
        "
        code => "
            if event['log_level']
                # keep original value
                log_level = event['log_level'].downcase
                if LOG_LEVELS_MAP.has_key?(log_level)
                    event['log_level_original'] = event['log_level']
                    event['log_level'] = LOG_LEVELS_MAP[log_level]
                else
                    event['log_level'] = log_level.capitalize
                end
            else
                event['log_level'] = 'Unknown'
            end
        "
    }

    mutate {
        add_field => {
            "[log][level]" => "%{log_level}"
        }

        # remove temporary fields
        remove_field => ["log_level", "message_tmp"]
    }
}

output {
    kafka {
        bootstrap_servers => "<%= @kafka_hosts %>"
        topic_id => "transformed-log"
    }
}