fluentdに流れてくるデータのキーを正規表現で置き換えるfilter plugin

filter_replace_key.rb

module Fluent
  class ReplaceKeyFilter < Filter
    Fluent::Plugin.register_filter('replace_key', self)

    config_param :pattern, :string
    config_param :replacement, :string

    attr_reader :pattern, :replacement

    def configure(conf)
      super

      @pattern = Regexp.compile(conf["pattern"])
      @replacement = conf["replacement"]
    end

    def filter_stream(tag, es)
      new_es = MultiEventStream.new

      es.each do |time, record|
        begin
          modified = record.each_with_object({}) do |(key, value), hash|
            hash[key.gsub(@pattern, @replacement)] = value
          end
          new_es.add(time, modified)
        rescue => e
          router.emit_error_event(tag, time, record, e)
        end
      end

      new_es
    end
  end
end