由于elasticsearch不支持关系型查询。运维审计系统的，审计log在如elasticsearch前需要对token字段进行匹配，开发这个小插件来完成匹配进入elasticsearch。下面是默认把获得的token字段通过api请求返回，name字段追加到fluentd的信息里入库es

out_api.rb

require 'socket'
require 'net/http'
require 'json'

class Fluent::ApiOutput < Fluent::Output
  Fluent::Plugin.register_output('api', self)

  include Fluent::SetTagKeyMixin
  config_set_default :include_tag_key, false

  include Fluent::SetTimeKeyMixin
  config_set_default :include_time_key, true

  config_param :add_prefix, :string, :default => nil
  config_param :api_url, :string, :default => 'http://localhost/api/?id='
  config_param :new_key_name, :string, :default => 'name'
  config_param :old_key_name, :string, :default => 'token'

  def configure(conf)
    super

    if @new_key_name.empty?
      raise Fluent::ConfigError, "new_key_name is must not be specified"
    end

    if @add_prefix
      @added_prefix_string = @add_prefix + '.'
    end
  end
  def getname(token)
    begin
        uri = URI("#{@apiurl}#{@old_key_name}")
        json = JSON.parse(Net::HTTP.get(uri))
        return json[0]['cn']
    rescue
        return ""
    end
  end
  def emit(tag, es, chain)
    if @add_prefix
      tag = if tag.length > 0
              @added_prefix_string + tag
            else
              @add_prefix
            end
    end

    es.each do |time,record|
      record[@new_key_name] = getname(record[@old_key_name])
      Fluent::Engine.emit(tag, time, record)
    end

    chain.next
  end
end