Fluentd Formatter プラグイン入門

Fluentd update 2014で言及されている通り、Fluentd に Formatter プラグイン というものが導入されているので、作り方を解説します。

2014/12/14 更新

• v0.10.58、v0.12.0以降で新APIが生えた[1][2]のでそちらを使うようにサンプルコードを更新

概説

現在サポートしているのは、out_file、out_s3 プラグインだけですが、例えば

<match foo.bar>
  type file
  path /path/to/file
  format ltsv
</match>

のように format パラメタに値(このサンプルでは ltsv)を指定することで、出力形式を変えることができます。これ自体は以前からできたのですが、Formatter プラグインを作成することによって、この format を自由に増やすことができるようになりました。

デフォルト formatter 一覧

現在デフォルトで bundle されている Formatter はコード的にはこの行で定義されており、以下のようになっています。

      'out_file' => Proc.new { OutFileFormatter.new },
      'json' => Proc.new { JSONFormatter.new },
      'msgpack' => Proc.new { MessagePackFormatter.new },
      'ltsv' => Proc.new { LabeledTSVFormatter.new },
      'csv' => Proc.new { CsvFormatter.new },
      'single_value' => Proc.new { SingleValueFormatter.new },

それぞれの使い方は公式ドキュメントが詳しいのでそちらを参照すると良いでしょう。

Formatter プラグインの作り方

さて、本題である Formatter プラグインの作り方ですが、サンプルとして新規に tsv フォーマットを作ってみることにします。cf. 公式ドキュメント

以下のコードを formatter_tsv.rb というファイル名で保存します。ディレクトリは任意ですが、今回は myplugin ディレクトリに保存してみます。

# myplugin/formatter_tsv.rb
module Fluent
  module TextFormatter
    class TSVFormatter < Formatter
      Plugin.register_formatter('tsv', self)

      include Configurable # This enables the use of config_param
      include HandleTagAndTimeMixin # If you wish to use tag_key, time_key, etc.

      def configure(conf)
        super
      end

      def format(tag, time, record)
        "#{record.values.join("\t")}\n"
      end
    end
  end
end

format メソッドに、tag, time, record の３つが渡されてくるので、それを加工して、実際に出力する文字列を返します。今回は TSV フォーマットなので、record ハッシュの値 (values) を取り出して、タブ区切りで出力しています。

また include HandleTagAndTimeMixin しておくと、include_time_key, time_key, time_format, include_tag_key, tag_key, localtime, timezone といったオプションが生えるので便利です。それらのオプションについて詳しくは http://docs.fluentd.org/articles/out_file#format のページを読むと良いでしょう。

動作確認として次のような fluent.conf を書いて

<source>
  forward
</source>

<match raw.**>
  type file
  format tsv
  path /tmp/tsv_test
</match>

fluentd を起動します。plugin をおいたディレクトリを -p オプションで指定します。

$ fluentd -c fluent.conf -p myplugin

fluent-cat でテストデータを送ってみましょう

$ echo '{"a":"foo","b":"bar"}' | fluent-cat raw.test

file プラグインの出力先 /tmp/tsv_test.* に

foo[TAB]bar

のように TSV 形式で出力されていれば成功です。

今回は適当なディレクトリにおいて -p オプションでプラグインディレクトリを指定して読み込みましたが、Input プラグインや Output プラグインのように gem にして公開することも可能です。cf. sonots/fluent-plugin-formatter_tsv

Formatter プラグインをサポートするプラグインの作り方

out_file のように Formatter プラグインをサポートする Output プラグインの作り方を解説します。サンプルとして Formatter プラグイン通りに標準出力に print するだけの out_print プラグインを定義してみます。

module Fluent
  class PrintOutput < Output
    Plugin.register_output('print', self)

    config_param :format, :string, :default => 'json'

    def configure(conf)
      super

      @formatter = Plugin.new_formatter(@format)
      @formatter.configure(conf)
    end

    def emit(tag, es, chain)
      es.each do |time, record|
        print @formatter.format(tag, time, record)
      end

      chain.next
    end
  end
end

Plugin.new_formatter でインスタンスを作ります。また、オプションをサポートしている Formatter プラグインのために、#configure メソッドに conf をそのまま渡してあげます。こうしないと大抵の Formatter プラグインが持つ include_time_key オプションなどが効きません。

あとは Formatter プラグインで実装した format メソッドを呼び出して print しているだけです。

おわりに

Formatter プラグインの作り方、および Formatter プラグインをサポートしたプラグインの作り方を紹介しました。

Formatter プラグインをサポートしているのは今の所 out_file や out_s3 ぐらいですが、カスタムフォーマットを独自に定義できるようになったので、是非ご活用ください。

PS. out_stdout も Formatter プラグインをサポートしたほうがよい気がしてきたので、そのうちやります。