FortiGate の CSV format な syslog を parse する fluentd plugin

fluentd-plugin-fortigate-csv-parser.rb

module Fluent
  class FortigateSyslogParseOutput < Output
    Fluent::Plugin.register_output('forti_log_parser', self)

    config_param :remove_prefix,   :string, :default => nil
    config_param :add_prefix,      :string, :default => nil
    config_param :message_key,     :string, :default => 'message'
    config_param :keys,            :string, :default => nil
    config_param :remove_keys,     :string, :default => nil

    def configure(conf)
      super

      if @remove_prefix
        @removed_prefix_string = @remove_prefix + '.'
        @removed_length = @removed_prefix_string.length
      end
      if @add_prefix
        @added_prefix_string = @add_prefix + '.'
      end

      if @keys
        if @remove_keys
          raise ConfigError, "forti_log_parser: 'keys' and 'remove_keys' parameters are exclusive"
        end
        @keys = Hash[@keys.split(',').map {|x| [x, 1] }]
      end
      if @remove_keys
        @remove_keys = Hash[@remove_keys.split(',').map {|x| [x, 1] }]
      end
    end

    def emit(tag, es, chain)
      _tag = tag.clone

      if @remove_prefix and
        ((tag.start_with?(@removed_prefix_string) && tag.length > @removed_length) || tag == @remove_prefix)
        tag = tag[@removed_length..-1] || ''
      end

      if @add_prefix
        tag = tag && tag.length > 0 ? @added_prefix_string + tag : @add_prefix
      end

      es.each do |time, record|
        time, record = parse(record)
        Engine.emit(tag, time, record)
      end

      chain.next
    end

    def parse(record)
      message = record[@message_key]
      record.delete(@message_key)
      data = message.split(/\s+/, 5).pop
      data.gsub(/\G[^,=]+=(:?"[^"]*"|[^,]+)(:?,|$)/) { |kv|
        (k, v) = kv.strip.split(/=/, 2)
        if (k == 'date' or k == 'time' or
           (@keys and @keys.has_key?(k)) or
           (@remove_keys and not @remove_keys.has_key?(k)) or
           (!@keys and !@remove_keys))
          record[k] = v.gsub(/,$/, '').gsub(/^"(.*)"$/, '\1')
        end
      }

      time = Time.strptime(record["date"] + " " + record["time"], '%Y-%m-%d %H: %M:%S').to_i

      record.delete("date")
      record.delete("time")

      [ time, record ]
    end
  end
end