rabbitmq_config

input {
    generator {
        lines => [
            "2017-02-12 14:50:98 DEBUG Main:12 - Test log file",
            "2017-02-12 14:51:98 INFO Main:13 - Test log file",
            "2017-02-12 14:52:98 WARN Main:14 - Test log file",
            "2017-02-12 14:53:98 ERROR Main:15 - Test log file",
            "2017-02-12 14:54:98 FATAL Main:16 - Test log file"
        ]
        add_field => { "service" => "a" }
        add_field => { "environment" => "development" }
        count => 50
    }
    generator {
        lines => [
            "2017-02-12 14:50:98 DEBUG Main:12 - Test log file",
            "2017-02-12 14:51:98 INFO Main:13 - Test log file",
            "2017-02-12 14:52:98 WARN Main:14 - Test log file",
            "2017-02-12 14:53:98 ERROR Main:15 - Test log file",
            "2017-02-12 14:54:98 FATAL Main:16 - Test log file"
        ]
        add_field => { "service" => "b" }
        add_field => { "environment" => "development" }
        count => 50
    }
    generator {
        lines => [
            "2017-02-12 14:50:98 DEBUG Main:12 - Test log file",
            "2017-02-12 14:51:98 INFO Main:13 - Test log file",
            "2017-02-12 14:52:98 WARN Main:14 - Test log file",
            "2017-02-12 14:53:98 ERROR Main:15 - Test log file",
            "2017-02-12 14:54:98 FATAL Main:16 - Test log file"
        ]
        add_field => { "service" => "a" }
        add_field => { "environment" => "production" }
        count => 50
    }
}
output {
    stdout { codec => rubydebug }
    rabbitmq {
        exchange => "logs"
        host => "localhost"
        exchange_type => "topic"
        key => "logs.%{environment}.%{service}"
        durable => false
        persistent => false
    }
}

read.rb

#!/usr/bin/env ruby

require 'bunny'
require 'json'
require 'optparse'


options = {}
OptionParser.new do |opts|
  opts.banner = "Usage: read.rb [options]"
  opts.on('-e', '--environment NAME', 'Environment name') { |v| options[:environment] = v }
  opts.on('-s', '--service HOST', 'Service name') { |v| options[:service] = v }
end.parse!

conn = Bunny.new
conn.start

ch  = conn.create_channel
x   = ch.topic("logs")
q   = ch.queue("", :exclusive => true)

environment = options[:environment] ? options[:environment] : '*'
service = options[:service] ? options[:service] : '*'
routing_key = "logs.#{environment}.#{service}"

q.bind(x, :routing_key => routing_key)


puts "Looking for logs matching '#{routing_key}', To exit press CTRL+C"

begin
  q.subscribe(:block => true) do |delivery_info, properties, body|
    puts "[#{delivery_info.routing_key}] #{JSON.parse(body)["message"]}"
  end
rescue Interrupt => _
  ch.close
  conn.close
end