fluentdのカスタムParser Pluginを作ってみる

なぜParser Pluginが必要なのか。

独自フォーマットのログをパースするときに、デフォルトのin_tail pluginの正規表現では対応が難しい場合ががあります。
例えば、下記のログが挙げられます。
このログは、Key=Valueが順不同で発生したり、そもそも発生してたりしていなかったりしてます。
さらに要求として、keyをカラム名にValueを値という形で使いたい場合にはどうすればよいでしょうか。

[2013-01-01 00:00:00] key1=value1&key2=value2
[2013-01-01 00:00:00] key2=value2&key1=value1&key3=value3

このような場合では、fluentdのParser Pluginを作るとで対応することができます。
その他の案として、Tailインプットプラグインパーサーをカスタマイズすることもできますが、
こちらの方法だとread_from_headerなどの最近追加されたオプションが使えません。
オプションが追加されたNewTailInput pluginを同様の方法で作成する方法もありますが、
少々NewTailInputが複雑そうなので今回はやめておきます。
それぞれの方法のサンプルとなるpluginを挙げます。

• parser plugin
  • fluent-plugin-grok-parser
  • fluent-plugin-multi-format-parser
• TailInput
  • Tailインプットプラグインパーサーをカスタマイズする
• NewTailInput
  • fluent-plugin-nata2

Parser Pluginを作る

それでは、Parser Pluginを作ります。
サンプルは下記です。

# filename: parser_test_format.rb
require 'fluent/parser'

module Fluent
  class TextParser
    class TestFormatParser # TestFormatParserを任意変更
      include Configurable

      def initialize
        super
      end
      def configure(conf)
        super
        @time_format = conf['time_format'] || '%Y-%m-%d %H:%M:%S'
      end
      def call(text)
        # begin: 1レコードを編集処理
        elements = /^\[(.+)\] (.+)$/.match(text)

        time = elements[1]
        time = Time.strptime(time, @time_format).to_i

        # key1=value1&key2=value2&... -> [key1=value1, key2=value2, ...]
        messages = elements[2].split('&')

        # [k1, v1, k2, v2, ...] -> {k1=>v1, k2=>v2, ...}
        record = {}
        messages.each do |message|
          kv = message.split('=') 
          while (k = kv.shift) && (v = kv.shift)
            record[k] = v
          end
        end
        # end

        if time && record
          if block_given?
            yield time, record
            return
          else
            return time, record
          end
        end
      rescue # ignore parser error
      end
    end

    # test_formatとTestFormatParserを任意変更
    register_template('test_format', Proc.new { TestFormatParser.new })
  end
end

Parser Pluginを使う

設定ファイルは下記の通りに記載します。
Gem化していない場合は、td-agentの場合には、/etc/fluent/pluginフォルダ以下に格納します。
あとは通常通りにfluentdを起動すれば大丈夫です。
NewTailInputを使うことができるので、read_from_head等のオプションも利用できます。

<source>
  type tail
  tag test.parser
  format test_format # registerで指定した名前
  path /var/log/test/test.log
  pos_file /var/log/td-agent/test.pos
  read_from_head true
</source>

まとめ

ログは、ログをパースすることも考えて、プログラミングでインプットが容易なログのフォーマットで保存するようにしましょう。
ログサイズは大きくなりますが、LTSV形式とかをおすすめしておきます。
ストレージを気にする時代でもないので、1レコードのサイズが大きくなっても問題ないでしょう。