Contents

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

Version data entries

1 entries across 1 versions & 1 rubygems

Version	Path
fluent-plugin-map-0.0.3.1	lib/fluent/plugin/out_map.rb