Keen.IO-Fluentd Http Output Plugin

lass KeenHttpOutput < Fluent::Output
  Fluent::Plugin.register_output('out-keen-http', self)

  def initialize
    super
    require 'rubygems'
    require 'eventmachine'
    require 'em-http-request'
    require 'cgi'
    require 'json'
  end

  # This method is called before starting.
  def configure(conf)
    super

    @project_id = conf['project_id']
    unless @project_id
      raise ConfigError, "'project_id' parameter is required on keen-agent-http output"
    end

    @collection = conf['collection']
    unless @collection
      raise ConfigError, "'collection' parameter is required on keen-agent-http output"
    end

    @write_key = conf['write_key']
    unless @write_key
      raise ConfigError, "'write_key' parameter is required on keen-agent-http output"
    end

    @api_url = 'https://api.keen.io/3.0/projects'
  end

  # This method is called when starting.
  def start
    super
  end

  # This method is called when shutting down.
  def shutdown
    super
  end

  # This method is called when an event reaches Fluentd.
  # 'es' is a Fluent::EventStream object that includes multiple events.
  # You can use 'es.each {|time,record| ... }' to retrieve events.
  # 'chain' is an object that manages transactions. Call 'chain.next' at
  # appropriate points and rollback if it raises an exception.
  def emit(tag, es, chain)
    EventMachine.run {
      multi = EventMachine::MultiRequest.new

      i = 0
      es.each do |time,record|
        http = handle_record(tag, time, record)
        multi.add i, http
        i = i + 1
      end
      multi.callback {
        EventMachine.stop
      }
    }
    chain.next
  end

  def handle_record(tag, time, record)
    url = "%s/%s/events/%s?api_key=%s" % [
      @api_url,
      @project_id,
      @collection,
      @write_key
    ]
    http = EventMachine::HttpRequest.new(url).post({
      :head => {'Content-Type' => 'application/json'},
      :body => JSON.dump(record)
    })
    http.errback {
      $log.error url
    }
    http.callback {
      if http.response_header.status == 201
        $log.info 'SUCCESS: ' + url
      else
        $log.info 'BAD RESPONSE: ' + url
      end
    }
    return http
  end
end

Yo, I've updated this and made some modifications over at https://gist.github.com/jdrydn/edfb7ab9ef094a812733 so that it can be used to dynamically set @collection with the tag 😉