fluentdで独自フォーマットを処理するParser Pluginを作る

Why?

fluentdはかなり便利なのですが、fluentdがデフォルトで対応していないフォーマットのログを出すソフトウェアもたくさんあります。そこで、そういったログを処理するために作ります。公式のドキュメントが充実しているので、さほど手間はかかりません。

フォーマット

Feb 01 00:12:30.000 [name1] key|value

みたいなログを、

2017-02-01...(省略) tag: 
{
  "[name1]":"key|value",
   "name2":"特定条件のvalue"
}

のようにパースするPluginを作ります。

考えるべきこと

• 時間のフォーマットの読み取り
• name, key, valueのparseメソッドでの処理

ベース

まずは、公式からParser Pluginをコピペしましょう。

module Fluent
  class TextParser
    class DokujiLogParser < Parser
      # Register this parser as "dokuji"
      Plugin.register_parser("dokuji", self)

      config_param :delimiter, :string, :default => " "

      (省略)

      def parse(text)
        time, key_values = text.split(" ", 2)

        (省略)

        record[k] = v
        yield time, record
      end
    end
  end
end

このように、インスタンス変数を設定し、parseメソッドに処理を書く、という形になっています。

時間

Rubyのstrptimeのようにフォーマットを指定して、

config_param :time_format, :string, :default => "%b %d %H:%M:%S.%L"
# Feb 01 00:12:30.000
# %b  %d %H:%M:%S.%L

のように設定します。

name, key, valueのパース

textには、行単位のデータが渡されるので、

# デリミタの指定
config_param :delimiter, :string, :default => " "

# parseメソッド
def parse(text)

  # テキストの分割
  splited = text.split(@delimiter, 4)
  time = splited[0..2].join(" ")
  data = splited[3]

  # 時間のパース
  time = @time_parser.parse(time)

  # recordの生成
  record = {}
  k, v = data.split(@delimiter, 2)
  record[k] = v

  yield time, record
end

このような感じでRubyのメソッドを使って普通に処理をしていきます。
また、特殊なvalueの処理は、

if value.macth(/条件/)
    record["name2"] = value.split("|")[1]
end

のようなことをparseメソッド内で行うだけです。

結果

formatにdokujiを指定するだけで、

2017-02-01 00:04:22.000000000 +0900 tag: 
{
  "[name1]":"key|value",
   "name2":"value"
}

このようにパースできます。

最後に

fluentd初心者なので、ここが間違っているとか、何かあったらコメントしていただけると幸いです。

参考文献

いまさらだけど fluentd に入門した - えいのうにっき
fluentdインストール（丁寧な説明つき） - Qiita
Writing plugins | Fluentd
Match and Handle Date/Time Formats in Td-Agent or Fluentd - Data Pipeline Architect