fluent-plugin-kestrel

out_kestrel.rb

module Fluent                                                                                                                                                                                                                                                                          
  class KestrelOutput < BufferedOutput
    Fluent::Plugin.register_output('kestrel', self)
    attr_reader :hosts, :kestrel

    def initialize
      super
      require 'msgpack'
      require 'kestrel'
    end

    def configure(conf)
      super

      hosts_source = conf.has_key?('hosts') ? conf['hosts'] : 'localhost:22133'
      @hosts = hosts_source.split(/\s*,\s*/)
    end

    def start
      super

      @kestrel = Kestrel::Client.new(@hosts)
    end

    def shutdown
    end

    def format(tag, time, record)
      event.record.to_msgpack
    end
----
    def format(tag, event)
      event.record.to_msgpack
    end

    def write(chunk)
      chunk.open { |io|
        begin
          MessagePack::Unpacker.new(io).each.each_with_index { |record, index|
            @kestrel.set(record['key'], record['value'],0)
          }
        rescue EOFError
          # EOFError always occured when reached end of chunk.
        end
      }
    end
  end
end