module Fluent class MapOutput < Fluent::Output Fluent::Plugin.register_output('map', self) config_param :map, :string, :default => nil config_param :tag, :string, :default => nil config_param :key, :string, :default => nil #deprected config_param :time, :string, :default => nil config_param :record, :string, :default => nil config_param :multi, :bool, :default => false def configure(conf) super if @map $log.debug { "map: #{@map}" } @mode = "tuple" elsif (@tag || @key) && @time && @record @tag ||= @key raise ConfigError, "multi and 3 parameters(tag, time, and record) are not compatible" if @multi $log.debug { "tag: #{@tag}, time: #{@time}, record: #{@record}" } @mode = "each" else raise ConfigError, "Either map or 3 parameters(key, time, and record) is required " end end def emit(tag, es, chain) begin tuples = [] es.each {|time, record| case @mode when "tuple" new_tuple = eval(@map) if @multi tuples.concat new_tuple else tuples << new_tuple end when "each" new_tag = eval(@tag) new_time = eval(@time) new_record = eval(@record) tuples << [new_tag, new_time, new_record] end } tuples.each do |tag, time, record| if time == nil or record == nil raise SyntaxError.new end $log.trace { [tag, time, record].inspect } Fluent::Engine::emit(tag, time, record) end chain.next tuples rescue SyntaxError => e chain.next $log.error "map command is syntax error: #{@map}" e #for test end end end end